orchestrata_ultimate.agents.volks.pm.goal_registry

Reference › orchestrata_ultimate › agents › volks › pm › goal_registry

goal_registry is the central store for goal lifecycle management used by the Project Manager (volk; specification 012, goals stories). GoalRegistry takes a session manager and reads and writes the portfolio goals through the context file manager (_load_goals, _save_goals); _migrate_legacy_goals folds an older top-level goals blob into the current layout. It creates goals with a description, priority, success criteria and assigned volks (create_goal), looks them up (get_goal, list_goals with an optional status filter, get_active_goal) and drives the state transitions activate_goal, complete_goal, block_goal, defer_goal and cancel_goal, plus update_goal. An illegal transition raises GoalTransitionError and an unknown id raises GoalNotFoundError. record_iteration stores an iteration result against a goal and updates its progress estimate with _estimate_progress, which scores criteria such as test results (_score_tests); _parse_criteria reads success criteria from a dictionary or applies defaults. The goal domain types come from orchestrata_ultimate.core.domain.goal.

At a glance

Class diagram

Dependency diagram

API

Classes

GoalTransitionError

Bases: Exception · Decorators: traces(spec='012', story='US-012-Goals', reqs=['REQ-012-027']) · Line 56

Raised when an illegal goal state transition is attempted.

GoalNotFoundError

Bases: Exception · Decorators: traces(spec='012', story='US-012-Goals', reqs=['REQ-012-033']) · Line 63

Raised when a goal ID is not found.

GoalRegistry

Decorators: traces(spec='012', story='US-012-Goals', reqs=['REQ-012-035', 'REQ-012-040']) · Line 99

Central registry for goal lifecycle management.

Functions

Source

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

# mypy: ignore-errors
"""Goal Registry for U:PM — central goal lifecycle management.

Manages portfolio goals with full lifecycle tracking: create, activate,
update progress, complete, block, defer, cancel.

SPEC-012: Goal-driven intent routing and lifecycle (REQ-012-026 through REQ-012-040).

Persistence (SPEC-023 unification): goals persist THROUGH the injected
``ContextFileManager`` — the canonical context/goal store — under
``ProjectState.portfolio_goals``, a typed list distinct from SPEC-023's
singular ``ProjectState.goal`` (the current session's goal). This registry
never does raw ``json.load``/``json.dump`` of context.json; that was the old
persistence path and it silently lost data (pydantic's ``ContextData`` drops
unknown top-level keys on every write). The one exception is a one-time,
read-only migration of a *legacy* top-level ``context.json["goals"]`` blob
(pre-unification format) into the typed field — see ``_migrate_legacy_goals``.
"""

from __future__ import annotations

import json
import logging
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

# Portfolio goal domain types live in core (layer 1) — importing them here from
# the agents layer is a permitted downward import (the pre-unification eager
# import was from session.models at layer 4, an upward violation).
from orchestrata_ultimate.core.domain.goal import (
    PortfolioGoal as Goal,
)
from orchestrata_ultimate.core.domain.goal import (
    PortfolioGoalPriority as GoalPriority,
)
from orchestrata_ultimate.core.domain.goal import (
    PortfolioGoalStatus as GoalStatus,
)
from orchestrata_ultimate.core.domain.goal import (
    PortfolioSuccessCriteria as SuccessCriteria,
)

# The context store is only referenced in annotations here (the instance is
# injected), and SessionManagerProtocol is a layer-0 core protocol — a
# permitted downward import. session.context.ContextFileManager (layer 4)
# satisfies it structurally; importing that concrete here, even under
# TYPE_CHECKING, would be an upward violation.
from orchestrata_ultimate.core.protocols.session import SessionManagerProtocol

logger = logging.getLogger(__name__)


@traces(spec="012", story="US-012-Goals", reqs=["REQ-012-027"])
class GoalTransitionError(Exception):
    """Raised when an illegal goal state transition is attempted."""

    pass


@traces(spec="012", story="US-012-Goals", reqs=["REQ-012-033"])
class GoalNotFoundError(Exception):
    """Raised when a goal ID is not found."""

    pass


# Valid state transitions map
_VALID_TRANSITIONS = {
    GoalStatus.ACTIVE: [
        GoalStatus.RUNNING,
        GoalStatus.BLOCKED,
        GoalStatus.DEFERRED,
        GoalStatus.CANCELLED,
    ],
    GoalStatus.RUNNING: [
        GoalStatus.ACTIVE,
        GoalStatus.COMPLETED,
        GoalStatus.BLOCKED,
        GoalStatus.DEFERRED,
        GoalStatus.CANCELLED,
    ],
    GoalStatus.COMPLETED: [],  # Terminal state
    GoalStatus.BLOCKED: [GoalStatus.ACTIVE, GoalStatus.DEFERRED, GoalStatus.CANCELLED],
    GoalStatus.DEFERRED: [GoalStatus.ACTIVE, GoalStatus.CANCELLED],
    GoalStatus.CANCELLED: [],  # Terminal state
}


