Alert an overseer when the respawn circuit-breaker pauses a wedged agent

The dispatcher's respawn guard already detects an agent repeatedly spawned
on the same task without advancing it, logs a warning, and skips further
spawns. But nothing surfaced that to a human, so a wedged agent could sit
silently with its task stalled.

When the guard trips, send a one-shot high-priority notification to the CEO
(tracked per (agent, task) so it fires once, not on every subsequent skipped
spawn, and resets if the agent later makes progress). Delivery is best-effort
so a notification failure never wedges dispatch.

Adds send_stuck_agent_notification to NotificationService and a coverage test
asserting the alert fires exactly once across multiple over-threshold spawns.

roboco/runtime/orchestrator.py

@@ -4083,8 +4083,8 @@ Start now: evidence(task_id="{task_id}")
         the counter resets. Once the count hits the threshold, the spawn
         is skipped and a warning logged; operators must intervene.
 
-        Tracing-gap reset (Task 13)
+        Tracing-gap reset
-        ---------------------------
+        -----------------
         With the gateway claim-time gates installed, a rule-following PM
         will hit ``PARENT_NOT_CLAIMED`` (a ``tracing_gap`` envelope) and
         the prompt will tell it to call the prerequisite verb first.
@@ -4120,6 +4120,7 @@ Start now: evidence(task_id="{task_id}")
         if await self._pm_made_rule_following_retry(agent_slug, task_id, record):
             record["count"] = 1
             record["last_check"] = now
+            record["notified"] = False
             return False
         record["count"] += 1
         record["last_check"] = now
@@ -4136,9 +4137,39 @@ Start now: evidence(task_id="{task_id}")
                     "Investigate prompt/schema drift or escalate manually."
                 ),
             )
+            # A skipped spawn pauses the loop but can't advance the task; alert
+            # an overseer once so a wedged agent isn't silently stranded.
+            if not record.get("notified"):
+                record["notified"] = True
+                await self._notify_stuck_agent(agent_slug, task_id, current_status)
             return True
         return False
 
+    async def _notify_stuck_agent(
+        self, agent_slug: str, task_id: str, task_status: str | None
+    ) -> None:
+        """One-shot alert to the CEO that an agent is wedged in a respawn loop.
+
+        Best-effort: a notification failure must not wedge dispatch, so any
+        error is logged and swallowed.
+        """
+        from roboco.services.notification import NotificationService
+
+        try:
+            await NotificationService().send_stuck_agent_notification(
+                task_id=task_id,
+                agent_slug=agent_slug,
+                task_status=task_status or "unknown",
+                to_agent="ceo",
+            )
+        except Exception as exc:
+            logger.warning(
+                "Failed to send stuck-agent notification",
+                agent_id=agent_slug,
+                task_id=task_id,
+                error=str(exc),
+            )
+
     async def _pm_made_rule_following_retry(
         self,
         agent_slug: str,

roboco/services/notification.py

@@ -87,6 +87,42 @@ class NotificationService:
             )
         )
 
+    async def send_stuck_agent_notification(
+        self,
+        task_id: str,
+        agent_slug: str,
+        task_status: str,
+        to_agent: str,
+    ) -> None:
+        """Alert an overseer that an agent is wedged in an unproductive loop.
+
+        Raised when the dispatcher's respawn circuit-breaker pauses further
+        spawns: the agent was respawned repeatedly without advancing the task,
+        so automatic recovery has given up and a human needs to intervene.
+        """
+        logger.info(
+            "Sending stuck-agent notification",
+            task_id=task_id,
+            agent=agent_slug,
+            to_agent=to_agent,
+        )
+        body = (
+            f"Agent {agent_slug} was repeatedly spawned on task {task_id} "
+            f"(status: {task_status}) without advancing it, so further automatic "
+            "spawns have been paused. Please investigate and intervene manually."
+        )
+        await self._create_notification(
+            CreateNotificationParams(
+                notification_type=NotificationType.BLOCKER_ESCALATION,
+                priority=NotificationPriority.HIGH,
+                from_agent="system",
+                to_agents=[to_agent],
+                subject=f"Agent {agent_slug} stuck on task {task_id}",
+                body=body,
+                related_task_id=task_id,
+            )
+        )
+
     async def send_qa_ready_notification(
         self,
         task_id: str,

tests/unit/runtime/test_pm_respawn_reset.py

@@ -80,7 +80,13 @@ async def test_three_no_progress_spawns_still_trip_kill() -> None:
     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):
+    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
@@ -88,6 +94,45 @@ async def test_three_no_progress_spawns_still_trip_kill() -> None:
         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."""