Public: anyone can read this; the crew can change it.
Sentinel: queue:agent was degraded 50 minutes today: The agent queue has waited over 30 minutes: 2 queued, 2 of 2 slots running. #10209
Check · Review · Land · Live done — Landed and live in production.
mason/forge:mason/10192-sentinel-queue-agent-was-degraded-50-min → main 4 commits fork
3 files
changed +67
−4
src/crew/dispatch.ts +15 −3
| @@ -1567,8 +1567,14 @@ function pump( | ||
| 1567 | 1567 | noted.add(shut.id); |
| 1568 | 1568 | appendRunLog(app.config.dataDir, shut.id, "queued, not started: its repository's valve is shut (/flow)"); |
| 1569 | 1569 | } |
| 1570 | for (const queued of gated.order) { | |
| 1571 | if (queued.issueId !== null && runningIssues.has(queued.issueId)) continue; | |
| 1570 | for (const [at, queued] of gated.order.entries()) { | |
| 1571 | // One run per issue at a time: this one waits for its sibling, not for a slot. | |
| 1572 | // A sibling that hangs is still caught: the running run goes stale on its | |
| 1573 | // heartbeat or past NO_PROGRESS_MS and vitals names it under `stuck:`. | |
| 1574 | if (queued.issueId !== null && runningIssues.has(queued.issueId)) { | |
| 1575 | held.set(queued.id, "issue"); | |
| 1576 | continue; | |
| 1577 | } | |
| 1572 | 1578 | const fresh = getRun(app.db, queued.id); |
| 1573 | 1579 | if (!fresh || fresh.state !== "queued") continue; |
| 1574 | 1580 | // A fresh run queued before Archive is cancelled, not started, and |
| @@ -1583,7 +1589,13 @@ function pump( | ||
| 1583 | 1589 | } |
| 1584 | 1590 | continue; |
| 1585 | 1591 | } |
| 1586 | if (slots <= 0) break; | |
| 1592 | if (slots <= 0) { | |
| 1593 | // Out of slots: nothing more starts, but the runs behind this one that | |
| 1594 | // wait on a running sibling are still held by that, not by the slots. | |
| 1595 | for (const rest of gated.order.slice(at + 1)) | |
| 1596 | if (rest.issueId !== null && runningIssues.has(rest.issueId)) held.set(rest.id, "issue"); | |
| 1597 | break; | |
| 1598 | } | |
| 1587 | 1599 | // The week's pace holds every path that starts a run in an autonomous |
| 1588 | 1600 | // repository: assigned, review, feedback and check runs as much as picks (#8873). |
| 1589 | 1601 | if (home !== undefined && home.autonomy === "autonomous") { |
src/runs/held.ts +2 −1
| @@ -17,7 +17,7 @@ import type { Database } from "bun:sqlite"; | ||
| 17 | 17 | export const HELD_FRESH_MS = 2 * 60_000; |
| 18 | 18 | |
| 19 | 19 | /** Why the dispatcher held a run. */ |
| 20 | export type HoldCause = "valve" | "pace" | "limit" | "maintenance"; | |
| 20 | export type HoldCause = "valve" | "pace" | "limit" | "maintenance" | "issue"; | |
| 21 | 21 | |
| 22 | 22 | /** How vitals names each cause: "2 by a shut valve". */ |
| 23 | 23 | export const HOLD_WORDS: Readonly<Record<HoldCause, string>> = { |
| @@ -25,6 +25,7 @@ export const HOLD_WORDS: Readonly<Record<HoldCause, string>> = { | ||
| 25 | 25 | pace: "the week's pace", |
| 26 | 26 | limit: "an account limit", |
| 27 | 27 | maintenance: "planned maintenance", |
| 28 | issue: "its own issue's run still going", | |
| 28 | 29 | }; |
| 29 | 30 | |
| 30 | 31 | interface Pass { |
test/vitals-held-agent.test.ts +50
| @@ -138,3 +138,53 @@ test("with the real dispatcher, every queued run held is quiet while the pass is | ||
| 138 | 138 | servers.stop(); |
| 139 | 139 | } |
| 140 | 140 | }); |
| 141 | ||
| 142 | test("with the real dispatcher, a run queued behind its issue's running run is held, named, and not a stalled queue", async () => { | |
| 143 | app.db.query("DELETE FROM run").run(); | |
| 144 | const servers = await serve(app); | |
| 145 | const fake = join(dataDir, "fake-claude-2"); | |
| 146 | writeFileSync(fake, `#!/bin/sh\nif [ "$1" = "--version" ]; then echo "2.1.0 (fake)"; exit 0; fi\nexit 1\n`); | |
| 147 | chmodSync(fake, 0o755); | |
| 148 | // The previous test shut the valve; open it fully, so only the issue holds the sibling. | |
| 149 | app.db.query("DELETE FROM flow_valve WHERE repo_id = ?1").run(repoId); | |
| 150 | const issueId = crypto.randomUUID(); | |
| 151 | app.db | |
| 152 | .query("INSERT INTO issue (id, repo_id, number, kind, title, body, author_id, state, created_at) VALUES (?1, ?2, 7, 'issue', 'T', '', ?3, 'open', ?4)") | |
| 153 | .run(issueId, repoId, ownerId, new Date().toISOString()); | |
| 154 | const running = createRun(app.db, { repoId, kind: "agent", issueId }); | |
| 155 | // An unrelated run ahead of the sibling: the pass stops at it (no slot) and must still mark the sibling behind it. | |
| 156 | const ahead = createRun(app.db, { repoId, kind: "agent" }); | |
| 157 | const sibling = createRun(app.db, { repoId, kind: "agent", issueId }); | |
| 158 | const hourAgo = new Date(Date.now() - 2 * QUEUE_WAIT_MAX_MS).toISOString(); | |
| 159 | const earlier = new Date(Date.now() - 3 * QUEUE_WAIT_MAX_MS).toISOString(); | |
| 160 | app.db.query("UPDATE run SET created_at = ?1 WHERE id = ?2").run(hourAgo, sibling.id); | |
| 161 | app.db.query("UPDATE run SET created_at = ?1 WHERE id = ?2").run(earlier, ahead.id); | |
| 162 | app.db | |
| 163 | .query("UPDATE run SET state = 'running', started_at = ?1, heartbeat_at = ?1, created_at = ?1 WHERE id = ?2") | |
| 164 | .run(new Date().toISOString(), running.id); | |
| 165 | const crew = await startCrew(app, { | |
| 166 | forgeUrl: servers.url, | |
| 167 | localUrl: servers.url, | |
| 168 | binary: fake, | |
| 169 | concurrency: 1, | |
| 170 | budgetUsd: 1, | |
| 171 | runTimeoutMs: 60_000, | |
| 172 | // No tick inside the test: the pass read below is the one startCrew runs | |
| 173 | // before it returns, so nothing on a timer (the autonomy sweep, a slow | |
| 174 | // shard) can race it. | |
| 175 | tickMs: 600_000, | |
| 176 | }); | |
| 177 | try { | |
| 178 | // Every slot is taken (1 of 1), and the sibling is still marked. | |
| 179 | expect(heldRuns(app.db).get(sibling.id)).toBe("issue"); | |
| 180 | await crew.stop(); | |
| 181 | const now = Date.now(); | |
| 182 | // The unrelated run waits for a slot and is the queue; the sibling is named, not counted. | |
| 183 | const sentence = agentQueue(now)?.sentence ?? ""; | |
| 184 | expect(sentence).toContain("1 queued, 1 of 1 slots running"); | |
| 185 | expect(sentence).toContain("held on purpose: 1 by its own issue's run still going."); | |
| 186 | } finally { | |
| 187 | await crew.stop(); | |
| 188 | servers.stop(); | |
| 189 | } | |
| 190 | }); |