orchestrata_ultimate.session.controller._core

Reference › orchestrata_ultimate › session › controller › _core

SessionController is the façade of the session package: it runs the lifecycle of a session (spec 006) with thread-safe access. It is composed from four mixins, GoalMixin, GitOpsMixin, ChangelogMixin and TestStateMixin. It holds the project root, the AI-model name, a ContextFileManager and a SessionGovernance, and guards the active session with a lock. The CLI (cli/lifecycle), the three Claude Code hooks and the event-emitting wrapper use it.

begin(purpose, objectives) makes sure the context files exist, auto-closes a previous session that was never closed (marking it auto-closed), takes the next session number, writes a new SessionRecord and sets the status line to ACTIVE. log_work, log_artifact, log_choice, log_achievement and set_handoff add to the active session. resume reopens a session, and get_session, all_sessions and session_count read.

end first runs the goal Stop gate (unless skip_goal_check is set; an active, unmet goal raises GoalNotAchievedError), records achievements, upcoming items and the session length, updates the project state to CLOSED with the test count, then commits and pushes code changes, updates the changelog, creates the session tag and pushes it. validate_and_end first runs _validate_checklist: work logged, artifacts recorded, decisions documented, handoff set and tests passing. It also asks the Prolog rule can_close_session/1 through the governance object, and when that rule cannot be consulted the Python observations stand. Any failure raises ChecklistError with the list.

At a glance

Class diagram

Dependency diagram

API

Classes

SessionController

