mason: fix channel2 lease and band reset

Author: ualtinok

docs/AUDIT-KNOWN-ISSUES.md

@@ -581,12 +581,22 @@ with the default, so they read `''` (verified `IS NULL → 0`), and the
 trigger/delivery CAS matches.
 
 A wedged `'claimed'` lease (crash mid-delivery) is still healed by
-`healWedgedChannel2Claims` (`storage-db.ts`), but the heal is now TTL-scoped via
-`channel2_nudge_claimed_at`: fresh claims are left alone so a sibling process
-boot cannot steal a live in-flight delivery; stale/legacy claims rewind to
-`'pending'`. After a send succeeds, confirm failures are not reverted to
-`'pending'` because the synthetic user message may already exist. Accepted as
-correct.
+`healWedgedChannel2Claims` (`storage-db.ts`), and `openDatabase()` now reruns that
+TTL-scoped heal on cached-handle reuse too so long-lived processes eventually
+unwind stale claims without a restart. The lease uses
+`channel2_nudge_claimed_at` as its liveness boundary: fresh claims are left alone
+so a sibling process boot cannot steal a live in-flight delivery; stale/legacy
+claims rewind to `'pending'`. If a send fails and the `claimed→pending` restore is
+locked, the row stays `claimed` with its timestamp intact and later TTL-heals back
+to `'pending'`.
+
+One rare duplicate window remains accepted by design: if a process is suspended or
+otherwise hangs for longer than the TTL after sending but before confirming, a
+sibling can heal that stale claim, redeliver the same reminder, and mark
+`'delivered'` first. The original sender now preserves an already-`'delivered'`
+row and logs that stolen-lease path distinctly for diagnosis, but it cannot
+unsend its already-queued reminder. The cost is one duplicate synthetic message,
+not extra cap consumption.
 
 ### A37. `NORMAL_HYSTERESIS_TOKENS` (256) eligible-head snap is deliberate (boundary-straddle wobble accepted)
 

packages/pi-plugin/src/context-handler.test.ts

@@ -11,6 +11,8 @@ import {
 	addNote,
 	appendNoteNudgeAnchor,
 	getHistorianFailureState,
+	getLastNudgeLevel,
+	getLastNudgeUndropped,
 	getNoteNudgeAnchors,
 	getOrCreateSessionMeta,
 	getPendingOps,
@@ -19,6 +21,8 @@ import {
 	incrementHistorianFailure,
 	insertTag,
 	queuePendingOp,
+	setLastNudgeLevel,
+	setLastNudgeUndropped,
 	setPendingPiCompactionMarkerState,
 	updateSessionMeta,
 } from "@magic-context/core/features/magic-context/storage";
@@ -164,6 +168,35 @@ describe("registerPiContextHandler", () => {
 		}
 	});
 
