From 78f70487e355b79bbd526a294d5c9dc162c4b442 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Wed, 7 Oct 2026 16:46:18 +0200 Subject: [PATCH 1/3] fix(server): stop a Pi approval from outliving its session Pi kept session approvals across thread switches and rollback, so a permission granted in one thread could silently authorize a later thread. Clear approvals and cancel pending prompts at session boundaries, reject late cancelled-prompt responses, and check the bound session at turn finalize. Key approvals by every payload field except the request id in canonical order, with wire-level regression coverage for all boundaries. Model: gpt-6.1-sol (Codex) --- .../Adapters/PiAdapterV2.test.ts | 222 +++++++++++++++++- .../orchestration-v2/Adapters/PiAdapterV2.ts | 47 +++- 2 files changed, 260 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts index 06728c041066..40733b909b95 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts @@ -2,6 +2,7 @@ import { assert, describe, it } from "@effect/vitest"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { EnvironmentId, + CheckpointId, NodeId, ProviderInstanceId, ProviderSessionId, @@ -1522,12 +1523,13 @@ describe("PiAdapterV2", () => { const uiResponse = yield* fake.takeRequest("extension_ui_response"); assert.equal(uiResponse["id"], "ui-trust"); assert.equal(uiResponse["confirmed"], true); + // Field order and request id do not change the confirmation content. yield* fake.emit({ - type: "extension_ui_request", - id: "ui-trust-again", - method: "confirm", - title: "Run project extensions?", message: "This project has .pi/extensions.", + title: "Run project extensions?", + method: "confirm", + id: "ui-trust-again", + type: "extension_ui_request", }); assert.equal((yield* fake.takeRequest("extension_ui_response"))["id"], "ui-trust-again"); yield* fake.emit({ @@ -1548,6 +1550,218 @@ describe("PiAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(layerTest)), ); + it.effect.each([ + "new thread", + "switch", + "rollback", + "empty rollback", + "fork", + "prompt metadata", + "pending switch", + "pending rollback", + ] as const)("does not reuse session approvals after %s", (boundary) => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const forkFake = yield* makeFakePi; + const { runtime, takeEvent } = yield* openRuntime( + fake, + "default", + THREAD_ID, + SESSION_ID, + forkFake, + ); + const source = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + const prompt = { + type: "extension_ui_request", + method: "confirm", + title: "Run project extensions?", + message: "This project has .pi/extensions.", + extensionPath: "/workspace/extension-a.ts", + }; + yield* fake.emit({ ...prompt, id: "grant-in-a" }); + const granted = yield* takeEvent( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", + ); + assert.equal(granted.type, "runtime_request.updated"); + if (granted.type !== "runtime_request.updated") return; + const pendingApproval = boundary === "pending switch" || boundary === "pending rollback"; + if (!pendingApproval) { + yield* runtime.respondToRuntimeRequest({ + requestId: granted.runtimeRequest.id, + decision: "acceptForSession", + }); + assert.equal((yield* fake.takeRequest("extension_ui_response"))["confirmed"], true); + } + const targetThreadId = ThreadId.make("thread-pi-b"); + if (boundary === "new thread") { + yield* runtime.ensureThread({ + threadId: targetThreadId, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + } else if (boundary === "switch" || boundary === "pending switch") { + fake.queueState({ sessionFile: "/fake/b.jsonl" }); + yield* runtime.resumeThread({ + providerThread: { + ...source, + id: ProviderThreadId.make("provider-thread-pi-b"), + appThreadId: targetThreadId, + nativeThreadRef: { driver: PI_PROVIDER, nativeId: "/fake/b.jsonl", strength: "strong" }, + }, + }); + } else if ( + boundary === "rollback" || + boundary === "empty rollback" || + boundary === "pending rollback" + ) { + fake.queueState({ sessionFile: "/fake/rolled-back.jsonl" }); + yield* runtime.rollbackThread({ + providerThread: source, + target: { + type: "thread_start", + checkpointId: CheckpointId.make("checkpoint-pi"), + appRunOrdinal: 0, + }, + providerThreadTurns: + boundary === "empty rollback" + ? [] + : [ + { + id: ProviderTurnId.make("pi-turn-to-discard"), + providerThreadId: source.id, + nodeId: NodeId.make("pi-discard-node"), + runAttemptId: null, + nativeTurnRef: { + driver: PI_PROVIDER, + nativeId: "user-entry", + strength: "strong", + }, + ordinal: 1, + status: "completed", + startedAt: null, + completedAt: null, + }, + ], + }); + } else if (boundary === "fork") { + forkFake.queueState({ sessionFile: "/fake/fork.jsonl" }); + fake.queueState({ sessionFile: "/fake/fork.jsonl" }); + yield* runtime.forkThread({ sourceProviderThread: source, targetThreadId }); + } + if (pendingApproval) { + const cancelled = yield* takeEvent( + (event) => + event.type === "runtime_request.updated" && + event.runtimeRequest.id === granted.runtimeRequest.id && + event.runtimeRequest.status === "cancelled", + ); + assert.equal(cancelled.type, "runtime_request.updated"); + assert.equal((yield* fake.takeRequest("extension_ui_response"))["cancelled"], true); + const lateResponse = yield* Effect.result( + runtime.respondToRuntimeRequest({ + requestId: granted.runtimeRequest.id, + decision: "acceptForSession", + }), + ); + assert.equal(lateResponse._tag, "Failure"); + } + yield* fake.emit({ + ...prompt, + id: "ask-again", + ...(boundary === "prompt metadata" ? { extensionPath: "/workspace/extension-b.ts" } : {}), + }); + // An automatic wire response must fail the test instead of waiting + // indefinitely for the pending request it would suppress. + const asked = yield* Effect.race( + takeEvent( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", + ), + fake.takeRequest("extension_ui_response").pipe(Effect.as(null)), + ); + assert.isNotNull(asked); + if (asked === null) return; + assert.equal(asked.type, "runtime_request.updated"); + if (asked.type !== "runtime_request.updated") return; + assert.equal(asked.runtimeRequest.nativeRequestRef?.nativeId, "ask-again"); + assert.isFalse(fake.allRequests().some((request) => request["id"] === "ask-again")); + if ( + boundary === "new thread" || + boundary === "switch" || + boundary === "pending switch" || + boundary === "fork" + ) { + assert.equal(asked.threadId, targetThreadId); + } + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + + it.effect("drops session approvals when pi moved to another session mid-turn", () => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime, takeEvent } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + const prompt = { + type: "extension_ui_request", + method: "confirm", + title: "Run project extensions?", + message: "This project has .pi/extensions.", + }; + yield* fake.emit({ ...prompt, id: "grant-in-turn" }); + const granted = yield* takeEvent( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", + ); + assert.equal(granted.type, "runtime_request.updated"); + if (granted.type !== "runtime_request.updated") return; + yield* runtime.respondToRuntimeRequest({ + requestId: granted.runtimeRequest.id, + decision: "acceptForSession", + }); + assert.equal((yield* fake.takeRequest("extension_ui_response"))["confirmed"], true); + + const runTurn = (ordinal: number) => + Effect.gen(function* () { + yield* startTurn(runtime, providerThread, "default", [], "Hello pi", undefined, ordinal); + yield* fake.takeRequest("prompt"); + yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ type: "agent_settled" }); + yield* takeEvent((event) => event.type === "turn.terminal"); + }); + + // Same session: the cached approval still applies on the next turn. + yield* runTurn(1); + yield* fake.emit({ ...prompt, id: "same-session" }); + assert.equal((yield* fake.takeRequest("extension_ui_response"))["id"], "same-session"); + + // A slash command in pi switched sessions with no adapter command. + // The settle probe reads the bound session, the finalize read the other. + fake.queueState({}); + fake.queueState({ sessionFile: "/fake/some-other-session.jsonl" }); + yield* runTurn(2); + yield* fake.emit({ ...prompt, id: "other-session" }); + // A cached approval answers pi directly and never raises a request, so + // racing the two outcomes names the regression instead of timing out. + const askedAgain = yield* Effect.race( + takeEvent( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", + ).pipe(Effect.as(true)), + fake.takeRequest("extension_ui_response").pipe(Effect.as(false)), + ); + assert.isTrue(askedAgain); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + it.effect("offers an explicit empty value for extension input dialogs", () => Effect.gen(function* () { const fake = yield* makeFakePi; diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts index 9270253d5c74..eee5915a678c 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts @@ -1213,8 +1213,13 @@ export function makePiAdapterV2( return; } if (nativeRequestId === undefined) return; - const approvalTitle = recordString(event, "title") ?? ""; - const approvalKey = `${approvalTitle.length}:${approvalTitle}${recordString(event, "message") ?? ""}`; + // Request ids change on repeats; every other supplied field describes the prompt. + // @effect-diagnostics-next-line preferSchemaOverJson:off - Cache identity for an already JSON-parsed event. + const approvalKey = JSON.stringify( + Object.entries(event) + .filter(([key]) => key !== "id") + .sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0)), + ); if (method === "confirm" && sessionApprovals.has(approvalKey)) { yield* connection.send({ type: "extension_ui_response", @@ -1407,6 +1412,29 @@ export function makePiAdapterV2( }; }); + /** + * Drop cached session approvals when Pi's live session is not the one + * this thread is bound to. Pi can switch sessions from inside its own + * process, including from a slash command forwarded verbatim in the + * user's first message, and no adapter command reports that switch, so + * the boundary is only observable at turn finalize. An unreadable state + * clears the approvals too: an unknown identity is not evidence of an + * unchanged one. + */ + const clearApprovalsForForeignSession = Effect.fnUntraced(function* (state: PiThreadState) { + // The common turn has no cached approval, so it pays no extra round trip. + if (sessionApprovals.size === 0) return; + const live = yield* request({ type: "get_state" }, 2_000).pipe( + Effect.orElseSucceed(() => undefined), + ); + if ( + live === undefined || + recordString(live, "sessionFile") !== state.providerThread.nativeThreadRef?.nativeId + ) { + sessionApprovals.clear(); + } + }); + const finalizeTurn = Effect.fnUntraced(function* (state: PiThreadState, readUsage = true) { const turn = state.activeTurn; if (turn === null) return; @@ -1432,6 +1460,9 @@ export function makePiAdapterV2( } } yield* cancelPendingPrompts(completedAt); + // A dead transport (the only caller that skips reading usage) cannot + // ask Pi anything; the next process reads its own identity. + if (readUsage) yield* clearApprovalsForForeignSession(state); const treeRefs = turn.stopTreeRefs !== undefined ? turn.stopTreeRefs : yield* captureTurnTreeRefs(); const tokenUsage = readUsage @@ -2011,6 +2042,8 @@ export function makePiAdapterV2( // Even a failed lifecycle operation can change Pi's native session. // Never leave the old app binding or model defaults usable afterward. threadState = null; + sessionApprovals.clear(); + yield* cancelPendingPrompts(yield* DateTime.now); appliedModel = null; appliedThinking = null; appliedSessionName = null; @@ -2537,6 +2570,8 @@ export function makePiAdapterV2( id: pending.nativeRequestId, ...response, }); + // A session switch can cancel this prompt while the send is pending. + if (pendingPrompts.get(String(requestInput.requestId)) !== pending) return; // Dropped only once Pi has the answer, so a failed send leaves the // request retryable and still cancellable during teardown. pendingPrompts.delete(String(requestInput.requestId)); @@ -2666,13 +2701,15 @@ export function makePiAdapterV2( // message, so the rollback boundary is the first user entry of // the earliest turn being discarded. const forkEntryId = piRollbackForkEntry(rollbackInput); + if (forkEntryId === undefined) { + return yield* protocolError("Pi rollback target has no captured session-tree entry"); + } + sessionApprovals.clear(); + yield* cancelPendingPrompts(yield* DateTime.now); if (forkEntryId === null) { // Nothing after the target: the conversation is already there. return piThreadSnapshot(state.providerThread); } - if (forkEntryId === undefined) { - return yield* protocolError("Pi rollback target has no captured session-tree entry"); - } const forkData = yield* lifecycleRequest({ type: "fork", entryId: forkEntryId }); if (recordField(forkData, "cancelled") === true) { return yield* protocolError("A Pi extension cancelled the session fork"); From 345352d28d909a604a6616cb2e098848c7162953 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Wed, 7 Oct 2026 17:10:19 +0200 Subject: [PATCH 2/3] fix(server): restore Pi approval keys and reuse settle state The audit found that the expanded approval key cannot distinguish real Pi confirm payloads, and the metadata test fabricated an unsupported field. Restore the original title/message key, remove that case and its redundant wire assertion, and preserve the runtime_request.updated assertions. Defer the transport send to test cancellation during thread replacement and empty rollback. Both cases fail when the adapter's late-response guard is removed. Reuse successful settle-probe state during finalization, and test same-session reuse, foreign-session invalidation, and failed-probe fallback. Keep the existing authorization fix, boundary clearing, pending-prompt cancellation, and late-response guard. Leave user-visible prompt text and state lookups on finalization paths without a successful probe alone. Model: gpt-6.1-sol (Codex) --- .../Adapters/PiAdapterV2.test.ts | 239 +++++++++++++----- .../orchestration-v2/Adapters/PiAdapterV2.ts | 32 ++- 2 files changed, 197 insertions(+), 74 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts index 40733b909b95..b4f52baaf88c 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts @@ -1,4 +1,4 @@ -import { assert, describe, it } from "@effect/vitest"; +import { assert, describe, it, vi } from "@effect/vitest"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { EnvironmentId, @@ -20,6 +20,7 @@ import { import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; @@ -43,6 +44,7 @@ import { import { handoffBudget } from "../ContextHandoffBudget.ts"; import { makePiAdapterV2, PI_PROVIDER } from "./PiAdapterV2.ts"; import { makePiRpcConnection, type PiRpcRecord } from "./PiRpc.ts"; +import * as PiRpc from "./PiRpc.ts"; const layerServerConfig = ServerConfig.layerTest(process.cwd(), { prefix: "t3-pi-v2-adapter-", @@ -1556,7 +1558,6 @@ describe("PiAdapterV2", () => { "rollback", "empty rollback", "fork", - "prompt metadata", "pending switch", "pending rollback", ] as const)("does not reuse session approvals after %s", (boundary) => @@ -1580,7 +1581,6 @@ describe("PiAdapterV2", () => { method: "confirm", title: "Run project extensions?", message: "This project has .pi/extensions.", - extensionPath: "/workspace/extension-a.ts", }; yield* fake.emit({ ...prompt, id: "grant-in-a" }); const granted = yield* takeEvent( @@ -1673,7 +1673,6 @@ describe("PiAdapterV2", () => { yield* fake.emit({ ...prompt, id: "ask-again", - ...(boundary === "prompt metadata" ? { extensionPath: "/workspace/extension-b.ts" } : {}), }); // An automatic wire response must fail the test instead of waiting // indefinitely for the pending request it would suppress. @@ -1689,7 +1688,6 @@ describe("PiAdapterV2", () => { assert.equal(asked.type, "runtime_request.updated"); if (asked.type !== "runtime_request.updated") return; assert.equal(asked.runtimeRequest.nativeRequestRef?.nativeId, "ask-again"); - assert.isFalse(fake.allRequests().some((request) => request["id"] === "ask-again")); if ( boundary === "new thread" || boundary === "switch" || @@ -1701,65 +1699,186 @@ describe("PiAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(layerTest)), ); - it.effect("drops session approvals when pi moved to another session mid-turn", () => - Effect.gen(function* () { - const fake = yield* makeFakePi; - const { runtime, takeEvent } = yield* openRuntime(fake); - const providerThread = yield* runtime.ensureThread({ - threadId: THREAD_ID, - modelSelection: modelSelection("default"), - runtimePolicy, - }); - const prompt = { - type: "extension_ui_request", - method: "confirm", - title: "Run project extensions?", - message: "This project has .pi/extensions.", - }; - yield* fake.emit({ ...prompt, id: "grant-in-turn" }); - const granted = yield* takeEvent( - (event) => - event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", - ); - assert.equal(granted.type, "runtime_request.updated"); - if (granted.type !== "runtime_request.updated") return; - yield* runtime.respondToRuntimeRequest({ - requestId: granted.runtimeRequest.id, - decision: "acceptForSession", - }); - assert.equal((yield* fake.takeRequest("extension_ui_response"))["confirmed"], true); - - const runTurn = (ordinal: number) => - Effect.gen(function* () { - yield* startTurn(runtime, providerThread, "default", [], "Hello pi", undefined, ordinal); - yield* fake.takeRequest("prompt"); - yield* fake.emit({ type: "agent_start" }); - yield* fake.emit({ type: "agent_settled" }); - yield* takeEvent((event) => event.type === "turn.terminal"); + it.effect.each(["switch", "rollback"] as const)( + "does not restore session approvals when a send finishes after %s", + (boundary) => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const sending = yield* Deferred.make(); + const releaseSend = yield* Deferred.make(); + const originalMakeConnection = PiRpc.makePiRpcConnection; + // PiRpc.send returns after enqueueing, so delaying the child stdin sink + // would not hold the adapter's send. Defer that transport call itself. + yield* Effect.acquireRelease( + Effect.sync(() => + vi.spyOn(PiRpc, "makePiRpcConnection").mockImplementation((options) => + originalMakeConnection(options).pipe( + Effect.map((connection) => ({ + ...connection, + send: (record) => + Effect.gen(function* () { + if (record["id"] === "late-grant" && record["confirmed"] === true) { + yield* Deferred.succeed(sending, undefined); + yield* Deferred.await(releaseSend); + } + yield* connection.send(record); + }), + })), + ), + ), + ), + (spy) => Effect.sync(() => spy.mockRestore()), + ); + const { runtime, takeEvent } = yield* openRuntime(fake); + const source = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, }); + const prompt = { + type: "extension_ui_request", + method: "confirm", + title: "Run project extensions?", + message: "This project has .pi/extensions.", + }; + yield* fake.emit({ ...prompt, id: "late-grant" }); + const granted = yield* takeEvent( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", + ); + assert.equal(granted.type, "runtime_request.updated"); + if (granted.type !== "runtime_request.updated") return; + // Call the adapter runtime directly, before any boundary cancels the request. + const response = yield* runtime + .respondToRuntimeRequest({ + requestId: granted.runtimeRequest.id, + decision: "acceptForSession", + }) + .pipe(Effect.forkScoped); + yield* Deferred.await(sending); + if (boundary === "switch") { + yield* runtime.ensureThread({ + threadId: ThreadId.make("thread-pi-after-late-grant"), + modelSelection: modelSelection("default"), + runtimePolicy, + }); + } else { + yield* runtime.rollbackThread({ + providerThread: source, + target: { + type: "thread_start", + checkpointId: CheckpointId.make("checkpoint-pi-late-grant"), + appRunOrdinal: 0, + }, + providerThreadTurns: [], + }); + } + yield* takeEvent( + (event) => + event.type === "runtime_request.updated" && + event.runtimeRequest.id === granted.runtimeRequest.id && + event.runtimeRequest.status === "cancelled", + ); + const cancellation = yield* fake.takeRequest("extension_ui_response"); + assert.equal(cancellation["id"], "late-grant"); + assert.equal(cancellation["cancelled"], true); + yield* Deferred.succeed(releaseSend, undefined); + yield* Fiber.join(response); + const lateResponse = yield* fake.takeRequest("extension_ui_response"); + assert.equal(lateResponse["id"], "late-grant"); + assert.equal(lateResponse["confirmed"], true); + + yield* fake.emit({ ...prompt, id: "after-late-grant" }); + const next = yield* Effect.race( + takeEvent((event) => event.type === "runtime_request.updated"), + fake.takeRequest("extension_ui_response").pipe(Effect.as(null)), + ); + assert.isNotNull(next); + if (next === null || next.type !== "runtime_request.updated") return; + // A late response must neither resolve the cancelled request nor cache its approval. + assert.equal(next.runtimeRequest.status, "pending"); + assert.equal(next.runtimeRequest.nativeRequestRef?.nativeId, "after-late-grant"); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); - // Same session: the cached approval still applies on the next turn. - yield* runTurn(1); - yield* fake.emit({ ...prompt, id: "same-session" }); - assert.equal((yield* fake.takeRequest("extension_ui_response"))["id"], "same-session"); - - // A slash command in pi switched sessions with no adapter command. - // The settle probe reads the bound session, the finalize read the other. - fake.queueState({}); - fake.queueState({ sessionFile: "/fake/some-other-session.jsonl" }); - yield* runTurn(2); - yield* fake.emit({ ...prompt, id: "other-session" }); - // A cached approval answers pi directly and never raises a request, so - // racing the two outcomes names the regression instead of timing out. - const askedAgain = yield* Effect.race( - takeEvent( + it.effect.each(["settle probe", "failed settle probe"] as const)( + "drops session approvals when pi moved to another session mid-turn via %s", + (path) => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime, takeEvent } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + const prompt = { + type: "extension_ui_request", + method: "confirm", + title: "Run project extensions?", + message: "This project has .pi/extensions.", + }; + yield* fake.emit({ ...prompt, id: "grant-in-turn" }); + const granted = yield* takeEvent( (event) => event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", - ).pipe(Effect.as(true)), - fake.takeRequest("extension_ui_response").pipe(Effect.as(false)), - ); - assert.isTrue(askedAgain); - }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + assert.equal(granted.type, "runtime_request.updated"); + if (granted.type !== "runtime_request.updated") return; + yield* runtime.respondToRuntimeRequest({ + requestId: granted.runtimeRequest.id, + decision: "acceptForSession", + }); + assert.equal((yield* fake.takeRequest("extension_ui_response"))["confirmed"], true); + + const runTurn = (ordinal: number) => + Effect.gen(function* () { + yield* startTurn( + runtime, + providerThread, + "default", + [], + path === "settle probe" ? "Hello pi" : "/extension-command", + undefined, + ordinal, + ); + yield* fake.takeRequest("prompt"); + if (path === "settle probe") { + yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ type: "agent_settled" }); + } else { + fake.failNextState(); + yield* fake.emit({ type: "response", command: "prompt", success: true }); + } + yield* takeEvent((event) => event.type === "turn.terminal"); + }); + + // A successful probe supplies the identity without another state request. + const stateRequestCount = () => + fake.allRequests().filter((request) => request["type"] === "get_state").length; + const before = stateRequestCount(); + yield* runTurn(1); + assert.equal(stateRequestCount() - before, path === "settle probe" ? 1 : 2); + // Same session: the cached approval still applies on the next turn. + yield* fake.emit({ ...prompt, id: "same-session" }); + assert.equal((yield* fake.takeRequest("extension_ui_response"))["id"], "same-session"); + + // A slash command in pi switched sessions with no adapter command. + // A failed probe leaves the queued identity for the fallback lookup. + fake.queueState({ sessionFile: "/fake/some-other-session.jsonl" }); + yield* runTurn(2); + yield* fake.emit({ ...prompt, id: "other-session" }); + // A cached approval answers pi directly and never raises a request, so + // racing the two outcomes names the regression instead of timing out. + const askedAgain = yield* Effect.race( + takeEvent( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", + ).pipe(Effect.as(true)), + fake.takeRequest("extension_ui_response").pipe(Effect.as(false)), + ); + assert.isTrue(askedAgain); + }).pipe(Effect.scoped, Effect.provide(layerTest)), ); it.effect("offers an explicit empty value for extension input dialogs", () => diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts index eee5915a678c..b3a18a372a3f 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts @@ -1213,13 +1213,8 @@ export function makePiAdapterV2( return; } if (nativeRequestId === undefined) return; - // Request ids change on repeats; every other supplied field describes the prompt. - // @effect-diagnostics-next-line preferSchemaOverJson:off - Cache identity for an already JSON-parsed event. - const approvalKey = JSON.stringify( - Object.entries(event) - .filter(([key]) => key !== "id") - .sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0)), - ); + const approvalTitle = recordString(event, "title") ?? ""; + const approvalKey = `${approvalTitle.length}:${approvalTitle}${recordString(event, "message") ?? ""}`; if (method === "confirm" && sessionApprovals.has(approvalKey)) { yield* connection.send({ type: "extension_ui_response", @@ -1421,12 +1416,17 @@ export function makePiAdapterV2( * clears the approvals too: an unknown identity is not evidence of an * unchanged one. */ - const clearApprovalsForForeignSession = Effect.fnUntraced(function* (state: PiThreadState) { + const clearApprovalsForForeignSession = Effect.fnUntraced(function* ( + state: PiThreadState, + stateData: Option.Option, + ) { // The common turn has no cached approval, so it pays no extra round trip. if (sessionApprovals.size === 0) return; - const live = yield* request({ type: "get_state" }, 2_000).pipe( - Effect.orElseSucceed(() => undefined), - ); + const live = Option.isSome(stateData) + ? stateData.value + : yield* request({ type: "get_state" }, 2_000).pipe( + Effect.orElseSucceed(() => undefined), + ); if ( live === undefined || recordString(live, "sessionFile") !== state.providerThread.nativeThreadRef?.nativeId @@ -1435,7 +1435,11 @@ export function makePiAdapterV2( } }); - const finalizeTurn = Effect.fnUntraced(function* (state: PiThreadState, readUsage = true) { + const finalizeTurn = Effect.fnUntraced(function* ( + state: PiThreadState, + readUsage = true, + stateData: Option.Option = Option.none(), + ) { const turn = state.activeTurn; if (turn === null) return; state.activeTurn = null; @@ -1462,7 +1466,7 @@ export function makePiAdapterV2( yield* cancelPendingPrompts(completedAt); // A dead transport (the only caller that skips reading usage) cannot // ask Pi anything; the next process reads its own identity. - if (readUsage) yield* clearApprovalsForForeignSession(state); + if (readUsage) yield* clearApprovalsForForeignSession(state, stateData); const treeRefs = turn.stopTreeRefs !== undefined ? turn.stopTreeRefs : yield* captureTurnTreeRefs(); const tokenUsage = readUsage @@ -1955,7 +1959,7 @@ export function makePiAdapterV2( (recordNumber(data, "pendingMessageCount") ?? 0) === 0 ) { turn.settleWhenIdle = false; - if (state !== null) yield* finalizeTurn(state); + if (state !== null) yield* finalizeTurn(state, true, Option.some(data)); } return; } From d9364c21e6cd1f335ea1f831992d923147ef5791 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Wed, 7 Oct 2026 21:21:55 +0200 Subject: [PATCH 3/3] test(server): prove unreadable Pi identity clears session approvals Model: gpt-6.1-sol (Codex harness) --- .../Adapters/PiAdapterV2.test.ts | 96 +++++++++++++++++-- 1 file changed, 89 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts index b4f52baaf88c..357c95b1a85a 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts @@ -89,8 +89,9 @@ interface FakePi { readonly deferNextState: () => void; /** Resolve the held `get_state` request. */ readonly resolveDeferredState: (data: unknown) => Effect.Effect; - /** Reject the next `get_state` request. */ - readonly failNextState: () => void; + /** Reject the next `count` state requests, defaulting to one. */ + readonly failNextState: (count?: number) => void; + readonly failedStateReads: () => number; readonly deferNextLifecycle: (type: "switch_session" | "new_session") => void; readonly queueModels: (models: ReadonlyArray) => void; readonly vetoNextNewSession: () => void; @@ -145,7 +146,8 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { const allRequests: Array = []; let deferState = false; let deferredStateRequest: PiRpcRecord | undefined; - let failState = false; + let stateFailuresRemaining = 0; + let failedStateReads = 0; let vetoSwitch = false; let vetoNewSession = false; let deferredLifecycle: string | undefined; @@ -169,8 +171,9 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { }; switch (record["type"]) { case "get_state": - if (failState) { - failState = false; + if (stateFailuresRemaining > 0) { + stateFailuresRemaining--; + failedStateReads++; return { ...base, success: false, error: "state unavailable" }; } // Queued data overrides fields of the recorded idle state, so a test @@ -288,9 +291,10 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { data, }); }), - failNextState: () => { - failState = true; + failNextState: (count = 1) => { + stateFailuresRemaining = count; }, + failedStateReads: () => failedStateReads, deferNextLifecycle: (type) => { deferredLifecycle = type; }, @@ -1881,6 +1885,84 @@ describe("PiAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(layerTest)), ); + it.effect("drops session approvals when both command-only state reads fail", () => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime, takeEvent } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + const prompt = { + type: "extension_ui_request", + method: "confirm", + title: "Run project extensions?", + message: "This project has .pi/extensions.", + }; + yield* fake.emit({ ...prompt, id: "grant-before-unreadable-state" }); + const granted = yield* takeEvent( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", + ); + assert.equal(granted.type, "runtime_request.updated"); + if (granted.type !== "runtime_request.updated") return; + yield* runtime.respondToRuntimeRequest({ + requestId: granted.runtimeRequest.id, + decision: "acceptForSession", + }); + assert.equal((yield* fake.takeRequest("extension_ui_response"))["confirmed"], true); + + const runCommand = Effect.fnUntraced(function* (ordinal: number, failedReads = 0) { + yield* startTurn( + runtime, + providerThread, + "default", + [], + "/extension-command", + undefined, + ordinal, + ); + yield* fake.takeRequest("prompt"); + if (failedReads > 0) fake.failNextState(failedReads); + yield* fake.emit({ type: "response", command: "prompt", success: true }); + const terminal = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed"); + }); + + // A readable, unchanged identity keeps the cached approval after a command-only turn. + yield* runCommand(1); + assert.equal(fake.failedStateReads(), 0); + yield* fake.emit({ ...prompt, id: "same-readable-session" }); + const cached = yield* fake.takeRequest("extension_ui_response"); + assert.equal(cached["id"], "same-readable-session"); + assert.equal(cached["confirmed"], true); + + const before = fake.allRequests().filter((request) => request["type"] === "get_state").length; + // No foreign session is supplied: both the settle probe and finalize lookup fail. + yield* runCommand(2, 2); + assert.equal( + fake.allRequests().filter((request) => request["type"] === "get_state").length - before, + 2, + ); + assert.equal(fake.failedStateReads(), 2); + + yield* fake.emit({ ...prompt, id: "ask-after-unreadable-state" }); + const asked = yield* Effect.race( + takeEvent( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", + ), + fake.takeRequest("extension_ui_response").pipe(Effect.as(null)), + ); + assert.isNotNull(asked); + if (asked === null || asked.type !== "runtime_request.updated") return; + assert.equal(asked.threadId, THREAD_ID); + assert.equal(asked.runtimeRequest.nativeRequestRef?.nativeId, "ask-after-unreadable-state"); + assert.equal(asked.runtimeRequest.status, "pending"); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + it.effect("offers an explicit empty value for extension input dialogs", () => Effect.gen(function* () { const fake = yield* makeFakePi;