fix: isolate concurrent Pi replays and committed tool caches

E2E.md

@@ -47,6 +47,29 @@ to verify the Claude Code preset remains optional. Meridian
 state is isolated in a temporary directory while the existing SDK auth is kept.
 Also run all four E41 modes to validate normal checkpoint resumes after changes.
 
+### Pi concurrent callers (#870 / #922)
+
+```bash
+bun scripts/e2e-pi-concurrent-replay.mjs
+bun scripts/e2e-pi-concurrent-replay.mjs --stream
+# Install Oh My Pi in a disposable directory; point at the package directory.
+E2E_OMP_PACKAGE=/path/to/node_modules/@oh-my-pi/pi-coding-agent bun scripts/e2e-omp-concurrent-client.mjs
+E2E_OMP_PACKAGE=/path/to/node_modules/@oh-my-pi/pi-coding-agent bun scripts/e2e-omp-concurrent-client.mjs --main-first
+```
+
+The first fixture controls queue admission while using real SDK responses. Both
+main-first and side-first orders must answer from their own request bodies,
+preserve source histories through the supported SDK API, and keep the next main
+turn correct. The last completed branch owns the mapping; following a side call
+can require another fresh replay.
+
+The second fixture uses Oh My Pi's actual session and title-generation APIs
+(validated with 18.0.3), provider serialization and read/write tools. It forces
+the main/title overlap using their real shared session metadata, then verifies
+that the title parses and the client copies a random fixture value through its
+tool loop. It does not mock client or model responses or exercise the terminal
+UI. Both fixtures isolate Meridian state and work only in temporary directories.
+
 ## Test Index
 
 | ID | Section | What It Proves | Verified |

docs/agents.md