@traces(
    spec="012",
    story="US-012-Goals",
    reqs=[
        "REQ-012-035",  # Persists through ContextFileManager, atomically
        "REQ-012-040",  # Loads existing goals on construction
    ],
)
class GoalRegistry:
    """Central registry for goal lifecycle management.

    Reads and writes portfolio goals through an injected ``ContextFileManager``
    (composition/DI — this registry never constructs its own file I/O).
    Provides CRUD operations and lifecycle transitions.

    Attributes:
        context: The ``SessionManagerProtocol`` store this registry persists
            through (concretely a session-layer ``ContextFileManager``).
    """

    @traces(
        spec="012", story="US-012-Goals", reqs=["REQ-012-040"]
    )  # Load existing goals
    def __init__(self, context: SessionManagerProtocol) -> None:
        """Initialize the goal registry.

        Args:
            context: The context store (SPEC-023 canonical store —
                structurally a ``ContextFileManager``) this registry
                reads/writes ``portfolio_goals`` through.
        """
        self.context = context
        self._migrate_legacy_goals()

    @traces(spec="012", story="US-012-Goals", reqs=["REQ-012-035"])
    def _migrate_legacy_goals(self) -> None:
        """Fold a legacy top-level ``context.json["goals"]`` blob into
        ``state.portfolio_goals``, once.

        Pre-unification ``GoalRegistry`` wrote goals as a raw top-level
        ``"goals"`` array via direct ``json.dump`` — a key pydantic's
        ``ContextData`` doesn't know about and silently drops on every read
        (and therefore on every subsequent write, permanently). This detects
        that legacy shape directly from the raw file (the only way to see it
        before the schema discards it), migrates it into the typed
        ``portfolio_goals`` field, and writes it back through
        ``ContextFileManager`` — after which it's schema-owned and this is a
        no-op. Advisory: any I/O or parse failure is swallowed so a corrupt
        legacy blob never blocks normal goal tracking.
        """
        json_path = self.context.json_path
        if not json_path.exists():
            return
        try:
            raw = json.loads(json_path.read_text())
        except OSError, json.JSONDecodeError:
            return
        legacy_goals = raw.get("goals")
        if not legacy_goals:
            return
        try:
            ctx = self.context.read()
            if ctx.state.portfolio_goals:
                return  # already migrated / already schema-owned
            ctx.state.portfolio_goals = [Goal.model_validate(g) for g in legacy_goals]
            self.context.write(ctx)
        except Exception as exc:
            logger.warning("Legacy goal migration skipped: %s", exc)

    @traces(spec="012", story="US-012-Goals", reqs=["REQ-012-040"])
    def _load_goals(self) -> list[Goal]:
        """Read the current portfolio goals through ``ContextFileManager``."""
        return self.context.read().state.portfolio_goals

    @traces(spec="012", story="US-012-Goals", reqs=["REQ-012-035"])
    def _save_goals(self, goals: list[Goal]) -> None:
        """Write the portfolio goals through ``ContextFileManager``."""
        self.context.update_state(portfolio_goals=goals)

    @traces(spec="012", story="US-012-Goals", reqs=["REQ-012-034"])  # list_goals
    def list_goals(self, status: GoalStatus | None = None) -> list[Goal]:
        """List all goals, optionally filtered by status.

        Args:
            status: Filter to only this status, or None for all.

        Returns:
            List of Goal instances.
        """
        goals = self._load_goals()
        if status is not None:
            goals = [g for g in goals if g.status == status]
        return goals

    @traces(spec="012", story="US-012-Goals", reqs=["REQ-012-033"])  # get_goal
    def get_goal(self, goal_id: str) -> Goal:
        """Get a goal by ID.

        Args:
            goal_id: The goal identifier.

        Returns:
            Goal instance.

        Raises:
            GoalNotFoundError: When goal ID does not exist.
        """
        for goal in self.list_goals():
            if goal.goal_id == goal_id:
                return goal
        raise GoalNotFoundError(f"Goal not found: {goal_id}")

    def get_active_goal(self) -> Goal | None:
        """Get the currently active or running goal.

        Returns:
            The active/running Goal, or None.
        """
        active_statuses = (GoalStatus.ACTIVE, GoalStatus.RUNNING)
        for goal in self.list_goals():
            if goal.status in active_statuses:
                return goal
        return None

    @traces(
        spec="012",
        story="US-012-Goals",
        reqs=[
            "REQ-012-026",  # create_goal creates new Goal
            "REQ-012-035",  # Persists through ContextFileManager
        ],
    )
    def create_goal(
        self,
        description: str,
        priority: GoalPriority = GoalPriority.P1,
        success_criteria: SuccessCriteria | None = None,
        assigned_volks: list[str] | None = None,
    ) -> Goal:
        """Create and register a new goal.

        Args:
            description: Goal description.
            priority: Priority level.
            success_criteria: Completion criteria.
            assigned_volks: Volk aliases to assign.

        Returns:
            The newly created Goal.
        """
        goal = Goal(
            description=description,
            priority=priority,
            success_criteria=success_criteria or SuccessCriteria(),
            assigned_volks=assigned_volks or [],
        )
        goals = self._load_goals()
        goals.append(goal)
        self._save_goals(goals)
        logger.info("Created goal %s: %s", goal.goal_id, description)
        return goal

    @traces(
        spec="012",
        story="US-012-Goals",
        reqs=[
            "REQ-012-027",  # activate() transitions to ACTIVE
            "REQ-012-035",  # Persists through ContextFileManager
        ],
    )
    def activate_goal(self, goal_id: str) -> Goal:
        """Activate a goal, transitioning it to ACTIVE state.

        Args:
            goal_id: The goal to activate.

        Returns:
            Updated Goal.

        Raises:
            GoalNotFoundError: When goal ID does not exist.
            GoalTransitionError: For illegal transitions.
        """
        goal = self.get_goal(goal_id)
        if goal.status not in _VALID_TRANSITIONS:
            raise GoalTransitionError(f"Cannot activate goal in {goal.status} state")
        if GoalStatus.ACTIVE not in _VALID_TRANSITIONS.get(goal.status, []):
            raise GoalTransitionError(
                f"Illegal transition from {goal.status} to ACTIVE"
            )
        return self.update_goal(goal_id, status=GoalStatus.ACTIVE)

    @traces(
        spec="012",
        story="US-012-Goals",
        reqs=[
            "REQ-012-028",  # update_goal accepts partial dict
            "REQ-012-035",  # Persists through ContextFileManager
        ],
    )
    def update_goal(self, goal_id: str, **updates: Any) -> Goal:
        """Update fields on an existing goal.

        Args:
            goal_id: The goal to update.
            **updates: Field name/value pairs to update.

        Returns:
            Updated Goal.

        Raises:
            GoalNotFoundError: When goal ID does not exist.
        """
        goals = self._load_goals()
        for goal in goals:
            if goal.goal_id == goal_id:
                for key, value in updates.items():
                    setattr(goal, key, value)
                self._save_goals(goals)
                return goal
        raise GoalNotFoundError(f"Goal not found: {goal_id}")

    @traces(
        spec="012",
        story="US-012-Goals",
        reqs=[
            "REQ-012-029",  # complete_goal transitions to COMPLETED
            "REQ-012-035",  # Persists through ContextFileManager
        ],
    )
    def complete_goal(self, goal_id: str, notes: str = "") -> Goal:
        """Mark a goal as completed.

        Args:
            goal_id: The goal to complete.
            notes: Completion notes.

        Returns:
            Updated Goal.

        Raises:
            GoalNotFoundError: When goal ID does not exist.
        """
        now = datetime.now(UTC).isoformat()
        return self.update_goal(
            goal_id,
            status=GoalStatus.COMPLETED,
            completed_at=now,
            progress_pct=100.0,
            notes=notes,
        )

    @traces(
        spec="012",
        story="US-012-Goals",
        reqs=[
            "REQ-012-030",  # block_goal transitions to BLOCKED
            "REQ-012-035",  # Persists through ContextFileManager
        ],
    )
    def block_goal(self, goal_id: str, reason: str) -> Goal:
        """Mark a goal as blocked.

        Args:
            goal_id: The goal to block.
            reason: Reason for blocking.

        Returns:
            Updated Goal.

        Raises:
            GoalNotFoundError: When goal ID does not exist.
        """
        goal = self.get_goal(goal_id)
        gaps = list(goal.gap_analysis)
        gaps.append(f"BLOCKED: {reason}")
        return self.update_goal(
            goal_id,
            status=GoalStatus.BLOCKED,
            gap_analysis=gaps,
        )

    @traces(
        spec="012",
        story="US-012-Goals",
        reqs=[
            "REQ-012-031",  # defer_goal transitions to DEFERRED
            "REQ-012-035",  # Persists through ContextFileManager
        ],
    )
    def defer_goal(self, goal_id: str, reason: str) -> Goal:
        """Mark a goal as deferred.

        Args:
            goal_id: The goal to defer.
            reason: Reason for deferring.

        Returns:
            Updated Goal.

        Raises:
            GoalNotFoundError: When goal ID does not exist.
        """
        goal = self.get_goal(goal_id)
        notes = goal.notes + f"\n[DEFERRED: {reason}]"
        return self.update_goal(
            goal_id,
            status=GoalStatus.DEFERRED,
            notes=notes,
        )

    @traces(
        spec="012",
        story="US-012-Goals",
        reqs=[
            "REQ-012-032",  # cancel_goal transitions to CANCELLED
            "REQ-012-035",  # Persists through ContextFileManager
        ],
    )
    def cancel_goal(self, goal_id: str, reason: str) -> Goal:
        """Mark a goal as cancelled.

        Args:
            goal_id: The goal to cancel.
            reason: Reason for cancelling.

        Returns:
            Updated Goal.

        Raises:
            GoalNotFoundError: When goal ID does not exist.
        """
        goal = self.get_goal(goal_id)
        notes = goal.notes + f"\n[CANCELLED: {reason}]"
        return self.update_goal(
            goal_id,
            status=GoalStatus.CANCELLED,
            notes=notes,
        )

    @traces(
        spec="012",
        story="US-012-Goals",
        reqs=[
            "REQ-012-035",  # Persists through ContextFileManager
        ],
    )
    def record_iteration(
        self,
        goal_id: str,
        result: dict[str, Any],
        gaps: list[str] | None = None,
    ) -> Goal:
        """Record an iteration result against a goal.

        Args:
            goal_id: The goal to update.
            result: Iteration result dict.
            gaps: New gaps identified in this iteration.

        Returns:
            Updated Goal.

        Raises:
            GoalNotFoundError: When goal ID does not exist.
        """
        goal = self.get_goal(goal_id)
        new_gaps = list(goal.gap_analysis)
        if gaps:
            new_gaps.extend(gaps)
        progress = _estimate_progress(goal, result)
        return self.update_goal(
            goal_id,
            iterations_spent=goal.iterations_spent + 1,
            last_result=result,
            gap_analysis=new_gaps,
            progress_pct=progress,
            status=GoalStatus.RUNNING,
        )


