orchestrata_ultimate.agents.volks.pm.volk

Reference › orchestrata_ultimate › agents › volks › pm › volk

PMVolk (alias U:PM, role Project Manager) is the scope gate of the pipeline (specification 012). It is advisory: it warns and estimates but does not block. It extends BaseAgent (base), is imported by the kernel pipeline, and ProjectManagerAgent is an alias subclass. Its budget logic is in budget, and goal tracking in goal_registry and goal_report.

What it receives: a scope_review packet from the Analyst (agent) carrying the requirements, user stories, project type and technology. What it produces: one forwarded packet carrying the requirements, task summary, project name, Prolog facts, a has_ui flag and the parsed scope data, plus the Analyst's structured fields passed through unchanged.

_handle_scope_review calls the LLM (_call_scope_llm) and parses the reply with _parse_scope_status (the verdict, defaulting to validated), _parse_estimate (a JSON block, tolerant of malformed JSON), _parse_warnings and _parse_task_summary; _apply_scope_threshold_warnings adds warnings when PMConfig thresholds are crossed. _detect_has_ui matches UI keywords as whole words. _build_forward_packet then routes: to U:Ux with subject ux_design when a UI is present, otherwise straight to U:Ar with subject implement.

Progress is reported as events: _emit_stage_progress for stage transitions, _emit_budget_events for fresh budget crossings, _emit_stall_if_due when an iteration idles for five minutes, all through _emit_guarded, which delivers to local observers and never raises on the CXP path. _handle_progress updates internal metrics from progress events, _record_goal_progress records an iteration against the active goal and regenerates its report, and finalize_iteration emits and persists the structured iteration summary. The constructor takes an optional session manager and a PMConfig.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

PMVolk (pydantic)

Bases: BaseAgent · Decorators: traces(spec='012', story='US-012-Volk', reqs=['REQ-012-001']) · Line 181

Project Manager volk — scope validation and effort estimation.

ProjectManagerAgent (pydantic)

Bases: PMVolk · Line 592

Agent alias for PMVolk.

Functions

Source

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

# mypy: ignore-errors
"""U:PM — Project Manager volk for scope validation and effort estimation."""

import json
import logging
import re
import time
from pathlib import Path

from pydantic import Field

from orchestrata_ultimate.agents.volks.base import BaseAgent
from orchestrata_ultimate.agents.volks.pm.budget import (
    BudgetLevel,
    BudgetTracker,
    PMConfig,
    build_iteration_summary,
    is_stalled,
)
from orchestrata_ultimate.agents.volks.pm.goal_registry import GoalRegistry
from orchestrata_ultimate.agents.volks.pm.goal_report import GoalReportGenerator
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph_directives import extract_graph_context
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus
from orchestrata_ultimate.core.protocols.session import SessionManagerProtocol

logger = logging.getLogger(__name__)

# REQ-012-006: PM_SYSTEM_PROMPT instructs LLM to count FEATURES, DATA ENTITIES, INTERFACES, DEPLOYMENT
# REQ-012-007: PM_SYSTEM_PROMPT specifies WARNING thresholds: >10 files, >8 features, >5 integrations, >20 LLM calls
PM_SYSTEM_PROMPT = """\
You are U:PM, the Project Manager — scope validation and effort estimation \
for AI-generated software projects. Advisory only — warn but never block.

SCOPE VALIDATION: Count FEATURES (functional capabilities), DATA ENTITIES \
(models/schemas), INTERFACES (CLI/API/UI), DEPLOYMENT concerns (services).

ESTIMATION HEURISTICS: Each feature ~2 src + 1 test file. Each entity ~1 \
model + 1 test. Each interface ~1-2 files. LLM calls ~N+5 (N=file count). \
Duration ~3 min per source file.

THRESHOLDS: >10 files, >8 features, >5 integrations, >20 LLM calls = WARNING.

HANDOFF: Route the validated scope to U:Ux (ux_design) when the project has a UI, otherwise to U:Ar (implement).

Output EXACTLY this format (JSON on ONE LINE, no markdown tables):
SCOPE_RESULT: VALIDATED or WARNING
ESTIMATE: {"files": N, "tests": N, "llm_calls": N, "duration_minutes": N, "features": N, "entities": N, "interfaces": N}
WARNINGS: ["warning1", "warning2"] or []
TASK_SUMMARY: <one-line summary>

CRITICAL: ESTIMATE must be a single-line JSON object. NOT a table. NOT multi-line.
"""

