mason: fix historian atomic publish and Pi project marker

Author: ualtinok

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

@@ -1519,13 +1519,6 @@ export function registerPiContextHandler(
 				previousModelKey !== undefined &&
 				currentModelKey !== undefined &&
 				previousModelKey !== currentModelKey;
-			if (modelChanged) {
-				clearPiM0Cache(
-					options.db,
-					sessionId,
-					`model switch ${previousModelKey} -> ${currentModelKey}`,
-				);
-			}
 			if (currentModelKey !== undefined) {
 				liveModelBySession.set(sessionId, currentModelKey);
 			}

packages/pi-plugin/src/index.ts

@@ -1094,7 +1094,6 @@ export default async function (pi: ExtensionAPI): Promise<void> {
 			});
 
 			if (result.hashChanged) {
-				clearPiM0Cache(db, sessionId, "system prompt hash change");
 				// Real prompt-content change. Cache prefix is already
 				// busted on this turn. Signal all three independent
 				// refresh sets so the next pi.on("context") event

packages/pi-plugin/src/inject-compartments-pi.test.ts

@@ -1451,6 +1451,160 @@ describe("mustMaterializePi — SOFT/HARD taxonomy (parity with OpenCode)", () =
 		}
 	});
 
+	it("lazy-adopts a NULL cached project marker without a no-switch HARD fold", () => {
+		const db = createTestDb();
+		const cwd = mkdtempSync(join(tmpdir(), "pi-tax-project-null-a-"));
+		const cwdB = mkdtempSync(join(tmpdir(), "pi-tax-project-null-b-"));
+		try {
+			const state = {
+				...piState("ses-pi-tax-project-null", cwd),
+				hardSignals: baseHard,
+			};
+			appendCompartments(db, state.sessionId, [compartment(0, "Alpha")]);
+			const first = [userMessage("hi", 10)];
+			injectM0M1Pi(state, db, first as never, ["entry-0"]);
+			const baselineM0 = textOf(first[0] as never);
+
+			db.prepare(
+				"UPDATE session_meta SET cached_m0_project_identity = NULL WHERE session_id = ?",
+			).run(state.sessionId);
+
+			expect(mustMaterializePi(state, db)).toEqual({
+				value: false,
+				reason: null,
+			});
+			const noSwitch = [userMessage("same project", 11)];
+			const noSwitchResult = injectM0M1Pi(state, db, noSwitch as never, [
+				"entry-0",
+			]);
+
+			expect(noSwitchResult.m0Materialized).toBe(false);
+			expect(noSwitchResult.m0Reason).not.toBe("first_render");
+			expect(noSwitchResult.m0Reason).not.toBe("project_change");
+			expect(textOf(noSwitch[0] as never)).toBe(baselineM0);
+			expect(
+				db
+					.prepare(
+						"SELECT cached_m0_project_identity FROM session_meta WHERE session_id = ?",
+					)
+					.get(state.sessionId),
+			).toEqual({ cached_m0_project_identity: state.projectIdentity });
+
+			const switched = {
+				...piState(state.sessionId, cwdB),
+				hardSignals: baseHard,
+			};
+			expect(mustMaterializePi(switched, db)).toEqual({
+				value: true,
+				reason: "project_change",
+			});
+		} finally {
+			rmSync(cwd, { recursive: true, force: true });
+			rmSync(cwdB, { recursive: true, force: true });
+			closeQuietly(db);
+		}
+	});
+
+	it("HARD: a genuine same-session project switch folds exactly once, then stabilizes", () => {
+		const db = createTestDb();
+		const cwdA = mkdtempSync(join(tmpdir(), "pi-tax-project-a-"));
+		const cwdB = mkdtempSync(join(tmpdir(), "pi-tax-project-b-"));
+		try {
+			const stateA = {
+				...piState("ses-pi-tax-project-switch", cwdA),
+				hardSignals: baseHard,
+			};
+			insertMemory(db, {
+				projectPath: stateA.projectIdentity,
+				category: "ARCHITECTURE",
+				content: "Project A memory must not replay after /cd.",
+			});
+			const first = [userMessage("hi", 10)];
+			injectM0M1Pi(stateA, db, first as never, undefined, true);
+			expect(textOf(first[0] as never)).toContain("Project A memory");
+
+			const stateB = {
+				...piState(stateA.sessionId, cwdB),
+				hardSignals: baseHard,
+			};
+			insertMemory(db, {
+				projectPath: stateB.projectIdentity,
+				category: "ARCHITECTURE",
+				content: "Project B memory is the switched project baseline.",
+			});
+
+			const switched = [userMessage("after cd", 11)];
+			const switchedResult = injectM0M1Pi(
+				stateB,
+				db,
+				switched as never,
+				undefined,
+				true,
+			);
+			expect(switchedResult.m0Materialized).toBe(true);
+			expect(switchedResult.m0Reason).toBe("project_change");
+			expect(textOf(switched[0] as never)).toContain("Project B memory");
+			expect(textOf(switched[0] as never)).not.toContain("Project A memory");
+
+			const stable = [userMessage("after cd stable", 12)];
+			const stableResult = injectM0M1Pi(
+				stateB,
+				db,
+				stable as never,
+				undefined,
+				false,
+			);
+			expect(stableResult.m0Materialized).toBe(false);
+			expect(stableResult.m0Reason).toBeNull();
+			expect(textOf(stable[0] as never)).toBe(textOf(switched[0] as never));
+		} finally {
+			rmSync(cwdA, { recursive: true, force: true });
+			rmSync(cwdB, { recursive: true, force: true });
+			closeQuietly(db);
+		}
+	});
+
+	it("model and system changes materialize with classified reasons, not first_render", () => {
+		const db = createTestDb();
+		const cwdModel = mkdtempSync(join(tmpdir(), "pi-tax-model-reason-"));
+		const cwdSystem = mkdtempSync(join(tmpdir(), "pi-tax-system-reason-"));
+		try {
+			const modelState = {
+				...piState("ses-pi-tax-model-reason", cwdModel),
+				hardSignals: baseHard,
+			};
+			injectM0M1Pi(modelState, db, [userMessage("hi", 10)] as never);
+			const modelChanged = {
+				...modelState,
+				hardSignals: { ...baseHard, modelKey: "anthropic/sonnet" },
+			};
+			const modelPass = [userMessage("model", 11)];
+			const modelResult = injectM0M1Pi(modelChanged, db, modelPass as never);
+			expect(modelResult.m0Materialized).toBe(true);
+			expect(modelResult.m0Reason).toBe("model_change");
+			expect(modelResult.m0Reason).not.toBe("first_render");
+
+			const systemState = {
+				...piState("ses-pi-tax-system-reason", cwdSystem),
+				hardSignals: baseHard,
+			};
+			injectM0M1Pi(systemState, db, [userMessage("hi", 10)] as never);
+			const systemChanged = {
+				...systemState,
+				hardSignals: { ...baseHard, systemHash: "sys-v2" },
+			};
+			const systemPass = [userMessage("system", 11)];
+			const systemResult = injectM0M1Pi(systemChanged, db, systemPass as never);
+			expect(systemResult.m0Materialized).toBe(true);
+			expect(systemResult.m0Reason).toBe("system_hash");
+			expect(systemResult.m0Reason).not.toBe("first_render");
+		} finally {
+			rmSync(cwdModel, { recursive: true, force: true });
+			rmSync(cwdSystem, { recursive: true, force: true });
+			closeQuietly(db);
+		}
+	});
+
 	it("an empty current HARD signal is never treated as a change", () => {
 		const db = createTestDb();
 		const cwd = mkdtempSync(join(tmpdir(), "pi-tax-empty-"));

packages/pi-plugin/src/inject-compartments-pi.ts

@@ -373,7 +373,7 @@ function injectHistoryBlockIntoFirstUserMessage(
 ): boolean {
 	for (let i = 0; i < piMessages.length; i++) {
 		const msg = piMessages[i];
-		if (!msg || msg.role !== "user") continue;
+		if (msg?.role !== "user") continue;
 
 		const userMsg = msg as PiUserMessage;
 		if (typeof userMsg.content === "string") {
@@ -610,6 +610,9 @@ export interface PiM0SnapshotMarkers {
 	// Captured from PiM0HardSignals at the injection call site.
 	systemHash: string;
 	modelKey: string;
+	// Pi sessions can switch projects in-process (`/cd`). NULL on legacy cached
+	// rows means unknown/lazy-adopted and must not force a HARD fold by itself.
+	projectIdentity: string | null;
 }
 
 /**
@@ -770,6 +773,7 @@ function getCachedMarkers(
 		lastBaselineEndMessageId: cachedBoundary,
 		systemHash: meta.cachedM0SystemHash ?? "",
 		modelKey: meta.cachedM0ModelKey ?? "",
+		projectIdentity: meta.cachedM0ProjectIdentity ?? null,
 	};
 }
 
@@ -859,6 +863,7 @@ function readCurrentMarkersFromCompartments(
 		lastBaselineEndMessageId: lastBaselineEndMessageId(compartments),
 		systemHash: (state.hardSignals ?? EMPTY_PI_HARD_SIGNALS).systemHash,
 		modelKey: (state.hardSignals ?? EMPTY_PI_HARD_SIGNALS).modelKey,
+		projectIdentity: state.projectIdentity,
 	};
 }
 
@@ -907,6 +912,15 @@ export function mustMaterializePi(
 	) {
 		return { value: true, reason: "system_hash" };
 	}
+	// Pi can switch projects within the same session (`/cd`). Legacy cached rows
+	// have a NULL marker: treat that as unknown/MATCH for lazy adoption so the
+	// first post-upgrade no-switch pass does not HARD-fold existing sessions.
+	if (
+		meta.cachedM0ProjectIdentity !== null &&
+		meta.cachedM0ProjectIdentity !== state.projectIdentity
+	) {
+		return { value: true, reason: "project_change" };
+	}
 	// Idle > TTL: self-consuming guard via cachedM0MaterializedAt (parity with
 	// OpenCode). cacheExpired stays true every pass until lastResponseTime
 	// updates, so fold only when the last response is newer than the last
@@ -1227,6 +1241,7 @@ function readFrozenM0InputsPi(
 			lastBaselineEndMessageId: lastBaselineEndMessageId(compartments),
 			systemHash: (state.hardSignals ?? EMPTY_PI_HARD_SIGNALS).systemHash,
 			modelKey: (state.hardSignals ?? EMPTY_PI_HARD_SIGNALS).modelKey,
+			projectIdentity: state.projectIdentity,
 		};
 		return { docs, markers, compartments, memories, userProfile, workspace };
 	});
@@ -1391,6 +1406,7 @@ export function materializeM0Pi(
 			current.maxCompartmentSeq !== snapshotMarkers.maxCompartmentSeq ||
 			current.maxMutationId !== snapshotMarkers.maxMutationId ||
 			current.maxMemoryMutationId !== snapshotMarkers.maxMemoryMutationId ||
+			current.projectIdentity !== snapshotMarkers.projectIdentity ||
 			// Inert today (both harnesses pin sessionFactsVersion to 0 — facts are
 			// retired in v2), but kept for structural parity with OpenCode
 			// materializeM0 so the two stale checks can't silently drift if either
@@ -1428,6 +1444,7 @@ export function materializeM0Pi(
 			upgradeState: snapshotMarkers.upgradeState,
 			systemHash: snapshotMarkers.systemHash,
 			modelKey: snapshotMarkers.modelKey,
+			projectIdentity: snapshotMarkers.projectIdentity,
 		});
 		// Persist the rendered-memory identity in the SAME transaction as the m[0]
 		// snapshot (parity with OpenCode materializeM0). `memory_block_ids` /
@@ -1743,6 +1760,7 @@ interface CachedPiM0M1Row {
 	cached_m0_upgrade_state: string | null;
 	cached_m0_system_hash: string | null;
 	cached_m0_model_key: string | null;
+	cached_m0_project_identity: string | null;
 	cached_m0_last_baseline_end_message_id: string | null;
 	memory_block_ids: string | null;
 }
@@ -1792,6 +1810,7 @@ function readCachedPiM0M1Row(
 					cached_m0_upgrade_state,
 					cached_m0_system_hash,
 					cached_m0_model_key,
+					cached_m0_project_identity,
 					cached_m0_last_baseline_end_message_id,
 					memory_block_ids
 			   FROM session_meta
@@ -1836,6 +1855,7 @@ function markersFromCachedPiRow(
 				: null,
 		systemHash: row.cached_m0_system_hash ?? "",
 		modelKey: row.cached_m0_model_key ?? "",
+		projectIdentity: row.cached_m0_project_identity ?? null,
 	};
 }
 
@@ -1870,6 +1890,8 @@ function cachedPiRowMatchesSnapshot(args: {
 		// this process's cached row so the soft-refresh CAS adopts the sibling's m[0].
 		(rowMarkers.systemHash ?? "") === (args.markers.systemHash ?? "") &&
 		(rowMarkers.modelKey ?? "") === (args.markers.modelKey ?? "") &&
+		(rowMarkers.projectIdentity ?? null) ===
+			(args.markers.projectIdentity ?? null) &&
 		// Workspace fingerprint (parity with OpenCode cachedRowMatchesState):
 		// projectMemoryEpoch above only tracks THIS session's own project, but a
 		// FOREIGN member's epoch bump changes the workspace fingerprint without
@@ -1881,6 +1903,15 @@ function cachedPiRowMatchesSnapshot(args: {
 	);
 }
 
+function adoptCachedPiProjectIdentity(
+	db: ContextDatabase,
+	state: PiM0M1State,
+): void {
+	db.prepare(
+		"UPDATE session_meta SET cached_m0_project_identity = ? WHERE session_id = ? AND cached_m0_project_identity IS NULL",
+	).run(state.projectIdentity, state.sessionId);
+}
+
 function decodeCachedM1(row: CachedPiM0M1Row, sessionId: string): string {
 	if (!row.cached_m1_bytes) {
 		throw new PiMaterializeContentionError(
@@ -2168,6 +2199,10 @@ export function injectM0M1Pi(
 			`missing m[0] markers for ${state.sessionId}`,
 		);
 	}
+	if (!materialized && markers.projectIdentity === null) {
+		adoptCachedPiProjectIdentity(db, state);
+		markers = { ...markers, projectIdentity: state.projectIdentity };
+	}
 
 	if (materialized) {
 		// m[1] was rendered and persisted atomically inside materializeM0Pi.

packages/pi-plugin/src/pi-historian-runner.test.ts

@@ -140,6 +140,9 @@ async function runHistorianWith(args: {
 	refreshBoundarySnapshot?: () => ProtectedTailBoundarySnapshot;
 	providerMessages?: ReturnType<typeof rawMessages>;
 	beforeRun?: (db: ReturnType<typeof createTestDb>) => void;
+	ensureProjectRegistered?: Parameters<
+		typeof runPiHistorian
+	>[0]["ensureProjectRegistered"];
 }) {
 	const db = createTestDb();
 	const runner = runnerReturning([...args.outputs]);
@@ -164,6 +167,7 @@ async function runHistorianWith(args: {
 		boundarySnapshot: args.boundarySnapshot,
 		refreshBoundarySnapshot: args.refreshBoundarySnapshot,
 		compartmentLeaseHolderId: holderId,
+		ensureProjectRegistered: args.ensureProjectRegistered,
 	});
 	return { db, runner };
 }
@@ -463,6 +467,39 @@ describe("runPiHistorian", () => {
 		}
 	});
 
+	it("keeps publish succeeded and signaled when post-commit project registration throws", async () => {
+		const onPublished = mock(() => undefined);
+		const ensureProjectRegistered = mock(async () => {
+			throw new Error("embedding provider unavailable");
+		});
+		const { db } = await runHistorianWith({
+			outputs: [successXml("Pi durable fact survives registration outage.")],
+			onPublished,
+			ensureProjectRegistered,
+		});
+		try {
+			await new Promise((resolve) => setTimeout(resolve, 0));
+			expect(onPublished).toHaveBeenCalledTimes(1);
+			expect(getCompartments(db, "ses-historian")).toHaveLength(1);
+			expect(
+				loadProtectedTailMeta(db, "ses-historian").priorBoundaryOrdinal,
+			).toBe(3);
+			const projectPath = resolveProjectIdentity(process.cwd());
+			expect(
+				getMemoriesByProject(db, projectPath).map((memory) => memory.content),
+			).toContain("Pi durable fact survives registration outage.");
+			expect(
+				db
+					.prepare(
+						"SELECT status FROM historian_runs WHERE session_id = ? ORDER BY id DESC LIMIT 1",
+					)
+					.get("ses-historian"),
+			).toEqual({ status: "success" });
+		} finally {
+			closeQuietly(db);
+		}
+	});
+
 	it("queues a Pi-native compaction marker after publication", async () => {
 		const appendCompaction = mock(() => "compact-1");
 		const entries = Array.from({ length: 6 }, (_, index) => ({