From 710b343aff25c8311fc07e5e2a777cf87a4a4c65 Mon Sep 17 00:00:00 2001 From: Naadir Jeewa Date: Thu, 3 Sep 2026 01:25:09 +0100 Subject: [PATCH] fix(pi): contain historian failures and busy-starved lease releases The Pi historian run promise is parked in inFlightHistorian with no catch, so a SQLITE_BUSY from the finally-block lease release escaped as an unhandled rejection via processTicksAndRejections. OpenCode's startCompartmentAgent carries its own .catch; the Pi port dropped it. - spawnPiHistorianRun: add .catch (session log) and guard both lease releases as best-effort teardown; the row expires via its 5-min TTL. - runPiWrapup: catch historian chunk failures and surface them as a Partial result instead of letting them escape the command handler, and guard the per-chunk lease release the same way. - Add a regression test that holds BEGIN IMMEDIATE on a second connection (shared file, not :memory:) and asserts wrapup survives the busy-starved release. --- .../pi-plugin/src/commands/ctx-wrapup.test.ts | 58 +++++++++++++++++++ packages/pi-plugin/src/commands/ctx-wrapup.ts | 20 ++++++- packages/pi-plugin/src/context-handler.ts | 43 +++++++++++--- 3 files changed, 111 insertions(+), 10 deletions(-) diff --git a/packages/pi-plugin/src/commands/ctx-wrapup.test.ts b/packages/pi-plugin/src/commands/ctx-wrapup.test.ts index 8c90aabdf..e41015d88 100644 --- a/packages/pi-plugin/src/commands/ctx-wrapup.test.ts +++ b/packages/pi-plugin/src/commands/ctx-wrapup.test.ts @@ -1,6 +1,8 @@ /// import { describe, expect, it, mock } from "bun:test"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { acquireCompartmentLease, releaseCompartmentLease, @@ -39,6 +41,14 @@ function createDb(): Database { return db; } +function createFileDb(name: string): Database { + const path = join(tmpdir(), `${name}-${crypto.randomUUID()}.db`); + const db = new Database(path); + initializeDatabase(db); + runMigrations(db); + return db; +} + function branch(count: number) { return Array.from({ length: count }, (_, index) => ({ id: `m-${index + 1}`, @@ -276,6 +286,54 @@ describe("Pi /ctx-wrapup", () => { } }); + it("survives a busy-database lease release failure without rejecting", async () => { + // Cross-connection lock contention requires a shared file; two :memory: + // databases never contend. Mirrors the production shape: a second + // connection (omp session, dreamer, FTS indexer) holds BEGIN IMMEDIATE + // past busy_timeout while wrapup releases its compartment lease. + const db = createFileDb("pi-wrapup-busy-release"); + try { + const sessionId = "pi-wrapup-busy-release"; + const runPiHistorianForWrapup = mock(async (args) => { + const start = getLastCompartmentEndMessage(db, sessionId) + 1; + const end = Math.min( + args.boundarySnapshot.eligibleEndOrdinal - 1, + start + 2, + ); + appendRange(db, sessionId, start, end); + args.onPublished?.(); + // Simulate the production contention window: another process holds + // the writer lock past busy_timeout while wrapup releases its lease. + const blocker = new Database(db.filename); + blocker.exec("PRAGMA busy_timeout=1; BEGIN IMMEDIATE;"); + try { + // Shrink the release connection's busy wait so the expected + // SQLITE_BUSY lands well inside bun's 5s default test timeout + // (initializeDatabase sets 5000ms; plain `bun test` in CI has + // no override — project constraint #5917). + db.exec("PRAGMA busy_timeout=250"); + releaseCompartmentLease(db, sessionId, "stale-holder"); + } finally { + blocker.exec("ROLLBACK"); + blocker.close(); + } + }); + + const result = await runPiWrapup( + pi().api, + deps(db, { runPiHistorianForWrapup }), + ctx(sessionId, 8), + sessionId, + 2, + ); + + expect(result).toContain("## Magic Wrapup"); + expect(getWrapupInProgressState(db, sessionId)).toBeNull(); + } finally { + closeQuietly(db); + } + }); + it("signals deferred history and materialization after a wrapup publish", async () => { const db = createDb(); try { diff --git a/packages/pi-plugin/src/commands/ctx-wrapup.ts b/packages/pi-plugin/src/commands/ctx-wrapup.ts index ce186c2ec..80b3e86c1 100644 --- a/packages/pi-plugin/src/commands/ctx-wrapup.ts +++ b/packages/pi-plugin/src/commands/ctx-wrapup.ts @@ -398,6 +398,7 @@ export async function runPiWrapup( ); } }, COMPARTMENT_LEASE_RENEWAL_MS); + let chunkFailure: string | null = null; try { const runHistorian = deps.runPiHistorianForWrapup ?? runPiHistorian; await runHistorian({ @@ -451,11 +452,28 @@ export async function runPiWrapup( signalPiDeferredMaterialization(sessionId); }, }); + } catch (err) { + // Parity with OpenCode: runCompartmentAgent carries its own .catch, + // so its wrapup finally only sees settled promises. Pi lacks that + // inner layer; record the failure instead of letting a busy-starved + // release (or any historian throw) escape the command handler. + chunkFailure = err instanceof Error ? err.message : String(err); } finally { clearInterval(leaseRenewal); - releaseCompartmentLease(deps.db, sessionId, leaseHolder); + try { + releaseCompartmentLease(deps.db, sessionId, leaseHolder); + } catch (err) { + // Best-effort teardown: the lease row expires via its own TTL. + console.warn( + `[magic-context][pi] /ctx-wrapup compartment lease release failed for ${sessionId} (expires via TTL): ${err instanceof Error ? err.message : String(err)}`, + ); + } } + if (chunkFailure) { + failure = `historian chunk failed (${chunkFailure}); wrapped up through message ${lastEnd}. Run /ctx-wrapup again to continue.`; + break; + } const afterEnd = getLastCompartmentEndMessage(deps.db, sessionId); if (afterEnd <= lastEnd) { failure = `No forward progress after chunk ${chunkIndex}; wrapped up through message ${lastEnd}. Run /ctx-wrapup again to continue.`; diff --git a/packages/pi-plugin/src/context-handler.ts b/packages/pi-plugin/src/context-handler.ts index c790049bf..f148c9fc0 100644 --- a/packages/pi-plugin/src/context-handler.ts +++ b/packages/pi-plugin/src/context-handler.ts @@ -3640,7 +3640,15 @@ function spawnPiHistorianRun(args: { // Close the cross-process check/lease race: /ctx-wrapup may have published // its marker after the first check but before this process won the lease. sessionLog(sessionId, "historian skipped: /ctx-wrapup became active"); - releaseCompartmentLease(db, sessionId, holderId); + try { + releaseCompartmentLease(db, sessionId, holderId); + } catch (err) { + // Same best-effort contract as the finally-block release below. + sessionLog( + sessionId, + `historian lease release failed (expires via TTL): ${err instanceof Error ? err.message : String(err)}`, + ); + } return; } const renewal = startPiCompartmentLeaseRenewal(db, sessionId, holderId); @@ -3726,15 +3734,32 @@ function spawnPiHistorianRun(args: { }); } finally { clearInterval(renewal); - releaseCompartmentLease(db, sessionId, holderId); - } - })().finally(() => { - inFlightHistorian.delete(sessionId); - unregister(); - if (isContextHandlerSessionActive(sessionId)) { - historian.onStatusChange?.(ctx, sessionId); + try { + releaseCompartmentLease(db, sessionId, holderId); + } catch (err) { + // Best-effort teardown: the lease row expires via its own TTL, so a + // busy_timeout-starved DELETE must not escape and reject this promise. + sessionLog( + sessionId, + `historian lease release failed (expires via TTL): ${err instanceof Error ? err.message : String(err)}`, + ); + } } - }); + })() + .catch((err) => { + // Parity with OpenCode's startCompartmentAgent .catch: the historian run + // promise is parked in inFlightHistorian for emergency/shutdown awaits, + // so an unhandled rejection here crashes the host via + // processTicksAndRejections instead of surfacing as a session log. + sessionLog(sessionId, "historian run failed:", err); + }) + .finally(() => { + inFlightHistorian.delete(sessionId); + unregister(); + if (isContextHandlerSessionActive(sessionId)) { + historian.onStatusChange?.(ctx, sessionId); + } + }); inFlightHistorian.set(sessionId, runPromise); historian.onStatusChange?.(ctx, sessionId); }