merge: Pi marker drain accepts m1-rendered coverage (liveness fix, panel-pinned)

Author: ualtinok

packages/e2e-tests/tests/pi-long-running-session.test.ts

@@ -364,7 +364,12 @@ describe("long-running Pi Magic Context session", () => {
 
             // Phase 5: Historian publishes. Pi uses native JSONL compaction entries, not OpenCode's deferred marker blob.
             await send("turn 17: Pi historian trigger pressure with eligible long tail", "pi phase 5 historian trigger", HISTORIAN_TRIGGER_USAGE);
-            await send("turn 18: Pi follow-up starts historian publication", "pi phase 5 historian follow-up");
+            // Turn 18's RESPONSE usage stays high so the first ordinary drive
+            // turn AFTER publication still sees >=85% and force-materializes
+            // (canConsumeDeferredLate) — that is the consuming pass which drains
+            // the staged marker. A low-usage turn 18 would leave the next pass
+            // in defer, where the deferred-history drain gate never opens.
+            await send("turn 18: Pi follow-up starts historian publication", "pi phase 5 historian follow-up", HISTORIAN_TRIGGER_USAGE);
             await h.waitFor(
                 () => {
                     const row = h
@@ -382,6 +387,15 @@ describe("long-running Pi Magic Context session", () => {
             expect(historianRange).not.toBeNull();
             expect(compartment.start_message).toBe(historianRange!.start);
             expect(compartment.end_message).toBe(historianRange!.end);
+            // Ordinary drive turn (NO HARD bust / m0-mutation injection): the
+            // publication's onPublished armed the deferred history-refresh +
+            // materialization signals, and this pass force-materializes (turn
+            // 18 kept pressure >=85%), rendering the new compartment into m[1].
+            // The widened coverage predicate (m[0] boundary OR current-pass
+            // m[1] delta) drains the staged marker into a native compaction
+            // entry on this consuming pass — parity with OpenCode's
+            // consuming-pass drain, which also rides ordinary send turns.
+            await send("turn 18b: ordinary Pi drive turn consumes the historian publication", "pi phase 5 consuming pass");
             const compactions = await h.waitFor(() => {
                 const entries = readCompactionEntries(h);
                 return entries.length > 0 ? entries : null;

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

@@ -3565,6 +3565,7 @@ describe("registerPiContextHandler", () => {
 					renderedBoundary: contention
 						? { endMessageId: "entry-1", ordinal: 1 }
 						: { endMessageId: "entry-2", ordinal: 2 },
+					m1RenderedCoverage: null,
 					syntheticLeadingCount: 0,
 				}),
 			);
@@ -3728,6 +3729,167 @@ describe("registerPiContextHandler", () => {
 				closeQuietly(db);
 			}
 		});
+
+		it("drains on m[1]-only coverage: fresh publication, no HARD fold", async () => {
+			const db = createTestDb();
+			const sessionId = "ses-pi-marker-m1-coverage";
+			const appendCompaction = mock(() => "compact-1");
+			// The exact shape a normal publication produces: m[0] still carries
+			// the empty pre-publication baseline (renderedBoundary <none> — no
+			// HARD fold has moved the compartment into m[0]), while the new
+			// compartment rendered into THIS pass's m[1] delta covers the marker.
+			const restoreInjection = contextHandlerInternals.setInjectM0M1PiForTests(
+				(_state, _db, _messages) => ({
+					injected: true,
+					compartmentCount: 1,
+					factCount: 0,
+					memoryCount: 0,
+					skippedVisibleMessages: 0,
+					m0Materialized: false,
+					m0Reason: null,
+					m0Bytes: 35,
+					m1Bytes: 518,
+					contentionExhausted: false,
+					renderedBoundary: { endMessageId: null, ordinal: null },
+					m1RenderedCoverage: { endMessageId: "entry-2", ordinal: 2 },
+					syntheticLeadingCount: 0,
+				}),
+			);
+			try {
+				seedCompartment(db, sessionId);
+				setPendingPiCompactionMarkerState(db, sessionId, {
+					firstKeptEntryId: "entry-3",
+					endMessageId: "entry-2",
+					ordinal: 2,
+					tokensBefore: 10,
+					summary: "summary",
+					publishedAt: 1,
+				});
+				signalPiDeferredHistoryRefresh(sessionId);
+				signalPiPendingMaterialization(sessionId);
+
+				await runDrainPass({
+					db,
+					sessionId,
+					appendCompaction,
+					contextPercent: 90,
+				});
+
+				expect(appendCompaction).toHaveBeenCalledTimes(1);
+				expect(getPendingPiCompactionMarkerState(db, sessionId)).toBeNull();
+				expect(consumeDeferredHistoryRefresh(sessionId)).toBe(false);
+			} finally {
+				restoreInjection();
+				clearContextHandlerSession(sessionId);
+				closeQuietly(db);
+			}
+		});
+
+		it("preserves the marker when a sibling-fallback serves stale m[1] (null coverage)", async () => {
+			const db = createTestDb();
+			const sessionId = "ses-pi-marker-sibling-fallback";
+			const appendCompaction = mock(() => "compact-1");
+			// softRefreshCachedM1Pi's sibling-fallback serves a sibling's stale
+			// cached m[1] with recomputed=false while contentionExhausted stays
+			// FALSE — the contention veto alone does not catch it, so the
+			// injection reports null m[1] coverage and the drain must skip.
+			const restoreInjection = contextHandlerInternals.setInjectM0M1PiForTests(
+				(_state, _db, _messages) => ({
+					injected: true,
+					compartmentCount: 1,
+					factCount: 0,
+					memoryCount: 0,
+					skippedVisibleMessages: 0,
+					m0Materialized: false,
+					m0Reason: null,
+					m0Bytes: 35,
+					m1Bytes: 518,
+					contentionExhausted: false,
+					renderedBoundary: { endMessageId: null, ordinal: null },
+					m1RenderedCoverage: null,
+					syntheticLeadingCount: 0,
+				}),
+			);
+			try {
+				seedCompartment(db, sessionId);
+				setPendingPiCompactionMarkerState(db, sessionId, {
+					firstKeptEntryId: "entry-3",
+					endMessageId: "entry-2",
+					ordinal: 2,
+					tokensBefore: 10,
+					summary: "summary",
+					publishedAt: 1,
+				});
+				signalPiDeferredHistoryRefresh(sessionId);
+				signalPiPendingMaterialization(sessionId);
+
+				await runDrainPass({
+					db,
+					sessionId,
+					appendCompaction,
+					contextPercent: 90,
+				});
+
+				expect(appendCompaction).not.toHaveBeenCalled();
+				expect(getPendingPiCompactionMarkerState(db, sessionId)).not.toBeNull();
+				// The deferred-history signal survives so the next FRESH render
+				// (non-fallback) retries the drain instead of losing the marker.
+				expect(consumeDeferredHistoryRefresh(sessionId)).toBe(true);
+			} finally {
+				restoreInjection();
+				clearContextHandlerSession(sessionId);
+				closeQuietly(db);
+			}
+		});
+
+		it("does not fire the drain on a pure defer pass even with coverage present", async () => {
+			const db = createTestDb();
+			const sessionId = "ses-pi-marker-defer-no-drain";
+			const appendCompaction = mock(() => "compact-1");
+			// Regression pin for the deferredHistoryDrainEligible gate: a pure
+			// SOFT+ defer/replay pass (no history-refresh consumption, no
+			// materialization this pass) must never drain — even when the
+			// pending marker exists and the injection reports full coverage.
+			const restoreInjection = contextHandlerInternals.setInjectM0M1PiForTests(
+				(_state, _db, _messages) => ({
+					injected: true,
+					compartmentCount: 1,
+					factCount: 0,
+					memoryCount: 0,
+					skippedVisibleMessages: 0,
+					m0Materialized: false,
+					m0Reason: null,
+					m0Bytes: 35,
+					m1Bytes: 518,
+					contentionExhausted: false,
+					renderedBoundary: { endMessageId: "entry-2", ordinal: 2 },
+					m1RenderedCoverage: { endMessageId: "entry-2", ordinal: 2 },
+					syntheticLeadingCount: 0,
+				}),
+			);
+			try {
+				seedCompartment(db, sessionId);
+				setPendingPiCompactionMarkerState(db, sessionId, {
+					firstKeptEntryId: "entry-3",
+					endMessageId: "entry-2",
+					ordinal: 2,
+					tokensBefore: 10,
+					summary: "summary",
+					publishedAt: 1,
+				});
+				// Deliberately NO deferred-history / materialization signals and
+				// no pressure: this is a replay pass, not a busting pass.
+
+				await runDrainPass({ db, sessionId, appendCompaction });
+
+				expect(appendCompaction).not.toHaveBeenCalled();
+				expect(getPendingPiCompactionMarkerState(db, sessionId)).not.toBeNull();
+			} finally {
+				restoreInjection();
+				clearContextHandlerSession(sessionId);
+				closeQuietly(db);
+			}
+		});
 	});
 });
 

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

@@ -4058,10 +4058,28 @@ function pendingPiMarkerCoveredByRenderedBoundary(
 	pending: PendingPiCompactionMarker,
 	injection: PiInjectionResult | null,
 ): boolean {
+	// Contention fallbacks must never authorize a native trim: the served
+	// bytes may lag the latest compartment snapshot.
 	if (!injection || injection.contentionExhausted) return false;
+	// m[0] arm: the boundary rendered into the m[0] snapshot.
 	const boundary = injection.renderedBoundary;
 	if (pending.endMessageId === boundary.endMessageId) return true;
-	return boundary.ordinal !== null && pending.ordinal <= boundary.ordinal;
+	if (boundary.ordinal !== null && pending.ordinal <= boundary.ordinal)
+		return true;
+	// m[1] arm (liveness fix): fresh publications render their compartment
+	// into the m[1] delta, not m[0] (which folds only on a HARD bust), so the
+	// m[0] snapshot boundary stays behind the pending marker until an
+	// unrelated HARD fold — starving the drain for hours in stable sessions.
+	// Accept coverage from a compartment actually rendered into m[1] THIS
+	// pass. The field is non-null only when m[1] was freshly recomputed this
+	// pass without a contention fallback (null on cached/sibling replay, where
+	// contentionExhausted alone would miss the sibling-fallback's stale
+	// bytes), so this restores OpenCode's consuming-pass drain parity without
+	// ever trimming getBranch() beyond content the model was shown this pass.
+	const m1Coverage = injection.m1RenderedCoverage;
+	if (!m1Coverage) return false;
+	if (pending.endMessageId === m1Coverage.endMessageId) return true;
+	return m1Coverage.ordinal !== null && pending.ordinal <= m1Coverage.ordinal;
 }
 
 function captureReasoningMutationRollback(
@@ -5184,9 +5202,10 @@ async function runPipeline(args: RunPipelineArgs): Promise<RunPipelineResult> {
 				suppressDeferredHistoryDrain = true;
 				preserveDeferredMaterializationForMarkerDrain = true;
 				const boundary = injectionResult?.renderedBoundary;
+				const m1Coverage = injectionResult?.m1RenderedCoverage;
 				sessionLog(
 					args.sessionId,
-					`Pi compaction-marker drain skipped: pending ordinal ${pending.ordinal} is newer than rendered boundary ${boundary?.ordinal ?? "<none>"} endMessageId=${boundary?.endMessageId ?? "<none>"}; preserving deferred signals`,
+					`Pi compaction-marker drain skipped: pending ordinal ${pending.ordinal} is newer than rendered boundary ${boundary?.ordinal ?? "<none>"} endMessageId=${boundary?.endMessageId ?? "<none>"} (m[1] coverage ${m1Coverage?.ordinal ?? "<none>"} endMessageId=${m1Coverage?.endMessageId ?? "<none>"}); preserving deferred signals`,
 				);
 			} else if (!args.appendCompaction || !args.readBranchEntries) {
 				suppressDeferredHistoryDrain = true;