backend/app/agents/loop.py

"""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 "")
        )