"""PM respawn loop guard counts no-progress only, not rule-following retries. Background ---------- ``_pm_respawn_should_gate`` is the orchestrator's circuit breaker against respawning the same PM on the same task forever. Before Task 13 it counted any spawn whose task status didn't advance as a "strike", and killed respawns after three of them. That works for an agent that's hung — but with the gateway claim-time gates installed in Phase 3, a rule-following PM that hits ``PARENT_NOT_CLAIMED`` (a ``tracing_gap`` envelope) and is told by the prompt to call ``i_will_plan`` first will keep returning to the same status. Each retry would trip a strike even though the agent did exactly what the gateway told it to do. The fix: when audit_log shows the agent emitted a ``gateway.rejected`` envelope with ``reason == "tracing_gap"`` since the last spawn, treat that as forward progress (rule-following) and reset the strike count instead of incrementing. """ from __future__ import annotations from datetime import UTC, datetime from typing import Any, cast from unittest.mock import AsyncMock, patch from uuid import uuid4 import pytest from roboco.runtime.orchestrator import AgentOrchestrator from roboco.seeds.initial_data import AGENT_UUIDS 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 = {} return orch @pytest.mark.asyncio async def test_three_tracing_gap_responses_do_not_trip_kill() -> None: """A PM rule-followed three times must NOT be killed. Each spawn the audit_log shows a fresh ``gateway.rejected`` row with ``reason == "tracing_gap"`` since the last check. That's the rule-following retry pattern — counter must reset every time. """ orch = _new_orchestrator() task_id = str(uuid4()) task = {"id": task_id, "status": "pending"} fake_audit = AsyncMock() fake_audit.has_recent_tracing_gap = AsyncMock(return_value=True) spawn_attempts = 5 with patch("roboco.services.audit.get_audit_service", return_value=fake_audit): for _ in range(spawn_attempts): should_gate = await orch._pm_respawn_should_gate("be-pm", task) assert should_gate is False # Audit was consulted on every call past the first record-creation. expected_audit_calls = spawn_attempts - 1 assert fake_audit.has_recent_tracing_gap.await_count >= expected_audit_calls @pytest.mark.asyncio async def test_unending_tracing_gap_is_bounded_and_eventually_trips() -> None: """A task whose EVERY respawn trips the same tracing_gap must still die. The rule-following reset is bounded by ``pm_respawn_max_tracing_resets``. Before this bound a permanently-wedged task (e.g. a cold-respawned PM that can never satisfy the unblock journal-decision gate) emitted a tracing_gap on every spawn, reset the strike counter every time, and respawned forever — the production bleed. With the cap, resets are exhausted and strikes accrue until the gate fires. """ orch = _new_orchestrator() 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), patch( "roboco.services.notification.NotificationService", return_value=AsyncMock(), ), ): results = [ await orch._pm_respawn_should_gate("main-pm", task) for _ in range(12) ] # The bug was that this list would be all-False forever. The gate must # trip at least once now that the reset budget is finite. assert any(results), "tracing_gap loop must eventually be gated" @pytest.mark.asyncio async def test_three_no_progress_spawns_still_trip_kill() -> None: """When there is NO tracing_gap envelope, the strike logic still bites. The classic stuck-loop case must still be detected: agent silently hung, no envelopes emitted, status doesn't change. Strike count increments each spawn and the kill fires after the threshold. """ orch = _new_orchestrator() 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(), ), ): # Spawns 1, 2, 3 — under threshold, all allowed. for _ in range(3): assert await orch._pm_respawn_should_gate("be-pm", task) is False # Spawn 4 — count now exceeds _PM_RESPAWN_MAX_UNPRODUCTIVE = 3. assert await orch._pm_respawn_should_gate("be-pm", task) is True @pytest.mark.asyncio async def test_stuck_loop_alerts_overseer_once() -> None: """When the loop guard bites, the CEO is alerted exactly once. The guard pauses further spawns but can't advance the task. Without an alert a wedged agent is silently stranded. The alert must fire on the first kill and NOT repeat on every subsequent skipped spawn. """ orch = _new_orchestrator() task_id = str(uuid4()) task = {"id": task_id, "status": "pending"} fake_audit = AsyncMock() fake_audit.has_recent_tracing_gap = AsyncMock(return_value=False) notifier = AsyncMock() notifier.send_stuck_agent_notification = AsyncMock() with ( patch("roboco.services.audit.get_audit_service", return_value=fake_audit), patch( "roboco.services.notification.NotificationService", return_value=notifier, ), ): # Under threshold — no kill, no alert. for _ in range(3): assert await orch._pm_respawn_should_gate("be-pm", task) is False notifier.send_stuck_agent_notification.assert_not_awaited() # Two over-threshold spawns — both killed, but only ONE alert. assert await orch._pm_respawn_should_gate("be-pm", task) is True assert await orch._pm_respawn_should_gate("be-pm", task) is True notifier.send_stuck_agent_notification.assert_awaited_once() kwargs = notifier.send_stuck_agent_notification.await_args.kwargs assert kwargs["to_agent"] == "ceo" assert kwargs["task_id"] == task_id assert kwargs["agent_slug"] == "be-pm" @pytest.mark.asyncio async def test_status_change_resets_strike_count() -> None: """Pre-existing reset path on real status change must keep working.""" orch = _new_orchestrator() task_id = str(uuid4()) 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): # Two strikes on pending. await orch._pm_respawn_should_gate( "be-pm", {"id": task_id, "status": "pending"} ) await orch._pm_respawn_should_gate( "be-pm", {"id": task_id, "status": "pending"} ) # Status advances — counter must drop back to 1. await orch._pm_respawn_should_gate( "be-pm", {"id": task_id, "status": "in_progress"} ) record = orch._pm_respawn_tracker[("be-pm", task_id)] assert record["count"] == 1 assert record["last_status"] == "in_progress" @pytest.mark.asyncio async def test_audit_query_uses_correct_agent_uuid_and_task_id() -> None: """The audit query must scope to (agent UUID, task UUID, since).""" orch = _new_orchestrator() task_id = str(uuid4()) task = {"id": task_id, "status": "pending"} 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): # First call seeds the record (no audit query yet). await orch._pm_respawn_should_gate("be-pm", task) # Second call should consult audit with be-pm's UUID + task UUID + since. await orch._pm_respawn_should_gate("be-pm", task) fake_audit.has_recent_tracing_gap.assert_awaited() kwargs = fake_audit.has_recent_tracing_gap.call_args.kwargs assert str(kwargs["agent_id"]) == AGENT_UUIDS["be-pm"] assert str(kwargs["task_id"]) == task_id assert isinstance(kwargs["since"], datetime) assert kwargs["since"].tzinfo is UTC @pytest.mark.asyncio async def test_unknown_slug_falls_back_to_status_only() -> None: """A slug not in AGENT_UUIDS must NOT crash; just skip the audit query. Defensive: if the slug map drifts, the kill loop guard still works, only it won't get the tracing_gap reset. """ orch = _new_orchestrator() 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): # Slug not present in AGENT_UUIDS — should not raise. for _ in range(3): assert await orch._pm_respawn_should_gate("not-a-real-slug", task) is False # Threshold trip path still reachable. assert await orch._pm_respawn_should_gate("not-a-real-slug", task) is True # Audit is never consulted because slug didn't resolve. fake_audit.has_recent_tracing_gap.assert_not_awaited() @pytest.mark.asyncio async def test_audit_query_failure_does_not_crash_gate() -> None: """If audit lookup raises, fall back to the legacy strike behavior. Audit is observability — it must never block orchestrator decisions. """ orch = _new_orchestrator() task_id = str(uuid4()) task = {"id": task_id, "status": "pending"} fake_audit = AsyncMock() fake_audit.has_recent_tracing_gap = AsyncMock(side_effect=RuntimeError("db down")) with patch("roboco.services.audit.get_audit_service", return_value=fake_audit): # Strikes 1-3: allowed. 4th: gated. Same as the no-tracing-gap case. for _ in range(3): assert await orch._pm_respawn_should_gate("be-pm", task) is False assert await orch._pm_respawn_should_gate("be-pm", task) is True