orchestrata_ultimate.session.handover

Reference › orchestrata_ultimate › session › handover

HandoverManager handles the handoff from one session to the next. A handoff is a HandoffData record (decisions, blockers, active work, key files, commands, alerts) defined in models. build_handoff builds one from lists, attach_to_session stores it on a numbered session in the ContextFileManager, and get_from_session and get_latest read one back. has_blockers and has_active_work answer quick questions about a session, format_display renders a handoff as text, merge_handoffs combines several into one, and extract_from_session derives a handoff from the data of a session. The controller's set_handoff and the handoff CLI command in cli/lifecycle produce the same data; a session needs one before it can close.

At a glance

Class diagram

Dependency diagram

API

Classes

HandoverManager

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

Manages session-to-session handover.

Source

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

"""Fresh implementation of session handover.

Manages session-to-session handover 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 HandoffData

# Spec: 006 - Session Management


logger = logging.getLogger(__name__)


@traces(spec="006", story="US-005", reqs=["REQ-006-US-005-001"])
class HandoverManager:
    """Manages session-to-session handover."""

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

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

    @traces(spec="006", story="US-005", reqs=["REQ-006-US-005-001"])
    def build_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,
    ) -> HandoffData:
        """Build handoff data.

        Args:
            decisions: Key decisions.
            blockers: Current blockers.
            active_work: Work in progress.
            key_files: Important files.
            cmds: Useful commands.
            alerts: Warnings.

        Returns:
            Handoff data.
        """
        return HandoffData(
            decisions=decisions or [],
            blockers=blockers or [],
            active_work=active_work or [],
            key_files=key_files or [],
            cmds=cmds or [],
            alerts=alerts or [],
        )

    @traces(spec="006", story="US-005", reqs=["REQ-006-US-005-005"])
    def attach_to_session(self, seq: int, handoff: HandoffData) -> bool:
        """Attach handoff to session.

        Args:
            seq: Session number.
            handoff: Handoff data.

        Returns:
            True if successful.
        """
        result = self.ctx.update_session(seq, handoff=handoff)
        return result is not None

    @traces(spec="006", story="US-005", reqs=["REQ-006-US-005-006"])
    def get_from_session(self, seq: int) -> HandoffData | None:
        """Get handoff from session.

        Args:
            seq: Session number.

        Returns:
            Handoff data or None.
        """
        sessions = self.ctx.read_sessions()
        for s in sessions:
            if s.seq == seq:
                return s.handoff
        return None

    @traces(spec="006", story="US-003", reqs=["REQ-006-US-003-001"])
    def get_latest(self) -> HandoffData | None:
        """Get latest session handoff.

        Returns:
            Latest handoff or None.
        """
        latest = self.ctx.read_latest_session()
        if latest:
            return latest.handoff
        return None

    def has_blockers(self, seq: int) -> bool:
        """Check if session has blockers.

        Args:
            seq: Session number.

        Returns:
            True if blockers exist.
        """
        handoff = self.get_from_session(seq)
        return bool(handoff and handoff.blockers)

    def has_active_work(self, seq: int) -> bool:
        """Check if session has active work.

        Args:
            seq: Session number.

        Returns:
            True if active work exists.
        """
        handoff = self.get_from_session(seq)
        return bool(handoff and handoff.active_work)

    def format_display(self, handoff: HandoffData) -> str:
        """Format handoff for display.

        Args:
            handoff: Handoff data.

        Returns:
            Formatted text.
        """
        sections = []

        if handoff.decisions:
            sections.append("**Key Decisions:**")
            sections.extend(f"- {d}" for d in handoff.decisions)
            sections.append("")

        if handoff.blockers:
            sections.append("**Blockers:**")
            sections.extend(f"- {b}" for b in handoff.blockers)
            sections.append("")

        if handoff.active_work:
            sections.append("**Active Work:**")
            sections.extend(f"- {w}" for w in handoff.active_work)
            sections.append("")

        if handoff.key_files:
            sections.append("**Key Files:**")
            sections.extend(f"- {f}" for f in handoff.key_files)
            sections.append("")

        if handoff.cmds:
            sections.append("**Useful Commands:**")
            sections.extend(f"- {c}" for c in handoff.cmds)
            sections.append("")

        if handoff.alerts:
            sections.append("**Alerts:**")
            sections.extend(f"- {a}" for a in handoff.alerts)
            sections.append("")

        return "\n".join(sections)

    def merge_handoffs(self, *handoffs: HandoffData) -> HandoffData:
        """Merge multiple handoffs.

        Args:
            *handoffs: Handoffs to merge.

        Returns:
            Merged handoff.
        """
        merged = HandoffData()

        for h in handoffs:
            merged.decisions.extend(h.decisions)
            merged.blockers.extend(h.blockers)
            merged.active_work.extend(h.active_work)
            merged.key_files.extend(h.key_files)
            merged.cmds.extend(h.cmds)
            merged.alerts.extend(h.alerts)

        merged.decisions = list(dict.fromkeys(merged.decisions))
        merged.blockers = list(dict.fromkeys(merged.blockers))
        merged.active_work = list(dict.fromkeys(merged.active_work))
        merged.key_files = list(dict.fromkeys(merged.key_files))
        merged.cmds = list(dict.fromkeys(merged.cmds))
        merged.alerts = list(dict.fromkeys(merged.alerts))

        return merged

    @traces(spec="006", story="US-005", reqs=["REQ-006-US-005-006"])
    def extract_from_session(self, seq: int) -> HandoffData:
        """Extract handoff from session data.

        Args:
            seq: Session number.

        Returns:
            Extracted handoff.
        """
        sessions = self.ctx.read_sessions()
        for s in sessions:
            if s.seq == seq:
                return HandoffData(
                    active_work=s.upcoming,
                    decisions=[c.choice for c in s.choices],
                )
        return HandoffData()