diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 39de1c99..010d56e8 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -313,3 +313,9 @@ E2E tests (`E2E.md`) should be run before releases or after major refactors. `sessionLifecycle.ts` persists a publication lease atomically with each new request target before SDK launch. The lease survives physical SDK writer shutdown and commit until the synchronous durable mapping CAS succeeds, or the request abandons its target. Failed publication restores the lease. Collectors in other processes cannot depend on a proxy instance's private request pins, so they consult these durable leases as well as durable mappings. Publication leases use the existing unarmed active-lease representation with `purpose: "publication"`. Older collectors also retain them while the owner process is alive; exact process-incarnation death permits recovery. They do not count as exclusive SDK writers, and abandoning publication never removes an actual writer lease. Published transcripts are retained by their durable mappings and become collectible after eviction. + +## Lineage hash encoding + +`session/lineage.ts` hashes structured v2 records with separate history, message and block domains. Records preserve roles, block and message boundaries, tool call identity/arguments, and result identity/error status. JSON object keys are canonicalized; plain text and a single text block remain equivalent, and opaque thinking/cache hints remain excluded. Display-oriented `normalizeContent` is not a lineage proof. + +Existing v1 digests cannot establish a v2 prefix. Their next request on an upgraded proxy safely replays the full supplied history and publishes v2 hashes; subsequent requests on upgraded proxies resume normally. Alternating between old and new proxy versions can repeat this replay cost until all participating proxies are upgraded. This migration relies on complete fresh replay, including completed tool calls/results and media. Stored transcript files are never rewritten to migrate hashes. diff --git a/E2E.md b/E2E.md index b420dea8..89dd46b3 100644 --- a/E2E.md +++ b/E2E.md @@ -3685,3 +3685,9 @@ unchanged through the matrix. Run `bun scripts/e2e-publication-lifetime.mjs` and again with `--stream` after lifecycle or publication changes. This gate uses real Claude Max queries and two concurrent HTTP conversations, each with a fresh and resumed turn. A timing hook pauses each request after its real SDK writer lease is released, promotes its request pin as the owning proxy would, and runs a separate collector process before publication. The collector uses zero grace periods and the supported SDK deleter, exercising the destructive race in an isolated session store and disposable project. Require four successful competing sweeps with no deletions, correct fixture identifiers in both answers, valid response envelopes, durable mappings, and unchanged source transcripts inspected through `getSessionMessages`. To verify rolling upgrades, set `E2E_COLLECTOR_MODULE` to the absolute `src/proxy/sessionLifecycle.ts` path in the previous checkout and repeat. The SDK itself is not mocked. + +## Lineage hash integrity and upgrade + +Run `bun scripts/e2e-lineage-hash-migration.mjs` and again with `--stream`; repeat with `E2E_MODEL=sonnet`. The gate establishes a real client tool loop whose result comes from a disposable fixture file, then changes only `is_error` in its supplied history. Require fresh replay with the revised metadata, a FAILED answer instead of the original SUCCEEDED answer, and unchanged source history. + +A second case installs a persisted mapping containing the legacy digest format for that real source. Require one complete replay with call identity/arguments and result data, then an ordinary continuation with the correct answer. All inspection uses supported SDK `getSessionMessages`. Run the E41 sequential/parallel and undo-gap controls too when changing the encoding. diff --git a/scripts/e2e-lineage-hash-migration.mjs b/scripts/e2e-lineage-hash-migration.mjs new file mode 100644 index 00000000..4031a1df --- /dev/null +++ b/scripts/e2e-lineage-hash-migration.mjs @@ -0,0 +1,119 @@ +#!/usr/bin/env bun +/** Real tool-result metadata changes and migration from pre-v2 lineage hashes. */ +import assert from "node:assert/strict" +import { createHash, randomUUID } from "node:crypto" +import { mkdtempSync, readFileSync, realpathSync, writeFileSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { getSessionMessages } from "@anthropic-ai/claude-agent-sdk" + +const stream = process.argv.includes("--stream") +const model = process.env.E2E_MODEL ?? "claude-haiku-4-5-20251001" +const root = realpathSync(mkdtempSync(join(tmpdir(), "meridian-hash-migration-"))) +for (const key of Object.keys(process.env)) { + if (key.startsWith("MERIDIAN_") || key.startsWith("CLAUDE_PROXY_")) delete process.env[key] +} +Object.assign(process.env, { MERIDIAN_CONFIG_DIR: join(root, "config"), MERIDIAN_SESSION_DIR: join(root, "sessions"), + MERIDIAN_WORKDIR: root, MERIDIAN_TELEMETRY_PERSIST: "0", MERIDIAN_PASSTHROUGH: "1" }) +const { startProxyServer } = await import("../src/proxy/server.ts") +const { storeSession } = await import("../src/proxy/session/cache.ts") +const { storeSharedSession, readSessionStoreSnapshot } = await import("../src/proxy/sessionStore.ts") +const { normalizeContent } = await import("../src/proxy/messages.ts") +const { telemetryStore } = await import("../src/telemetry/index.ts") +const marker = `fixture_${randomUUID()}` +const fixturePath = join(root, "fixture.json") +writeFileSync(fixturePath, JSON.stringify({ record: marker })) +const tools = [{ name: "get_fixture", description: "Read the JavaScript test fixture record.", + input_schema: { type: "object", properties: {}, additionalProperties: false } }] +const firstMessage = { role: "user", content: "Call get_fixture once. Then report its record identifier and execution status: SUCCEEDED for is_error=false (or absent), FAILED for is_error=true. Judge status from the tool result metadata, not the payload. Do not call it again after receiving its result." } +const instance = await startProxyServer({ port: 0, host: "127.0.0.1", silent: true }) +const address = instance.server.address() +assert(address && typeof address === "object") + +async function request(key, messages) { + const response = await fetch(`http://127.0.0.1:${address.port}/v1/messages`, { + method: "POST", headers: { "content-type": "application/json", "x-opencode-session": key }, + body: JSON.stringify({ model, max_tokens: 256, stream, tools, messages }), signal: AbortSignal.timeout(90_000), + }) + const raw = await response.text() + assert.equal(response.status, 200, raw) + if (!stream) return JSON.parse(raw).content + const events = raw.split("\n").filter(line => line.startsWith("data:")).map(line => JSON.parse(line.slice(5))) + assert(!events.some(event => event.type === "error"), raw) + assert.equal(events.filter(event => event.type === "message_stop").length, 1, raw) + const blocks = [] + for (const event of events) { + if (event.type === "content_block_start") blocks[event.index] = { ...event.content_block, json: "" } + if (event.delta?.type === "text_delta") blocks[event.index].text = (blocks[event.index].text ?? "") + event.delta.text + if (event.delta?.type === "input_json_delta") blocks[event.index].json += event.delta.partial_json + } + return blocks.filter(Boolean).map(({ json, ...block }) => block.type === "tool_use" + ? { ...block, input: json ? JSON.parse(json) : block.input } : block) +} +const answerText = blocks => blocks.filter(block => block.type === "text").map(block => block.text).join("") +const inputText = row => typeof row?.message?.content === "string" ? row.message.content + : (row?.message?.content ?? []).filter(block => block.type === "text").map(block => block.text).join("\n") +const digest = value => createHash("sha256").update(value).digest("hex").slice(0, 32) + +try { + const sourceKey = `hash-source-${randomUUID()}` + const first = await request(sourceKey, [firstMessage]) + const calls = first.filter(block => block.type === "tool_use") + assert.equal(calls.length, 1, JSON.stringify(first)) + assert.equal(calls[0].name, "get_fixture") + // The client executes its real fixture read and returns the real file bytes. + const result = { type: "tool_result", tool_use_id: calls[0].id, content: readFileSync(fixturePath, "utf8"), is_error: false } + const history = [firstMessage, { role: "assistant", content: first }, { role: "user", content: [result] }] + const initialAnswer = await request(sourceKey, history) + assert(!initialAnswer.some(block => block.type === "tool_use"), JSON.stringify(initialAnswer)) + assert(answerText(initialAnswer).includes(marker) && answerText(initialAnswer).includes("SUCCEEDED"), answerText(initialAnswer)) + const source = readSessionStoreSnapshot()[sourceKey] + assert(source?.claudeSessionId && source.currentTranscript, "source mapping was not published") + const sourceRows = await getSessionMessages(source.claudeSessionId, { dir: root }) + assert(JSON.stringify(sourceRows).includes(marker), "real tool payload is absent from source") + const failures = [] + + for (const mode of ["changed-error-status", "legacy-hashes"]) { + const key = `${mode}-${randomUUID()}` + if (mode === "legacy-hashes") { + const strings = history.map(message => `${message.role}:${normalizeContent(message.content)}`) + assert(storeSharedSession(key, source.claudeSessionId, history.length, digest(strings.join("\n")), + strings.map(digest), source.sdkMessageUuids, undefined, + history.map(message => (Array.isArray(message.content) ? message.content : [message.content]) + .filter(block => !["thinking", "redacted_thinking"].includes(block?.type)) + .map(block => digest(normalizeContent([block])))), + source.passthroughToolCallAssistantUuid, source.passthroughToolCallIds, source.currentTranscript)) + } else { + assert(storeSession(key, history, source.claudeSessionId, root, source.sdkMessageUuids, undefined, + source.passthroughToolCallAssistantUuid, source.passthroughToolCallIds, source.currentTranscript)) + } + const changed = mode === "changed-error-status" + const messages = [history[0], history[1], { role: "user", content: [{ ...result, is_error: changed }] }, + { role: "assistant", content: initialAnswer }, + { role: "user", content: "Re-evaluate the supplied get_fixture result's is_error flag. Report FAILED if true, SUCCEEDED if false, and the record identifier. Use the supplied result; do not read the fixture again." }] + const response = await request(key, messages) + const answer = answerText(response) + const lineage = telemetryStore.getRecent({ limit: 1 })[0]?.lineageType + const active = readSessionStoreSnapshot()[key] + assert(active?.claudeSessionId, "fresh replay mapping missing") + const rows = await getSessionMessages(active.claudeSessionId, { dir: root }) + const firstInput = inputText(rows.find(row => row.type === "user")) + const completeInput = firstInput.includes(calls[0].id) && firstInput.includes(JSON.stringify(calls[0].input)) + && firstInput.includes(marker) && firstInput.includes(`"is_error":${changed}`) + const expectedStatus = changed ? "FAILED" : "SUCCEEDED" + console.log(JSON.stringify({ mode, stream, model, lineage, completeInput, answer })) + if (lineage !== "new" || !completeInput || !answer.includes(marker) || !answer.includes(expectedStatus) + || response.some(block => block.type === "tool_use")) failures.push(`${mode}: stale history or incomplete replay`) + + const continuation = await request(key, [...messages, { role: "assistant", content: response }, + { role: "user", content: "Repeat that execution status and record identifier without calling any tool." }]) + const nextLineage = telemetryStore.getRecent({ limit: 1 })[0]?.lineageType + if (nextLineage !== "continuation" || !answerText(continuation).includes(marker) + || !answerText(continuation).includes(expectedStatus) || continuation.some(block => block.type === "tool_use")) { + failures.push(`${mode}: follow-up did not safely resume the migrated history`) + } + assert.deepEqual(await getSessionMessages(source.claudeSessionId, { dir: root }), sourceRows, "source transcript changed") + } + assert.deepEqual(failures, [], "lineage hash integrity or migration failed") + console.log(`PASS: changed result status and legacy hashes replay completely, then resume (stream=${stream}, model=${model})`) +} finally { await instance.close() } diff --git a/src/__tests__/lineage-hash-domains.test.ts b/src/__tests__/lineage-hash-domains.test.ts new file mode 100644 index 00000000..c47de82c --- /dev/null +++ b/src/__tests__/lineage-hash-domains.test.ts @@ -0,0 +1,97 @@ +import { describe, expect, it } from "bun:test" +import { createHash } from "node:crypto" +import { normalizeContent } from "../proxy/messages" +import { + computeLineageHash, computeMessageHashes, computeMessageBlockHashes, + hashMessage, verifyLineage, type SessionState, +} from "../proxy/session/lineage" + +type Message = { role: string; content: unknown } +const text = (value: string) => ({ type: "text", text: value }) +const result = (id: string, content: unknown, is_error = false) => ({ type: "tool_result", tool_use_id: id, content, is_error }) +function session(messages: Message[]): SessionState { + return { + claudeSessionId: "source", lastAccess: 0, messageCount: messages.length, + lineageHash: computeLineageHash(messages), messageHashes: computeMessageHashes(messages), + messageBlockHashes: computeMessageBlockHashes(messages), + } +} + +describe("lineage hash domains", () => { + for (const [label, left, right] of [ + ["text versus tool result", text("tool_result:call_1:ok"), result("call_1", "ok")], + ["text versus tool use", text('tool_use:call_1:read:{}'), { type: "tool_use", id: "call_1", name: "read", input: {} }], + ["tool result delimiter", result("call_1:a", "b"), result("call_1", "a:b")], + ["tool result error status", result("call_1", "ok", false), result("call_1", "ok", true)], + ["nested result text versus block array", result("call_1", '[{"type":"text","text":"ok"}]'), result("call_1", [text("ok")])], + ] as const) { + it(`separates ${label} at every hash level`, () => { + const a = { role: "user", content: [left] } + const b = { role: "user", content: [right] } + expect(computeMessageBlockHashes([a])).not.toEqual(computeMessageBlockHashes([b])) + expect(hashMessage(a)).not.toBe(hashMessage(b)) + expect(computeLineageHash([a])).not.toBe(computeLineageHash([b])) + expect(verifyLineage(session([a]), [b, { role: "assistant", content: "next" }]).type).toBe("diverged") + }) + } + + it("does not let content inject message boundaries into the aggregate hash", () => { + const a = [{ role: "user", content: "one\nassistant:two" }, { role: "assistant", content: "three" }] + const b = [{ role: "user", content: "one" }, { role: "assistant", content: "two\nassistant:three" }] + expect(computeLineageHash(a)).not.toBe(computeLineageHash(b)) + expect(verifyLineage(session(a), [...b, { role: "user", content: "next" }]).type).toBe("diverged") + }) + + it("keeps block boundaries distinct from a newline in text", () => { + expect(hashMessage({ role: "user", content: [text("one"), text("two")] })) + .not.toBe(hashMessage({ role: "user", content: [text("one\ntwo")] })) + }) + + it("does not reinterpret a cached assistant as a user during a block append", () => { + const a = [{ role: "assistant", content: [text("old")] }] + const b = [{ role: "user", content: [text("old"), result("new", "new result")] }] + expect(verifyLineage(session(a), b).type).toBe("diverged") + }) + + it("preserves plain-string versus text-block equivalence and ignores cache hints", () => { + const a = { role: "user", content: "hello" } + const b = { role: "user", content: [{ ...text("hello"), cache_control: { type: "ephemeral" } }] } + expect(hashMessage(a)).toBe(hashMessage(b)) + expect(computeMessageBlockHashes([a])).toEqual(computeMessageBlockHashes([b])) + }) + + it("continues only from the newly appended parallel result", () => { + const a = [{ role: "user", content: [result("a", "first")] }] + const b = [{ role: "user", content: [result("a", "first"), result("b", "second")] }] + expect(verifyLineage(session(a), b)).toMatchObject({ type: "continuation", resumeFrom: 0, resumeContentFrom: 1 }) + }) + + it("treats JSON key order as immaterial without discarding tool arguments named cache_control", () => { + const call = (input: unknown) => ({ role: "assistant", content: [{ type: "tool_use", id: "a", name: "write", input }] }) + expect(hashMessage(call({ path: "a", nested: { x: 1, y: 2 } }))) + .toBe(hashMessage(call({ nested: { y: 2, x: 1 }, path: "a" }))) + expect(hashMessage(call({ cache_control: "first" }))) + .not.toBe(hashMessage(call({ cache_control: "second" }))) + }) + + it("ignores nested content cache hints but retains image payload identity", () => { + const image = { type: "image", source: { type: "base64", media_type: "image/png", data: "first" } } + const a = { role: "user", content: [result("a", [image])] } + const hinted = { role: "user", content: [result("a", [{ ...image, cache_control: { type: "ephemeral" } }])] } + const changed = { role: "user", content: [result("a", [{ ...image, source: { ...image.source, data: "second" } }])] } + expect(hashMessage(a)).toBe(hashMessage(hinted)) + expect(hashMessage(a)).not.toBe(hashMessage(changed)) + }) + + it("safely replays a legacy cache instead of trusting its ambiguous hashes", () => { + const a = [{ role: "user", content: [text("tool_result:call_1:ok")] }] + const legacyDigest = (value: string) => createHash("sha256").update(value).digest("hex").slice(0, 32) + const old = session(a) + old.lineageHash = legacyDigest(a.map(m => `${m.role}:${normalizeContent(m.content)}`).join("\n")) + old.messageHashes = a.map(m => legacyDigest(`${m.role}:${normalizeContent(m.content)}`)) + old.messageBlockHashes = [[legacyDigest(normalizeContent(a[0]!.content))]] + const b = [{ role: "user", content: [result("call_1", "ok")] }, { role: "assistant", content: "next" }] + expect(verifyLineage(old, b).type).toBe("diverged") + expect(verifyLineage(old, [...a, { role: "assistant", content: "next" }]).type).toBe("diverged") + }) +}) diff --git a/src/__tests__/proxy-lineage-hash-domains.test.ts b/src/__tests__/proxy-lineage-hash-domains.test.ts new file mode 100644 index 00000000..2709d633 --- /dev/null +++ b/src/__tests__/proxy-lineage-hash-domains.test.ts @@ -0,0 +1,93 @@ +import { beforeEach, describe, expect, it } from "bun:test" +import { createHash } from "node:crypto" +import { installSdkMock } from "./sdkMock" +import { installLoggerMock } from "./loggerMock" +import { installMcpToolsMock } from "./mcpToolsMock" +import { assistantMessage, messageStart, textBlockStart, textDelta, blockStop, messageDelta, messageStop, withMockSdkSessionId } from "./helpers" +import { normalizeContent } from "../proxy/messages" + +type Message = { role: string; content: unknown } +type Input = { prompt: string | AsyncIterable<{ message: { content: unknown } }>; options?: { sessionId?: string; resume?: string } } +let captured: { prompt: string; options: Input["options"] }[] = [] +installSdkMock(() => ({ + query: (input: Input) => (async function* () { + let prompt = "" + if (typeof input.prompt === "string") prompt = input.prompt + else for await (const row of input.prompt) prompt += JSON.stringify(row.message.content) + captured.push({ prompt, options: input.options }) + for (const event of [messageStart(), textBlockStart(0), textDelta(0, "ok"), blockStop(0), messageDelta(), messageStop(), assistantMessage([{ type: "text", text: "ok" }])]) { + yield withMockSdkSessionId(event, input.options) + } + })(), + createSdkMcpServer: () => ({ type: "sdk", name: "test", instance: {} }), tool: () => ({}), +}), "proxy-lineage-hash-domains.test.ts") +installLoggerMock(() => ({ claudeLog: () => {}, withClaudeLogContext: (_context: unknown, fn: () => unknown) => fn() })) +installMcpToolsMock(() => ({ createOpencodeMcpServer: () => ({ type: "sdk", name: "opencode", instance: {} }) })) +const { createProxyServer, clearSessionCache } = await import("../proxy/server") +const { storeSession } = await import("../proxy/session/cache") +const { storeSharedSession } = await import("../proxy/sessionStore") +const result = (is_error: boolean) => ({ type: "tool_result", tool_use_id: "call-1", content: "ok", is_error }) +const text = (value: string) => ({ type: "text", text: value }) +const cases: { name: string; stored: Message[]; incoming: Message[]; expected: string }[] = [ + { name: "text reinterpreted as a tool result", stored: [{ role: "user", content: [text("tool_result:call-1:ok")] }], + incoming: [{ role: "user", content: [result(false)] }], expected: '"tool_use_id":"call-1"' }, + { name: "changed result error status", stored: [{ role: "user", content: [result(false)] }], + incoming: [{ role: "user", content: [result(true)] }], expected: '"is_error":true' }, + { name: "injected aggregate message boundaries", stored: [{ role: "user", content: "one\nassistant:two" }, { role: "assistant", content: "three" }], + incoming: [{ role: "user", content: "one" }, { role: "assistant", content: "two\nassistant:three" }], expected: "two\nassistant:three" }, + { name: "assistant rewritten as a user during append", stored: [{ role: "assistant", content: [text("ORIGINAL_ASSISTANT")] }], + incoming: [{ role: "user", content: [text("ORIGINAL_ASSISTANT"), result(false)] }], expected: "ORIGINAL_ASSISTANT" }, +] + +describe("lineage hash integrity through HTTP", () => { + beforeEach(() => { captured = []; clearSessionCache() }) + for (const stream of [false, true]) { + async function post(app: ReturnType["app"], key: string, messages: Message[]) { + const response = await app.fetch(new Request("http://localhost/v1/messages", { + method: "POST", headers: { "content-type": "application/json", "x-opencode-session": key }, + body: JSON.stringify({ model: "haiku", stream, messages }), + })) + const body = await response.text() + expect(response.status, body).toBe(200) + if (stream) { expect(body).toContain("event: message_stop"); expect(body).not.toContain("event: error") } + } + + for (const item of cases) { + it(`replays ${item.name} (stream=${stream})`, async () => { + const key = crypto.randomUUID() + expect(storeSession(key, item.stored, "mock-source")).not.toBe(false) + const { app } = createProxyServer({ silent: true }) + await post(app, key, [...item.incoming, { role: "assistant", content: "prior reply" }, { role: "user", content: "Explain the revised history." }]) + expect(captured).toHaveLength(1) + expect(captured[0]!.options?.resume).toBeUndefined() + expect(captured[0]!.prompt).toContain(item.expected) + }) + } + + it(`migrates a legacy mapping once, preserving tool history before resuming (stream=${stream})`, async () => { + const key = crypto.randomUUID() + const history: Message[] = [ + { role: "user", content: "Look up the fixture" }, + { role: "assistant", content: [{ type: "tool_use", id: "old-call", name: "lookup", input: { fixture: "violet" } }] }, + { role: "user", content: [{ type: "tool_result", tool_use_id: "old-call", content: "EXACT_LEGACY_RESULT" }] }, + ] + const digest = (value: string) => createHash("sha256").update(value).digest("hex").slice(0, 32) + const messageStrings = history.map(message => `${message.role}:${normalizeContent(message.content)}`) + expect(storeSharedSession(key, "legacy-source", history.length, digest(messageStrings.join("\n")), + messageStrings.map(digest), undefined, undefined, + history.map(message => (Array.isArray(message.content) ? message.content : [message.content]) + .map(block => digest(normalizeContent([block])))))).not.toBe(false) + const { app } = createProxyServer({ silent: true }) + const messages = [...history, { role: "assistant", content: "prior answer" }, { role: "user", content: "Explain that fixture." }] + await post(app, key, messages) + expect(captured[0]!.options?.resume).toBeUndefined() + expect(captured[0]!.prompt).toContain('"fixture":"violet"') + expect(captured[0]!.prompt).toContain("EXACT_LEGACY_RESULT") + await post(app, key, [...messages, { role: "assistant", content: "ok" }, { role: "user", content: "Continue." }]) + expect(captured).toHaveLength(2) + expect(captured[1]!.options?.resume).toBe(captured[0]!.options?.sessionId) + expect(captured[1]!.prompt).toContain("Continue.") + expect(captured[1]!.prompt).not.toContain("EXACT_LEGACY_RESULT") + }) + } +}) diff --git a/src/proxy/messages.ts b/src/proxy/messages.ts index d19cf27c..a16a3bbf 100644 --- a/src/proxy/messages.ts +++ b/src/proxy/messages.ts @@ -75,12 +75,13 @@ export const HASH_SERIALIZED_BLOCK_TYPES = new Set([ ]) /** - * Normalize message content to a string for hashing and comparison. + * Legacy content rendering used for adapter compatibility and diagnostics. * Handles both string content and array content (Anthropic content blocks). - * Strips cache_control metadata to ensure hash stability across requests. + * Omits cache_control metadata and opaque thinking blocks. * - * Used only for lineage hashing — see {@link HASH_IGNORED_BLOCK_TYPES}, which - * drops content a display-oriented normalizer would need to keep. + * This representation is ambiguous across block types and boundaries. Never + * use it to prove lineage: session/lineage.ts uses structured domain-separated + * hashes instead. * * NOTE: OpenCode sends content as a string on the first request but as * an array on subsequent ones. This normalizer handles both formats. diff --git a/src/proxy/session/lineage.ts b/src/proxy/session/lineage.ts index 83968fc5..2a32100c 100644 --- a/src/proxy/session/lineage.ts +++ b/src/proxy/session/lineage.ts @@ -134,6 +134,46 @@ export type LineageDivergenceReason = // --- Hashing --- +/** Preserve JSON structure without letting object key order affect identity. */ +function canonicalJson(value: unknown): unknown { + if (Array.isArray(value)) return value.map(canonicalJson) + if (value && typeof value === "object") { + const object = value as Record + return Object.fromEntries(Object.keys(object).sort() + .map(key => [key, canonicalJson(object[key])])) + } + return value +} + +function semanticBlock(value: unknown): unknown { + if (!value || typeof value !== "object" || Array.isArray(value)) return ["value", typeof value, value] + const block = value as Record + switch (block.type) { + case "text": return ["text", block.text] + case "tool_use": return ["tool_use", block.id, block.name, canonicalJson(block.input)] + case "tool_result": return ["tool_result", block.tool_use_id, block.is_error ?? false, semanticContent(block.content)] + default: { + const { cache_control, ...content } = block + return ["block", canonicalJson(content)] + } + } +} + +function semanticContent(content: unknown): unknown[] { + // NOTE: OpenCode changes plain strings to text blocks between requests. + if (typeof content === "string") return [["text", content]] + if (!Array.isArray(content)) return [["value", typeof content, content]] + return hashableContentBlocks(content).map(semanticBlock) +} + +function lineageDigest(domain: string, value: unknown): string { + // Domain-separated structured encoding prevents text, delimiters, roles, + // and content blocks from impersonating one another (#887). Old digests + // cannot prove this representation and safely fall back to a full replay. + return createHash("sha256").update(JSON.stringify(["meridian-lineage-v2", domain, value])) + .digest("hex").slice(0, 32) +} + /** * Compute a lineage hash of an ordered message array. * Used as a fast-path check: if the aggregate hash matches, the messages @@ -141,8 +181,7 @@ export type LineageDivergenceReason = */ export function computeLineageHash(messages: Array<{ role: string; content: any }>): string { if (!messages || messages.length === 0) return "" - const parts = messages.map(m => `${m.role}:${normalizeContent(m.content)}`) - return createHash("sha256").update(parts.join("\n")).digest("hex").slice(0, 32) + return lineageDigest("history", messages.map(m => [m.role, semanticContent(m.content)])) } /** @@ -150,10 +189,7 @@ export function computeLineageHash(messages: Array<{ role: string; content: any * Used to build per-message hash arrays for precise diff-based verification. */ export function hashMessage(message: { role: string; content: any }): string { - return createHash("sha256") - .update(`${message.role}:${normalizeContent(message.content)}`) - .digest("hex") - .slice(0, 32) + return lineageDigest("message", [message.role, semanticContent(message.content)]) } /** A message's shape, for diagnostics that must never carry its content. */ @@ -272,13 +308,6 @@ export function computeMessageHashes(messages: Array<{ role: string; content: an return messages.map(hashMessage) } -function hashNormalizedContent(content: any): string { - return createHash("sha256") - .update(normalizeContent(content)) - .digest("hex") - .slice(0, 32) -} - function hashableContentBlocks(content: any): any[] { if (!Array.isArray(content)) return [content] return content.filter((block: any) => !HASH_IGNORED_BLOCK_TYPES.has(block?.type)) @@ -287,9 +316,8 @@ function hashableContentBlocks(content: any): any[] { /** Compute semantic hashes for each content block in every message. */ export function computeMessageBlockHashes(messages: Array<{ role: string; content: any }>): string[][] { if (!messages || messages.length === 0) return [] - return messages.map((message) => - hashableContentBlocks(message.content).map((block) => - hashNormalizedContent(Array.isArray(message.content) ? [block] : block))) + return messages.map((message) => semanticContent(message.content) + .map(block => lineageDigest("block", [message.role, block]))) } // --- Overlap measurement --- @@ -499,7 +527,7 @@ export function verifyLineage( const storedBlocks = cached.messageBlockHashes[boundary] if (incomingBoundary?.role === "user" && storedBlocks && Array.isArray(incomingBoundary.content)) { const incomingBlocks = hashableContentBlocks(incomingBoundary.content) - const incomingBlockHashes = incomingBlocks.map((block) => hashNormalizedContent([block])) + const incomingBlockHashes = computeMessageBlockHashes([incomingBoundary])[0]! const preservesStoredBlocks = incomingBlocks.length === incomingBoundary.content.length && incomingBlockHashes.length > storedBlocks.length &&