orchestrata_ultimate.kernel.pipeline_session

Reference › orchestrata_ultimate › kernel › pipeline_session

The session and terminal-outcome side of PipelineRunner. PipelineSessionMixin creates the session controller, records each stage transition and artifact, checkpoints stage output for resume, receives everything addressed to the human, and ends the session when the pipeline reaches a terminal result. pipeline is its only importer; it also imports the artifact store from pipeline_artifacts.

_init_session_manager is non-fatal. It skips session management when ORCHESTRATA_KERNEL_SESSION_DISABLED is set to a true value, when the session module is unavailable, or when ORCHESTRATA_KERNEL_SESSION_ROOT is missing from the environment mapping given to boot; it never derives a root from the package location (the docstring cites the RUL-011 incident). Otherwise it builds a SessionController at that root.

_human_handler is the postman handler registered for the HUMAN alias. It appends the packet to the bounded human inbox, collects artifacts, and acts on terminal subjects: qa_complete emits a completion event with the test result and ends the session through _end_session_on_qa; compliance_complete and compliance_failed go to _handle_compliance_complete, which emits pipeline.complete with the real pass or fail outcome, logs the coverage and missing requirements on failure, and ends the session. It always returns None because the human does not auto-reply. _session_log, _session_log_transition, _collect_artifacts and _collect_decisions record work in the session and pass artifacts and decisions on, and _fire_session_audit writes audit records. _checkpoint_stage saves each stage's output through the checkpoint store and swallows failures, since resume is an optimisation (see pipeline_checkpoint). shutdown and ashutdown emit a failed completion event, stop the audit background task and end the session; shutdown is safe to call repeatedly.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

PipelineSessionMixin

Line 80

Session lifecycle helpers for PipelineRunner.

Source

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

"""PipelineRunner session management mixin.

Extracted from pipeline.py to keep PipelineRunner under the
300-line class limit. Contains session lifecycle helpers.
"""

import json
import logging
import os
import threading
from collections import deque
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket
from orchestrata_ultimate.kernel.audit_bootstrap import drain_audit_tasks
from orchestrata_ultimate.kernel.pipeline_artifacts import ArtifactStore
from orchestrata_ultimate.kernel.pipeline_events import PipelineEventBus

if TYPE_CHECKING:
    from orchestrata_ultimate.session.controller import SessionController

_logger = logging.getLogger(__name__)


# Map a stage-transition packet subject to the artifacts it carries:
#   subject -> list of (payload_key, artifact_type, format)
# The producing stage is taken from packet.from_agent so the artifact is
# attributed to the volk that emitted it.
_ARTIFACT_MAP: dict[str, list[tuple[str, str, str]]] = {
    "review_and_test": [
        ("task_summary", "task_summary", "text"),
        ("user_stories", "user_stories", "json"),
        ("structured_requirements", "requirements_analysis", "json"),
    ],
    "implement": [
        ("architecture", "architecture", "mermaid"),
        ("file_tree", "file_tree", "text"),
        ("api_spec", "api_spec", "text"),
        ("requirements", "requirements", "text"),
    ],
    # qa/volk.py emits the LLM review body under ``review`` (only when issues
    # were found) — without this mapping the review work product was
    # discarded at the kernel engine too (twin of the orchestrator
    # extractor's phantom ``review_report`` key defect).
    "compliance_check": [
        ("review", "qa_report", "json"),
        ("test_output", "qa_report", "text"),
    ],
    "qa_complete": [
        ("review", "qa_report", "json"),
        ("test_output", "qa_report", "text"),
    ],
    "compliance_complete": [
        ("report", "compliance_report", "text"),
    ],
    "compliance_failed": [
        ("report", "compliance_report", "text"),
    ],
}

# Session management is provided by the in-project session package. Imported
# defensively (never fatal to kernel boot) in case that package is ever split
# out again, as it was in the pre-port monorepo. The runtime import is
# guarded by TYPE_CHECKING above so mypy sees exactly one binding of the
# name `SessionController` (the real class) instead of complaining about
# a redefinition against the `= None` fallback below.
_SESSION_AVAILABLE: bool
if not TYPE_CHECKING:
    try:
        from orchestrata_ultimate.session.controller import SessionController

        _SESSION_AVAILABLE = True
    except ImportError:
        SessionController: type[Any] | None = None
        _SESSION_AVAILABLE = False


