From 24ee45e0e19b4a9f2fb670d9939519a61ae61137 Mon Sep 17 00:00:00 2001 From: Yichen Wu Date: Tue, 21 Jul 2026 14:58:35 -0700 Subject: [PATCH] Make loop lifecycle explicit --- ARCHITECTURE.md | 52 ++ README.md | 6 + .../versions/0011_loop_state_machine.py | 63 ++ apps/api/app/db/models/task.py | 6 + apps/api/app/domain/loop.py | 178 ++++++ apps/api/app/repositories/task.py | 29 +- apps/api/app/schemas/task.py | 13 + apps/api/app/services/agent_react.py | 540 +++++++++--------- apps/api/app/services/loop/__init__.py | 34 ++ apps/api/app/services/loop/context.py | 60 ++ apps/api/app/services/loop/decisions.py | 78 +++ apps/api/app/services/loop/delegation.py | 38 ++ apps/api/app/services/loop/dispatch.py | 51 ++ apps/api/app/services/loop/verification.py | 83 +++ apps/api/app/services/runner.py | 8 + apps/api/app/services/task.py | 12 +- apps/api/tests/test_agent_react.py | 13 +- apps/api/tests/test_chat.py | 3 + apps/api/tests/test_deployment_boundaries.py | 2 +- apps/api/tests/test_loop_components.py | 320 +++++++++++ apps/api/tests/test_runner.py | 37 ++ apps/api/tests/test_tasks.py | 10 + apps/web/app/tasks/[id]/page.tsx | 2 + apps/web/components/authority-panel.test.tsx | 1 + apps/web/components/loop-state.test.tsx | 18 + apps/web/components/loop-state.tsx | 36 ++ docs/STRATEGY.md | 9 + docs/adr/0011-explicit-loop-state-machine.md | 46 ++ docs/adr/README.md | 1 + docs/loop.md | 38 ++ packages/api-contract/src/index.ts | 17 + scripts/k8s-deployment-acceptance.sh | 2 +- 32 files changed, 1519 insertions(+), 287 deletions(-) create mode 100644 apps/api/alembic/versions/0011_loop_state_machine.py create mode 100644 apps/api/app/domain/loop.py create mode 100644 apps/api/app/services/loop/__init__.py create mode 100644 apps/api/app/services/loop/context.py create mode 100644 apps/api/app/services/loop/decisions.py create mode 100644 apps/api/app/services/loop/delegation.py create mode 100644 apps/api/app/services/loop/dispatch.py create mode 100644 apps/api/app/services/loop/verification.py create mode 100644 apps/api/tests/test_loop_components.py create mode 100644 apps/web/components/loop-state.test.tsx create mode 100644 apps/web/components/loop-state.tsx create mode 100644 docs/adr/0011-explicit-loop-state-machine.md diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 95c35da..78ca4d5 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -7,6 +7,7 @@ boundary. - [Design principles](#design-principles) - [System topology](#system-topology) - [Backend: layered architecture](#backend-layered-architecture) +- [Loop orchestration](#loop-orchestration) - [Request lifecycle](#request-lifecycle) - [Data & persistence](#data--persistence) - [Caching & background work](#caching--background-work) @@ -102,6 +103,57 @@ slice rather than bypassing the service/repository boundary. --- +## Loop orchestration + +`AgentReactService` is the runtime coordinator, not the source of every orchestration +decision. It wires model clients, sandboxes, gateways, repositories, and Receipts around +small policy components with explicit inputs and no side effects: + +| Component | Owns | +| ---------------------- | --------------------------------------------------------------- | +| `domain/loop.py` | allowed lifecycle transitions and status/stop-reason projection | +| `loop/decisions.py` | extraction and validation of one model action | +| `loop/context.py` | planner/verifier token allocation and bounded history | +| `progress.py` | semantic no-progress and repeated-branch detection | +| `loop/dispatch.py` | invalid, blocked, approval, or executable routing | +| `loop/verification.py` | evidence gates, coverage, verifier verdict, and acceptance | +| `loop/delegation.py` | child step/token allocation within the parent's remainder | + +The lifecycle is persisted as `loop_state`, `transition_reason`, and a monotonically +increasing `transition_sequence`. `status` remains the coarse external compatibility +projection. A transition not present in the policy fails closed. + +```mermaid +stateDiagram-v2 + [*] --> queued + queued --> preparing: claim + preparing --> understanding: contract or rubric required + preparing --> planning: runtime prepared + understanding --> planning: contract or rubric ready + planning --> acting: action selected + acting --> planning: action recorded + planning --> verifying: verification requested + verifying --> planning: evidence rejected + verifying --> completed: evidence accepted + preparing --> awaiting_input: clarification required + acting --> awaiting_input: input required + acting --> awaiting_approval: approval required + awaiting_input --> queued: user responded + awaiting_approval --> queued: user responded + preparing --> stopped: bounded limit + understanding --> stopped: bounded limit + planning --> stopped: bounded limit + acting --> stopped: bounded limit + verifying --> stopped: bounded limit +``` + +Every active phase also allows cancellation and failure. Every working phase can recover +to `preparing`; the durable runner normally requeues an interrupted task first, while +atomic claiming also repairs a legacy or partially upgraded `pending`/working-state row. +Waiting phases are deliberately excluded from automatic claiming. + +--- + ## Request lifecycle ```mermaid diff --git a/README.md b/README.md index 37a6466..4d457a6 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,9 @@ an instruction to an accepted result. It separates **work** from **acceptance**: - **Interruption is a first-class state.** In-flight cancellation tears down the active execution boundary, while a durable operation journal refuses crash recovery that could repeat a mutation with an unknown outcome. +- **The loop is an explicit state machine.** Phase, transition reason, and sequence are + persisted rather than inferred from logs; recovery, the API, and the task UI therefore + share the same semantics. If the evidence fails, the task does not become `completed` merely because the model called `finish`. Limit, stuck, cancelled, and error outcomes still receive an @@ -158,6 +161,9 @@ and residual risks. first mutation, and an explicit pause when risk, confidence, or authority is not safe to infer. - ReAct planning with independent executor/verifier provider selection and fallback. +- A persisted fail-closed loop state machine with independently tested policies for + transitions, decision parsing, context budgeting, progress control, dispatch, + verification, and delegation. - Strict acceptance contracts with required final artifacts, independent check snapshots, baseline regression checks, criterion-to-evidence mappings, Receipt replay, and offline verification. diff --git a/apps/api/alembic/versions/0011_loop_state_machine.py b/apps/api/alembic/versions/0011_loop_state_machine.py new file mode 100644 index 0000000..e29182f --- /dev/null +++ b/apps/api/alembic/versions/0011_loop_state_machine.py @@ -0,0 +1,63 @@ +"""Persist the explicit loop state and latest transition reason. + +Revision ID: 0011_loop_state_machine +Revises: 0010_operation_journal +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0011_loop_state_machine" +down_revision: str | None = "0010_operation_journal" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column( + "tasks", + sa.Column("loop_state", sa.String(length=30), nullable=False, server_default="queued"), + ) + op.add_column("tasks", sa.Column("transition_reason", sa.Text(), nullable=True)) + op.add_column( + "tasks", + sa.Column("transition_sequence", sa.Integer(), nullable=False, server_default="0"), + ) + op.create_index("ix_tasks_loop_state", "tasks", ["loop_state"], unique=False) + op.execute( + """ + UPDATE tasks + SET loop_state = CASE + WHEN status = 'pending' THEN 'queued' + WHEN status = 'running' THEN 'preparing' + WHEN status = 'awaiting_input' THEN 'awaiting_input' + WHEN status = 'completed' AND stop_reason = 'goal_achieved' THEN 'completed' + WHEN status = 'completed' THEN 'stopped' + WHEN status = 'stopped' THEN 'stopped' + WHEN status = 'cancelled' THEN 'cancelled' + WHEN status = 'failed' THEN 'failed' + ELSE 'failed' + END, + status = CASE + WHEN status = 'completed' AND COALESCE(stop_reason, '') <> 'goal_achieved' + THEN 'stopped' + WHEN status IN ( + 'pending', 'running', 'awaiting_input', 'completed', + 'stopped', 'cancelled', 'failed' + ) THEN status + ELSE 'failed' + END, + transition_reason = 'migrated_from_task_status' + """ + ) + + +def downgrade() -> None: + op.drop_index("ix_tasks_loop_state", table_name="tasks") + op.drop_column("tasks", "transition_sequence") + op.drop_column("tasks", "transition_reason") + op.drop_column("tasks", "loop_state") diff --git a/apps/api/app/db/models/task.py b/apps/api/app/db/models/task.py index 4671d72..71cf3c7 100644 --- a/apps/api/app/db/models/task.py +++ b/apps/api/app/db/models/task.py @@ -14,6 +14,7 @@ from sqlalchemy.orm import Mapped, mapped_column from app.db.base import Base, TimestampMixin, UUIDPrimaryKeyMixin +from app.domain.loop import LoopState from app.domain.task import TaskStatus @@ -29,6 +30,11 @@ class TaskModel(UUIDPrimaryKeyMixin, TimestampMixin, Base): status: Mapped[str] = mapped_column( String(20), nullable=False, default=TaskStatus.PENDING.value, index=True ) + loop_state: Mapped[str] = mapped_column( + String(30), nullable=False, default=LoopState.QUEUED.value, index=True + ) + transition_reason: Mapped[str | None] = mapped_column(Text, nullable=True) + transition_sequence: Mapped[int] = mapped_column(Integer, nullable=False, default=0) rubric: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list) criteria_source: Mapped[str] = mapped_column(String(20), nullable=False, default="generated") verification_mode: Mapped[str] = mapped_column(String(20), nullable=False, default="judgment") diff --git a/apps/api/app/domain/loop.py b/apps/api/app/domain/loop.py new file mode 100644 index 0000000..515f3c7 --- /dev/null +++ b/apps/api/app/domain/loop.py @@ -0,0 +1,178 @@ +"""The authoritative task-loop transition policy.""" + +from __future__ import annotations + +import enum +from dataclasses import dataclass +from typing import Protocol + +from app.domain.task import StopReason, TaskStatus + + +class LoopState(enum.StrEnum): + QUEUED = "queued" + PREPARING = "preparing" + UNDERSTANDING = "understanding" + PLANNING = "planning" + ACTING = "acting" + VERIFYING = "verifying" + AWAITING_INPUT = "awaiting_input" + AWAITING_APPROVAL = "awaiting_approval" + COMPLETED = "completed" + STOPPED = "stopped" + CANCELLED = "cancelled" + FAILED = "failed" + + +class LoopEvent(enum.StrEnum): + CLAIM = "claim" + RECOVER = "recover" + CONTRACT_REQUIRED = "contract_required" + CONTRACT_READY = "contract_ready" + RUBRIC_REQUIRED = "rubric_required" + PREPARATION_COMPLETED = "preparation_completed" + RUBRIC_READY = "rubric_ready" + ACTION_SELECTED = "action_selected" + ACTION_RECORDED = "action_recorded" + VERIFICATION_REQUESTED = "verification_requested" + VERIFICATION_REJECTED = "verification_rejected" + VERIFICATION_ACCEPTED = "verification_accepted" + INPUT_REQUIRED = "input_required" + APPROVAL_REQUIRED = "approval_required" + USER_RESPONDED = "user_responded" + LIMIT_REACHED = "limit_reached" + CANCEL = "cancel" + FAIL = "fail" + + +class InvalidLoopTransitionError(ValueError): + pass + + +class TransitionTarget(Protocol): + loop_state: str + transition_reason: str | None + transition_sequence: int + status: str + stop_reason: str | None + + +@dataclass(frozen=True, slots=True) +class LoopTransition: + sequence: int + source: LoopState + target: LoopState + event: LoopEvent + reason: str + + +_WORKING = { + LoopState.PREPARING, + LoopState.UNDERSTANDING, + LoopState.PLANNING, + LoopState.ACTING, + LoopState.VERIFYING, +} +_ACTIVE = {LoopState.QUEUED, *_WORKING, LoopState.AWAITING_INPUT, LoopState.AWAITING_APPROVAL} + +_RULES: dict[tuple[LoopState, LoopEvent], LoopState] = { + (LoopState.QUEUED, LoopEvent.CLAIM): LoopState.PREPARING, + (LoopState.PREPARING, LoopEvent.CONTRACT_REQUIRED): LoopState.UNDERSTANDING, + (LoopState.UNDERSTANDING, LoopEvent.CONTRACT_READY): LoopState.PLANNING, + (LoopState.PREPARING, LoopEvent.RUBRIC_REQUIRED): LoopState.UNDERSTANDING, + (LoopState.PREPARING, LoopEvent.PREPARATION_COMPLETED): LoopState.PLANNING, + (LoopState.UNDERSTANDING, LoopEvent.RUBRIC_READY): LoopState.PLANNING, + (LoopState.PLANNING, LoopEvent.ACTION_SELECTED): LoopState.ACTING, + (LoopState.ACTING, LoopEvent.ACTION_RECORDED): LoopState.PLANNING, + (LoopState.PLANNING, LoopEvent.VERIFICATION_REQUESTED): LoopState.VERIFYING, + (LoopState.VERIFYING, LoopEvent.VERIFICATION_REJECTED): LoopState.PLANNING, + (LoopState.VERIFYING, LoopEvent.VERIFICATION_ACCEPTED): LoopState.COMPLETED, + (LoopState.PREPARING, LoopEvent.INPUT_REQUIRED): LoopState.AWAITING_INPUT, + (LoopState.UNDERSTANDING, LoopEvent.INPUT_REQUIRED): LoopState.AWAITING_INPUT, + (LoopState.ACTING, LoopEvent.INPUT_REQUIRED): LoopState.AWAITING_INPUT, + (LoopState.ACTING, LoopEvent.APPROVAL_REQUIRED): LoopState.AWAITING_APPROVAL, + (LoopState.AWAITING_INPUT, LoopEvent.USER_RESPONDED): LoopState.QUEUED, + (LoopState.AWAITING_APPROVAL, LoopEvent.USER_RESPONDED): LoopState.QUEUED, +} +for state in _WORKING: + _RULES[(state, LoopEvent.RECOVER)] = LoopState.PREPARING + _RULES[(state, LoopEvent.LIMIT_REACHED)] = LoopState.STOPPED +for state in _ACTIVE: + _RULES[(state, LoopEvent.CANCEL)] = LoopState.CANCELLED + _RULES[(state, LoopEvent.FAIL)] = LoopState.FAILED + +_STATUS_BY_STATE = { + LoopState.QUEUED: TaskStatus.PENDING, + LoopState.PREPARING: TaskStatus.RUNNING, + LoopState.UNDERSTANDING: TaskStatus.RUNNING, + LoopState.PLANNING: TaskStatus.RUNNING, + LoopState.ACTING: TaskStatus.RUNNING, + LoopState.VERIFYING: TaskStatus.RUNNING, + LoopState.AWAITING_INPUT: TaskStatus.AWAITING_INPUT, + LoopState.AWAITING_APPROVAL: TaskStatus.AWAITING_INPUT, + LoopState.COMPLETED: TaskStatus.COMPLETED, + LoopState.STOPPED: TaskStatus.STOPPED, + LoopState.CANCELLED: TaskStatus.CANCELLED, + LoopState.FAILED: TaskStatus.FAILED, +} + + +class LoopTransitionPolicy: + @staticmethod + def claimable_states() -> frozenset[LoopState]: + return frozenset({LoopState.QUEUED, *_WORKING}) + + @staticmethod + def allowed_events(state: LoopState) -> frozenset[LoopEvent]: + return frozenset(event for source, event in _RULES if source is state) + + @staticmethod + def status_for(state: LoopState) -> TaskStatus: + return _STATUS_BY_STATE[state] + + def apply( + self, + task: TransitionTarget, + event: LoopEvent, + reason: str, + *, + stop_reason: StopReason | None = None, + ) -> LoopTransition: + try: + source = LoopState(task.loop_state) + except ValueError as exc: + raise InvalidLoopTransitionError(f"Unknown loop state {task.loop_state!r}") from exc + target = _RULES.get((source, event)) + if target is None: + raise InvalidLoopTransitionError(f"{event.value} is not allowed from {source.value}") + if event is LoopEvent.LIMIT_REACHED and stop_reason not in { + StopReason.MAX_STEPS, + StopReason.BUDGET_EXHAUSTED, + StopReason.STUCK, + }: + raise InvalidLoopTransitionError("A limit transition needs a bounded stop reason") + + sequence = task.transition_sequence + 1 + transition = LoopTransition( + sequence=sequence, + source=source, + target=target, + event=event, + reason=reason.strip() or event.value, + ) + task.loop_state = target.value + task.transition_reason = transition.reason + task.transition_sequence = sequence + task.status = _STATUS_BY_STATE[target].value + if target is LoopState.COMPLETED: + task.stop_reason = StopReason.GOAL_ACHIEVED.value + elif target is LoopState.STOPPED: + assert stop_reason is not None + task.stop_reason = stop_reason.value + elif target is LoopState.CANCELLED: + task.stop_reason = StopReason.CANCELLED.value + elif target is LoopState.FAILED: + task.stop_reason = StopReason.ERROR.value + else: + task.stop_reason = None + return transition diff --git a/apps/api/app/repositories/task.py b/apps/api/app/repositories/task.py index 7080fd7..52ead3c 100644 --- a/apps/api/app/repositories/task.py +++ b/apps/api/app/repositories/task.py @@ -4,11 +4,12 @@ import uuid -from sqlalchemy import func, select, update +from sqlalchemy import case, func, select, update from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql.elements import ColumnElement from app.db.models.task import TaskModel +from app.domain.loop import LoopState, LoopTransitionPolicy from app.repositories.base import BaseRepository @@ -44,17 +45,37 @@ async def get_by_idempotency_key(self, key: str, *, owner_id: str) -> TaskModel return result async def claim_pending(self, task_id: uuid.UUID) -> TaskModel | None: + claimable_states = tuple(state.value for state in LoopTransitionPolicy.claimable_states()) claimed = await self.session.scalar( update(TaskModel) - .where(TaskModel.id == task_id, TaskModel.status == "pending") - .values(status="running") + .where( + TaskModel.id == task_id, + TaskModel.status == "pending", + TaskModel.loop_state.in_(claimable_states), + ) + .values( + status="running", + loop_state=LoopState.PREPARING.value, + transition_reason=case( + ( + TaskModel.loop_state == LoopState.QUEUED.value, + "worker_claimed_task", + ), + else_="worker_recovered_interrupted_task", + ), + transition_sequence=TaskModel.transition_sequence + 1, + stop_reason=None, + ) .returning(TaskModel.id) ) if claimed is None: await self.session.rollback() return None await self.session.commit() - return await self.get(task_id) + task = await self.get(task_id) + if task is not None: + await self.session.refresh(task) + return task async def count_roots(self, *, owner_id: str | None = None) -> int: conditions: list[ColumnElement[bool]] = [TaskModel.parent_id.is_(None)] diff --git a/apps/api/app/schemas/task.py b/apps/api/app/schemas/task.py index 16f606d..0f002c7 100644 --- a/apps/api/app/schemas/task.py +++ b/apps/api/app/schemas/task.py @@ -14,6 +14,7 @@ from app.core.config import settings from app.domain.authority_token import AuthorityTokenError, normalize_hosts from app.domain.capability import CAPABILITY_SCHEMA_VERSION, Capability +from app.domain.loop import LoopState from app.domain.task import StopReason, TaskStatus from app.schemas.contract import ContractDraft from app.schemas.file import FileEntry @@ -184,6 +185,12 @@ class AuthorityRead(BaseModel): enforcement: AuthorityEnforcementRead +class LoopStateRead(BaseModel): + state: LoopState + transition_reason: str | None + sequence: int + + class TaskRead(BaseModel): model_config = ConfigDict(from_attributes=True) @@ -192,6 +199,7 @@ class TaskRead(BaseModel): owner_id: str project_id: str status: str + loop: LoopStateRead rubric: list[str] criteria_source: str verification_mode: str @@ -244,6 +252,11 @@ def from_model(cls, model: object) -> TaskRead: owner_id=m.owner_id, # type: ignore[attr-defined] project_id=m.project_id, # type: ignore[attr-defined] status=status, + loop=LoopStateRead( + state=m.loop_state, # type: ignore[attr-defined] + transition_reason=m.transition_reason, # type: ignore[attr-defined] + sequence=m.transition_sequence, # type: ignore[attr-defined] + ), rubric=m.rubric or [], # type: ignore[attr-defined] criteria_source=m.criteria_source, # type: ignore[attr-defined] verification_mode=m.verification_mode, # type: ignore[attr-defined] diff --git a/apps/api/app/services/agent_react.py b/apps/api/app/services/agent_react.py index f44448c..65b3bf2 100644 --- a/apps/api/app/services/agent_react.py +++ b/apps/api/app/services/agent_react.py @@ -21,8 +21,6 @@ import asyncio import contextlib -import json -import re import shutil import uuid from collections.abc import Awaitable, Callable @@ -64,7 +62,6 @@ discover_project_checks, mark_supplementary_agent_checks, merge_completion_checks, - regressions, ) from app.services.contract import ( compile_project_contract, @@ -73,8 +70,21 @@ verify_contract_hash, ) from app.services.ledger import genesis_hash, step_hash +from app.services.loop import ( + ActionDispatchPolicy, + ContextBudget, + DecisionParser, + DelegationPolicy, + DispatchKind, + HistoryWindow, + LoopEvent, + LoopState, + LoopTransitionPolicy, + VerificationPolicy, + extract_json, +) from app.services.memory import MemoryStore, scoped_memory_root -from app.services.progress import HistoryEntry, ProgressGuard, compact_history +from app.services.progress import HistoryEntry, ProgressGuard from app.services.prompts import plan_prompts, understand_prompts, verify_prompts from app.services.receipt import RECEIPT_SCHEMA, build_receipt, refresh_receipt_authority from app.services.skills import SkillStore @@ -99,9 +109,6 @@ log = get_logger("agent") -# How many recent steps the planner sees in full; older steps become a bounded state summary. -_HISTORY_WINDOW = 6 - _SIBYL_TOOLS = frozenset({"gather_bundle", "gather_sources", "quick_search", "read_url"}) _ARGUS_TOOLS = frozenset( { @@ -118,10 +125,6 @@ } ) -# Below this many tokens left, a spawn is refused rather than floored — flooring -# would let the sub-tree overshoot the parent's (and the global) token ceiling. -_MIN_SPAWN_BUDGET = 1_000 - # A reasoning model (deepseek-reasoner) spends the max_tokens budget on its chain of # thought BEFORE the answer, so a tight cap returns finish_reason=length with an EMPTY # content field — the decision/verdict is never emitted. That surfaced as intermittent @@ -133,78 +136,7 @@ _VERDICT_MAX_TOKENS = 1_500 # rubric + verify: reasoning then a short JSON verdict -def _verification_reserve(token_budget: int) -> int: - return min( - settings.agent_verification_token_reserve, - max(250, token_budget // 5), - max(0, token_budget // 2), - ) - - -def _balanced_json_objects(text: str) -> list[str]: - """Every balanced-brace ``{...}`` substring, in order, ignoring braces inside - string literals. A greedy ``\\{.*\\}`` regex fails on the two things reasoning - models do constantly — dict-like braces in prose (``a map {k: v}``) before the - real JSON, and emitting more than one object — so scan properly instead.""" - spans: list[str] = [] - depth, start = 0, -1 - in_str = esc = False - for i, ch in enumerate(text): - if in_str: - if esc: - esc = False - elif ch == "\\": - esc = True - elif ch == '"': - in_str = False - continue - if ch == '"': - in_str = True - elif ch == "{": - if depth == 0: - start = i - depth += 1 - elif ch == "}" and depth > 0: - depth -= 1 - if depth == 0 and start >= 0: - spans.append(text[start : i + 1]) - start = -1 - return spans - - -def _extract_json(text: str) -> Any: - """Best-effort: pull the model's JSON decision out of a reply that may be wrapped - in prose or chain-of-thought. Prefers the LAST parseable object — a reasoning - model reasons first and states its decision at the end.""" - cleaned = re.sub(r"```(?:json)?", "", text).strip() - try: - return json.loads(cleaned) - except json.JSONDecodeError: - pass - for span in reversed(_balanced_json_objects(cleaned)): - try: - obj = json.loads(span) - except json.JSONDecodeError: - continue - if isinstance(obj, dict): - return obj - return None - - -def _as_int(value: object, default: int) -> int: - if isinstance(value, (int, float, str)): - try: - return int(value) - except (TypeError, ValueError): - return default - return default - - -def _clamp_score(value: object) -> int: - try: - return max(0, min(100, int(float(value)))) # type: ignore[arg-type] - except (TypeError, ValueError): - return 0 +_extract_json = extract_json class AgentReactService: @@ -245,6 +177,30 @@ def __init__( self._active_authority_clients: list[tuple[str, Any]] = [] self._active_run_id: str | None = None self._authority_revoked = False + self._decision_parser = DecisionParser() + self._history_window = HistoryWindow() + self._dispatch_policy = ActionDispatchPolicy() + self._verification_policy = VerificationPolicy() + self._transition_policy = LoopTransitionPolicy() + self._delegation_policy = DelegationPolicy( + minimum_budget=1_000, + default_steps=settings.agent_max_steps_default, + max_steps=settings.agent_max_steps_cap, + ) + + def _transition( + self, + task: TaskModel, + event: LoopEvent, + reason: str, + *, + stop_reason: StopReason | None = None, + ) -> None: + self._transition_policy.apply(task, event, reason, stop_reason=stop_reason) + + def _fail_task(self, task: TaskModel, error: str, reason: str) -> None: + self._transition(task, LoopEvent.FAIL, reason) + task.error = error[:1000] async def run(self, task_id: uuid.UUID) -> None: if self._cancellation_probe is None: @@ -397,11 +353,11 @@ async def _run_claimed(self, task_id: uuid.UUID) -> None: operation = task.operation_journal tool = str(operation.get("tool", "unknown")) operation_id = str(operation.get("id", "unknown")) - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = ( + self._fail_task( + task, f"Interrupted with {tool!r} operation {operation_id} in flight. Its outcome " - "is unknown, so Loop refused to replay it and risk a duplicate mutation." + "is unknown, so Loop refused to replay it and risk a duplicate mutation.", + "operation_outcome_unknown", ) task.workspace_path = str(workspace.root) await self._rebuild_history(task.id) @@ -427,11 +383,11 @@ async def _run_claimed(self, task_id: uuid.UUID) -> None: store = SkillStore(Path(settings.agent_skills_root), settings.trust_public_key_pem()) skill = store.load(task.skill) if skill is None: - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = ( + self._fail_task( + task, f"Skill '{task.skill}' could not be loaded (unsigned, tampered, or " - "not found). Refusing to run." + "not found). Refusing to run.", + "skill_verification_failed", ) task.workspace_path = str(workspace.root) # so the refusal is auditable self._ensure_unverified_receipt(task) @@ -477,9 +433,7 @@ async def _run_claimed(self, task_id: uuid.UUID) -> None: else intersect_host_policies(requested_egress_hosts, skill_egress_hosts) ) except AuthorityTokenError as exc: - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = f"Invalid destination authority: {exc}" + self._fail_task(task, f"Invalid destination authority: {exc}", "invalid_authority") task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) await self._commit() @@ -515,9 +469,11 @@ async def _run_claimed(self, task_id: uuid.UUID) -> None: ): unavailable_host_mcp.append("Argus MCP") if unavailable_host_mcp: - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = f"Requested capability is unavailable: {', '.join(unavailable_host_mcp)}." + self._fail_task( + task, + f"Requested capability is unavailable: {', '.join(unavailable_host_mcp)}.", + "host_mcp_unavailable", + ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) await self._commit() @@ -527,9 +483,11 @@ async def _run_claimed(self, task_id: uuid.UUID) -> None: and resolved_capabilities & destination_capabilities and not envelope.egress_hosts ): - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = "Shell/browser network authority requires explicit egress_hosts." + self._fail_task( + task, + "Shell/browser network authority requires explicit egress_hosts.", + "egress_hosts_required", + ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) await self._commit() @@ -557,11 +515,11 @@ async def _run_claimed(self, task_id: uuid.UUID) -> None: if unavailable_gateways: gateway_names = " and ".join(unavailable_gateways) verb = "is" if len(unavailable_gateways) == 1 else "are" - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = ( + self._fail_task( + task, "Requested capabilities are disabled because no isolated " - f"{gateway_names} {verb} configured." + f"{gateway_names} {verb} configured.", + "isolated_gateway_unavailable", ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) @@ -590,9 +548,11 @@ def configured_provider_hosts(value: str) -> frozenset[str]: ), } except AuthorityTokenError as exc: - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = f"Invalid provider destination authority: {exc}" + self._fail_task( + task, + f"Invalid provider destination authority: {exc}", + "invalid_provider_authority", + ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) await self._commit() @@ -622,10 +582,11 @@ def configured_provider_hosts(value: str) -> frozenset[str]: if resolved_capabilities & capabilities and gateway_url and not hosts ] if missing_provider_hosts: - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = "Provider gateways require configured egress hosts: " + ", ".join( - missing_provider_hosts + self._fail_task( + task, + "Provider gateways require configured egress hosts: " + + ", ".join(missing_provider_hosts), + "provider_egress_hosts_required", ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) @@ -635,9 +596,7 @@ def configured_provider_hosts(value: str) -> frozenset[str]: try: sandbox_image, sandbox_label, sandbox_backend = self._resolve_sandbox() except RuntimeError as exc: - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = str(exc) + self._fail_task(task, str(exc), "sandbox_unavailable") task.sandbox = "unavailable" task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) @@ -651,11 +610,11 @@ def configured_provider_hosts(value: str) -> frozenset[str]: or not settings.agent_egress_proxy_url or not settings.agent_egress_proxy_audit_url ): - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = ( + self._fail_task( + task, "Shell network authority requires an isolated sandbox and " - "destination-enforcing egress proxy." + "destination-enforcing egress proxy.", + "shell_network_isolation_required", ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) @@ -678,9 +637,11 @@ def configured_provider_hosts(value: str) -> frozenset[str]: needs_authority_token = uses_gateway or Capability.NET_SHELL in resolved_capabilities authority_key = settings.authority_signing_key_pem() if needs_authority_token and not authority_key: - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = "Isolated network/provider execution requires an authority signing key." + self._fail_task( + task, + "Isolated network/provider execution requires an authority signing key.", + "authority_signing_key_missing", + ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) await self._commit() @@ -864,9 +825,11 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any try: await provider_gateway.start() except Exception as exc: - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = f"Isolated gateway is unavailable: {str(exc)[:300]}" + self._fail_task( + task, + f"Isolated gateway is unavailable: {str(exc)[:300]}", + "isolated_gateway_start_failed", + ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) await self._commit() @@ -894,9 +857,11 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any missing.append(Capability.NET_BROWSER.value) if missing: await provider_gateway.stop() - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = f"Isolated gateways lack requested capabilities: {', '.join(missing)}" + self._fail_task( + task, + f"Isolated gateways lack requested capabilities: {', '.join(missing)}", + "isolated_gateway_capability_missing", + ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) await self._commit() @@ -911,9 +876,11 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any elif envelope.permits_capability(Capability.NET_BROWSER): if provider_gateway is not None: await provider_gateway.stop() - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = "The requested browser capability is unavailable." + self._fail_task( + task, + "The requested browser capability is unavailable.", + "browser_unavailable", + ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) await self._commit() @@ -936,9 +903,11 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any await browser.stop() if provider_gateway is not None: await provider_gateway.stop() - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = "The requested email capability is not configured." + self._fail_task( + task, + "The requested email capability is not configured.", + "email_unavailable", + ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) await self._commit() @@ -961,9 +930,11 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any await browser.stop() if provider_gateway is not None: await provider_gateway.stop() - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = "The requested calendar capability is not configured." + self._fail_task( + task, + "The requested calendar capability is not configured.", + "calendar_unavailable", + ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) await self._commit() @@ -994,9 +965,11 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any await browser.stop() if provider_gateway is not None: await provider_gateway.stop() - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = "The requested vision capability is not configured." + self._fail_task( + task, + "The requested vision capability is not configured.", + "vision_unavailable", + ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) await self._commit() @@ -1010,9 +983,11 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any await browser.stop() if provider_gateway is not None: await provider_gateway.stop() - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = f"Requested MCP provider is unavailable: {str(exc)[:300]}" + self._fail_task( + task, + f"Requested MCP provider is unavailable: {str(exc)[:300]}", + "host_mcp_start_failed", + ) task.workspace_path = str(workspace.root) self._ensure_unverified_receipt(task) await self._commit() @@ -1032,7 +1007,6 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any "production must keep host providers disabled.\n" ) - task.status = TaskStatus.RUNNING.value task.workspace_path = str(workspace.root) # Rebuild the working memory from whatever has already happened so a # resumed run sees its own past actions (and the user's answer). @@ -1070,9 +1044,7 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any raise except Exception as exc: # any unhandled error fails the task cleanly log.exception("agent.failed", task_id=str(task.id)) - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = str(exc)[:1000] + self._fail_task(task, str(exc), "unhandled_runtime_error") self._ensure_unverified_receipt(task) # a crash is auditable too await self._commit() finally: @@ -1109,9 +1081,11 @@ async def _prepare_project_contract( or not draft.critique.accepted or not verify_contract_hash(draft, task.contract_hash) ): - task.status = TaskStatus.FAILED.value - task.stop_reason = StopReason.ERROR.value - task.error = "The locked acceptance contract failed its content hash check." + self._fail_task( + task, + "The locked acceptance contract failed its content hash check.", + "contract_hash_invalid", + ) self._ensure_unverified_receipt(task) await self._commit() return False @@ -1135,10 +1109,13 @@ async def _prepare_project_contract( clarifications = [ str(item) for item in previous.get("clarifications", []) if str(item).strip() ] - compilation_budget = max( - 0, - task.token_budget - task.tokens_used - _verification_reserve(task.token_budget), - ) + compilation_budget = ContextBudget.allocate( + task.token_budget, + task.tokens_used, + settings.agent_verification_token_reserve, + ).planning + self._transition(task, LoopEvent.CONTRACT_REQUIRED, "contract_compilation_started") + await self._commit() try: compiled = await compile_project_contract( goal=task.goal, @@ -1162,7 +1139,7 @@ async def _prepare_project_contract( task.contract_hash = None task.contract_status = "awaiting_input" task.pending_question = draft.critique.question - task.status = TaskStatus.AWAITING_INPUT.value + self._transition(task, LoopEvent.INPUT_REQUIRED, "contract_compilation_failed") await self._commit() return False @@ -1172,7 +1149,7 @@ async def _prepare_project_contract( if compiled.contract_hash is None: task.contract_status = "awaiting_input" task.pending_question = compiled.draft.critique.question - task.status = TaskStatus.AWAITING_INPUT.value + self._transition(task, LoopEvent.INPUT_REQUIRED, "contract_clarification_required") await self._commit() return False @@ -1180,6 +1157,7 @@ async def _prepare_project_contract( task.criteria_source = "compiled" task.pending_question = None self._apply_locked_contract(task, compiled.draft) + self._transition(task, LoopEvent.CONTRACT_READY, "contract_locked") await self._commit() return True @@ -1315,9 +1293,16 @@ async def _start_host_mcp(self, envelope: CapabilityEnvelope) -> McpPool: async def _run_loop( self, task: TaskModel, workspace: Workspace, executor: ToolExecutor, *, start: int ) -> None: - verification_reserve = _verification_reserve(task.token_budget) + context_budget = ContextBudget.allocate( + task.token_budget, + task.tokens_used, + settings.agent_verification_token_reserve, + ) + verification_reserve = context_budget.verification_reserve if not task.rubric: # only on a fresh run, not a resume - understand_budget = max(0, task.token_budget - task.tokens_used - verification_reserve) + self._transition(task, LoopEvent.RUBRIC_REQUIRED, "rubric_generation_started") + await self._commit() + understand_budget = context_budget.planning try: rubric, understand_result = await self._understand(task.goal, understand_budget) tokens = understand_result.tokens @@ -1330,13 +1315,23 @@ async def _run_loop( rubric, tokens = ["Fully and correctly satisfies the task"], exc.tokens_spent task.rubric = rubric task.tokens_used += tokens + self._transition(task, LoopEvent.RUBRIC_READY, "rubric_ready") await self._commit() + else: + if LoopState(task.loop_state) is LoopState.PREPARING: + self._transition(task, LoopEvent.PREPARATION_COMPLETED, "runtime_prepared") + await self._commit() # Resuming from an approved action: run it now as this step, then continue. if task.pending_action is not None: action = dict(task.pending_action) task.pending_action = None approved_tool = str(action["tool"]) + self._transition( + task, + LoopEvent.ACTION_SELECTED, + f"approved_action_selected:{approved_tool}", + ) args = await self._begin_operation( task, number=start, @@ -1346,6 +1341,7 @@ async def _run_loop( tokens=0, ) result = await executor.execute(approved_tool, args) + self._transition(task, LoopEvent.ACTION_RECORDED, f"action_recorded:{approved_tool}") await self._record_step( task, start, @@ -1389,8 +1385,12 @@ async def _run_loop( await self._finish(task, StopReason.BUDGET_EXHAUSTED) return - tokens_left = max(0, task.token_budget - task.tokens_used) - planning_budget = max(0, tokens_left - verification_reserve) + context_budget = ContextBudget.allocate( + task.token_budget, + task.tokens_used, + settings.agent_verification_token_reserve, + ) + planning_budget = context_budget.planning if planning_budget <= 0: await self._finish(task, StopReason.BUDGET_EXHAUSTED) return @@ -1435,7 +1435,16 @@ async def _run_loop( raise self._record_model_use(task, decision) step_tokens = decision.tokens - thought, tool, args = self._parse_decision(_extract_json(decision.content)) + parsed_decision = self._decision_parser.parse( + decision.content, + valid_tools=VALID_TOOLS, + dynamic_tools=self._mcp_tools, + ) + thought, tool, args = ( + parsed_decision.thought, + parsed_decision.tool, + parsed_decision.args, + ) if tool == "finish": finish_marker = workspace.state_marker() @@ -1454,11 +1463,18 @@ async def _run_loop( ) await self._finish(task, StopReason.STUCK) return + self._transition(task, LoopEvent.VERIFICATION_REQUESTED, "finish_requested") accepted, score, summary, _ = await self._handle_finish( task, workspace, args, thought, number, step_tokens ) if accepted: return + self._transition( + task, + LoopEvent.VERIFICATION_REJECTED, + f"verification_rejected:score={score}", + ) + await self._commit() last_rejected_finish_marker = finish_marker finish_retries += 1 if finish_retries > settings.agent_max_finish_retries: @@ -1468,11 +1484,17 @@ async def _run_loop( return continue + self._transition( + task, + LoopEvent.ACTION_SELECTED, + f"action_selected:{tool or 'invalid'}", + ) if tool == "ask_user": await self._pause_for_user(task, args, thought, number, step_tokens) return # the run resumes when the user answers if tool in {"remember", "spawn"} and not executor.envelope.permits(tool): + self._transition(task, LoopEvent.ACTION_RECORDED, f"action_blocked:{tool}") await self._record_step( task, number, @@ -1507,6 +1529,7 @@ async def _run_loop( observation = self.memory.remember(note, str(topic) if topic else None) if note: # make it visible to the rest of this run too self._memory_snapshot = f"{self._memory_snapshot}\n- {note}".strip() + self._transition(task, LoopEvent.ACTION_RECORDED, "action_recorded:remember") await self._record_step( task, number, thought, "remember", args, observation, ToolStatus.OK, step_tokens ) @@ -1523,6 +1546,7 @@ async def _run_loop( if tool == "spawn": guard_block = guard.preflight(tool, args) if guard_block: + self._transition(task, LoopEvent.ACTION_RECORDED, "action_blocked:spawn") await self._record_step( task, number, @@ -1601,38 +1625,36 @@ async def _run_loop( guard_block = guard.preflight(tool, args) if tool is not None else None workspace_before = workspace.state_marker() - - if tool is None: - observation, status = ( - "Could not parse a valid action. Respond with one JSON object " - f"using a valid tool: {sorted(VALID_TOOLS)}.", - ToolStatus.ERROR, - ) - elif guard_block: - observation, status = guard_block, ToolStatus.BLOCKED - elif tool in ("write_file", "edit_file") and same_path_writes >= 3: - observation, status = ( - f"Blocked: you have written '{last_write_path}' {same_path_writes} times " - "without running it. Writing it again is not allowed — run it with " - "run_command, call finish with checks, or take a different action.", - ToolStatus.BLOCKED, - ) - elif tool == "run_command" and approval_required: + approval_reason: str | None = None + if tool == "run_command" and approval_required: verdict, reason = evaluate_command(str(args.get("command", ""))) if verdict is Verdict.NEEDS_APPROVAL: - await self._pause_for_approval(task, args, thought, number, step_tokens, reason) - return # resumes when the user approves or denies - args = await self._begin_operation( + approval_reason = reason + route = self._dispatch_policy.route( + tool, + valid_tools=VALID_TOOLS, + guard_block=guard_block, + repeated_write_count=same_path_writes, + approval_reason=approval_reason, + last_write_path=last_write_path, + ) + if route.kind is DispatchKind.INVALID: + observation, status = route.observation or "Invalid action.", ToolStatus.ERROR + elif route.kind is DispatchKind.BLOCKED: + observation, status = route.observation or "Action blocked.", ToolStatus.BLOCKED + elif route.kind is DispatchKind.APPROVAL: + assert tool is not None and route.approval_reason is not None + await self._pause_for_approval( task, - number=number, - thought=thought, - tool=tool, - args=args, - tokens=step_tokens, + args, + thought, + number, + step_tokens, + route.approval_reason, ) - tool_result = await executor.execute(tool, args) - observation, status = tool_result.observation, tool_result.status + return else: + assert tool is not None args = await self._begin_operation( task, number=number, @@ -1649,6 +1671,11 @@ async def _run_loop( "do not rewrite it again.]" ) + self._transition( + task, + LoopEvent.ACTION_RECORDED, + f"action_recorded:{tool or 'invalid'}:{status.value}", + ) await self._record_step( task, number, thought, tool or "invalid", args, observation, status, step_tokens ) @@ -1681,6 +1708,11 @@ async def _run_loop( "Loop submitted the workspace after every user-required check passed." ) } + self._transition( + task, + LoopEvent.VERIFICATION_REQUESTED, + "contract_evidence_ready", + ) accepted, score, summary, _ = await self._handle_finish( task, workspace, @@ -1691,6 +1723,12 @@ async def _run_loop( ) if accepted: return + self._transition( + task, + LoopEvent.VERIFICATION_REJECTED, + f"verification_rejected:score={score}", + ) + await self._commit() skipped_step_number = auto_number last_rejected_finish_marker = workspace.state_marker() finish_retries += 1 @@ -1735,7 +1773,7 @@ async def _pause_for_user( tokens, ) task.pending_question = question - task.status = TaskStatus.AWAITING_INPUT.value + self._transition(task, LoopEvent.INPUT_REQUIRED, "agent_requested_input") await self._commit() log.info("agent.awaiting_input", task_id=str(task.id), number=number) @@ -1784,7 +1822,7 @@ async def _pause_for_action( ) task.pending_action = {"tool": tool, "args": args} task.pending_question = f"Approve this action? Answer yes or no.\n {summary}" - task.status = TaskStatus.AWAITING_INPUT.value + self._transition(task, LoopEvent.APPROVAL_REQUIRED, f"approval_required:{tool}") await self._commit() log.info("agent.awaiting_approval", task_id=str(task.id), number=number, tool=tool) @@ -1795,6 +1833,7 @@ async def _handle_spawn( sandboxed loop under a sub-budget, and its result + output files come back as this step's observation. The child's tokens count against this task.""" if task.depth >= settings.agent_max_spawn_depth: + self._transition(task, LoopEvent.ACTION_RECORDED, "action_blocked:spawn_depth") await self._record_step( task, number, @@ -1809,6 +1848,7 @@ async def _handle_spawn( return goal = str(args.get("goal", "")).strip() if len(goal) < 4: + self._transition(task, LoopEvent.ACTION_RECORDED, "action_failed:spawn_goal") await self._record_step( task, number, @@ -1822,9 +1862,11 @@ async def _handle_spawn( return remaining = max(0, task.token_budget - task.tokens_used - plan_tokens) - if remaining < _MIN_SPAWN_BUDGET: + allocation = self._delegation_policy.allocate(args, remaining) + if allocation is None: # Flooring the child at 1000 here would let the sub-tree overshoot the # parent's (and the global) token ceiling. Refuse instead. + self._transition(task, LoopEvent.ACTION_RECORDED, "action_blocked:spawn_budget") await self._record_step( task, number, @@ -1837,15 +1879,8 @@ async def _handle_spawn( plan_tokens, ) return - # Never exceed what the parent actually has left, so the ceiling holds. - child_budget = max(1, min(_as_int(args.get("token_budget"), remaining), remaining)) - child_steps = max( - 1, - min( - _as_int(args.get("max_steps"), settings.agent_max_steps_default), - settings.agent_max_steps_cap, - ), - ) + child_budget = allocation.token_budget + child_steps = allocation.max_steps allowed = args.get("allowed_tools") raw_capabilities = args.get("capabilities") if isinstance(raw_capabilities, list): @@ -1878,6 +1913,7 @@ async def _handle_spawn( if any(host == allowed or host.endswith(f".{allowed}") for allowed in parent_hosts) ) if child_capabilities & {Capability.NET_SHELL, Capability.NET_BROWSER} and not child_hosts: + self._transition(task, LoopEvent.ACTION_RECORDED, "action_blocked:spawn_egress") await self._record_step( task, number, @@ -1944,6 +1980,7 @@ async def _handle_spawn( f"stop={child.stop_reason}, verified_by={child.verified_by}, " f"score={child.verification_score}.\nSummary: {child.summary or '(none)'}.{files_line}" ) + self._transition(task, LoopEvent.ACTION_RECORDED, "action_recorded:spawn") await self._record_step( task, number, @@ -2073,9 +2110,6 @@ async def _handle_finish( contract_substantiation_authoritative = mark_supplementary_agent_checks( check_results, len(task.rubric or []) ) - gates_passed = completion_gates_pass(check_results) - coverage_complete = execution_coverage_complete(check_results, len(task.rubric or [])) - system, user = verify_prompts( task.goal, task.rubric, @@ -2085,7 +2119,11 @@ async def _handle_finish( workspace.contents_digest(), today=date.today().isoformat(), ) - verify_budget = max(0, task.token_budget - task.tokens_used - plan_tokens) + verify_budget = ContextBudget.allocate( + task.token_budget, + task.tokens_used, + settings.agent_verification_token_reserve, + ).verification_after(plan_tokens) try: result = await self.verifier_llm.complete( system, @@ -2111,40 +2149,19 @@ async def _handle_finish( return True, 0, summary, exc.tokens_spent raise self._record_model_use(task, result, verifier=True) - parsed = _extract_json(result.content) - if isinstance(parsed, dict): - score = _clamp_score(parsed.get("score")) - raw_missing = parsed.get("missing") or [] - missing = [str(item) for item in raw_missing] if isinstance(raw_missing, list) else [] - llm_met = bool(parsed.get("met")) - # Strict parse: omit -> True (prior behavior), but the stronger label - # must be EARNED, so a string "false"/"no" or a null doesn't slip through - # bool() as truthy. Only an explicit true earns it. - _sub = parsed.get("checks_substantiate", True) - substantiate = _sub is True or ( - isinstance(_sub, str) and _sub.strip().lower() in {"true", "yes", "1"} - ) - else: - score, missing, llm_met, substantiate = 0, ["verifier returned no verdict"], False, True - - execution_ready = bool( - check_results - and coverage_complete - and gates_passed - and (substantiate or contract_substantiation_authoritative) + decision = self._verification_policy.evaluate( + _extract_json(result.content), + check_results, + criterion_count=len(task.rubric or []), + acceptance_score=settings.agent_acceptance_score, + strict=task.verification_mode == "strict", + contract_substantiation_authoritative=contract_substantiation_authoritative, ) - verified_by = "execution" if execution_ready else "judgment" - met = llm_met and score >= settings.agent_acceptance_score and gates_passed - if task.verification_mode == "strict" and not execution_ready: - met = False - if not coverage_complete: - missing.append("Every success criterion needs a mapped execution check.") - if not gates_passed: - missing.append("A required check failed or a project quality gate regressed.") - if check_results and not substantiate and not contract_substantiation_authoritative: - missing.append("The proposed checks do not substantiate the task goal.") - for regression in regressions(check_results): - missing.append(f"Regression in {regression.target}.") + score = decision.score + missing = list(decision.missing) + met = decision.accepted + verified_by = decision.verified_by + coverage_complete = decision.coverage_complete verdict = ( f"verifier: score {score}, met={met}, verified_by={verified_by}, " @@ -2226,18 +2243,6 @@ def _record_model_use(task: TaskModel, result: LLMResult, *, verifier: bool = Fa if identity not in used: task.executor_models = [*used, identity] - def _parse_decision(self, decision: Any) -> tuple[str, str | None, dict[str, Any]]: - if not isinstance(decision, dict): - return "", None, {} - thought = str(decision.get("thought", "")).strip() - tool = decision.get("tool") - args = decision.get("args") - if not isinstance(args, dict): - args = {} - if tool not in VALID_TOOLS and tool not in self._mcp_tools: - return thought, None, {} - return thought, str(tool), args - async def _record_step( self, task: TaskModel, @@ -2320,29 +2325,7 @@ async def _rebuild_history(self, task_id: uuid.UUID) -> None: self._last_hash = steps[-1].hash if steps else genesis_hash(task_id) def _history_view(self) -> str: - """The history the planner sees: recent steps in full, older ones - compacted into durable decisions/evidence so a long run stays bounded - without forgetting failed branches.""" - if len(self._history) <= _HISTORY_WINDOW: - rendered = [entry.render() for entry in self._history[:-1]] - if self._history: - rendered.append(self._render_latest_history(self._history[-1])) - return "\n".join(rendered) or "(nothing yet)" - older = self._history[:-_HISTORY_WINDOW] - recent = self._history[-_HISTORY_WINDOW:] - return ( - compact_history(older) - + "\n\n[RECENT STEPS]\n" - + "\n".join( - self._render_latest_history(entry) if index == len(recent) - 1 else entry.render() - for index, entry in enumerate(recent) - ) - ) - - @staticmethod - def _render_latest_history(entry: HistoryEntry) -> str: - limit = 2_400 if entry.tool in {"run_command", "read_file"} else None - return entry.render(observation_limit=limit) + return self._history_window.render(self._history) @staticmethod def _required_checks_view(checks: list[dict[str, Any]]) -> str: @@ -2381,12 +2364,15 @@ def _stop_summary(task: TaskModel, reason: StopReason) -> str: return "Stopped without a verified result." async def _finish(self, task: TaskModel, reason: StopReason) -> None: - task.status = ( - TaskStatus.COMPLETED.value - if reason is StopReason.GOAL_ACHIEVED - else TaskStatus.STOPPED.value - ) - task.stop_reason = reason.value + if reason is StopReason.GOAL_ACHIEVED: + self._transition(task, LoopEvent.VERIFICATION_ACCEPTED, "verification_accepted") + else: + self._transition( + task, + LoopEvent.LIMIT_REACHED, + f"bounded_stop:{reason.value}", + stop_reason=reason, + ) # The accepted path sets its own summary; for the other stops the agent # never wrote one, so give the user a reason-specific explanation. if not task.summary and reason is not StopReason.GOAL_ACHIEVED: diff --git a/apps/api/app/services/loop/__init__.py b/apps/api/app/services/loop/__init__.py new file mode 100644 index 0000000..76f8887 --- /dev/null +++ b/apps/api/app/services/loop/__init__.py @@ -0,0 +1,34 @@ +"""Explicit policies used by the production agent loop.""" + +from app.domain.loop import ( + InvalidLoopTransitionError, + LoopEvent, + LoopState, + LoopTransition, + LoopTransitionPolicy, +) +from app.services.loop.context import ContextBudget, HistoryWindow +from app.services.loop.decisions import Decision, DecisionParser, extract_json +from app.services.loop.delegation import DelegationAllocation, DelegationPolicy +from app.services.loop.dispatch import ActionDispatchPolicy, DispatchKind, DispatchRoute +from app.services.loop.verification import VerificationDecision, VerificationPolicy + +__all__ = [ + "ActionDispatchPolicy", + "ContextBudget", + "Decision", + "DecisionParser", + "DelegationAllocation", + "DelegationPolicy", + "DispatchKind", + "DispatchRoute", + "HistoryWindow", + "InvalidLoopTransitionError", + "LoopEvent", + "LoopState", + "LoopTransition", + "LoopTransitionPolicy", + "VerificationDecision", + "VerificationPolicy", + "extract_json", +] diff --git a/apps/api/app/services/loop/context.py b/apps/api/app/services/loop/context.py new file mode 100644 index 0000000..b7043c9 --- /dev/null +++ b/apps/api/app/services/loop/context.py @@ -0,0 +1,60 @@ +"""Token allocation and bounded planner history.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from app.services.progress import HistoryEntry, compact_history + + +@dataclass(frozen=True, slots=True) +class ContextBudget: + total: int + used: int + verification_reserve: int + + @classmethod + def allocate(cls, total: int, used: int, reserve_cap: int) -> ContextBudget: + reserve = min(reserve_cap, max(250, total // 5), max(0, total // 2)) + return cls(total=total, used=used, verification_reserve=reserve) + + @property + def remaining(self) -> int: + return max(0, self.total - self.used) + + @property + def planning(self) -> int: + return max(0, self.remaining - self.verification_reserve) + + def verification_after(self, planning_tokens: int) -> int: + return max(0, self.remaining - planning_tokens) + + def delegation_after(self, planning_tokens: int) -> int: + return max(0, self.remaining - planning_tokens) + + +@dataclass(frozen=True, slots=True) +class HistoryWindow: + recent_steps: int = 6 + + def render(self, entries: list[HistoryEntry]) -> str: + if len(entries) <= self.recent_steps: + rendered = [entry.render() for entry in entries[:-1]] + if entries: + rendered.append(self._render_latest(entries[-1])) + return "\n".join(rendered) or "(nothing yet)" + older = entries[: -self.recent_steps] + recent = entries[-self.recent_steps :] + return ( + compact_history(older) + + "\n\n[RECENT STEPS]\n" + + "\n".join( + self._render_latest(entry) if index == len(recent) - 1 else entry.render() + for index, entry in enumerate(recent) + ) + ) + + @staticmethod + def _render_latest(entry: HistoryEntry) -> str: + limit = 2_400 if entry.tool in {"run_command", "read_file"} else None + return entry.render(observation_limit=limit) diff --git a/apps/api/app/services/loop/decisions.py b/apps/api/app/services/loop/decisions.py new file mode 100644 index 0000000..d401335 --- /dev/null +++ b/apps/api/app/services/loop/decisions.py @@ -0,0 +1,78 @@ +"""Parse one model response into a bounded loop decision.""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass +from typing import Any + + +def _balanced_json_objects(text: str) -> list[str]: + spans: list[str] = [] + depth, start = 0, -1 + in_string = escaped = False + for index, character in enumerate(text): + if in_string: + if escaped: + escaped = False + elif character == "\\": + escaped = True + elif character == '"': + in_string = False + continue + if character == '"': + in_string = True + elif character == "{": + if depth == 0: + start = index + depth += 1 + elif character == "}" and depth > 0: + depth -= 1 + if depth == 0 and start >= 0: + spans.append(text[start : index + 1]) + start = -1 + return spans + + +def extract_json(text: str) -> Any: + cleaned = re.sub(r"```(?:json)?", "", text).strip() + try: + return json.loads(cleaned) + except json.JSONDecodeError: + pass + for span in reversed(_balanced_json_objects(cleaned)): + try: + parsed = json.loads(span) + except json.JSONDecodeError: + continue + if isinstance(parsed, dict): + return parsed + return None + + +@dataclass(frozen=True, slots=True) +class Decision: + thought: str + tool: str | None + args: dict[str, Any] + + +class DecisionParser: + def parse( + self, + content: str, + *, + valid_tools: set[str] | frozenset[str], + dynamic_tools: set[str] | frozenset[str] = frozenset(), + ) -> Decision: + payload = extract_json(content) + if not isinstance(payload, dict): + return Decision("", None, {}) + thought = str(payload.get("thought", "")).strip() + tool = payload.get("tool") + args = payload.get("args") + parsed_args = dict(args) if isinstance(args, dict) else {} + if not isinstance(tool, str) or tool not in valid_tools | dynamic_tools: + return Decision(thought, None, {}) + return Decision(thought, tool, parsed_args) diff --git a/apps/api/app/services/loop/delegation.py b/apps/api/app/services/loop/delegation.py new file mode 100644 index 0000000..cefeb3b --- /dev/null +++ b/apps/api/app/services/loop/delegation.py @@ -0,0 +1,38 @@ +"""Bound delegation so a child cannot overspend its parent.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +def _as_int(value: object, default: int) -> int: + if isinstance(value, (int, float, str)): + try: + return int(value) + except (TypeError, ValueError): + return default + return default + + +@dataclass(frozen=True, slots=True) +class DelegationAllocation: + token_budget: int + max_steps: int + + +class DelegationPolicy: + def __init__(self, *, minimum_budget: int, default_steps: int, max_steps: int) -> None: + self.minimum_budget = minimum_budget + self.default_steps = default_steps + self.max_steps = max_steps + + def allocate(self, args: dict[str, Any], remaining: int) -> DelegationAllocation | None: + if remaining < self.minimum_budget: + return None + token_budget = max(1, min(_as_int(args.get("token_budget"), remaining), remaining)) + max_steps = max( + 1, + min(_as_int(args.get("max_steps"), self.default_steps), self.max_steps), + ) + return DelegationAllocation(token_budget=token_budget, max_steps=max_steps) diff --git a/apps/api/app/services/loop/dispatch.py b/apps/api/app/services/loop/dispatch.py new file mode 100644 index 0000000..f98e51e --- /dev/null +++ b/apps/api/app/services/loop/dispatch.py @@ -0,0 +1,51 @@ +"""Route a parsed decision without executing side effects.""" + +from __future__ import annotations + +import enum +from dataclasses import dataclass + + +class DispatchKind(enum.StrEnum): + INVALID = "invalid" + BLOCKED = "blocked" + APPROVAL = "approval" + EXECUTE = "execute" + + +@dataclass(frozen=True, slots=True) +class DispatchRoute: + kind: DispatchKind + observation: str | None = None + approval_reason: str | None = None + + +class ActionDispatchPolicy: + def route( + self, + tool: str | None, + *, + valid_tools: set[str] | frozenset[str], + guard_block: str | None, + repeated_write_count: int, + approval_reason: str | None, + last_write_path: str | None, + ) -> DispatchRoute: + if tool is None: + return DispatchRoute( + DispatchKind.INVALID, + "Could not parse a valid action. Respond with one JSON object using a valid " + f"tool: {sorted(valid_tools)}.", + ) + if guard_block: + return DispatchRoute(DispatchKind.BLOCKED, guard_block) + if tool in {"write_file", "edit_file"} and repeated_write_count >= 3: + return DispatchRoute( + DispatchKind.BLOCKED, + f"Blocked: you have written '{last_write_path}' {repeated_write_count} times " + "without running it. Writing it again is not allowed — run it with " + "run_command, call finish with checks, or take a different action.", + ) + if approval_reason is not None: + return DispatchRoute(DispatchKind.APPROVAL, approval_reason=approval_reason) + return DispatchRoute(DispatchKind.EXECUTE) diff --git a/apps/api/app/services/loop/verification.py b/apps/api/app/services/loop/verification.py new file mode 100644 index 0000000..1f03bd1 --- /dev/null +++ b/apps/api/app/services/loop/verification.py @@ -0,0 +1,83 @@ +"""Turn verifier output and execution evidence into one acceptance decision.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from app.services.completion import completion_gates_pass, regressions +from app.services.verification import CheckResult, execution_coverage_complete + + +def _score(value: object) -> int: + try: + return max(0, min(100, int(float(value)))) # type: ignore[arg-type] + except (TypeError, ValueError): + return 0 + + +@dataclass(frozen=True, slots=True) +class VerificationDecision: + accepted: bool + score: int + missing: tuple[str, ...] + verified_by: str + coverage_complete: bool + gates_passed: bool + + +class VerificationPolicy: + def evaluate( + self, + payload: Any, + results: list[CheckResult], + *, + criterion_count: int, + acceptance_score: int, + strict: bool, + contract_substantiation_authoritative: bool, + ) -> VerificationDecision: + if isinstance(payload, dict): + score = _score(payload.get("score")) + raw_missing = payload.get("missing") or [] + missing = [str(item) for item in raw_missing] if isinstance(raw_missing, list) else [] + llm_met = bool(payload.get("met")) + raw_substantiation = payload.get("checks_substantiate", True) + substantiates = raw_substantiation is True or ( + isinstance(raw_substantiation, str) + and raw_substantiation.strip().lower() in {"true", "yes", "1"} + ) + else: + score = 0 + missing = ["verifier returned no verdict"] + llm_met = False + substantiates = True + + gates_passed = completion_gates_pass(results) + coverage_complete = execution_coverage_complete(results, criterion_count) + execution_ready = bool( + results + and coverage_complete + and gates_passed + and (substantiates or contract_substantiation_authoritative) + ) + verified_by = "execution" if execution_ready else "judgment" + accepted = llm_met and score >= acceptance_score and gates_passed + if strict and not execution_ready: + accepted = False + if not coverage_complete: + missing.append("Every success criterion needs a mapped execution check.") + if not gates_passed: + missing.append("A required check failed or a project quality gate regressed.") + if results and not substantiates and not contract_substantiation_authoritative: + missing.append("The proposed checks do not substantiate the task goal.") + for regression in regressions(results): + missing.append(f"Regression in {regression.target}.") + return VerificationDecision( + accepted=accepted, + score=score, + missing=tuple(missing), + verified_by=verified_by, + coverage_complete=coverage_complete, + gates_passed=gates_passed, + ) diff --git a/apps/api/app/services/runner.py b/apps/api/app/services/runner.py index 5917a93..3c4b067 100644 --- a/apps/api/app/services/runner.py +++ b/apps/api/app/services/runner.py @@ -24,6 +24,7 @@ from app.core.llm import get_llm_client, get_verifier_client from app.core.logging import get_logger from app.db.session import get_sessionmaker +from app.domain.loop import LoopState from app.domain.task import StopReason, TaskStatus from app.observability.metrics import TASK_RUN_DURATION, TASK_RUNS from app.repositories.step import StepRepository @@ -129,12 +130,19 @@ async def reconcile_interrupted_tasks( values: dict[str, object] = ( { "status": TaskStatus.PENDING.value, + "loop_state": LoopState.QUEUED.value, + "transition_reason": "worker_requeued_interrupted_task", + "transition_sequence": TaskModel.transition_sequence + 1, + "stop_reason": None, "error": "Interrupted — recovered by the durable worker queue.", "attempt": TaskModel.attempt + 1, } if requeue else { "status": TaskStatus.FAILED.value, + "loop_state": LoopState.FAILED.value, + "transition_reason": "runner_interrupted_task", + "transition_sequence": TaskModel.transition_sequence + 1, "stop_reason": StopReason.ERROR.value, "error": "Interrupted — the runner crashed or restarted mid-run.", } diff --git a/apps/api/app/services/task.py b/apps/api/app/services/task.py index 2f4a90d..2894413 100644 --- a/apps/api/app/services/task.py +++ b/apps/api/app/services/task.py @@ -39,6 +39,7 @@ save_patch, ) from app.services.ledger import genesis_hash, step_hash, verify_chain +from app.services.loop import LoopEvent, LoopTransitionPolicy from app.tools.base import ToolError from app.tools.workspace import Workspace @@ -76,6 +77,7 @@ def __init__( self.tasks = tasks self.steps = steps self.subject = subject + self._transition_policy = LoopTransitionPolicy() def _resolve_limits(self, limits: LimitsIn) -> tuple[int, int]: """Apply defaults for omitted fields, then clamp to the hard caps so no @@ -721,10 +723,14 @@ async def cancel(self, task_id: uuid.UUID) -> TaskModel: ) if task.status not in active: raise ConflictError(f"Task is {task.status} and cannot be cancelled") - task.status = TaskStatus.CANCELLED.value + self._transition_policy.apply(task, LoopEvent.CANCEL, "cancelled_by_user") for descendant in await self.tasks.list_descendants(task.id): if descendant.status in active: - descendant.status = TaskStatus.CANCELLED.value + self._transition_policy.apply( + descendant, + LoopEvent.CANCEL, + "cancelled_with_parent", + ) await self.tasks.session.flush() await self.tasks.session.refresh(task) return task @@ -775,7 +781,7 @@ async def respond(self, task_id: uuid.UUID, answer: str) -> TaskModel: ) task.pending_question = None - task.status = TaskStatus.PENDING.value # pending == ready to (re)run + self._transition_policy.apply(task, LoopEvent.USER_RESPONDED, "user_response_recorded") # flush+refresh pulls the server-side onupdate ``updated_at`` before it is # serialized (otherwise it lazy-loads in a sync context and 500s). Commit # before the caller schedules the resume so the agent's own session sees diff --git a/apps/api/tests/test_agent_react.py b/apps/api/tests/test_agent_react.py index 848d529..ff7ed2e 100644 --- a/apps/api/tests/test_agent_react.py +++ b/apps/api/tests/test_agent_react.py @@ -20,6 +20,7 @@ from app.repositories.step import StepRepository from app.repositories.task import TaskRepository from app.services.agent_react import AgentReactService, _extract_json +from app.services.loop import LoopState from app.services.progress import HistoryEntry, ProgressGuard, compact_history from app.services.task import TaskService from app.tools import ToolStatus, Workspace @@ -170,6 +171,8 @@ async def test_goal_achieved_when_verifier_accepts_finish(session: AsyncSession) await session.refresh(task) assert task.status == TaskStatus.COMPLETED.value + assert task.loop_state == LoopState.COMPLETED.value + assert task.transition_reason == "verification_accepted" assert task.stop_reason == StopReason.GOAL_ACHIEVED.value assert task.verification_score == 92 assert task.summary == "wrote result.txt" @@ -192,7 +195,7 @@ async def test_running_shell_is_killed_when_task_is_cancelled( async def cancellation_probe(_task_id: object) -> bool: if not requested: return False - task.status = TaskStatus.CANCELLED.value + await TaskService(TaskRepository(session), StepRepository(session)).cancel(task.id) await session.commit() return True @@ -226,6 +229,7 @@ async def cancellation_probe(_task_id: object) -> bool: os.kill(child_pid, 0) await session.refresh(task) assert task.status == TaskStatus.CANCELLED.value + assert task.loop_state == LoopState.CANCELLED.value assert task.stop_reason == StopReason.CANCELLED.value assert (Path(task.workspace_path) / "receipt.json").is_file() @@ -324,6 +328,7 @@ async def test_cancelling_parent_cancels_active_descendants(session: AsyncSessio for item in (parent, child, grandchild): await session.refresh(item) assert item.status == TaskStatus.CANCELLED.value + assert item.loop_state == LoopState.CANCELLED.value async def test_agent_scopes_protocol_and_browser_gateway_tokens( @@ -640,6 +645,7 @@ async def test_rejected_finish_then_gives_up_stuck(session: AsyncSession) -> Non await session.refresh(task) assert task.stop_reason == StopReason.STUCK.value assert task.status == TaskStatus.STOPPED.value + assert task.loop_state == LoopState.STOPPED.value assert task.steps_used == 2 assert llm.verify_calls == 1 steps = await StepRepository(session).list_for_task(task.id) @@ -686,6 +692,7 @@ async def test_stops_at_step_cap(session: AsyncSession) -> None: await session.refresh(task) assert task.stop_reason == StopReason.MAX_STEPS.value assert task.status == TaskStatus.STOPPED.value + assert task.loop_state == LoopState.STOPPED.value assert task.steps_used == 3 @@ -766,6 +773,7 @@ async def test_ask_user_pauses_then_resumes_to_completion(session: AsyncSession) await service.run(task.id) await session.refresh(task) assert task.status == TaskStatus.AWAITING_INPUT.value + assert task.loop_state == LoopState.AWAITING_INPUT.value assert task.pending_question == "Which language?" assert task.steps_used == 1 @@ -774,12 +782,14 @@ async def test_ask_user_pauses_then_resumes_to_completion(session: AsyncSession) await tasks_service.respond(task.id, "Python") await session.refresh(task) assert task.status == TaskStatus.PENDING.value + assert task.loop_state == LoopState.QUEUED.value assert task.pending_question is None # Second run resumes from history and finishes. await service.run(task.id) await session.refresh(task) assert task.stop_reason == StopReason.GOAL_ACHIEVED.value + assert task.loop_state == LoopState.COMPLETED.value assert task.summary == "built it" @@ -1198,6 +1208,7 @@ async def test_approval_pauses_then_runs_on_approve(session: AsyncSession) -> No await svc.run(task.id) await session.refresh(task) assert task.status == TaskStatus.AWAITING_INPUT.value + assert task.loop_state == LoopState.AWAITING_APPROVAL.value assert task.pending_action is not None assert "whoami" in (task.pending_question or "") diff --git a/apps/api/tests/test_chat.py b/apps/api/tests/test_chat.py index c74a8e6..5f8761c 100644 --- a/apps/api/tests/test_chat.py +++ b/apps/api/tests/test_chat.py @@ -9,6 +9,7 @@ from app.db.models.task import TaskModel from app.domain.task import StopReason, TaskStatus +from app.services.loop import LoopState async def test_chat_endpoint_replies(client: AsyncClient, monkeypatch: pytest.MonkeyPatch) -> None: @@ -56,6 +57,7 @@ async def fake_execute(task_id: object) -> None: async with sm() as s: t = await TaskRepository(s).get(task_id) # type: ignore[arg-type] t.status = "completed" + t.loop_state = LoopState.COMPLETED.value t.stop_reason = "goal_achieved" t.summary = "done" await s.commit() @@ -71,6 +73,7 @@ async def fake_execute(task_id: object) -> None: svc = TaskService(TaskRepository(s), StepRepository(s)) awaiting = await svc.publish(TaskCreate(goal="do the thing here", chat_id="conv-2")) awaiting.status = "awaiting_input" + awaiting.loop_state = LoopState.AWAITING_INPUT.value awaiting.pending_question = "what colour?" await s.commit() awaiting_id = awaiting.id diff --git a/apps/api/tests/test_deployment_boundaries.py b/apps/api/tests/test_deployment_boundaries.py index 43e6258..6be7c76 100644 --- a/apps/api/tests/test_deployment_boundaries.py +++ b/apps/api/tests/test_deployment_boundaries.py @@ -305,7 +305,7 @@ def test_kubernetes_acceptance_migrates_runs_task_and_rolls_back() -> None: assert "kubectl rollout undo deployment/api" in script assert 'task["sandbox"] != "kubernetes"' in script assert 'report.get("authentic")' in script - assert "0010_operation_journal" in script + assert "0011_loop_state_machine" in script assert "run_cluster_probe after-postgres-restart true" in script assert "gosu postgres pg_ctl -D /var/lib/postgresql/data -m immediate stop" in script assert "kill -9 1" not in script diff --git a/apps/api/tests/test_loop_components.py b/apps/api/tests/test_loop_components.py new file mode 100644 index 0000000..a08fa8f --- /dev/null +++ b/apps/api/tests/test_loop_components.py @@ -0,0 +1,320 @@ +from __future__ import annotations + +from dataclasses import dataclass + +import pytest + +from app.domain.task import StopReason, TaskStatus +from app.services.loop import ( + ActionDispatchPolicy, + ContextBudget, + DecisionParser, + DelegationPolicy, + DispatchKind, + InvalidLoopTransitionError, + LoopEvent, + LoopState, + LoopTransitionPolicy, + VerificationPolicy, + extract_json, +) +from app.services.verification import CheckResult + + +@dataclass +class TaskState: + loop_state: str = LoopState.QUEUED.value + transition_reason: str | None = None + transition_sequence: int = 0 + status: str = TaskStatus.PENDING.value + stop_reason: str | None = None + + +def transition( + policy: LoopTransitionPolicy, + task: TaskState, + event: LoopEvent, + *, + stop_reason: StopReason | None = None, +) -> None: + policy.apply(task, event, f"because:{event.value}", stop_reason=stop_reason) + + +def test_fresh_loop_success_path_is_explicit() -> None: + policy = LoopTransitionPolicy() + task = TaskState() + for event in ( + LoopEvent.CLAIM, + LoopEvent.RUBRIC_REQUIRED, + LoopEvent.RUBRIC_READY, + LoopEvent.ACTION_SELECTED, + LoopEvent.ACTION_RECORDED, + LoopEvent.VERIFICATION_REQUESTED, + LoopEvent.VERIFICATION_ACCEPTED, + ): + transition(policy, task, event) + + assert task.loop_state == LoopState.COMPLETED.value + assert task.status == TaskStatus.COMPLETED.value + assert task.stop_reason == StopReason.GOAL_ACHIEVED.value + assert task.transition_sequence == 7 + assert task.transition_reason == "because:verification_accepted" + + +def test_contract_compilation_success_path_is_explicit() -> None: + policy = LoopTransitionPolicy() + task = TaskState() + for event in ( + LoopEvent.CLAIM, + LoopEvent.CONTRACT_REQUIRED, + LoopEvent.CONTRACT_READY, + LoopEvent.ACTION_SELECTED, + LoopEvent.ACTION_RECORDED, + LoopEvent.VERIFICATION_REQUESTED, + LoopEvent.VERIFICATION_ACCEPTED, + ): + transition(policy, task, event) + + assert task.loop_state == LoopState.COMPLETED.value + assert task.status == TaskStatus.COMPLETED.value + + +@pytest.mark.parametrize( + ("pause_event", "pause_state"), + [ + (LoopEvent.INPUT_REQUIRED, LoopState.AWAITING_INPUT), + (LoopEvent.APPROVAL_REQUIRED, LoopState.AWAITING_APPROVAL), + ], +) +def test_every_pause_path_resumes_through_queue( + pause_event: LoopEvent, pause_state: LoopState +) -> None: + policy = LoopTransitionPolicy() + task = TaskState() + transition(policy, task, LoopEvent.CLAIM) + transition(policy, task, LoopEvent.PREPARATION_COMPLETED) + transition(policy, task, LoopEvent.ACTION_SELECTED) + transition(policy, task, pause_event) + assert task.loop_state == pause_state.value + assert task.status == TaskStatus.AWAITING_INPUT.value + + transition(policy, task, LoopEvent.USER_RESPONDED) + transition(policy, task, LoopEvent.CLAIM) + assert task.loop_state == LoopState.PREPARING.value + assert task.status == TaskStatus.RUNNING.value + + +@pytest.mark.parametrize( + ("source", "event", "target"), + [ + (LoopState.PREPARING, LoopEvent.INPUT_REQUIRED, LoopState.AWAITING_INPUT), + (LoopState.UNDERSTANDING, LoopEvent.INPUT_REQUIRED, LoopState.AWAITING_INPUT), + (LoopState.ACTING, LoopEvent.INPUT_REQUIRED, LoopState.AWAITING_INPUT), + (LoopState.ACTING, LoopEvent.APPROVAL_REQUIRED, LoopState.AWAITING_APPROVAL), + (LoopState.AWAITING_INPUT, LoopEvent.USER_RESPONDED, LoopState.QUEUED), + (LoopState.AWAITING_APPROVAL, LoopEvent.USER_RESPONDED, LoopState.QUEUED), + ], +) +def test_every_human_resumable_transition_is_explicit( + source: LoopState, + event: LoopEvent, + target: LoopState, +) -> None: + policy = LoopTransitionPolicy() + task = TaskState(loop_state=source.value, status=policy.status_for(source).value) + transition(policy, task, event) + assert task.loop_state == target.value + assert task.status == policy.status_for(target).value + + +@pytest.mark.parametrize( + "source", + [ + LoopState.PREPARING, + LoopState.UNDERSTANDING, + LoopState.PLANNING, + LoopState.ACTING, + LoopState.VERIFYING, + ], +) +def test_every_working_state_can_recover_to_preparing(source: LoopState) -> None: + policy = LoopTransitionPolicy() + task = TaskState(loop_state=source.value, status=TaskStatus.RUNNING.value) + transition(policy, task, LoopEvent.RECOVER) + assert task.loop_state == LoopState.PREPARING.value + assert task.status == TaskStatus.RUNNING.value + assert source in policy.claimable_states() + + +@pytest.mark.parametrize( + "source", + [ + LoopState.PREPARING, + LoopState.UNDERSTANDING, + LoopState.PLANNING, + LoopState.ACTING, + LoopState.VERIFYING, + ], +) +@pytest.mark.parametrize( + "reason", [StopReason.MAX_STEPS, StopReason.BUDGET_EXHAUSTED, StopReason.STUCK] +) +def test_every_working_state_has_each_bounded_terminal_path( + source: LoopState, reason: StopReason +) -> None: + policy = LoopTransitionPolicy() + task = TaskState(loop_state=source.value, status=TaskStatus.RUNNING.value) + transition(policy, task, LoopEvent.LIMIT_REACHED, stop_reason=reason) + assert task.loop_state == LoopState.STOPPED.value + assert task.status == TaskStatus.STOPPED.value + assert task.stop_reason == reason.value + + +@pytest.mark.parametrize( + "source", + [ + LoopState.QUEUED, + LoopState.PREPARING, + LoopState.UNDERSTANDING, + LoopState.PLANNING, + LoopState.ACTING, + LoopState.VERIFYING, + LoopState.AWAITING_INPUT, + LoopState.AWAITING_APPROVAL, + ], +) +@pytest.mark.parametrize( + ("event", "target", "status", "stop"), + [ + (LoopEvent.CANCEL, LoopState.CANCELLED, TaskStatus.CANCELLED, StopReason.CANCELLED), + (LoopEvent.FAIL, LoopState.FAILED, TaskStatus.FAILED, StopReason.ERROR), + ], +) +def test_every_active_state_has_cancel_and_failure_paths( + source: LoopState, + event: LoopEvent, + target: LoopState, + status: TaskStatus, + stop: StopReason, +) -> None: + policy = LoopTransitionPolicy() + task = TaskState(loop_state=source.value, status=policy.status_for(source).value) + transition(policy, task, event) + assert task.loop_state == target.value + assert task.status == status.value + assert task.stop_reason == stop.value + + +def test_invalid_or_underspecified_transitions_fail_closed() -> None: + policy = LoopTransitionPolicy() + with pytest.raises(InvalidLoopTransitionError, match="not allowed"): + transition(policy, TaskState(), LoopEvent.VERIFICATION_ACCEPTED) + task = TaskState(loop_state=LoopState.PLANNING.value, status=TaskStatus.RUNNING.value) + with pytest.raises(InvalidLoopTransitionError, match="bounded stop reason"): + transition(policy, task, LoopEvent.LIMIT_REACHED) + + +def test_transition_table_has_no_unreachable_noninitial_state() -> None: + policy = LoopTransitionPolicy() + targets = { + target + for source in LoopState + for event in policy.allowed_events(source) + if (target := _target(policy, source, event)) is not None + } + assert set(LoopState) - {LoopState.QUEUED} <= targets + + +def _target(policy: LoopTransitionPolicy, source: LoopState, event: LoopEvent) -> LoopState | None: + task = TaskState(loop_state=source.value, status=policy.status_for(source).value) + try: + policy.apply( + task, + event, + event.value, + stop_reason=StopReason.STUCK if event is LoopEvent.LIMIT_REACHED else None, + ) + except InvalidLoopTransitionError: + return None + return LoopState(task.loop_state) + + +def test_decision_parser_prefers_final_json_and_enforces_tool_registry() -> None: + assert extract_json('{"thought":"draft"}\nthen {"tool":"finish","args":{}}')["tool"] == ( + "finish" + ) + parser = DecisionParser() + parsed = parser.parse( + 'reasoning {not: json}\n{"thought":"done","tool":"finish","args":{"x":1}}', + valid_tools={"finish"}, + ) + assert (parsed.thought, parsed.tool, parsed.args) == ("done", "finish", {"x": 1}) + assert parser.parse('{"tool":"unknown"}', valid_tools={"finish"}).tool is None + + +def test_context_budget_preserves_verification_reserve() -> None: + budget = ContextBudget.allocate(total=4_000, used=2_900, reserve_cap=800) + assert budget.verification_reserve == 800 + assert budget.remaining == 1_100 + assert budget.planning == 300 + assert budget.verification_after(300) == 800 + + +def test_dispatch_policy_routes_invalid_blocked_approval_and_execute() -> None: + policy = ActionDispatchPolicy() + common = { + "valid_tools": {"run_command", "write_file"}, + "guard_block": None, + "repeated_write_count": 0, + "approval_reason": None, + "last_write_path": None, + } + assert policy.route(None, **common).kind is DispatchKind.INVALID + assert policy.route("run_command", **{**common, "guard_block": "duplicate"}).kind is ( + DispatchKind.BLOCKED + ) + assert ( + policy.route("run_command", **{**common, "approval_reason": "unsafe command"}).kind + is DispatchKind.APPROVAL + ) + assert policy.route("run_command", **common).kind is DispatchKind.EXECUTE + + +def test_delegation_policy_refuses_flooring_and_caps_child() -> None: + policy = DelegationPolicy(minimum_budget=1_000, default_steps=8, max_steps=20) + assert policy.allocate({}, 999) is None + allocation = policy.allocate({"token_budget": 9_000, "max_steps": 99}, 2_000) + assert allocation is not None + assert (allocation.token_budget, allocation.max_steps) == (2_000, 20) + + +def test_verification_policy_requires_execution_evidence_in_strict_mode() -> None: + policy = VerificationPolicy() + result = CheckResult( + kind="command", + target="pytest", + passed=True, + evidence="exit code 0", + criterion_ids=("criterion-001",), + ) + accepted = policy.evaluate( + {"score": 95, "met": True, "missing": []}, + [result], + criterion_count=1, + acceptance_score=80, + strict=True, + contract_substantiation_authoritative=True, + ) + assert accepted.accepted is True + assert accepted.verified_by == "execution" + + rejected = policy.evaluate( + {"score": 95, "met": True, "missing": []}, + [], + criterion_count=1, + acceptance_score=80, + strict=True, + contract_substantiation_authoritative=False, + ) + assert rejected.accepted is False + assert "mapped execution check" in " ".join(rejected.missing) diff --git a/apps/api/tests/test_runner.py b/apps/api/tests/test_runner.py index c00bc2c..96b8dda 100644 --- a/apps/api/tests/test_runner.py +++ b/apps/api/tests/test_runner.py @@ -82,6 +82,43 @@ async def claim() -> bool: await engine.dispose() +async def test_claim_recovers_working_state_but_never_human_wait_state(session) -> None: + from app.domain.loop import LoopState + from app.repositories.task import TaskRepository + + repo = TaskRepository(session) + common = { + "status": "pending", + "rubric": [], + "max_steps": 4, + "token_budget": 10_000, + "summary": None, + "verification_score": 0, + "steps_used": 0, + "tokens_used": 0, + "workspace_path": None, + } + interrupted = await repo.create( + goal="recover me", + loop_state=LoopState.ACTING.value, + transition_sequence=3, + **common, + ) + waiting = await repo.create( + goal="wait for me", + loop_state=LoopState.AWAITING_INPUT.value, + **common, + ) + await session.commit() + + claimed = await repo.claim_pending(interrupted.id) + assert claimed is not None + assert claimed.loop_state == LoopState.PREPARING.value + assert claimed.transition_reason == "worker_recovered_interrupted_task" + assert claimed.transition_sequence == 4 + assert await repo.claim_pending(waiting.id) is None + + async def test_worker_run_task_handler_dispatches(monkeypatch: pytest.MonkeyPatch) -> None: """The worker's run_task handler runs the loop for the payload's task id.""" import app.workers.worker as worker diff --git a/apps/api/tests/test_tasks.py b/apps/api/tests/test_tasks.py index 68ff881..680746b 100644 --- a/apps/api/tests/test_tasks.py +++ b/apps/api/tests/test_tasks.py @@ -19,6 +19,11 @@ async def test_publish_creates_pending_task_with_clamped_limits(client: AsyncCli assert resp.status_code == 201 task = resp.json() assert task["status"] == "pending" + assert task["loop"] == { + "state": "queued", + "transition_reason": None, + "sequence": 0, + } assert task["goal"] == "build a small script" assert task["limits"]["max_steps"] == 40 # clamped to the cap assert task["limits"]["token_budget"] == 200_000 # clamped to the cap @@ -124,6 +129,11 @@ async def test_cancel_pending_task(client: AsyncClient) -> None: cancelled = await client.post(f"/api/v1/tasks/{task_id}/cancel") assert cancelled.status_code == 200 assert cancelled.json()["status"] == "cancelled" + assert cancelled.json()["loop"] == { + "state": "cancelled", + "transition_reason": "cancelled_by_user", + "sequence": 1, + } again = await client.post(f"/api/v1/tasks/{task_id}/cancel") assert again.status_code == 409 diff --git a/apps/web/app/tasks/[id]/page.tsx b/apps/web/app/tasks/[id]/page.tsx index a13b1ec..75d1bbd 100644 --- a/apps/web/app/tasks/[id]/page.tsx +++ b/apps/web/app/tasks/[id]/page.tsx @@ -12,6 +12,7 @@ import { ContractPanel } from '@/components/contract-panel'; import { ReceiptPanel } from '@/components/receipt-panel'; import { StepItem } from '@/components/step-item'; import { StatusPill, stopReasonLabel } from '@/components/status-pill'; +import { LoopState } from '@/components/loop-state'; import { WorkspaceFiles } from '@/components/workspace-files'; import { ApiError, tasksApi } from '@/lib/api-client'; import { apiBaseUrl } from '@/lib/env'; @@ -157,6 +158,7 @@ export default function TaskDetail() {

{task.goal}

+ {reason && task.status === 'stopped' && (

Stopped: {reason}.

diff --git a/apps/web/components/authority-panel.test.tsx b/apps/web/components/authority-panel.test.tsx index e43ec94..c02898a 100644 --- a/apps/web/components/authority-panel.test.tsx +++ b/apps/web/components/authority-panel.test.tsx @@ -10,6 +10,7 @@ const baseTask: Task = { owner_id: 'owner-1', project_id: 'default', status: 'completed', + loop: { state: 'completed', transition_reason: 'verification_accepted', sequence: 4 }, rubric: [], criteria_source: 'generated', verification_mode: 'judgment', diff --git a/apps/web/components/loop-state.test.tsx b/apps/web/components/loop-state.test.tsx new file mode 100644 index 0000000..b953b63 --- /dev/null +++ b/apps/web/components/loop-state.test.tsx @@ -0,0 +1,18 @@ +import { render, screen } from '@testing-library/react'; +import { describe, expect, it } from 'vitest'; +import { LoopState } from './loop-state'; + +describe('LoopState', () => { + it('shows the persisted phase and transition reason', () => { + render( + , + ); + + expect(screen.getByTestId('loop-state')).toHaveTextContent( + 'Loop phase: Verifying evidence · transition #7 · Finish requested', + ); + expect(screen.getByTitle('Transition reason: finish_requested')).toBeInTheDocument(); + }); +}); diff --git a/apps/web/components/loop-state.tsx b/apps/web/components/loop-state.tsx new file mode 100644 index 0000000..29a3c59 --- /dev/null +++ b/apps/web/components/loop-state.tsx @@ -0,0 +1,36 @@ +import type { Task } from '@repo/api-contract'; + +const LABELS: Record = { + queued: 'Queued', + preparing: 'Preparing runtime', + understanding: 'Understanding goal', + planning: 'Planning next action', + acting: 'Executing action', + verifying: 'Verifying evidence', + awaiting_input: 'Waiting for input', + awaiting_approval: 'Waiting for approval', + completed: 'Completed', + stopped: 'Stopped', + cancelled: 'Cancelled', + failed: 'Failed', +}; + +function reasonLabel(reason: string) { + const words = reason.replace(/[_:]+/g, ' ').trim(); + return words.charAt(0).toUpperCase() + words.slice(1); +} + +export function LoopState({ loop }: { loop: Task['loop'] }) { + return ( +

+ Loop phase: {LABELS[loop.state]} + · transition #{loop.sequence} + {loop.transition_reason && ( + + {' '} + · {reasonLabel(loop.transition_reason)} + + )} +

+ ); +} diff --git a/docs/STRATEGY.md b/docs/STRATEGY.md index 287c3f4..f08d361 100644 --- a/docs/STRATEGY.md +++ b/docs/STRATEGY.md @@ -188,6 +188,15 @@ duplicate mutation, and no cross-workspace leakage under at least 20 concurrent ### Gate 3 — Explicit loop state machine +**Implementation status (2026-07-21): gate complete.** The production coordinator +now composes independently tested transition, decision, context, progress, dispatch, +verification, and delegation policies. The authoritative phase, latest reason, and +monotonic transition sequence are persisted on every task and exposed through the API +and task UI. Fresh success, both human pause/resume paths, all bounded stop reasons +from every working phase, cancel/failure from every active phase, and interrupted-run +recovery are transition-tested. The deterministic HTTP benchmark remains at 1/1 solved, +zero false acceptances, two steps, 24 tokens, valid Receipt integrity, and passing replay. + - Add characterization tests before moving behavior. - Split the current orchestration service into explicit transition policy, decision parsing, context budgeting, progress control, action dispatch, verification, and diff --git a/docs/adr/0011-explicit-loop-state-machine.md b/docs/adr/0011-explicit-loop-state-machine.md new file mode 100644 index 0000000..e47958a --- /dev/null +++ b/docs/adr/0011-explicit-loop-state-machine.md @@ -0,0 +1,46 @@ +# 0011 — Persist an explicit loop state machine + +- Status: Accepted +- Date: 2026-07-21 + +## Context + +Task `status` was sufficient for list views but too coarse for a durable autonomous +loop. `running` did not say whether Loop was preparing authority, understanding the +goal, planning, acting, or independently verifying. Pause and recovery behavior was +spread across service branches, and the UI had to infer progress from Steps. Moving or +repairing one branch could therefore create a status combination that another branch +did not understand. + +The orchestration service also owned parsing, token allocation, dispatch, acceptance, +and delegation decisions directly. Those decisions were difficult to exhaustively test +without constructing the entire runtime. + +## Decision + +Persist `loop_state`, `transition_reason`, and `transition_sequence` on every task. +`LoopTransitionPolicy` enumerates allowed events, maps each phase to the compatibility +`status`, and assigns terminal stop reasons. Unlisted transitions fail closed. The +runner's atomic claim and interrupted-task updates persist equivalent reasons and +monotonic sequence changes so recovery does not rely on log interpretation. + +Keep `AgentReactService` as the side-effect coordinator, but move pure decisions into +separate transition, decision parsing, context/history, progress, action dispatch, +verification, and delegation policies. Production execution must use those policies; +they are not a parallel test-only model. Expose the persisted state object through the +API contract and task UI. + +The state graph explicitly covers fresh completion, input and approval pause/resume, +all bounded stops from every working phase, cancellation and failure from every active +phase, and recovery from interrupted working phases. Automatic claiming excludes human +waiting phases. + +## Consequences + +Recovery, API clients, and the UI now share one durable lifecycle vocabulary. Transition +tests can enumerate terminal and resumable paths without provider or sandbox setup, and +parsing/budget/dispatch/verification/delegation regressions have small deterministic +tests. The public `status` remains for compatibility, so persistence code must update it +with the explicit phase rather than writing it independently. The coordinator remains +large because it owns runtime resource wiring and side effects; further extraction can +shrink that surface without changing lifecycle semantics. diff --git a/docs/adr/README.md b/docs/adr/README.md index 9e00088..50d1869 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -16,3 +16,4 @@ a past ADR — supersede it with a new one. | [0008](./0008-agent-behavior-and-runtime-safety-tuning.md) | Empirically-tuned agent behaviour, reliability & runtime safety | Accepted | | [0009](./0009-contract-first-verified-completion.md) | Contract-first Verified Completion | Accepted | | [0010](./0010-generated-contract-locking.md) | Generated contracts are criticized and locked before mutation | Accepted | +| [0011](./0011-explicit-loop-state-machine.md) | Persist an explicit loop state machine | Accepted | diff --git a/docs/loop.md b/docs/loop.md index ec40e70..e5a5f5f 100644 --- a/docs/loop.md +++ b/docs/loop.md @@ -6,6 +6,44 @@ loop. The product contract is in the README; the deployment boundary is in ## State machine +The authoritative lifecycle is not inferred from the coarse task status or recent Step. +Each task persists: + +- `loop_state`: `queued`, `preparing`, `understanding`, `planning`, `acting`, + `verifying`, `awaiting_input`, `awaiting_approval`, or a terminal state; +- `transition_reason`: the stable machine-readable reason for the latest transition; +- `transition_sequence`: a monotonic counter that distinguishes successive phase changes. + +`LoopTransitionPolicy` is the only in-process policy allowed to project these phases to +the public task status and stop reason. Invalid transitions raise instead of silently +repairing state. Atomic SQL claim/recovery paths reproduce the same state, reason, and +sequence semantics so duplicate workers cannot race through an ORM read-modify-write. +The API and task page expose the persisted values directly. + +```mermaid +flowchart LR + Q["queued"] --> P["preparing"] + P --> U["understanding"] + P --> L["planning"] + U --> L + L --> A["acting"] + A --> L + L --> V["verifying"] + V -->|"rejected"| L + V -->|"accepted"| C["completed"] + P --> I["awaiting_input"] + A --> I + A --> H["awaiting_approval"] + I -->|"response"| Q + H -->|"response"| Q +``` + +All five working phases have explicit `max_steps`, `budget_exhausted`, and `stuck` +paths to `stopped`. All eight active/resumable phases have `cancelled` and `failed` +paths. Interrupted working phases can be recovered to `preparing`; waiting phases are +never automatically claimed. Transition tests exhaust these terminal, resumable, and +recovery paths rather than testing only the happy path. + The same `AgentReactService.run` is driven by inline and Redis worker execution: ```text diff --git a/packages/api-contract/src/index.ts b/packages/api-contract/src/index.ts index 375103d..878b615 100644 --- a/packages/api-contract/src/index.ts +++ b/packages/api-contract/src/index.ts @@ -140,6 +140,23 @@ export interface Task { owner_id: string; project_id: string; status: TaskStatus; + loop: { + state: + | 'queued' + | 'preparing' + | 'understanding' + | 'planning' + | 'acting' + | 'verifying' + | 'awaiting_input' + | 'awaiting_approval' + | 'completed' + | 'stopped' + | 'cancelled' + | 'failed'; + transition_reason: string | null; + sequence: number; + }; rubric: string[]; criteria_source: 'user' | 'generated' | 'compiled'; verification_mode: 'strict' | 'judgment'; diff --git a/scripts/k8s-deployment-acceptance.sh b/scripts/k8s-deployment-acceptance.sh index fa9e910..8eba5e0 100755 --- a/scripts/k8s-deployment-acceptance.sh +++ b/scripts/k8s-deployment-acceptance.sh @@ -312,7 +312,7 @@ migration="$( kubectl exec deployment/postgres --namespace "$namespace" -- \ psql -U app -d app -tAc 'SELECT version_num FROM alembic_version' )" -if [[ "$migration" != "0010_operation_journal" ]]; then +if [[ "$migration" != "0011_loop_state_machine" ]]; then echo "Unexpected Alembic revision: $migration" >&2 exit 1 fi