"""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 # ── Input Contract ────────────────────────────────────────────────────── 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.", ) # ── Budget / Kill Switches ───────────────────────────────────────────── 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) # ── Output Contract ───────────────────────────────────────────────────── 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 # ── Kill switch reasons ───────────────────────────────────────────────── class BudgetExceededError(RuntimeError): """Raised when any kill switch trips.""" # ── Bounded Agent Loop ────────────────────────────────────────────────── 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() # Load long-term memory from fact_store at loop start. 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) # Check verify_command after each step (cheap path). if await self._verify(): break # Final verification. 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, ) # ── To be overridden by subclasses ────────────────────────────────── async def _step(self, context: Any) -> dict[str, Any]: """One iteration of the agent loop. Subclass and implement.""" raise NotImplementedError # ── Built-in budget enforcement ───────────────────────────────────── 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" ) # ── Helpers (override-friendly) ───────────────────────────────────── 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 "") )