orchestrata_ultimate.session.facts

Reference › orchestrata_ultimate › session › facts

FactGenerator turns the contents of context.json into Prolog fact strings. session_fact renders one SessionRecord and work_fact one WorkItem; generate_all_facts renders everything, session_facts_from_list renders a given list, and write_facts_file writes the result to a file and returns the number of facts. sessions_by_purpose and work_by_priority return the facts matching a purpose pattern or a priority. The records come from models via the ContextFileManager. This is a separate path from the typed runtime facts of the FactEmitter; the PrologSessionInterface uses it to load facts into the engine.

At a glance

Class diagram

Dependency diagram

API

Classes

FactGenerator

Decorators: traces(spec='006', story='FR-003', reqs=['REQ-006-FR3-001']) · Line 20

Generates Prolog facts from session data.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/facts.py

"""Fresh implementation of Prolog fact generation.

Generates Prolog facts from session data.
"""

import logging
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.context import ContextFileManager
from orchestrata_ultimate.session.models import SessionRecord, WorkItem

# Spec: 006 - Session Management


logger = logging.getLogger(__name__)


@traces(spec="006", story="FR-003", reqs=["REQ-006-FR3-001"])
class FactGenerator:
    """Generates Prolog facts from session data."""

    def __init__(self, root: str | Path) -> None:
        """Initialize generator.

        Args:
            root: Project root directory.
        """
        self.root = Path(root)
        self.ctx = ContextFileManager(root)

    @traces(spec="006", story="FR-008", reqs=["REQ-006-FR8-002"])
    def generate_all_facts(self) -> list[str]:
        """Generate all Prolog facts.

        Returns:
            List of Prolog facts.
        """
        facts = []
        facts.extend(self._state_facts())
        facts.extend(self._session_facts())
        facts.extend(self._work_facts())
        return facts

    @traces(spec="006", story="US-003", reqs=["REQ-006-US-003-005"])
    def write_facts_file(self, output_path: str | Path) -> int:
        """Write facts to file.

        Args:
            output_path: Output path.

        Returns:
            Number of facts written.
        """
        facts = self.generate_all_facts()
        content = "\n".join(facts) + "\n"
        Path(output_path).write_text(content, encoding="utf-8")
        return len(facts)

    @traces(spec="006", story="FR-003", reqs=["REQ-006-FR3-001"])
    def _state_facts(self) -> list[str]:
        """Generate state facts."""
        try:
            state = self.ctx.read_state()
        except FileNotFoundError:
            return []

        facts = []

        if state.active_session is not None:
            facts.append(f"project_current_session({state.active_session}).")

        facts.append(f"project_version('{state.version}').")
        facts.append(f"project_branch('{state.git_branch}').")
        facts.append(f"project_status('{state.status_line}').")

        if state.test_state:
            facts.append(f"project_test_state('{state.test_state}').")

        return facts

    @traces(spec="006", story="FR-003", reqs=["REQ-006-FR3-001"])
    def _session_facts(self) -> list[str]:
        """Generate session facts from context file."""
        try:
            sessions = self.ctx.read_sessions()
        except FileNotFoundError:
            return []
        return self.session_facts_from_list(sessions)

    @traces(spec="006", story="FR-003", reqs=["REQ-006-FR3-001"])
    def session_facts_from_list(self, sessions: list) -> list[str]:
        """Generate facts from a list of sessions.

        Args:
            sessions: List of SessionRecord objects.

        Returns:
            List of Prolog facts.
        """
        facts = []

        for session in sessions:
            ts = int(session.when.timestamp())
            facts.append(
                f"session({session.seq}, '{session.purpose}', '{session.ai_model}', {ts})."
            )

            facts.extend(
                f"session_objective({session.seq}, '{self._escape(obj)}')."
                for obj in session.objectives
            )

            facts.extend(
                f"session_achievement({session.seq}, '{self._escape(ach)}')."
                for ach in session.achievements
            )

            facts.extend(
                f"session_artifact({session.seq}, '{self._escape(art.filepath)}')."
                for art in session.artifacts
            )

            facts.extend(
                f"session_decision({session.seq}, '{self._escape(choice.choice)}')."
                for choice in session.choices
            )

            if session.handoff:
                facts.extend(
                    f"session_blocker({session.seq}, '{self._escape(b)}')."
                    for b in session.handoff.blockers
                )
                facts.extend(
                    f"session_active_work({session.seq}, '{self._escape(w)}')."
                    for w in session.handoff.active_work
                )

        return facts

    @traces(spec="006", story="FR-003", reqs=["REQ-006-FR3-002"])
    def _work_facts(self) -> list[str]:
        """Generate work item facts."""
        try:
            items = self.ctx.read_work_items()
        except FileNotFoundError:
            return []

        facts = []

        for idx, item in enumerate(items):
            ts = int(item.timestamp.timestamp())
            priority = item.priority or "none"
            state = item.state.value

            facts.append(
                f"work_item({idx}, '{self._escape(item.text)}', '{state}', '{priority}', {ts})."
            )

            if item.notes:
                facts.append(f"work_item_notes({idx}, '{self._escape(item.notes)}').")

        return facts

    @traces(spec="006", story="FR-003", reqs=["REQ-006-FR3-002"])
    def session_fact(self, session: SessionRecord) -> str:
        """Generate single session fact."""
        ts = int(session.when.timestamp())
        return (
            f"session({session.seq}, '{session.purpose}', '{session.ai_model}', {ts})."
        )

    @traces(spec="006", story="FR-003", reqs=["REQ-006-FR3-002"])
    def work_fact(self, item: WorkItem, idx: int = 0) -> str:
        """Generate single work item fact."""
        ts = int(item.timestamp.timestamp())
        priority = item.priority or "none"
        state = item.state.value
        return f"work_item({idx}, '{self._escape(item.text)}', '{state}', '{priority}', {ts})."

    def _escape(self, text: str) -> str:
        """Escape quotes for Prolog."""
        return text.replace("'", "\\'")

    @traces(spec="006", story="FR-009", reqs=["REQ-006-FR9-002"])
    def sessions_by_purpose(self, pattern: str) -> list[str]:
        """Get session facts matching purpose pattern."""
        sessions = self.ctx.read_sessions()
        return [
            self.session_fact(s)
            for s in sessions
            if pattern.lower() in s.purpose.lower()
        ]

    def work_by_priority(self, priority: str) -> list[str]:
        """Get work item facts by priority."""
        items = self.ctx.read_work_items()
        return [
            self.work_fact(item, idx)
            for idx, item in enumerate(items)
            if item.priority == priority
        ]