diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 785c4b1c9ec1..9d1ee058b0cc 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,116 @@ 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 = Effect.fnUntraced(function* (frame: SDKMessage, attempt: RunAttemptId) { + return claudeSdkFrame({ + ...frame, + user_message_uuid: yield* 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( + yield* stamp( + makeAssistantTextFrame({ uuid: "00000000-0000-4000-8000-000000000801", text: "One." }), + firstAttempt, + ), + ); + yield* harness.offerAndWait( + yield* 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(yield* stamp(planFrame, thirdAttempt)); + yield* harness.offerAndWait( + yield* 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({ diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 63f0cbdae66b..bbac14185dfb 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) @@ -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..750e405429fd 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) { @@ -5712,6 +5726,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi subagent: null, startedAt: now, itemPositions: new Map(), + planDeltas: new Map(), }; const providerTurn = { id: context.providerTurnId, @@ -6293,6 +6308,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" diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 46792897e789..a88f2dc663a4 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,119 @@ 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 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, + 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 }), + // Cancellation fails on the first unload or next turn, then is retried. + cancelOut, + reply("session.inbox.cancel", { + status: 500, + body: { _tag: "UnknownError", message: "busy" }, + }), + ...(unload + ? [ + cancelOut, + reply("session.inbox.cancel", null), + out("session.get", { sessionID: SESSION }), + replyData("session.get", sessionInfo()), + ...noOpenRequests, + ] + : []), + ...(unload ? withInstructions(nextPrompt) : nextPrompt), + ...(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 }); + // 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)); + 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,40 +3403,32 @@ 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(); - // 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* () { 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); + if (seen.length === 2) yield* Deferred.succeed(nextEnded, event.status); }), ), Stream.runDrain, @@ -3335,16 +3451,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); - 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); + // The replay rejects any stale steer or extra cancellation on the reattached turn. + assert.equal(yield* Deferred.await(nextEnded), "completed"); }).pipe(Effect.scoped), ); diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts index 7716b75baaf9..b22ffa3a0d4d 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts @@ -905,6 +905,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr // 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(); + // `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) => { @@ -3564,7 +3569,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 +3998,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); 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)); diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts index 06728c041066..4f48cc1fbcf7 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,211 @@ 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(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: "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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(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 87e24f92b0e8..46a26e1c1ee2 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");