def _parse_criteria(
    criteria_data: Any,
) -> SuccessCriteria:
    """Parse success criteria from dict or return defaults.

    Args:
        criteria_data: Raw criteria dict or non-dict value.

    Returns:
        SuccessCriteria instance.
    """
    if not isinstance(criteria_data, dict):
        return SuccessCriteria()
    return SuccessCriteria(
        tests_pass=criteria_data.get("tests_pass", True),
        min_coverage_pct=criteria_data.get("min_coverage_pct", 90.0),
        min_spec_coverage_pct=criteria_data.get("min_spec_coverage_pct", 0.0),
        max_iterations=criteria_data.get("max_iterations", 20),
        custom=criteria_data.get("custom", {}),
    )


@traces(spec="012", story="US-012-Reporting", reqs=["REQ-012-049"])
def _estimate_progress(goal: Goal, result: dict[str, Any]) -> float:
    """Estimate goal progress percentage from iteration result.

    Args:
        goal: The current goal.
        result: Latest iteration result.

    Returns:
        Estimated progress as a float 0-100.
    """
    criteria = goal.success_criteria
    score = 0.0
    total = 0.0

    if criteria.tests_pass:
        total += 50.0
        score += _score_tests(result)

    if criteria.min_coverage_pct > 0:
        total += 50.0
        actual = result.get("coverage_pct", 0.0)
        ratio = min(actual / criteria.min_coverage_pct, 1.0)
        score += 50.0 * ratio

    if total == 0.0:
        return 0.0
    return round((score / total) * 100, 1)


@traces(spec="012", story="US-012-Reporting", reqs=["REQ-012-049"])
def _score_tests(result: dict[str, Any]) -> float:
    """Score the test-pass criterion from a result dict.

    Args:
        result: Iteration result with test counts.

    Returns:
        Score out of 50.0 for the tests criterion.
    """
    if result.get("tests_passed", False):
        return 50.0
    pass_count = result.get("pass_count", 0)
    fail_count = result.get("fail_count", 0)
    total_tests = pass_count + fail_count
    if total_tests > 0:
        return 50.0 * (pass_count / total_tests)
    return 0.0