diff --git a/CHANGELOG.md b/CHANGELOG.md index 89f2f6a1..b080ba04 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ## [Unreleased] +### Fixed + +- **The PM-respawn loop breaker now survives an orchestrator restart.** The circuit breaker that stops RoboCo from respawning the same PM on the same wedged task forever (`_pm_respawn_tracker`) lived only in memory, so a deploy/crash/OOM reset a task's strike count to 1 and re-burned the whole threshold — four full agent spawns × container cost — against the still-broken task before the gate fired again. The counter is now write-through-persisted to a new `respawn_tracker` table (migration 051) on every mutation and restored at startup, validated against live tasks so a stale counter can't resurrect against a fixed one. Best-effort and inert when empty (a DB hiccup degrades to exactly the prior in-memory behaviour); it can only ever suppress a spawn, never manufacture one. + ## [0.13.0] - 2026-06-26 ### Added diff --git a/CLAUDE.md b/CLAUDE.md index 2d334ffa..b8fa34a6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -512,6 +512,7 @@ Server-side events reach these sockets through `roboco/api/websocket_bridge.py`, - **Provider overloads** reuse the same park-and-probe break. A persistent model-API overload (HTTP 529 / 500 / 503 — the SDK already retries transient ones) parks the provider exactly like a 429 instead of crash-retrying the agent straight back into the overload and burning tokens; the overload is detected orchestrator-side from the dead container's log markers, and the background loop revives the parked work when it recovers. The same break also catches the **Claude session-limit** 429 (the org's 5-hour usage window): an agent exiting with a 0-token session-limit rejection parks the provider and is auto-revived when the window resets, instead of fleet-wide crash-respawning straight back into the limit. Gated by `ROBOCO_OVERLOAD_BREAK_ENABLED` (default-on). - **Gateway-health recovery** closes a blind spot in the stale-claim reaper: the heartbeat is bumped only by gateway verbs, so a broken-but-alive agent (a corrupted `/app/.venv` so no gateway tool imports) goes heartbeat-stale yet keeps its container up, and the reaper's live-skip would protect it forever. On a stale-heartbeat live container the reaper now probes the gateway out-of-band (`_probe_gateway_health` → `docker exec` the gateway venv imports) and, once broken past `ROBOCO_GATEWAY_HEALTH_GRACE_SECONDS` (a transient probe miss is tolerated), kills + evicts it (`_maybe_recover_broken_gateway`) so it falls through to release + respawn; healthy or inconclusive probes spare it. Gated by `ROBOCO_GATEWAY_HEALTH_ENABLED` (default-on). It is the third leg beside the shipped bash-guard `/app` block (prevents the self-corruption) and the reaper Docker-liveness fallback (stops over-reaping live containers). - **PM coordinator concurrency.** A Main / Cell PM plans and delegates many root tasks in parallel — the actual work then runs in the delegated children/cells, not in the PM's own hands. The claim-time concurrency guards that keep a *developer* to one task at a time (`already_active` / `paused`, in `roboco/services/gateway/claim_guards.py`) are therefore **skipped for the coordinator PM roles** (`_COORDINATOR_ROLES = {main_pm, cell_pm}`, consulted in `_run_claim_guards`); only a genuine upstream **sequence dependency** (`unmet_dependency`, which parks the task back to `pending`) holds a PM's root back. Without this a single PM that claimed one root could never plan a second — it thrashed between its claimed roots and respawned forever, burning tokens for zero progress (the live `i_am_idle`-auto-paused-umbrella deadlock). The `paused` guard also excludes the target task itself, so a PM re-entering its own paused umbrella never self-blocks. +- **Orchestrator runtime-state durability.** The PM-respawn loop breaker (`_pm_respawn_tracker`, the `(agent_slug, task_id) → strike-count` circuit breaker) is **DB-durable** via the `respawn_tracker` table (migration 051): each gate mutation write-throughs fire-and-forget on the `_bg_tasks` set (`_schedule_respawn_persist` → `_persist_respawn_record`), and `restore_respawn_tracker()` repopulates it at `start()`, validating each row against live tasks (terminal/missing rows are evicted). Kept only in memory it reset to `count=1` on every restart and re-burned the whole strike threshold (4 spawns) against a still-wedged task. It mirrors the `WaitingRecordTable` / `restore_waiting_records` pattern: best-effort (a DB hiccup degrades to in-memory-only — it can only ever *suppress* a spawn, never manufacture one) and inert when the table is empty. The companion `_instances` registry is **reconciled-from-Docker** (not persisted) at startup via `_readopt_running_agents`, so the reaper's liveness path and the spawn gate's `_is_agent_active` check see surviving containers immediately after a restart. - **Token usage** is captured per agent session from the Claude Code transcript via the SDK server's `/usage/sync` (hook → orchestrator finalize → `agent_spawn_sessions` → `daily_usage_rollups` → dashboard). Cost uses provider-aware pricing in `roboco/billing/pricing.py` (Anthropic priced; local/Ollama intentionally `$0`). The token sweep also publishes `USAGE_SNAPSHOT` to `/ws/system`, so the dashboard's "Token Usage & Cost" panel updates live and falls back to HTTP polling when the stream is down. - **Delivery observability** (the panel's Metrics → "Delivery" tab) shows how work *flows*, computed by `MetricsService` from data already captured — no new feature flag. Per-stage cycle time and the bottleneck distribution are reconstructed from the `audit_log` transition journey (each generic `task.` event marks entry into a status; the named `task.qa_fail`/`task.pr_fail` events are excluded from the reconstruction). Rework rate reads `tasks.revision_count` — incremented once per transition into `needs_revision` at the single chokepoint `TaskService._emit_status_transition_audit` — and attributes each bounce to the QA / PR-reviewer via those named audit events; rework cost joins `agent_spawn_sessions.task_id`. Read-only endpoints: `/dashboard/metrics/{cycle-time,bottlenecks,rework,scorecard/agent/{id},scorecard/team/{team}}`. diff --git a/alembic/versions/051_respawn_tracker.py b/alembic/versions/051_respawn_tracker.py new file mode 100644 index 00000000..8d867d3f --- /dev/null +++ b/alembic/versions/051_respawn_tracker.py @@ -0,0 +1,51 @@ +"""Add the respawn_tracker table — durable PM-respawn loop counter. + +``AgentOrchestrator._pm_respawn_tracker`` is the circuit breaker against +respawning the same PM on the same task forever. Kept only in memory it reset +to ``count=1`` on every orchestrator restart, re-burning the whole strike +threshold against a still-wedged task. This table is its write-through mirror, +restored at startup. Composite PK ``(agent_slug, task_id)`` matches the +in-memory dict key. ``task_id`` is deliberately NOT a FK to ``tasks``: the +startup loader validates against live tasks instead, so a stale counter cannot +resurrect against a fixed/deleted task. + +Revision ID: 051_respawn_tracker +Revises: 050_playbooks +Create Date: 2026-06-26 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "051_respawn_tracker" +down_revision = "050_playbooks" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "respawn_tracker", + sa.Column("agent_slug", sa.String(length=64), nullable=False), + sa.Column("task_id", sa.UUID(as_uuid=True), nullable=False), + sa.Column("count", sa.Integer(), nullable=False, server_default="1"), + sa.Column("last_status", sa.String(length=64), nullable=True), + sa.Column("last_check", sa.DateTime(timezone=True), nullable=False), + sa.Column("tracing_resets", sa.Integer(), nullable=False, server_default="0"), + sa.Column("notified", sa.Boolean(), nullable=False, server_default=sa.false()), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("now()"), + ), + sa.PrimaryKeyConstraint("agent_slug", "task_id"), + ) + op.create_index("ix_respawn_tracker_last_check", "respawn_tracker", ["last_check"]) + + +def downgrade() -> None: + op.drop_index("ix_respawn_tracker_last_check", table_name="respawn_tracker") + op.drop_table("respawn_tracker") diff --git a/roboco/api/routes/project.py b/roboco/api/routes/project.py index ffe502dd..397e85f8 100644 --- a/roboco/api/routes/project.py +++ b/roboco/api/routes/project.py @@ -4,7 +4,7 @@ Project API Routes CRUD operations for managing git projects/repositories. """ -from typing import TYPE_CHECKING, Annotated +from typing import TYPE_CHECKING, Annotated, cast from uuid import UUID from fastapi import APIRouter, HTTPException, Query, status @@ -221,7 +221,7 @@ async def update_project( is_active=data.is_active, ) - updated = await service.update(project.id, update_data) + updated = await service.update(cast("UUID", project.id), update_data) await db.commit() if not updated: @@ -269,7 +269,7 @@ async def delete_project( require_cell_access(agent, project.assigned_cell, "delete") - deleted = await service.delete(project.id) + deleted = await service.delete(cast("UUID", project.id)) await db.commit() if not deleted: @@ -309,7 +309,7 @@ async def set_workspace( status_code=status.HTTP_404_NOT_FOUND, detail=f"Project not found: {project_id}", ) from None - uuid = project.id + uuid = cast("UUID", project.id) updated = await service.set_workspace_path(uuid, data.workspace_path) await db.commit() @@ -346,7 +346,7 @@ async def update_sync_state( status_code=status.HTTP_404_NOT_FOUND, detail=f"Project not found: {project_id}", ) from None - uuid = project.id + uuid = cast("UUID", project.id) updated = await service.update_sync_state(uuid, data.head_commit) await db.commit() @@ -391,7 +391,7 @@ async def add_agent_access( status_code=status.HTTP_404_NOT_FOUND, detail=f"Project not found: {project_id}", ) from None - uuid = project.id + uuid = cast("UUID", project.id) updated = await service.add_allowed_agent(uuid, agent_id) await db.commit() @@ -428,7 +428,7 @@ async def remove_agent_access( status_code=status.HTTP_404_NOT_FOUND, detail=f"Project not found: {project_id}", ) from None - uuid = project.id + uuid = cast("UUID", project.id) updated = await service.remove_allowed_agent(uuid, agent_id) await db.commit() diff --git a/roboco/db/tables.py b/roboco/db/tables.py index 29c2b2bf..6e7ea6c6 100644 --- a/roboco/db/tables.py +++ b/roboco/db/tables.py @@ -1843,6 +1843,39 @@ class WaitingRecordTable(Base): __table_args__ = (Index("ix_waiting_records_waiting_for", "waiting_for"),) +class RespawnTrackerTable(Base): + """Persistent backing for the orchestrator's PM-respawn loop counter. + + ``AgentOrchestrator._pm_respawn_tracker`` is the circuit breaker against + respawning the same PM on the same task forever. Kept only in memory it + reset to ``count=1`` on every orchestrator restart, re-burning the whole + strike threshold against a still-wedged task. This mirror survives a + restart. Default-on; inert when empty (degrades to in-memory-only). + + ``task_id`` is deliberately NOT a FK to ``tasks``: the startup loader + validates each row against live tasks (skipping terminal/missing ones), so + a stale counter can never resurrect against a fixed/deleted task and the + deletion authority stays in one explicit, tested place. + """ + + __tablename__ = "respawn_tracker" + + agent_slug: Mapped[str] = mapped_column(String(64), primary_key=True) + task_id: Mapped[UUID] = mapped_column(UUID(as_uuid=True), primary_key=True) + count: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + last_status: Mapped[str | None] = mapped_column(String(64), nullable=True) + last_check: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False + ) + tracing_resets: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + notified: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=lambda: datetime.now(UTC), nullable=False + ) + + __table_args__ = (Index("ix_respawn_tracker_last_check", "last_check"),) + + # ============================================================================= # AUDIT LOG # ============================================================================= diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 251de10e..5e0fde88 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -26,7 +26,7 @@ from typing import TYPE_CHECKING, Any, ClassVar, cast import httpx if TYPE_CHECKING: - from collections.abc import Callable, Coroutine + from collections.abc import Callable, Coroutine, Iterable from uuid import UUID from sqlalchemy.ext.asyncio import AsyncSession @@ -810,6 +810,12 @@ class AgentOrchestrator: # agents that were WAITING_LONG at shutdown can still be resolved. await self.restore_waiting_records() + # Restore the PM-respawn loop counter so a task wedged at the strike + # threshold trips immediately after a restart instead of resetting to + # count=1 and re-burning the whole budget. Validates against live tasks + # (drops terminal/missing rows); inert when the table is empty. + await self.restore_respawn_tracker() + # Self-heal: roll back orphan claims left over from a prior crash. # Tasks that show CLAIMED/IN_PROGRESS but have NO # branch_name set indicate _finalize_claim flushed the status before @@ -3885,6 +3891,20 @@ class AgentOrchestrator: self._bg_tasks.add(bg) bg.add_done_callback(self._bg_tasks.discard) + def _schedule_respawn_persist( + self, agent_slug: str, task_id: str, record: dict[str, Any] + ) -> None: + """Fire-and-forget a write-through of one PM-respawn counter row. + + Copies ``record`` so a later in-place mutation can't race the background + write, then schedules it on the strong-ref ``_bg_tasks`` set — the + dispatcher hot path never blocks on the DB, and a write failure degrades + to in-memory-only (today's behaviour). + """ + self._schedule_bg( + self._persist_respawn_record(agent_slug, task_id, dict(record)) + ) + def _schedule_intake_first_message(self, session_id: str, text: str) -> None: """Fire-and-forget the opening message once the container is reachable.""" self._schedule_bg(self._deliver_when_ready(session_id, text)) @@ -4079,6 +4099,83 @@ class AgentOrchestrator: error=str(e), ) + async def _persist_respawn_record( + self, agent_slug: str, task_id: str, record: dict[str, Any] + ) -> None: + """Write-through one PM-respawn counter row (delete-then-insert upsert). + + Best-effort, mirroring ``_persist_waiting_record``: a persistence failure + must never gate or un-gate a spawn, so any error is logged and swallowed. + The counter stays authoritative in memory regardless. + """ + try: + from uuid import UUID as _UUID + + from sqlalchemy import delete + + from roboco.db.base import get_session_factory + from roboco.db.tables import RespawnTrackerTable + + tid = _UUID(task_id) + session_factory = get_session_factory() + async with session_factory() as db: + await db.execute( + delete(RespawnTrackerTable).where( + RespawnTrackerTable.agent_slug == agent_slug, + RespawnTrackerTable.task_id == tid, + ) + ) + db.add( + RespawnTrackerTable( + agent_slug=agent_slug, + task_id=tid, + count=int(record["count"]), + last_status=record.get("last_status"), + last_check=record["last_check"], + tracing_resets=int(record.get("tracing_resets", 0)), + notified=bool(record.get("notified", False)), + ) + ) + await db.commit() + except Exception as e: + logger.error( + "Failed to persist respawn record", + agent_id=agent_slug, + task_id=task_id, + error=str(e), + ) + + async def _clear_respawn_record(self, agent_slug: str, task_id: str) -> None: + """Delete one PM-respawn counter row (best-effort). + + Used by the startup loader to evict a row whose task is gone or + terminal, so a stale counter never resurrects against a fixed task. + """ + try: + from uuid import UUID as _UUID + + from sqlalchemy import delete + + from roboco.db.base import get_session_factory + from roboco.db.tables import RespawnTrackerTable + + session_factory = get_session_factory() + async with session_factory() as db: + await db.execute( + delete(RespawnTrackerTable).where( + RespawnTrackerTable.agent_slug == agent_slug, + RespawnTrackerTable.task_id == _UUID(task_id), + ) + ) + await db.commit() + except Exception as e: + logger.error( + "Failed to clear respawn record", + agent_id=agent_slug, + task_id=task_id, + error=str(e), + ) + # ========================================================================= # PROVIDER QUERY HELPERS (used by the choreographer rate-limit path) # ========================================================================= @@ -4826,6 +4923,83 @@ class AgentOrchestrator: logger.error("Failed to restore waiting records", error=str(e)) return 0 + @staticmethod + def _partition_respawn_rows( + rows: "Iterable[Any]", status_by_id: dict[Any, Any] + ) -> tuple[dict[tuple[str, str], dict[str, Any]], list[tuple[str, Any]]]: + """Split persisted respawn rows into (restorable entries, stale keys). + + Pure: a row is **stale** when its task is missing from ``status_by_id`` + or terminal (completed/cancelled) — a stale counter must never resurrect + against a fixed/deleted task. Restorable entries are keyed + ``(agent_slug, str(task_id))`` to match the in-memory dict; stale keys + carry the raw ``task_id`` for deletion. + """ + from roboco.models.base import TaskStatus + + terminal = {TaskStatus.COMPLETED.value, TaskStatus.CANCELLED.value} + restored: dict[tuple[str, str], dict[str, Any]] = {} + stale: list[tuple[str, Any]] = [] + for r in rows: + status = status_by_id.get(r.task_id) + norm = getattr(status, "value", status) + if status is None or norm in terminal: + stale.append((r.agent_slug, r.task_id)) + continue + restored[(r.agent_slug, str(r.task_id))] = { + "count": r.count, + "last_status": r.last_status, + "last_check": r.last_check, + "tracing_resets": r.tracing_resets, + "notified": r.notified, + } + return restored, stale + + async def restore_respawn_tracker(self) -> int: + """Load the persisted PM-respawn counter into memory on startup. + + Mirrors ``restore_waiting_records``: read every ``respawn_tracker`` row, + keep only those whose task is still live and non-terminal, evict the + rest, and populate ``_pm_respawn_tracker`` so a wedged-task counter trips + at its persisted threshold instead of resetting to 1 and re-burning the + whole budget. Best-effort — any failure starts with an empty tracker + (exactly today's behaviour) and never blocks startup. + """ + try: + from sqlalchemy import select + + from roboco.db.base import get_session_factory + from roboco.db.tables import RespawnTrackerTable, TaskTable + + session_factory = get_session_factory() + async with session_factory() as db: + rows = (await db.execute(select(RespawnTrackerTable))).scalars().all() + if not rows: + return 0 + ids = [r.task_id for r in rows] + live = ( + await db.execute( + select(TaskTable.id, TaskTable.status).where( + TaskTable.id.in_(ids) + ) + ) + ).all() + status_by_id = {row.id: row.status for row in live} + restored, stale = self._partition_respawn_rows(rows, status_by_id) + self._pm_respawn_tracker.update(restored) + for agent_slug, task_id in stale: + await self._clear_respawn_record(agent_slug, str(task_id)) + if restored: + logger.info( + "Restored PM-respawn records from database", + count=len(restored), + evicted=len(stale), + ) + return len(restored) + except Exception as e: + logger.error("Failed to restore respawn records", error=str(e)) + return 0 + async def resolve_wait( self, agent_id: str, @@ -7819,6 +7993,9 @@ Start now: evidence(task_id="{task_id}") "last_status": current_status, "last_check": now, } + self._schedule_respawn_persist( + agent_slug, str(task_id), self._pm_respawn_tracker[key] + ) return False # Same status as last spawn — could be a stuck loop OR a # rule-following retry. A tracing_gap normally means the agent is @@ -7835,6 +8012,9 @@ Start now: evidence(task_id="{task_id}") record["count"] = 1 record["last_check"] = now record["notified"] = False + self._schedule_respawn_persist( + agent_slug, str(task_id), self._pm_respawn_tracker[key] + ) return False logger.warning( "PM respawn tracing_gap reset budget exhausted — " @@ -7846,6 +8026,9 @@ Start now: evidence(task_id="{task_id}") ) record["count"] += 1 record["last_check"] = now + self._schedule_respawn_persist( + agent_slug, str(task_id), self._pm_respawn_tracker[key] + ) if record["count"] > self._PM_RESPAWN_MAX_UNPRODUCTIVE: logger.warning( "PM respawn loop detected — skipping spawn", @@ -7863,6 +8046,9 @@ Start now: evidence(task_id="{task_id}") # an overseer once so a wedged agent isn't silently stranded. if not record.get("notified"): record["notified"] = True + self._schedule_respawn_persist( + agent_slug, str(task_id), self._pm_respawn_tracker[key] + ) await self._notify_stuck_agent(agent_slug, task_id, current_status) return True return False diff --git a/roboco/services/self_heal_engine.py b/roboco/services/self_heal_engine.py index d99ac812..cb6b7a80 100644 --- a/roboco/services/self_heal_engine.py +++ b/roboco/services/self_heal_engine.py @@ -25,7 +25,7 @@ from __future__ import annotations import hashlib from dataclasses import dataclass -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, cast from roboco.config import settings from roboco.foundation import identity as _foundation @@ -43,6 +43,8 @@ from roboco.services.task import ( from roboco.services.telemetry import get_ci_telemetry_source if TYPE_CHECKING: + from uuid import UUID + from sqlalchemy.ext.asyncio import AsyncSession from roboco.services.telemetry import TelemetrySource @@ -188,7 +190,7 @@ class SelfHealEngine(BaseService): task_type=TaskType.CODE, nature=TaskNature.TECHNICAL, estimated_complexity=Complexity.MEDIUM, - project_id=project.id, + project_id=cast("UUID", project.id), status=TaskStatus.PENDING, source=SELF_HEAL_SOURCE, confirmed_by_human=True, diff --git a/tests/unit/db/test_respawn_tracker_table.py b/tests/unit/db/test_respawn_tracker_table.py new file mode 100644 index 00000000..5e92956d --- /dev/null +++ b/tests/unit/db/test_respawn_tracker_table.py @@ -0,0 +1,52 @@ +"""The respawn_tracker table — durable backing for the PM-respawn counter. + +Mirrors WaitingRecordTable: a composite-PK row per (agent_slug, task_id) the +orchestrator's loop-breaker counter is keyed on, so it survives a restart +instead of resetting to count=1 and re-burning the strike threshold. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, cast + +from roboco.db.tables import RespawnTrackerTable + +if TYPE_CHECKING: + from sqlalchemy import Table + +_TABLE = cast("Table", RespawnTrackerTable.__table__) + + +def test_table_name() -> None: + assert RespawnTrackerTable.__tablename__ == "respawn_tracker" + + +def test_composite_primary_key_is_agent_slug_and_task_id() -> None: + pk_cols = {col.name for col in _TABLE.primary_key.columns} + assert pk_cols == {"agent_slug", "task_id"} + + +def test_payload_columns_present() -> None: + cols = set(_TABLE.columns.keys()) + assert { + "agent_slug", + "task_id", + "count", + "last_status", + "last_check", + "tracing_resets", + "notified", + "updated_at", + } <= cols + + +def test_task_id_has_no_foreign_key() -> None: + # Deliberately NOT a FK to tasks: the startup loader validates against live + # tasks instead, so a cascade can never silently resurrect/erase a counter. + task_id = _TABLE.columns["task_id"] + assert task_id.foreign_keys == set() + + +def test_last_check_index_present() -> None: + index_names = {idx.name for idx in _TABLE.indexes} + assert "ix_respawn_tracker_last_check" in index_names diff --git a/tests/unit/runtime/test_pm_respawn_reset.py b/tests/unit/runtime/test_pm_respawn_reset.py index 73d098aa..c95ac6ca 100644 --- a/tests/unit/runtime/test_pm_respawn_reset.py +++ b/tests/unit/runtime/test_pm_respawn_reset.py @@ -36,6 +36,10 @@ def _new_orchestrator() -> AgentOrchestrator: """Bypass __init__ so tests don't need a full DI graph.""" orch = AgentOrchestrator.__new__(AgentOrchestrator) cast("Any", orch)._pm_respawn_tracker = {} + # The gate now write-throughs each mutation to the respawn_tracker table via + # _schedule_respawn_persist; these tests cover gate LOGIC only, so stub the + # scheduler to a no-op (persistence is covered in test_respawn_persistence). + cast("Any", orch)._schedule_respawn_persist = lambda *_a, **_k: None return orch diff --git a/tests/unit/runtime/test_respawn_persistence.py b/tests/unit/runtime/test_respawn_persistence.py new file mode 100644 index 00000000..b1831f1a --- /dev/null +++ b/tests/unit/runtime/test_respawn_persistence.py @@ -0,0 +1,315 @@ +"""The PM-respawn counter survives an orchestrator restart. + +`_pm_respawn_tracker` is the loop breaker against respawning the same PM on the +same task forever. Kept only in memory it reset to count=1 on every restart, +re-burning the whole strike threshold against a still-wedged task. These tests +cover the write-through persist on each gate mutation, the startup loader (which +validates against live tasks and drops terminal/missing rows), and the safety +property that a restored counter trips at the persisted threshold — never +manufacturing a spawn. +""" + +from __future__ import annotations + +import copy +from datetime import UTC, datetime +from types import SimpleNamespace +from typing import Any, cast +from unittest.mock import AsyncMock, MagicMock, patch +from uuid import uuid4 + +import pytest +from roboco.runtime.orchestrator import AgentOrchestrator + +_SEEDED_COUNT = 3 # a persisted strike count, one below the trip threshold +_STRIKE_COUNT = 2 +_MIN_PERSISTS = 2 +_TRIP_COUNT = 4 # count > _PM_RESPAWN_MAX_UNPRODUCTIVE (3) fires the gate + + +def _new_orchestrator() -> AgentOrchestrator: + orch = AgentOrchestrator.__new__(AgentOrchestrator) + cast("Any", orch)._pm_respawn_tracker = {} + cast("Any", orch)._bg_tasks = set() + return orch + + +def _row(task_id: Any, **over: Any) -> SimpleNamespace: + base = { + "agent_slug": "be-pm", + "task_id": task_id, + "count": 2, + "last_status": "blocked", + "last_check": datetime(2026, 6, 26, tzinfo=UTC), + "tracing_resets": 0, + "notified": False, + } + base.update(over) + return SimpleNamespace(**base) + + +# --------------------------------------------------------------------------- # +# Pure partition helper +# --------------------------------------------------------------------------- # + + +def test_partition_keeps_live_nonterminal_rows() -> None: + tid = uuid4() + rows = [_row(tid, count=3)] + restored, stale = AgentOrchestrator._partition_respawn_rows( + rows, {tid: "in_progress"} + ) + assert stale == [] + assert restored[("be-pm", str(tid))]["count"] == _SEEDED_COUNT + assert restored[("be-pm", str(tid))]["last_status"] == "blocked" + + +def test_partition_drops_terminal_and_missing_rows() -> None: + done, cancelled, gone = uuid4(), uuid4(), uuid4() + rows = [_row(done), _row(cancelled), _row(gone)] + restored, stale = AgentOrchestrator._partition_respawn_rows( + rows, + {done: "completed", cancelled: "cancelled"}, # gone absent entirely + ) + assert restored == {} + assert {(s, t) for s, t in stale} == { + ("be-pm", done), + ("be-pm", cancelled), + ("be-pm", gone), + } + + +# --------------------------------------------------------------------------- # +# Startup loader +# --------------------------------------------------------------------------- # + + +def _mock_session_factory(respawn_rows: list[Any], live: list[Any]) -> Any: + """A get_session_factory() stub: 1st execute -> respawn rows, 2nd -> live + tasks, any further (deletes) -> a throwaway result.""" + db = AsyncMock() + rows_result = MagicMock() + rows_result.scalars.return_value.all.return_value = respawn_rows + live_result = MagicMock() + live_result.all.return_value = live + db.execute = AsyncMock(side_effect=[rows_result, live_result, MagicMock()]) + db.commit = AsyncMock() + ctx = MagicMock() + ctx.__aenter__ = AsyncMock(return_value=db) + ctx.__aexit__ = AsyncMock(return_value=False) + factory = MagicMock(return_value=ctx) + return factory, db + + +@pytest.mark.asyncio +async def test_loader_populates_dict_with_str_keys() -> None: + orch = _new_orchestrator() + tid = uuid4() + factory, _db = _mock_session_factory( + [_row(tid, count=3)], [SimpleNamespace(id=tid, status="in_progress")] + ) + with patch("roboco.db.base.get_session_factory", return_value=factory): + restored = await orch.restore_respawn_tracker() + assert restored == 1 + assert orch._pm_respawn_tracker[("be-pm", str(tid))]["count"] == _SEEDED_COUNT + + +@pytest.mark.asyncio +async def test_loader_skips_terminal_rows_and_deletes_them() -> None: + orch = _new_orchestrator() + done = uuid4() + factory, db = _mock_session_factory( + [_row(done)], [SimpleNamespace(id=done, status="completed")] + ) + with patch("roboco.db.base.get_session_factory", return_value=factory): + restored = await orch.restore_respawn_tracker() + assert restored == 0 + assert orch._pm_respawn_tracker == {} + db.commit.assert_awaited() # stale row deleted + committed + + +@pytest.mark.asyncio +async def test_loader_empty_on_exception() -> None: + orch = _new_orchestrator() + with patch( + "roboco.db.base.get_session_factory", side_effect=RuntimeError("db down") + ): + restored = await orch.restore_respawn_tracker() + assert restored == 0 + assert orch._pm_respawn_tracker == {} + + +# --------------------------------------------------------------------------- # +# Write-through on each gate mutation +# --------------------------------------------------------------------------- # + + +def _capture_persists(orch: AgentOrchestrator) -> list[tuple[str, str, dict[str, Any]]]: + """Record each _schedule_respawn_persist call, snapshotting the payload. + + The real scheduler copies the record (dict(record)) before the background + write, so the test must snapshot too — the gate mutates the same dict object + in place, so capturing the reference would show only its final state. + """ + captured: list[tuple[str, str, dict[str, Any]]] = [] + + def _cap(agent_slug: str, task_id: str, record: dict[str, Any]) -> None: + captured.append((agent_slug, task_id, dict(record))) + + cast("Any", orch)._schedule_respawn_persist = _cap + return captured + + +@pytest.mark.asyncio +async def test_new_entry_and_strike_schedule_persist() -> None: + orch = _new_orchestrator() + captured = _capture_persists(orch) + task_id = str(uuid4()) + task = {"id": task_id, "status": "pending"} + fake_audit = AsyncMock() + fake_audit.has_recent_tracing_gap = AsyncMock(return_value=False) + with ( + patch("roboco.services.audit.get_audit_service", return_value=fake_audit), + patch( + "roboco.services.notification.NotificationService", + return_value=AsyncMock(), + ), + ): + await orch._pm_respawn_should_gate("be-pm", task) # new entry + await orch._pm_respawn_should_gate("be-pm", task) # strike -> count 2 + assert len(captured) >= _MIN_PERSISTS + assert captured[0][0] == "be-pm" and captured[0][1] == task_id + assert captured[0][2]["count"] == 1 + assert captured[1][2]["count"] == _STRIKE_COUNT + + +@pytest.mark.asyncio +async def test_notified_flip_schedules_persist_with_notified_true() -> None: + orch = _new_orchestrator() + captured = _capture_persists(orch) + task_id = str(uuid4()) + task = {"id": task_id, "status": "pending"} + fake_audit = AsyncMock() + fake_audit.has_recent_tracing_gap = AsyncMock(return_value=False) + with ( + patch("roboco.services.audit.get_audit_service", return_value=fake_audit), + patch( + "roboco.services.notification.NotificationService", + return_value=AsyncMock(), + ), + ): + for _ in range(4): # 4th trips the gate + flips notified + await orch._pm_respawn_should_gate("be-pm", task) + assert any(c[2].get("notified") for c in captured), ( + "the notified flip must schedule a persist with notified=True" + ) + + +@pytest.mark.asyncio +async def test_tracing_reset_schedules_persist_with_reset_count() -> None: + orch = _new_orchestrator() + captured = _capture_persists(orch) + task_id = str(uuid4()) + task = {"id": task_id, "status": "blocked"} + fake_audit = AsyncMock() + fake_audit.has_recent_tracing_gap = AsyncMock(return_value=True) + with patch("roboco.services.audit.get_audit_service", return_value=fake_audit): + await orch._pm_respawn_should_gate("be-pm", task) # new entry + await orch._pm_respawn_should_gate("be-pm", task) # tracing reset + assert captured[-1][2]["count"] == 1 + assert captured[-1][2]["tracing_resets"] == 1 + + +# --------------------------------------------------------------------------- # +# Persist helper safety +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_persist_record_swallows_db_failure() -> None: + orch = _new_orchestrator() + with patch( + "roboco.db.base.get_session_factory", side_effect=RuntimeError("db down") + ): + # Must not raise — a persistence failure can never gate/un-gate a spawn. + await orch._persist_respawn_record( + "be-pm", + str(uuid4()), + {"count": 2, "last_check": datetime.now(UTC)}, + ) + + +# --------------------------------------------------------------------------- # +# Safety regression + transparency +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_restored_counter_trips_at_persisted_threshold_not_from_one() -> None: + """A restart mid-loop must NOT reset the strike count to 1.""" + orch = _new_orchestrator() + cast("Any", orch)._schedule_respawn_persist = MagicMock() + task_id = str(uuid4()) + # Simulate restore: count=3 (one below the trip), status-stable, resets spent. + orch._pm_respawn_tracker[("be-pm", task_id)] = { + "count": 3, + "last_status": "blocked", + "last_check": datetime(2026, 6, 26, tzinfo=UTC), + "tracing_resets": 3, + "notified": False, + } + task = {"id": task_id, "status": "blocked"} + fake_audit = AsyncMock() + fake_audit.has_recent_tracing_gap = AsyncMock(return_value=False) + with ( + patch("roboco.services.audit.get_audit_service", return_value=fake_audit), + patch( + "roboco.services.notification.NotificationService", + return_value=AsyncMock(), + ), + ): + gated = await orch._pm_respawn_should_gate("be-pm", task) + assert gated is True # fires on the next spawn, not re-counted from 1 + assert orch._pm_respawn_tracker[("be-pm", task_id)]["count"] == _TRIP_COUNT + + +@pytest.mark.asyncio +async def test_restart_midloop_continues_identically_to_no_restart() -> None: + """Transparency: the gate decision depends only on the dict contents. + + Drive a fresh orchestrator through N spawns; separately drive a second one + for K spawns, snapshot its dict (the restart point), load that snapshot into + a third orchestrator and continue — the tail must equal the no-restart tail. + """ + task = {"id": "t1", "status": "pending"} + spawns = 5 + restart_after = 2 + + async def _spawn(orch: AgentOrchestrator) -> bool: + fake_audit = AsyncMock() + fake_audit.has_recent_tracing_gap = AsyncMock(return_value=False) + with ( + patch("roboco.services.audit.get_audit_service", return_value=fake_audit), + patch( + "roboco.services.notification.NotificationService", + return_value=AsyncMock(), + ), + ): + return await orch._pm_respawn_should_gate("be-pm", task) + + no_restart = _new_orchestrator() + cast("Any", no_restart)._schedule_respawn_persist = MagicMock() + full = [await _spawn(no_restart) for _ in range(spawns)] + + pre = _new_orchestrator() + cast("Any", pre)._schedule_respawn_persist = MagicMock() + for _ in range(restart_after): + await _spawn(pre) + snapshot = copy.deepcopy(pre._pm_respawn_tracker) + + loaded = _new_orchestrator() + cast("Any", loaded)._schedule_respawn_persist = MagicMock() + loaded._pm_respawn_tracker = snapshot + tail = [await _spawn(loaded) for _ in range(spawns - restart_after)] + + assert tail == full[restart_after:]