RMI Platform
feat(v3): ship new system β€” skip _legacy_main, mount v1 routers
3cf0daf
Raw
History Blame Contribute Delete
10.3 kB
"""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 "")
)