+	it("resets the persisted Channel 1 band when baseline refresh sees a smaller tail", async () => {
+		const db = createTestDb();
+		try {
+			const sessionId = "ses-pi-band-reset";
+			setLastNudgeUndropped(db, sessionId, 80_000);
+			setLastNudgeLevel(db, sessionId, "urgent");
+
+			const fake = createFakePi();
+			registerPiContextHandler(fake.pi as never, {
+				db,
+				ctxReduceEnabled: true,
+			});
+			const handler = fake.handlers.get("context") as (
+				event: { messages: never[] },
+				ctx: never,
+			) => Promise<{ messages: never[] } | undefined>;
+
+			await handler(
+				{ messages: [userMessage("hello", 1)] as never[] },
+				fakeContext(sessionId) as never,
+			);
+
+			expect(getLastNudgeUndropped(db, sessionId)).toBe(0);
+			expect(getLastNudgeLevel(db, sessionId)).toBe("");
+		} finally {
+			closeQuietly(db);
+		}
+	});
+
 	it("clears stale compartmentInProgress on first context pass after restart", async () => {
 		const db = createTestDb();
 		try {
@@ -958,8 +991,21 @@ describe("registerPiContextHandler", () => {
 						piInputs.commitClusterTrigger,
 					);
 
+					const stripCreatedAtDeep = (value: unknown): unknown => {
+						if (Array.isArray(value)) {
+							return value.map(stripCreatedAtDeep);
+						}
+						if (!value || typeof value !== "object") return value;
+						const entries = Object.entries(value as Record<string, unknown>)
+							.filter(([key]) => key !== "createdAt")
+							.map(([key, inner]) => [key, stripCreatedAtDeep(inner)]);
+						return Object.fromEntries(entries);
+					};
+
 					expect(piInputs.triggerBudget).toBe(triggerBudget);
-					expect(piDecision).toEqual(opencodeDecision);
+					expect(stripCreatedAtDeep(piDecision)).toEqual(
+						stripCreatedAtDeep(opencodeDecision),
+					);
 					expect(piDecision).toMatchObject({
 						shouldFire: true,
 						reason: "projected_headroom",

packages/pi-plugin/src/context-handler.ts

@@ -80,6 +80,7 @@ import {
 	peekDeferredExecutePending,
 	pruneAutoSearchHintDecisions,
 	pruneNoteNudgeAnchors,
+	resetLastNudgeCycleIfTailShrank,
 	setDeferredExecutePendingIfAbsent,
 } from "@magic-context/core/features/magic-context/storage-meta-persisted";
 import {
@@ -2220,6 +2221,14 @@ export function registerPiContextHandler(
 						0,
 						executeThresholdTokensPi - usageInputTokens + liveTailTokens,
 					);
+					// Same rationale as OpenCode: a historian publish, emergency drop,
+					// or pending-op replay can shrink the tail without a ctx_reduce
+					// tool call, so a regrowth must not inherit a stale persisted band.
+					resetLastNudgeCycleIfTailShrank(
+						options.db,
+						sessionId,
+						tailToolTokens,
+					);
 					setPiChannel1Baseline(sessionId, {
 						tailToolTokens,
 						historyBudgetTokens: historyBudgetTokens ?? 0,

packages/pi-plugin/src/ctx-reduce-nudge-pi.test.ts

@@ -1,10 +1,12 @@
-import { describe, expect, it } from "bun:test";
+import { afterEach, describe, expect, it, mock, spyOn } from "bun:test";
 import { readFileSync } from "node:fs";
 import { join } from "node:path";
 import {
+	getChannel2NudgeClaimedAt,
 	getChannel2NudgeState,
 	setChannel2NudgeState,
 } from "@magic-context/core/features/magic-context/storage";
+import * as loggerModule from "@magic-context/core/shared/logger";
 import {
 	clearPiChannel1State,
 	computeTailTokenEstimatePi,
@@ -19,6 +21,10 @@ function toolResultMsg(text: string) {
 	return { role: "toolResult", content: [{ type: "text", text }] };
 }
 
+afterEach(() => {
+	mock.restore();
+});
+
 describe("computeTailToolTokensPi", () => {
 	it("sums non-dropped toolResult text, excludes sentinels", () => {
 		const big = "x".repeat(40_000); // ~10k tokens
@@ -262,6 +268,90 @@ describe("maybeDeliverChannel2Pi", () => {
 		expect(getChannel2NudgeState(db, SESSION)).toBe("pending");
 	});
 
+	it("returns false and leaves the claim healable when claimed→pending CAS throws", async () => {
+		const db = createTestDb();
+		const session = "ses-ch2-pi-revert-throw";
+		setChannel2NudgeState(db, session, "pending");
+		armStrongBaseline(session);
+
+		const originalPrepare = db.prepare.bind(db);
+		(db as unknown as { prepare: typeof db.prepare }).prepare = (
+			sql: string,
+		) => {
+			const statement = originalPrepare(sql);
+			if (
+				sql ===
+				"UPDATE session_meta SET channel2_nudge_state = ?, channel2_nudge_claimed_at = ? WHERE session_id = ? AND channel2_nudge_state = ?"
+			) {
+				return {
+					...statement,
+					run: (...args: unknown[]) => {
+						if (
+							args[0] === "pending" &&
+							args[1] === 0 &&
+							args[2] === session &&
+							args[3] === "claimed"
+						) {
+							throw new Error("SQLITE_BUSY: database is locked");
+						}
+						return statement.run(
+							...(args as [unknown, unknown, unknown, unknown]),
+						);
+					},
+				} as typeof statement;
+			}
+			return statement;
+		};
+
+		const delivered = maybeDeliverChannel2Pi(
+			{
+				sendUserMessage: () => {
+					throw new Error("transient");
+				},
+			},
+			db,
+			session,
+		);
+
+		expect(delivered).toBe(false);
+		expect(getChannel2NudgeState(db, session)).toBe("claimed");
+		expect(getChannel2NudgeClaimedAt(db, session)).toBeGreaterThan(0);
+	});
+
+	it("preserves a sibling's delivered claim and logs the duplicate window distinctly", async () => {
+		const db = createTestDb();
+		const session = "ses-ch2-pi-duplicate";
+		setChannel2NudgeState(db, session, "pending");
+		armStrongBaseline(session);
+
+		const sessionLog = spyOn(loggerModule, "sessionLog").mockImplementation(
+			() => {},
+		);
+
+		const delivered = maybeDeliverChannel2Pi(
+			{
+				sendUserMessage: () => {
+					db.prepare(
+						"UPDATE session_meta SET channel2_nudge_state = 'delivered', channel2_nudge_claimed_at = 0 WHERE session_id = ?",
+					).run(session);
+				},
+			},
+			db,
+			session,
+		);
+
+		expect(delivered).toBe(false);
+		expect(getChannel2NudgeState(db, session)).toBe("delivered");
+		expect(
+			sessionLog.mock.calls.some(
+				(call) =>
+					call[0] === session &&
+					typeof call[1] === "string" &&
+					call[1].includes("duplicate window"),
+			),
+		).toBe(true);
+	});
+
 	it("does not re-deliver after success (one nudge per lifetime)", () => {
 		const db = createTestDb();
 		setChannel2NudgeState(db, SESSION, "delivered");

packages/pi-plugin/src/ctx-reduce-nudge-pi.ts

@@ -40,10 +40,32 @@ import {
 	tailToolTokensFromStrings,
 	toolOutputTokens,
 } from "@magic-context/core/hooks/magic-context/ctx-reduce-nudge";
+import { sessionLog } from "@magic-context/core/shared/logger";
 import type { Database } from "@magic-context/core/shared/sqlite";
 
 export type { Channel1State };
 
+function sealDeliveredAfterUnconfirmedSend(
+	db: Database,
+	sessionId: string,
+): "already-delivered" | "sealed" | "stuck-claimed" {
+	try {
+		if (getChannel2NudgeState(db, sessionId) === "delivered") {
+			return "already-delivered";
+		}
+	} catch {
+		// Best-effort probe only — if the read fails, still try to seal the cap.
+	}
+
+	try {
+		setChannel2NudgeState(db, sessionId, "delivered");
+		return "sealed";
+	} catch {
+		// Preserve the stale claim so the shared TTL heal can recover it later.
+		return "stuck-claimed";
+	}
+}
+
 // Per-session Channel 1 metric baseline. Written at the end of each pipeline
 // pass (post-drop), read in the `tool_result` handler. Primary-only: subagents
 // never get a baseline, which is how Channel 1 stays off for them (matches
@@ -322,8 +344,22 @@ export function maybeDeliverChannel2Pi(
 		pi.sendUserMessage(buildChannel2Reminder(undropped), {
 			deliverAs,
 		});
-	} catch {
-		casChannel2NudgeState(db, sessionId, "claimed", "pending");
+	} catch (error) {
+		try {
+			casChannel2NudgeState(db, sessionId, "claimed", "pending");
+		} catch (revertError) {
+			sessionLog(
+				sessionId,
+				"channel2 ceiling nudge delivery failed; pending restore was busy so the stale claim will heal later:",
+				{ deliveryError: error, revertError },
+			);
+			return false;
+		}
+		sessionLog(
+			sessionId,
+			"channel2 ceiling nudge delivery failed (will retry):",
+			error,
+		);
 		return false;
 	}
 
@@ -334,22 +370,50 @@ export function maybeDeliverChannel2Pi(
 			"claimed",
 			"delivered",
 		);
-		if (confirmed) return true;
-		try {
-			// The nudge has already been handed to Pi; seal the cap if another
-			// process rewound the claim before our authoritative confirm.
-			setChannel2NudgeState(db, sessionId, "delivered");
-		} catch {
-			// Best-effort; never re-arm after send.
+		if (confirmed) {
+			sessionLog(sessionId, "channel2 ceiling nudge delivered");
+			return true;
+		}
+
+		const outcome = sealDeliveredAfterUnconfirmedSend(db, sessionId);
+		if (outcome === "already-delivered") {
+			sessionLog(
+				sessionId,
+				"channel2 ceiling nudge duplicate window: our send returned after a sibling reclaimed the stale lease and already delivered",
+			);
+		} else if (outcome === "sealed") {
+			sessionLog(
+				sessionId,
+				"channel2 ceiling nudge sent but claim confirmation was lost; sealed delivered without an authoritative confirm",
+			);
+		} else {
+			sessionLog(
+				sessionId,
+				"channel2 ceiling nudge sent but claim confirmation was lost; lease stayed claimed and will heal later",
+			);
 		}
 		return false;
-	} catch {
+	} catch (error) {
 		// The nudge has already been handed to Pi; never re-arm on a post-send
 		// confirm failure, or a transient DB error can duplicate the one-shot cap.
-		try {
-			setChannel2NudgeState(db, sessionId, "delivered");
-		} catch {
-			// Best-effort; never re-arm after send.
+		const outcome = sealDeliveredAfterUnconfirmedSend(db, sessionId);
+		if (outcome === "already-delivered") {
+			sessionLog(
+				sessionId,
+				"channel2 ceiling nudge duplicate window: our send returned after a sibling reclaimed the stale lease and already delivered",
+			);
+		} else if (outcome === "sealed") {
+			sessionLog(
+				sessionId,
+				"channel2 ceiling nudge sent but confirm failed:",
+				error,
+			);
+		} else {
+			sessionLog(
+				sessionId,
+				"channel2 ceiling nudge sent but confirm failed; lease stayed claimed and will heal later:",
+				error,
+			);
 		}
 		return false;
 	}