From 58131f0d19c29dc55ec51c814e4c22fac6b9c7fe Mon Sep 17 00:00:00 2001
From: Mate Remias <materemias@gmail.com>
Date: Wed, 2 Sep 2026 12:54:00 +0200
Subject: [PATCH 1/4] fix(session): stop refusing the loser of a declared
 concurrent race (#870)

Oh My Pi drives one conversation from several callers at once: the main
turn, title generation, and side questions asked mid-turn. All of them
carry the same metadata.user_id.session_id, because the protocol has no
per-flow signal to send. Turn coordination serializes them correctly,
but whichever one lost the commit race arrived holding a branch the
mapping had moved past, and was refused with a 400.

That 400 is a hard error on the client side. In omp it triggers a model
fallback, so a turn that could have run switched models and threw away
the prompt cache instead.

Adapters can now declare that their clients share one session key across
concurrent turns, which routes them into the existing
declaresConcurrentFlow escape that fork and subagent sources already
use. The loser is answered by replaying its own history: one
prompt-cache miss instead of a failed turn. Nothing else changes, and
every other adapter still refuses a stale arrival.

Admitting such a turn is not the same as trusting its lineage. A short
side call carries a prefix of the main history, so after the main turn
commits the loser reads as an undo against it, and honouring that would
resume and rewind the session that just committed. A loser admitted by
the protocol declaration is therefore reclassified as diverged before
the undo path sees it. Per-request fork and subagent signals keep their
lineage, since those callers name their own session boundary.

Two tests cover it. The first reproduces the reported shape, two
branches sharing a prefix and differing at the last message, and fails
on main with the request never reaching the SDK. The second commits a
winning turn with SDK message UUIDs while a prefix-carrying side call
waits, and fails without the reclassification, with resume pointing at
the winner's session.
---
 docs/agents.md                                |  18 +++
 .../proxy-concurrency-coordination.test.ts    | 107 +++++++++++++++++-
 src/proxy/adapter.ts                          |  13 +++
 src/proxy/adapters/pi.ts                      |  11 ++
 src/proxy/server.ts                           |  31 ++++-
 src/proxy/session/lineage.ts                  |   1 +
 6 files changed, 175 insertions(+), 6 deletions(-)

diff --git a/docs/agents.md b/docs/agents.md
index b8320941..c4c49668 100644
--- a/docs/agents.md
+++ b/docs/agents.md
@@ -325,6 +325,24 @@ 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 the turn that
+loses the commit race replays from its own history instead of resuming, which
+costs a prompt-cache hit but never fails the turn.
+
 ### Prime Agent
 
 [Prime Agent](https://www.npmjs.com/package/prime-agent) is a fork of Pi with a
diff --git a/src/__tests__/proxy-concurrency-coordination.test.ts b/src/__tests__/proxy-concurrency-coordination.test.ts
index 16a731e3..5d46660c 100644
--- a/src/__tests__/proxy-concurrency-coordination.test.ts
+++ b/src/__tests__/proxy-concurrency-coordination.test.ts
@@ -25,7 +25,7 @@ let activeQueries = 0
 let maxActiveQueries = 0
 let queryCalls = 0
 let controls: AttemptControl[] = []
-let capturedParams: Array<{ options?: { resume?: string; sessionId?: string; env?: Record<string, string> } }> = []
+let capturedParams: Array<{ options?: { resume?: string; resumeSessionAt?: string; sessionId?: string; env?: Record<string, string> } }> = []
 let rateLimitWorkQueries = false
 
 function deferredAttempt(): AttemptControl & { wait: Promise<void>; markStarted: () => void } {
@@ -81,8 +81,9 @@ installMcpToolsMock(() => ({
 const { createProxyServer, clearSessionCache } = await import("../proxy/server")
 const { resetProcessSdkSemaphoreForTests } = await import("../proxy/concurrency")
 const { telemetryStore } = await import("../telemetry")
-const { setSessionStoreDir, storeSharedSession } = await import("../proxy/sessionStore")
+const { setSessionStoreDir, storeSharedSession, readSessionStoreSnapshot } = await import("../proxy/sessionStore")
 const { processSessionTurns } = await import("../proxy/session/turnCoordinator")
+const { computeLineageHash, computeMessageHashes } = await import("../proxy/session/lineage")
 
 function request(
   messages: Array<{ role: string; content: unknown }>,
@@ -103,6 +104,30 @@ function request(
   })
 }
 
+/**
+ * Oh My Pi has no per-flow header: every caller in one conversation, main turn
+ * and side calls alike, stamps the same id in `metadata.user_id`.
+ */
+function piRequest(
+  messages: Array<{ role: string; content: unknown }>,
+  sessionId: string,
+): Request {
+  return new Request("http://localhost/v1/messages", {
+    method: "POST",
+    headers: {
+      "Content-Type": "application/json",
+      "x-meridian-agent": "pi",
+    },
+    body: JSON.stringify({
+      model: "claude-sonnet-4-6",
+      max_tokens: 128,
+      stream: false,
+      messages,
+      metadata: { user_id: JSON.stringify({ session_id: sessionId }) },
+    }),
+  })
+}
+
 async function waitForControl(index: number, timeoutMs = 3000): Promise<AttemptControl> {
   const deadline = Date.now() + timeoutMs
   while (!controls[index]) {
@@ -241,6 +266,84 @@ describe("SDK and Session concurrency coordination", () => {
     expect(conflicts[0]!.upstreamDurationMs).toBe(0)
   })
 
+  it("answers, instead of refusing, the loser of a race an adapter declares (#870)", async () => {
+    // Reproduces the omp report: a side question asked mid-turn and the main
+    // tool loop reach the proxy under one session id, holding branches that
+    // share a prefix and differ at the last message. Serializing them is
+    // right; refusing the loser is not, because the 400 is a hard error that
+    // pushes the client onto a fallback model for a turn it could have run.
+    const app = createProxyServer({ port: 0, host: "127.0.0.1", silent: true }).app
+    const shared = [
+      { role: "user", content: "start the task" },
+      { role: "assistant", content: "ok" },
+    ]
+    const sideQuestion = [...shared, { role: "user", content: "by the way, which branch is this?" }]
+    const mainLoop = [...shared, { role: "user", content: "tool result for step 12" }]
+
+    const sideP = app.fetch(piRequest(sideQuestion, "omp-session"))
+    const sideControl = await waitForControl(0)
+    const mainP = app.fetch(piRequest(mainLoop, "omp-session"))
+
+    sideControl.release()
+    expect((await sideP).status).toBe(200)
+    const mainControl = await waitForControl(1)
+    mainControl.release()
+    expect((await mainP).status).toBe(200)
+    // One session id still means one turn at a time: the second SDK query only
+    // started once the first had finished.
+    expect(maxActiveQueries).toBe(1)
+    expect(queryCalls).toBe(2)
+
+    // The loser carries a branch the winner never had, so it runs fresh rather
+    // than resuming the winner's session and merging two histories.
+    expect(capturedParams[1]?.options?.resume).toBeUndefined()
+    expect(telemetryStore.getRecent().filter(m => m.error === "session_turn_conflict")).toHaveLength(0)
+
+    // The mapping follows the turn that ran last, so the next main-loop request
+    // resumes instead of paying a second fresh replay.
+    const loserSessionId = capturedParams[1]?.options?.sessionId
+    expect(loserSessionId).toMatch(/^[0-9a-f-]{36}$/)
+    const stored = Object.values(readSessionStoreSnapshot()).map(s => s.claudeSessionId)
+    expect(stored).toContain(loserSessionId!)
+  })
+
+  it("replays a declared-flow loser instead of rewinding the turn it lost to (#870)", async () => {
+    // A side call carries a prefix of the main history, so once the main turn
+    // commits the loser reads as an undo against it. Honouring that would roll
+    // the winner's SDK session back to serve a turn that merely arrived late,
+    // so a declared flow is admitted on its own body, never on that lineage.
+    const app = createProxyServer({ port: 0, host: "127.0.0.1", silent: true }).app
+    const sessionId = `omp-undo-${crypto.randomUUID()}`
+    const committed = [
+      { role: "user", content: "start the task" },
+      { role: "assistant", content: "ok" },
+      { role: "user", content: "tool result for step 12" },
+    ]
+    const lease = await processSessionTurns.acquire(`session:${sessionId}`)
+    const sideP = app.fetch(piRequest(committed.slice(0, 2), sessionId))
+
+    // Let handleWithQueue take its coherent arrival snapshot, then commit the
+    // winning turn, UUIDs and all, before the queued side call is granted.
+    await Bun.sleep(20)
+    storeSharedSession(
+      sessionId,
+      "winner-sdk",
+      committed.length,
+      computeLineageHash(committed),
+      computeMessageHashes(committed),
+      ["winner-uuid-1", "winner-uuid-2", "winner-uuid-3"],
+    )
+    lease.markCommitted(sessionId)
+    lease.release()
+
+    const sideControl = await waitForControl(0)
+    sideControl.release()
+    expect((await sideP).status).toBe(200)
+    // Neither resumed nor rolled back: the committed session is left alone.
+    expect(capturedParams[0]?.options?.resume).toBeUndefined()
+    expect(capturedParams[0]?.options?.resumeSessionAt).toBeUndefined()
+  })
+
   it("does not refuse a turn because a DIFFERENT profile advanced the same session id", async () => {
     // One client session id backs an independent conversation per profile, each
     // with its own cache scope. A commit under "work" says nothing about the
diff --git a/src/proxy/adapter.ts b/src/proxy/adapter.ts
index 9e647e42..e06cd679 100644
--- a/src/proxy/adapter.ts
+++ b/src/proxy/adapter.ts
@@ -53,6 +53,19 @@ export interface AgentIdentity {
    */
   getAgentMode?(c: Context, body?: unknown): string | undefined
 
+  /**
+   * True when the client is known to run several turns concurrently under one
+   * session key. Turn coordination still serializes them; what changes is the
+   * loser of a commit race. It is reclassified (`diverged`) and answered from
+   * its own body instead of refused with a 400 (#870). The pattern predates
+   * turn coordination and the upstream API answers both turns.
+   *
+   * Set this only for clients whose protocol makes the sharing unavoidable.
+   * For everyone else, a key that advanced under a waiting request carries a
+   * stale branch, and refusing it is what stops two histories from merging.
+   */
+  readonly runsConcurrentTurnsPerSessionKey?: boolean
+
   /**
    * Optional trusted identity for a visible human turn.
    *
diff --git a/src/proxy/adapters/pi.ts b/src/proxy/adapters/pi.ts
index fbc5f278..4fb8e611 100644
--- a/src/proxy/adapters/pi.ts
+++ b/src/proxy/adapters/pi.ts
@@ -70,6 +70,17 @@ function extractPiCwd(body: any): string | undefined {
 export const piAdapter: AgentAdapter = {
   name: "pi",
 
+  /**
+   * Oh My Pi drives one conversation from several callers at once, the main
+   * turn, title generation, and side questions asked mid-turn, and they all
+   * carry the same `metadata.user_id.session_id`. Whichever one commits first
+   * advances the mapping, so the other arrives holding a branch that no longer
+   * matches and used to be refused with a 400 (#870). Its own history is
+   * complete, so it is answered by fresh replay: one prompt-cache miss instead
+   * of a failed turn that pushes the client onto a fallback model.
+   */
+  runsConcurrentTurnsPerSessionKey: true,
+
   /**
    * Pi itself sends no session header — continuity normally comes from the
    * fingerprint cache. Orchestrators driving the pi runtime (pylon) can opt
diff --git a/src/proxy/server.ts b/src/proxy/server.ts
index 18223a32..da8a561c 100644
--- a/src/proxy/server.ts
+++ b/src/proxy/server.ts
@@ -2150,8 +2150,15 @@ export function createProxyServer(config: Partial<ProxyConfig> = {}): ProxyServe
         // the keyed fork/subagent note above. Serializing them is still worth
         // doing, but a reclassification is their normal cost; refusing them
         // would break flows that worked before turn coordination existed.
+        //
+        // An adapter can declare the same fact for its whole protocol when the
+        // client has no per-flow signal to send: pi carries one session id for
+        // the main turn and its side calls alike, so the loser of that race is
+        // reclassified rather than refused. See runsConcurrentTurnsPerSessionKey.
         const declaresConcurrentFlow =
-          requestSource?.startsWith("fork-") === true || isSubagentRequest
+          requestSource?.startsWith("fork-") === true
+          || isSubagentRequest
+          || adapter.runsConcurrentTurnsPerSessionKey === true
         // NOTE: agent-specific (opencode) — OpenCode begins the tool-result
         // request as soon as the visible checkpoint closes, while Meridian is
         // still draining and committing that checkpoint. Its prompt envelope
@@ -2180,15 +2187,17 @@ export function createProxyServer(config: Partial<ProxyConfig> = {}): ProxyServe
             }
           }
         }
-        if (
+        // The account this turn holds was rewritten while it waited: it lost a
+        // commit race, so the branch its body carries is no longer authoritative.
+        const lostRaceWhileWaiting = Boolean(
           agentSessionId &&
           profileSessionId &&
-          !declaresConcurrentFlow &&
           !advancesDurableCheckpoint &&
           (requestMeta.sessionTurnLease?.advancedWhileWaiting(profileSessionId) || advancedAcrossProcesses) &&
           lineageResult.type !== "continuation" &&
           lineageResult.type !== "compaction"
-        ) {
+        )
+        if (lostRaceWhileWaiting && !declaresConcurrentFlow) {
           const reason = lineageResult.type === "diverged" ? lineageResult.reason : lineageResult.type
           const message = "This session advanced while the request was waiting. Retry with the latest conversation history or use a distinct session ID."
           claudeLog("session.concurrent_conflict", {
@@ -2246,6 +2255,20 @@ export function createProxyServer(config: Partial<ProxyConfig> = {}): ProxyServe
             },
           )
         }
+        // Admitting a protocol-declared flow is not the same as trusting its
+        // lineage. A loser holding a prefix of the committed history reads as an
+        // undo, and honouring that would rewind the session that just committed
+        // to serve a turn which merely arrived late — pi's title generation is
+        // exactly that shape. Replay its own body instead. Per-request fork and
+        // subagent signals keep their lineage: those callers name their own
+        // session boundary, so an undo from them is deliberate.
+        if (
+          lostRaceWhileWaiting &&
+          adapter.runsConcurrentTurnsPerSessionKey === true &&
+          lineageResult.type === "undo"
+        ) {
+          lineageResult = { type: "diverged", reason: "concurrent-race" }
+        }
         if (options.forceFreshPriorityReplay) {
           if (!options.priorityPublication || !agentSessionId || !durableMappingKey) {
             throw new Error("Fresh priority replay requires trusted keyed durable publication")
diff --git a/src/proxy/session/lineage.ts b/src/proxy/session/lineage.ts
index a719928a..600e33da 100644
--- a/src/proxy/session/lineage.ts
+++ b/src/proxy/session/lineage.ts
@@ -131,6 +131,7 @@ export type LineageDivergenceReason =
   | "independent-request"
   | "priority-failback"
   | "missing-session-header"
+  | "concurrent-race"
 
 // --- Hashing ---
 

From c5f8ac8611a152c7d76d76d691bc20162b981cc1 Mon Sep 17 00:00:00 2001
From: Mate Remias <materemias@gmail.com>
Date: Wed, 2 Sep 2026 16:19:56 +0200
Subject: [PATCH 2/4] fix(session): keep a per-request fork signal's undo
 through a lost race

The undo downgrade added for the pi adapter applied to every request that
adapter serves, including the ones that carry their own concurrency
declaration. A fork source or subagent signal names its own session
boundary, so an undo from it is deliberate and honouring it is the point;
only pi's unmarkable side calls need their arrival-order undo replaced by
a replay. Gate the downgrade on the absence of a per-request signal.

Also corrects docs/agents.md: the loser misses the prompt cache, it does
not cost a hit.
---
 docs/agents.md                                |  4 +-
 .../proxy-concurrency-coordination.test.ts    | 40 +++++++++++++++++++
 src/proxy/server.ts                           | 11 +++--
 3 files changed, 49 insertions(+), 6 deletions(-)

diff --git a/docs/agents.md b/docs/agents.md
index c4c49668..b2ba4739 100644
--- a/docs/agents.md
+++ b/docs/agents.md
@@ -340,8 +340,8 @@ providers:
 
 omp runs its main turn, title generation and mid-turn side questions
 concurrently under one session id. Meridian serializes them, and the turn that
-loses the commit race replays from its own history instead of resuming, which
-costs a prompt-cache hit but never fails the turn.
+loses the commit race replays from its own history instead of resuming, so it
+misses the prompt cache for that one turn but never fails.
 
 ### Prime Agent
 
diff --git a/src/__tests__/proxy-concurrency-coordination.test.ts b/src/__tests__/proxy-concurrency-coordination.test.ts
index 5d46660c..aff115df 100644
--- a/src/__tests__/proxy-concurrency-coordination.test.ts
+++ b/src/__tests__/proxy-concurrency-coordination.test.ts
@@ -111,12 +111,14 @@ function request(
 function piRequest(
   messages: Array<{ role: string; content: unknown }>,
   sessionId: string,
+  extraHeaders: Record<string, string> = {},
 ): Request {
   return new Request("http://localhost/v1/messages", {
     method: "POST",
     headers: {
       "Content-Type": "application/json",
       "x-meridian-agent": "pi",
+      ...extraHeaders,
     },
     body: JSON.stringify({
       model: "claude-sonnet-4-6",
@@ -344,6 +346,44 @@ describe("SDK and Session concurrency coordination", () => {
     expect(capturedParams[0]?.options?.resumeSessionAt).toBeUndefined()
   })
 
+  it("keeps a per-request fork signal's undo when it loses the same race (#870)", async () => {
+    // The protocol-level declaration exists because pi cannot mark its own side
+    // calls, so an undo shape there is an accident of arrival order. A fork
+    // source is the opposite: that caller named the boundary itself, so its
+    // rollback is deliberate and must still be honoured after losing a race.
+    const app = createProxyServer({ port: 0, host: "127.0.0.1", silent: true }).app
+    const sessionId = `omp-fork-${crypto.randomUUID()}`
+    const committed = [
+      { role: "user", content: "start the task" },
+      { role: "assistant", content: "ok" },
+      { role: "user", content: "tool result for step 12" },
+    ]
+    const lease = await processSessionTurns.acquire(`session:${sessionId}`)
+    const forkP = app.fetch(piRequest(committed.slice(0, 2), sessionId, {
+      "x-meridian-source": "fork-memory-extract",
+    }))
+
+    // Same shape as the test above: nothing observable fires between the
+    // arrival snapshot and the queue grant, so the window has to be timed.
+    await Bun.sleep(20)
+    storeSharedSession(
+      sessionId,
+      "winner-sdk",
+      committed.length,
+      computeLineageHash(committed),
+      computeMessageHashes(committed),
+      ["winner-uuid-1", "winner-uuid-2", "winner-uuid-3"],
+    )
+    lease.markCommitted(sessionId)
+    lease.release()
+
+    const forkControl = await waitForControl(0)
+    forkControl.release()
+    expect((await forkP).status).toBe(200)
+    expect(capturedParams[0]?.options?.resume).toBe("winner-sdk")
+    expect(capturedParams[0]?.options?.resumeSessionAt).toBe("winner-uuid-2")
+  })
+
   it("does not refuse a turn because a DIFFERENT profile advanced the same session id", async () => {
     // One client session id backs an independent conversation per profile, each
     // with its own cache scope. A commit under "work" says nothing about the
diff --git a/src/proxy/server.ts b/src/proxy/server.ts
index da8a561c..c76e7f19 100644
--- a/src/proxy/server.ts
+++ b/src/proxy/server.ts
@@ -2155,9 +2155,11 @@ export function createProxyServer(config: Partial<ProxyConfig> = {}): ProxyServe
         // client has no per-flow signal to send: pi carries one session id for
         // the main turn and its side calls alike, so the loser of that race is
         // reclassified rather than refused. See runsConcurrentTurnsPerSessionKey.
-        const declaresConcurrentFlow =
+        const declaresPerRequestConcurrentFlow =
           requestSource?.startsWith("fork-") === true
           || isSubagentRequest
+        const declaresConcurrentFlow =
+          declaresPerRequestConcurrentFlow
           || adapter.runsConcurrentTurnsPerSessionKey === true
         // NOTE: agent-specific (opencode) — OpenCode begins the tool-result
         // request as soon as the visible checkpoint closes, while Meridian is
@@ -2258,12 +2260,13 @@ export function createProxyServer(config: Partial<ProxyConfig> = {}): ProxyServe
         // Admitting a protocol-declared flow is not the same as trusting its
         // lineage. A loser holding a prefix of the committed history reads as an
         // undo, and honouring that would rewind the session that just committed
-        // to serve a turn which merely arrived late — pi's title generation is
-        // exactly that shape. Replay its own body instead. Per-request fork and
-        // subagent signals keep their lineage: those callers name their own
+        // to serve a turn which merely arrived late. Pi's title generation is
+        // exactly that shape. Replay its own body instead. A per-request fork
+        // or subagent signal keeps its lineage: those callers name their own
         // session boundary, so an undo from them is deliberate.
         if (
           lostRaceWhileWaiting &&
+          !declaresPerRequestConcurrentFlow &&
           adapter.runsConcurrentTurnsPerSessionKey === true &&
           lineageResult.type === "undo"
         ) {

From 686c7cc09ed5aae8ad3483aad10129ab9d3948cd Mon Sep 17 00:00:00 2001
From: Trevor Walker <rynfar@gmail.com>
Date: Fri, 4 Sep 2026 15:32:21 -0600
Subject: [PATCH 3/4] test: exercise valid concurrent undo boundaries
 deterministically

---
 .../proxy-concurrency-coordination.test.ts    | 42 ++++++++++++++-----
 1 file changed, 32 insertions(+), 10 deletions(-)

diff --git a/src/__tests__/proxy-concurrency-coordination.test.ts b/src/__tests__/proxy-concurrency-coordination.test.ts
index aff115df..c2a51c1f 100644
--- a/src/__tests__/proxy-concurrency-coordination.test.ts
+++ b/src/__tests__/proxy-concurrency-coordination.test.ts
@@ -1,4 +1,4 @@
-import { afterEach, beforeEach, describe, expect, it, mock } from "bun:test"
+import { afterEach, beforeEach, describe, expect, it, mock, spyOn } from "bun:test"
 import { installSdkMock } from "./sdkMock"
 import { installLoggerMock } from "./loggerMock"
 import { installMcpToolsMock } from "./mcpToolsMock"
@@ -83,7 +83,7 @@ const { resetProcessSdkSemaphoreForTests } = await import("../proxy/concurrency"
 const { telemetryStore } = await import("../telemetry")
 const { setSessionStoreDir, storeSharedSession, readSessionStoreSnapshot } = await import("../proxy/sessionStore")
 const { processSessionTurns } = await import("../proxy/session/turnCoordinator")
-const { computeLineageHash, computeMessageHashes } = await import("../proxy/session/lineage")
+const { computeLineageHash, computeMessageHashes, verifyLineage } = await import("../proxy/session/lineage")
 
 function request(
   messages: Array<{ role: string; content: unknown }>,
@@ -130,6 +130,18 @@ function piRequest(
   })
 }
 
+function observeTurnArrival(sessionId: string) {
+  let markArrived = () => {}
+  const arrived = new Promise<void>(resolve => { markArrived = resolve })
+  const acquire = processSessionTurns.acquire.bind(processSessionTurns)
+  const observer = spyOn(processSessionTurns, "acquire").mockImplementation((key, signal) => {
+    const pending = acquire(key, signal)
+    if (key === `session:${sessionId}`) markArrived()
+    return pending
+  })
+  return { arrived, restore: () => observer.mockRestore() }
+}
+
 async function waitForControl(index: number, timeoutMs = 3000): Promise<AttemptControl> {
   const deadline = Date.now() + timeoutMs
   while (!controls[index]) {
@@ -322,11 +334,12 @@ describe("SDK and Session concurrency coordination", () => {
       { role: "user", content: "tool result for step 12" },
     ]
     const lease = await processSessionTurns.acquire(`session:${sessionId}`)
-    const sideP = app.fetch(piRequest(committed.slice(0, 2), sessionId))
+    const incoming = [...committed.slice(0, 2), { role: "user", content: "side question from the earlier turn" }]
+    const arrival = observeTurnArrival(sessionId)
+    const sideP = app.fetch(piRequest(incoming, sessionId))
+    try { await arrival.arrived } finally { arrival.restore() }
 
-    // Let handleWithQueue take its coherent arrival snapshot, then commit the
-    // winning turn, UUIDs and all, before the queued side call is granted.
-    await Bun.sleep(20)
+    // The request has taken its coherent snapshot and joined the real queue.
     storeSharedSession(
       sessionId,
       "winner-sdk",
@@ -335,6 +348,10 @@ describe("SDK and Session concurrency coordination", () => {
       computeMessageHashes(committed),
       ["winner-uuid-1", "winner-uuid-2", "winner-uuid-3"],
     )
+    // Ensure this fixture actually reaches the undo path under current proofs.
+    const winner = readSessionStoreSnapshot()[sessionId]
+    if (!winner?.lineageHash) throw new Error("Expected a verifiable winner mapping")
+    expect(verifyLineage({ ...winner, lineageHash: winner.lineageHash, lastAccess: 0 }, incoming).type).toBe("undo")
     lease.markCommitted(sessionId)
     lease.release()
 
@@ -359,13 +376,14 @@ describe("SDK and Session concurrency coordination", () => {
       { role: "user", content: "tool result for step 12" },
     ]
     const lease = await processSessionTurns.acquire(`session:${sessionId}`)
-    const forkP = app.fetch(piRequest(committed.slice(0, 2), sessionId, {
+    const incoming = [...committed.slice(0, 2), { role: "user", content: "extract memory at this fork boundary" }]
+    const arrival = observeTurnArrival(sessionId)
+    const forkP = app.fetch(piRequest(incoming, sessionId, {
       "x-meridian-source": "fork-memory-extract",
     }))
+    try { await arrival.arrived } finally { arrival.restore() }
 
-    // Same shape as the test above: nothing observable fires between the
-    // arrival snapshot and the queue grant, so the window has to be timed.
-    await Bun.sleep(20)
+    // Same admission boundary as the unmarked side call; no timing sleep.
     storeSharedSession(
       sessionId,
       "winner-sdk",
@@ -374,6 +392,10 @@ describe("SDK and Session concurrency coordination", () => {
       computeMessageHashes(committed),
       ["winner-uuid-1", "winner-uuid-2", "winner-uuid-3"],
     )
+    // Ensure this fixture actually reaches the undo path under current proofs.
+    const winner = readSessionStoreSnapshot()[sessionId]
+    if (!winner?.lineageHash) throw new Error("Expected a verifiable winner mapping")
+    expect(verifyLineage({ ...winner, lineageHash: winner.lineageHash, lastAccess: 0 }, incoming).type).toBe("undo")
     lease.markCommitted(sessionId)
     lease.release()
 

From 554b531134dfb092382383592356c2f6f19714e5 Mon Sep 17 00:00:00 2001
From: Trevor Walker <rynfar@gmail.com>
Date: Fri, 4 Sep 2026 15:49:30 -0600
Subject: [PATCH 4/4] fix: inherit tools only from the committed conversation
 branch

---
 E2E.md                                        |  23 +++
 docs/agents.md                                |  13 +-
 scripts/e2e-omp-concurrent-client.mjs         | 142 ++++++++++++++++
 scripts/e2e-pi-concurrent-replay.mjs          | 155 ++++++++++++++++++
 .../proxy-cross-process-coordination.test.ts  |  41 ++++-
 src/__tests__/proxy-tool-cache.test.ts        |  43 +++++
 .../session-tool-cache-eviction.test.ts       |   6 +-
 src/proxy/adapters/pi.ts                      |   2 +-
 src/proxy/server.ts                           |  31 ++--
 9 files changed, 437 insertions(+), 19 deletions(-)
 create mode 100644 scripts/e2e-omp-concurrent-client.mjs
 create mode 100644 scripts/e2e-pi-concurrent-replay.mjs

diff --git a/E2E.md b/E2E.md
index 32408325..d0db24eb 100644
--- a/E2E.md
+++ b/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 |
diff --git a/docs/agents.md b/docs/agents.md
index b2ba4739..32c8c764 100644
--- a/docs/agents.md
+++ b/docs/agents.md
@@ -339,9 +339,16 @@ providers:
 ```
 
 omp runs its main turn, title generation and mid-turn side questions
-concurrently under one session id. Meridian serializes them, and the turn that
-loses the commit race replays from its own history instead of resuming, so it
-misses the prompt cache for that one turn but never fails.
+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
 
diff --git a/scripts/e2e-omp-concurrent-client.mjs b/scripts/e2e-omp-concurrent-client.mjs
new file mode 100644
index 00000000..d2c16ac2
--- /dev/null
+++ b/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()
+}
diff --git a/scripts/e2e-pi-concurrent-replay.mjs b/scripts/e2e-pi-concurrent-replay.mjs
new file mode 100644
index 00000000..8c9c47e3
--- /dev/null
+++ b/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()
+}
diff --git a/src/__tests__/proxy-cross-process-coordination.test.ts b/src/__tests__/proxy-cross-process-coordination.test.ts
index 6bd403ff..359ecafc 100644
--- a/src/__tests__/proxy-cross-process-coordination.test.ts
+++ b/src/__tests__/proxy-cross-process-coordination.test.ts
@@ -144,7 +144,7 @@ const request = new Request("http://localhost/v1/messages", {
   method: "POST",
   headers: {
     "Content-Type": "application/json",
-    "x-opencode-session": process.env.CLIENT_SESSION_ID,
+    ...JSON.parse(process.env.REQUEST_HEADERS),
   },
   body: process.env.REQUEST_BODY,
 })
@@ -217,13 +217,15 @@ function spawnProxyWorker(
   id: string,
   clientSessionId: string,
   messages: Array<{ role: string, content: unknown }>,
+  options: { adapter?: "pi"; stream?: boolean } = {},
 ): WorkerHandle {
   const releaseFile = join(paths.base, `release-${id}`)
   const requestBody = JSON.stringify({
     model: "claude-sonnet-4-6",
     max_tokens: 128,
-    stream: false,
+    stream: options.stream ?? false,
     messages,
+    ...(options.adapter === "pi" ? { metadata: { user_id: JSON.stringify({ session_id: clientSessionId }) } } : {}),
   })
   const child = Bun.spawn([process.execPath, "-e", workerSource], {
     env: {
@@ -241,6 +243,8 @@ function spawnProxyWorker(
       RELEASE_FILE: releaseFile,
       WORKER_ID: id,
       CLIENT_SESSION_ID: clientSessionId,
+      REQUEST_HEADERS: JSON.stringify(options.adapter === "pi"
+        ? { "x-meridian-agent": "pi" } : { "x-opencode-session": clientSessionId }),
       REQUEST_BODY: requestBody,
     },
     stdout: "ignore",
@@ -312,6 +316,39 @@ function conflictBody(event: WorkerEvent): unknown {
 }
 
 describe("proxy coordination across OS processes", () => {
+  for (const stream of [false, true]) {
+    for (const firstName of ["main", "side"] as const) {
+      test(`Pi keeps both branches valid across processes: ${firstName} first, stream=${stream}`, async () => {
+        const paths = await makePaths()
+        const key = `pi-race-${firstName}-${stream}`
+        const prefix = [{ role: "user", content: "shared fixture" }, { role: "assistant", content: "ok" }]
+        const main = [...prefix, { role: "user", content: "main-only fixture field" }]
+        const side = [...prefix, { role: "user", content: "side-only fixture field" }]
+        const owner = spawnProxyWorker(paths, "pi-owner", key, firstName === "main" ? main : side, { adapter: "pi", stream })
+        const ownerSdk = await waitForEvent(paths.events, owner.id, "sdk-start")
+        const waiter = spawnProxyWorker(paths, "pi-waiter", key, firstName === "main" ? side : main, { adapter: "pi", stream })
+        await waitForEvent(paths.events, waiter.id, "arrival-snapshot")
+        expect((await readEvents(paths.events)).some(row => row.workerId === waiter.id && row.name === "sdk-start")).toBe(false)
+        await release(owner)
+        expect((await result(paths, owner)).status).toBe(200)
+        const waiterSdk = await waitForEvent(paths.events, waiter.id, "sdk-start")
+        expect(waiterSdk.resume).toBeNull()
+        expect(waiterSdk.sdkSessionId).not.toBe(ownerSdk.sdkSessionId)
+        await release(waiter)
+        expect((await result(paths, waiter)).status).toBe(200)
+
+        const followup = spawnProxyWorker(paths, "pi-followup", key, [...main,
+          { role: "assistant", content: "ok" }, { role: "user", content: "continue the main fixture" }], { adapter: "pi", stream })
+        const followupSdk = await waitForEvent(paths.events, followup.id, "sdk-start")
+        // A single key stores the most recent branch. A later main turn either
+        // resumes that main branch or safely replays after a side branch.
+        expect(followupSdk.resume).toBe(firstName === "side" ? waiterSdk.sdkSessionId! : null)
+        await release(followup)
+        expect((await result(paths, followup)).status).toBe(200)
+      }, 20_000)
+    }
+  }
+
   test("serializes one session and rejects a stale arrival after durable advancement", async () => {
     const paths = await makePaths()
     const opening = [{ role: "user", content: "hello" }]
diff --git a/src/__tests__/proxy-tool-cache.test.ts b/src/__tests__/proxy-tool-cache.test.ts
index 22d56d68..82e6dd1e 100644
--- a/src/__tests__/proxy-tool-cache.test.ts
+++ b/src/__tests__/proxy-tool-cache.test.ts
@@ -12,11 +12,13 @@ import { assistantMessage, makeRequest, READ_TOOL, withMockSdkSessionId } from "
 
 let capturedQueryParams: any = null
 let mockMessages: any[] = []
+let mockQueryError: Error | undefined
 
 installSdkMock(() => ({
   query: (params: any) => {
     capturedQueryParams = params
     return (async function* () {
+      if (mockQueryError) throw mockQueryError
       for (const msg of mockMessages) {
         yield withMockSdkSessionId(msg, params.options)
       }
@@ -75,6 +77,7 @@ describe("Session tool cache", () => {
   beforeEach(() => {
     clearSessionCache()
     capturedQueryParams = null
+    mockQueryError = undefined
     mockMessages = [
       assistantMessage([{ type: "text", text: "Done." }]),
     ]
@@ -111,6 +114,46 @@ describe("Session tool cache", () => {
     expect(opts2?.mcpServers).toBeDefined()
   })
 
+  for (const stream of [false, true]) {
+    it(`keeps a fresh side conversation and its continuation free of the main tools, stream=${stream}`, async () => {
+      const app = createTestApp()
+      expect((await post(app, makeRequest({ stream: false, tools: [TOOL_A],
+        messages: [{ role: "user", content: "main task with file access" }],
+      }))).status).toBe(200)
+      expect(capturedQueryParams?.options?.mcpServers?.oc).toBeDefined()
+      const side = [{ role: "user", content: "Give this conversation a title. No tools." }]
+      const response = await post(app, makeRequest({ stream, tools: [], messages: side }))
+      expect(response.status).toBe(200)
+      await response.text()
+      expect(capturedQueryParams?.options?.resume).toBeUndefined()
+      expect(capturedQueryParams?.options?.mcpServers?.oc).toBeUndefined()
+      const followup = await post(app, makeRequest({ stream, tools: [], messages: [...side,
+        { role: "assistant", content: "Done." }, { role: "user", content: "Shorten that title." }],
+      }))
+      expect(followup.status).toBe(200)
+      await followup.text()
+      expect(capturedQueryParams?.options?.resume).toBeDefined()
+      expect(capturedQueryParams?.options?.mcpServers?.oc).toBeUndefined()
+    })
+  }
+
+  it("keeps the committed main tool set when a fresh side request fails", async () => {
+    const app = createTestApp()
+    const main = [{ role: "user", content: "main task with read access" }]
+    expect((await post(app, makeRequest({ stream: false, tools: [TOOL_A], messages: main }))).status).toBe(200)
+    mockQueryError = new Error("side request failed before publication")
+    expect((await post(app, makeRequest({ stream: false, tools: [TOOL_B],
+      messages: [{ role: "user", content: "independent side task" }],
+    }))).status).toBe(500)
+    mockQueryError = undefined
+    expect((await post(app, makeRequest({ stream: false, tools: [], messages: [...main,
+      { role: "assistant", content: "Done." }, { role: "user", content: "continue reading" }],
+    }))).status).toBe(200)
+    expect(capturedQueryParams?.options?.resume).toBeDefined()
+    expect(capturedQueryParams?.options?.allowedTools).toContain("mcp__oc__read_file")
+    expect(capturedQueryParams?.options?.allowedTools).not.toContain("mcp__oc__write_file")
+  })
+
   it("does not reuse tools for a different session", async () => {
     const app = createTestApp()
 
diff --git a/src/__tests__/session-tool-cache-eviction.test.ts b/src/__tests__/session-tool-cache-eviction.test.ts
index 71d6dee0..316f4215 100644
--- a/src/__tests__/session-tool-cache-eviction.test.ts
+++ b/src/__tests__/session-tool-cache-eviction.test.ts
@@ -72,7 +72,11 @@ async function post(app: any, session: string, tools: any[]) {
     body: JSON.stringify(makeRequest({
       stream: false,
       tools,
-      messages: [{ role: "user", content: "hi" }],
+      messages: tools.length > 0 ? [{ role: "user", content: "hi" }] : [
+        { role: "user", content: "hi" },
+        { role: "assistant", content: "ok" },
+        { role: "user", content: "continue" },
+      ],
     })),
   }))
 }
diff --git a/src/proxy/adapters/pi.ts b/src/proxy/adapters/pi.ts
index 4fb8e611..c8d239d2 100644
--- a/src/proxy/adapters/pi.ts
+++ b/src/proxy/adapters/pi.ts
@@ -71,7 +71,7 @@ export const piAdapter: AgentAdapter = {
   name: "pi",
 
   /**
-   * Oh My Pi drives one conversation from several callers at once, the main
+   * NOTE: agent-specific (pi) — Oh My Pi drives one conversation from several callers at once, the main
    * turn, title generation, and side questions asked mid-turn, and they all
    * carry the same `metadata.user_id.session_id`. Whichever one commits first
    * advances the mapping, so the other arrives holding a branch that no longer
diff --git a/src/proxy/server.ts b/src/proxy/server.ts
index c76e7f19..7ee87bfe 100644
--- a/src/proxy/server.ts
+++ b/src/proxy/server.ts
@@ -582,7 +582,7 @@ export function createProxyServer(config: Partial<ProxyConfig> = {}): ProxyServe
 
   // Cache last-seen tool definitions per agent session to prevent prompt cache
   // invalidation when clients intermittently omit tools on continuation requests.
-  const sessionToolCache = new LRUMap<string, any[]>(getMaxSessionsLimit())
+  const sessionToolCache = new LRUMap<string, { sdkSessionId: string; tools: any[] }>(getMaxSessionsLimit())
   // Cache the passthrough MCP server per session. Reusing the same server
   // across turns (when the tool set is unchanged) avoids subtle prompt-cache
   // invalidation from MCP server re-creation. Key hashes tool name + schema
@@ -2044,8 +2044,16 @@ export function createProxyServer(config: Partial<ProxyConfig> = {}): ProxyServe
         // can differ per attempt), but "did this session advance" is only
         // meaningful within one profile's cache scope — so commits and the
         // conflict check below both carry profileSessionId.
-        const commitSessionTurn = () => {
-          if (profileSessionId) requestMeta.sessionTurnLease?.markCommitted(profileSessionId)
+        const commitSessionTurn = (committedSdkSessionId: string | undefined) => {
+          if (profileSessionId) {
+            requestMeta.sessionTurnLease?.markCommitted(profileSessionId)
+            // Tool inheritance belongs to the successfully published branch.
+            // Failed side calls must not replace the main turn's tool set, and
+            // an empty fresh branch must not inherit or retain another's tools.
+            if (passthrough && committedSdkSessionId) {
+              sessionToolCache.set(profileSessionId, { sdkSessionId: committedSdkSessionId, tools: requestTools })
+            }
+          }
         }
         // Use the client-local CWD for fingerprint bucketing so that two
         // independent client projects don't collide on the same first-user-
@@ -2828,11 +2836,11 @@ export function createProxyServer(config: Partial<ProxyConfig> = {}): ProxyServe
       if (advisorModel) {
         requestTools = stripAdvisorTools(requestTools)
       }
-      if (passthrough && requestTools.length === 0 && profileSessionId) {
+      if (passthrough && isResume && requestTools.length === 0 && profileSessionId) {
         const cached = sessionToolCache.get(profileSessionId)
-        if (cached && cached.length > 0) {
-          requestTools = cached
-          plog(`[PROXY] ${requestMeta.requestId} tools_restored: client sent 0 tools but session had ${cached.length} — reusing cached tools to preserve prompt cache`)
+        if (cached && cached.sdkSessionId === resumeSessionId && cached.tools.length > 0) {
+          requestTools = cached.tools
+          plog(`[PROXY] ${requestMeta.requestId} tools_restored: client sent 0 tools but continued branch had ${cached.tools.length} — reusing cached tools to preserve prompt cache`)
         }
       }
       if (passthrough && requestTools.length > 0) {
@@ -2849,7 +2857,6 @@ export function createProxyServer(config: Partial<ProxyConfig> = {}): ProxyServe
             }
           }
         }
-        if (profileSessionId) sessionToolCache.set(profileSessionId, requestTools)
       }
       const hasDeferredTools = passthroughMcp?.hasDeferredTools ?? false
       // Count deferred tools: when auto-defer is active, non-core tools are deferred
@@ -4018,7 +4025,7 @@ export function createProxyServer(config: Partial<ProxyConfig> = {}): ProxyServe
                       releaseManagedPins()
                       void sweepSessionGc()
                     }
-                    commitSessionTurn()
+                    commitSessionTurn(currentSessionId)
                   }
                 }
               }
@@ -5070,7 +5077,7 @@ export function createProxyServer(config: Partial<ProxyConfig> = {}): ProxyServe
                       releaseManagedPins()
                       void sweepSessionGc()
                     }
-                    commitSessionTurn()
+                    commitSessionTurn(currentSessionId)
                   }
                 }
               }
@@ -5398,7 +5405,7 @@ export function createProxyServer(config: Partial<ProxyConfig> = {}): ProxyServe
                     mappingExpectedGeneration = recoveryMappingStored
                     recoveryForkPublished = true
                     void sweepSessionGc()
-                    commitSessionTurn()
+                    commitSessionTurn(recoverySessionId)
                   }
                 }
                 if (!recoveryForkPublished) {
@@ -5956,7 +5963,7 @@ export function createProxyServer(config: Partial<ProxyConfig> = {}): ProxyServe
                       releaseManagedPins()
                       void sweepSessionGc()
                     }
-                    commitSessionTurn()
+                    commitSessionTurn(currentSessionId)
                   }
                   if (mappingStored) {
                     claudeLog("passthrough.checkpoint_persisted", {