_UI_KEYWORDS = frozenset(
    [
        "web app",
        "dashboard",
        "frontend",
        "ui",
        "user interface",
        "react",
        "vue",
        "htmx",
        "html",
        "css",
        "browser",
        "responsive",
    ]
)


_UI_PATTERNS: tuple[re.Pattern[str], ...] = tuple(
    re.compile(rf"\b{re.escape(keyword)}\b")
    for keyword in sorted(_UI_KEYWORDS, key=len, reverse=True)
)


@traces(
    spec="012",
    story="US-012-UI",
    reqs=[
        "REQ-012-016",  # _detect_has_ui scans for UI keywords
        "REQ-012-017",  # Keyword matching is case-insensitive
        "REQ-012-018",  # Returns True if keyword matches
    ],
)
def _detect_has_ui(text: str) -> bool:
    """Detect whether requirements indicate a UI project.

    Whole-word matching: a keyword must appear as its own word so a short
    token like "ui" cannot false-positive inside "build" or "css" inside
    "scss".
    """
    lower = text.lower()
    return any(pattern.search(lower) for pattern in _UI_PATTERNS)


@traces(
    spec="012",
    story="US-012-Scope",
    reqs=[
        "REQ-012-009",  # LLM response includes SCOPE_RESULT token
    ],
)
def _parse_scope_status(response: str) -> str:
    """Extract the SCOPE_RESULT verdict, defaulting to VALIDATED."""
    if "SCOPE_RESULT:" not in response:
        return "VALIDATED"
    return response.split("SCOPE_RESULT:")[1].split("\n")[0].strip()


@traces(spec="012", story="US-012-Scope", reqs=["REQ-012-010"])
def _parse_estimate(response: str) -> dict:
    """Extract the ESTIMATE JSON block, tolerating multi-line and malformed JSON."""
    if "ESTIMATE:" not in response:
        return {}
    # G217: Handle multi-line JSON — extract until next marker or end
    est_block = response.split("ESTIMATE:")[1]
    for marker in ["WARNINGS:", "TASK_SUMMARY:", "ESCALATE_TO:"]:
        if marker in est_block:
            est_block = est_block.split(marker)[0]
    est_block = est_block.strip()
    try:
        return json.loads(est_block)
    except json.JSONDecodeError, ValueError:
        # Try first line only as fallback
        est_line = est_block.split("\n")[0].strip()
        try:
            return json.loads(est_line)
        except json.JSONDecodeError, ValueError:
            return {"raw": est_line}


@traces(spec="012", story="US-012-Scope", reqs=["REQ-012-011"])
def _parse_warnings(response: str) -> list:
    """Extract the WARNINGS list, falling back to a single-item list on bad JSON."""
    if "WARNINGS:" not in response:
        return []
    warn_line = response.split("WARNINGS:")[1].split("\n")[0].strip()
    try:
        return json.loads(warn_line)
    except json.JSONDecodeError, ValueError:
        if warn_line and warn_line != "[]":
            return [warn_line]
        return []


@traces(spec="012", story="US-012-Scope", reqs=["REQ-012-012"])
def _parse_task_summary(response: str) -> str:
    """Extract the TASK_SUMMARY block up to the next major section."""
    if "TASK_SUMMARY:" not in response:
        return ""
    # REQ-012-012: Parse TASK_SUMMARY block
    summary_block = response.split("TASK_SUMMARY:")[1]
    # Extract until next major section or end
    for marker in ["SCOPE_RESULT:", "ESTIMATE:", "WARNINGS:"]:
        if marker in summary_block:
            summary_block = summary_block.split(marker)[0]
    return summary_block.strip()