@@ -325,6 +325,31 @@ Pi mimics Claude Code's User-Agent, so automatic detection isn't possible. The `
 
 Pi runs in passthrough mode by default — it executes its own tools and Meridian just forwards the `tool_use` blocks. Opt out with `MERIDIAN_PASSTHROUGH=0`.
 
+[Oh My Pi](https://www.npmjs.com/package/@oh-my-pi/pi-coding-agent) (`omp`) is
+built on the same runtime and uses the Pi adapter. Its config lives in
+`~/.omp/agent/models.yml` and takes the same three keys:
+
+```yaml
+providers:
+  anthropic:
+    baseUrl: http://127.0.0.1:3456
+    apiKey: x
+    headers:
+      x-meridian-agent: pi
+```
+
+omp runs its main turn, title generation and mid-turn side questions
+concurrently under one session id. Meridian serializes them and answers a
+conflicting late request by replaying its own history. It does not reject that
+request merely because another caller committed while it waited. Normal
+upstream errors and cancellation still apply. The mapping follows the last
+completed caller: if that is a side call, the next main turn may also need a
+fresh replay. Separate session identities avoid this extra replay cost.
+
+Fresh side requests use their own tool declarations. Tool definitions omitted
+on a continuation can be inherited only from that same published SDK branch;
+a failed side request does not replace its tool cache.
+
 ### Prime Agent
 
 [Prime Agent](https://www.npmjs.com/package/prime-agent) is a fork of Pi with a

scripts/e2e-omp-concurrent-client.mjs

@@ -0,0 +1,142 @@
+#!/usr/bin/env bun
+// Exercises the real Oh My Pi session/title/tool-loop APIs against the real SDK.
+// E2E_OMP_PACKAGE points to an isolated installed @oh-my-pi/pi-coding-agent directory.
+import assert from "node:assert/strict"
+import { randomUUID } from "node:crypto"
+import { mkdirSync, mkdtempSync, readFileSync, realpathSync, writeFileSync } from "node:fs"
+import { tmpdir } from "node:os"
+import { join } from "node:path"
+import { pathToFileURL } from "node:url"
+import { spyOn } from "bun:test"
+import * as sdk from "@anthropic-ai/claude-agent-sdk"
+
+const packageDir = process.env.E2E_OMP_PACKAGE
+assert(packageDir, "Set E2E_OMP_PACKAGE to the installed Oh My Pi package directory")
+const firstKind = process.argv.includes("--main-first") ? "main" : "title"
+const root = realpathSync(mkdtempSync(join(tmpdir(), "meridian-omp-client-")))
+const agentDir = join(root, "omp")
+console.log(JSON.stringify({ root, firstKind }))
+mkdirSync(agentDir)
+for (const key of Object.keys(process.env)) {
+  if (key.startsWith("MERIDIAN_") || key.startsWith("CLAUDE_PROXY_") || key.startsWith("PI_")) delete process.env[key]
+}
+Object.assign(process.env, { MERIDIAN_CONFIG_DIR: join(root, "meridian"), MERIDIAN_SESSION_DIR: join(root, "sessions"),
+  MERIDIAN_WORKDIR: root, MERIDIAN_TELEMETRY_PERSIST: "0", MERIDIAN_PASSTHROUGH: "1", PI_CODING_AGENT_DIR: agentDir })
+process.chdir(root)
+function deferred() {
+  let resolve
+  const promise = new Promise(done => { resolve = done })
+  return { promise, resolve }
+}
+const started = deferred()
+const queued = deferred()
+let firstQuery = true
+let queryCount = 0
+const originalQuery = sdk.query
+const querySpy = spyOn(sdk, "query").mockImplementation(input => {
+  writeFileSync(join(root, `sdk-query-${++queryCount}.json`), JSON.stringify({
+    resume: input.options.resume, systemPrompt: input.options.systemPrompt,
+    allowedTools: input.options.allowedTools, mcpServers: Object.keys(input.options.mcpServers ?? {}),
+    ...(typeof input.prompt === "string" ? { prompt: input.prompt } : {}),
+  }, null, 2))
+  const actual = originalQuery(input)
+  if (!firstQuery) return actual
+  firstQuery = false
+  started.resolve()
+  return new Proxy(actual, { get(target, property) {
+    if (property === Symbol.asyncIterator) return async function* () { await queued.promise; yield* actual }
+    const value = Reflect.get(target, property, target)
+    return typeof value === "function" ? value.bind(target) : value
+  } })
+})
+const { startProxyServer } = await import("../src/proxy/server.ts")
+const { processSessionTurns } = await import("../src/proxy/session/turnCoordinator.ts")
+const { telemetryStore } = await import("../src/telemetry/index.ts")
+const proxy = await startProxyServer({ port: 0, host: "127.0.0.1", silent: true })
+const address = proxy.server.address()
+assert(address && typeof address === "object")
+let arrivals = 0
+const acquire = processSessionTurns.acquire.bind(processSessionTurns)
+const arrivalSpy = spyOn(processSessionTurns, "acquire").mockImplementation((key, signal) => {
+  const pending = acquire(key, signal)
+  if (++arrivals === 2) queued.resolve()
+  return pending
+})
+const requests = []
+const firstPair = []
+let session
+const relay = Bun.serve({ hostname: "127.0.0.1", port: 0, idleTimeout: 120, async fetch(request) {
+  const path = new URL(request.url).pathname
+  if (request.method !== "POST" || path !== "/v1/messages") {
+    return fetch(`http://127.0.0.1:${address.port}${path}`, { method: request.method, headers: request.headers })
+  }
+  const raw = await request.text()
+  const body = JSON.parse(raw)
+  const kind = JSON.stringify(body.system).includes("<title>") ? "title" : "main"
+  const row = { kind, model: body.model, stream: body.stream, session: body.metadata?.user_id,
+    messages: body.messages.length, tools: body.tools?.map(tool => tool.name) ?? [] }
+  requests.push(row)
+  writeFileSync(join(root, `client-request-${requests.length}-${kind}.json`), JSON.stringify(body, null, 2))
+  console.log(JSON.stringify({ request: row }))
+  const forward = async () => {
+    const response = await fetch(`http://127.0.0.1:${address.port}/v1/messages`, {
+      method: "POST", headers: request.headers, body: raw, signal: AbortSignal.timeout(120_000),
+    })
+    const output = await response.text()
+    if (kind === "title") console.log(JSON.stringify({ titleWireResponse: output }))
+    row.status = response.status
+    row.hasStreamError = /"type"\s*:\s*"error"/.test(output)
+    console.log(JSON.stringify({ response: row, ...(response.status !== 200 ? { error: output } : {}) }))
+    return new Response(output, { status: response.status, headers: response.headers })
+  }
+  if (firstPair.length >= 2) return forward()
+  const done = deferred()
+  firstPair.push({ kind, forward, done })
+  if (firstPair.length === 2) {
+    assert.deepEqual(firstPair.map(item => item.kind).sort(), ["main", "title"])
+    assert.equal(requests[0].session, requests[1].session, "Actual client must use the same session identity")
+    const first = firstPair.find(item => item.kind === firstKind)
+    const second = firstPair.find(item => item.kind !== firstKind)
+    void first.forward().then(first.done.resolve)
+    await started.promise
+    void second.forward().then(second.done.resolve)
+  }
+  return done.promise
+} })
+const modelId = "claude-haiku-4-5-20251001"
+writeFileSync(join(agentDir, "models.yml"), `providers:\n  anthropic:\n    baseUrl: http://127.0.0.1:${relay.port}\n    apiKey: x\n    headers:\n      x-meridian-agent: pi\n`)
+const marker = `record_${randomUUID()}`
+writeFileSync(join(root, "fixture.json"), JSON.stringify({ record: marker }))
+const timeout = setTimeout(() => { console.error(JSON.stringify({ timeout: true, root, requests })); process.exit(1) }, 180_000)
+try {
+  const { createAgentSession } = await import(pathToFileURL(join(packageDir, "src/sdk.ts")).href)
+  const { Settings } = await import(pathToFileURL(join(packageDir, "src/config/settings.ts")).href)
+  const settings = await Settings.init({ cwd: root, agentDir, overrides: {
+    "providers.tinyModel": "online", modelRoles: { default: `anthropic/${modelId}`, smol: `anthropic/${modelId}`, tiny: `anthropic/${modelId}` },
+  } })
+  const created = await createAgentSession({ cwd: root, agentDir, settings, modelPattern: `anthropic/${modelId}`,
+    systemPrompt: "You copy JavaScript test fixtures accurately. Follow the user's read and write instructions.",
+    disableExtensionDiscovery: true, skills: [], rules: [], contextFiles: [], promptTemplates: [], slashCommands: [],
+    enableMCP: false, enableLsp: false, enableIrc: false, skipPythonPreflight: true,
+    toolNames: ["read", "write"], restrictToolNames: true,
+  })
+  session = created.session
+  const prompt = "Read fixture.json, then write copied.json with the same record field and value. Do not invent the value. Finally say FIXTURE_COPIED."
+  const [title] = await Promise.all([session.generateTitle(prompt), session.prompt(prompt)])
+  assert(firstPair.length === 2, "Main and title must both reach Meridian")
+  assert(requests.every(row => row.status === 200 && !row.hasStreamError), JSON.stringify(requests))
+  assert(title && title.length > 0, "Oh My Pi must parse the title response")
+  assert.deepEqual(JSON.parse(readFileSync(join(root, "copied.json"), "utf8")), { record: marker })
+  assert(requests.filter(row => row.kind === "main").length >= 3, "Must complete the actual read/write tool loop")
+  assert.equal(telemetryStore.getRecent().filter(row => row.error === "session_turn_conflict").length, 0)
+  console.log(JSON.stringify({ valid: true, firstKind, root, title, requests: requests.length,
+    ompVersion: JSON.parse(readFileSync(join(packageDir, "package.json"), "utf8")).version }))
+} finally {
+  clearTimeout(timeout)
+  queued.resolve()
+  if (session) await session.dispose()
+  relay.stop(true)
+  arrivalSpy.mockRestore()
+  querySpy.mockRestore()
+  await proxy.close()
+}

