Persist the PM-respawn counter across orchestrator restarts (#275)

* feat(orchestrator): add respawn_tracker table + migration 051

Durable backing for AgentOrchestrator._pm_respawn_tracker (the PM-respawn
loop breaker). Kept only in memory it reset to count=1 on every restart,
re-burning the strike threshold against a still-wedged task. RespawnTrackerTable
mirrors WaitingRecordTable: composite PK (agent_slug, task_id) matching the
in-memory key; task_id is intentionally NOT a FK (the startup loader validates
against live tasks so a stale counter can't resurrect). Migration 051 verified
with a real alembic upgrade head + downgrade -1 + re-upgrade on Postgres.

* feat(orchestrator): persist the PM-respawn counter across restarts

The PM-respawn loop breaker (_pm_respawn_tracker) lived only in memory, so an
orchestrator restart reset a wedged task's strike count to 1 and re-burned the
whole threshold (4 spawns x container cost) before the gate fired again.

Write-through each gate mutation to the respawn_tracker table via a
fire-and-forget _schedule_respawn_persist (on the existing _bg_tasks strong-ref
set; a DB hiccup degrades to in-memory-only, never gates/un-gates a spawn), and
restore_respawn_tracker() repopulates the counter at startup, validating each
row against live tasks (drops terminal/missing) so a stale counter can't
resurrect against a fixed task. Best-effort + inert when the table is empty.
Cannot manufacture a spawn — the counter only ever suppresses one.

(_instances reconcile, the spec's other goal, already shipped as
_readopt_running_agents.)

* fix(types): cast Mapped[UUID] columns in project routes + self_heal

A clean `mypy roboco/ tests/` run surfaces 7 pre-existing errors in files this
branch doesn't touch: project-route handlers and self_heal_engine pass a
ProjectTable.id (declared Mapped[UUID] against SQLAlchemy's dialect UUID, so
mypy infers sqlalchemy.sql.sqltypes.UUID[Any]) where a uuid.UUID is expected.
An incremental .mypy_cache had hidden them. Apply the same targeted cast unblock
used for the prior batch; the deeper fix (migrating the ~88 Mapped[UUID]
columns to Mapped[uuid.UUID]) remains a separate dedicated task.

* docs(orchestrator): document respawn_tracker durability

Add the orchestrator runtime-state durability note to CLAUDE.md (respawn_tracker
write-through + restore; _instances reconciled-from-Docker) + the migration-051
narrative, and a CHANGELOG [Unreleased] Fixed entry. Also type-clean the new
respawn_tracker table test (cast __table__ to Table under TYPE_CHECKING).

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

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

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.<status>` 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}}`.
 

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

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

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
 # =============================================================================

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

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,

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

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
 
 

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:]