@traces(spec="012", story="US-012-Scope", reqs=["REQ-012-014"])
def _parse_scope_result(response: str) -> dict:
    """Extract structured scope data from LLM response."""
    return {
        "scope_status": _parse_scope_status(response),
        "estimate": _parse_estimate(response),
        "warnings": _parse_warnings(response),
        "task_summary": _parse_task_summary(response),
    }


@traces(
    spec="012",
    story="US-012-Volk",
    reqs=[
        "REQ-012-001",  # PMVolk extends BaseAgent with alias U:PM
    ],
)
class PMVolk(BaseAgent):
    """Project Manager volk — scope validation and effort estimation."""

    alias: str = Field(default="U:PM")
    role: str = "Project Manager"

    def __init__(
        self,
        context_manager: SessionManagerProtocol | None = None,
        config: PMConfig | None = None,
        **kwargs,
    ):
        """Initialize the PM volk.

        Args:
            context_manager: Optional session context store (SPEC-023
                canonical store, structurally a session-layer
                ``ContextFileManager``) that a ``GoalRegistry`` reads/writes
                portfolio goals through (SPEC-012 goal lifecycle,
                REQ-012-026..REQ-012-041). Injected, never constructed here:
                the concrete lives at layer 4 (session) and the agents layer
                takes no import on it — callers above the seam (kernel, CLI,
                tests) build it and pass it in. None (default) keeps goal
                tracking inert — advisory, opt-in, mirrors the kernel
                dispatcher's staged D3 wiring (construct only when
                configured).
            config: Budget/scope configuration (SPEC-012 US-004 §
                Configuration). None uses the spec defaults. Budget
                enforcement is advisory-only: crossings emit events, never
                block (SPEC-012 NFR-002).
        """
        super().__init__(**kwargs)
        self.config = config or PMConfig()
        self._budget = BudgetTracker(self.config)
        self._last_progress_ts: float | None = None
        self._stall_emitted = False
        self._project_dir: str | None = None
        self.escalation_target = "U:Ar"
        self.metrics = {
            "llm_calls": 0,
            "design_cycles": 0,
            "fix_cycles": 0,
            "start_time": 0.0,
            "files_generated": 0,
            "tests_run": 0,
            "tests_passed": 0,
            "tests_failed": 0,
        }
        self.scope_data = {}
        self.task_summary = ""
        # SPEC-012 goal lifecycle (REQ-012-026, REQ-012-040, REQ-012-041):
        # GoalRegistry is only constructed when a context store was injected
        # (see the __init__ docstring above); it persists through that store
        # — composition, not ad-hoc file I/O inside GoalRegistry.
        # GoalReportGenerator is pure/stateless, so it is always available.
        self._context_manager = context_manager
        self._goal_registry = (
            GoalRegistry(self._context_manager) if self._context_manager else None
        )
        self._report_generator = GoalReportGenerator()
        self.latest_goal_report = None

    @traces(
        spec="012",
        story="US-012-Routing",
        reqs=[
            "REQ-012-003",  # process() accepts CXPPacket
            "REQ-012-004",  # Handles subjects: implement, scope_review, progress_update
            "REQ-012-005",  # Unrecognised subjects return error without exception
        ],
    )
    async def process(self, packet: CXPPacket) -> CXPPacket | None:
        """Route packet to appropriate handler."""
        self._graph_context = extract_graph_context(packet.payload)
        if packet.subject == "implement":
            return await self._handle_scope_review(packet)
        if packet.subject == "scope_review":
            return await self._handle_scope_review(packet)
        if packet.subject == "progress_update":
            return await self._handle_progress(packet)
        # REQ-012-005: Unrecognised subjects return None without raising
        return None

    @traces(
        spec="012",
        story="US-012-Scope",
        reqs=[
            "REQ-012-008",  # _handle_scope_review calls LLM with PM_SYSTEM_PROMPT
            "REQ-012-013",  # Scope validation operates in advisory mode
            "REQ-012-020",  # _detect_has_ui called once per cycle, not recomputed
        ],
    )
    async def _handle_scope_review(self, packet: CXPPacket) -> CXPPacket:
        """Validate scope and estimate effort, then forward."""
        self.status = WorkStatus.IN_PROGRESS
        self.current_thread = packet.thread_id
        self.metrics["start_time"] = time.time()
        self.summary = "Validating scope and estimating effort"

        self._project_dir = packet.payload.get("project_dir") or self._project_dir
        requirements = packet.payload.get("requirements", "")
        self.task_summary = packet.payload.get("task_summary", "Implement as specified")
        original_intent = packet.payload.get("original_intent", "")
        has_ui = packet.payload.get(
            "has_ui", _detect_has_ui(requirements + " " + original_intent)
        )
        print(f"  [U:PM] Reviewing scope for: {self.task_summary}", flush=True)

        response = await self._call_scope_llm(requirements)
        self.scope_data = _parse_scope_result(response)
        self._apply_scope_threshold_warnings()
        self._log_scope_result(
            self.scope_data["scope_status"],
            self.scope_data.get("warnings", []),
        )
        self.status = WorkStatus.RESOLVED
        self.summary = f"Scope {self.scope_data['scope_status']} — forwarding"
        # US-003: progress_update at the PM stage transition, then budget
        # checks (the scope LLM call consumes budget). Advisory-only.
        await self._emit_stage_progress("scope_review", self.scope_data["scope_status"])
        await self._emit_budget_events()
        return self._build_forward_packet(packet, has_ui, requirements, original_intent)

    @traces(
        spec="012",
        story="US-012-Scope",
        reqs=[
            "REQ-012-008",  # Calls LLM with PM_SYSTEM_PROMPT
        ],
    )
    async def _call_scope_llm(self, requirements: str) -> str:
        """Call LLM for scope validation and record result."""
        prompt = (
            f"Requirements:\n{requirements}\n\n"
            f"Task: {self.task_summary}\n\n"
            "Validate scope and estimate effort."
        )
        response = await self._call_llm(
            PM_SYSTEM_PROMPT, prompt, temperature=0.3, graph_context=self._graph_context
        )
        self.metrics["llm_calls"] += 1
        self.output_artifacts.append(f"=== SCOPE REVIEW ===\n{response}")
        return response

    def _log_scope_result(self, scope_status: str, warnings: list) -> None:
        """Print scope validation results."""
        print(f"  [U:PM] Scope: {scope_status}", flush=True)
        if warnings:
            for w in warnings:
                print(f"  [U:PM] WARNING: {w}", flush=True)
        estimate = self.scope_data.get("estimate", {})
        if estimate:
            print(f"  [U:PM] Estimate: {estimate}", flush=True)

    @traces(
        spec="012",
        story="US-012-Routing",
        reqs=[
            "REQ-012-014",  # Parsed SCOPE_RESULT/ESTIMATE/WARNINGS attached to payload
            "REQ-012-019",  # has_ui boolean stored in outbound packet
            "REQ-012-021",  # Routes to U:Ux when has_ui is True
            "REQ-012-022",  # Routes to U:Ar when has_ui is False
            "REQ-012-023",  # Preserve thread_id, correlation_id, context
            "REQ-012-024",  # Forward ESTIMATE JSON verbatim
            "REQ-012-025",  # Log routing destination
        ],
    )
    def _build_forward_packet(
        self,
        packet: CXPPacket,
        has_ui: bool,
        requirements: str,
        original_intent: str,
    ) -> CXPPacket:
        """Build the CXP packet to forward to UX or Architect."""
        target = "U:Ux" if has_ui else "U:Ar"
        subject = "ux_design" if has_ui else "implement"

        print(f"  [U:PM] Forwarding to {target}", flush=True)
        forward_payload = {
            "requirements": requirements,
            "original_intent": original_intent,
            "task_summary": self.task_summary,
            "project_name": packet.payload.get("project_name"),
            "prolog_facts": packet.payload.get("prolog_facts", []),
            "has_ui": has_ui,
            "scope_data": self.scope_data,
        }
        # Forward BA enrichments if present
        for key in (
            "user_stories",
            "structured_requirements",
            "nfrs",
            "project_type",
            "technology",
            "spec_metadata",
            "analyst_output",
        ):
            if key in packet.payload:
                forward_payload[key] = packet.payload[key]
        return CXPPacket(
            from_agent="U:PM",
            to_agent=target,
            thread_id=packet.thread_id,
            subject=subject,
            payload=forward_payload,
            work_status=WorkStatus.PENDING,
        )

    @traces(spec="012", story="US-012-Routing", reqs=["REQ-012-004"])
    async def _handle_progress(self, packet: CXPPacket) -> None:
        """Update internal metrics from progress events.

        Also records the iteration against the active goal and refreshes
        its completion report when a GoalRegistry is configured (SPEC-012
        goal lifecycle wiring). Advisory only: any failure here (no
        registry configured, no active goal, unreadable context.json) is
        swallowed so goal tracking can never affect the metrics-update
        contract this method already had.

        US-003/004/005 additions (all advisory, SPEC-012 NFR-002): the
        stall check runs BEFORE the merge so a progress packet that
        arrives after the 5-minute idle window reports the stall of the
        window it just closed, then records the fresh progress; budget
        crossings are emitted after the merge; an
        ``iteration_complete`` payload ends the iteration with a
        structured summary.
        """
        payload = packet.payload
        await self._emit_stall_if_due()
        for key in self.metrics:
            if key in payload:
                self.metrics[key] = payload[key]
        self._last_progress_ts = time.time()
        self._stall_emitted = False
        self._record_goal_progress(payload)
        if payload.get("iteration_complete"):
            await self.finalize_iteration(
                project_dir=payload.get("project_dir") or self._project_dir
            )
        else:
            await self._emit_budget_events()
        return None

    @traces(spec="012", story="US-012-Reporting", reqs=["REQ-012-041"])
    def _record_goal_progress(self, result: dict) -> None:
        """Record an iteration against the active goal and regenerate its report.

        No-op when no GoalRegistry is configured (no context store
        injected) or when no goal is currently ACTIVE/RUNNING. Never raises — goal
        tracking is advisory, matching this volk's "warn but never block"
        contract.

        Args:
            result: The progress_update payload, forwarded verbatim as the
                iteration result to GoalRegistry.record_iteration and
                GoalReportGenerator.generate.
        """
        if self._goal_registry is None:
            return
        try:
            goal = self._goal_registry.get_active_goal()
            if goal is None:
                return
            updated = self._goal_registry.record_iteration(goal.goal_id, result=result)
            self.latest_goal_report = self._report_generator.generate(
                goal_id=updated.goal_id,
                description=updated.description,
                criteria=updated.success_criteria.model_dump(),
                result=result,
                iteration_number=updated.iterations_spent,
            )
        except Exception as exc:
            print(f"  [U:PM] goal progress tracking skipped: {exc}", flush=True)

    @traces(spec="012", story="US-012-Scope", reqs=["REQ-012-056"])
    def _apply_scope_threshold_warnings(self) -> None:
        """Append PMConfig scope-threshold warnings to the parsed warnings.

        The LLM prompt carries its own threshold guidance; this is the
        deterministic backstop (SPEC-012 § Configuration:
        ``scope_file_threshold``/``scope_feature_threshold``) so an
        estimate that exceeds the configured limits warns even when the
        LLM response omitted the warning.
        """
        estimate = self.scope_data.get("estimate", {})
        warnings = self.scope_data.setdefault("warnings", [])
        files = estimate.get("files")
        features = estimate.get("features")
        if isinstance(files, (int, float)) and files > self.config.scope_file_threshold:
            warnings.append(
                f"Estimated {int(files)} files exceed threshold "
                f"{self.config.scope_file_threshold}"
            )
        if (
            isinstance(features, (int, float))
            and features > self.config.scope_feature_threshold
        ):
            warnings.append(
                f"Estimated {int(features)} features exceed threshold "
                f"{self.config.scope_feature_threshold}"
            )

    @traces(spec="012", story="US-012-Progress", reqs=["REQ-012-052"])
    async def _emit_stage_progress(self, stage: str, status: str) -> None:
        """Emit a progress_update for a PM stage transition (advisory)."""
        await self._emit_guarded(
            "progress_update",
            {
                "stage": stage,
                "status": status,
                "metrics_snapshot": dict(self.metrics),
            },
        )

    @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
        ],
    )
    async def _emit_budget_events(self) -> None:
        """Emit fresh budget crossings for llm_calls/fix_cycles/duration.

        Edge-triggered via BudgetTracker: each metric crosses 80%
        (budget_warning) and 100% (budget_exceeded) at most once per
        iteration. Duration is measured from ``metrics["start_time"]``;
        an unset start time consumes no duration budget.
        """
        start = self.metrics.get("start_time") or 0.0
        elapsed = time.time() - start if start else 0.0
        for event in self._budget.consume(
            int(self.metrics.get("llm_calls", 0)),
            int(self.metrics.get("fix_cycles", 0)),
            elapsed,
        ):
            subject = (
                "budget_warning"
                if event.level is BudgetLevel.WARNING
                else "budget_exceeded"
            )
            await self._emit_guarded(
                subject,
                {
                    "metric": event.metric,
                    "threshold": event.threshold,
                    "current": event.current,
                },
            )

    @traces(spec="012", story="US-012-Progress", reqs=["REQ-012-051"])
    async def _emit_stall_if_due(self) -> None:
        """Emit iteration_stalled once when progress idled 5+ minutes."""
        now = time.time()
        if self._stall_emitted or not is_stalled(self._last_progress_ts, now):
            return
        self._stall_emitted = True
        await self._emit_guarded(
            "iteration_stalled",
            {"idle_seconds": round(now - self._last_progress_ts, 2)},
        )

    async def _emit_guarded(self, event_type: str, payload: dict) -> None:
        """Advisory event emission: observers always get it; CXP never raises.

        ``BaseAgent.emit_event`` delivers to local observers first and then
        sends the CXP packet; without a postman the CXP leg raises, which
        this guard swallows so budget reporting can never fail the
        pipeline (SPEC-012 NFR-002).
        """
        try:
            await self.emit_event(event_type, payload)
        except Exception as exc:
            logger.debug(
                "U:PM event %s not sent over CXP (advisory): %s",
                event_type,
                exc,
            )

    @traces(spec="012", story="US-012-Summary", reqs=["REQ-012-055"])
    async def finalize_iteration(self, project_dir: str | None = None) -> None:
        """Emit the structured iteration_summary and persist it.

        US-005: the summary is a structured dict (build_iteration_summary)
        emitted as an ``iteration_summary`` event and appended to the
        iteration ``conversation.md`` under the target project directory.
        Advisory-only: a missing project_dir skips persistence with a
        debug log, and file I/O failures never raise.
        """
        now = time.time()
        summary = build_iteration_summary(
            self.metrics, self.metrics.get("start_time") or 0.0, now
        )
        await self._emit_guarded("iteration_summary", summary)
        target = project_dir or self._project_dir
        if not target:
            logger.debug("U:PM iteration_summary not persisted: no project_dir known")
            return
        try:
            conversation = Path(target) / "conversation.md"
            block = json.dumps(summary, indent=2)
            conversation.parent.mkdir(parents=True, exist_ok=True)
            with conversation.open("a", encoding="utf-8") as fh:
                fh.write(f"\n## Iteration Summary\n```json\n{block}\n```\n")
        except OSError as exc:
            logger.warning("U:PM iteration_summary persist failed (advisory): %s", exc)


# Class alias for orchestrata-ultimate naming
class ProjectManagerAgent(PMVolk):
    """Agent alias for PMVolk."""

    pass