fix(agent99): rotate outcome reconciliation cohorts
This commit is contained in:
@@ -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",
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user