From 416ef988f10b9a5551d22149ac55b29fad3ddf23 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Sun, 4 Oct 2026 19:21:45 +0200 Subject: [PATCH 1/9] fix(server): fix Pi session approval authorization Keep approval identity tied to every prompt field except the native request id, and clear cached approvals and pending prompts wherever the bound native session changes: every thread registration and every rollback. An empty rollback clears them too, even though it moves no session, so the boundary stays uniform and fail-closed. Preserve the empty rollback boundary check. Implemented with gpt-6.1-sol through the Codex harness (initial implementation). --- .../Adapters/PiAdapterV2.test.ts | 145 ++++++++++++++++++ .../orchestration-v2/Adapters/PiAdapterV2.ts | 21 ++- 2 files changed, 161 insertions(+), 5 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..26b31ccd62ac 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, @@ -1548,6 +1549,150 @@ 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: "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/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: "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/fake/b.jsonl" }); + yield* runtime.resumeThread({ + providerThread: { + ...source, + id: ProviderThreadId.make("provider-thread-pi-b"), + appThreadId: targetThreadId, + nativeThreadRef: { driver: PI_PROVIDER, nativeId: "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/fake/b.jsonl", strength: "strong" }, + }, + }); + } else if ( + boundary === "rollback" || + boundary === "empty rollback" || + boundary === "pending rollback" + ) { + fake.queueState({ sessionFile: "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/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: "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/fake/fork.jsonl" }); + fake.queueState({ sessionFile: "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/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: "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/workspace/extension-b.ts" } : {}), + }); + const asked = yield* takeEvent( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", + ); + 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(testLayer)), + ); + 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 87e24f92b0e8..ca5cac259340 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts @@ -99,6 +99,7 @@ export const PI_PROVIDER = ProviderDriverKind.make("pi"); const PI_DRIVER_KIND = PI_PROVIDER; const PI_DEFAULT_INSTANCE_ID = defaultInstanceIdForDriver(PI_DRIVER_KIND); const DEFAULT_PI_SETTINGS = Schema.decodeSync(PiSettings)({}); +const encodeApprovalKey = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); /** * Sentinel model slug meaning "do not call set_model": Pi resolves the model @@ -1213,8 +1214,12 @@ 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. + const approvalKey = encodeApprovalKey( + Object.entries(event) + .filter(([key]) => key !== "id") + .sort(([left], [right]) => left.localeCompare(right)), + ); if (method === "confirm" && sessionApprovals.has(approvalKey)) { yield* connection.send({ type: "extension_ui_response", @@ -2011,6 +2016,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 +2544,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 +2675,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 f7fec27aa36473cf411c2bf4c125267ac527ed58 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Sun, 4 Oct 2026 19:22:01 +0200 Subject: [PATCH 2/9] fix(server): scope Codex plan deltas and reclaim native thread state Scope streamed plan text to its turn and preserve token accounting only for tracked native threads. Release usage and model metadata on close or unload, and prevent an older child metadata lookup from restoring released state. Implemented with gpt-6.1-sol through the Codex harness (initial implementation). --- .../Adapters/CodexAdapterV2.test.ts | 217 ++++++++++++++++-- .../Adapters/CodexAdapterV2.ts | 49 +++- 2 files changed, 230 insertions(+), 36 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 63f0cbdae66b..7de904a25883 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2498,6 +2498,135 @@ describe("CodexAdapterV2 post-settle continuation", () => { : [], ); + it.effect.each([false, true])( + "scopes plan deltas to turns and resets usage on native close=%s", + (closed) => + Effect.gen(function* () { + const nativeThreadId = "state-scope-thread"; + const first = "state-scope-first"; + const second = "state-scope-second"; + const secondStart = codexReplayPreamble({ + nativeThreadId, + nativeTurnId: second, + prompt: "second", + }) + .slice(5) + .map((entry) => + "frame" in entry + ? { + ...entry, + frame: + typeof entry.frame === "object" && entry.frame !== null && "id" in entry.frame + ? { ...entry.frame, id: 4 } + : entry.frame, + } + : entry, + ); + const notification = ( + method: string, + params: unknown, + ): CodexReplay.CodexAppServerReplayEntry => ({ + type: "emit_inbound", + label: method, + frame: { method, params }, + }); + const usage = (inputTokens: number) => ({ + totalTokens: inputTokens, + inputTokens, + cachedInputTokens: 0, + outputTokens: 0, + reasoningOutputTokens: 0, + }); + const transcript = makeCodexReplayTranscript({ + scenario: `state-scope-${closed}`, + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId: first, prompt: "first" }), + notification("item/plan/delta", { + threadId: nativeThreadId, + turnId: first, + itemId: "reused-plan", + delta: "OLD", + }), + notification("thread/tokenUsage/updated", { + threadId: nativeThreadId, + turnId: first, + tokenUsage: { total: usage(100), last: usage(10), modelContextWindow: 200_000 }, + }), + notification("turn/completed", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: first, status: "interrupted" }), + }), + ...(closed ? [notification("thread/closed", { threadId: nativeThreadId })] : []), + ...secondStart, + notification("item/plan/delta", { + threadId: nativeThreadId, + turnId: second, + itemId: "reused-plan", + delta: "NEW", + }), + notification("item/completed", { + threadId: nativeThreadId, + turnId: second, + item: { type: "plan", id: "reused-plan", text: "" }, + }), + notification("thread/tokenUsage/updated", { + threadId: nativeThreadId, + turnId: second, + tokenUsage: { total: usage(120), last: usage(5), modelContextWindow: 200_000 }, + }), + notification("turn/completed", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: second, status: "completed" }), + }), + ], + }); + const secondTerminal = yield* Deferred.make(); + let terminals = 0; + const harness = yield* makeCodexReplayHarness(transcript, (event) => { + if (event.type !== "turn.terminal") return Effect.void; + terminals += 1; + return terminals === 2 ? Deferred.succeed(secondTerminal, undefined) : Effect.void; + }); + const now = yield* DateTime.now; + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("state-first"), + text: "first", + }), + ); + yield* harness.firstTerminal; + yield* harness.runtime.startTurn({ + ...makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("state-second"), + text: "second", + }), + providerTurnOrdinal: 2, + runOrdinal: 2, + }); + yield* Deferred.await(secondTerminal); + const plans = harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "proposed_plan" + ? [event.turnItem.markdown] + : [], + ); + assert.deepEqual(plans, ["OLD", "NEW", "NEW"]); + const completed = harness.events.find( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "completed", + ); + assert.equal(completed?.type, "provider_turn.updated"); + if (completed?.type === "provider_turn.updated") { + assert.equal(completed.providerTurn.turnTokenUsage?.inputTokens, closed ? 5 : 20); + } + }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ); + it.effect("keeps an asynchronous Codex question actionable after the turn completes", () => Effect.scoped( Effect.gen(function* () { @@ -6717,6 +6846,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { { name: "missing", model: null }, { name: "invalid", model: null }, { name: "wrong child", model: null }, + { name: "closed", model: null }, ])("reads $name child metadata without using the parent model", ({ name, model }) => Effect.scoped( Effect.gen(function* () { @@ -6724,7 +6854,39 @@ describe("CodexAdapterV2 post-settle continuation", () => { const modelReported = yield* Deferred.make(); let metadataRequests = 0; const harness = yield* makeCodexReplayHarness( - resumeSubagentTranscript, + name === "closed" + ? { + ...resumeSubagentTranscript, + entries: resumeSubagentTranscript.entries.flatMap((entry) => + entry.type === "emit_inbound" && + entry.label === "item/completed/subAgentActivity-started" + ? [ + { + type: "emit_inbound" as const, + frame: { + method: "model/rerouted", + params: { + threadId: RESUME_CHILD_THREAD, + turnId: RESUME_CHILD_TURN_1, + fromModel: "gpt-5.6-sol", + toModel: "gpt-6-astra", + reason: "highRiskCyberActivity", + }, + }, + }, + { + type: "emit_inbound" as const, + frame: { + method: "thread/closed", + params: { threadId: RESUME_CHILD_THREAD }, + }, + }, + entry, + ] + : [entry], + ), + } + : resumeSubagentTranscript, (event) => event.type === "subagent.updated" && event.subagent.model === model ? Deferred.succeed(modelReported, undefined) @@ -6771,7 +6933,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); - it.effect.each(["failed", "malformed", "wrong child", "blank model"] as const)( +it.effect.each(["failed", "malformed", "wrong child", "blank model"] as const)( "resumes child metadata after a %s read", (readResult) => Effect.scoped( @@ -6828,39 +6990,41 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); - it.effect.each(["thread/settings/updated", "model/rerouted"] as const)( + it.effect.each(["thread/settings/updated", "model/rerouted", "thread/closed"] as const)( "keeps %s child metadata when an older lookup finishes later", (method) => Effect.scoped( Effect.gen(function* () { const releaseMetadata = yield* Deferred.make(); const observed = yield* Deferred.make(); - const model = "gpt-5.6-sol"; + const model = method === "thread/closed" ? null : "gpt-5.6-sol"; const notification: CodexReplay.CodexAppServerReplayEntry = { type: "emit_inbound", frame: { method, params: - method === "model/rerouted" - ? { - threadId: RESUME_CHILD_THREAD, - turnId: RESUME_CHILD_TURN_1, - fromModel: "gpt-6-astra", - toModel: model, - reason: "highRiskCyberActivity", - } - : { - threadId: RESUME_CHILD_THREAD, - threadSettings: { - model, - modelProvider: "openai", - cwd: "/workspace", - approvalPolicy: "never", - approvalsReviewer: "auto_review", - collaborationMode: { mode: "default", settings: { model } }, - sandboxPolicy: { type: "dangerFullAccess" }, + method === "thread/closed" + ? { threadId: RESUME_CHILD_THREAD } + : method === "model/rerouted" + ? { + threadId: RESUME_CHILD_THREAD, + turnId: RESUME_CHILD_TURN_1, + fromModel: "gpt-6-astra", + toModel: model, + reason: "highRiskCyberActivity", + } + : { + threadId: RESUME_CHILD_THREAD, + threadSettings: { + model, + modelProvider: "openai", + cwd: "/workspace", + approvalPolicy: "never", + approvalsReviewer: "auto_review", + collaborationMode: { mode: "default", settings: { model } }, + sandboxPolicy: { type: "dangerFullAccess" }, + }, }, - }, }, }; const harness = yield* makeCodexReplayHarness( @@ -6868,7 +7032,9 @@ describe("CodexAdapterV2 post-settle continuation", () => { ...resumeSubagentTranscript, entries: resumeSubagentTranscript.entries.flatMap((entry) => entry.type === "emit_inbound" && entry.label === "turn/completed/root" - ? [entry, notification] + ? method === "thread/closed" + ? [notification, entry] + : [entry, notification] : [entry], ), }, @@ -6892,7 +7058,8 @@ describe("CodexAdapterV2 post-settle continuation", () => { }), ); yield* TestClock.adjust("100 millis"); - yield* Deferred.await(observed); + if (method === "thread/closed") yield* harness.firstTerminal; + else yield* Deferred.await(observed); assert.equal(harness.subagentUpdates().at(-1)?.subagent.status, "completed"); yield* Deferred.succeed(releaseMetadata, undefined); yield* TestClock.adjust("30 seconds"); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 19517c349116..3a5b09679e5b 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1085,6 +1085,7 @@ function codexErrorInfoCode(value: unknown): string | null { } interface ActiveCodexTurnContext { + readonly planDeltas: Map; latestProviderFailure?: { readonly nativeMessage: string; readonly failure: OrchestrationV2ProviderFailure; @@ -1169,6 +1170,7 @@ interface CodexSubagentThreadContext { readonly turnItemId: OrchestrationV2TurnItem["id"]; readonly turnItemOrdinal: number; task: OrchestrationV2Subagent; + modelLookupActive: boolean; } /** @@ -1788,8 +1790,6 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi const providerRetries = yield* Ref.make( new Map(), ); - // Streamed plan text per plan item, dropped when the item completes. - const planDeltas = new Map(); const planIds = yield* Ref.make(new Map()); const pendingRuntimeRequests = yield* Ref.make( new Map(), @@ -1938,6 +1938,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi subagent: null, startedAt: input.startedAt, itemPositions: new Map(), + planDeltas: new Map(), }; yield* Ref.update(limitedTurnItems, (current) => { const next = new Map(current); @@ -2498,6 +2499,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi subagent, startedAt: turn.startedAt, itemPositions: new Map(), + planDeltas: new Map(), }; beginTurnTokenUsage(activeContext); yield* Ref.update(activeTurns, (current) => { @@ -2718,6 +2720,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi }), turnItemOrdinal, task, + modelLookupActive: true, } satisfies CodexSubagentThreadContext; yield* Ref.update(subagentThreads, (current) => { @@ -2848,6 +2851,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi Effect.timeout("5 seconds"), Effect.flatMap((response) => response.thread.id === input.nativeThreadId && + subagent.modelLookupActive && !subagentModels.has(input.nativeThreadId) ? updateSubagentModel(input.nativeThreadId, response.model) : Effect.void, @@ -3924,8 +3928,8 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi return; } yield* completeProviderRetry(context, yield* DateTime.now); - const markdown = `${planDeltas.get(payload.itemId) ?? ""}${payload.delta}`; - planDeltas.set(payload.itemId, markdown); + const markdown = `${context.planDeltas.get(payload.itemId) ?? ""}${payload.delta}`; + context.planDeltas.set(payload.itemId, markdown); const artifacts = yield* buildProposedPlanArtifacts({ context, nativeItemId: payload.itemId, @@ -4021,11 +4025,10 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi yield* client.handleServerNotification("thread/tokenUsage/updated", (payload) => Effect.gen(function* () { - accumulateCodexTurnTokenUsage( - usageStateForThread(payload.threadId), - payload.turnId, - payload.tokenUsage, - ); + const usageState = turnTokenUsageByThread.get(payload.threadId); + if (usageState !== undefined) { + accumulateCodexTurnTokenUsage(usageState, payload.turnId, payload.tokenUsage); + } const context = yield* awaitActiveTurn(payload.turnId); if (context === undefined) { return; @@ -4067,6 +4070,16 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi updateSubagentModel(payload.threadId, payload.toModel), ); + const releaseThreadState = Effect.fnUntraced(function* (nativeThreadId: string) { + turnTokenUsageByThread.delete(nativeThreadId); + subagentModels.delete(nativeThreadId); + const subagent = (yield* Ref.get(subagentThreads)).get(nativeThreadId); + if (subagent !== undefined) subagent.modelLookupActive = false; + }); + yield* client.handleServerNotification("thread/closed", (payload) => + releaseThreadState(payload.threadId), + ); + yield* client.handleServerNotification("turn/started", (payload) => Effect.gen(function* () { const context = (yield* Ref.get(activeTurns)).get(payload.turn.id); @@ -4589,8 +4602,8 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi const markdown = payload.item.text.length > 0 ? payload.item.text - : (planDeltas.get(payload.item.id) ?? ""); - planDeltas.delete(payload.item.id); + : (context.planDeltas.get(payload.item.id) ?? ""); + context.planDeltas.delete(payload.item.id); // A finished proposal stays active until Implement consumes it. const artifacts = yield* buildProposedPlanArtifacts({ context, @@ -5410,6 +5423,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi if (current !== input.context) { return false; } + input.context.planDeltas.clear(); const providerRetry = yield* Ref.modify(providerRetries, (current) => { const retry = current.get(input.context.providerTurnId); if (retry === undefined) { @@ -6293,6 +6307,19 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi Effect.gen(function* () { const nativeThreadId = yield* getNativeThreadId(unloadInput.providerThread); yield* client.request("thread/unsubscribe", { threadId: nativeThreadId }); + yield* releaseThreadState(nativeThreadId); + for (const [childId, child] of yield* Ref.get(subagentThreads)) { + for ( + let owner: ActiveCodexTurnContext | undefined = child.parentContext; + owner !== undefined; + owner = owner.subagent?.parentContext + ) { + if (owner.providerThread.id === unloadInput.providerThread.id) { + yield* releaseThreadState(childId); + break; + } + } + } }).pipe( Effect.mapError((cause) => cause._tag === "ProviderAdapterProtocolError" From 3f8150b5b8f8e4b20227650415c88bd99df29f39 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Sun, 4 Oct 2026 19:22:18 +0200 Subject: [PATCH 3/9] fix(server): retain and retry stranded OpenCode2 steers Keep stranded steers and unfinished revert work visible to idle release. Cancel retained inbox entries before unload and retain them when cancellation fails; release the session-owned state after successful cleanup. Implemented with gpt-6.1-sol through the Codex harness (initial implementation). --- .../Adapters/OpenCode2AdapterV2.test.ts | 294 ++++++++++++------ .../Adapters/OpenCode2AdapterV2.ts | 80 +++-- 2 files changed, 263 insertions(+), 111 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 46792897e789..b02a89081b42 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -2947,74 +2947,89 @@ describe("OpenCode2 adapter", () => { }).pipe(Effect.scoped), ); - it.effect("clears a staged revert whose commit failed, before the next prompt too", () => - Effect.gen(function* () { - const kept = "msg_t3_turn_run-attempt:kept:1"; - const dropped = "msg_t3_turn_run-attempt:dropped:1"; - const { runtime, thread } = yield* resumed([ - out("message.list", ""), - reply("message.list", { - data: [ - { id: dropped, time: { created: 2 }, text: "second", type: "user" }, - { id: kept, time: { created: 1 }, text: "first", type: "user" }, - ], - cursor: {}, - }), - out("session.revert.stage", { sessionID: SESSION, messageID: dropped, files: false }), - replyData("session.revert.stage", { messageID: dropped, files: [] }), - out("session.revert.commit", { sessionID: SESSION }), - reply("session.revert.commit", { - status: 500, - body: { _tag: "UnknownError", message: "disk I/O error" }, - }), - // Uncleared, OpenCode would commit the stage on the next prompt. - out("session.revert.clear", { sessionID: SESSION }), - reply("session.revert.clear", { - status: 500, - body: { _tag: "UnknownError", message: "busy" }, - }), - // The next turn clears it first, then waits out the empty execution `clear` runs. - out("session.revert.clear", { sessionID: SESSION }), - reply("session.revert.clear", null), - event("session.execution.started", { sessionID: SESSION }), - event("session.execution.succeeded", { sessionID: SESSION }), - out("session.prompt", { sessionID: SESSION, id: "", text: "" }), - promptAccepted, - event("session.execution.started", { sessionID: SESSION }), - event("session.execution.succeeded", { sessionID: SESSION }), - ]); - const now = yield* DateTime.now; - const turn = (attempt: string, ordinal: number) => ({ - id: ProviderTurnId.make(`provider-turn:${attempt}`), - providerThreadId: thread.id, - nodeId: NodeId.make(`node:${attempt}`), - runAttemptId: RunAttemptId.make(`run-attempt:${attempt}:1`), - nativeTurnRef: { - driver: OPENCODE_PROVIDER, - nativeId: `msg_t3_turn_run-attempt:${attempt}:1`, - strength: "weak" as const, - }, - ordinal, - status: "completed" as const, - startedAt: now, - completedAt: now, - }); - const rollback = yield* runtime - .rollbackThread({ - providerThread: thread, - target: { - type: "provider_turn", - checkpointId: CheckpointId.make("checkpoint:kept"), - appRunOrdinal: 1, - providerTurn: turn("kept", 1), + it.effect.each([false, true])( + "clears a failed staged revert before the next prompt, unload=%s", + (unload) => + Effect.gen(function* () { + const kept = "msg_t3_turn_run-attempt:kept:1"; + const dropped = "msg_t3_turn_run-attempt:dropped:1"; + const { runtime, thread } = yield* resumed([ + out("message.list", ""), + reply("message.list", { + data: [ + { id: dropped, time: { created: 2 }, text: "second", type: "user" }, + { id: kept, time: { created: 1 }, text: "first", type: "user" }, + ], + cursor: {}, + }), + out("session.revert.stage", { sessionID: SESSION, messageID: dropped, files: false }), + replyData("session.revert.stage", { messageID: dropped, files: [] }), + out("session.revert.commit", { sessionID: SESSION }), + reply("session.revert.commit", { + status: 500, + body: { _tag: "UnknownError", message: "disk I/O error" }, + }), + // Uncleared, OpenCode would commit the stage on the next prompt. + out("session.revert.clear", { sessionID: SESSION }), + reply("session.revert.clear", { + status: 500, + body: { _tag: "UnknownError", message: "busy" }, + }), + // The next turn clears it first, then waits out the empty execution `clear` runs. + out("session.revert.clear", { sessionID: SESSION }), + reply("session.revert.clear", null), + event("session.execution.started", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: SESSION }), + ...(unload + ? [ + out("session.get", { sessionID: SESSION }), + replyData("session.get", sessionInfo()), + ...noOpenRequests, + ] + : []), + out("session.prompt", { sessionID: SESSION, id: "", text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: SESSION }), + ]); + const now = yield* DateTime.now; + const turn = (attempt: string, ordinal: number) => ({ + id: ProviderTurnId.make(`provider-turn:${attempt}`), + providerThreadId: thread.id, + nodeId: NodeId.make(`node:${attempt}`), + runAttemptId: RunAttemptId.make(`run-attempt:${attempt}:1`), + nativeTurnRef: { + driver: OPENCODE_PROVIDER, + nativeId: `msg_t3_turn_run-attempt:${attempt}:1`, + strength: "weak" as const, }, - providerThreadTurns: [turn("kept", 1), turn("dropped", 2)], - }) - .pipe(Effect.exit); - assert.isTrue(Exit.isFailure(rollback)); - yield* runtime.startTurn(turnInput(thread)); - assert.equal((yield* terminalOf(runtime))?.status, "completed"); - }).pipe(Effect.scoped), + ordinal, + status: "completed" as const, + startedAt: now, + completedAt: now, + }); + const rollback = yield* runtime + .rollbackThread({ + providerThread: thread, + target: { + type: "provider_turn", + checkpointId: CheckpointId.make("checkpoint:kept"), + appRunOrdinal: 1, + providerTurn: turn("kept", 1), + }, + providerThreadTurns: [turn("kept", 1), turn("dropped", 2)], + }) + .pipe(Effect.exit); + assert.isTrue(Exit.isFailure(rollback)); + assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + if (unload) { + yield* runtime.unloadThread!({ providerThread: thread }); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + yield* runtime.resumeThread({ providerThread: thread }); + } + yield* runtime.startTurn(turnInput(thread)); + assert.equal((yield* terminalOf(runtime))?.status, "completed"); + }).pipe(Effect.scoped), ); it.effect("offers no follow-up for the execution a cleared revert starts", () => @@ -3257,10 +3272,113 @@ describe("OpenCode2 adapter", () => { }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), ); - it.effect("takes back a stranded steer again when its first cancel failed", () => + it.effect.each([false, true])( + "retries a stranded steer after cancellation fails, unload=%s", + (unload) => + Effect.gen(function* () { + const steerId = `msg_t3_steer_${SESSION}:message:opencode2-adapter:steer`; + const cancelOut = out("session.inbox.cancel", { sessionID: SESSION, inboxID: steerId }); + const { runtime, thread } = yield* resumed([ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + out("session.prompt", { + sessionID: SESSION, + id: steerId, + text: "", + delivery: "steer", + }), + replyData("session.prompt", { + id: steerId, + sessionID: SESSION, + time: { created: 1790656601500 }, + type: "user", + payload: { text: "Also say STEERED." }, + delivery: "steer", + }), + // Stopped before the steer was read: it stays in OpenCode's inbox. + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: SESSION }), + // The next turn's cancel fails, so the turn after it tries again. + cancelOut, + reply("session.inbox.cancel", { + status: 500, + body: { _tag: "UnknownError", message: "busy" }, + }), + ...(unload + ? [ + out("session.get", { sessionID: SESSION }), + replyData("session.get", sessionInfo()), + cancelOut, + reply("session.inbox.cancel", null), + ] + : []), + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: SESSION }), + ...(unload ? [] : [cancelOut, reply("session.inbox.cancel", null)]), + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: SESSION }), + ]); + const seen: Array = []; + const ended = yield* Deferred.make(); + const nextEnded = yield* Deferred.make(); + const thirdEnded = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.tap((event) => + Effect.gen(function* () { + if (event.type !== "turn.terminal") return; + seen.push(event.status); + if (seen.length === 1) yield* Deferred.succeed(ended, undefined); + if (seen.length === 2) yield* Deferred.succeed(nextEnded, undefined); + if (seen.length === 3) yield* Deferred.succeed(thirdEnded, undefined); + }), + ), + Stream.runDrain, + Effect.forkScoped, + ); + yield* runtime.startTurn(turnInput(thread)); + yield* runtime.steerTurn({ + threadId, + runId: RunId.make("run:opencode2-adapter"), + providerThread: thread, + providerTurnId: yield* providerTurnId, + message: { + ...turnInput(thread).message, + messageId: MessageId.make("message:opencode2-adapter:steer"), + text: "Also say STEERED.", + }, + }); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + }); + yield* Deferred.await(ended); + if (unload) { + yield* runtime.unloadThread!({ providerThread: thread }); + assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + yield* runtime.resumeThread({ providerThread: thread }); + } + yield* runtime.startTurn(secondTurn(thread)); + yield* Deferred.await(nextEnded); + yield* runtime.startTurn({ + ...secondTurn(thread), + runId: RunId.make("run:opencode2-adapter:3"), + runOrdinal: 3, + providerTurnOrdinal: 3, + attemptId: RunAttemptId.make("attempt:opencode2-adapter:3"), + }); + yield* Deferred.await(thirdEnded); + }).pipe(Effect.scoped), + ); + + it.effect("takes back a stranded steer before unloading and reattaching its session", () => Effect.gen(function* () { const steerId = `msg_t3_steer_${SESSION}:message:opencode2-adapter:steer`; - const cancelOut = out("session.inbox.cancel", { sessionID: SESSION, inboxID: steerId }); const { runtime, thread } = yield* resumed([ out("session.prompt", { sessionID: SESSION, text: "" }), promptAccepted, @@ -3279,30 +3397,24 @@ describe("OpenCode2 adapter", () => { payload: { text: "Also say STEERED." }, delivery: "steer", }), - // Stopped before the steer was read: it stays in OpenCode's inbox. out("session.interrupt", { sessionID: SESSION }), reply("session.interrupt", { interrupted: true }), event("session.execution.interrupted", { sessionID: SESSION }), - // The next turn's cancel fails, so the turn after it tries again. - cancelOut, - reply("session.inbox.cancel", { - status: 500, - body: { _tag: "UnknownError", message: "busy" }, - }), - out("session.prompt", { sessionID: SESSION, text: "" }), - promptAccepted, - event("session.execution.started", { sessionID: SESSION }), - event("session.execution.succeeded", { sessionID: SESSION }), - cancelOut, + // Unload must cancel this inbox entry before discarding its id. + out("session.inbox.cancel", { sessionID: SESSION, inboxID: steerId }), reply("session.inbox.cancel", null), - out("session.prompt", { sessionID: SESSION, text: "" }), - promptAccepted, - event("session.execution.started", { sessionID: SESSION }), - event("session.execution.succeeded", { sessionID: SESSION }), + out("session.get", { sessionID: SESSION }), + replyData("session.get", sessionInfo()), + ...noOpenRequests, + ...withInstructions([ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: SESSION }), + ]), ]); - const seen: Array = []; const ended = yield* Deferred.make(); - const nextEnded = yield* Deferred.make(); +const nextEnded = yield* Deferred.make(); // One consumer reads every terminal; a second reader would race it for the queue. const thirdEnded = yield* Deferred.make(); yield* runtime.events.pipe( @@ -3311,7 +3423,7 @@ describe("OpenCode2 adapter", () => { if (event.type !== "turn.terminal") return; seen.push(event.status); if (seen.length === 1) yield* Deferred.succeed(ended, undefined); - if (seen.length === 2) yield* Deferred.succeed(nextEnded, undefined); + if (seen.length === 2) yield* Deferred.succeed(nextEnded, event.status); if (seen.length === 3) yield* Deferred.succeed(thirdEnded, undefined); }), ), @@ -3335,8 +3447,12 @@ describe("OpenCode2 adapter", () => { providerTurnId: yield* providerTurnId, }); yield* Deferred.await(ended); + yield* runtime.unloadThread!({ providerThread: thread }); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + yield* runtime.resumeThread({ providerThread: thread }); yield* runtime.startTurn(secondTurn(thread)); - yield* Deferred.await(nextEnded); +// The replay rejects any stale steer or extra cancellation on the reattached turn. + assert.equal(yield* Deferred.await(nextEnded), "completed"); yield* runtime.startTurn({ ...secondTurn(thread), runId: RunId.make("run:opencode2-adapter:3"), diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts index 7716b75baaf9..20ca3c52a901 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts @@ -904,13 +904,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const stagedReverts = new Set(); // Starting a turn and cutting the history take turns on a session: each // checks that the other is not running before its own requests yield. - const sessionGates = yield* KeyedLock.make(); +const sessionGates = yield* KeyedLock.make(); const exclusive = (providerThread: OrchestrationV2ProviderThread) => (effect: Effect.Effect) => { const sessionId = providerThread.nativeThreadRef?.nativeId; if (sessionId == null) return effect; - return sessionGates.withLock(sessionId, effect); +return sessionGates.withLock(sessionId, effect); }; const emit = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); @@ -3564,7 +3564,15 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr events: Stream.fromQueue(events), // A background subagent keeps its session busy after its parent's turn, // and a held wake still needs its turn: idle release must wait for both. - hasPendingBackgroundWork: Effect.sync(() => [...threads.values()].some(owesWork)), + hasPendingBackgroundWork: Effect.sync(() => + [...threads.values()].some( + (state) => + owesWork(state) || + state.strandedSteers.size > 0 || + stagedReverts.has(state.sessionId) || + clearing.has(state.sessionId), + ), + ), hasPendingBackgroundWorkForThread: (providerThread) => Effect.sync(() => { const nativeId = providerThread.nativeThreadRef?.nativeId; @@ -3985,25 +3993,53 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ), ), unloadThread: ({ providerThread }) => - Effect.gen(function* () { - const nativeId = providerThread.nativeThreadRef?.nativeId; - const state = nativeId == null ? undefined : threads.get(nativeId); - if ( - nativeId == null || - state === undefined || - state.active !== undefined || - hasBackground(state) - ) { - return; - } - threads.delete(nativeId); - for (const [child, owner] of childOwners) { - if (owner !== state) continue; - childOwners.delete(child); - threads.delete(child); - } - if (state.mcp !== undefined) yield* removeMcp(state.mcp); - }), + exclusive(providerThread)( + Effect.gen(function* () { + const nativeId = providerThread.nativeThreadRef?.nativeId; + const state = nativeId == null ? undefined : threads.get(nativeId); + if ( + nativeId == null || + state === undefined || + state.active !== undefined || + hasBackground(state) + ) { + return; + } + const owned = new Set(sessionsOf(state).map((child) => child.sessionId)); + for (const [child, owner] of childOwners) { + if (owned.has(owner.sessionId)) owned.add(child); + } + for (const id of owned) { + const child = threads.get(id); + if (child !== undefined) { + yield* cancelStrandedSteers(child); + // A failed cancellation still owns an inbox entry. Retain it for retry. + if (child.strandedSteers.size > 0) return; + } + if (stagedReverts.has(id)) yield* clearRevert(id); + } + for (const id of owned) { + const child = threads.get(id); + threads.delete(id); + childOwners.delete(id); + announced.delete(id); + busy.delete(id); + stagedReverts.delete(id); + clearing.delete(id); + if (child?.mcp !== undefined) yield* removeMcp(child.mcp); + } + }), + ).pipe( + Effect.mapError((cause) => + isProviderAdapterError(cause) + ? cause + : new ProviderAdapter.ProviderAdapterProtocolError({ + driver, + detail: "Could not release OpenCode thread state.", + cause, + }), + ), + ), respondToRuntimeRequest: (requestInput) => Effect.gen(function* () { const entry = pending.get(requestInput.requestId); From 0e9d7c5e87a52c6b26149e2d5089371c6aa90607 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Sun, 4 Oct 2026 19:22:46 +0200 Subject: [PATCH 4/9] fix(server): reclaim OpenCode1 state with its native session Release the native session and its owned descendant state on unload without changing the adapter-specific ownership model. The write-only `messages` map is gone; nothing read it. `session.deleted` now cancels active turns instead of only clearing bookkeeping, so a turn cannot hang on a session OpenCode has already deleted. Implemented with gpt-6.1-sol through the Codex harness (initial implementation). --- .../Adapters/OpenCodeAdapterV2.test.ts | 111 ++++++++++++++++++ .../Adapters/OpenCodeAdapterV2.ts | 78 ++++++++++-- 2 files changed, 181 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.test.ts index 44632e02c66d..3d063eeb60e5 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.test.ts @@ -237,6 +237,117 @@ const makeOpenCodeRuntimeHarness = Effect.fn("makeOpenCodeRuntimeHarness")(funct }); describe("OpenCodeAdapterV2", () => { + it.effect.each(["unload", "deleted"] as const)( + "releases idle thread state on %s and reloads it on resume", + (release) => + Effect.gen(function* () { + const nativeEvents = asyncEventStream(); + const nativeSession = { id: "root", time: { created: 1, updated: 1 } }; + const harness = yield* makeOpenCodeRuntimeHarness(`release-${release}`, "root", { + event: { subscribe: async () => ({ stream: nativeEvents.stream }) }, + session: { + create: async () => ({ data: nativeSession }), + get: async () => ({ data: nativeSession }), + promptAsync: async () => ({ data: true }), + }, + }); + if (release === "unload") { + assert.isDefined(harness.runtime.unloadThread); + yield* harness.runtime.unloadThread!({ providerThread: harness.providerThread }); + } else { + yield* Effect.promise(() => + nativeEvents.push({ type: "session.deleted", properties: { info: nativeSession } }), + ); + } + const failure = yield* harness.startTurn().pipe(Effect.flip); + assert.equal(failure._tag, "ProviderAdapterTurnStartError"); + yield* harness.runtime.resumeThread({ providerThread: harness.providerThread }); + yield* harness.startTurn(); + }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ); + + it.effect("keeps an idle root loaded while a related session is busy", () => + Effect.gen(function* () { + const nativeEvents = asyncEventStream(); + const harness = yield* makeOpenCodeRuntimeHarness("release-busy-child", "root", { + event: { subscribe: async () => ({ stream: nativeEvents.stream }) }, + session: { + create: async () => ({ data: { id: "root", time: { created: 1, updated: 1 } } }), + promptAsync: async () => ({ data: true }), + }, + }); + yield* Effect.promise(() => + nativeEvents.push({ + type: "session.created", + properties: { + info: { id: "child", parentID: "root", time: { created: 1, updated: 1 } }, + }, + }), + ); + yield* Effect.promise(() => + nativeEvents.push({ + type: "session.status", + properties: { + sessionID: "child", + status: { type: "busy" }, + }, + }), + ); + yield* harness.runtime.unloadThread!({ providerThread: harness.providerThread }); + assert.isTrue(yield* harness.runtime.hasPendingBackgroundWork!); + yield* harness.startTurn(); + yield* harness.runtime.unloadThread!({ providerThread: harness.providerThread }); + // Active turns survive a detach request too. + const activeError = yield* harness.startTurn().pipe(Effect.flip); + assert.equal(activeError._tag, "ProviderAdapterTurnStartError"); + }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ); + + it.effect("cancels a deleted session's active turn and clears descendant busy state", () => + Effect.gen(function* () { + const nativeEvents = asyncEventStream(); + const nativeSession = { id: "root", time: { created: 1, updated: 1 } }; + const harness = yield* makeOpenCodeRuntimeHarness("delete-active", "root", { + event: { subscribe: async () => ({ stream: nativeEvents.stream }) }, + session: { + create: async () => ({ data: nativeSession }), + promptAsync: async () => ({ data: true }), + }, + }); + yield* harness.startTurn(); + const terminal = yield* harness.runtime.events.pipe( + Stream.filter((event) => event.type === "turn.terminal"), + Stream.runHead, + Effect.forkScoped, + ); + yield* Effect.promise(() => + nativeEvents.push({ + type: "session.created", + properties: { + info: { id: "child", parentID: "root", time: { created: 1, updated: 1 } }, + }, + }), + ); + yield* Effect.promise(() => + nativeEvents.push({ + type: "session.status", + properties: { sessionID: "child", status: { type: "busy" } }, + }), + ); + yield* Effect.promise(() => + nativeEvents.push({ type: "session.deleted", properties: { info: nativeSession } }), + ); + const ended = Option.getOrThrow(yield* Fiber.join(terminal)); + assert.equal(ended.type, "turn.terminal"); + if (ended.type === "turn.terminal") assert.equal(ended.status, "cancelled"); + assert.isFalse(yield* harness.runtime.hasPendingBackgroundWork!); + assert.equal( + (yield* harness.startTurn().pipe(Effect.flip))._tag, + "ProviderAdapterTurnStartError", + ); + }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ); + it.effect.each(["completed", "failed", "unresolved", "unavailable", "reconnect"] as const)( "normalizes OpenCode step usage for %s turns", (ending) => diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts index aad31d636325..2ccc91260dee 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts @@ -386,7 +386,6 @@ interface OpenCodeThreadState { appThread: OrchestrationV2AppThread | null; activeTurn: ActiveOpenCodeTurn | null; readonly providerTurns: Map; - readonly messages: Map; readonly runtimeRequests: Map; readonly messageRoles: Map; readonly userMessageIds: Array; @@ -1258,7 +1257,6 @@ export function makeOpenCodeAdapterV2( createdAt: startedAt, updatedAt: emittedAt, }; - state.messages.set(String(message.id), message); yield* emitProviderEvent({ type: "message.updated", driver: OPENCODE_PROVIDER, @@ -1429,7 +1427,6 @@ export function makeOpenCodeAdapterV2( appThread: childThread, activeTurn: null, providerTurns: new Map(), - messages: new Map(), runtimeRequests: new Map(), messageRoles: new Map(), userMessageIds: [], @@ -2336,7 +2333,6 @@ export function makeOpenCodeAdapterV2( createdAt: turn.startedAt, updatedAt: now, }; - state.messages.set(String(messageId), projected); yield* emitProviderEvent({ type: "message.updated", driver: OPENCODE_PROVIDER, @@ -2550,6 +2546,55 @@ export function makeOpenCodeAdapterV2( } }); + const sessionsOwnedBy = (sessionId: string) => { + const owned = new Set([sessionId]); + let changed = true; + while (changed) { + changed = false; + for (const [id, state] of threads) { + const parentId = state.parentSubagent?.parentState.nativeSessionId; + if (!owned.has(id) && parentId !== undefined && owned.has(parentId)) { + owned.add(id); + changed = true; + } + } + for (const [id, owner] of relatedSessionOwners) { + if (!owned.has(id) && owned.has(owner.nativeSessionId)) { + owned.add(id); + changed = true; + } + } + } + return owned; + }; + + // Thread-owned state is released together, after its native work has stopped. + const releaseSessions = Effect.fnUntraced(function* (sessionIds: ReadonlySet) { + for (const request of Array.from(pendingRequests.values())) { + if (sessionIds.has(request.nativeSessionId)) { + rememberSettledRequest(request.nativeRequestId); + yield* resolveRuntimeRequest(request.nativeRequestId, "cancelled"); + } + } + for (const [id, subagent] of subagentsByNativeItemId) { + if ( + sessionIds.has(subagent.parentState.nativeSessionId) || + (subagent.childSessionId !== null && sessionIds.has(subagent.childSessionId)) + ) { + subagentsByNativeItemId.delete(id); + if (subagent.childSessionId !== null) + subagentsByChildSessionId.delete(subagent.childSessionId); + } + } + for (const id of sessionIds) { + for (const controller of commandControllers.get(id) ?? []) controller.abort(); + commandControllers.delete(id); + threads.delete(id); + relatedSessionOwners.delete(id); + busySessionIds.delete(id); + } + }); + const handleEvent = Effect.fnUntraced(function* (event: OpenCodeEvent) { yield* logProtocolEvent({ direction: "incoming", @@ -2622,10 +2667,17 @@ export function makeOpenCodeAdapterV2( } return; } - case "session.deleted": - relatedSessionOwners.delete(event.properties.info.id); - busySessionIds.delete(event.properties.info.id); + case "session.deleted": { + const owned = sessionsOwnedBy(event.properties.info.id); + for (const id of owned) { + const state = threads.get(id); + if (state?.activeTurn != null) { + yield* finalizeTurn(state, state.activeTurn, "cancelled"); + } + } + yield* releaseSessions(owned); return; + } case "session.status": { const sessionId = event.properties.sessionID; switch (event.properties.status.type) { @@ -2834,7 +2886,6 @@ export function makeOpenCodeAdapterV2( appThread, activeTurn: null, providerTurns: new Map(), - messages: new Map(), runtimeRequests: new Map(), messageRoles: new Map(), userMessageIds: [], @@ -3490,6 +3541,17 @@ export function makeOpenCodeAdapterV2( }), ), ), + unloadThread: ({ providerThread }) => + Effect.gen(function* () { + const owned = sessionsOwnedBy(nativeThreadId(providerThread)); + if ( + Array.from(owned).some( + (id) => threads.get(id)?.activeTurn != null || busySessionIds.has(id), + ) + ) + return; + yield* releaseSessions(owned); + }), respondToRuntimeRequest: (requestInput) => Effect.gen(function* () { const pending = pendingRequests.get(String(requestInput.requestId)); From 54eb0c2cc597293884b924d582ca872da2435860 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Sun, 4 Oct 2026 19:23:03 +0200 Subject: [PATCH 5/9] fix(server): release Claude wake state on native process transitions Clear wake state when its native process transitions. The wakeBuffers audit is only partially closed: clearWakeStateForNativeThread releases the entry on every native-process transition but adds no bound; openQuery sibling-close keeps the count small. Implemented with gpt-6.1-sol through the Codex harness (initial implementation). --- .../Adapters/ClaudeAdapterV2.test.ts | 120 +++++++++++++++++- .../Adapters/ClaudeAdapterV2.ts | 15 +++ 2 files changed, 132 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 785c4b1c9ec1..fb4714457227 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -44,6 +44,7 @@ import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { Tool } from "effect/ai"; import { formatClaudeResumeCompactionQuestion } from "@t3tools/shared/claudeCompaction"; import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; @@ -2085,6 +2086,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { readonly close?: (sdkMessages: Queue.Queue) => Effect.Effect; readonly interrupt?: Effect.Effect; readonly environment?: NodeJS.ProcessEnv; + readonly freshMessageQueue?: boolean; }) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -2092,7 +2094,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { const attachmentsDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-claude-v2-wake-", }); - const sdkMessages = yield* Queue.unbounded(); + let sdkMessages = yield* Queue.unbounded(); const processedMessages = new WeakMap>(); const offerAndWait = Effect.fnUntraced(function* (message: SDKMessage) { const processed = yield* Deferred.make(); @@ -2126,7 +2128,8 @@ describe("ClaudeAdapterV2 background wake turns", () => { queryRunner: { allocateSessionId: Effect.succeed(WAKE_NATIVE_SESSION), open: (input) => - Effect.sync(() => { + Effect.gen(function* () { + if (options?.freshMessageQueue) sdkMessages = yield* Queue.unbounded(); openedOptions = input.options; return { messages: Stream.fromQueue(sdkMessages).pipe( @@ -2204,7 +2207,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { runtime, providerThread, threadId, - sdkMessages, + get sdkMessages() { + return sdkMessages; + }, offerAndWait, offeredMessages, permissionModeChanges, @@ -4996,6 +5001,115 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect("discards held plans when an interrupt finalizes the turn before its echo", () => + Effect.gen(function* () { + const interrupted = yield* Deferred.make(); + const harness = yield* makeWakeHarnessWithOptions({ + freshMessageQueue: true, + interrupt: Deferred.succeed(interrupted, undefined).pipe(Effect.asVoid), + }); + const now = yield* DateTime.now; + const firstAttempt = RunAttemptId.make("held-plan-first"); + const secondAttempt = RunAttemptId.make("held-plan-second"); + const thirdAttempt = RunAttemptId.make("held-plan-third"); + const stamp = (frame: SDKMessage, attempt: RunAttemptId) => + claudeSdkFrame({ + ...frame, + user_message_uuid: ClaudeAdapterV2.claudePromptUuid(attempt), + }); + const start = (attemptId: RunAttemptId, providerTurnOrdinal: number) => + harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId, + providerTurnOrdinal, + text: "Plan the task.", + attachments: [], + }), + ); + yield* start(firstAttempt, 1); + yield* harness.offerAndWait( + stamp( + makeAssistantTextFrame({ uuid: "00000000-0000-4000-8000-000000000801", text: "One." }), + firstAttempt, + ), + ); + yield* harness.offerAndWait( + stamp( + makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000802", result: "One." }), + firstAttempt, + ), + ); + yield* Queue.take(harness.terminalReceipts); + yield* start(secondAttempt, 2); + const toolUseId = "held-exit-plan"; + const planFrame = claudeSdkFrame({ + type: "assistant", + message: { + model: "claude-sonnet-4-6", + id: "msg_held_plan", + type: "message", + role: "assistant", + content: [{ type: "tool_use", id: toolUseId, name: "ExitPlanMode", input: {} }], + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000803", + session_id: WAKE_NATIVE_SESSION, + }); + yield* harness.offerAndWait(planFrame); + const canUseTool = harness.getOpenedOptions()?.canUseTool; + assert.isFunction(canUseTool); + const signal = yield* Effect.abortSignal; + yield* Effect.promise(() => + canUseTool!( + "ExitPlanMode", + { plan: "OLD PLAN" }, + { + signal, + toolUseID: toolUseId, + requestId: "held-plan-request", + }, + ), + ); + const active = harness.events.findLast( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "running", + ); + assert.equal(active?.type, "provider_turn.updated"); + if (active?.type !== "provider_turn.updated") return; + const stopping = yield* harness.runtime + .interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: active.providerTurn.id, + }) + .pipe(Effect.forkScoped); + yield* Deferred.await(interrupted); + // This process never closes; exercise the adapter's forced-finalization path. + yield* TestClock.adjust("10 seconds"); + yield* Fiber.join(stopping); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "interrupted"); + yield* start(thirdAttempt, 3); + yield* harness.offerAndWait(stamp(planFrame, thirdAttempt)); + yield* harness.offerAndWait( + stamp( + makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000804", result: "Three." }), + thirdAttempt, + ), + ); + yield* Queue.take(harness.terminalReceipts); + assert.isFalse( + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "proposed_plan" && + event.turnItem.markdown === "OLD PLAN", + ), + ); + }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ); + it.effect("buffers wake output and requests a single continuation run", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 9af1c0efde4b..c8a9c7f84aeb 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -3655,16 +3655,29 @@ export function makeClaudeAdapterV2( yield* endClaudeMonitorTasks((_taskId, task) => task.nativeThreadId === nativeThreadId); }); + const clearHeldPlansForFrames = (frames: ReadonlyArray) => { + for (const frame of frames) { + for (const toolUse of claudeToolUseBlocksFromAssistantMessage(frame)) { + heldProposedPlansByToolUseId.delete(toolUse.id); + } + } + }; // Drop idle wake traffic for a dead native process so it cannot pin // session-wide pending work after sibling query replacement. const clearWakeStateForNativeThread = (nativeThreadId: string) => Effect.gen(function* () { yield* clearWakeReports(nativeThreadId); + yield* Ref.update(wakeReportsByNativeThread, (current) => { + const userTurns = new Map(current.userTurns); + userTurns.delete(nativeThreadId); + return { userTurns, reports: current.reports }; + }); yield* Ref.update(wakeBuffers, (current) => { if (!current.has(nativeThreadId)) { return current; } const updated = new Map(current); + clearHeldPlansForFrames(current.get(nativeThreadId)?.messages ?? []); updated.delete(nativeThreadId); return updated; }); @@ -4902,6 +4915,8 @@ export function makeClaudeAdapterV2( readonly threadDisposition?: "reusable" | "broken"; readonly result?: SDKResultMessage; }) { + clearHeldPlansForFrames(input.context.heldRootFrames); + input.context.heldRootFrames.length = 0; yield* reasoningDeltas.flushTurn(input.context.nativeTurnId); for (const toolCall of input.context.toolCalls.values()) { const artifacts = buildToolCallArtifacts({ From dae8807bc68f85e80818b7b8ccc3b4de7048eaf8 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Sun, 4 Oct 2026 19:52:53 +0200 Subject: [PATCH 6/9] fix(server): tighten the Pi approval key and the OpenCode2 unload test JSON.stringify over the sorted entries is byte-identical for an already-parsed event and keeps a Schema round-trip out of a hot prompt path. Sort by code unit: locale collation should not decide cache identity. The stranded-steer unload test could only pass because unloadThread early-returned; it now supplies noOpenRequests for the unload arm and asserts the retry actually releases the failed cancellation. Drops the second sessionGates eviction rule, since the ensuring-decrement already covers it. Implemented with opencode/space-bunny-free through the OpenCode harness (audit fixes). --- .../Adapters/OpenCode2AdapterV2.test.ts | 22 ++++++++++++------- .../orchestration-v2/Adapters/PiAdapterV2.ts | 6 ++--- 2 files changed, 17 insertions(+), 11 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index b02a89081b42..b85663b950fb 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -3278,6 +3278,12 @@ describe("OpenCode2 adapter", () => { Effect.gen(function* () { const steerId = `msg_t3_steer_${SESSION}:message:opencode2-adapter:steer`; const cancelOut = out("session.inbox.cancel", { sessionID: SESSION, inboxID: steerId }); + const nextPrompt = [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: SESSION }), + ]; const { runtime, thread } = yield* resumed([ out("session.prompt", { sessionID: SESSION, text: "" }), promptAccepted, @@ -3300,7 +3306,7 @@ describe("OpenCode2 adapter", () => { out("session.interrupt", { sessionID: SESSION }), reply("session.interrupt", { interrupted: true }), event("session.execution.interrupted", { sessionID: SESSION }), - // The next turn's cancel fails, so the turn after it tries again. + // Cancellation fails on the first unload or next turn, then is retried. cancelOut, reply("session.inbox.cancel", { status: 500, @@ -3308,16 +3314,14 @@ describe("OpenCode2 adapter", () => { }), ...(unload ? [ - out("session.get", { sessionID: SESSION }), - replyData("session.get", sessionInfo()), cancelOut, reply("session.inbox.cancel", null), + out("session.get", { sessionID: SESSION }), + replyData("session.get", sessionInfo()), + ...noOpenRequests, ] : []), - out("session.prompt", { sessionID: SESSION, text: "" }), - promptAccepted, - event("session.execution.started", { sessionID: SESSION }), - event("session.execution.succeeded", { sessionID: SESSION }), + ...(unload ? withInstructions(nextPrompt) : nextPrompt), ...(unload ? [] : [cancelOut, reply("session.inbox.cancel", null)]), out("session.prompt", { sessionID: SESSION, text: "" }), promptAccepted, @@ -3360,7 +3364,9 @@ describe("OpenCode2 adapter", () => { yield* Deferred.await(ended); if (unload) { yield* runtime.unloadThread!({ providerThread: thread }); - assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + // The first unload retained the failed cancellation; the retry must release it. + yield* runtime.unloadThread!({ providerThread: thread }); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); yield* runtime.resumeThread({ providerThread: thread }); } yield* runtime.startTurn(secondTurn(thread)); diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts index ca5cac259340..5be43c361676 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts @@ -99,7 +99,6 @@ export const PI_PROVIDER = ProviderDriverKind.make("pi"); const PI_DRIVER_KIND = PI_PROVIDER; const PI_DEFAULT_INSTANCE_ID = defaultInstanceIdForDriver(PI_DRIVER_KIND); const DEFAULT_PI_SETTINGS = Schema.decodeSync(PiSettings)({}); -const encodeApprovalKey = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); /** * Sentinel model slug meaning "do not call set_model": Pi resolves the model @@ -1215,10 +1214,11 @@ export function makePiAdapterV2( } if (nativeRequestId === undefined) return; // Request ids change on repeats; every other supplied field describes the prompt. - const approvalKey = encodeApprovalKey( + // @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.localeCompare(right)), + .sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0)), ); if (method === "confirm" && sessionApprovals.has(approvalKey)) { yield* connection.send({ From d00d7e9e3df3b4e75bc2bcfa97f2690b1ab5a12b Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Sun, 4 Oct 2026 21:47:23 +0200 Subject: [PATCH 7/9] fix(server): drop cached Pi approvals when pi switched sessions Precautionary, not a fix for a confirmed defect. Pi is an external CLI, so its command set cannot be read from this repo: the adapter forwards a leading "/" so pi's own slash commands keep expanding, and a /new or /resume there would change the native session inside pi, with no adapter command and no event to observe. Cached session approvals would survive that switch into a conversation the user never authorized, reopening the hole the rollback and registration boundaries closed. Turn finalize now compares the session file pi reports against the one the thread is bound to and clears the approvals when they differ, and also when the state cannot be read, because an unknown identity is not evidence of an unchanged one. It costs one get_state per turn, and only when an approval is actually cached. Implemented with opencode/space-bunny-free through the OpenCode harness (audit fixes). --- .../Adapters/PiAdapterV2.test.ts | 61 +++++++++++++++++++ .../orchestration-v2/Adapters/PiAdapterV2.ts | 26 ++++++++ 2 files changed, 87 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts index 26b31ccd62ac..d095a5b61f03 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts @@ -1693,6 +1693,67 @@ describe("PiAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(testLayer)), ); + 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: "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/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(testLayer)), + ); + 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 5be43c361676..46a26e1c1ee2 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts @@ -1412,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; @@ -1437,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 From a5b65dbc1853230a8ddbf4766389841d30e3ef60 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Wed, 7 Oct 2026 00:19:34 +0200 Subject: [PATCH 8/9] test(server): repair rebased adapter state-scope coverage The hand-rebase left three files that did not compile and a security test that had never run: - PiAdapterV2.test.ts referenced an undefined "testLayer"; the file defines "layerTest". All 8 rows of the session-approval boundary matrix and the mid-turn foreign-session test threw ReferenceError, so the Pi authorization fix had no executing test at all. - OpenCode2AdapterV2.test.ts gained a third startTurn on a transcript that scripts only two prompts, so it hung for 120s, and used "seen" without declaring it. Dropped the unscripted turn; kept the reattached-turn "completed" assertion that actually proves the stranded-steer fix. - CodexAdapterV2.ts had a third ActiveCodexTurnContext literal missing "planDeltas", so context.planDeltas was undefined on that path. Also documents the KeyedLock non-reentrancy constraint at "exclusive", and restores indentation at five sites the conflict resolution flattened. Claude's "discards held plans when an interrupt finalizes the turn before its echo" still fails and is left as-is: reaching the held-plan path needs the turn's own prompt echo to set promptEchoMode "early" first, which this harness does not reproduce. Fixing it needs a decision about the harness rather than another stamping attempt. Model: opencode/space-bunny-free (OpenCode) --- .../Adapters/CodexAdapterV2.test.ts | 2 +- .../orchestration-v2/Adapters/CodexAdapterV2.ts | 1 + .../Adapters/OpenCode2AdapterV2.test.ts | 16 +++------------- .../Adapters/OpenCode2AdapterV2.ts | 9 +++++++-- .../Adapters/PiAdapterV2.test.ts | 4 ++-- 5 files changed, 14 insertions(+), 18 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 7de904a25883..bbac14185dfb 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -6933,7 +6933,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); -it.effect.each(["failed", "malformed", "wrong child", "blank model"] as const)( + it.effect.each(["failed", "malformed", "wrong child", "blank model"] as const)( "resumes child metadata after a %s read", (readResult) => Effect.scoped( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 3a5b09679e5b..750e405429fd 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -5726,6 +5726,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi subagent: null, startedAt: now, itemPositions: new Map(), + planDeltas: new Map(), }; const providerTurn = { id: context.providerTurnId, diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index b85663b950fb..a88f2dc663a4 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -3420,9 +3420,8 @@ describe("OpenCode2 adapter", () => { ]), ]); const ended = yield* Deferred.make(); -const nextEnded = yield* Deferred.make(); - // One consumer reads every terminal; a second reader would race it for the queue. - const thirdEnded = yield* Deferred.make(); + const seen: Array = []; + const nextEnded = yield* Deferred.make(); yield* runtime.events.pipe( Stream.tap((event) => Effect.gen(function* () { @@ -3430,7 +3429,6 @@ const nextEnded = yield* Deferred.make(); seen.push(event.status); if (seen.length === 1) yield* Deferred.succeed(ended, undefined); if (seen.length === 2) yield* Deferred.succeed(nextEnded, event.status); - if (seen.length === 3) yield* Deferred.succeed(thirdEnded, undefined); }), ), Stream.runDrain, @@ -3457,16 +3455,8 @@ const nextEnded = yield* Deferred.make(); assert.isFalse(yield* runtime.hasPendingBackgroundWork!); yield* runtime.resumeThread({ providerThread: thread }); yield* runtime.startTurn(secondTurn(thread)); -// The replay rejects any stale steer or extra cancellation on the reattached turn. + // The replay rejects any stale steer or extra cancellation on the reattached turn. assert.equal(yield* Deferred.await(nextEnded), "completed"); - yield* runtime.startTurn({ - ...secondTurn(thread), - runId: RunId.make("run:opencode2-adapter:3"), - runOrdinal: 3, - providerTurnOrdinal: 3, - attemptId: RunAttemptId.make("attempt:opencode2-adapter:3"), - }); - yield* Deferred.await(thirdEnded); }).pipe(Effect.scoped), ); diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts index 20ca3c52a901..b22ffa3a0d4d 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts @@ -904,13 +904,18 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const stagedReverts = new Set(); // Starting a turn and cutting the history take turns on a session: each // checks that the other is not running before its own requests yield. -const sessionGates = yield* KeyedLock.make(); + const sessionGates = yield* KeyedLock.make(); + // `KeyedLock` is not reentrant: taking a key while already holding it on + // the same session deadlocks. Nothing called from inside an `exclusive` + // block may re-enter `exclusive` for that same session. The blocks below + // (startTurn, unloadThread, rollback, fork) only call helpers that issue + // their own requests directly; keep it that way. const exclusive = (providerThread: OrchestrationV2ProviderThread) => (effect: Effect.Effect) => { const sessionId = providerThread.nativeThreadRef?.nativeId; if (sessionId == null) return effect; -return sessionGates.withLock(sessionId, effect); + return sessionGates.withLock(sessionId, effect); }; const emit = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts index d095a5b61f03..4f48cc1fbcf7 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts @@ -1690,7 +1690,7 @@ describe("PiAdapterV2", () => { ) { assert.equal(asked.threadId, targetThreadId); } - }).pipe(Effect.scoped, Effect.provide(testLayer)), + }).pipe(Effect.scoped, Effect.provide(layerTest)), ); it.effect("drops session approvals when pi moved to another session mid-turn", () => @@ -1751,7 +1751,7 @@ describe("PiAdapterV2", () => { fake.takeRequest("extension_ui_response").pipe(Effect.as(false)), ); assert.isTrue(askedAgain); - }).pipe(Effect.scoped, Effect.provide(testLayer)), + }).pipe(Effect.scoped, Effect.provide(layerTest)), ); it.effect("offers an explicit empty value for extension input dialogs", () => From 26f4b636bfcb94dd5a0e8c6c5ddd701011b0a93f Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Wed, 7 Oct 2026 00:36:29 +0200 Subject: [PATCH 9/9] test(server): resolve prompt UUIDs in held-plan regression --- .../Adapters/ClaudeAdapterV2.test.ts | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index fb4714457227..9d1ee058b0cc 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -5012,11 +5012,12 @@ describe("ClaudeAdapterV2 background wake turns", () => { const firstAttempt = RunAttemptId.make("held-plan-first"); const secondAttempt = RunAttemptId.make("held-plan-second"); const thirdAttempt = RunAttemptId.make("held-plan-third"); - const stamp = (frame: SDKMessage, attempt: RunAttemptId) => - claudeSdkFrame({ + const stamp = Effect.fnUntraced(function* (frame: SDKMessage, attempt: RunAttemptId) { + return claudeSdkFrame({ ...frame, - user_message_uuid: ClaudeAdapterV2.claudePromptUuid(attempt), + user_message_uuid: yield* ClaudeAdapterV2.claudePromptUuid(attempt), }); + }); const start = (attemptId: RunAttemptId, providerTurnOrdinal: number) => harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -5031,13 +5032,13 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); yield* start(firstAttempt, 1); yield* harness.offerAndWait( - stamp( + yield* stamp( makeAssistantTextFrame({ uuid: "00000000-0000-4000-8000-000000000801", text: "One." }), firstAttempt, ), ); yield* harness.offerAndWait( - stamp( + yield* stamp( makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000802", result: "One." }), firstAttempt, ), @@ -5091,9 +5092,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* Fiber.join(stopping); assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "interrupted"); yield* start(thirdAttempt, 3); - yield* harness.offerAndWait(stamp(planFrame, thirdAttempt)); + yield* harness.offerAndWait(yield* stamp(planFrame, thirdAttempt)); yield* harness.offerAndWait( - stamp( + yield* stamp( makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000804", result: "Three." }), thirdAttempt, ),