| """Agent Loop Specification (M1 β P2 #30 in v3 unfuck plan). |
| |
| Frozen on creation. Modified only via bounded-task delegation with Task ID |
| in /home/z/my-project/worklog.md. See DESIGN.md Β§M1. |
| |
| Defines the formal contract for any AI agent loop that runs on RMI infrastructure: |
| Hermes, claude-code, aider, GLM-5.2. Without this spec, loops are unbounded β |
| they iterate forever, burn tokens, produce no verifiable artifacts. |
| |
| Usage from a Hermes cron task: |
| from app.agents.loop import BoundedAgentLoop, TaskInput, LoopBudget |
| loop = BoundedAgentLoop( |
| task=TaskInput( |
| task_id="30-a", |
| description="Add the new typed error class", |
| success_criteria="error class exists and passes mypy", |
| verify_command="python -c 'from app.core.errors import NewError'", |
| ), |
| budget=LoopBudget(max_iterations=5, max_tokens=10_000), |
| ) |
| result = await loop.run() |
| if result.verify_passed: |
| ... |
| """ |
| from __future__ import annotations |
|
|
| import time |
| from typing import Any |
|
|
| from pydantic import BaseModel, ConfigDict, Field |
|
|
|
|
| |
| class TaskInput(BaseModel): |
| """What every agent loop receives.""" |
|
|
| model_config = ConfigDict(strict=True, frozen=True) |
|
|
| task_id: str = Field( |
| ..., |
| pattern=r"^\d+-[a-z0-9-]+$", |
| description="Globally-unique ID matching ^\\d+-[a-z0-9-]+$. Logged before delegation.", |
| ) |
| description: str = Field( |
| ..., |
| max_length=500, |
| description="One-paragraph task description. No multi-page briefs β split the task.", |
| ) |
| context_budget_tokens: int = Field( |
| default=8000, |
| le=32000, |
| description="How much context the agent loads. Larger = more expensive.", |
| ) |
| allowed_tools: list[str] = Field( |
| ..., |
| min_length=1, |
| description="Whitelist of MCP tools the agent may call. Anything else trips a kill switch.", |
| ) |
| success_criteria: str = Field( |
| ..., |
| max_length=300, |
| description="One-sentence definition of 'done.' Verifiable by verify_command.", |
| ) |
| verify_command: str = Field( |
| ..., |
| max_length=200, |
| description="Shell command that returns 0 if success_criteria is met.", |
| ) |
|
|
|
|
| |
| class LoopBudget(BaseModel): |
| """The four kill switches. Checked every iteration.""" |
|
|
| model_config = ConfigDict(strict=True, frozen=True) |
|
|
| max_iterations: int = Field(default=20, le=100) |
| max_tokens: int = Field(default=50_000, le=500_000) |
| max_wallclock_seconds: int = Field(default=900, le=3600) |
| max_spend_usd: float = Field(default=5.0, le=50.0) |
|
|
|
|
| |
| class TaskOutput(BaseModel): |
| """What every agent loop returns.""" |
|
|
| model_config = ConfigDict(strict=True) |
|
|
| task_id: str |
| files_touched: list[str] = Field(default_factory=list) |
| lines_added: int = 0 |
| lines_removed: int = 0 |
| verify_passed: bool = False |
| worklog_entry: str = "" |
| spend_usd: float = 0.0 |
| iterations_used: int = 0 |
| aborted: bool = False |
| abort_reason: str | None = None |
|
|
|
|
| |
| class BudgetExceededError(RuntimeError): |
| """Raised when any kill switch trips.""" |
|
|
|
|
| |
| class BoundedAgentLoop: |
| """Wraps any agent loop with kill switches and a verifiable output contract. |
| |
| This is the production runtime. For the actual loop body, subclass and |
| override _step(). The base class enforces the budget and produces the |
| output contract. |
| """ |
|
|
| def __init__(self, task: TaskInput, budget: LoopBudget | None = None) -> None: |
| self.task = task |
| self.budget = budget or LoopBudget() |
| self._iterations_used = 0 |
| self._tokens_used = 0 |
| self._spend_usd = 0.0 |
| self._files_touched: list[str] = [] |
| self._lines_added = 0 |
| self._lines_removed = 0 |
| self._start_time = 0.0 |
| self._aborted = False |
| self._abort_reason: str | None = None |
|
|
| async def run(self) -> TaskOutput: |
| """Execute the loop until done, budget exceeded, or verify passes.""" |
| self._start_time = time.monotonic() |
|
|
| |
| facts = await self._load_facts() |
| context = self._build_initial_context(facts) |
|
|
| while not self._aborted: |
| self._check_budget() |
| self._iterations_used += 1 |
|
|
| try: |
| step_result = await self._step(context) |
| except BudgetExceededError as exc: |
| self._aborted = True |
| self._abort_reason = str(exc) |
| break |
|
|
| self._record_step(step_result) |
| context = self._update_context(context, step_result) |
|
|
| |
| if await self._verify(): |
| break |
|
|
| |
| verify_passed = await self._verify() |
|
|
| return TaskOutput( |
| task_id=self.task.task_id, |
| files_touched=self._files_touched, |
| lines_added=self._lines_added, |
| lines_removed=self._lines_removed, |
| verify_passed=verify_passed, |
| worklog_entry=self._build_worklog_entry(), |
| spend_usd=self._spend_usd, |
| iterations_used=self._iterations_used, |
| aborted=self._aborted, |
| abort_reason=self._abort_reason, |
| ) |
|
|
| |
| async def _step(self, context: Any) -> dict[str, Any]: |
| """One iteration of the agent loop. Subclass and implement.""" |
| raise NotImplementedError |
|
|
| |
| def _check_budget(self) -> None: |
| """Throws BudgetExceededError if any kill switch is tripped.""" |
| if self._iterations_used + 1 > self.budget.max_iterations: |
| raise BudgetExceededError( |
| f"max_iterations={self.budget.max_iterations} exceeded" |
| ) |
|
|
| elapsed = time.monotonic() - self._start_time |
| if elapsed > self.budget.max_wallclock_seconds: |
| raise BudgetExceededError( |
| f"max_wallclock_seconds={self.budget.max_wallclock_seconds} exceeded" |
| ) |
|
|
| if self._tokens_used > self.budget.max_tokens: |
| raise BudgetExceededError( |
| f"max_tokens={self.budget.max_tokens} exceeded" |
| ) |
|
|
| if self._spend_usd > self.budget.max_spend_usd: |
| raise BudgetExceededError( |
| f"max_spend_usd={self.budget.max_spend_usd} exceeded" |
| ) |
|
|
| |
| async def _load_facts(self) -> dict[str, Any]: |
| """Load facts from fact_store at loop start.""" |
| from app.agents.fact_store import load_facts |
|
|
| return await load_facts(namespace="agents") |
|
|
| async def _verify(self) -> bool: |
| """Run verify_command and return True if exit code is 0.""" |
| import asyncio |
|
|
| try: |
| proc = await asyncio.create_subprocess_shell( |
| self.task.verify_command, |
| stdout=asyncio.subprocess.PIPE, |
| stderr=asyncio.subprocess.PIPE, |
| ) |
| stdout, stderr = await asyncio.wait_for( |
| proc.communicate(), timeout=120 |
| ) |
| return proc.returncode == 0 |
| except (asyncio.TimeoutError, OSError): |
| return False |
|
|
| def _build_initial_context(self, facts: dict[str, Any]) -> Any: |
| """Build the initial context for step 0. Override for custom merging.""" |
| return { |
| "task": self.task.model_dump(), |
| "facts": facts, |
| "budget": self.budget.model_dump(), |
| "iteration": 0, |
| } |
|
|
| def _update_context(self, context: Any, step_result: dict[str, Any]) -> Any: |
| """Update context for the next iteration.""" |
| context = dict(context) |
| context["iteration"] = context.get("iteration", 0) + 1 |
| context["last_step"] = step_result |
| return context |
|
|
| def _record_step(self, step_result: dict[str, Any]) -> None: |
| """Update internal counters from step result.""" |
| self._tokens_used += int(step_result.get("tokens_used", 0)) |
| self._spend_usd += float(step_result.get("spend_usd", 0)) |
| self._files_touched.extend(step_result.get("files_touched", [])) |
| self._lines_added += int(step_result.get("lines_added", 0)) |
| self._lines_removed += int(step_result.get("lines_removed", 0)) |
|
|
| def _build_worklog_entry(self) -> str: |
| """Build the worklog entry for this task.""" |
| return ( |
| f"task_id: {self.task.task_id}\n" |
| f"description: {self.task.description}\n" |
| f"iterations_used: {self._iterations_used}\n" |
| f"tokens_used: {self._tokens_used}\n" |
| f"spend_usd: ${self._spend_usd:.3f}\n" |
| f"files_touched: {len(self._files_touched)}\n" |
| f"lines_added: {self._lines_added}\n" |
| f"lines_removed: {self._lines_removed}\n" |
| f"aborted: {self._aborted}" |
| + (f" (reason: {self._abort_reason})" if self._abort_reason else "") |
| ) |
|
|