fix: close 5 correctness findings from Oracle audit wave 2

ualtinok · ualtinok · commit c94798cc7060 · 2026-06-19T00:45:12.000+02:00
Three parallel read-only Oracles (nudges/emergency-drop/reclaim; telemetry/
forward-pressure/boundary; historian pipeline/decay/recomp/migration). Cores
verified safe (Ch1/emergency-math/two-pass/defer-safety; forward-pressure floor;
protected-tail open-arc fence; decay defer-replay; parse/validate). Five
source-confirmed fixes; two design questions banked to
.alfonso/oracle-loop-findings.md (D3 emergency skeleton policy, D4 historian
publish atomicity — both structural/cache-path, need review before changing).

1. Channel 2 stale-delivery (event-handler.ts). deliverChannel2IfPending ignored
   the baseline's reducedSinceRefresh flag, so if the agent already called
   ctx_reduce since the last transform, delivery used stale-high tail tokens and
   nudged the agent to drop output it just dropped. Skip when reducedSinceRefresh
   (the pending intent stays armed for the next fresh re-eval). PARITY: Pi's
   maybeDeliverChannel2Pi already had this guard — this is an OpenCode parity fix.

2. Discard-last user-observation gap (compartment-runner-incremental.ts + Pi
   pi-historian-runner.ts). Facts are skipped on a discard-last run (the
   provisional last compartment re-emits next run), but user_observations were
   stored unconditionally → a reworded re-emission next run double-stores the
   candidate. Gate observations on !discardedLast too (hoisted the Pi compute
   above the observations block). Both harnesses.

3. Memory-migration apply+guard atomicity (memory-migration.ts + Pi
   pi-memory-migration.ts). applyMemoryMigration (delete+reinsert+epoch bump) and
   markMemoryMigrationDone ran separately; a crash between them left the project
   migrated-to-v2 but UNGUARDED → a retry re-migrates v2 rows (new ids/stats/
   embeddings, dup observation candidates). Wrap both in one db.transaction()
   (nested = savepoint, inner tx still works). Both harnesses.

4. transform_decisions retention cap missing (transform-decision-log.ts). The
   documented 2000-row/(session,harness) cap was never enforced — only full
   session-delete pruned. A long session's cache-affecting passes grew the
   telemetry table unbounded (dashboard loads all rows for attribution). Prune
   newest-2000 after each insert on the same non-blocking handle. +tests.

5. Pi telemetry older-entry misattribution (transform-decision-log.ts). The
   backward value-skip resolver returned an OLDER assistant when no new one had
   arrived (branch still ends at snapshot) → this pass's cache decision recorded
   against the wrong message. Made it index-aware: resolve the snapshot's index,
   bind only to the first assistant AFTER it; refuse (null, stay pending) if the
   snapshot is absent. Dashboard-attribution only (no runtime/cache corruption).
   +tests.

