orchestrata_ultimate.session.controller._goals

Reference › orchestrata_ultimate › session › controller › _goals

GoalMixin adds the goal-driven session system (SPEC-023) to the SessionController. A Goal with exit criteria and tasks (see models) is stored in context.json. The methods are set_goal, get_goal, get_goal_status (formatted for display), update_goal_status (achieved, blocked or abandoned), log_goal_progress, add_blocker and clear_blocker, advance_goal_step and set_goal_step, and load_goal_tasks_from_plan, which fills the goal's tasks from an execution plan.

goal_evaluate runs every criterion's verify_command and grades it. A criterion without a command is manual and unchanged. A verifier passes when the command exits 0 and, if a verify_pattern is set, the pattern matches the output; each command has a timeout (120 seconds by default). Changes from pending to met or from met to failed are logged as progress entries and printed. When every criterion is met the goal becomes achieved and AudioAlarm plays the success tone, unless play_alarm is off.

check_goal_before_stop is the Stop gate. It returns an empty string when there is no goal, the goal is achieved or abandoned, or all criteria are met; otherwise it returns a message that lists the remaining criteria and asks the agent to continue. The controller's end raises GoalNotAchievedError on that condition.

At a glance

Class diagram

Dependency diagram

API

Classes

GoalMixin

Line 23

SPEC-023 goal-driven session system methods.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/controller/_goals.py

# mypy: ignore-errors
"""GoalMixin — SPEC-023 goal-driven session methods for SessionController."""

import os
import re
import shlex
import subprocess  # noqa: S404
from collections.abc import Mapping
from datetime import datetime

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.planning.plan_parser import PlanStep, parse_plan_file
from orchestrata_ultimate.session.context import ContextFileManager
from orchestrata_ultimate.session.exceptions import GoalNotAchievedError
from orchestrata_ultimate.session.models import (
    Goal,
    GoalCriterion,
    GoalProgress,
    GoalTask,
)


