diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index d97bd7d9..643ada57 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -861,6 +861,19 @@ class AgentOrchestrator: # is in a loop — without this gate the orchestrator re-spawns every # tick forever (seen in production on 2026-04-22). self._pm_respawn_tracker: dict[tuple[str, str], dict[str, Any]] = {} + # Serializes the fire-and-forget respawn-tracker upserts so same-key + # persists COMMIT in schedule (logical) order — not whatever order their + # DB transactions resolve in. A respawn loop fires count 1->2->3->4 in + # quick succession, one fire-and-forget persist per increment; without + # serialization a slow stale persist (count=2) can commit AFTER a fast + # fresh one (count=4), leaving the durable row at the stale low count + # and re-burning the strike threshold on restart. The lock is acquired + # as the FIRST await in _persist_respawn_record, so acquisition order + # matches task creation order (FIFO ready queue), which is the logical + # schedule order. Persists are best-effort background writes, so + # serializing them never blocks the dispatcher hot path (the lock lives + # in the bg task, not the caller). + self._respawn_persist_lock = asyncio.Lock() # Board agents (Product Owner / Head of Marketing) get exactly ONE # review pass per assigned task: they have no verb to claim, plan, # delegate, or complete, so a respawn cannot advance the task and would @@ -4398,54 +4411,61 @@ class AgentOrchestrator: ``pk_respawn_tracker`` UniqueViolation, the durable count stuck at the first INSERT's value, and a restart re-burned the strike threshold — the exact re-burn this feature was built to stop (2026-06-27 live meltdown). - The single ``ON CONFLICT DO UPDATE`` upsert is race-free: concurrent - upserts on the same key serialize at row level. + The single ``ON CONFLICT DO UPDATE`` upsert is race-free at row level, BUT + fire-and-forget tasks can still COMMIT out of order: a slow stale persist + (count=2) scheduled first can resolve AFTER a fast fresh one (count=4) + scheduled second, leaving the durable row at the stale low count (same + re-burn on restart). The ``_respawn_persist_lock`` is acquired as the + FIRST await below, so acquisition order = task creation order = logical + schedule order, and commits land in that order — the durable row always + ends at the latest logical value. """ - try: - from uuid import UUID as _UUID + async with self._respawn_persist_lock: + try: + from uuid import UUID as _UUID - from sqlalchemy.dialects.postgresql import insert as pg_insert + from sqlalchemy.dialects.postgresql import insert as pg_insert - from roboco.db.base import get_session_factory - from roboco.db.tables import RespawnTrackerTable + from roboco.db.base import get_session_factory + from roboco.db.tables import RespawnTrackerTable - tid = _UUID(task_id) - now = datetime.now(UTC) - stmt = pg_insert(RespawnTrackerTable).values( - 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)), - updated_at=now, - ) - stmt = stmt.on_conflict_do_update( - index_elements=[ - RespawnTrackerTable.agent_slug, - RespawnTrackerTable.task_id, - ], - set_={ - "count": stmt.excluded.count, - "last_status": stmt.excluded.last_status, - "last_check": stmt.excluded.last_check, - "tracing_resets": stmt.excluded.tracing_resets, - "notified": stmt.excluded.notified, - "updated_at": stmt.excluded.updated_at, - }, - ) - session_factory = get_session_factory() - async with session_factory() as db: - await db.execute(stmt) - 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), - ) + tid = _UUID(task_id) + now = datetime.now(UTC) + stmt = pg_insert(RespawnTrackerTable).values( + 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)), + updated_at=now, + ) + stmt = stmt.on_conflict_do_update( + index_elements=[ + RespawnTrackerTable.agent_slug, + RespawnTrackerTable.task_id, + ], + set_={ + "count": stmt.excluded.count, + "last_status": stmt.excluded.last_status, + "last_check": stmt.excluded.last_check, + "tracing_resets": stmt.excluded.tracing_resets, + "notified": stmt.excluded.notified, + "updated_at": stmt.excluded.updated_at, + }, + ) + session_factory = get_session_factory() + async with session_factory() as db: + await db.execute(stmt) + 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). diff --git a/tests/unit/runtime/test_respawn_persistence.py b/tests/unit/runtime/test_respawn_persistence.py index 9dcba2d9..586f485b 100644 --- a/tests/unit/runtime/test_respawn_persistence.py +++ b/tests/unit/runtime/test_respawn_persistence.py @@ -11,6 +11,7 @@ manufacturing a spawn. from __future__ import annotations +import asyncio import copy from datetime import UTC, datetime from types import SimpleNamespace @@ -32,6 +33,7 @@ def _new_orchestrator() -> AgentOrchestrator: orch = AgentOrchestrator.__new__(AgentOrchestrator) cast("Any", orch)._pm_respawn_tracker = {} cast("Any", orch)._bg_tasks = set() + cast("Any", orch)._respawn_persist_lock = asyncio.Lock() return orch @@ -375,3 +377,78 @@ async def test_restart_midloop_continues_identically_to_no_restart() -> None: tail = [await _spawn(loaded) for _ in range(spawns - restart_after)] assert tail == full[restart_after:] + + +# --------------------------------------------------------------------------- # +# F096 — fire-and-forget persist commit ordering +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_persist_commits_in_schedule_order_not_out_of_order() -> None: + """Two fire-and-forget persists for the same ``(agent_slug, task_id)`` must + COMMIT in the order they were scheduled (logical order: count 1 -> 2 -> 3 + -> 4), not in whatever order their DB transactions happen to resolve. + + Without serialization, a slow stale persist (count=2) scheduled first can + commit AFTER a fast fresh persist (count=4) scheduled second, leaving the + durable row at the stale low count — and a restart re-burns the strike + threshold from that stale value (the exact re-burn this feature stops). The + fix serializes same-key persists so commit order = schedule order, and the + durable row ends at the latest logical value (4), never the stale one (2). + """ + orch = _new_orchestrator() + tid = str(uuid4()) + key = ("be-pm", tid) + now = datetime.now(UTC) + + committed_counts: list[int] = [] + # Per-execute delay, popped in execute-call order. The FIRST execute called + # (the stale count=2 persist, scheduled first = FIFO) is SLOW; the second + # (the fresh count=4 persist) is fast. Without serialization the fast one + # commits first -> durable ends at the stale 2. + delays = [0.05, 0.0] + + def _make_db() -> Any: + captured: dict[str, int] = {} + + async def _execute(stmt: Any) -> Any: + compiled = stmt.compile(dialect=postgresql.dialect()) + captured["count"] = int(dict(compiled.params)["count"]) + await asyncio.sleep(delays.pop(0)) + return MagicMock() + + async def _commit() -> None: + committed_counts.append(captured["count"]) + + db = MagicMock() + db.execute = _execute + db.commit = _commit + db.add = MagicMock() + ctx = MagicMock() + ctx.__aenter__ = AsyncMock(return_value=db) + ctx.__aexit__ = AsyncMock(return_value=False) + return ctx + + factory = MagicMock(side_effect=_make_db) + with patch("roboco.db.base.get_session_factory", return_value=factory): + # Schedule the stale persist (count=2) first. + orch._pm_respawn_tracker[key] = { + "count": 2, + "last_status": "pending", + "last_check": now, + "tracing_resets": 0, + "notified": False, + } + orch._schedule_respawn_persist("be-pm", tid, orch._pm_respawn_tracker[key]) + # Then mutate to the fresh value (count=4) and schedule the fresh + # persist second — logical order is 2 -> 4. + orch._pm_respawn_tracker[key]["count"] = 4 + orch._schedule_respawn_persist("be-pm", tid, orch._pm_respawn_tracker[key]) + + # Let both fire-and-forget bg tasks run to completion. + await asyncio.sleep(0.2) + + # Commits happened in schedule order (2 then 4), so the durable row ends at + # the LATEST logical value (4) — never the stale 2. + assert committed_counts == [2, 4]