diff --git a/docs/superpowers/plans/2026-08-22-dcp-reliability-troubleshooting/phase-05-persistence-lifecycle.md b/docs/superpowers/plans/2026-08-22-dcp-reliability-troubleshooting/phase-05-persistence-lifecycle.md index bcc4286..aea99db 100644 --- a/docs/superpowers/plans/2026-08-22-dcp-reliability-troubleshooting/phase-05-persistence-lifecycle.md +++ b/docs/superpowers/plans/2026-08-22-dcp-reliability-troubleshooting/phase-05-persistence-lifecycle.md @@ -15,7 +15,9 @@ - Keep `DcpSnapshotV1`, parser behavior, and serialized snapshot content unchanged. - Exclude both `messageIds.byRawId` and `messageIds.nextRefIndex` from the semantic fingerprint. - Preserve forced recovery writes and explicit command/compression/compaction writes. -- Accept the projection only if all lifecycle reconstruction tests pass. +- Use the current Pi 0.84.2 lifecycle contract for the gate: `session_tree` changes the active branch, `session_compact` fires after a successful compaction, and `agent_settled` fires only after retries, compaction, and queued follow-ups are finished. Keep the historical v0.83.0 corpus numbers separate from current lifecycle behavior. +- Treat synthetic message refs as a reconstruction contract. OpenCode DCP is a reference for using host-stable message IDs; it does not justify removing Pi's synthetic-ID proof. +- Accept the projection only if the complete lifecycle reconstruction matrix passes. - Do not combine the projection and `agent_settled` fallback; select one design from evidence. --- @@ -23,10 +25,12 @@ ### Task 1: Define the semantic fingerprint contract **Files:** + - Modify: `tests/persistence.test.ts` - Modify: `src/state/persistence.ts` **Interfaces:** + - Consumes: `serializeDcpSnapshot(state, "owner")`. - Produces: `durableStateFingerprint(state): string | undefined` that ignores only `messageIds`. @@ -58,15 +62,33 @@ Add: ```typescript it("changes the durable fingerprint for semantic mutations", () => { - const mutations: Array<(state: ReturnType) => void> = [ - (state) => { state.manualMode = "active"; }, - (state) => { state.compressPermission = "deny"; }, - (state) => { state.stats.totalPruneTokens = 1; }, - (state) => { state.lastCompaction = 1; }, - (state) => { state.prune.tools.set("call", 1); }, - (state) => { state.prune.messages.nextBlockId = 2; }, - (state) => { state.prune.messages.nextRunId = 2; }, - (state) => { state.nudges.turnAnchors.add("user:1:0"); }, + const mutations: Array< + (state: ReturnType) => void + > = [ + (state) => { + state.manualMode = "active"; + }, + (state) => { + state.compressPermission = "deny"; + }, + (state) => { + state.stats.totalPruneTokens = 1; + }, + (state) => { + state.lastCompaction = 1; + }, + (state) => { + state.prune.tools.set("call", 1); + }, + (state) => { + state.prune.messages.nextBlockId = 2; + }, + (state) => { + state.prune.messages.nextRunId = 2; + }, + (state) => { + state.nudges.turnAnchors.add("user:1:0"); + }, ]; for (const mutate of mutations) { @@ -96,7 +118,9 @@ Expected: the message-ID exclusion test FAILS; semantic mutation test PASSES. Replace `durableStateFingerprint()` in `src/state/persistence.ts` with: ```typescript -export function durableStateFingerprint(state: SessionState): string | undefined { +export function durableStateFingerprint( + state: SessionState, +): string | undefined { const snapshot = serializeDcpSnapshot(state, "owner"); if (!snapshot) return undefined; const { messageIds: _messageIds, ...durable } = snapshot; @@ -139,18 +163,25 @@ Expected: PASS. ### Task 2: Prove deterministic reconstruction without ID-only checkpoints **Files:** + - Modify: `tests/stable-ids.test.ts` +- Modify: `tests/index.test.ts` +- Modify: `tests/pipeline.test.ts` **Interfaces:** + - Consumes: `serializeDcpSnapshot()`, `restoreDcpSnapshot()`, and `assignMessageRefs()`. -- Produces: lifecycle evidence required to accept or reject the projection. +- Produces: lifecycle evidence required to accept or reject the projection, including Pi branch/compaction events and compression-block boundary rebuilding. - [ ] **Step 1: Add persistence imports and a message helper** Add imports: ```typescript -import { restoreDcpSnapshot, serializeDcpSnapshot } from "../src/state/persistence.ts"; +import { + restoreDcpSnapshot, + serializeDcpSnapshot, +} from "../src/state/persistence.ts"; ``` Use existing inline messages or existing `makeUserMessage`/`makeAssistantMessage`; do not create a second production helper. @@ -166,8 +197,16 @@ it("reconstructs the same refs after ID-only growth was not checkpointed", () => const snapshot = serializeDcpSnapshot(baseline); if (!snapshot) throw new Error("expected snapshot"); const messages: AgentMessage[] = [ - { role: "user", content: [{ type: "text", text: "one" }], timestamp: 1 } as AgentMessage, - { role: "user", content: [{ type: "text", text: "two" }], timestamp: 2 } as AgentMessage, + { + role: "user", + content: [{ type: "text", text: "one" }], + timestamp: 1, + } as AgentMessage, + { + role: "user", + content: [{ type: "text", text: "two" }], + timestamp: 2, + } as AgentMessage, ]; assignMessageRefs(baseline, messages); @@ -224,10 +263,12 @@ it("reconstructs stable refs independently on sibling branches", () => { restoreDcpSnapshot(snapshot, returnedA, "owner"); assignMessageRefs(returnedA, [prefix, branchA]); - expect([...returnedA.messageIds.byIndex.values()]).toEqual( - [...firstA.messageIds.byIndex.values()], + expect([...returnedA.messageIds.byIndex.values()]).toEqual([ + ...firstA.messageIds.byIndex.values(), + ]); + expect(siblingB.messageIds.byIndex.get(0)).toBe( + firstA.messageIds.byIndex.get(0), ); - expect(siblingB.messageIds.byIndex.get(0)).toBe(firstA.messageIds.byIndex.get(0)); }); ``` @@ -239,8 +280,16 @@ Add: it("uses the semantic compaction checkpoint to preserve retained refs", () => { const state = createSessionState(); state.sessionId = "owner"; - const old = { role: "user", content: [{ type: "text", text: "old" }], timestamp: 1 } as AgentMessage; - const retained = { role: "user", content: [{ type: "text", text: "retained" }], timestamp: 2 } as AgentMessage; + const old = { + role: "user", + content: [{ type: "text", text: "old" }], + timestamp: 1, + } as AgentMessage; + const retained = { + role: "user", + content: [{ type: "text", text: "retained" }], + timestamp: 2, + } as AgentMessage; assignMessageRefs(state, [old, retained]); expect(state.messageIds.byIndex.get(1)).toBe("m0002"); state.lastCompaction = 10; @@ -262,22 +311,43 @@ it("uses the semantic compaction checkpoint to preserve retained refs", () => { }); ``` -- [ ] **Step 5: Run the lifecycle gate** +- [ ] **Step 5: Test Pi tree navigation and compaction restart** + +Add extension-level regressions in `tests/index.test.ts` using the existing `createMockApi()`: + +1. Build two valid branch paths from the same semantic checkpoint. Drive `session_tree` from branch A to B and back to A, then run `context` on each path. Assert that the shared prefix and each branch's messages receive the same refs as independent reconstruction, and that ID-only context growth does not append a state entry. +2. Drive `session_compact` on a state containing an old message and a retained message. Run `context` with a `compactionSummary` plus the retained tail, capture the resulting refs, create a fresh extension instance from the persisted checkpoint, and run the same context again. Assert that retained refs and the next allocated ref are unchanged across the restart. + +The test must use the actual registered handlers, not direct calls to `restoreDcpSnapshot()` alone. It should assert that the compaction handler clears runtime indexes while the persisted raw-key map remains available for reconstruction. + +- [ ] **Step 6: Test compression-block boundary reconstruction** + +Add in `tests/pipeline.test.ts`: + +1. Create a compression block with stable `startKey`, `endKey`, `anchorKey`, and `compressToolCallId` boundaries, then serialize that semantic checkpoint. +2. Run a later pipeline pass that adds messages and grows only `messageIds`; do not serialize that ID-only change. +3. Restore the semantic checkpoint into a fresh state and run the pipeline over the later message list. + +Assert that `startIndex`, `endIndex`, `anchorIndex`, `effectiveMessageIndices`, and the resulting pruned messages match the uninterrupted state. This proves that omitted ID-only checkpoints do not invalidate persisted compression boundaries. + +- [ ] **Step 7: Run the lifecycle gate** Run: ```bash -pnpm vitest run tests/stable-ids.test.ts tests/persistence.test.ts +pnpm vitest run tests/stable-ids.test.ts tests/persistence.test.ts tests/index.test.ts tests/pipeline.test.ts ``` -Expected: PASS. If any expected ref changes, stop and execute Task 5 instead of Task 3/4. +Expected: PASS for same-owner resume, fork with reset statistics, both tree directions, compaction followed by restart, and compression-block boundary reconstruction. If any expected ref or block result changes, stop and execute Task 5 instead of Task 3/4. ### Task 3: Stop context writes caused only by growing messages **Files:** + - Modify: `tests/index.test.ts` **Interfaces:** + - Consumes: the semantic fingerprint from Task 1. - Produces: extension-level evidence that growing contexts do not append state without semantic changes. @@ -295,21 +365,40 @@ it("skips state writes when growing context changes only message IDs", async () getSessionId: () => "session", getBranch: () => [] as unknown[], }, - getContextUsage: () => ({ tokens: 20_000, contextWindow: 1_000_000, percent: 2 }), + getContextUsage: () => ({ + tokens: 20_000, + contextWindow: 1_000_000, + percent: 2, + }), hasUI: false, }; const start = handlers.get("session_start")?.[0]; - await (start as (...args: unknown[]) => Promise)({ reason: "new" }, ctx); + await (start as (...args: unknown[]) => Promise)( + { reason: "new" }, + ctx, + ); entries.length = 0; const context = handlers.get("context")?.[0]; - const first = [{ role: "user", content: [{ type: "text", text: "hello" }], timestamp: 1 }]; + const first = [ + { role: "user", content: [{ type: "text", text: "hello" }], timestamp: 1 }, + ]; const second = [ ...first, - { role: "assistant", content: [{ type: "text", text: "hi" }], timestamp: 2 }, + { + role: "assistant", + content: [{ type: "text", text: "hi" }], + timestamp: 2, + }, ]; - await (context as (...args: unknown[]) => Promise)({ messages: first }, ctx); - await (context as (...args: unknown[]) => Promise)({ messages: second }, ctx); + await (context as (...args: unknown[]) => Promise)( + { messages: first }, + ctx, + ); + await (context as (...args: unknown[]) => Promise)( + { messages: second }, + ctx, + ); expect(entries).toHaveLength(0); }); @@ -329,18 +418,29 @@ it("persists one context snapshot when a nudge anchor changes", async () => { getSessionId: () => "session", getBranch: () => [] as unknown[], }, - getContextUsage: () => ({ tokens: 800_000, contextWindow: 1_000_000, percent: 80 }), + getContextUsage: () => ({ + tokens: 800_000, + contextWindow: 1_000_000, + percent: 80, + }), hasUI: false, }; - await (handlers.get("session_start")?.[0] as (...args: unknown[]) => Promise)( - { reason: "new" }, - ctx, - ); + await ( + handlers.get("session_start")?.[0] as (...args: unknown[]) => Promise + )({ reason: "new" }, ctx); entries.length = 0; const event = { - messages: [{ role: "user", content: [{ type: "text", text: "hello" }], timestamp: 1 }], + messages: [ + { + role: "user", + content: [{ type: "text", text: "hello" }], + timestamp: 1, + }, + ], }; - const context = handlers.get("context")?.[0] as (...args: unknown[]) => Promise; + const context = handlers.get("context")?.[0] as ( + ...args: unknown[] + ) => Promise; await context(event, ctx); await context(event, ctx); @@ -362,15 +462,18 @@ pnpm vitest run tests/index.test.ts -t "state writes|context snapshot|persists c ``` Expected: PASS with no `src/index.ts` production change. +The accepted projection path must not add an `agent_settled` persistence handler; that handler belongs only to Task 5's rejected-projection fallback. ### Task 4: Verify the accepted projection design **Files:** + - Verify: `src/state/persistence.ts` - Verify: `src/index.ts` - Verify: persistence and lifecycle tests **Interfaces:** + - Produces: the selected persistence design and evidence for approximately 56 semantic checkpoints in the historical corpus. - [ ] **Step 1: Run all persistence/lifecycle tests** @@ -420,18 +523,20 @@ Expected: PASS; `DcpSnapshotV1` and serialized `messageIds` remain unchanged. - [ ] **Step 4: Commit the accepted projection** ```bash -git add src/state/persistence.ts tests/persistence.test.ts tests/stable-ids.test.ts tests/index.test.ts +git add src/state/persistence.ts tests/persistence.test.ts tests/stable-ids.test.ts tests/index.test.ts tests/pipeline.test.ts git commit -m "fix: skip message-id-only dcp snapshots" ``` ### Task 5: Fallback only if deterministic reconstruction fails **Files:** + - Revert Task 1 production change: `src/state/persistence.ts` - Modify: `src/index.ts` - Modify: `tests/index.test.ts` **Interfaces:** + - Consumes: full existing fingerprint. - Produces: one ordinary checkpoint at `agent_settled` instead of one per context pass. @@ -442,7 +547,9 @@ Execute this task only when Task 2 produces a failing reference-stability case t Restore: ```typescript -export function durableStateFingerprint(state: SessionState): string | undefined { +export function durableStateFingerprint( + state: SessionState, +): string | undefined { return JSON.stringify(serializeDcpSnapshot(state, "owner")); } ``` @@ -451,7 +558,7 @@ Keep the failing lifecycle regression that rejected the projection. - [ ] **Step 2: Move ordinary persistence to `agent_settled`** -Remove the unconditional `persistIfChanged()` call at the end of the `context` handler. Add: +Remove the unconditional `persistIfChanged()` call at the end of the `context` handler. Register the fallback alongside the other top-level lifecycle handlers: ```typescript pi.on("agent_settled", async () => { @@ -463,7 +570,7 @@ Keep command, compression completion, compaction, shutdown, start, and tree pers - [ ] **Step 3: Replace the growing-context test expectation** -After two growing context calls, assert zero entries; invoke the registered `agent_settled` handler and assert one full snapshot containing the latest message IDs. Invoke it again and assert the count remains one. +Assert that `handlers.get("agent_settled")` contains exactly one handler. After two growing context calls, assert zero entries; invoke the registered `agent_settled` handler and assert one full snapshot containing the latest `messageIds.byRawId` and `nextRefIndex`. Invoke it again and assert the count remains one. Also assert that a semantic mutation still persists once before `agent_settled` and does not duplicate at settlement. - [ ] **Step 4: Run the fallback lifecycle suite** @@ -480,6 +587,6 @@ Expected: all PASS, including the lifecycle regression that rejected the project - [ ] **Step 5: Commit the fallback instead of Task 4's commit** ```bash -git add src/index.ts src/state/persistence.ts tests/index.test.ts tests/persistence.test.ts tests/stable-ids.test.ts +git add src/index.ts src/state/persistence.ts tests/index.test.ts tests/persistence.test.ts tests/stable-ids.test.ts tests/pipeline.test.ts git commit -m "fix: checkpoint dcp state after settled agent runs" ``` diff --git a/src/state/persistence.ts b/src/state/persistence.ts index ab21d45..2c3bc89 100644 --- a/src/state/persistence.ts +++ b/src/state/persistence.ts @@ -70,7 +70,10 @@ export function serializeDcpSnapshot( /** Stable comparison key for deciding whether a custom entry must be appended. */ export function durableStateFingerprint(state: SessionState): string | undefined { - return JSON.stringify(serializeDcpSnapshot(state, "owner")); + const snapshot = serializeDcpSnapshot(state, "owner"); + if (!snapshot) return undefined; + const { messageIds: _messageIds, ...durable } = snapshot; + return JSON.stringify(durable); } type SnapshotWarning = (message: string) => void; diff --git a/tests/index.test.ts b/tests/index.test.ts index 6ba6fbb..c2b270f 100644 --- a/tests/index.test.ts +++ b/tests/index.test.ts @@ -2,10 +2,12 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import * as fs from "node:fs"; import * as os from "node:os"; import * as path from "node:path"; +import type { AgentMessage } from "@earendil-works/pi-agent-core"; import createExtension from "../src/index.ts"; import * as subagentResults from "../src/subagents/subagent-results.ts"; import { createSessionState } from "../src/state/state.ts"; -import { serializeDcpSnapshot } from "../src/state/persistence.ts"; +import { restoreDcpSnapshot, serializeDcpSnapshot } from "../src/state/persistence.ts"; +import { assignMessageRefs } from "../src/messages/inject.ts"; import { makeAssistantMessage } from "./helpers.ts"; const agentDir = vi.hoisted(() => `/tmp/dcp-index-test-${Date.now()}-${Math.random()}`); @@ -42,6 +44,16 @@ function createMockApi() { return { api, handlers, entries, commands, tools }; } +function registeredHandler(handlers: Map, event: string): Handler { + const handler = handlers.get(event)?.[0]; + if (!handler) throw new Error(`missing ${event} handler`); + return handler; +} + +function messageRefs(result: { messages: Array<{ content?: Array<{ text?: string }> }> }) { + return result.messages.map((message) => message.content?.[0]?.text?.match(/m\d+/)?.[0]); +} + describe("dcp extension", () => { it("exports a function", () => { expect(typeof createExtension).toBe("function"); @@ -56,6 +68,7 @@ describe("dcp extension", () => { expect(handlers.has("session_compact")).toBe(true); expect(handlers.has("session_shutdown")).toBe(true); expect(handlers.has("context")).toBe(true); + expect(handlers.has("agent_settled")).toBe(false); }); it("registers before_agent_start handler", () => { @@ -401,6 +414,78 @@ describe("dcp extension", () => { }); }); + it("reconstructs refs through same-owner resume and fork lifecycle handlers", async () => { + const prefix = { + role: "user", + content: [{ type: "text" as const, text: "prefix" }], + timestamp: 1, + } as AgentMessage; + const continuation = { + role: "assistant", + content: [{ type: "text" as const, text: "continuation" }], + timestamp: 2, + } as AgentMessage; + const messages = [prefix, continuation]; + const saved = createSessionState(); + saved.sessionId = "parent"; + saved.stats.totalPruneTokens = 99; + assignMessageRefs(saved, [prefix]); + const checkpoint = serializeDcpSnapshot(saved); + if (!checkpoint) throw new Error("expected checkpoint"); + + const runContext = async ( + sessionId: string, + branch: unknown[], + reason: "new" | "resume" | "fork", + ) => { + const mock = createMockApi(); + createExtension(mock.api); + const ctx = { + sessionManager: { + getSessionDir: () => "/tmp/test-session-dir", + getSessionId: () => sessionId, + getBranch: () => branch, + }, + getContextUsage: () => undefined, + hasUI: false, + }; + await ( + registeredHandler(mock.handlers, "session_start") as (...args: unknown[]) => Promise + )({ reason }, ctx); + const result = (await ( + registeredHandler(mock.handlers, "context") as (...args: unknown[]) => Promise + )({ messages }, ctx)) as { + messages: Array<{ content?: Array<{ text?: string }> }>; + }; + return { mock, result }; + }; + + const uninterrupted = await runContext("parent", [], "new"); + const resumed = await runContext( + "parent", + [{ type: "custom", customType: "pi-dcp-state", data: checkpoint }], + "resume", + ); + const forked = await runContext( + "child", + [{ type: "custom", customType: "pi-dcp-state", data: checkpoint }], + "fork", + ); + + expect(messageRefs(resumed.result)).toEqual(messageRefs(uninterrupted.result)); + expect(messageRefs(forked.result)).toEqual(messageRefs(uninterrupted.result)); + expect(resumed.mock.entries).toHaveLength(0); + expect(forked.mock.entries).toHaveLength(1); + expect(forked.mock.entries[0]?.data).toMatchObject({ + ownerSessionId: "child", + stats: { totalPruneTokens: 0 }, + messageIds: { + byRawId: [["user:1:0", "m0001"]], + nextRefIndex: 2, + }, + }); + }); + it("does not append an unchanged snapshot on clean resume", async () => { const { api, handlers, entries } = createMockApi(); const saved = createSessionState(); @@ -511,7 +596,81 @@ describe("dcp extension", () => { expect(entries[1]?.data).toMatchObject({ compressPermission: "deny" }); }); - it("persists one context mutation and skips an unchanged repeated pass", async () => { + it("reconstructs branch refs through session_tree without ID-only snapshots", async () => { + const { api, handlers, entries } = createMockApi(); + const sharedState = createSessionState(); + sharedState.sessionId = "session"; + sharedState.lastCompaction = 1; + const sharedCheckpoint = serializeDcpSnapshot(sharedState); + if (!sharedCheckpoint) throw new Error("expected checkpoint"); + const branchAMessages: AgentMessage[] = [ + { role: "user", content: [{ type: "text", text: "prefix" }], timestamp: 1 } as AgentMessage, + { role: "user", content: [{ type: "text", text: "A" }], timestamp: 2 } as AgentMessage, + ]; + const branchBMessages: AgentMessage[] = [ + ...branchAMessages.slice(0, 1), + { + role: "assistant", + content: [{ type: "text", text: "hidden" }], + timestamp: 2, + } as AgentMessage, + { role: "user", content: [{ type: "text", text: "B" }], timestamp: 3 } as AgentMessage, + ]; + let branch: unknown[] = [ + { type: "custom", customType: "pi-dcp-state", data: sharedCheckpoint }, + { type: "message", message: branchAMessages[1] }, + ]; + const ctx = { + sessionManager: { + getSessionDir: () => "/tmp/test-session-dir", + getSessionId: () => "session", + getBranch: () => branch, + }, + getContextUsage: () => undefined, + hasUI: false, + }; + createExtension(api); + await (registeredHandler(handlers, "session_start") as (...args: unknown[]) => Promise)( + { reason: "resume" }, + ctx, + ); + entries.length = 0; + const context = registeredHandler(handlers, "context") as (...args: unknown[]) => Promise<{ + messages: Array<{ content: Array<{ text?: string }> }>; + }>; + const tree = registeredHandler(handlers, "session_tree") as ( + ...args: unknown[] + ) => Promise; + const expectedA = createSessionState(); + expect(restoreDcpSnapshot(sharedCheckpoint, expectedA, "session")).toBe(true); + assignMessageRefs(expectedA, branchAMessages); + const expectedB = createSessionState(); + expect(restoreDcpSnapshot(sharedCheckpoint, expectedB, "session")).toBe(true); + assignMessageRefs(expectedB, branchBMessages); + + const firstA = await context({ messages: branchAMessages }, ctx); + expect(messageRefs(firstA)).toEqual([...expectedA.messageIds.byIndex.values()]); + branch = [ + { type: "custom", customType: "pi-dcp-state", data: sharedCheckpoint }, + { type: "message", message: branchBMessages[1] }, + { type: "message", message: branchBMessages[2] }, + ]; + await tree({}, ctx); + const siblingB = await context({ messages: branchBMessages }, ctx); + expect(messageRefs(siblingB)).toEqual([...expectedB.messageIds.byIndex.values()]); + branch = [ + { type: "custom", customType: "pi-dcp-state", data: sharedCheckpoint }, + { type: "message", message: branchAMessages[1] }, + ]; + await tree({}, ctx); + const returnedA = await context({ messages: branchAMessages }, ctx); + + expect(messageRefs(returnedA)).toEqual(messageRefs(firstA)); + expect(messageRefs(siblingB)[0]).toBe(messageRefs(firstA)[0]); + expect(entries).toHaveLength(0); + }); + + it("reconstructs retained refs after compaction through registered lifecycle handlers", async () => { const { api, handlers, entries } = createMockApi(); createExtension(api); const ctx = { @@ -520,19 +679,136 @@ describe("dcp extension", () => { getSessionId: () => "session", getBranch: () => [] as unknown[], }, - getContextUsage: () => ({ tokens: 200_000, contextWindow: 1_000_000, percent: 20 }), + getContextUsage: () => undefined, hasUI: false, }; - const start = handlers.get("session_start")?.[0]; - await (start as (...args: unknown[]) => Promise)({ reason: "new" }, ctx); + await (registeredHandler(handlers, "session_start") as (...args: unknown[]) => Promise)( + { reason: "new" }, + ctx, + ); + entries.length = 0; + const context = registeredHandler(handlers, "context") as (...args: unknown[]) => Promise<{ + messages: Array<{ content: Array<{ text?: string }> }>; + }>; + await context( + { + messages: [ + { role: "user", content: [{ type: "text", text: "old" }], timestamp: 1 }, + { role: "user", content: [{ type: "text", text: "retained" }], timestamp: 2 }, + ], + }, + ctx, + ); + entries.length = 0; + await (registeredHandler(handlers, "session_compact") as (...args: unknown[]) => Promise)( + {}, + ctx, + ); + const checkpoint = entries[0]?.data; + expect(checkpoint).toMatchObject({ + messageIds: { + byRawId: [ + ["user:1:0", "m0001"], + ["user:2:0", "m0002"], + ], + nextRefIndex: 3, + }, + }); + + const postCompactionEvent = { + messages: [ + { role: "compactionSummary", summary: "summary", tokensBefore: 100, timestamp: 3 }, + { role: "user", content: [{ type: "text", text: "retained" }], timestamp: 2 }, + { role: "user", content: [{ type: "text", text: "new" }], timestamp: 4 }, + ], + }; + const liveResult = await context(postCompactionEvent, ctx); + expect(entries).toHaveLength(1); + + const restarted = createMockApi(); + createExtension(restarted.api); + const restartCtx = { + ...ctx, + sessionManager: { + ...ctx.sessionManager, + getBranch: () => [{ type: "custom", customType: "pi-dcp-state", data: checkpoint }], + }, + }; + await ( + registeredHandler(restarted.handlers, "session_start") as ( + ...args: unknown[] + ) => Promise + )({ reason: "resume" }, restartCtx); + const result = await ( + registeredHandler(restarted.handlers, "context") as (...args: unknown[]) => Promise<{ + messages: Array<{ content: Array<{ text?: string }> }>; + }> + )(postCompactionEvent, restartCtx); + + expect(messageRefs(result)).toEqual(messageRefs(liveResult)); + expect(result.messages).toEqual(liveResult.messages); + expect(result.messages[1]?.content[0]?.text).toContain("m0002"); + expect(result.messages[2]?.content[0]?.text).toContain("m0004"); + }); + + it("skips state writes when growing context changes only message IDs", async () => { + const { api, handlers, entries } = createMockApi(); + createExtension(api); + const ctx = { + sessionManager: { + getSessionDir: () => "/tmp/test-session-dir", + getSessionId: () => "session", + getBranch: () => [] as unknown[], + }, + getContextUsage: () => ({ tokens: 20_000, contextWindow: 1_000_000, percent: 2 }), + hasUI: false, + }; + await (registeredHandler(handlers, "session_start") as (...args: unknown[]) => Promise)( + { reason: "new" }, + ctx, + ); + entries.length = 0; + const context = registeredHandler(handlers, "context") as ( + ...args: unknown[] + ) => Promise; + const first = [{ role: "user", content: [{ type: "text", text: "hello" }], timestamp: 1 }]; + const second = [ + ...first, + { role: "assistant", content: [{ type: "text", text: "hi" }], timestamp: 2 }, + ]; + + await context({ messages: first }, ctx); + await context({ messages: second }, ctx); + + expect(entries).toHaveLength(0); + }); + + it("persists one context snapshot when a nudge anchor changes", async () => { + const { api, handlers, entries } = createMockApi(); + createExtension(api); + const ctx = { + sessionManager: { + getSessionDir: () => "/tmp/test-session-dir", + getSessionId: () => "session", + getBranch: () => [] as unknown[], + }, + getContextUsage: () => ({ tokens: 800_000, contextWindow: 1_000_000, percent: 80 }), + hasUI: false, + }; + await (registeredHandler(handlers, "session_start") as (...args: unknown[]) => Promise)( + { reason: "new" }, + ctx, + ); entries.length = 0; const event = { messages: [{ role: "user", content: [{ type: "text", text: "hello" }], timestamp: 1 }], }; - const context = handlers.get("context")?.[0]; + const context = registeredHandler(handlers, "context") as ( + ...args: unknown[] + ) => Promise; - await (context as (...args: unknown[]) => Promise)(event, ctx); - await (context as (...args: unknown[]) => Promise)(event, ctx); + await context(event, ctx); + await context(event, ctx); expect(entries).toHaveLength(1); expect(entries[0]?.data).toMatchObject({ diff --git a/tests/persistence.test.ts b/tests/persistence.test.ts index 05d48f9..40580e3 100644 --- a/tests/persistence.test.ts +++ b/tests/persistence.test.ts @@ -251,6 +251,59 @@ describe("persistence", () => { expect(snapshot?.blocks[0]).not.toHaveProperty("active"); }); + it("excludes message-id bookkeeping from the durable fingerprint", () => { + const state = createSessionState(); + state.sessionId = "owner"; + const before = persistence.durableStateFingerprint(state); + + state.messageIds.byRawId.set("user:1:0", "m0001"); + state.messageIds.byRef.set("m0001", "user:1:0"); + state.messageIds.nextRefIndex = 2; + + expect(persistence.durableStateFingerprint(state)).toBe(before); + expect(persistence.serializeDcpSnapshot(state)?.messageIds).toEqual({ + byRawId: [["user:1:0", "m0001"]], + nextRefIndex: 2, + }); + }); + + it("changes the durable fingerprint for semantic mutations", () => { + const mutations: Array<(state: ReturnType) => void> = [ + (state) => { + state.manualMode = "active"; + }, + (state) => { + state.compressPermission = "deny"; + }, + (state) => { + state.stats.totalPruneTokens = 1; + }, + (state) => { + state.lastCompaction = 1; + }, + (state) => { + state.prune.tools.set("call", 1); + }, + (state) => { + state.prune.messages.nextBlockId = 2; + }, + (state) => { + state.prune.messages.nextRunId = 2; + }, + (state) => { + state.nudges.turnAnchors.add("user:1:0"); + }, + ]; + + for (const mutate of mutations) { + const state = createSessionState(); + state.sessionId = "owner"; + const before = persistence.durableStateFingerprint(state); + mutate(state); + expect(persistence.durableStateFingerprint(state)).not.toBe(before); + } + }); + it("restores in place and resets statistics for a forked owner", () => { const saved = createSessionState(); saved.sessionId = "parent"; diff --git a/tests/pipeline.test.ts b/tests/pipeline.test.ts index 3dea6ec..c9ca4f3 100644 --- a/tests/pipeline.test.ts +++ b/tests/pipeline.test.ts @@ -6,6 +6,7 @@ import type { AgentMessage } from "@earendil-works/pi-agent-core"; import type { ContextUsage } from "../src/state/types.ts"; import { applyCompressionState, allocateBlockId, allocateRunId } from "../src/compress/state.ts"; import { countTokens, extractMessageText } from "../src/utils/tokens.ts"; +import { restoreDcpSnapshot, serializeDcpSnapshot } from "../src/state/persistence.ts"; describe("runPipeline", () => { it("returns messages unchanged when no pruning applies", () => { @@ -381,6 +382,58 @@ describe("runPipeline", () => { expect(result.messages.length).toBe(2); }); + it("rebuilds compression boundaries after omitted ID-only checkpoints", () => { + const config = makeDefaultConfig(); + const initial: AgentMessage[] = [ + makeUserMessage("old user", 1), + makeAssistantMessage("old assistant", 2), + { + ...makeAssistantMessage("", 3), + content: [{ type: "toolCall", id: "compress-call", name: "compress", arguments: {} }], + } as AgentMessage, + ]; + const uninterrupted = createSessionState(); + uninterrupted.sessionId = "owner"; + runPipeline(uninterrupted, config, initial, undefined); + applyCompressionState(uninterrupted, { + blockId: allocateBlockId(uninterrupted), + runId: allocateRunId(uninterrupted), + topic: "old context", + mode: "range", + startIndex: 0, + endIndex: 1, + anchorIndex: 0, + compressToolCallId: "compress-call", + startKey: "user:1:0", + endKey: "assistant:2:0", + anchorKey: "user:1:0", + summary: "compressed summary", + summaryTokens: 2, + consumedBlockIds: [], + }); + const checkpoint = serializeDcpSnapshot(uninterrupted); + if (!checkpoint) throw new Error("expected checkpoint"); + const later = [...initial, makeUserMessage("later user", 4)]; + + const uninterruptedResult = runPipeline(uninterrupted, config, later, undefined); + const uninterruptedBlock = uninterrupted.prune.messages.blocksById.get(1); + + const restored = createSessionState(); + expect(restoreDcpSnapshot(checkpoint, restored, "owner")).toBe(true); + const restoredResult = runPipeline(restored, config, later, undefined); + const restoredBlock = restored.prune.messages.blocksById.get(1); + + expect(restoredBlock).toMatchObject({ + startIndex: uninterruptedBlock?.startIndex, + endIndex: uninterruptedBlock?.endIndex, + anchorIndex: uninterruptedBlock?.anchorIndex, + effectiveMessageIndices: uninterruptedBlock?.effectiveMessageIndices, + }); + expect(restoredResult.messages.map(extractMessageText)).toEqual( + uninterruptedResult.messages.map(extractMessageText), + ); + }); + it("is a pure function of its inputs (no Pi mock needed)", () => { const state1 = createSessionState(); const state2 = createSessionState(); diff --git a/tests/stable-ids.test.ts b/tests/stable-ids.test.ts index e67826b..fbdda73 100644 --- a/tests/stable-ids.test.ts +++ b/tests/stable-ids.test.ts @@ -2,6 +2,7 @@ import { describe, it, expect } from "vitest"; import { getMessageKey } from "../src/utils/message-ids.ts"; import { createSessionState } from "../src/state/state.ts"; import { assignMessageRefs } from "../src/messages/inject.ts"; +import { restoreDcpSnapshot, serializeDcpSnapshot } from "../src/state/persistence.ts"; import type { AgentMessage } from "@earendil-works/pi-agent-core"; describe("getMessageKey", () => { @@ -163,6 +164,111 @@ describe("assignMessageRefs (stable)", () => { expect(state.messageIds.byRawId.get("toolResult:call_2")).toBe("m0002"); }); + it("reconstructs the same refs after ID-only growth was not checkpointed", () => { + const baseline = createSessionState(); + baseline.sessionId = "owner"; + const snapshot = serializeDcpSnapshot(baseline); + if (!snapshot) throw new Error("expected snapshot"); + const messages: AgentMessage[] = [ + { + role: "user", + content: [{ type: "text", text: "one" }], + timestamp: 1, + } as AgentMessage, + { + role: "user", + content: [{ type: "text", text: "two" }], + timestamp: 2, + } as AgentMessage, + ]; + + assignMessageRefs(baseline, messages); + const original = [...baseline.messageIds.byIndex.values()]; + + const resumed = createSessionState(); + expect(restoreDcpSnapshot(snapshot, resumed, "owner")).toBe(true); + assignMessageRefs(resumed, messages); + + const forked = createSessionState(); + expect(restoreDcpSnapshot(snapshot, forked, "child")).toBe(true); + assignMessageRefs(forked, messages); + + expect([...resumed.messageIds.byIndex.values()]).toEqual(original); + expect([...forked.messageIds.byIndex.values()]).toEqual(original); + }); + + it("reconstructs stable refs independently on sibling branches", () => { + const baseline = createSessionState(); + baseline.sessionId = "owner"; + const snapshot = serializeDcpSnapshot(baseline); + if (!snapshot) throw new Error("expected snapshot"); + const prefix = { + role: "user", + content: [{ type: "text", text: "prefix" }], + timestamp: 1, + } as AgentMessage; + const branchA = { + ...prefix, + content: [{ type: "text" as const, text: "A" }], + timestamp: 2, + } as AgentMessage; + const branchB = { + ...prefix, + content: [{ type: "text" as const, text: "B" }], + timestamp: 3, + } as AgentMessage; + + const firstA = createSessionState(); + restoreDcpSnapshot(snapshot, firstA, "owner"); + assignMessageRefs(firstA, [prefix, branchA]); + + const siblingB = createSessionState(); + restoreDcpSnapshot(snapshot, siblingB, "owner"); + assignMessageRefs(siblingB, [prefix, branchB]); + + const returnedA = createSessionState(); + restoreDcpSnapshot(snapshot, returnedA, "owner"); + assignMessageRefs(returnedA, [prefix, branchA]); + + expect([...returnedA.messageIds.byIndex.values()]).toEqual([ + ...firstA.messageIds.byIndex.values(), + ]); + expect(siblingB.messageIds.byIndex.get(0)).toBe(firstA.messageIds.byIndex.get(0)); + }); + + it("uses the semantic compaction checkpoint to preserve retained refs", () => { + const state = createSessionState(); + state.sessionId = "owner"; + const old = { + role: "user", + content: [{ type: "text", text: "old" }], + timestamp: 1, + } as AgentMessage; + const retained = { + role: "user", + content: [{ type: "text", text: "retained" }], + timestamp: 2, + } as AgentMessage; + assignMessageRefs(state, [old, retained]); + expect(state.messageIds.byIndex.get(1)).toBe("m0002"); + state.lastCompaction = 10; + const checkpoint = serializeDcpSnapshot(state); + if (!checkpoint) throw new Error("expected checkpoint"); + + const restored = createSessionState(); + restoreDcpSnapshot(checkpoint, restored, "owner"); + const summary = { + role: "compactionSummary", + summary: "summary", + tokensBefore: 100, + timestamp: 3, + } as unknown as AgentMessage; + assignMessageRefs(restored, [summary, retained]); + + expect(restored.messageIds.byIndex.get(1)).toBe("m0002"); + expect(restored.messageIds.byIndex.get(0)).toBe("m0003"); + }); + it("rebuilds byIndex on every call (runtime cache)", () => { const state = createSessionState(); const msg = {