class GoalMixin:
    """SPEC-023 goal-driven session system methods."""

    # Provided by SessionController.__init__ (composed via multiple inheritance).
    ctx: ContextFileManager

    def set_goal(
        self, statement: str, criteria: list[GoalCriterion] | None = None
    ) -> Goal:
        """Set a new goal for the session.

        Args:
            statement: Goal statement describing what to achieve.
            criteria: Optional list of exit criteria.

        Returns:
            The newly created Goal.
        """
        goal = Goal(
            statement=statement,
            criteria=criteria or [],
        )
        self.ctx.update_state(goal=goal)
        return goal

    def _steps_to_tasks(self, steps: list[PlanStep]) -> list[GoalTask]:
        """Convert parsed plan steps into goal tasks."""
        tasks: list[GoalTask] = []
        for step in steps:
            description = f"{step.step_num}. {step.title}"
            if step.goal:
                description += f" — {step.goal}"
            tasks.append(
                GoalTask(
                    id=f"step-{step.step_num}",
                    description=description,
                )
            )
        return tasks

    def load_goal_tasks_from_plan(
        self,
        plan_path: str | None = None,
        *,
        env: Mapping[str, str] | None = None,
    ) -> Goal | None:
        """Hydrate the current goal's tasks from an execution plan.

        Args:
            plan_path: Path to the plan markdown. If None, resolves via
                ``ORCHESTRATA_PLAN_PATH`` or defaults to ``plans/execution-plan.md``.
            env: Optional environment mapping for testing.

        Returns:
            Updated Goal or None if no goal exists.
        """
        goal = self.get_goal()
        if goal is None:
            return None

        resolved_env: dict[str, str] = (
            dict(env) if env is not None else dict(os.environ)
        )
        steps = parse_plan_file(plan_path, env=resolved_env)
        goal.tasks = self._steps_to_tasks(steps)
        goal.current_step = 0
        self.ctx.update_state(goal=goal)
        return goal

    def get_goal(self) -> Goal | None:
        """Get the current goal from context.json.

        Returns:
            The current Goal if one exists, None otherwise.
        """
        ctx = self.ctx.read()
        return ctx.state.goal

    def get_goal_status(self) -> str:
        """Get formatted goal status for display.

        Returns:
            Formatted string showing goal statement and criteria status.
        """
        goal = self.get_goal()
        if goal is None:
            return "No active goal. Use controller.set_goal() to define one."

        lines = [
            "## Active Goal",
            "",
            f"**Statement**: {goal.statement}",
            f"**Status**: {goal.status}",
            "",
            "### Exit Criteria",
        ]
        for criterion in goal.criteria:
            status_icon = "[x]" if criterion.status == "met" else "[ ]"
            lines.append(f"- {status_icon} {criterion.description}")

        lines.append("")
        lines.append("Continue working toward this goal.")

        return "\n".join(lines)

    def update_goal_status(self, status: str) -> Goal | None:
        """Update goal status (achieved, blocked, abandoned).

        Args:
            status: New status (achieved, blocked, abandoned).

        Returns:
            Updated Goal or None if no goal exists.
        """
        goal = self.get_goal()
        if goal is None:
            return None

        goal.status = status
        if status == "achieved":
            goal.achieved_at = datetime.now().isoformat()
        elif status == "abandoned":
            goal.abandoned_at = datetime.now().isoformat()

        self.ctx.update_state(goal=goal)
        return goal

    def log_goal_progress(
        self, action: str, note: str, criteria_changed: dict[str, str] | None = None
    ) -> Goal | None:
        """Log progress toward the current goal.

        Args:
            action: Action taken (e.g., "CI run", "Fixed tests").
            note: Details about the action.
            criteria_changed: Optional dict of criterion_id -> new status.

        Returns:
            Updated Goal or None if no goal exists.
        """
        goal = self.get_goal()
        if goal is None:
            return None

        progress_entry = GoalProgress(
            action=action,
            note=note,
            criteria_changed=criteria_changed or {},
        )
        goal.progress.append(progress_entry)

        # Update criteria statuses if provided
        if criteria_changed:
            for criterion in goal.criteria:
                if criterion.id in criteria_changed:
                    criterion.status = criteria_changed[criterion.id]
                    if criteria_changed[criterion.id] == "met":
                        criterion.met_at = datetime.now().isoformat()

        self.ctx.update_state(goal=goal)
        return goal

    def check_goal_before_stop(self) -> str:
        """Check goal status before allowing the agent to stop.

        Returns:
            Message to inject if goal is not achieved, empty string if stop allowed.
        """
        goal = self.get_goal()
        if goal is None:
            return ""  # No goal, allow stop

        if goal.status in ("achieved", "abandoned"):
            return ""  # Goal resolved, allow stop

        # Check if all criteria are met
        all_met = goal.criteria and all(c.status == "met" for c in goal.criteria)
        if all_met:
            return ""  # All criteria met, allow stop

        # Goal not achieved, prevent premature stopping
        remaining = [c.description for c in goal.criteria if c.status != "met"]
        message = "Goal not achieved. Remaining criteria:\n"
        for desc in remaining:
            message += f"- {desc}\n"
        message += "\nContinue working toward this goal."

        return message

    @traces(
        spec="023",
        story="US-005",
        reqs=[
            "REQ-023-022",  # the Stop gate reads state.goal from context.json
            "REQ-023-023",  # active + incomplete + unblocked => inject and continue
            "REQ-023-024",  # blocked => inject the blocker reason + attention alarm
            "REQ-023-025",  # no active goal => do nothing
        ],
    )
    def _enforce_goal_stop_gate(self) -> None:
        """SPEC-023 US-005 Stop gate, run by ``SessionController.end``.

        Refuses to end while the goal is active, incomplete and
        unblocked (raises :class:`GoalNotAchievedError` carrying the
        injected message); a blocked goal legitimately ends the run and
        triggers the attention alarm; an achieved/abandoned/absent goal
        does nothing.
        """
        goal = self.get_goal()
        if goal is None or goal.status in ("achieved", "abandoned"):
            return  # REQ-023-025: no active goal — existing behavior
        if goal.status == "blocked":
            # REQ-023-024: inject the blocker reason + attention alarm.
            reason = goal.blockers[-1] if goal.blockers else "unknown reason"
            print(f"  [goal] BLOCKED: {reason}", flush=True)
            from orchestrata_ultimate.session.audio_alarm import AudioAlarm

            AudioAlarm().play_attention()
            return
        message = self.check_goal_before_stop()
        if message:
            raise GoalNotAchievedError(message)

    @traces(
        spec="023",
        story="US-002",
        reqs=[
            "REQ-023-007",  # evaluate all exit criteria after each run
            "REQ-023-008",  # pending->met transitions are logged and announced
            "REQ-023-009",  # met->failed regressions are logged and announced
            "REQ-023-010",  # all criteria met => goal transitions to achieved
            "REQ-023-011",  # evaluation is automatic, not user-triggered
            "REQ-023-012",  # each evaluation logged as a progress entry
        ],
    )
    def goal_evaluate(
        self,
        *,
        env: Mapping[str, str] | None = None,
        timeout_seconds: int = 120,
        play_alarm: bool = True,
    ) -> Goal | None:
        """Run every criterion's ``verify_command`` and grade it (US-002).

        Criteria without a ``verify_command`` are manual and stay as they
        are. A verifier passes when the command exits 0 and (when a
        ``verify_pattern`` is set) the pattern matches the combined
        stdout/stderr. Status transitions are logged as progress entries
        with timestamps; achieving every criterion transitions the goal
        to ``achieved`` and plays the success alarm. ``play_alarm=False``
        is the test seam that skips audio without skipping evaluation.
        """
        goal = self.get_goal()
        if goal is None or goal.status != "active":
            return goal
        changed: dict[str, str] = {}
        announcements: list[str] = []
        for criterion in goal.criteria:
            if not criterion.verify_command:
                continue
            passed, output = self._run_criterion_verifier(
                criterion, env=env, timeout_seconds=timeout_seconds
            )
            new_status = "met" if passed else "failed"
            if criterion.status != new_status:
                changed[criterion.id] = new_status
                announcements.append(
                    f"- {criterion.description}: {criterion.status} -> {new_status}"
                )
                if not passed and output.strip():
                    announcements.append(f"  output: {output.strip()[:500]}")
        if changed:
            self.log_goal_progress(
                "goal_evaluate",
                "\n".join(announcements),
                criteria_changed=changed,
            )
            for line in announcements:
                print(f"  [goal] {line}", flush=True)
        goal = self.get_goal()
        all_met = goal.criteria and all(c.status == "met" for c in goal.criteria)
        if all_met:
            self.update_goal_status("achieved")
            if play_alarm:
                from orchestrata_ultimate.session.audio_alarm import AudioAlarm

                AudioAlarm().play_success()
        return self.get_goal()

    @traces(spec="023", story="US-002", reqs=["REQ-023-007"])
    def _run_criterion_verifier(
        self,
        criterion: GoalCriterion,
        *,
        env: Mapping[str, str] | None = None,
        timeout_seconds: int = 120,
    ) -> tuple[bool, str]:
        """Run one criterion's verify_command and grade the output."""
        parts = shlex.split(criterion.verify_command or "")
        if not parts:
            return False, "empty verify_command"
        resolved_env = dict(env) if env is not None else None
        root = getattr(self.ctx, "root", None)
        try:
            process = subprocess.run(  # noqa: S603
                parts,
                capture_output=True,
                text=True,
                timeout=timeout_seconds,
                env=resolved_env,
                cwd=root,
            )
        except (OSError, subprocess.TimeoutExpired) as exc:
            return False, str(exc)
        output = f"{process.stdout or ''}{process.stderr or ''}"
        if process.returncode != 0:
            return False, output
        if criterion.verify_pattern:
            if re.search(criterion.verify_pattern, output) is None:
                return False, output
        return True, output

    def add_blocker(self, reason: str) -> Goal | None:
        """Add a blocker to the current goal.

        Args:
            reason: Reason for blocking.

        Returns:
            Updated Goal or None if no goal exists.
        """
        goal = self.get_goal()
        if goal is None:
            return None

        goal.blockers.append(reason)
        goal.status = "blocked"

        self.ctx.update_state(goal=goal)
        return goal

    def clear_blocker(self) -> Goal | None:
        """Clear blocker status and resume goal.

        Returns:
            Updated Goal or None if no goal exists.
        """
        goal = self.get_goal()
        if goal is None:
            return None

        goal.blockers = []
        goal.status = "active"

        self.ctx.update_state(goal=goal)
        return goal

    def advance_goal_step(self) -> Goal | None:
        """Mark the current task done and move to the next step.

        ``current_step`` is incremented by one and clamped to the number of
        tasks so it never points past the end of the list. The task at the
        previous position is marked ``done``.

        Returns:
            Updated Goal or None if no goal exists.
        """
        goal = self.get_goal()
        if goal is None:
            return None

        if 0 <= goal.current_step < len(goal.tasks):
            task = goal.tasks[goal.current_step]
            task.status = "done"
            task.done_at = datetime.now().isoformat()

        goal.current_step = min(goal.current_step + 1, len(goal.tasks))
        self.ctx.update_state(goal=goal)
        return goal

    def set_goal_step(self, index: int) -> Goal | None:
        """Set the current step index explicitly.

        Args:
            index: 0-based task index. Negative values or values beyond the
                task list are clamped to the valid range.

        Returns:
            Updated Goal or None if no goal exists.
        """
        goal = self.get_goal()
        if goal is None:
            return None

        max_index = max(0, len(goal.tasks) - 1)
        goal.current_step = max(0, min(index, max_index))
        self.ctx.update_state(goal=goal)
        return goal