scripts/e2e-pi-concurrent-replay.mjs

@@ -0,0 +1,155 @@
+#!/usr/bin/env bun
+// Real HTTP + SDK validation. Only scheduling is controlled; model replies are real.
+import assert from "node:assert/strict"
+import { randomUUID } from "node:crypto"
+import { mkdtempSync, realpathSync } from "node:fs"
+import { tmpdir } from "node:os"
+import { join } from "node:path"
+import { spyOn } from "bun:test"
+import * as sdk 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-pi-race-")))
+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" })
+
+function deferred() {
+  let resolve
+  const promise = new Promise(done => { resolve = done })
+  return { promise, resolve }
+}
+async function bounded(promise, label) {
+  let timer
+  try { return await Promise.race([promise, new Promise((_, reject) => {
+    timer = setTimeout(() => reject(new Error(`Timed out: ${label}`)), 90_000)
+  })]) } finally { clearTimeout(timer) }
+}
+let gates
+let nextGate = 0
+const originalQuery = sdk.query
+const querySpy = spyOn(sdk, "query").mockImplementation(input => {
+  const actual = originalQuery(input)
+  const gate = gates?.[nextGate++]
+  if (!gate) return actual
+  gate.options = input.options
+  gate.started.resolve()
+  return new Proxy(actual, { get(target, property) {
+    if (property === Symbol.asyncIterator) return async function* () {
+      await gate.release.promise
+      yield* actual
+    }
+    const value = Reflect.get(target, property, target)
+    return typeof value === "function" ? value.bind(target) : value
+  } })
+})
+const { startProxyServer } = await import("../src/proxy/server.ts")
+const { processSessionTurns } = await import("../src/proxy/session/turnCoordinator.ts")
+const { lookupSharedSession } = await import("../src/proxy/sessionStore.ts")
+const { telemetryStore } = await import("../src/telemetry/index.ts")
+const instance = await startProxyServer({ port: 0, host: "127.0.0.1", silent: true })
+const address = instance.server.address()
+assert(address && typeof address === "object")
+const url = `http://127.0.0.1:${address.port}/v1/messages`
+async function request(key, messages) {
+  const response = await fetch(url, {
+    method: "POST", headers: { "content-type": "application/json", "x-meridian-agent": "pi" },
+    body: JSON.stringify({ model, stream, max_tokens: 160, tools: [], messages,
+      metadata: { user_id: JSON.stringify({ session_id: key }) } }), signal: AbortSignal.timeout(90_000),
+  })
+  const raw = await response.text()
+  if (response.status !== 200) return { status: response.status, raw }
+  if (!stream) return { status: response.status, content: 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 content = []
+  for (const event of events) {
+    if (event.type === "content_block_start") content[event.index] = { ...event.content_block }
+    if (event.delta?.type === "text_delta") content[event.index].text += event.delta.text
+  }
+  return { status: response.status, content: content.filter(Boolean) }
+}
+function checkAnswer(response, expected, excluded) {
+  assert.equal(response.status, 200, response.raw)
+  assert(!response.content.some(block => block.type === "tool_use"), JSON.stringify(response))
+  const answer = response.content.filter(block => block.type === "text").map(block => block.text).join("")
+  for (const field of expected) assert(answer.includes(field), answer)
+  if (excluded) assert(!answer.includes(excluded), answer)
+  return answer
+}
+async function snapshot(key) {
+  const mapping = lookupSharedSession(key)
+  assert(mapping?.claudeSessionId, "Missing durable session")
+  const rows = await sdk.getSessionMessages(mapping.claudeSessionId, { dir: root })
+  assert(rows.length, "SDK history must be readable through the supported API")
+  return { id: mapping.claudeSessionId, rows }
+}
+async function unchanged(source) {
+  assert.deepEqual(await sdk.getSessionMessages(source.id, { dir: root }), source.rows, "Source history changed")
+}
+const failures = []
+try {
+  for (const firstName of ["main", "side"]) {
+    gates = undefined
+    const key = `pi-${randomUUID()}`
+    const base = `base_${randomUUID().slice(0, 8)}`
+    const main = `main_${randomUUID().slice(0, 8)}`
+    const side = `side_${randomUUID().slice(0, 8)}`
+    const opening = [{ role: "user", content: `A JavaScript fixture has field ${base}. For now reply only ACK. No tools.` }]
+    const initial = await request(key, opening)
+    assert.equal(initial.status, 200, initial.raw)
+    const source = await snapshot(key)
+    const prefix = [...opening, { role: "assistant", content: initial.content }]
+    const histories = Object.fromEntries([["main", main], ["side", side]].map(([name, field]) => [name,
+      [...prefix, { role: "user", content: `Also declare field ${field}. List only the fixture field names declared in this conversation, as one JSON array. No tools.` }]]))
+    const secondName = firstName === "main" ? "side" : "main"
+    gates = Array.from({ length: 2 }, () => ({ started: deferred(), release: deferred() }))
+    nextGate = 0
+    const firstP = request(key, histories[firstName])
+    await bounded(gates[0].started.promise, "first SDK query")
+    const queued = deferred()
+    const acquire = processSessionTurns.acquire.bind(processSessionTurns)
+    const arrivalSpy = spyOn(processSessionTurns, "acquire").mockImplementation((turnKey, signal) => {
+      const result = acquire(turnKey, signal)
+      if (turnKey === `session:${key}`) queued.resolve()
+      return result
+    })
+    const secondP = request(key, histories[secondName])
+    try { await bounded(queued.promise, "second request queue admission") } finally { arrivalSpy.mockRestore() }
+    gates[0].release.resolve()
+    const first = await firstP
+    checkAnswer(first, [base, firstName === "main" ? main : side], firstName === "main" ? side : main)
+    await bounded(Promise.race([gates[1].started.promise, secondP]), "second SDK query or refusal")
+    const winner = await snapshot(key)
+    gates[1].release.resolve()
+    const second = await secondP
+    console.log(JSON.stringify({ firstName, stream, model, secondStatus: second.status, secondError: second.raw,
+      secondResume: gates[1].options?.resume ?? null, secondRollback: gates[1].options?.resumeSessionAt ?? null }))
+    if (second.status !== 200) { failures.push(`${firstName}-first refused`); await unchanged(source); continue }
+    assert.equal(gates[1].options?.resume, undefined, "Unmarked race loser must replay its own body")
+    assert.equal(gates[1].options?.resumeSessionAt, undefined)
+    checkAnswer(second, [base, secondName === "main" ? main : side], secondName === "main" ? side : main)
+    const loser = await snapshot(key)
+    await unchanged(winner)
+    await unchanged(source)
+    gates = undefined
+    const mainAnswer = firstName === "main" ? first : second
+    const followup = await request(key, [...histories.main, { role: "assistant", content: mainAnswer.content },
+      { role: "user", content: "List those same fixture field names again as one JSON array. No tools." }])
+    const answer = checkAnswer(followup, [base, main], side)
+    await unchanged(winner)
+    await unchanged(loser)
+    await unchanged(source)
+    console.log(JSON.stringify({ firstName, stream, model, valid: true, answer,
+      followupLineage: telemetryStore.getRecent({ limit: 1 })[0]?.lineageType }))
+  }
+  assert.deepEqual(failures, [], "Pi concurrency replay failed")
+} finally {
+  for (const gate of gates ?? []) gate.release.resolve()
+  querySpy.mockRestore()
+  await instance.close()
+}