Also: corrected the stale Channel-1 denominator comment (ctx-reduce-nudge.ts:
historyBudget → workingWindow). Gate: plugin 2174/0 (+6), Pi 471/0, tsc+biome
clean both harnesses.
diff --git a/packages/pi-plugin/src/pi-historian-runner.ts b/packages/pi-plugin/src/pi-historian-runner.ts
@@ -940,12 +940,19 @@ export async function runPiHistorian(deps: PiHistorianDeps): Promise<void> {
 			// rendered m[0]/m[1] bytes.
 			queueDropsForCompartmentalizedMessages(db, sessionId, lastNewEnd);
 
+			// discard-last: when the provisional last compartment was dropped, its
+			// facts AND observations are not durable yet — skip both this run
+			// (unanchored; re-derived next run). Computed before the observations
+			// block so we can gate it too (parity with OpenCode).
+			const discardedLast = newCompartments.length < emittedCompartments.length;
+
 			// user observations are inserted POST-COMMIT,
 			// best-effort, so an auxiliary failure never rolls back the publish.
 			// Gated on the user-memory feature so opted-out users never have
 			// behavioral candidates persisted (privacy parity with OpenCode).
 			if (
 				userMemoriesEnabled === true &&
+				!discardedLast &&
 				validatedPass.userObservations?.length
 			) {
 				try {
@@ -971,11 +978,6 @@ export async function runPiHistorian(deps: PiHistorianDeps): Promise<void> {
 				}
 			}
 
-			// discard-last: when the provisional last
-			// compartment was dropped, its facts are not durable yet — skip fact
-			// promotion this run (facts are unanchored; re-derived next run).
-			const discardedLast = newCompartments.length < emittedCompartments.length;
-
 			// register the project for embeddings against the
 			// LIVE directory ONCE up front (not inside the promotion block), so a
 			// discard-last pass that skips promotion still registers before the
diff --git a/packages/pi-plugin/src/pi-memory-migration.ts b/packages/pi-plugin/src/pi-memory-migration.ts
@@ -193,13 +193,15 @@ export async function runPiMemoryMigration(
 		routed = parsed.userObservations.length;
 	}
 
-	const { removed, inserted } = applyMemoryMigration(
-		deps.db,
-		projectPath,
-		parsed,
-	);
-
-	markMemoryMigrationDone(deps.db, projectPath);
+	// Apply the destructive rewrite AND set the done-guard atomically (parity with
+	// OpenCode). Separate, a crash between them leaves the project migrated-to-v2
+	// but UNGUARDED, so a retry re-migrates v2 rows. A nested db.transaction() runs
+	// as a savepoint, so applyMemoryMigration's inner transaction still works.
+	const { removed, inserted } = deps.db.transaction(() => {
+		const counts = applyMemoryMigration(deps.db, projectPath, parsed);
+		markMemoryMigrationDone(deps.db, projectPath);
+		return counts;
+	})();
 	return {
 		ran: true,
 		summary: `Re-evaluated ${removed} memor${removed === 1 ? "y" : "ies"} into ${inserted} v2-taxonomy memor${inserted === 1 ? "y" : "ies"}${routed > 0 ? `, routed ${routed} user trait${routed === 1 ? "" : "s"} to your profile` : ""}.`,
diff --git a/packages/plugin/src/features/magic-context/memory/memory-migration.ts b/packages/plugin/src/features/magic-context/memory/memory-migration.ts
@@ -510,9 +510,17 @@ export async function runMemoryMigration(
             routed = result.userObservations.length;
         }
 
-        const { removed, inserted } = applyMemoryMigration(db, projectPath, result);
-
-        markMemoryMigrationDone(db, projectPath);
+        // Apply the destructive rewrite AND set the done-guard atomically. If they
+        // were separate and a crash landed between them, the project would be left
+        // already-migrated-to-v2 but UNGUARDED — a retry would re-migrate v2 rows
+        // (new ids/stats/embeddings, possible duplicate observation candidates).
+        // A nested db.transaction() runs as a savepoint, so applyMemoryMigration's
+        // own inner transaction still works.
+        const { removed, inserted } = db.transaction(() => {
+            const counts = applyMemoryMigration(db, projectPath, result);
+            markMemoryMigrationDone(db, projectPath);
+            return counts;
+        })();
         return {
             ran: true,
             removed,
diff --git a/packages/plugin/src/features/magic-context/transform-decision-log.test.ts b/packages/plugin/src/features/magic-context/transform-decision-log.test.ts
@@ -0,0 +1,121 @@
+import { afterEach, beforeEach, describe, expect, it } from "bun:test";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { Database } from "../../shared/sqlite";
+import { closeQuietly } from "../../shared/sqlite-helpers";
+import { runMigrations } from "./migrations";
+import { initializeDatabase } from "./storage-db";
+import { __test, TRANSFORM_DECISIONS_RETENTION } from "./transform-decision-log";
+
+let dir: string;
+let dbPath: string;
+let db: Database;
+
+beforeEach(() => {
+    dir = mkdtempSync(join(tmpdir(), "mc-txn-decision-"));
+    dbPath = join(dir, "context.db");
+    db = new Database(dbPath);
+    initializeDatabase(db);
+    runMigrations(db);
+    __test.reset();
+});
+
+afterEach(() => {
+    closeQuietly(db);
+    rmSync(dir, { recursive: true, force: true });
+    __test.reset();
+});
+
+function baseRow(messageId: string, tsMs: number) {
+    return {
+        sessionId: "ses-1",
+        harness: "opencode" as const,
+        messageId,
+        tsMs,
+        decision: "execute" as const,
+        materialized: true,
+        materializeReason: "model_change" as const,
+        emergency: false,
+        droppedTokens: 0,
+        droppedCount: 0,
+        inputTokens: 100,
+        bustedThisPass: true,
+    };
+}
+
+function rowCount(): number {
+    return (
+        db
+            .prepare(
+                "SELECT COUNT(*) AS c FROM transform_decisions WHERE session_id = 'ses-1' AND harness = 'opencode'",
+            )
+            .get() as { c: number }
+    ).c;
+}
+
+describe("transform_decisions retention cap", () => {
+    it("prunes to the newest TRANSFORM_DECISIONS_RETENTION rows per (session,harness)", () => {
+        // Write cap + 5 rows with strictly increasing ts and distinct message ids.
+        const total = TRANSFORM_DECISIONS_RETENTION + 5;
+        for (let i = 0; i < total; i++) {
+            __test.writeRow(dbPath, baseRow(`msg-${i}`, 1000 + i));
+        }
+        expect(rowCount()).toBe(TRANSFORM_DECISIONS_RETENTION);
+
+        // The oldest (smallest ts) must be gone; the newest must remain.
+        const oldest = db
+            .prepare("SELECT 1 FROM transform_decisions WHERE message_id = 'msg-0'")
+            .get();
+        const newest = db
+            .prepare(`SELECT 1 FROM transform_decisions WHERE message_id = 'msg-${total - 1}'`)
+            .get();
+        expect(oldest ?? null).toBeNull();
+        expect(newest ?? null).not.toBeNull();
+    });
+
+    it("does not prune below the cap", () => {
+        for (let i = 0; i < 10; i++) {
+            __test.writeRow(dbPath, baseRow(`m-${i}`, 1000 + i));
+        }
+        expect(rowCount()).toBe(10);
+    });
+});
+
+describe("findNewestPiAssistantEntryIdAfter (index-aware binding)", () => {
+    const asst = (id: string) => ({
+        id,
+        type: "message",
+        message: { role: "assistant" },
+    });
+    const user = (id: string) => ({
+        id,
+        type: "message",
+        message: { role: "user" },
+    });
+
+    it("binds to the first assistant AFTER the snapshot", () => {
+        const entries = [asst("a1"), user("u1"), asst("a2")];
+        expect(__test.findNewestPiAssistantEntryIdAfter(entries, "a1")).toBe("a2");
+    });
+
+    it("returns null when no assistant exists after the snapshot (no older-entry fallback)", () => {
+        // Branch still ends at the snapshot — a value-skip scan would wrongly
+        // return the older a1; the index-aware version must refuse.
+        const entries = [asst("a1"), asst("a2")];
+        expect(__test.findNewestPiAssistantEntryIdAfter(entries, "a2")).toBeNull();
+    });
+
+    it("refuses to bind when the snapshot id is absent (compacted/reordered)", () => {
+        const entries = [asst("a1"), asst("a2")];
+        expect(__test.findNewestPiAssistantEntryIdAfter(entries, "missing-snapshot")).toBeNull();
+    });
+
+    it("with a null snapshot, binds to the FIRST assistant (recorded when none existed)", () => {
+        // Null snapshot = no assistant at record time, so the first assistant to
+        // arrive is the one this decision belongs to (not the newest — that would
+        // misattribute to a later pass's message if resolve lagged).
+        const entries = [asst("a1"), user("u1"), asst("a2"), user("u2")];
+        expect(__test.findNewestPiAssistantEntryIdAfter(entries, null)).toBe("a1");
+    });
+});
diff --git a/packages/plugin/src/features/magic-context/transform-decision-log.ts b/packages/plugin/src/features/magic-context/transform-decision-log.ts
@@ -5,6 +5,14 @@ import { getDatabasePath } from "./storage-db";
 export type TransformDecisionHarness = "opencode" | "pi";
 export type TransformSchedulerDecision = "execute" | "defer";
 
+/**
+ * Max transform_decisions rows kept per (session_id, harness). Pruned newest-first
+ * after every insert so a long session's cache-affecting passes never grow this
+ * telemetry table without bound (the dashboard loads all matching rows for cause
+ * attribution).
+ */
+export const TRANSFORM_DECISIONS_RETENTION = 2000;
+
 export type CanonicalMaterializeReason =
     | "system_hash"
     | "model_change"
@@ -247,16 +255,41 @@ function findNewestPiAssistantEntryIdAfter(
     snapshotNewestAssistantEntryId: string | null,
 ): string | null {
     if (!Array.isArray(entries)) return null;
-    for (let i = entries.length - 1; i >= 0; i--) {
+
+    // Bind only to an assistant entry positioned AFTER the snapshot. A pure
+    // value-skip backward scan misattributes when NO new assistant has arrived
+    // yet (the branch still ends at the snapshot): it would skip the snapshot by
+    // value and fall back to an OLDER assistant, recording THIS pass's cache
+    // decision against the wrong message. Resolve the snapshot's INDEX first, then
+    // return the first assistant after it. If the snapshot id is absent (compacted
+    // away / reordered), refuse to bind (return null) — the pending row stays for
+    // a later pass (at most one per session; overwritten by the next bust), never
+    // attaching to an older entry.
+    let startIndex = 0;
+    if (snapshotNewestAssistantEntryId !== null) {
+        let snapshotIndex = -1;
+        for (let i = entries.length - 1; i >= 0; i--) {
+            const entry = entries[i];
+            if (
+                entry &&
+                typeof entry === "object" &&
+                (entry as { id?: unknown }).id === snapshotNewestAssistantEntryId
+            ) {
+                snapshotIndex = i;
+                break;
+            }
+        }
+        if (snapshotIndex === -1) return null;
+        startIndex = snapshotIndex + 1;
+    }
+
+    for (let i = startIndex; i < entries.length; i++) {
         const entry = entries[i];
         if (!entry || typeof entry !== "object") continue;
         const row = entry as { id?: unknown; type?: unknown; message?: unknown };
         if (row.type !== "message" || typeof row.id !== "string" || row.id.length === 0) {
             continue;
         }
-        if (snapshotNewestAssistantEntryId !== null && row.id === snapshotNewestAssistantEntryId) {
-            continue;
-        }
         const message = row.message;
         if (
             message &&
@@ -301,6 +334,27 @@ function writeTransformDecisionRow(dbPath: string, row: TransformDecisionRow): v
             Math.max(0, Math.floor(row.droppedCount)),
             Math.max(0, Math.floor(row.inputTokens)),
         );
+        // Enforce the per-(session,harness) retention cap so a long session's
+        // cache-affecting passes can't grow this telemetry table unbounded (the
+        // dashboard loads all matching rows for cause attribution). Keep the
+        // newest TRANSFORM_DECISIONS_RETENTION rows by (ts_ms, rowid). Best-effort
+        // on the same non-blocking handle; a failure just defers the prune.
+        db.prepare(
+            `DELETE FROM transform_decisions
+             WHERE session_id = ? AND harness = ?
+               AND rowid NOT IN (
+                 SELECT rowid FROM transform_decisions
+                 WHERE session_id = ? AND harness = ?
+                 ORDER BY ts_ms DESC, rowid DESC
+                 LIMIT ?
+               )`,
+        ).run(
+            row.sessionId,
+            row.harness,
+            row.sessionId,
+            row.harness,
+            TRANSFORM_DECISIONS_RETENTION,
+        );
     } finally {
         closeQuietly(db);
     }
@@ -323,4 +377,8 @@ export const __test = {
     setWriterForTests(writer: TransformDecisionWriter | null): void {
         writerOverrideForTests = writer;
     },
+    writeRow(dbPath: string, row: TransformDecisionRow): void {
+        writeTransformDecisionRow(dbPath, row);
+    },
+    findNewestPiAssistantEntryIdAfter,
 };
diff --git a/packages/plugin/src/hooks/magic-context/compartment-runner-incremental.ts b/packages/plugin/src/hooks/magic-context/compartment-runner-incremental.ts
@@ -762,8 +762,13 @@ export async function runCompartmentAgent(deps: CompartmentRunnerDeps): Promise<
         // Store user behavior observations as candidates ONLY when the user-memory
         // feature is enabled. Without this gate we'd persist behavioral candidates
         // for users who opted out of user memories entirely (privacy).
+        // discard-last: skip observation candidates for the discarded provisional
+        // compartment for the SAME reason facts are skipped above — observations
+        // are unanchored, so a reworded re-emission next run would double-store.
+        // (`discardedLast` computed above for the fact-promotion gate.)
         if (
             deps.experimentalUserMemories === true &&
+            !discardedLast &&
             validatedPass.userObservations &&
             validatedPass.userObservations.length > 0
         ) {
diff --git a/packages/plugin/src/hooks/magic-context/ctx-reduce-nudge.ts b/packages/plugin/src/hooks/magic-context/ctx-reduce-nudge.ts
@@ -4,10 +4,13 @@
 // later transform, so this is "free sticky" — no anchor store, no CAS, no
 // replay machinery (unlike the deleted assistant/user-anchored nudges).
 //
-// The metric is `severity = (undropped / historyBudget) × pressure`:
+// The metric is `severity = (undropped / workingWindow) × pressure`:
 //   - `undropped` = approximate tokens of NON-dropped tool output in the live
-//     tail (dropped outputs are `[dropped …]` / `[truncated]` sentinels, so a
-//     simple tail walk excludes them — no agent-vs-heuristic attribution).
+//     tail (dropped outputs are `[dropped …]` sentinels, so a simple tail walk
+//     excludes them — no agent-vs-heuristic attribution).
+//   - `workingWindow` = the execute-threshold token budget (the range the agent
+//     actually moves in), NOT the tiny historyBudget — using historyBudget here
+//     saturated severity and nagged constantly (the prior denominator bug).
 //   - `pressure`  = current usage% / execute-threshold%.
 // Either factor low ⇒ quiet, so a disciplined agent and an early-exploring
 // agent are both spared; only "lots of reclaimable space AND near compaction"
diff --git a/packages/plugin/src/hooks/magic-context/event-handler.ts b/packages/plugin/src/hooks/magic-context/event-handler.ts
@@ -131,6 +131,13 @@ async function deliverChannel2IfPending(deps: EventHandlerDeps, sessionId: strin
         // self-gates on the live-server probe (channel2-delivery.ts), so plain
         // TUI still 404s out; it no-ops unless a `pending` intent exists.
         const baseline = deps.channel1StateBySession?.get(sessionId);
+        // If the agent already called ctx_reduce since the last transform refreshed
+        // the baseline, the tailToolTokens/turnToolTokens here are STALE-HIGH (they
+        // predate the reduction) — delivering now would nudge the agent to drop
+        // output it just dropped. Skip until the next transform recomputes a fresh
+        // baseline. Parity with Pi's maybeDeliverChannel2Pi (reducedSinceRefresh
+        // guard). The pending intent stays armed for that fresh re-evaluation.
+        if (baseline?.reducedSinceRefresh) return;
         await maybeDeliverChannel2(sessionId, {
             db: deps.db,
             serverUrl: deps.serverUrl,
