orchestrata_ultimate.agents.volks.pm.budget

Reference › orchestrata_ultimate › agents › volks › pm › budget

budget implements the budget and progress rules of the Project Manager volk (specification 012, user stories US-003 to US-005). It is used by volk and has no other importers. PMConfig is a frozen dataclass of limits: 20 LLM calls, 30 minutes, 3 fix cycles, and scope thresholds of 10 files and 8 features. BudgetTracker watches three counters (LLM calls, fix cycles, elapsed seconds). readings returns every crossing currently due, and consume returns only fresh crossings, so each warning or exceeded event (BudgetLevel, carried in a BudgetEvent with metric, threshold and current value) is emitted at most once per iteration. is_stalled reports whether an iteration has seen no progress for a timeout, and build_iteration_summary builds the structured summary dictionary from the metrics and the start and end times.

At a glance

Class diagram

Dependency diagram

API

Classes

PMConfig (dataclass)

Decorators: dataclass(frozen=True) · Line 31

Budget + scope configuration (SPEC-012 § Configuration, verbatim).

BudgetLevel (enum)

Bases: StrEnum · Line 41

Consumption-crossing level for one budget metric.

BudgetEvent (dataclass)

Decorators: dataclass(frozen=True) · Line 49

One budget crossing: metric name, level, limit, and current value.

BudgetTracker (dataclass)

Decorators: dataclass · Line 59

Crossing-triggered budget monitor over an iteration's counters.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/agents/volks/pm/budget.py

"""PM budget enforcement + iteration metrics (SPEC-012 US-003/004/005).

Advisory-only by contract (SPEC-012 NFR-002): nothing in this module raises
into the pipeline, and every budget verdict is a *warning* event — budget
exceeded never kills an iteration (SPEC-012 US-004).

Budget events are crossing-triggered, not level-triggered: each metric
emits at most one ``budget_warning`` (the 80% line was crossed) and one
``budget_exceeded`` (the 100% line was crossed) per iteration. When a single
update jumps past both lines, both events fire, warning first — the warning
would have fired had an update landed in between, and skipping it would
hide the crossing.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from enum import StrEnum

from orchestrata_ultimate.core.base.traces import traces

# SPEC-012 US-003: "Emits iteration_stalled if no progress in 5 minutes".
STALL_TIMEOUT_SECONDS = 300

# SPEC-012 US-004: warning at 80% consumption, exceeded at 100%.
WARNING_RATIO = 0.8
EXCEEDED_RATIO = 1.0


@dataclass(frozen=True)
class PMConfig:
    """Budget + scope configuration (SPEC-012 § Configuration, verbatim)."""

    max_llm_calls: int = 20
    max_duration_minutes: int = 30
    max_fix_cycles: int = 3
    scope_file_threshold: int = 10
    scope_feature_threshold: int = 8


class BudgetLevel(StrEnum):
    """Consumption-crossing level for one budget metric."""

    WARNING = "warning"
    EXCEEDED = "exceeded"


@dataclass(frozen=True)
class BudgetEvent:
    """One budget crossing: metric name, level, limit, and current value."""

    metric: str
    level: BudgetLevel
    threshold: float
    current: float


@dataclass
class BudgetTracker:
    """Crossing-triggered budget monitor over an iteration's counters.

    Tracks the three SPEC-012 budgets (LLM calls, duration, fix cycles)
    against a :class:`PMConfig`. ``consume`` returns only crossings that
    have not been emitted yet, so a caller that emits one event per
    returned item can never double-emit the same verdict.
    """

    config: PMConfig
    _emitted: set[tuple[str, BudgetLevel]] = field(default_factory=set)

    @traces(
        spec="012",
        story="US-012-Budget",
        reqs=[
            "REQ-012-053",  # emits budget_warning at 80% consumption
            "REQ-012-054",  # emits budget_exceeded at 100% consumption
        ],
    )
    def readings(
        self, llm_calls: int, fix_cycles: int, elapsed_seconds: float
    ) -> list[BudgetEvent]:
        """All budget crossings currently due, including already-emitted ones.

        A reading crossing the 100% line also crossed the 80% line on the
        way, so both events are returned (warning first).
        """
        minutes = max(elapsed_seconds, 0.0) / 60.0
        budgets: tuple[tuple[str, float, float], ...] = (
            ("llm_calls", float(llm_calls), float(self.config.max_llm_calls)),
            ("fix_cycles", float(fix_cycles), float(self.config.max_fix_cycles)),
            (
                "duration_minutes",
                minutes,
                float(self.config.max_duration_minutes),
            ),
        )
        events: list[BudgetEvent] = []
        for metric, current, limit in budgets:
            if limit <= 0:
                continue
            ratio = current / limit
            if ratio >= WARNING_RATIO:
                events.append(
                    BudgetEvent(
                        metric,
                        BudgetLevel.WARNING,
                        threshold=limit * WARNING_RATIO,
                        current=current,
                    )
                )
            if ratio >= EXCEEDED_RATIO:
                events.append(
                    BudgetEvent(
                        metric,
                        BudgetLevel.EXCEEDED,
                        threshold=limit,
                        current=current,
                    )
                )
        return events

    def consume(
        self, llm_calls: int, fix_cycles: int, elapsed_seconds: float
    ) -> list[BudgetEvent]:
        """Fresh crossings only (each emitted at most once per iteration).

        Calling again with unchanged counters returns an empty list — the
        tracker is edge-triggered, so steady over-budget state does not
        re-emit on every progress packet.
        """
        fresh = [
            event
            for event in self.readings(llm_calls, fix_cycles, elapsed_seconds)
            if (event.metric, event.level) not in self._emitted
        ]
        for event in fresh:
            self._emitted.add((event.metric, event.level))
        return fresh


@traces(
    spec="012",
    story="US-012-Progress",
    reqs=[
        "REQ-012-051",  # monitors CXP packet flow for progress tracking
    ],
)
def is_stalled(
    last_progress_ts: float | None,
    now: float,
    timeout_s: float = STALL_TIMEOUT_SECONDS,
) -> bool:
    """True when the iteration saw no progress for ``timeout_s`` seconds.

    ``last_progress_ts`` of ``None`` means progress was never recorded
    (the iteration has not really started) — not stalled.
    """
    if last_progress_ts is None:
        return False
    return now - last_progress_ts >= timeout_s


@traces(
    spec="012",
    story="US-012-Summary",
    reqs=[
        "REQ-012-055",  # PM produces structured iteration_summary
    ],
)
def build_iteration_summary(metrics: dict, start_time: float, now: float) -> dict:
    """Build the SPEC-012 US-005 structured iteration summary.

    A plain dict (not prose): files generated, tests passed/failed, pass
    rate, LLM calls made, duration, design cycles, fix cycles. Pass rate
    is ``passed / (passed + failed)`` and 0.0 when no tests ran.
    """
    tests_passed = int(metrics.get("tests_passed", 0))
    tests_failed = int(metrics.get("tests_failed", 0))
    tests_total = tests_passed + tests_failed
    pass_rate = round(tests_passed / tests_total, 4) if tests_total else 0.0
    return {
        "files_generated": int(metrics.get("files_generated", 0)),
        "tests_passed": tests_passed,
        "tests_failed": tests_failed,
        "pass_rate": pass_rate,
        "llm_calls": int(metrics.get("llm_calls", 0)),
        "duration_seconds": round(now - start_time, 2) if start_time else 0.0,
        "design_cycles": int(metrics.get("design_cycles", 0)),
        "fix_cycles": int(metrics.get("fix_cycles", 0)),
    }