fix(orchestrator): stop auditor alert-spawn rotation — ack as auditor on dispatch

The auditor respawned every ~3 min on the same stale rework alerts.

Root cause: _dispatch_audit_work's alert path fetched the SYSTEM-wide
"not fully acked" view (list_system_notifications), but the auditor is
read-only (no ack verb) and auditor_triage never acks — so once an alert
existed the CEO was the only party who could clear it, and the CEO hadn't
acked. The per-alert cooldown (PR #499) only paced a rotation through the
N un-acked alerts; it was a damper, not a fix.

Fix: fetch the auditor's OWN pending-ack view (GET /notifications authed
as the auditor -> list_for_agent, which filters acked_by for the auditor)
and ack the alert as the auditor on dispatch. Each alert is now a
one-shot, DB-persistent: the next tick cannot respawn on an alert the
auditor already observed — even one the CEO hasn't acked. Authed as the
auditor (not the system identity) so the route selects the per-recipient
view; HTTP rather than DB-direct so it shares the orchestrator's loop in
prod and stays loop-safe in the e2e harness (which runs _dispatch_audit_work
in its own asyncio.run loop, away from the app's DB engine).

e2e now asserts the alert is in acked_by for the auditor after dispatch —
the rotation-stopper itself, not just the spawn.

roboco/runtime/orchestrator.py

@@ -13578,22 +13578,32 @@ Never `commit`, never write code, never run `git`. PMs coordinate.
         Monitors: quality alert notifications + scheduled periodic sweeps
         Spawns: auditor
         """
-        alerts = await self._fetch_notifications(client, "alert")
+        # Alert path: dispatch the auditor ONCE per alert targeting it that the
-
+        # auditor has not observed yet. The auditor is read-only (no ack verb)
-        for alert in alerts:
+        # and auditor_triage never acks, so under the old system-wide fetch
-            targets = alert.get("to_agents", [])
+        # (every not-fully-acked alert) a stale rework alert stayed pending
-            # Resolve UUIDs to slugs and check if auditor is a target
+        # forever and the per-alert cooldown only paced a rotation that
-            target_slugs = [self._resolve_agent_slug(str(t)) for t in targets]
+        # respawned the auditor every ~3 min. Fetching the auditor's own
-            if "auditor" in target_slugs and not self._is_agent_active("auditor"):
+        # not-yet-acked-by-me view and acking on dispatch makes each alert a
-                if self._notification_spawn_cooled("auditor", alert.get("id")):
+        # one-shot, DB-persistent — the rotation cannot restart on a tick.
-                    continue
+        if not self._is_agent_active("auditor"):
-                await self.spawn_agent(
+            alert = await self._next_unobserved_audit_alert(client)
-                    agent_id="auditor",
+            if alert is not None:
-                    initial_prompt=self._build_audit_prompt(alert),
+                alert_id = str(alert["id"])
-                    spawned_by="_dispatch_audit_work",
+                if not self._notification_spawn_cooled("auditor", alert_id):
-                )
+                    await self.spawn_agent(
-                self._last_audit_spawn_at = datetime.now(UTC)
+                        agent_id="auditor",
-                return
+                        initial_prompt=self._build_audit_prompt(
+                            {
+                                "subject": alert.get("subject", ""),
+                                "body": alert.get("body", ""),
+                            }
+                        ),
+                        spawned_by="_dispatch_audit_work",
+                    )
+                    await self._ack_alert_as_auditor(client, alert_id)
+                    self._last_audit_spawn_at = datetime.now(UTC)
+                    return
 
         # Scheduled periodic audit sweep. Reuse the notification cooldown
         # pattern with a sentinel key as a one-tick breaker so a single
@@ -13627,6 +13637,68 @@ Never `commit`, never write code, never run `git`. PMs coordinate.
             return False
         return (datetime.now(UTC) - last).total_seconds() < interval
 
+    async def _next_unobserved_audit_alert(
+        self, client: httpx.AsyncClient
+    ) -> dict[str, Any] | None:
+        """Newest ALERT targeting the auditor that the auditor has not acked.
+
+        Fetches the auditor's OWN pending-ack view: ``GET /notifications`` authed
+        as the auditor routes through ``list_for_agent``, which filters
+        ``acked_by`` for the auditor — not the system-wide "not fully acked"
+        view ``list_system_notifications`` returns. The auditor is read-only (no
+        ack verb) and ``auditor_triage`` never acks, so under the system view a
+        stale rework alert stayed pending forever (the CEO hadn't acked either)
+        and the per-alert cooldown only paced a rotation that respawned the
+        auditor every ~3 min. The auditor's own view excludes alerts it has
+        already observed — even ones the CEO has not acked yet — so each alert
+        is a one-shot, not a rotation source. Authed as the auditor (not the
+        system identity) so the route selects the per-recipient view.
+        """
+        auditor_uuid = AGENT_UUIDS.get("auditor")
+        if not auditor_uuid:
+            return None
+        try:
+            resp = await client.get(
+                f"{self._api_url}/notifications",
+                params={"type_filter": "alert", "pending_ack_only": "true"},
+                headers=_agent_api_headers(auditor_uuid, "auditor"),
+            )
+            if resp.status_code == http_status.HTTP_200_OK:
+                items = resp.json().get("items", [])
+                return items[0] if items else None
+        except Exception as e:
+            logger.error("Fetch unobserved audit alert failed", error=str(e))
+        return None
+
+    async def _ack_alert_as_auditor(
+        self, client: httpx.AsyncClient, notification_id: str
+    ) -> None:
+        """Acknowledge an alert as the auditor on dispatch (HTTP, loop-safe).
+
+        The spawn IS the auditor's observation; the auditor has no ack verb
+        (read-only), so the orchestrator acks on its behalf via the API authed
+        as the auditor. This is the terminal one-shot response that clears the
+        alert from the auditor's pending view so the next tick cannot respawn
+        on the same alert. Best-effort: a failed ack degrades to the per-alert
+        cooldown guarding the next tick — it never blocks the dispatch.
+        """
+        auditor_uuid = AGENT_UUIDS.get("auditor")
+        if not auditor_uuid:
+            return
+        try:
+            resp = await client.post(
+                f"{self._api_url}/notifications/{notification_id}/ack",
+                headers=_agent_api_headers(auditor_uuid, "auditor"),
+            )
+            if resp.status_code != http_status.HTTP_200_OK:
+                logger.warning(
+                    "Ack audit alert as auditor failed",
+                    notification_id=notification_id,
+                    status=resp.status_code,
+                )
+        except Exception as e:
+            logger.warning("Ack audit alert as auditor failed", error=str(e))
+
     async def _has_recent_delivery_activity(
         self,
         client: httpx.AsyncClient,

tests/e2e_smoke/test_auditor_triggers.py

@@ -228,3 +228,30 @@ def test_reactive_alert_producer_spawns_auditor(
     prompt = call.kwargs["initial_prompt"]
     assert "QUALITY ALERT" in prompt
     assert "missing edge-case coverage" in prompt
+
+    # Rotation-stopper: dispatch acks the alert as the auditor (the auditor is
+    # read-only, no ack verb), so the auditor's pending-ack view no longer
+    # returns it and the next tick cannot respawn on this same alert. Without
+    # this ack the per-alert cooldown only paced a rotation through every
+    # stale not-fully-acked alert — the loop being fixed.
+    async def _acked_by_auditor(session: AsyncSession) -> bool:
+        from roboco.db.tables import NotificationTable
+        from sqlalchemy import select
+
+        row = (
+            (
+                await session.execute(
+                    select(NotificationTable).where(
+                        NotificationTable.related_task_id == task_id,
+                        NotificationTable.type == NotificationType.ALERT,
+                    )
+                )
+            )
+            .scalars()
+            .one()
+        )
+        return auditor_id in row.acked_by
+
+    assert stack.run_db(_acked_by_auditor), (
+        "alert not acked as auditor after dispatch — rotation not stopped"
+    )

tests/unit/runtime/test_scheduled_audit_trigger.py

@@ -26,10 +26,13 @@ def orch() -> AgentOrchestrator:
 async def _run_dispatch(
     orch: AgentOrchestrator, *, tasks: list[dict] | None = None
 ) -> MagicMock:
-    """Run _dispatch_audit_work with notifications empty and return the spawn mock."""
+    """Run _dispatch_audit_work with no unobserved alerts and return the spawn mock."""
     client = MagicMock()
     with (
-        patch.object(orch, "_fetch_notifications", new=AsyncMock(return_value=[])),
+        patch.object(
+            orch, "_next_unobserved_audit_alert", new=AsyncMock(return_value=None)
+        ),
+        patch.object(orch, "_ack_alert_as_auditor", new=AsyncMock()),
         patch.object(orch, "_fetch_tasks", new=AsyncMock(return_value=tasks or [])),
         patch.object(orch, "_is_agent_active", return_value=False),
         patch.object(orch, "spawn_agent", new=AsyncMock()) as spawn_mock,
@@ -111,7 +114,10 @@ async def test_breaker_skips_when_auditor_active(orch: AgentOrchestrator) -> Non
     with (
         patch.object(settings, "audit_interval_seconds", 21600),
         patch("roboco.runtime.orchestrator.datetime", wraps=datetime) as dt_mock,
-        patch.object(orch, "_fetch_notifications", new=AsyncMock(return_value=[])),
+        patch.object(
+            orch, "_next_unobserved_audit_alert", new=AsyncMock(return_value=None)
+        ),
+        patch.object(orch, "_ack_alert_as_auditor", new=AsyncMock()),
         patch.object(orch, "_fetch_tasks", new=AsyncMock(return_value=[])),
         patch.object(orch, "_is_agent_active", return_value=True),
         patch.object(orch, "spawn_agent", new=AsyncMock()) as spawn_mock,
@@ -129,8 +135,7 @@ async def test_reactive_alert_stamps_last_spawn_and_blocks_scheduled(
     """A reactive alert spawn records _last_audit_spawn_at and returns early."""
     client = MagicMock()
     alert = {
-        "id": "a1",
+        "id": "11111111-1111-1111-1111-111111111111",
-        "to_agents": ["auditor"],
         "subject": "Coverage gap",
         "body": "Test",
     }
@@ -138,7 +143,10 @@ async def test_reactive_alert_stamps_last_spawn_and_blocks_scheduled(
     with (
         patch.object(settings, "audit_interval_seconds", 21600),
         patch("roboco.runtime.orchestrator.datetime", wraps=datetime) as dt_mock,
-        patch.object(orch, "_fetch_notifications", new=AsyncMock(return_value=[alert])),
+        patch.object(
+            orch, "_next_unobserved_audit_alert", new=AsyncMock(return_value=alert)
+        ),
+        patch.object(orch, "_ack_alert_as_auditor", new=AsyncMock()),
         patch.object(orch, "_fetch_tasks", new=AsyncMock(return_value=[])),
         patch.object(orch, "_is_agent_active", return_value=False),
         patch.object(orch, "spawn_agent", new=AsyncMock()) as spawn_mock,
@@ -149,6 +157,43 @@ async def test_reactive_alert_stamps_last_spawn_and_blocks_scheduled(
     assert orch._last_audit_spawn_at == now
 
 
+@pytest.mark.anyio
+async def test_reactive_alert_acks_so_it_cannot_rotate(
+    orch: AgentOrchestrator,
+) -> None:
+    """The alert is acked as the auditor on dispatch, so the next tick —
+    which would otherwise re-fetch the same stale alert and respawn — finds
+    no unobserved alert and falls through to the scheduled path (blocked by
+    the just-stamped _last_audit_spawn_at). This is the rotation-stopper.
+    """
+    client = MagicMock()
+    alert_id = "22222222-2222-2222-2222-222222222222"
+    alert = {
+        "id": alert_id,
+        "subject": "Rework alert",
+        "body": "Task entered needs_revision",
+    }
+    now = datetime.now(UTC)
+    ack_mock = AsyncMock()
+    fetch_mock = AsyncMock(side_effect=[alert, None])
+    with (
+        patch.object(settings, "audit_interval_seconds", 21600),
+        patch("roboco.runtime.orchestrator.datetime", wraps=datetime) as dt_mock,
+        patch.object(orch, "_next_unobserved_audit_alert", new=fetch_mock),
+        patch.object(orch, "_ack_alert_as_auditor", new=ack_mock),
+        patch.object(orch, "_fetch_tasks", new=AsyncMock(return_value=[])),
+        patch.object(orch, "_is_agent_active", return_value=False),
+        patch.object(orch, "spawn_agent", new=AsyncMock()) as spawn_mock,
+    ):
+        dt_mock.now.return_value = now
+        await orch._dispatch_audit_work(client)  # tick 1: alert -> spawn + ack
+        await orch._dispatch_audit_work(client)  # tick 2: no alert -> scheduled
+    assert spawn_mock.await_count == 1  # not respawned on tick 2
+    assert ack_mock.await_count == 1
+    # acked as the auditor saw it: (client, alert_id)
+    assert ack_mock.await_args.args[1] == alert_id
+
+
 @pytest.mark.anyio
 async def test_cooldown_zero_disables_scheduled_sweeps(orch: AgentOrchestrator) -> None:
     """ROBOCO_AUDIT_INTERVAL_SECONDS=0 disables scheduled sweeps entirely."""