fix(pi): contain historian failures and busy-starved lease releases

packages/pi-plugin/src/commands/ctx-wrapup.test.ts

@@ -1,6 +1,8 @@
 /// <reference types="bun-types" />
 
 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 {

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.`;

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);
 }