orchestrata_ultimate.context.providers.mission_context

Reference › orchestrata_ultimate › context › providers › mission_context

This provider renders a small mission block from the session's ContextData, so volks receive project state, the active goal, pending work and the last handoff without the whole context file. The assembler calls it for the mission layer. Tests are in tests/context/providers/test_mission_context.py. render_mission_context carries @traces for REQ-006-FR14-006 and REQ-006-US-005-001.

The block starts with a project line (branch, version, status line, test state), then the active session number, the active goal with up to five unmet criteria, up to five pending work items (formatted by _format_work_item with priority and notes), and handoff sections for decisions, blockers, active work, key files and alerts, five items each. When the line count exceeds max_lines (default 40) lines are dropped from the end, so the handoff goes first. A missing context gives an empty string.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/context/providers/mission_context.py

"""Render a compact mission/session context block from ``ContextData``.

This provider closes Gap D from ``docs/context-injection.md``: volks receive
a small, high-signal block summarising project state, the active goal, pending
work, and the latest handoff. It deliberately avoids dumping the full
``context.md``; the output is sized to fit inside the assembler's ``mission``
budget.
"""

from __future__ import annotations

from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces

if TYPE_CHECKING:
    from orchestrata_ultimate.session.models import ContextData, WorkItem


@traces(spec="006", reqs=["REQ-006-FR14-006", "REQ-006-US-005-001"])
def render_mission_context(ctx: ContextData | None, max_lines: int = 40) -> str:
    """Render a compact mission block for volk prompts.

    Args:
        ctx: Loaded project context. If None, returns an empty string.
        max_lines: Hard cap on the number of lines emitted. The provider
            drops lower-priority sections to stay under this cap.

    Returns:
        Markdown mission block or "" when no context is available.
    """
    if ctx is None:
        return ""

    lines: list[str] = ["# MISSION CONTEXT"]

    state = ctx.state
    lines.append(
        f"Project: branch={state.git_branch} version={state.version} "
        f"status={state.status_line} tests={state.test_state or 'unknown'}"
    )
    if state.active_session is not None:
        lines.append(f"Active session: {state.active_session:03d}")

    goal = state.goal
    if goal is not None:
        lines.append("")
        lines.append(f"## Active goal ({goal.status})")
        lines.append(goal.statement)
        unmet = [c.description for c in (goal.criteria or []) if c.status != "met"]
        if unmet:
            lines.append("Unmet criteria:")
            lines.extend(f"- {c}" for c in unmet[:5])

    pending = ctx.pending_work()
    if pending:
        lines.append("")
        lines.append("## Pending work")
        for item in pending[:5]:
            lines.append(_format_work_item(item))

    latest = ctx.latest_session()
    if latest is not None and latest.handoff is not None:
        handoff = latest.handoff
        sections = [
            ("Decisions", handoff.decisions),
            ("Blockers", handoff.blockers),
            ("Active work", handoff.active_work),
            ("Key files", handoff.key_files),
            ("Alerts", handoff.alerts),
        ]
        for heading, items in sections:
            if items:
                lines.append("")
                lines.append(f"## Handoff: {heading}")
                lines.extend(f"- {item}" for item in items[:5])

    # Hard line cap: drop from the end (handoff is lowest priority) until fit.
    while len(lines) > max_lines:
        lines.pop()

    return "\n".join(lines)


def _format_work_item(item: WorkItem) -> str:
    prefix = f"[{item.priority}] " if item.priority else ""
    notes = f" ({item.notes})" if item.notes else ""
    return f"- {prefix}{item.text}{notes}"