From c27bf04a71f51aa6fca9195dc8ffb0b0a4e9a653 Mon Sep 17 00:00:00 2001 From: Your Name Date: Thu, 2 Jul 2026 12:12:41 +0800 Subject: [PATCH] feat: recover stale running agent runs --- agent-run-gateway.mjs | 81 ++++++++++++++ agent-run-gateway.test.mjs | 100 ++++++++++++++++++ docs/agent-run-worker-rollout-runbook.md | 12 +++ ...2-runtime-execution-assessment-20260702.md | 47 +++++--- .../memind-2-streaming-agent-runtime-plan.md | 69 ++++++++++++ scripts/agent-run-worker.mjs | 29 +++++ scripts/check-agent-run-worker.mjs | 11 ++ 7 files changed, 334 insertions(+), 15 deletions(-) diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index e7ebd70..7ba2723 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -493,6 +493,17 @@ export function createAgentRunGateway({ for (const row of rows) { statusCounts[row.status] = Number(row.count ?? 0); } + const [oldestRunningRows] = await pool.query( + `SELECT MIN(started_at) AS oldest_started_at + FROM h5_agent_runs + WHERE status = 'running'`, + ); + const oldestRunningStartedAt = oldestRunningRows[0]?.oldest_started_at == null + ? null + : Number(oldestRunningRows[0].oldest_started_at); + const oldestRunningAgeMs = oldestRunningStartedAt == null + ? 0 + : Math.max(0, nowMs() - oldestRunningStartedAt); return { autoDispatch, maxConcurrentRuns, @@ -500,18 +511,86 @@ export function createAgentRunGateway({ inFlight: inFlight.size, pendingDispatches: queuedDispatches.length, statusCounts, + oldestRunningStartedAt, + oldestRunningAgeMs, terminalStatuses: [...TERMINAL_STATUSES], toolGateway: toolGateway?.getStatus ? toolGateway.getStatus() : null, }; } + async function recoverStaleRunningRuns({ + staleMs = runTimeoutMs, + limit = maxConcurrentRuns, + dryRun = true, + reason = 'stale_running_timeout', + } = {}) { + const normalizedStaleMs = positiveInteger(staleMs, runTimeoutMs); + const normalizedLimit = positiveInteger(limit, maxConcurrentRuns); + const cutoff = nowMs() - normalizedStaleMs; + const [rows] = await pool.query( + `SELECT id, request_id, started_at, updated_at, attempts + FROM h5_agent_runs + WHERE status = 'running' AND started_at IS NOT NULL AND started_at <= ? + ORDER BY started_at ASC + LIMIT ?`, + [cutoff, normalizedLimit], + ); + const recovered = []; + for (const row of rows) { + const ageMs = Math.max(0, nowMs() - Number(row.started_at ?? 0)); + const item = { + id: row.id, + requestId: row.request_id, + startedAt: Number(row.started_at ?? 0), + updatedAt: Number(row.updated_at ?? 0), + attempts: Number(row.attempts ?? 0), + ageMs, + reason, + }; + if (!dryRun) { + const message = `agent run recovered from stale running state after ${ageMs}ms`; + const completedAt = nowMs(); + const [update] = await pool.query( + `UPDATE h5_agent_runs + SET status = 'failed', error_message = ?, updated_at = ?, completed_at = ? + WHERE id = ? AND status = 'running' AND started_at <= ?`, + [message, completedAt, completedAt, row.id, cutoff], + ); + if (Number(update?.affectedRows ?? 0) === 0) continue; + await appendEvent(row.id, 'stale_recovered', { + reason, + staleMs: normalizedStaleMs, + ageMs, + status: 'failed', + error: message, + }); + } + recovered.push(item); + } + return { + dryRun, + staleMs: normalizedStaleMs, + cutoff, + considered: rows.length, + recovered: dryRun ? 0 : recovered.length, + runs: recovered, + }; + } + async function dispatchQueuedRuns({ limit = maxConcurrentRuns } = {}) { + const staleRecovery = await recoverStaleRunningRuns({ + staleMs: runTimeoutMs, + limit: Math.max(maxConcurrentRuns, Number(limit) || maxConcurrentRuns), + dryRun: false, + reason: 'worker_dispatch_stale_recovery', + }); const dispatchLimit = Math.max(1, Number(limit) || maxConcurrentRuns); const available = Math.max(0, maxConcurrentRuns - inFlight.size - queuedDispatches.length); if (available <= 0) { return { considered: 0, dispatched: 0, + staleRecovery, queue: await getQueueStatus(), }; } @@ -530,6 +609,7 @@ export function createAgentRunGateway({ return { considered: rows.length, dispatched: rows.length, + staleRecovery, queue: await getQueueStatus(), }; } @@ -540,5 +620,6 @@ export function createAgentRunGateway({ dispatchRun, getQueueStatus, dispatchQueuedRuns, + recoverStaleRunningRuns, }; } diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index ee9808b..f78ddc8 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -33,6 +33,26 @@ function createFakePool() { } return [[...counts].map(([status, count]) => ({ status, count }))]; } + if (sql.includes('SELECT MIN(started_at) AS oldest_started_at')) { + const started = [...runs.values()] + .filter((row) => row.status === 'running' && row.started_at != null) + .map((row) => Number(row.started_at)); + return [[{ oldest_started_at: started.length ? Math.min(...started) : null }]]; + } + if (sql.includes('SELECT id, request_id, started_at, updated_at, attempts')) { + const [cutoff, limit = 1] = params; + return [[...runs.values()] + .filter((row) => row.status === 'running' && row.started_at != null && Number(row.started_at) <= Number(cutoff)) + .sort((a, b) => Number(a.started_at) - Number(b.started_at)) + .slice(0, Number(limit)) + .map((row) => ({ + id: row.id, + request_id: row.request_id, + started_at: row.started_at, + updated_at: row.updated_at, + attempts: row.attempts, + }))]; + } if (sql.includes('SELECT id') && sql.includes("status IN ('queued', 'retryable')")) { const limit = Number(params[0] ?? 1); return [[...runs.values()] @@ -87,6 +107,20 @@ function createFakePool() { }); return [{ affectedRows: 1 }]; } + if (sql.includes("WHERE id = ? AND status = 'running' AND started_at <= ?")) { + const [errorMessage, updatedAt, completedAt, id, cutoff] = params; + const row = runs.get(id); + if (!row || row.status !== 'running' || Number(row.started_at) > Number(cutoff)) { + return [{ affectedRows: 0 }]; + } + Object.assign(row, { + status: 'failed', + error_message: errorMessage, + updated_at: updatedAt, + completed_at: completedAt, + }); + return [{ affectedRows: 1 }]; + } if (sql.includes('UPDATE h5_agent_runs SET')) { const id = params.at(-1); const row = runs.get(id); @@ -610,3 +644,69 @@ test('external worker does not dispatch more runs when local queue is full', asy await waitFor(() => pool.runs.get(run1.id)?.status === 'succeeded'); assert.equal(pool.runs.get(run2.id).status, 'queued'); }); + +test('stale running recovery dry-run reports rows without mutating them', async () => { + const pool = createFakePool(); + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: {}, + autoDispatch: false, + runTimeoutMs: 1000, + }); + + const run = await gateway.createRun('user-1', { + requestId: 'req-stale-dry-run', + userMessage: { role: 'user', content: [] }, + }); + Object.assign(pool.runs.get(run.id), { + status: 'running', + attempts: 1, + started_at: Date.now() - 5000, + updated_at: Date.now() - 5000, + }); + + const result = await gateway.recoverStaleRunningRuns({ staleMs: 1000, dryRun: true }); + + assert.equal(result.considered, 1); + assert.equal(result.recovered, 0); + assert.equal(result.runs[0].id, run.id); + assert.equal(pool.runs.get(run.id).status, 'running'); + assert.equal( + pool.events.some((event) => event.runId === run.id && event.eventType === 'stale_recovered'), + false, + ); +}); + +test('stale running recovery marks old running rows failed with an event', async () => { + const pool = createFakePool(); + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: {}, + autoDispatch: false, + runTimeoutMs: 1000, + }); + + const run = await gateway.createRun('user-1', { + requestId: 'req-stale-apply', + userMessage: { role: 'user', content: [] }, + }); + Object.assign(pool.runs.get(run.id), { + status: 'running', + attempts: 1, + started_at: Date.now() - 5000, + updated_at: Date.now() - 5000, + }); + + const result = await gateway.recoverStaleRunningRuns({ staleMs: 1000, dryRun: false }); + + assert.equal(result.considered, 1); + assert.equal(result.recovered, 1); + assert.equal(pool.runs.get(run.id).status, 'failed'); + assert.match(pool.runs.get(run.id).error_message, /stale running state/); + assert.equal( + pool.events.some((event) => event.runId === run.id && event.eventType === 'stale_recovered'), + true, + ); +}); diff --git a/docs/agent-run-worker-rollout-runbook.md b/docs/agent-run-worker-rollout-runbook.md index 6ee7666..bf804da 100644 --- a/docs/agent-run-worker-rollout-runbook.md +++ b/docs/agent-run-worker-rollout-runbook.md @@ -180,6 +180,18 @@ MEMIND_AGENT_RUN_WORKER_EXPECT_RUNNING=1 node /Users/john/Project/Memind/scripts The worker LaunchAgent must have `KeepAlive=true` during all-user gray so a clean worker exit does not leave code-run queue processing stopped. +Check stale running recovery before and after worker changes: + +```bash +node /Users/john/Project/Memind/scripts/agent-run-worker.mjs --recover-stale --stale-ms 900000 --limit 5 +``` + +Apply only when the listed running rows are known stale: + +```bash +node /Users/john/Project/Memind/scripts/agent-run-worker.mjs --recover-stale --apply-recovery --stale-ms 900000 --limit 5 +``` + 8. Submit only code runs with validation metadata. Required user message metadata: diff --git a/docs/architecture/memind-2-runtime-execution-assessment-20260702.md b/docs/architecture/memind-2-runtime-execution-assessment-20260702.md index a4a41ad..c34c262 100644 --- a/docs/architecture/memind-2-runtime-execution-assessment-20260702.md +++ b/docs/architecture/memind-2-runtime-execution-assessment-20260702.md @@ -19,12 +19,12 @@ - Stream Controller 已具备 SSE headers、abort propagation、backpressure pipeline。 - Redis Router 已启用，承担 worker runtime state。 - Goose Worker Pool 已从固定单点走向四 worker 可观测调度。 -- Aider/OpenHands 已从普通聊天默认能力中剥离，进入 code mode 和后端灰度门禁；P6.3 已把 code run 从 goosed session extension 外移到 `agent-run-v1` Tool Gateway 协议，P6.4 已完成 Aider 真实执行 canary，P6.5 已完成 OpenHands 真实执行 canary，P6.6 已完成 external worker 精确接管 code-run canary，P6.7 已加入 Tool Gateway 产物校验与输出审计，P6.8 已安装 external worker LaunchAgent，P6.9 已完成带 validation 的 external worker 灰度 canary，P6.10 已加入 external worker 只读观测脚本，P6.11 已加入放量策略门禁，P6.12 已开启全用户长期灰度并通过普通测试用户真实路径，P6.13 已安装自动暂停 guard，P6.15 已让 H5 code-run 自动补 receipt validation 并恢复后端 required validation，P6.17/P6.18 已把 external worker 并发 2 通过 canary 并固化为当前 all-user gray 策略。 +- Aider/OpenHands 已从普通聊天默认能力中剥离，进入 code mode 和后端灰度门禁；P6.3 已把 code run 从 goosed session extension 外移到 `agent-run-v1` Tool Gateway 协议，P6.4 已完成 Aider 真实执行 canary，P6.5 已完成 OpenHands 真实执行 canary，P6.6 已完成 external worker 精确接管 code-run canary，P6.7 已加入 Tool Gateway 产物校验与输出审计，P6.8 已安装 external worker LaunchAgent，P6.9 已完成带 validation 的 external worker 灰度 canary，P6.10 已加入 external worker 只读观测脚本，P6.11 已加入放量策略门禁，P6.12 已开启全用户长期灰度并通过普通测试用户真实路径，P6.13 已安装自动暂停 guard，P6.15 已让 H5 code-run 自动补 receipt validation 并恢复后端 required validation，P6.17/P6.18 已把 external worker 并发 2 通过 canary 并固化为当前 all-user gray 策略，P8.1 已加入 stale running recovery。 - PG 和 MindSpace 仍保持生产数据边界，SLO 报告只做统计读取；P6.3-P6.15 不新增 schema migration，不删除或修改既有用户数据。 整体执行评分: 9.95 / 10。 -可以支撑当前 H5 streaming 稳定性改造的基础目标。自动采样、worker sidecar heartbeat、SLO 只读快照、SLO 日报定时器、SLO 日报保留策略、first-token EWMA、first-token p50/p95 窗口趋势、Tool Gateway Queue v0、外部 worker 接管入口、真实 worker canary、后端 code-mode canary、code-run 用户级灰度 gate、H5 页面编辑 UI canary、P6.3 Tool Gateway 协议化、P6.4 Aider 真实 canary、P6.5 OpenHands 真实 canary、P6.6 external worker code-run canary、P6.7 Tool Gateway guardrails、P6.8 worker LaunchAgent、P6.9 validated external worker canary、P6.10 worker observability、P6.11 rollout policy gates、P6.12 all-user gray、P6.13 auto-pause guard、P6.15 H5 validation metadata 和 P6.17/P6.18 worker concurrency 2 已经落地。主要剩余差距转为 running/stuck recovery、DB/session auth 瞬时错误收敛、更细粒度的任务级产物规范和用户可见进度/失败说明。 +可以支撑当前 H5 streaming 稳定性改造的基础目标。自动采样、worker sidecar heartbeat、SLO 只读快照、SLO 日报定时器、SLO 日报保留策略、first-token EWMA、first-token p50/p95 窗口趋势、Tool Gateway Queue v0、外部 worker 接管入口、真实 worker canary、后端 code-mode canary、code-run 用户级灰度 gate、H5 页面编辑 UI canary、P6.3 Tool Gateway 协议化、P6.4 Aider 真实 canary、P6.5 OpenHands 真实 canary、P6.6 external worker code-run canary、P6.7 Tool Gateway guardrails、P6.8 worker LaunchAgent、P6.9 validated external worker canary、P6.10 worker observability、P6.11 rollout policy gates、P6.12 all-user gray、P6.13 auto-pause guard、P6.15 H5 validation metadata、P6.17/P6.18 worker concurrency 2 和 P8.1 stale running recovery 已经落地。主要剩余差距转为 DB/session auth 瞬时错误收敛、更细粒度的任务级产物规范和用户可见进度/失败说明。 ## 实测结果 @@ -891,21 +891,32 @@ Data boundary: - observation: - Portal had one restart-window `read EADDRNOTAVAIL`, then LaunchAgent recovered and status returned `ok`. -## 下一步执行建议 - ### P8.1 Queue Lease / Stuck Run Recovery -- 为 `running` run 增加 lease / stale recovery: - - worker 启动和轮询时识别超过 TTL 的 running run。 - - 将超时 running 标记 failed 或 retryable，避免 worker/进程异常退出后永久卡住。 - - guard 同步读取 stale running 作为暂停依据。 -- 增加只读检查脚本输出: - - oldest running age - - stale running count - - recovery dry-run actions -- 同时修补 Portal auth/session DB 瞬时错误: - - `read EADDRNOTAVAIL` 不应作为 uncaught exception 杀掉 Portal。 - - session attach 失败应降级为 503/401 或跳过认证态，而不是进程退出。 +结果: 通过，worker dispatch 前会自动回收超过 timeout 的 running run，并提供手动 dry-run/apply 运维入口。 + +- 新增 `recoverStaleRunningRuns()`。 +- `dispatchQueuedRuns()` 自动执行 stale recovery。 +- `agent-run-worker.mjs` 新增: + - `--recover-stale` + - `--apply-recovery` + - `--stale-ms` +- `check-agent-run-worker.mjs` 和 runtime/status queue 摘要新增 running age。 +- 测试: + - 29 tests pass。 + - stale dry-run 不改状态。 + - stale apply 标记 failed 并写 `stale_recovered`。 +- 生产 synthetic stale run 验证: + - run id `9e22ad3d-ad5d-42b5-9ce3-24fe7aace3f5` + - status `failed` + - event `stale_recovered` +- post-check: + - queue empty + - worker running `--limit 2` + - guard `shouldPause=false` + - SLO `ok=true`, `failures=[]` + +## 下一步执行建议 ### P6.19 Task-level Artifact Validation @@ -913,6 +924,12 @@ Data boundary: - 页面编辑任务追加目标 `public/*.html` 校验。 - 文件/仓库任务根据任务参数声明目标相对路径。 +### P8.2 Portal DB/Auth Transient Error Hardening + +- `read EADDRNOTAVAIL` 不应作为 uncaught exception 杀掉 Portal。 +- session attach / auth verify 的 DB 瞬时错误应返回 503/401 或跳过认证态。 +- 增加回归测试，证明 DB read transient error 不会触发进程级 unhandled rejection。 + ### P6.22 Task Artifact UX and Failure Messages - 用户侧显示 queued/running/validation/succeeded/failed。 diff --git a/docs/architecture/memind-2-streaming-agent-runtime-plan.md b/docs/architecture/memind-2-streaming-agent-runtime-plan.md index 39947d2..621ef02 100644 --- a/docs/architecture/memind-2-streaming-agent-runtime-plan.md +++ b/docs/architecture/memind-2-streaming-agent-runtime-plan.md @@ -26,6 +26,7 @@ - P6.15 H5 Code-run Validation Metadata: 已完成，H5 code-run 自动声明 `.memind/agent-runs/.json` receipt 校验，生产已恢复 `MEMIND_AGENT_CODE_RUNS_REQUIRE_VALIDATION=1` 并通过普通测试用户真实路径。 - P6.17 Controlled Worker Concurrency Canary: 已完成，external worker 短窗口提升到并发 2，两条真实 code-run 同时 running 并全部通过 receipt validation。 - P6.18 Concurrency Rollout Policy: 已完成，生产 worker 和 Portal runtime/status 已对齐到并发 2，worker KeepAlive 保持开启，保留回滚备份。 +- P8.1 Queue Lease / Stuck Run Recovery: 已完成第一步，worker dispatch 前自动回收超时 running run，新增 `--recover-stale` dry-run/apply 运维入口，生产 synthetic stale run 验证通过。 - P5.15 Active Stream TTL Reconcile: 已按用户要求跳过，暂不做报表/定时 reconcile。 - P5 Worker Pool 运维化: 已完成第一步，Redis Router 支持 worker drain。 - P5.9 First-token Latency EWMA: 已完成，StreamController 会把首个 SSE chunk 延迟写入 Redis，SLO 报告已展示。 @@ -2736,6 +2737,74 @@ runtime/status: - Portal 在一次重启窗口中出现 `read EADDRNOTAVAIL` 并由 LaunchAgent 自动拉起，随后 `/api/status` 和 `/api/runtime/status` 恢复 `ok`。 - 后续建议优先做 running/stuck recovery 与 DB/session auth 异常不致命化，避免瞬时 DB/socket 错误扩大为进程退出。 +### 2026-07-02 P8.1 Queue Lease / Stuck Run Recovery + +目标: + +- 防止 worker/进程在 run 进入 `running` 后异常退出，导致该 run 永久卡住。 +- 在不新增 schema、不修改真实用户数据的前提下，先用 `started_at + runTimeoutMs` 作为最小 lease。 + +新增能力: + +- `agent-run-gateway.mjs`: + - 新增 `recoverStaleRunningRuns()`。 + - `dispatchQueuedRuns()` 每轮 dispatch 前自动回收超过 `runTimeoutMs` 的 `running` rows。 + - stale row 会标记为 `failed`，写入 `stale_recovered` event。 +- `scripts/agent-run-worker.mjs`: + - 新增只读入口: + - `node scripts/agent-run-worker.mjs --recover-stale --stale-ms 900000 --limit 5` + - 新增应用入口: + - `node scripts/agent-run-worker.mjs --recover-stale --apply-recovery --stale-ms 900000 --limit 5` +- `scripts/check-agent-run-worker.mjs`: + - queue 摘要增加: + - `oldestRunningStartedAt` + - `oldestRunningAgeMs` +- runtime/status 的 queue 摘要增加 running age 字段。 + +测试: + +- `node --check agent-run-gateway.mjs scripts/agent-run-worker.mjs scripts/check-agent-run-worker.mjs` 通过。 +- `node --test agent-run-gateway.test.mjs agent-run-routes.test.mjs` 通过，29 tests pass。 +- 新增测试: + - dry-run reports stale running rows without mutating them。 + - apply marks stale running rows failed and writes `stale_recovered` event。 + +生产部署: + +- 备份: + - `/Users/john/Project/memind_backups/20260702-120900-p81-stale-recovery` +- 已部署: + - bundled `server.mjs` + - bundled `scripts/agent-run-worker.mjs` + - `scripts/check-agent-run-worker.mjs` + - `RUNBOOK.txt` +- 已重启: + - `cn.tkmind.memind-portal` + - `cn.tkmind.memind-agent-run-worker` + +生产验证: + +- `/api/status` 返回 `ok`。 +- `/api/runtime/status.toolRuntime.queue`: + - `maxConcurrentRuns=2` + - `statusCounts={}` + - `oldestRunningStartedAt=null` + - `oldestRunningAgeMs=0` +- stale dry-run: + - `considered=0` + - `recovered=0` +- synthetic stale apply: + - run id `9e22ad3d-ad5d-42b5-9ce3-24fe7aace3f5` + - request id `p81-stale-recovery-1782965482040` + - recovery `considered=1`, `recovered=1` + - final status `failed` + - event chain includes `stale_recovered` +- post-check: + - worker running with `--limit 2` + - queue empty + - guard `shouldPause=false` + - SLO `ok=true`, `failures=[]` + ## 回滚策略 - P0: 修改前保留 `server.mjs` 备份；如启动失败，恢复备份并 `launchctl kickstart` Portal。 diff --git a/scripts/agent-run-worker.mjs b/scripts/agent-run-worker.mjs index cce6317..1b7ba0e 100644 --- a/scripts/agent-run-worker.mjs +++ b/scripts/agent-run-worker.mjs @@ -36,17 +36,26 @@ function parseArgs(argv) { const args = { once: false, status: false, + recoverStale: false, + applyRecovery: false, runId: '', pollMs: Number(process.env.MEMIND_AGENT_RUN_WORKER_POLL_MS ?? 1000), limit: Number(process.env.MEMIND_AGENT_RUN_WORKER_BATCH_SIZE ?? 1), + staleMs: Number(process.env.MEMIND_AGENT_RUN_STALE_RUNNING_MS ?? process.env.MEMIND_AGENT_RUN_TIMEOUT_MS ?? 15 * 60 * 1000), }; for (let i = 0; i < argv.length; i += 1) { const item = argv[i]; if (item === '--once') args.once = true; else if (item === '--status') args.status = true; + else if (item === '--recover-stale') args.recoverStale = true; + else if (item === '--apply-recovery') { + args.recoverStale = true; + args.applyRecovery = true; + } else if (item === '--run-id') args.runId = String(argv[++i] ?? '').trim(); else if (item === '--poll-ms') args.pollMs = Number(argv[++i] ?? args.pollMs); else if (item === '--limit') args.limit = Number(argv[++i] ?? args.limit); + else if (item === '--stale-ms') args.staleMs = Number(argv[++i] ?? args.staleMs); else if (item === '--help' || item === '-h') args.help = true; } return args; @@ -56,9 +65,12 @@ function printHelp() { console.log([ 'Usage:', ' node scripts/agent-run-worker.mjs [--once] [--status] [--run-id ] [--poll-ms 1000] [--limit 1]', + ' node scripts/agent-run-worker.mjs --recover-stale [--apply-recovery] [--stale-ms 900000]', '', 'Notes:', ' - Processes existing h5_agent_runs queued/retryable rows through the Tool Gateway queue.', + ' - Regular dispatch automatically fails stale running rows older than MEMIND_AGENT_RUN_TIMEOUT_MS.', + ' - --recover-stale is dry-run by default; --apply-recovery marks stale running rows failed.', ' - Does not create agent runs by itself.', ' - --run-id dispatches exactly one known run and avoids scanning the global queue.', ' - Set MEMIND_AGENT_RUN_AUTODISPATCH=0 on Portal only when this worker is ready to take over.', @@ -140,6 +152,21 @@ async function runOnce() { })); } +async function runStaleRecovery() { + const result = await gateway.recoverStaleRunningRuns({ + staleMs: args.staleMs, + limit: args.limit, + dryRun: !args.applyRecovery, + reason: args.applyRecovery ? 'manual_stale_recovery' : 'manual_stale_recovery_dry_run', + }); + console.log(JSON.stringify({ + ok: true, + mode: args.applyRecovery ? 'recover-stale-apply' : 'recover-stale-dry-run', + recovery: result, + queue: await gateway.getQueueStatus(), + }, null, 2)); +} + async function waitForIdle() { const startedAt = Date.now(); const maxWaitMs = Number(process.env.MEMIND_AGENT_RUN_WORKER_DRAIN_WAIT_MS ?? 20 * 60 * 1000); @@ -157,6 +184,8 @@ let exitCode = 0; try { if (args.status) { console.log(JSON.stringify({ ok: true, queue: await gateway.getQueueStatus() }, null, 2)); + } else if (args.recoverStale) { + await runStaleRecovery(); } else if (args.once || args.runId) { await runOnce(); } else { diff --git a/scripts/check-agent-run-worker.mjs b/scripts/check-agent-run-worker.mjs index 8b63b2e..ec3220b 100644 --- a/scripts/check-agent-run-worker.mjs +++ b/scripts/check-agent-run-worker.mjs @@ -108,6 +108,15 @@ async function readQueueSummary() { ? null : Number(lagRows[0].oldest_updated_at); + const [runningRows] = await conn.query( + `SELECT MIN(started_at) AS oldest_started_at + FROM h5_agent_runs + WHERE status = 'running'`, + ); + const oldestRunningStartedAt = runningRows[0]?.oldest_started_at == null + ? null + : Number(runningRows[0].oldest_started_at); + const [failedRows] = await conn.query( `SELECT id, request_id, error_message, updated_at FROM h5_agent_runs @@ -120,6 +129,8 @@ async function readQueueSummary() { statusCounts, oldestPendingUpdatedAt: oldestUpdatedAt, oldestPendingAgeMs: oldestUpdatedAt == null ? 0 : Math.max(0, Date.now() - oldestUpdatedAt), + oldestRunningStartedAt, + oldestRunningAgeMs: oldestRunningStartedAt == null ? 0 : Math.max(0, Date.now() - oldestRunningStartedAt), latestFailedRun: failedRows[0] ? { id: failedRows[0].id, requestId: failedRows[0].request_id,