class PipelineSessionMixin:
    """Session lifecycle helpers for PipelineRunner.

    Expects ``self.session`` to be set (or None) by the host class.
    Also expects ``self.events`` (PipelineEventBus) and
    ``self.human_inbox`` (a bounded ``deque[CXPPacket]``, D-H4) from the
    host class.
    """

    session: Any | None
    events: PipelineEventBus
    human_inbox: deque[CXPPacket]
    _human_inbox_lock: threading.Lock
    artifacts: ArtifactStore
    checkpoint: Any

    def _init_session_manager(self, env: Mapping[str, str] | None = None) -> None:
        """Initialize session management from orchestrata_ultimate.session.

        Creates the sessions directory and a SessionController instance.
        Non-fatal — self.session is left as None if unavailable.

        ``SessionController.end()`` unconditionally stages/commits/pushes
        the whole working tree at its root and creates+pushes a git tag
        (see ``session/controller/_git.py``). Deriving that root from
        ``__file__`` (this package's own install location) is exactly the
        defect fixed by fix/session-root (RUL-011, 2026-07-27 incident: 16
        fixture commits landed on this repo's live master, one pushed to
        GitLab) — this method NEVER does that. The root now comes only from
        ``ORCHESTRATA_KERNEL_SESSION_ROOT`` in *env* (an explicit,
        caller-supplied injection, e.g. via ``PipelineRunner.boot(env=...)``);
        absent that, session management is skipped exactly like the
        already-established "module not available" and
        ``ORCHESTRATA_KERNEL_SESSION_DISABLED`` paths below — a caller that
        cannot say where to keep session state must not guess. Automated
        callers (benchmarks, dogfood/E2E runners, tests) also keep the
        pre-existing ``ORCHESTRATA_KERNEL_SESSION_DISABLED=1`` opt-out (read
        at boot time, not import time; the whole test suite sets it in
        tests/conftest.py). ``env`` is the injection seam — defaults to
        ``os.environ``.
        """
        env = os.environ if env is None else env
        if env.get("ORCHESTRATA_KERNEL_SESSION_DISABLED", "").lower() in (
            "1",
            "true",
            "yes",
        ):
            print("  [Session] Disabled via ORCHESTRATA_KERNEL_SESSION_DISABLED")
            return
        # Checking the class (not just the _SESSION_AVAILABLE flag) also lets
        # the type checker narrow SessionController to non-None below.
        if not _SESSION_AVAILABLE or SessionController is None:
            print("  [Session] Module not available — skipping")
            return
        sessions_dir = env.get("ORCHESTRATA_KERNEL_SESSION_ROOT")
        if not sessions_dir:
            print(
                "  [Session] No ORCHESTRATA_KERNEL_SESSION_ROOT configured "
                "— skipping"
            )
            return
        # _SESSION_AVAILABLE is True only when the optional import above
        # succeeded, so SessionController is the real class (never None) here.
        assert SessionController is not None
        try:
            os.makedirs(sessions_dir, exist_ok=True)
            controller = SessionController(
                root=sessions_dir,
                ai_model="Spike Kernel",
                explicit_root=True,
            )
            self.session = controller
            if not controller.ctx.file_exists():
                controller.ctx.initialize()
                print(f"  [Session] Created context files: {sessions_dir}")
            print(f"  [Session] Initialized: {sessions_dir}")
            self._fire_session_audit(
                "start", "", context={"sessions_dir": sessions_dir}
            )
        except Exception as exc:
            print(f"  [Session] Init failed: {exc}")

    def _fire_session_audit(
        self, event: str, thread_id: str, context: dict[str, Any] | None = None
    ) -> None:
        """Fire a REQ-008-004 session-lifecycle audit entry, best-effort.

        Reads ``self.audit``/``self._audit_fire`` via getattr (this mixin
        does not own that state — PipelineAuditMixin does; the two are
        composed together on PipelineRunner) so this file stays decoupled
        from the audit mixin's presence, matching this module's existing
        ``getattr(self, "artifacts", None)``-style cross-mixin access.
        """
        audit = getattr(self, "audit", None)
        fire = getattr(self, "_audit_fire", None)
        if audit is None or not callable(fire):
            return
        fire(
            audit.log_session_lifecycle(
                session_id=thread_id or "kernel-boot",
                volk_id="kernel",
                event=event,
                context=context or {},
            )
        )

    def _session_log(self, desc: str, details: dict | None = None) -> None:
        """Log a work item to the active session.

        Args:
            desc: Short description of the work item.
            details: Optional extra detail dict. The session controller's
                ``log_work`` stores details as ``list[str]``, so the dict is
                flattened to ``"key: value"`` lines before being passed on.
        """
        if not self.session:
            return
        try:
            detail_lines = (
                [f"{k}: {v}" for k, v in details.items()] if details else None
            )
            self.session.log_work(desc=desc, details=detail_lines)
        except Exception as exc:
            _logger.debug("session log_work failed: %s", exc)

    def _end_session_on_qa(self, packet: CXPPacket) -> None:
        """End the active session after a qa_complete packet.

        Args:
            packet: The qa_complete CXP packet.
        """
        if not self.session:
            return
        test_passed = packet.payload.get("test_passed", False)
        test_files = packet.payload.get("test_files", [])
        review_cycles = packet.payload.get("review_cycles", 0)
        achievements = [
            f"QA {'PASS' if test_passed else 'FAIL'}",
            f"{len(test_files)} test files generated",
            f"{review_cycles} review cycle(s)",
        ]
        for filepath in test_files:
            try:
                self.session.log_artifact(filepath=filepath)
            except Exception as exc:
                _logger.debug("session log_artifact: %s", exc)
        try:
            self.session.end(achievements=achievements)
            print("  [Session] Ended", flush=True)
        except Exception as exc:
            print(f"  [Session] End failed: {exc}", flush=True)
        self._fire_session_audit(
            "end",
            packet.thread_id,
            context={"reason": "qa_complete", "test_passed": test_passed},
        )

    def _session_log_transition(self, packet: CXPPacket) -> None:
        """Log rich session data from key CXP packet transitions.

        Records design feedback, implementation artifacts, and fix
        cycles to the active session.

        Args:
            packet: The CXP packet representing the transition.
        """
        if not self.session:
            return
        try:
            subj = packet.subject
            payload = packet.payload

            if subj == "design_feedback":
                verdict = payload.get("verdict", "")
                cycle = payload.get("design_cycle", 0)
                self.session.log_choice(
                    choice=f"Design review: {verdict}",
                    reason=f"Cycle {cycle}",
                )

            elif subj == "implementation_complete":
                files = payload.get("files_created", [])
                for filepath in files:
                    self.session.log_artifact(filepath=filepath)

            elif subj == "fix_issues":
                issues = payload.get("issues", [])
                self.session.log_choice(
                    choice="Fix cycle triggered",
                    reason=f"{len(issues)} issues",
                )
        except Exception as exc:
            _logger.debug("session log_transition failed: %s", exc)

    @traces(spec="001", story="US-006", reqs=["REQ-001-US-006-018"])
    def _collect_artifacts(self, packet: CXPPacket) -> None:
        """Record any artifacts carried by a stage-transition packet.

        Driven by ``_ARTIFACT_MAP``: for the packet's subject, each mapped
        payload key with a non-empty value is added to ``self.artifacts``,
        attributed to ``packet.from_agent`` (the producing stage). Also
        checkpoints the stage output so a resumed run can reuse it. Best-effort
        and non-fatal — the pipeline must never crash on artifact capture.

        F8 fix (kernel.pipeline_escalation.approve/may_complete): this is
        now the ONLY place ``task_has_output`` is asserted for the
        Prolog completion gate — previously ``approve()`` asserted it for
        itself immediately before checking ``may_complete``, making the
        acceptance-criteria/output-verified gates self-fulfilling (they
        could never fail). Asserting it here, keyed to the SAME
        ``f"task_{thread_id}"`` id ``approve()`` queries, ties it to real
        pipeline output instead: a volk whose packet carried no mapped
        artifact content never marks its task's output produced, and
        ``may_complete`` genuinely denies until it does.

        Args:
            packet: The stage-transition CXP packet.
        """
        store = getattr(self, "artifacts", None)
        if store is None:
            return
        try:
            mappings = _ARTIFACT_MAP.get(packet.subject)
            if not mappings:
                return
            payload = packet.payload or {}
            if not isinstance(payload, dict):
                return
            stage = packet.from_agent or "unknown"
            new_artifacts = []
            for key, art_type, fmt in mappings:
                value = payload.get(key)
                if not value:
                    continue
                content = (
                    value if isinstance(value, str) else json.dumps(value, default=str)
                )
                new_artifacts.append(
                    store.add_content(
                        stage=stage,
                        type=art_type,
                        content=content,
                        format=fmt,
                        thread_id=packet.thread_id,
                    )
                )
            # Write artifact provenance ((:Artifact)-[:GENERATED_BY]->(:Volk)) so
            # the semantic layer (QueryEngine.volk_work_context) is populated as
            # stages complete. Best-effort — never break artifact capture.
            sync_artifacts = getattr(self, "sync_artifacts_to_kg", None)
            if new_artifacts and callable(sync_artifacts):
                try:
                    sync_artifacts(new_artifacts)
                except Exception as exc:  # provenance is auxiliary
                    _logger.debug("artifact KG provenance failed: %s", exc)
            if new_artifacts:
                prolog = getattr(self, "prolog", None)
                if prolog is not None:
                    try:
                        prolog.assert_task_output(stage, f"task_{packet.thread_id}")
                    except Exception as exc:  # F8: never break artifact capture
                        _logger.debug("assert_task_output failed: %s", exc)
            self._checkpoint_stage(packet)
        except Exception as exc:
            _logger.debug("artifact collect failed: %s", exc)

    @traces(spec="008", story="US-005", reqs=["REQ-008-006"])
    def _collect_decisions(self, packet: CXPPacket) -> None:
        """Persist any decisions a stage packet carries to the semantic graph.

        A volk records a decision by putting a ``decisions`` list
        (``[{rationale, files, requirements}]``) in its packet payload; this
        writes each as ``(:Volk)-[:MADE]->(:Decision)-[:APPLIES_TO]->(:File)`` via
        ``sync_decisions_to_kg``, attributed to ``packet.from_agent``. Best-effort
        and non-fatal — decision capture must never break the pipeline.

        Args:
            packet: The stage-transition CXP packet.
        """
        payload = packet.payload if isinstance(packet.payload, dict) else {}
        decisions = payload.get("decisions")
        sync_decisions = getattr(self, "sync_decisions_to_kg", None)
        if not decisions or not callable(sync_decisions):
            return
        try:
            sync_decisions(packet.from_agent or "unknown", decisions)
        except Exception as exc:  # decision capture is auxiliary
            _logger.debug("decision KG capture failed: %s", exc)

    @traces(spec="001", story="US-006", reqs=["REQ-001-US-006-016"])
    def _checkpoint_stage(self, packet: CXPPacket) -> None:
        """Persist a completed stage's output for resume.

        Best-effort: checkpoint is a resume optimisation, never load-bearing for
        the live run, so any failure is swallowed.

        Args:
            packet: The stage-transition CXP packet whose payload is the output.
        """
        cp = getattr(self, "checkpoint", None)
        if cp is None:
            return
        payload = packet.payload if isinstance(packet.payload, dict) else {}
        try:
            cp.save_stage(
                thread_id=packet.thread_id,
                stage=packet.from_agent or "unknown",
                subject=packet.subject,
                output=payload,
            )
        except Exception as exc:
            _logger.debug("checkpoint save failed: %s", exc)

    def _human_handler(self, packet: CXPPacket) -> CXPPacket | None:
        """Handle packets addressed to HUMAN — store in inbox.

        Terminal pipeline outcomes are surfaced here:
        - ``qa_complete``         → emit_complete(success=test_passed)
        - ``compliance_complete`` → emit_complete(success=True), final stage passed
        - ``compliance_failed``   → emit_complete(success=False), violation surfaced

        Compliance is the final pipeline stage (U:Cp runs after U:Qa); a
        compliance violation is therefore a first-class terminal FAILED outcome
        with its report surfaced, not a silent RESOLVED.

        Args:
            packet: Incoming CXP packet for the human operator.

        Returns:
            Always None (human does not auto-reply).
        """
        with self._human_inbox_lock:
            self.human_inbox.append(packet)
        self._collect_artifacts(packet)
        if packet.subject == "qa_complete":
            test_passed = packet.payload.get("test_passed", False)
            self.events.emit_complete(
                packet.thread_id, success=test_passed, reason="qa_complete"
            )
            self._end_session_on_qa(packet)
        elif packet.subject in ("compliance_complete", "compliance_failed"):
            self._handle_compliance_complete(packet)
        return None

    def _handle_compliance_complete(self, packet: CXPPacket) -> None:
        """Finalise the pipeline on a compliance result.

        Emits ``pipeline.complete`` with the real pass/fail outcome and ends the
        active session. A violation (``passed=False`` / ``compliance_failed``
        subject) is a terminal FAILED outcome — the compliance report is logged
        and carried on the emitted event so it is surfaced, not swallowed.

        Args:
            packet: The ``compliance_complete``/``compliance_failed`` packet.
        """
        payload = packet.payload or {}
        passed = bool(payload.get("passed", packet.subject == "compliance_complete"))
        report = payload.get("report", "")
        reason = "compliance_pass" if passed else "compliance_failed"

        self.events.emit_complete(packet.thread_id, success=passed, reason=reason)
        self._fire_session_audit(
            "end", packet.thread_id, context={"reason": reason, "passed": passed}
        )

        if not passed:
            _logger.warning(
                "Compliance FAILED for thread %s: %s",
                packet.thread_id,
                payload.get("missing_reqs") or payload.get("missing_classes") or "",
            )
            self._session_log(
                f"COMPLIANCE FAILED ({payload.get('coverage_pct', 0)}% RTM)",
                {"report": report} if report else None,
            )

        if not self.session:
            return
        try:
            achievements = [
                f"Compliance {'PASS' if passed else 'FAIL'}",
                f"{payload.get('coverage_pct', 0)}% RTM coverage",
            ]
            self.session.end(achievements=achievements)
            print("  [Session] Ended (compliance)", flush=True)
        except Exception as exc:
            print(f"  [Session] Compliance end failed: {exc}", flush=True)

    @traces(
        spec="001", story="US-004", reqs=["REQ-001-US-004-021", "REQ-001-US-008-020"]
    )
    def shutdown(self, reason: str = "Manual shutdown") -> None:
        """End session on failure/timeout — safe to call multiple times.

        Args:
            reason: Human-readable shutdown reason.
        """
        self.events.emit_complete("", success=False, reason=reason)
        stop_audit = getattr(self, "_stop_audit_background", None)
        if callable(stop_audit):
            stop_audit()
        if not self.session:
            return
        try:
            self.session.log_work(desc=f"SHUTDOWN: {reason}")
            self.session.end(achievements=[f"FAILED: {reason}"])
            print(f"  [Session] Ended (shutdown: {reason})", flush=True)
        except Exception as exc:
            print(f"  [Session] Shutdown end failed: {exc}", flush=True)

    @traces(
        spec="001", story="US-004", reqs=["REQ-001-US-004-021", "REQ-001-US-008-020"]
    )
    async def ashutdown(
        self, reason: str = "Manual shutdown", timeout: float = 30.0
    ) -> int:
        """Shut down, then WAIT for the audit writes shutdown scheduled.

        :meth:`shutdown` is synchronous and its final act — flushing the audit
        logger via ``_stop_audit_background`` — is fire-and-forget. From a
        caller that owns the event loop, returning immediately means the loop
        can close (or the process exit) while that flush is still in flight,
        which drops exactly the shutdown-time audit records a compliance trail
        most needs. Any async caller shutting a runner down should await this
        instead of calling :meth:`shutdown` directly.

        Returns the number of audit tasks drained. Never raises: the drain
        itself only logs (see :func:`drain_audit_tasks`).

        Args:
            reason: Human-readable shutdown reason, forwarded to
                :meth:`shutdown`.
            timeout: Seconds to wait for in-flight audit writes before
                reporting the remainder and returning.
        """
        self.shutdown(reason=reason)
        return await drain_audit_tasks(timeout=timeout)