orchestrata_ultimate.context.providers.plan_context

Reference › orchestrata_ultimate › context › providers › plan_context

This provider fills the plan layer of the assembler from the active goal's tasks (SPEC-107, REQ-107-003). No test module imports it directly.

render_plan_context returns an empty string when there is no context, no goal or no tasks. Otherwise it prints the goal statement and a progress count of done tasks. The current task is found by starting at the goal's persisted current_step and scanning forward to the first task that is not done. The block then lists up to three recently completed tasks before it and up to two upcoming tasks after it. Over max_lines (default 25) lines are dropped from the end, in the order upcoming, done, current, progress.

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/plan_context.py

"""Render a compact plan-window context block from ``ContextData``.

This provider closes FR-001 / REQ-107-003 of SPEC-107: the ``plan`` layer of
``UnifiedContextAssembler`` is populated from the active goal's tasks instead
of staying empty. It reports the current task, the most recently completed
tasks, and the next upcoming tasks, all capped to fit the assembler's plan
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, GoalTask


@traces(spec="107", reqs=["REQ-107-003"])
def render_plan_context(ctx: ContextData | None, max_lines: int = 25) -> str:
    """Render a compact plan window for volk prompts.

    Args:
        ctx: Loaded project context. If None, or if the active goal has no
            tasks, returns an empty string.
        max_lines: Hard cap on the number of lines emitted. Lower-priority
            sections (upcoming, then done) are dropped to stay under the cap.

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

    goal = ctx.state.goal
    if goal is None:
        return ""

    tasks = goal.tasks
    if not tasks:
        return ""

    total = len(tasks)
    done = [t for t in tasks if t.status == "done"]

    # Start from the persisted current_step, then scan forward for the first
    # task that is not yet done. This honors explicit step tracking while still
    # coping with manually edited task statuses.
    start_index = max(0, min(goal.current_step, total - 1)) if total else 0
    current: GoalTask | None = None
    current_index = -1
    for i in range(start_index, total):
        if tasks[i].status != "done":
            current = tasks[i]
            current_index = i
            break

    lines: list[str] = ["# PLAN WINDOW"]
    lines.append(f"Goal: {goal.statement}")
    lines.append(f"Progress: {len(done)}/{total} tasks done")

    if current is not None:
        lines.append("")
        lines.append(f"## Current task ({current.status})")
        lines.append(f"- {current.description}")

    # Recent done tasks: last up to 3 completed tasks before the current one.
    recent_done = [t for t in tasks[:current_index] if t.status == "done"][-3:]
    if recent_done:
        lines.append("")
        lines.append("## Recent done")
        for task in recent_done:
            lines.append(f"- {task.description}")

    # Upcoming tasks: non-done tasks after the current one, up to 2.
    upcoming: list[GoalTask] = []
    if current_index >= 0:
        upcoming = [t for t in tasks[current_index + 1 :] if t.status != "done"][:2]
    if upcoming:
        lines.append("")
        lines.append("## Upcoming")
        for task in upcoming:
            lines.append(f"- {task.description}")

    # Hard line cap: drop from the end (upcoming < done < current < progress).
    while len(lines) > max_lines:
        lines.pop()

    # If we've dropped everything but the header, still return the header only
    # when there is real task state to report.
    if len(lines) == 1:
        return ""

    return "\n".join(lines)