fix: separate lineage hash domains and preserve result semantics

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.

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.

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() }

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")
+  })
+})