Bases: GoalMixin, GitOpsMixin, ChangelogMixin, TestStateMixin · Decorators: traces(spec='006', story='FR-010', reqs=['REQ-006-FR10-001', 'REQ-006-FR10-002', 'REQ-006-FR10-003', 'REQ-006-FR10-004'…, traces(spec='006', story='FR-009', reqs=['REQ-006-FR9-001', 'REQ-006-FR9-002', 'REQ-006-FR9-003', 'REQ-006-FR9-004', 'R…, traces(spec='006', story='FR-007', reqs=['REQ-006-FR7-001', 'REQ-006-FR7-002', 'REQ-006-FR7-003', 'REQ-006-FR7-004', 'R…, traces(spec='006', story='FR-006', reqs=['REQ-006-FR6-001', 'REQ-006-FR6-002', 'REQ-006-FR6-003', 'REQ-006-FR6-004', 'R…, traces(spec='006', story='FR-005', reqs=['REQ-006-FR5-001', 'REQ-006-FR5-002', 'REQ-006-FR5-003', 'REQ-006-FR5-004', 'R…, traces(spec='006', story='FR-004', reqs=['REQ-006-FR4-001', 'REQ-006-FR4-002', 'REQ-006-FR4-003', 'REQ-006-FR4-004', 'R…, traces(spec='006', story='FR-001', reqs=['REQ-006-FR1-001', 'REQ-006-FR1-002', 'REQ-006-FR1-003', 'REQ-006-FR1-004', 'R…, traces(spec='006', story='US-002', reqs=['REQ-006-US-002-001', 'REQ-006-US-002-002', 'REQ-006-US-002-003', 'REQ-006-US-… · Line 123

Controls session lifecycle with thread-safe access.

Source

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

# mypy: ignore-errors
"""SessionController core — lifecycle, logging, and validation.

Handles session lifecycle operations with thread safety. Git, changelog,
test-state, and goal responsibilities live in sibling mixin modules.
"""

import logging
import threading
from datetime import datetime
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.context import ContextFileManager
from orchestrata_ultimate.session.controller._changelog import ChangelogMixin
from orchestrata_ultimate.session.controller._git import GitOpsMixin
from orchestrata_ultimate.session.controller._goals import GoalMixin
from orchestrata_ultimate.session.controller._tests import TestStateMixin
from orchestrata_ultimate.session.exceptions import ChecklistError
from orchestrata_ultimate.session.governance import SessionGovernance
from orchestrata_ultimate.session.models import (
    FileArtifact,
    HandoffData,
    LogEntry,
    SessionRecord,
    TechChoice,
    default_ai_model,
)

logger = logging.getLogger(__name__)


@traces(
    spec="006",
    story="FR-010",
    reqs=[
        "REQ-006-FR10-001",
        "REQ-006-FR10-002",
        "REQ-006-FR10-003",
        "REQ-006-FR10-004",
        "REQ-006-FR10-005",
    ],
)
@traces(
    spec="006",
    story="FR-009",
    reqs=[
        "REQ-006-FR9-001",
        "REQ-006-FR9-002",
        "REQ-006-FR9-003",
        "REQ-006-FR9-004",
        "REQ-006-FR9-005",
    ],
)
@traces(
    spec="006",
    story="FR-007",
    reqs=[
        "REQ-006-FR7-001",
        "REQ-006-FR7-002",
        "REQ-006-FR7-003",
        "REQ-006-FR7-004",
        "REQ-006-FR7-005",
        "REQ-006-FR7-006",
    ],
)
@traces(
    spec="006",
    story="FR-006",
    reqs=[
        "REQ-006-FR6-001",
        "REQ-006-FR6-002",
        "REQ-006-FR6-003",
        "REQ-006-FR6-004",
        "REQ-006-FR6-005",
    ],
)
@traces(
    spec="006",
    story="FR-005",
    reqs=[
        "REQ-006-FR5-001",
        "REQ-006-FR5-002",
        "REQ-006-FR5-003",
        "REQ-006-FR5-004",
        "REQ-006-FR5-005",
    ],
)
@traces(
    spec="006",
    story="FR-004",
    reqs=[
        "REQ-006-FR4-001",
        "REQ-006-FR4-002",
        "REQ-006-FR4-003",
        "REQ-006-FR4-004",
        "REQ-006-FR4-005",
    ],
)
@traces(
    spec="006",
    story="FR-001",
    reqs=[
        "REQ-006-FR1-001",
        "REQ-006-FR1-002",
        "REQ-006-FR1-003",
        "REQ-006-FR1-004",
        "REQ-006-FR1-005",
    ],
)
@traces(
    spec="006",
    story="US-002",
    reqs=[
        "REQ-006-US-002-001",
        "REQ-006-US-002-002",
        "REQ-006-US-002-003",
        "REQ-006-US-002-004",
        "REQ-006-US-002-005",
        "REQ-006-US-002-006",
    ],
)
class SessionController(GoalMixin, GitOpsMixin, ChangelogMixin, TestStateMixin):
    """Controls session lifecycle with thread-safe access."""

    def __init__(
        self,
        root: str | Path,
        ai_model: str | None = None,
        *,
        explicit_root: bool = True,
        governance: SessionGovernance | None = None,
    ) -> None:
        """Initialize controller.

        Args:
            root: Project root directory.
            ai_model: AI model name. Defaults to ``default_ai_model()`` (call-time,
                env-overridable) when not given.
            explicit_root: Whether *root* was deliberately supplied by the
                caller (the default -- every existing caller, incl. the
                session CLI, keeps pushing exactly as before) rather than
                guessed by the calling code (cwd, ``__file__``, ...). Callers
                that cannot vouch for *root* should pass ``False``:
                ``GitOpsMixin`` then still commits locally but skips the
                network push (fix/session-root, RUL-011).
            governance: The ``governance.pl`` seam consulted at lifecycle
                boundaries. Defaults to a real one that builds its engine
                LAZILY, so a controller that never opens or closes a session
                pays no Prolog startup. Injectable so tests can bind a
                scoped bridge instead of the process-global engine.
        """
        self.root = Path(root)
        self.root_explicit = explicit_root
        self.ai_model = ai_model or default_ai_model()
        self.ctx = ContextFileManager(root)
        self._active: SessionRecord | None = None
        self._lock = threading.Lock()
        self._governance = governance or SessionGovernance()

    @property
    @traces(spec="006", story="US-003", reqs=["REQ-006-US-003-001"])
    def active_session(self) -> SessionRecord | None:
        """Get currently active session (thread-safe)."""
        with self._lock:
            return self._active

    @traces(
        spec="006",
        story="US-001",
        reqs=[
            "REQ-006-FR7-001",
            "REQ-006-US-001-001",
            "REQ-006-US-001-002",
            "REQ-006-US-001-003",
            "REQ-006-US-001-005",
        ],
    )
    def begin(
        self,
        purpose: str,
        objectives: list[str] | None = None,
    ) -> SessionRecord:
        """Begin new session (thread-safe).

        Auto-creates context files if they don't exist and the project root
        is detected (via ``.venv`` or ``pyproject.toml``).

        Args:
            purpose: Session focus/purpose.
            objectives: Optional goals.

        Returns:
            Created session.

        Note:
            Auto-closes any unclosed session from previous runs.
        """
        with self._lock:
            # Ensure context files exist (auto-create if .venv is present)
            self.ctx.ensure_initialized()

            # Auto-close any unclosed session first
            self._auto_close_unclosed()

            # Get next session number (checks both context.md and history)
            seq = self.ctx.next_session_num()

            session = SessionRecord(
                seq=seq,
                when=datetime.now(),
                ai_model=self.ai_model,
                purpose=purpose,
                objectives=objectives or [],
            )

            self.ctx.add_session(session)
            self.ctx.update_state(status_line="ACTIVE")
            self._active = session
            return session

    @traces(
        spec="006",
        story="US-002",
        reqs=[
            "REQ-006-FR10-002",
            "REQ-006-FR12-001",
            "REQ-006-FR12-004",
            "REQ-006-FR5-003",
            "REQ-006-FR5-004",
            "REQ-006-FR7-004",
            "REQ-006-US-002-001",
            "REQ-006-US-002-004",
        ],
    )
    def end(
        self,
        achievements: list[str] | None = None,
        upcoming: list[str] | None = None,
        *,
        skip_goal_check: bool = False,
    ) -> SessionRecord | None:
        """End current session (thread-safe).

        Args:
            achievements: What was completed.
            upcoming: Next steps.
            skip_goal_check: Escape hatch — skip the SPEC-023 US-005 Stop
                gate (an active, incomplete, unblocked goal refuses the
                end). Use only for explicit user overrides like
                ``/goal abandon`` flows that already changed the goal.

        Returns:
            Ended session or None.

        Raises:
            GoalNotAchievedError: If the Stop gate refuses the end.
        """
        with self._lock:
            if self._active is None:
                return None

            if not skip_goal_check:
                self._enforce_goal_stop_gate()

            if achievements:
                self._active.achievements = achievements
            if upcoming:
                self._active.upcoming = upcoming

            start = self._active.when
            end = datetime.now()
            delta = end - start
            hours = delta.seconds // 3600
            mins = (delta.seconds % 3600) // 60
            self._active.length = f"{hours}h {mins}m"

            self.ctx.update_session(
                self._active.seq,
                achievements=self._active.achievements,
                upcoming=self._active.upcoming,
                length=self._active.length,
            )
            self.ctx.update_state(
                status_line="CLOSED",
                active_session=self._active.seq,
                test_state=self._get_test_state(),
            )

            session = self._active
            self._active = None

            # Commit and push any code changes before tagging
            self._commit_and_push_changes(session)

            # Update CHANGELOG.md with session achievements
            self._update_changelog(session)

            # Create git tag for this session (after changelog commit)
            self._create_session_tag(session)

            # Push session tag to origin
            self._push_session_tag(session)

            return session

    @traces(spec="006", story="FR-012", reqs=["REQ-006-FR12-001", "REQ-006-FR12-005"])
    def log_work(self, desc: str, details: list[str] | None = None) -> None:
        """Add work log entry (thread-safe).

        Args:
            desc: Work description.
            details: Optional details.
        """
        with self._lock:
            if self._active is None:
                return

            entry = LogEntry(desc=desc, details=details or [])
            self._active.log.append(entry)
            self.ctx.update_session(self._active.seq, log=self._active.log)

    @traces(spec="006", story="FR-012", reqs=["REQ-006-FR12-001", "REQ-006-FR12-005"])
    def log_artifact(
        self,
        filepath: str,
        line_count: int | None = None,
        summary: str | None = None,
    ) -> None:
        """Log created file (thread-safe).

        Args:
            filepath: File path.
            line_count: Optional line count.
            summary: Optional description.
        """
        with self._lock:
            if self._active is None:
                return

            artifact = FileArtifact(
                filepath=filepath,
                line_count=line_count,
                summary=summary,
            )
            self._active.artifacts.append(artifact)
            self.ctx.update_session(self._active.seq, artifacts=self._active.artifacts)

    @traces(spec="006", story="FR-012", reqs=["REQ-006-FR12-001", "REQ-006-FR12-005"])
    def log_choice(self, choice: str, reason: str) -> None:
        """Log technical decision (thread-safe).

        Args:
            choice: Decision made.
            reason: Rationale.
        """
        with self._lock:
            if self._active is None:
                return

            tech = TechChoice(choice=choice, reason=reason)
            self._active.choices.append(tech)
            self.ctx.update_session(self._active.seq, choices=self._active.choices)

    @traces(spec="006", story="FR-012", reqs=["REQ-006-FR12-001", "REQ-006-FR12-005"])
    def log_achievement(self, item: str) -> None:
        """Log completed item (thread-safe).

        Args:
            item: Completed item.
        """
        with self._lock:
            if self._active is None:
                return

            self._active.achievements.append(item)
            self.ctx.update_session(
                self._active.seq, achievements=self._active.achievements
            )

    @traces(spec="006", story="US-005", reqs=["REQ-006-US-005-001"])
    def set_handoff(
        self,
        decisions: list[str] | None = None,
        blockers: list[str] | None = None,
        active_work: list[str] | None = None,
        key_files: list[str] | None = None,
        cmds: list[str] | None = None,
        alerts: list[str] | None = None,
    ) -> None:
        """Set handoff data (thread-safe).

        Args:
            decisions: Key decisions.
            blockers: Current blockers.
            active_work: In-progress work.
            key_files: Important files.
            cmds: Useful commands.
            alerts: Warnings.
        """
        with self._lock:
            if self._active is None:
                return

            self._active.handoff = HandoffData(
                decisions=decisions or [],
                blockers=blockers or [],
                active_work=active_work or [],
                key_files=key_files or [],
                cmds=cmds or [],
                alerts=alerts or [],
            )
            self.ctx.update_session(self._active.seq, handoff=self._active.handoff)

    @traces(spec="006", story="US-007", reqs=["REQ-006-US-007-004"])
    def resume(self, seq: int | None = None) -> SessionRecord | None:
        """Resume a session (thread-safe).

        Args:
            seq: Session number or None for latest.

        Returns:
            Resumed session or None.
        """
        with self._lock:
            try:
                ctx = self.ctx.read()
            except FileNotFoundError:
                return None

            if seq is None:
                session = ctx.latest_session()
            else:
                session = None
                for s in ctx.session_history:
                    if s.seq == seq:
                        session = s
                        break

            if session:
                self._active = session
                return session

    @traces(spec="006", story="US-003", reqs=["REQ-006-FR10-003", "REQ-006-US-003-001"])
    def get_session(self, seq: int) -> SessionRecord | None:
        """Get session by number.

        Args:
            seq: Session number.

        Returns:
            Session or None.
        """
        ctx = self.ctx.read()
        for s in ctx.session_history:
            if s.seq == seq:
                return s
        return None

    @traces(spec="006", story="US-007", reqs=["REQ-006-US-007-005"])
    def all_sessions(self) -> list[SessionRecord]:
        """Get all sessions."""
        return self.ctx.read_sessions()

    def session_count(self) -> int:
        """Get number of sessions."""
        return len(self.ctx.read_sessions())

    def _auto_close_unclosed(self) -> SessionRecord | None:
        """Auto-close any unclosed session from previous runs.

        Returns:
            The auto-closed session if one was found, None otherwise.
        """
        try:
            ctx = self.ctx.read()
        except FileNotFoundError:
            return None
        latest = ctx.latest_session()

        if latest and latest.length is None:
            # Session was never closed properly
            latest.length = "auto-closed"
            latest.achievements.append("(auto-closed by new session)")
            self.ctx.update_session(
                latest.seq,
                length=latest.length,
                achievements=latest.achievements,
            )
            return latest
        return None

    def _validate_checklist(self) -> list[str]:
        """Validate the session checklist and return the list of failures.

        Python OBSERVES; ``governance.pl`` DECIDES. Each condition below is a
        fact about session state that only this process can see — whether a
        work log is empty, whether the test command passed. Which combination
        of them permits a close is policy, and policy lives in
        ``can_close_session/1``.

        Until 2026-08-21 the combination was the implicit ``if`` chain here
        and ``can_close_session/1`` was queried by nothing, its input facts
        asserted by nothing. The substitution is exact — checklist_complete
        is work AND choices AND handoff; blocking-issues is false with no
        ``session_issue/2`` facts — so the rule's verdict equals this
        function's old verdict, and a divergence means the rule was edited,
        which is the point.

        The failure STRINGS stay here because they are diagnostics, not
        policy: the rule base can say no, it cannot say which empty list to
        go and fill.
        """
        failures = []

        if self._active is None:
            failures.append("No active session")
            return failures

        work_logged = bool(self._active.log)
        artifacts_recorded = bool(self._active.artifacts)
        decisions_documented = bool(self._active.choices)
        handoff_set = self._active.handoff is not None
        tests_passing = self._run_tests()

        if not work_logged:
            failures.append("No work logged (session.log is empty)")

        if not artifacts_recorded:
            failures.append("No artifacts recorded (session.artifacts is empty)")

        if not decisions_documented:
            failures.append(
                "No technical decisions documented (session.choices is empty)"
            )

        if not handoff_set:
            failures.append("Handoff not set (call set_handoff() before closing)")

        if not tests_passing:
            failures.append("Tests do not pass (uv run pytest tests/ failed)")

        verdict = self._governance.may_close(
            self._active.session_id,
            checklist_complete=work_logged and decisions_documented and handoff_set,
            tests_passing=tests_passing,
            artifacts_valid=artifacts_recorded,
        )
        # None means the rule base could not be consulted — the observations
        # above stand rather than blocking every close on a load failure.
        if verdict is False and not failures:
            failures.append(
                "governance.pl can_close_session/1 refused this session — a "
                "rule denies it that the checklist above does not model "
                "(e.g. a critical session_issue/2)"
            )

        return failures

    def validate_and_end(
        self,
        achievements: list[str] | None = None,
        upcoming: list[str] | None = None,
        *,
        skip_goal_check: bool = False,
    ) -> SessionRecord:
        """End session with strict checklist validation.

        Args:
            achievements: What was completed.
            upcoming: Next steps.
            skip_goal_check: Escape hatch for the SPEC-023 Stop gate.

        Returns:
            Ended session.

        Raises:
            ChecklistError: If checklist validation fails.
        """
        failures = self._validate_checklist()
        if failures:
            raise ChecklistError(failures)

        session = self.end(achievements, upcoming, skip_goal_check=skip_goal_check)
        if session is None:
            raise ChecklistError(["No active session to end"])
        return session