From 096f61ed69987636b77f1dfe307171c1bed052e2 Mon Sep 17 00:00:00 2001
From: ogt <ogt@awoooi.local>
Date: Wed, 15 Jul 2026 05:03:57 +0800
Subject: [PATCH] fix(agent99): rotate outcome reconciliation cohorts

---
 .../agent99_controlled_dispatch_ledger.py     | 25 ++++++++++-
 .../test_agent99_controlled_dispatch_p1.py    | 43 +++++++++++++++++++
 2 files changed, 67 insertions(+), 1 deletion(-)

diff --git a/apps/api/src/services/agent99_controlled_dispatch_ledger.py b/apps/api/src/services/agent99_controlled_dispatch_ledger.py
index f345d5e37..16c7f7dae 100644
--- a/apps/api/src/services/agent99_controlled_dispatch_ledger.py
+++ b/apps/api/src/services/agent99_controlled_dispatch_ledger.py
@@ -2215,14 +2215,37 @@ class PostgresAgent99DispatchLedger:
                         ]),
                     )
                     # Prioritize observed delivery/outcome lanes; pending crash
-                    # recovery candidates are polled after them.
+                    # recovery candidates are polled after them. Heartbeat is
+                    # also the durable poll cursor: each selected cohort is
+                    # touched below so unresolved old runs cannot starve newer
+                    # outcomes forever.
                     .order_by(
                         (AwoooPRunState.state == "pending").asc(),
+                        AwoooPRunState.heartbeat_at.asc().nullsfirst(),
                         AwoooPRunState.created_at.asc(),
                     )
                     .limit(max(1, min(int(limit), 100)))
                 )
                 rows = result.all()
+                if rows:
+                    await db.execute(
+                        update(AwoooPRunState)
+                        .where(
+                            AwoooPRunState.project_id
+                            == (project_id or "awoooi"),
+                            AwoooPRunState.agent_id
+                            == AGENT99_DISPATCH_AGENT_ID,
+                            AwoooPRunState.run_id.in_([
+                                row.run_id for row in rows
+                            ]),
+                            AwoooPRunState.state.in_([
+                                "pending",
+                                "running",
+                                "waiting_tool",
+                            ]),
+                        )
+                        .values(heartbeat_at=_utc_now_naive())
+                    )
         except Exception as exc:
             logger.warning(
                 "agent99_dispatch_reconciliation_list_failed",
diff --git a/apps/api/tests/test_agent99_controlled_dispatch_p1.py b/apps/api/tests/test_agent99_controlled_dispatch_p1.py
index f7bfc4b73..eeea4cd7d 100644
--- a/apps/api/tests/test_agent99_controlled_dispatch_p1.py
+++ b/apps/api/tests/test_agent99_controlled_dispatch_p1.py
@@ -148,6 +148,49 @@ class _Context:
         return False
 
 
+@pytest.mark.asyncio
+async def test_reconciliation_poll_cohort_rotates_by_durable_heartbeat(
+    monkeypatch,
+) -> None:
+    identity = _identity()
+    statements = []
+
+    class RowsResult:
+        def all(self):
+            return [
+                SimpleNamespace(
+                    run_id=identity.run_id,
+                    state="waiting_tool",
+                    error_detail=json.dumps({"identity": identity.public_dict()}),
+                )
+            ]
+
+    class ReconcileDb:
+        async def execute(self, statement):
+            statements.append(statement)
+            return RowsResult() if len(statements) == 1 else _ScalarResult()
+
+    monkeypatch.setattr(
+        ledger_module,
+        "get_db_context",
+        lambda _project_id: _Context(ReconcileDb()),
+    )
+
+    items = await PostgresAgent99DispatchLedger().list_reconcilable(
+        project_id="awoooi",
+        limit=20,
+    )
+
+    assert len(items) == 1
+    assert items[0]["identity"] == identity
+    assert len(statements) == 2
+    select_sql = str(statements[0])
+    update_sql = str(statements[1])
+    assert "awooop_run_state.heartbeat_at ASC NULLS FIRST" in select_sql
+    assert "UPDATE awooop_run_state" in update_sql
+    assert "heartbeat_at" in update_sql
+
+
 @pytest.mark.asyncio
 async def test_reservation_claim_tokens_are_unique_and_running_is_reconcile_only(
     monkeypatch,