From 3aae12d51160ad7225d4bc9c231ad6f7bd5b92e2 Mon Sep 17 00:00:00 2001 From: Nick M Date: Sun, 12 Jul 2026 04:16:56 -0500 Subject: [PATCH] feat(miner): add lease + expiry sweep to reclaim stuck portfolio-queue items MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Portfolio-queue items were flipped to in_progress with no lease and no expiry, so a crashed or killed process stranded an item in_progress forever with no recovery — unlike the claim-ledger and worktree-allocator stores in the same package, which already sweep their own stuck rows. Adds a leased_at column (idempotently migrated onto existing stores) stamped when an item is claimed and cleared when it leaves in_progress, a listInProgress()/reclaimStuckItem() pair on the store, and a pure portfolio-queue-expiry module (findStuckItems + sweepStuckItems) mirroring claim-ledger-expiry. The base QueueEntry shape every existing caller relies on is unchanged; lease data is exposed via a separate projection. Closes #4827 --- .../lib/portfolio-queue-expiry.d.ts | 20 ++ .../lib/portfolio-queue-expiry.js | 49 +++++ .../lib/portfolio-queue-manager.d.ts | 2 + .../lib/portfolio-queue-manager.js | 11 + .../gittensory-miner/lib/portfolio-queue.d.ts | 10 + .../gittensory-miner/lib/portfolio-queue.js | 58 +++++- packages/gittensory-miner/package.json | 2 +- .../unit/miner-portfolio-queue-expiry.test.ts | 190 ++++++++++++++++++ 8 files changed, 335 insertions(+), 7 deletions(-) create mode 100644 packages/gittensory-miner/lib/portfolio-queue-expiry.d.ts create mode 100644 packages/gittensory-miner/lib/portfolio-queue-expiry.js create mode 100644 test/unit/miner-portfolio-queue-expiry.test.ts diff --git a/packages/gittensory-miner/lib/portfolio-queue-expiry.d.ts b/packages/gittensory-miner/lib/portfolio-queue-expiry.d.ts new file mode 100644 index 0000000000..f52ae5ed8c --- /dev/null +++ b/packages/gittensory-miner/lib/portfolio-queue-expiry.d.ts @@ -0,0 +1,20 @@ +import type { QueueEntry, QueueLeaseEntry } from "./portfolio-queue.js"; + +export declare const DEFAULT_MAX_LEASE_MS: number; + +export type PortfolioQueueExpiryStore = { + listInProgress(): QueueLeaseEntry[]; + reclaimStuckItem(repoFullName: string, identifier: string): QueueEntry | null; +}; + +export function findStuckItems( + items: QueueLeaseEntry[], + nowMs: number, + maxLeaseMs: number, +): QueueLeaseEntry[]; + +export function sweepStuckItems( + store: PortfolioQueueExpiryStore, + nowMs: number, + maxLeaseMs?: number, +): QueueEntry[]; diff --git a/packages/gittensory-miner/lib/portfolio-queue-expiry.js b/packages/gittensory-miner/lib/portfolio-queue-expiry.js new file mode 100644 index 0000000000..209fae0e71 --- /dev/null +++ b/packages/gittensory-miner/lib/portfolio-queue-expiry.js @@ -0,0 +1,49 @@ +/** PURE — no IO, no Date, no random (#4827). Mirror of claim-ledger-expiry.js for the portfolio-queue store: a + * crashed/killed process leaves its item stuck 'in_progress' forever, so sweep leases older than a bound back to + * 'queued'. */ + +// A generous default: a real attempt rarely holds a single portfolio item for long, so 30 minutes without the row +// leaving 'in_progress' strongly implies the owning process died rather than that it is still working. +export const DEFAULT_MAX_LEASE_MS = 30 * 60 * 1000; + +function leaseAgeMs(item, nowMs) { + const leasedAtMs = Date.parse(item.leasedAt); + if (!Number.isFinite(leasedAtMs)) return null; + return nowMs - leasedAtMs; +} + +/** + * Return in-flight items whose lease age is strictly greater than `maxLeaseMs`. An item whose age equals + * `maxLeaseMs` exactly is still within the window (not stuck). Items that are not 'in_progress', or whose + * `leasedAt` is missing/unparseable, are never returned. + */ +export function findStuckItems(items, nowMs, maxLeaseMs) { + if (!Number.isFinite(nowMs) || nowMs < 0) throw new Error("invalid_now_ms"); + if (!Number.isFinite(maxLeaseMs) || maxLeaseMs < 0) throw new Error("invalid_max_lease_ms"); + if (!Array.isArray(items)) throw new Error("invalid_items"); + + const stuck = []; + for (const item of items) { + if (item?.status !== "in_progress") continue; + const ageMs = leaseAgeMs(item, nowMs); + if (ageMs === null) continue; + if (ageMs > maxLeaseMs) stuck.push(item); + } + return stuck; +} + +/** + * Reclaim every stuck in-flight item back to 'queued', returning the reclaimed entries. `store.listInProgress()` + * supplies the lease-annotated rows and `store.reclaimStuckItem()` performs the atomic per-item flip — the same + * store/sweep split sweepExpiredClaims uses. + */ +export function sweepStuckItems(store, nowMs, maxLeaseMs = DEFAULT_MAX_LEASE_MS) { + const inProgress = store.listInProgress(); + const stuck = findStuckItems(inProgress, nowMs, maxLeaseMs); + const reclaimed = []; + for (const item of stuck) { + const updated = store.reclaimStuckItem(item.repoFullName, item.identifier); + if (updated) reclaimed.push(updated); + } + return reclaimed; +} diff --git a/packages/gittensory-miner/lib/portfolio-queue-manager.d.ts b/packages/gittensory-miner/lib/portfolio-queue-manager.d.ts index 1f578a5d27..49ce435943 100644 --- a/packages/gittensory-miner/lib/portfolio-queue-manager.d.ts +++ b/packages/gittensory-miner/lib/portfolio-queue-manager.d.ts @@ -32,6 +32,7 @@ export type PortfolioQueueManager = { listQueue(repoFullName?: string | null): QueueEntry[]; markDone(repoFullName: string, identifier: string): QueueEntry | null; markFailed(repoFullName: string, identifier: string): QueueEntry | null; + reclaimStuckItems(maxLeaseMs?: number): QueueEntry[]; claimNextBatch(): QueueEntry[]; close(): void; }; @@ -40,6 +41,7 @@ export type InitPortfolioQueueManagerOptions = { caps?: Partial; store?: PortfolioQueueStore; dbPath?: string; + staleLeaseMs?: number; }; export function initPortfolioQueueManager(options?: InitPortfolioQueueManagerOptions): PortfolioQueueManager; diff --git a/packages/gittensory-miner/lib/portfolio-queue-manager.js b/packages/gittensory-miner/lib/portfolio-queue-manager.js index ded8372027..06cd2fdec9 100644 --- a/packages/gittensory-miner/lib/portfolio-queue-manager.js +++ b/packages/gittensory-miner/lib/portfolio-queue-manager.js @@ -4,6 +4,7 @@ // single-row dequeue. Caps are plain constructor arguments — not wired to .gittensory-miner.yml here. import { nextEligibleItems } from "@jsonbored/gittensory-engine"; import { initPortfolioQueueStore } from "./portfolio-queue.js"; +import { DEFAULT_MAX_LEASE_MS, sweepStuckItems } from "./portfolio-queue-expiry.js"; const ITEM_ID_SEPARATOR = "::"; @@ -74,6 +75,9 @@ export function selectEligibleBatch(entries, caps) { export function initPortfolioQueueManager(options = {}) { const caps = normalizePortfolioCaps(options.caps ?? { globalWipCap: 1, perRepoWipCap: 1 }); const store = options.store ?? initPortfolioQueueStore(options.dbPath); + // A lease older than this means the process that claimed the item almost certainly died; the item is swept back + // to 'queued' so it no longer occupies WIP capacity forever (#4827). + const staleLeaseMs = Number.isFinite(options.staleLeaseMs) ? options.staleLeaseMs : DEFAULT_MAX_LEASE_MS; return { caps, @@ -91,7 +95,14 @@ export function initPortfolioQueueManager(options = {}) { markFailed(repoFullName, identifier) { return store.markFailed(repoFullName, identifier); }, + /** Sweep leases orphaned by a crashed/killed process back to 'queued', returning the reclaimed items (#4827). */ + reclaimStuckItems(maxLeaseMs = staleLeaseMs) { + return sweepStuckItems(store, Date.now(), maxLeaseMs); + }, claimNextBatch() { + // Reclaim orphaned leases first, so an item stranded 'in_progress' by a dead process becomes eligible again + // instead of permanently consuming a WIP slot and starving the queue. + sweepStuckItems(store, Date.now(), staleLeaseMs); return store.batchClaim((entries) => selectEligibleBatch(entries, caps)); }, close() { diff --git a/packages/gittensory-miner/lib/portfolio-queue.d.ts b/packages/gittensory-miner/lib/portfolio-queue.d.ts index bc8e30f3ed..54d6a827ba 100644 --- a/packages/gittensory-miner/lib/portfolio-queue.d.ts +++ b/packages/gittensory-miner/lib/portfolio-queue.d.ts @@ -14,13 +14,23 @@ export type EnqueueItem = { priority?: number | null; }; +/** Lease-annotated view of an in-flight row: when it was claimed, for the expiry sweep (#4827). */ +export type QueueLeaseEntry = { + repoFullName: string; + identifier: string; + status: QueueStatus; + leasedAt: string | null; +}; + export type PortfolioQueueStore = { dbPath: string; enqueue(item: EnqueueItem): QueueEntry; dequeueNext(): QueueEntry | null; listQueue(repoFullName?: string | null): QueueEntry[]; + listInProgress(): QueueLeaseEntry[]; markDone(repoFullName: string, identifier: string): QueueEntry | null; markFailed(repoFullName: string, identifier: string): QueueEntry | null; + reclaimStuckItem(repoFullName: string, identifier: string): QueueEntry | null; batchClaim( selectFn: (entries: QueueEntry[]) => Array<{ repoFullName: string; identifier: string }>, ): QueueEntry[]; diff --git a/packages/gittensory-miner/lib/portfolio-queue.js b/packages/gittensory-miner/lib/portfolio-queue.js index 8d64e962ea..923e9a3382 100644 --- a/packages/gittensory-miner/lib/portfolio-queue.js +++ b/packages/gittensory-miner/lib/portfolio-queue.js @@ -53,6 +53,17 @@ function rowToEntry(row) { }; } +/** Lease-annotated projection of an in-flight row (adds `leasedAt`), consumed by the expiry sweep. Kept separate + * from `rowToEntry` so the base entry shape every existing caller relies on is unchanged. */ +function rowToLeaseEntry(row) { + return { + repoFullName: row.repo_full_name, + identifier: row.identifier, + status: row.status, + leasedAt: row.leased_at ?? null, + }; +} + /** * Opens the local portfolio/queue store, creating the table on first use. Rows are ordered highest-priority-first * with an insertion-order tie-break: `priority DESC, enqueued_at ASC, rowid ASC` — the implicit `rowid` guarantees @@ -69,9 +80,20 @@ export function initPortfolioQueueStore(dbPath = resolvePortfolioQueueDbPath()) priority REAL NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT 'queued' CHECK (status IN ('queued', 'in_progress', 'done')), enqueued_at TEXT NOT NULL, + leased_at TEXT, PRIMARY KEY (repo_full_name, identifier) ) `); + // `leased_at` records when an item was flipped to 'in_progress', so a crashed/killed process's stuck lease can be + // swept back to 'queued' by age (see portfolio-queue-expiry.js) instead of stranding the item forever — the same + // recovery the claim-ledger and worktree-allocator stores already provide for their own tables (#4827). Additive + // migration for stores created before this column: CREATE TABLE IF NOT EXISTS never adds a column to a pre-existing + // table, so add it idempotently. + const hasLeasedAtColumn = db + .prepare("PRAGMA table_info(miner_portfolio_queue)") + .all() + .some((column) => column.name === "leased_at"); + if (!hasLeasedAtColumn) db.exec("ALTER TABLE miner_portfolio_queue ADD COLUMN leased_at TEXT"); // `rowid` is a stable, unique key assigned once at first insert (re-enqueue updates in place, never re-inserts), // so it is a deterministic total-order tie-break: two items sharing a priority AND an `enqueued_at` timestamp @@ -95,18 +117,20 @@ export function initPortfolioQueueStore(dbPath = resolvePortfolioQueueDbPath()) // Claim the highest-priority queued item ATOMICALLY: one UPDATE selects the ordered top row in a subquery and // flips it to 'in_progress', RETURNING it — so two processes sharing the file can't both claim the same row (a // separate SELECT-then-UPDATE would race). + // Claiming stamps `leased_at` with the caller-supplied claim time; leaving 'in_progress' (done/failed/reclaim) + // clears it back to NULL so only genuinely in-flight rows carry a lease. const dequeueStatement = db.prepare(` - UPDATE miner_portfolio_queue SET status = 'in_progress' + UPDATE miner_portfolio_queue SET status = 'in_progress', leased_at = ? WHERE rowid = ( SELECT rowid FROM miner_portfolio_queue WHERE status = 'queued' ${ORDER} LIMIT 1 ) RETURNING * `); const markDoneStatement = db.prepare( - "UPDATE miner_portfolio_queue SET status = 'done' WHERE repo_full_name = ? AND identifier = ? AND status <> 'done'", + "UPDATE miner_portfolio_queue SET status = 'done', leased_at = NULL WHERE repo_full_name = ? AND identifier = ? AND status <> 'done'", ); const markFailedStatement = db.prepare(` - UPDATE miner_portfolio_queue SET status = 'queued' + UPDATE miner_portfolio_queue SET status = 'queued', leased_at = NULL WHERE repo_full_name = ? AND identifier = ? AND status = 'in_progress' RETURNING * `); @@ -117,8 +141,16 @@ export function initPortfolioQueueStore(dbPath = resolvePortfolioQueueDbPath()) const listActiveStatement = db.prepare( `SELECT * FROM miner_portfolio_queue WHERE status IN ('queued', 'in_progress') ${ORDER}`, ); + const listInProgressStatement = db.prepare( + `SELECT * FROM miner_portfolio_queue WHERE status = 'in_progress' ${ORDER}`, + ); + const reclaimStatement = db.prepare(` + UPDATE miner_portfolio_queue SET status = 'queued', leased_at = NULL + WHERE repo_full_name = ? AND identifier = ? AND status = 'in_progress' + RETURNING * + `); const claimTargetStatement = db.prepare(` - UPDATE miner_portfolio_queue SET status = 'in_progress' + UPDATE miner_portfolio_queue SET status = 'in_progress', leased_at = ? WHERE repo_full_name = ? AND identifier = ? AND status = 'queued' RETURNING * `); @@ -134,7 +166,20 @@ export function initPortfolioQueueStore(dbPath = resolvePortfolioQueueDbPath()) return rowToEntry(getStatement.get(repoFullName, identifier)); }, dequeueNext() { - const row = dequeueStatement.get(); + const row = dequeueStatement.get(new Date().toISOString()); + return row ? rowToEntry(row) : null; + }, + /** In-flight ('in_progress') rows with their `leasedAt` claim time, for the expiry sweep (#4827). */ + listInProgress() { + return listInProgressStatement.all().map(rowToLeaseEntry); + }, + /** Reclaim a single stuck in-flight item back to 'queued' (clearing its lease), returning it — or null if it is + * no longer 'in_progress' (already finished/reclaimed by another sweep). The sweep target of #4827. */ + reclaimStuckItem(repoFullName, identifier) { + const row = reclaimStatement.get( + normalizeRepoFullName(repoFullName), + normalizeIdentifier(identifier), + ); return row ? rowToEntry(row) : null; }, listQueue(repoFullName) { @@ -169,11 +214,12 @@ export function initPortfolioQueueStore(dbPath = resolvePortfolioQueueDbPath()) const entries = listActiveStatement.all().map(rowToEntry); const targets = selectFn(entries); if (!Array.isArray(targets)) throw new Error("invalid_batch_claim_selection"); + const leasedAt = new Date().toISOString(); const claimed = []; for (const target of targets) { const repoFullName = normalizeRepoFullName(target?.repoFullName); const identifier = normalizeIdentifier(target?.identifier); - const row = claimTargetStatement.get(repoFullName, identifier); + const row = claimTargetStatement.get(leasedAt, repoFullName, identifier); if (row) claimed.push(rowToEntry(row)); } db.exec("COMMIT"); diff --git a/packages/gittensory-miner/package.json b/packages/gittensory-miner/package.json index d0943e54d9..9f2879722f 100644 --- a/packages/gittensory-miner/package.json +++ b/packages/gittensory-miner/package.json @@ -32,7 +32,7 @@ "expected-engine.version" ], "scripts": { - "build": "node --check bin/gittensory-miner.js && node --check lib/attempt-cli.js && node --check lib/attempt-log.js && node --check lib/attempt-runner.js && node --check lib/calibration-types.js && node --check lib/calibration.js && node --check lib/ci-poller.js && node --check lib/claim-adjudication.js && node --check lib/claim-ledger-cli.js && node --check lib/claim-ledger-expiry.js && node --check lib/claim-ledger.js && node --check lib/cli.js && node --check lib/coding-agent-construction.js && node --check lib/coding-agent-house-rules.js && node --check lib/deny-check.js && node --check lib/deny-hook-synthesis.js && node --check lib/deny-hooks.js && node --check lib/discover-cli.js && node --check lib/event-ledger-cli.js && node --check lib/event-ledger.js && node --check lib/execute-local-write.js && node --check lib/feasibility-cli.js && node --check lib/gate-verdict-poller.js && node --check lib/governor-action-mode.js && node --check lib/governor-chokepoint-persisted.js && node --check lib/governor-chokepoint.js && node --check lib/governor-kill-switch.js && node --check lib/governor-ledger-cli.js && node --check lib/governor-ledger.js && node --check lib/governor-open-pr.js && node --check lib/governor-run-halt.js && node --check lib/governor-state.js && node --check lib/governor-write-rate-limit.js && node --check lib/harness-submission-trigger.js && node --check lib/laptop-init.js && node --check lib/live-issue-snapshot.js && node --check lib/local-store.js && node --check lib/loop-closure.js && node --check lib/loop-reentry.js && node --check lib/manage-poll.js && node --check lib/manage-status.js && node --check lib/opportunity-fanout.js && node --check lib/opportunity-ranker.js && node --check lib/orb-export.js && node --check lib/plan-store-cli.js && node --check lib/plan-store.js && node --check lib/portfolio-dashboard.js && node --check lib/portfolio-discovery.js && node --check lib/portfolio-queue-cli.js && node --check lib/portfolio-queue-manager.js && node --check lib/portfolio-queue.js && node --check lib/pr-outcome.js && node --check lib/prediction-ledger.js && node --check lib/pretooluse-hook.js && node --check lib/rejection-state-machine.js && node --check lib/rejection-templates.js && node --check lib/replay-objective-anchor.js && node --check lib/replay-snapshot.js && node --check lib/replay-task-generation.js && node --check lib/run-state-cli.js && node --check lib/run-state.js && node --check lib/slop-assessment.js && node --check lib/status.js && node --check lib/submission-freshness-check.js && node --check lib/update-check.js && node --check lib/version.js && node --check lib/worktree-allocator.js" + "build": "node --check bin/gittensory-miner.js && node --check lib/attempt-cli.js && node --check lib/attempt-log.js && node --check lib/attempt-runner.js && node --check lib/calibration-types.js && node --check lib/calibration.js && node --check lib/ci-poller.js && node --check lib/claim-adjudication.js && node --check lib/claim-ledger-cli.js && node --check lib/claim-ledger-expiry.js && node --check lib/claim-ledger.js && node --check lib/cli.js && node --check lib/coding-agent-construction.js && node --check lib/coding-agent-house-rules.js && node --check lib/deny-check.js && node --check lib/deny-hook-synthesis.js && node --check lib/deny-hooks.js && node --check lib/discover-cli.js && node --check lib/event-ledger-cli.js && node --check lib/event-ledger.js && node --check lib/execute-local-write.js && node --check lib/feasibility-cli.js && node --check lib/gate-verdict-poller.js && node --check lib/governor-action-mode.js && node --check lib/governor-chokepoint-persisted.js && node --check lib/governor-chokepoint.js && node --check lib/governor-kill-switch.js && node --check lib/governor-ledger-cli.js && node --check lib/governor-ledger.js && node --check lib/governor-open-pr.js && node --check lib/governor-run-halt.js && node --check lib/governor-state.js && node --check lib/governor-write-rate-limit.js && node --check lib/harness-submission-trigger.js && node --check lib/laptop-init.js && node --check lib/live-issue-snapshot.js && node --check lib/local-store.js && node --check lib/loop-closure.js && node --check lib/loop-reentry.js && node --check lib/manage-poll.js && node --check lib/manage-status.js && node --check lib/opportunity-fanout.js && node --check lib/opportunity-ranker.js && node --check lib/orb-export.js && node --check lib/plan-store-cli.js && node --check lib/plan-store.js && node --check lib/portfolio-dashboard.js && node --check lib/portfolio-discovery.js && node --check lib/portfolio-queue-cli.js && node --check lib/portfolio-queue-manager.js && node --check lib/portfolio-queue.js && node --check lib/portfolio-queue-expiry.js && node --check lib/pr-outcome.js && node --check lib/prediction-ledger.js && node --check lib/pretooluse-hook.js && node --check lib/rejection-state-machine.js && node --check lib/rejection-templates.js && node --check lib/replay-objective-anchor.js && node --check lib/replay-snapshot.js && node --check lib/replay-task-generation.js && node --check lib/run-state-cli.js && node --check lib/run-state.js && node --check lib/slop-assessment.js && node --check lib/status.js && node --check lib/submission-freshness-check.js && node --check lib/update-check.js && node --check lib/version.js && node --check lib/worktree-allocator.js" }, "dependencies": { "@jsonbored/gittensory-engine": "*" diff --git a/test/unit/miner-portfolio-queue-expiry.test.ts b/test/unit/miner-portfolio-queue-expiry.test.ts new file mode 100644 index 0000000000..3172d79ed9 --- /dev/null +++ b/test/unit/miner-portfolio-queue-expiry.test.ts @@ -0,0 +1,190 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { initPortfolioQueueStore } from "../../packages/gittensory-miner/lib/portfolio-queue.js"; +import type { QueueLeaseEntry } from "../../packages/gittensory-miner/lib/portfolio-queue.js"; +import { + DEFAULT_MAX_LEASE_MS, + findStuckItems, + sweepStuckItems, +} from "../../packages/gittensory-miner/lib/portfolio-queue-expiry.js"; +import { initPortfolioQueueManager } from "../../packages/gittensory-miner/lib/portfolio-queue-manager.js"; + +const roots: string[] = []; +const stores: Array<{ close(): void }> = []; + +function tempStore() { + const root = mkdtempSync(join(tmpdir(), "gittensory-miner-portfolio-expiry-")); + roots.push(root); + const store = initPortfolioQueueStore(join(root, "portfolio-queue.sqlite3")); + stores.push(store); + return store; +} + +afterEach(() => { + for (const store of stores.splice(0)) store.close(); + vi.useRealTimers(); + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); +}); + +const leaseItem = (overrides: Partial = {}): QueueLeaseEntry => ({ + repoFullName: "o/a", + identifier: "x", + status: "in_progress", + leasedAt: "2026-01-01T00:00:00.000Z", + ...overrides, +}); + +describe("portfolio-queue lease bookkeeping (#4827)", () => { + it("stamps leased_at when an item is claimed and exposes it via listInProgress", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-12T10:00:00.000Z")); + const store = tempStore(); + store.enqueue({ repoFullName: "o/a", identifier: "x" }); + expect(store.listInProgress()).toEqual([]); // queued item carries no lease + + const claimed = store.dequeueNext(); + expect(claimed).toMatchObject({ identifier: "x", status: "in_progress" }); + expect(store.listInProgress()).toEqual([ + { repoFullName: "o/a", identifier: "x", status: "in_progress", leasedAt: "2026-07-12T10:00:00.000Z" }, + ]); + }); + + it("clears the lease when an item leaves in_progress (done, failed, reclaimed)", () => { + const store = tempStore(); + for (const id of ["done", "failed", "reclaimed"]) { + store.enqueue({ repoFullName: "o/a", identifier: id }); + } + // Claim all three, then release each a different way. + store.dequeueNext(); + store.dequeueNext(); + store.dequeueNext(); + expect(store.listInProgress()).toHaveLength(3); + + store.markDone("o/a", "done"); + store.markFailed("o/a", "failed"); + const reclaimed = store.reclaimStuckItem("o/a", "reclaimed"); + + expect(reclaimed).toMatchObject({ identifier: "reclaimed", status: "queued" }); + expect(store.listInProgress()).toEqual([]); // every lease cleared + }); + + it("reclaimStuckItem is a no-op (null) for an item that is not in_progress", () => { + const store = tempStore(); + store.enqueue({ repoFullName: "o/a", identifier: "x" }); // still 'queued' + expect(store.reclaimStuckItem("o/a", "x")).toBeNull(); + expect(store.reclaimStuckItem("o/a", "missing")).toBeNull(); + }); + + it("batchClaim stamps a lease on every claimed row", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-12T11:30:00.000Z")); + const store = tempStore(); + store.enqueue({ repoFullName: "o/a", identifier: "1" }); + store.enqueue({ repoFullName: "o/b", identifier: "2" }); + store.batchClaim((entries) => entries.map((e) => ({ repoFullName: e.repoFullName, identifier: e.identifier }))); + expect(store.listInProgress().map((e) => e.leasedAt)).toEqual([ + "2026-07-12T11:30:00.000Z", + "2026-07-12T11:30:00.000Z", + ]); + }); +}); + +describe("findStuckItems (#4827)", () => { + const now = Date.parse("2026-07-12T12:00:00.000Z"); + const max = 30 * 60 * 1000; + + it("returns an in-flight item whose lease age strictly exceeds the bound", () => { + const stuck = leaseItem({ leasedAt: new Date(now - max - 1).toISOString() }); + expect(findStuckItems([stuck], now, max)).toEqual([stuck]); + }); + + it("treats an item exactly at the bound as still within the window", () => { + const atBound = leaseItem({ leasedAt: new Date(now - max).toISOString() }); + expect(findStuckItems([atBound], now, max)).toEqual([]); + }); + + it("ignores fresh, non-in_progress, and unparseable-lease items", () => { + const fresh = leaseItem({ identifier: "fresh", leasedAt: new Date(now - 1).toISOString() }); + const queued = leaseItem({ identifier: "queued", status: "queued", leasedAt: new Date(now - max - 5).toISOString() }); + const noLease = leaseItem({ identifier: "nolease", leasedAt: null }); + const bogus = leaseItem({ identifier: "bogus", leasedAt: "not-a-date" }); + expect(findStuckItems([fresh, queued, noLease, bogus], now, max)).toEqual([]); + }); + + it("validates its arguments", () => { + expect(() => findStuckItems([], Number.NaN, max)).toThrow("invalid_now_ms"); + expect(() => findStuckItems([], -1, max)).toThrow("invalid_now_ms"); + expect(() => findStuckItems([], now, Number.NaN)).toThrow("invalid_max_lease_ms"); + expect(() => findStuckItems([], now, -1)).toThrow("invalid_max_lease_ms"); + expect(() => findStuckItems("nope" as unknown as [], now, max)).toThrow("invalid_items"); + }); +}); + +describe("sweepStuckItems (#4827)", () => { + it("reclaims only the stuck in-flight items back to queued against a real store", () => { + vi.useFakeTimers(); + const store = tempStore(); + + // Claim `old` long ago, `recent` just now. + vi.setSystemTime(new Date("2026-07-12T09:00:00.000Z")); + store.enqueue({ repoFullName: "o/a", identifier: "old" }); + store.dequeueNext(); + vi.setSystemTime(new Date("2026-07-12T09:59:59.000Z")); + store.enqueue({ repoFullName: "o/a", identifier: "recent" }); + store.dequeueNext(); + + const nowMs = Date.parse("2026-07-12T10:00:00.000Z"); + const reclaimed = sweepStuckItems(store, nowMs, 30 * 60 * 1000); + + expect(reclaimed.map((e) => e.identifier)).toEqual(["old"]); + expect(store.listInProgress().map((e) => e.identifier)).toEqual(["recent"]); + // The reclaimed item is back in the queue for another attempt. + expect(store.listQueue("o/a").find((e) => e.identifier === "old")?.status).toBe("queued"); + }); + + it("defaults the bound to DEFAULT_MAX_LEASE_MS and reclaims nothing when all leases are fresh", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-12T10:00:00.000Z")); + const store = tempStore(); + store.enqueue({ repoFullName: "o/a", identifier: "fresh" }); + store.dequeueNext(); + const nowMs = Date.parse("2026-07-12T10:00:01.000Z"); // 1s old, well under the default + expect(sweepStuckItems(store, nowMs)).toEqual([]); + expect(DEFAULT_MAX_LEASE_MS).toBe(30 * 60 * 1000); + }); +}); + +describe("PortfolioQueueManager stuck-lease reclaim wiring (#4827)", () => { + it("claimNextBatch sweeps an orphaned lease back to queued before selecting, so it is claimable again", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-12T09:00:00.000Z")); + const store = tempStore(); + const manager = initPortfolioQueueManager({ store, caps: { globalWipCap: 1, perRepoWipCap: 1 } }); + manager.enqueue({ repoFullName: "acme/alpha", identifier: "work" }); + + expect(manager.claimNextBatch().map((e) => e.identifier)).toEqual(["work"]); + // The owning process "dies": the item stays in_progress and keeps the only WIP slot, so nothing else claims. + expect(manager.claimNextBatch()).toEqual([]); + + vi.setSystemTime(new Date("2026-07-12T10:00:00.000Z")); // +1h, past the 30m default lease bound + // The next claim sweeps the orphaned lease back to queued, then re-claims the now-eligible item. + expect(manager.claimNextBatch().map((e) => e.identifier)).toEqual(["work"]); + }); + + it("reclaimStuckItems() returns the swept items and leaves fresh leases alone", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-12T09:00:00.000Z")); + const store = tempStore(); + const manager = initPortfolioQueueManager({ store, caps: { globalWipCap: 5, perRepoWipCap: 5 } }); + manager.enqueue({ repoFullName: "acme/alpha", identifier: "work" }); + manager.claimNextBatch(); + + expect(manager.reclaimStuckItems()).toEqual([]); // fresh lease → nothing stuck + vi.setSystemTime(new Date("2026-07-12T10:00:00.000Z")); + const reclaimed = manager.reclaimStuckItems(); + expect(reclaimed.map((e) => e.identifier)).toEqual(["work"]); + expect(reclaimed[0]?.status).toBe("queued"); + }); +});