From fce9e60290fa4bdf799e09f9b975e77a6ebd6e50 Mon Sep 17 00:00:00 2001 From: Nathan Shan Date: Sun, 25 Oct 2026 12:55:33 +0800 Subject: [PATCH 01/14] =?UTF-8?q?fix(acp):=20=E9=87=8D=E5=BC=80=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D=E6=97=B6=E4=BF=9D=E7=95=99=E5=B7=B2=E6=8A=95=E9=80=92?= =?UTF-8?q?=E6=8C=87=E4=BB=A4=E8=AE=B0=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 将指令摘要与原生会话标识保存到可选元数据，避免关闭后重开重复注入。 - 在完成事件前记录投递状态，并在压缩或回滚时失效，保留模式和工具变更的重新注入。 - 使用真实 mock 子进程验证重开、两类压缩和 ACP 工作状态时序；80 项测试及 check:changed 通过。 --- apps/server/scripts/acp-mock-agent.ts | 9 ++ .../Adapters/AcpAdapterV2.test.ts | 83 +++++++++++++- packages/contracts/src/orchestrationV2.ts | 8 ++ packages/provider-acp/src/server/adapter.ts | 108 ++++++++++++++---- 4 files changed, 181 insertions(+), 27 deletions(-) diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 51e89523d031e..f31c5f8e57598 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -958,6 +958,15 @@ const program = Effect.gen(function* () { yield* agent.handlePrompt((request) => Effect.gen(function* () { const requestedSessionId = String(request.sessionId ?? sessionId); + if (process.env.T3_ACP_ECHO_ACCEPTED_PROMPT === "1") { + yield* agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "Accepted test prompt" }, + }, + }); + } beginAcpMockPrompt(cancelledSessions, requestedSessionId); promptCount += 1; const idleError = process.env.T3_ACP_IDLE_ERROR; diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 73b66f2dc39c3..d35a4956ce36b 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -306,6 +306,7 @@ const pollProtocolMethods = (events: Queue.Queue>, + followAgentWorkState = false, ) { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const path = yield* Path.Path; @@ -331,6 +332,7 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( driver: ACP_TEST_DRIVER, capabilities: AcpProviderCapabilitiesV2, supportsCompaction: true, + followAgentWorkState, makeRuntime: makeMockRuntime({ childProcessSpawner, mockAgentPath, @@ -345,6 +347,7 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( }), }, selfInvocation, + continuationRequests: { offer: () => Effect.void }, }); const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ runtimeMode: "full-access", @@ -352,16 +355,45 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( cwd: process.cwd(), }); const modelSelection = { instanceId, model: "default" } as const; - const runtime = yield* adapter.openSession({ + const open = Effect.fnUntraced(function* (initialNativeThreadId?: string) { + const scope = yield* Scope.make(); + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)); + const runtime = yield* adapter + .openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-acp-instruction-compaction"), + modelSelection, + runtimePolicy, + ...(initialNativeThreadId === undefined ? {} : { initialNativeThreadId }), + }) + .pipe(Effect.provideService(Scope.Scope, scope)); + return { runtime, scope }; + }); + let session = yield* open(); + let providerThread = yield* session.runtime.ensureThread({ threadId, - providerSessionId: ProviderSessionId.make("provider-session-acp-instruction-compaction"), modelSelection, runtimePolicy, }); - const providerThread = yield* runtime.ensureThread({ threadId, modelSelection, runtimePolicy }); const now = yield* DateTime.now; let ordinal = 0; - return Effect.fnUntraced(function* (messageText: string, expectedStatus = "completed") { + return Effect.fnUntraced(function* ( + messageText: string, + expectedStatus = "completed", + reopen = false, + ) { + if (reopen) { + const nativeSessionId = providerThread.nativeThreadRef?.nativeId; + if (nativeSessionId == null) return yield* Effect.die("Expected a persisted native session"); + yield* Scope.close(session.scope, Exit.void); + session = yield* open(nativeSessionId); + providerThread = yield* session.runtime.resumeThread({ + providerThread, + modelSelection, + runtimePolicy, + }); + } + const { runtime } = session; yield* runtime.startTurn( makeTurnInput({ threadId, @@ -374,6 +406,11 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( }), ); const terminal = yield* runtime.events.pipe( + Stream.tap((event) => + Effect.sync(() => { + if (event.type === "provider_thread.updated") providerThread = event.providerThread; + }), + ), Stream.filter((event) => event.type === "turn.terminal"), Stream.runHead, ); @@ -1010,6 +1047,44 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.live("records instructions before ACP work-state completion ends the run", () => + Effect.gen(function* () { + const runTurn = yield* makeInstructionCompactionHarness( + { T3_ACP_ECHO_ACCEPTED_PROMPT: "1" }, + true, + ); + assertReinjectedInstructions(yield* runTurn("Start work."), "Start work."); + assert.equal( + (yield* runTurn("Continue work.", "completed", true))[0]?.text, + "Continue work.", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + + it.live.each(["native", "protocol"] as const)( + "retains ACP instructions across reopen but reinjects after %s compaction", + (compaction) => + Effect.gen(function* () { + const runTurn = yield* makeInstructionCompactionHarness( + compaction === "protocol" + ? { + T3_ACP_COMPACTION_PROMPT_ORDINALS: "2", + } + : {}, + ); + assertReinjectedInstructions(yield* runTurn("Start work."), "Start work."); + assert.equal( + (yield* runTurn("Continue work.", "completed", true))[0]?.text, + "Continue work.", + ); + yield* runTurn(compaction === "native" ? "/compact" : "Compact while working."); + assertReinjectedInstructions( + yield* runTurn("Continue after compact.", "completed", true), + "Continue after compact.", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.live( "reinjects ACP instructions after protocol compaction, including during reinjection", () => diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 6aa4ea7bbe66a..dd2ebc984905d 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -908,6 +908,14 @@ export const OrchestrationV2PendingBackgroundTask = kindUnionWithFallback( export type OrchestrationV2PendingBackgroundTask = typeof OrchestrationV2PendingBackgroundTask.Type; export const OrchestrationV2ProviderThreadNativeMetadata = Schema.Struct({ + acpInstructions: Schema.optional( + Schema.NullOr( + Schema.Struct({ + nativeSessionId: TrimmedNonEmptyString, + digest: TrimmedNonEmptyString, + }), + ), + ), modelSelection: Schema.optional(ModelSelection), title: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), updatedAt: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 3ac79ddc614f6..caf039d927e73 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -46,6 +46,7 @@ import type * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; +import * as Hex from "effect/encoding/Hex"; import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; @@ -1022,6 +1023,11 @@ interface AcpNativeBuildConfiguration { } interface ActiveAcpTurn { + readonly instructionDelivery?: { + readonly state: T3AcpInstructionState; + readonly digest: string; + readonly compactionMarker: symbol | undefined; + }; readonly input: ProviderAdapter.ProviderAdapterV2TurnInput; readonly providerTurnId: OrchestrationV2ProviderTurn["id"]; readonly nativeThreadId: string; @@ -1922,6 +1928,31 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const emitProviderEvent = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); + const recordInstructions = Effect.fnUntraced(function* ( + context: ActiveAcpTurn, + receipt: NonNullable< + OrchestrationV2ProviderThreadNativeMetadata["acpInstructions"] + > | null, + ) { + const metadata = { ...context.nativeMetadata, acpInstructions: receipt }; + context.nativeMetadata = metadata; + yield* Ref.update(nativeMetadataBySessionId, (current) => + new Map(current).set(context.nativeThreadId, metadata), + ); + const known = (yield* Ref.get(providerThreadByNativeSessionId)).get( + context.nativeThreadId, + ); + if (known === undefined) return; + const providerThread = { + ...known, + nativeMetadata: metadata, + updatedAt: yield* DateTime.now, + }; + yield* Ref.update(providerThreadByNativeSessionId, (current) => + new Map(current).set(context.nativeThreadId, providerThread), + ); + yield* emitProviderEvent({ type: "provider_thread.updated", driver, providerThread }); + }); let scheduleDeferredFinalize: (context: ActiveAcpTurn) => Effect.Effect = () => Effect.void; @@ -3650,6 +3681,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( : { beforeTokenCount: context.contextUsage.usedTokens }), }; if (status === "completed") { + yield* recordInstructions(context, null); context.contextUsage = null; yield* Ref.update(contextUsageBySessionId, (current) => { const updated = new Map(current); @@ -6914,6 +6946,26 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* emitProviderRetry(context, settledStatus); } const now = yield* DateTime.now; + const delivery = context.instructionDelivery; + if ( + settledStatus === "completed" && + delivery !== undefined && + (yield* Ref.get(compactionMarkers)).get(context.nativeThreadId) === + delivery.compactionMarker + ) { + yield* recordInstructions(context, { + nativeSessionId: context.nativeThreadId, + digest: delivery.digest, + }); + yield* Ref.update(promptInstructionStates, (current) => + new Map(current).set(context.nativeThreadId, delivery.state), + ); + yield* Ref.update(compactionMarkers, (current) => { + const updated = new Map(current); + updated.delete(context.nativeThreadId); + return updated; + }); + } if ( flavor.supportsCompaction === true && context.input.message.text.trim() === "/compact" && @@ -6921,6 +6973,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( settledStatus === "completed" ) { const nativeItemId = `${context.nativeTurnId}:compaction`; + yield* recordInstructions(context, null); yield* Ref.update(compactionMarkers, (current) => { const updated = new Map(current); updated.set(context.nativeThreadId, Symbol()); @@ -7092,7 +7145,27 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( interactionMode: turnInput.runtimePolicy.interactionMode, hasT3Mcp: (yield* readMcpContext(turnInput.threadId)).servers.length > 0, } satisfies T3AcpInstructionState; - const previousInstructionState = (yield* Ref.get(promptInstructionStates)).get(sessionId); + const instructionDigest = yield* crypto + .digest( + "SHA-256", + new TextEncoder().encode( + t3AcpPromptWithInstructions({ prompt: "", state: instructionState }), + ), + ) + .pipe(Effect.map(Hex.encode), Effect.orDie); + const rememberedInstructions = (yield* Ref.get(nativeMetadataBySessionId)).get( + sessionId, + )?.acpInstructions; + const recordedInstructions = + rememberedInstructions === undefined + ? turnInput.providerThread.nativeMetadata?.acpInstructions + : rememberedInstructions; + const previousInstructionState = + (yield* Ref.get(promptInstructionStates)).get(sessionId) ?? + (recordedInstructions?.nativeSessionId === sessionId && + recordedInstructions.digest === instructionDigest + ? instructionState + : undefined); const compactionMarker = (yield* Ref.get(compactionMarkers)).get(sessionId); const userText = flavor.rewriteUserText === undefined @@ -7173,6 +7246,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( return { prompt, instructionState: text === messageText ? undefined : instructionState, + instructionDigest, compactionMarker, }; }); @@ -7308,6 +7382,15 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ); } const context: ActiveAcpTurn = { + ...(promptParts?.instructionState === undefined + ? {} + : { + instructionDelivery: { + state: promptParts.instructionState, + digest: promptParts.instructionDigest, + compactionMarker: promptParts.compactionMarker, + }, + }), input: turnInput, providerTurnId, nativeThreadId: requestedSessionId, @@ -7477,28 +7560,6 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( } const promptGeneration = yield* Ref.get(runtimeCallbackGeneration); yield* runtime.prompt({ prompt: promptParts!.prompt }).pipe( - Effect.tap(() => - Ref.update(promptInstructionStates, (current) => { - if (promptParts?.instructionState === undefined) return current; - const updated = new Map(current); - updated.set(requestedSessionId, promptParts.instructionState); - return updated; - }), - ), - Effect.tap(() => - Ref.update(compactionMarkers, (current) => { - if (promptParts?.instructionState === undefined) return current; - const observedMarker = promptParts.compactionMarker; - if ( - observedMarker === undefined || - current.get(requestedSessionId) !== observedMarker - ) - return current; - const updated = new Map(current); - updated.delete(requestedSessionId); - return updated; - }), - ), // Wire settlement precedes the completion callback's permit request so // settled-soft classification can observe the native return even when // the completion fiber has not yet set promptSettled under the permit. @@ -8334,6 +8395,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( nativeConversationHeadRef: null, nativeMetadata: { ...rollbackInput.providerThread.nativeMetadata, + acpInstructions: null, itemIdentityVersion: 2 as const, }, status: "idle" as const, From 6072d2826e05b38e520fa5ae99a5a404b6b03973 Mon Sep 17 00:00:00 2001 From: Nathan Shan Date: Sun, 25 Oct 2026 13:01:12 +0800 Subject: [PATCH 02/14] =?UTF-8?q?fix(acp):=20=E4=BF=9D=E7=95=99=E5=BC=95?= =?UTF-8?q?=E5=AF=BC=E4=B8=AD=E6=96=AD=E5=90=8E=E7=9A=84=E8=BF=9B=E7=A8=8B?= =?UTF-8?q?=E5=86=85=E6=8C=87=E4=BB=A4=E7=8A=B6=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 恢复已确认 prompt 的进程内状态更新，避免 steer 重启路径重复注入指令。 - 持久记录仍只在正常完成时建立，保持重开与压缩回归保护。 - 复现并修复 CI 的 ACP Registry 引导回放失败；83 项针对性测试及 check:changed 通过。 --- packages/provider-acp/src/server/adapter.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index caf039d927e73..3ae5f94adb1f8 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -7560,6 +7560,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( } const promptGeneration = yield* Ref.get(runtimeCallbackGeneration); yield* runtime.prompt({ prompt: promptParts!.prompt }).pipe( + Effect.tap(() => + Ref.update(promptInstructionStates, (current) => { + if (promptParts?.instructionState === undefined) return current; + return new Map(current).set(requestedSessionId, promptParts.instructionState); + }), + ), // Wire settlement precedes the completion callback's permit request so // settled-soft classification can observe the native return even when // the completion fiber has not yet set promptSettled under the permit. From 10d2a225ebde5dcdf994f02424f66f1ffe832132 Mon Sep 17 00:00:00 2001 From: Nathan Shan Date: Sun, 25 Oct 2026 13:05:32 +0800 Subject: [PATCH 03/14] =?UTF-8?q?fix(acp):=20=E6=96=B0=E6=8C=87=E4=BB=A4?= =?UTF-8?q?=E5=8F=91=E9=80=81=E5=89=8D=E5=A4=B1=E6=95=88=E6=97=A7=E6=8A=95?= =?UTF-8?q?=E9=80=92=E8=AE=B0=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在发送替代指令前清除持久与进程内旧状态，避免模式切换失败后错误复用旧指令。 - 添加同一进程及重开会话后的 Plan 模式回归用例，确认失败后重新注入。 - 100 项跨适配器回放测试与包含新增用例的 check:changed 均通过。 --- .../Adapters/AcpAdapterV2.test.ts | 25 +++++++++++++++++-- packages/provider-acp/src/server/adapter.ts | 8 ++++++ 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index d35a4956ce36b..98c68d18105ba 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -381,7 +381,9 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( messageText: string, expectedStatus = "completed", reopen = false, + interactionMode: "default" | "plan" = "default", ) { + const turnRuntimePolicy = { ...runtimePolicy, interactionMode }; if (reopen) { const nativeSessionId = providerThread.nativeThreadRef?.nativeId; if (nativeSessionId == null) return yield* Effect.die("Expected a persisted native session"); @@ -390,7 +392,7 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( providerThread = yield* session.runtime.resumeThread({ providerThread, modelSelection, - runtimePolicy, + runtimePolicy: turnRuntimePolicy, }); } const { runtime } = session; @@ -399,7 +401,7 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( threadId, providerThread, instanceId, - runtimePolicy, + runtimePolicy: turnRuntimePolicy, now, ordinal: ++ordinal, messageText, @@ -1047,6 +1049,25 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.live.each([false, true])( + "invalidates old instructions after a failed mode change (reopen: %s)", + (reopen) => + Effect.gen(function* () { + const runTurn = yield* makeInstructionCompactionHarness({ + T3_ACP_FAIL_PROMPT_ORDINAL: "2", + }); + assert.include( + (yield* runTurn("Plan first.", "completed", false, "plan"))[0]?.text ?? "", + "T3 Code interaction mode: Plan", + ); + yield* runTurn("Implement now.", "failed"); + assert.include( + (yield* runTurn("Plan again.", "completed", reopen, "plan"))[0]?.text ?? "", + "T3 Code interaction mode: Plan", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.live("records instructions before ACP work-state completion ends the run", () => Effect.gen(function* () { const runTurn = yield* makeInstructionCompactionHarness( diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 3ae5f94adb1f8..8e8c08019cfcf 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -7559,6 +7559,14 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( return; } const promptGeneration = yield* Ref.get(runtimeCallbackGeneration); + if (promptParts?.instructionState !== undefined) { + yield* recordInstructions(context, null); + yield* Ref.update(promptInstructionStates, (current) => { + const updated = new Map(current); + updated.delete(requestedSessionId); + return updated; + }); + } yield* runtime.prompt({ prompt: promptParts!.prompt }).pipe( Effect.tap(() => Ref.update(promptInstructionStates, (current) => { From fd2a787c2ac65b8f404e7769fe7452b2d87ec7ae Mon Sep 17 00:00:00 2001 From: Nathan Shan Date: Sun, 25 Oct 2026 13:16:46 +0800 Subject: [PATCH 04/14] =?UTF-8?q?fix(acp):=20=E6=B8=85=E7=90=86=E7=A9=BA?= =?UTF-8?q?=E9=97=B2=E5=8E=8B=E7=BC=A9=E5=92=8C=E5=9B=9E=E6=BB=9A=E5=90=8E?= =?UTF-8?q?=E7=9A=84=E6=8C=87=E4=BB=A4=E7=BC=93=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 在空闲根会话完成压缩时同步失效持久指令记录和进程内缓存，保留原有后续任务处理。\n\n回滚替换会话时清除同名原生会话的旧投递记录，避免遗漏新会话的指令。\n\n补充提示词行为回归；72 项 ACP 测试、31 项 Registry/Grok 回放及 check:changed 均通过。 --- .../Adapters/AcpAdapterV2.test.ts | 57 ++++++++++++++++++- packages/provider-acp/src/server/adapter.ts | 36 +++++++++++- 2 files changed, 89 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 98c68d18105ba..36397e85b6d03 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -307,6 +307,7 @@ const pollProtocolMethods = (events: Queue.Queue>, followAgentWorkState = false, + drainPromptEvents = true, ) { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const path = yield* Path.Path; @@ -317,6 +318,9 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( const protocolEvents = yield* Queue.unbounded(); const instanceId = ProviderInstanceId.make("acp-test-instruction-compaction"); const threadId = ThreadId.make("thread-acp-instruction-compaction"); + let sessionUpdate: + | Parameters[0] + | undefined; yield* (yield* McpProviderSessions.McpProviderSessions).set({ environmentId: EnvironmentId.make("environment-acp-instruction-compaction"), threadId, @@ -340,9 +344,15 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( environment, wrapRuntime: (runtime) => ({ ...runtime, + handleSessionUpdate: (handler) => { + sessionUpdate = handler; + return runtime.handleSessionUpdate(handler); + }, // Make compaction during prompt completion deterministic at the adapter boundary. prompt: (payload, options) => - runtime.prompt(payload, options).pipe(Effect.tap(() => runtime.drainEvents)), + runtime + .prompt(payload, options) + .pipe(Effect.tap(() => (drainPromptEvents ? runtime.drainEvents : Effect.void))), }), }), }, @@ -380,11 +390,36 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( return Effect.fnUntraced(function* ( messageText: string, expectedStatus = "completed", - reopen = false, + reopen: boolean | "rollback" | "compact" = false, interactionMode: "default" | "plan" = "default", ) { const turnRuntimePolicy = { ...runtimePolicy, interactionMode }; - if (reopen) { + if (reopen === "compact") { + if (sessionUpdate === undefined) return yield* Effect.die("Missing session handler"); + const sessionId = providerThread.nativeThreadRef?.nativeId; + if (sessionId == null) return yield* Effect.die("Missing native session"); + yield* sessionUpdate({ + sessionId, + update: { + sessionUpdate: "compaction_update", + compactionId: "idle-compact", + status: "completed", + }, + }); + } + if (reopen === "rollback") { + const result = yield* session.runtime.rollbackThread({ + providerThread, + providerThreadTurns: [], + target: { + type: "thread_start", + checkpointId: CheckpointId.make("instruction-rollback"), + appRunOrdinal: 0, + }, + }); + providerThread = result.providerThread; + } + if (reopen === true) { const nativeSessionId = providerThread.nativeThreadRef?.nativeId; if (nativeSessionId == null) return yield* Effect.die("Expected a persisted native session"); yield* Scope.close(session.scope, Exit.void); @@ -438,6 +473,22 @@ function assertReinjectedInstructions( } describe("AcpAdapterV2", () => { + it.live.each(["rollback", "compact"] as const)( + "reinjects instructions after idle %s with no continuation consumer", + (transition) => + Effect.gen(function* () { + const runTurn = yield* makeInstructionCompactionHarness( + {}, + false, + transition !== "rollback", + ); + assertReinjectedInstructions(yield* runTurn("Start work."), "Start work."); + assertReinjectedInstructions( + yield* runTurn("Continue after reset.", "completed", transition), + "Continue after reset.", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); it.live.each(["details", "none"] as const)( "fails the turn on an ACP v2 idle error with %s, settles its open tool, and accepts the next prompt", (details) => diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 8e8c08019cfcf..c798922e486a0 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -1929,7 +1929,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const emitProviderEvent = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); const recordInstructions = Effect.fnUntraced(function* ( - context: ActiveAcpTurn, + context: Pick, receipt: NonNullable< OrchestrationV2ProviderThreadNativeMetadata["acpInstructions"] > | null, @@ -4507,6 +4507,34 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) { const context = yield* Ref.get(activeTurn); const update = notification.update; + if ( + (context === null || context.finalized) && + notification.sessionId === (yield* Ref.get(activeSessionId)) && + update.sessionUpdate === "compaction_update" && + update.status === "completed" + ) { + const known = (yield* Ref.get(providerThreadByNativeSessionId)).get( + notification.sessionId, + ); + yield* recordInstructions( + { + nativeThreadId: notification.sessionId, + nativeMetadata: + (yield* Ref.get(nativeMetadataBySessionId)).get(notification.sessionId) ?? + known?.nativeMetadata ?? + null, + }, + null, + ); + yield* Ref.update(promptInstructionStates, (current) => { + const updated = new Map(current); + updated.delete(notification.sessionId); + return updated; + }); + yield* Ref.update(compactionMarkers, (current) => + new Map(current).set(notification.sessionId, Symbol()), + ); + } if ( followsAgentWorkState && flavor.extractSubagentSessionEvent !== undefined && @@ -8370,6 +8398,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* Ref.set(activeSelection, null); yield* Ref.set(activeInteractionMode, null); yield* Ref.set(promptInstructionStates, new Map()); + yield* Ref.update(nativeMetadataBySessionId, (current) => + new Map(current).set(candidate.sessionId, { + ...current.get(candidate.sessionId), + acpInstructions: null, + }), + ); yield* Ref.set(compactionMarkers, new Map()); yield* Ref.set(providerTurns, new Map()); yield* Ref.set(snapshot, { From 8bf89eb08e56cfe18a0e4890800e15d2c913470f Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 02:51:50 +0000 Subject: [PATCH 05/14] fix(acp): persist instruction invalidation across idle sessions --- .../Adapters/AcpAdapterV2.test.ts | 36 +++++++- .../ProviderSessionManager.test.ts | 86 ++++++++++++++++++- .../ProviderSessionManager.ts | 54 ++++++++++++ packages/provider-acp/src/server/adapter.ts | 67 ++++++++++----- 4 files changed, 217 insertions(+), 26 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index dd26806eb129c..0f8d7c3f2efcc 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -392,7 +392,7 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( return Effect.fnUntraced(function* ( messageText: string, expectedStatus = "completed", - reopen: boolean | "rollback" | "compact" = false, + reopen: boolean | "rollback" | "compact" | "resume-compact-reopen" = false, interactionMode: "default" | "plan" = "default", ) { const turnRuntimePolicy = { ...runtimePolicy, interactionMode }; @@ -421,7 +421,7 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( }); providerThread = result.providerThread; } - if (reopen === true) { + const resume = Effect.fnUntraced(function* () { const nativeSessionId = providerThread.nativeThreadRef?.nativeId; if (nativeSessionId == null) return yield* Effect.die("Expected a persisted native session"); yield* Scope.close(session.scope, Exit.void); @@ -431,6 +431,27 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( modelSelection, runtimePolicy: turnRuntimePolicy, }); + }); + if (reopen === true || reopen === "resume-compact-reopen") yield* resume(); + if (reopen === "resume-compact-reopen") { + if (sessionUpdate === undefined) return yield* Effect.die("Missing session handler"); + const sessionId = providerThread.nativeThreadRef?.nativeId; + if (sessionId == null) return yield* Effect.die("Missing native session"); + yield* sessionUpdate({ + sessionId, + update: { + sessionUpdate: "compaction_update", + compactionId: "resumed-idle-compact", + status: "completed", + }, + }); + const updated = yield* session.runtime.events.pipe( + Stream.filter((event) => event.type === "provider_thread.updated"), + Stream.runHead, + ); + providerThread = Option.getOrThrow(updated).providerThread; + assert.isNull(providerThread.nativeMetadata?.acpInstructions); + yield* resume(); } const { runtime } = session; yield* runtime.startTurn( @@ -1135,6 +1156,17 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.live("persists idle compaction after resume before another turn", () => + Effect.gen(function* () { + const runTurn = yield* makeInstructionCompactionHarness({}); + yield* runTurn("Start work."); + assertReinjectedInstructions( + yield* runTurn("Continue after idle compaction.", "completed", "resume-compact-reopen"), + "Continue after idle compaction.", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.live.each(["native", "protocol"] as const)( "retains ACP instructions across reopen but reinjects after %s compaction", (compaction) => diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 3c635be6753df..033a809348ae3 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -14,7 +14,7 @@ import { ProjectId, ProviderDriverKind, ProviderInstanceId, - type ProviderSessionId, + ProviderSessionId, ThreadId, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; @@ -2900,6 +2900,90 @@ it.effect("ProviderSessionManagerV2 settles a request the event pump persists du }), ); +it.effect.each(["current", "newer-run", "newer-session"] as const)( + "ProviderSessionManagerV2 persists instruction invalidation only for its owner (%s)", + (owner) => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("instruction-invalidation"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const providerThread = { + ...makeProviderThread({ idAllocator, threadId, providerSessionId, now }), + lastRunOrdinal: 1, + nativeMetadata: { + title: "current title", + acpInstructions: { + nativeSessionId: "native-thread", + digest: "delivered-instructions", + }, + }, + }; + const stored = { + ...providerThread, + ...(owner === "newer-run" ? { lastRunOrdinal: 2 } : {}), + ...(owner === "newer-session" + ? { providerSessionId: ProviderSessionId.make("newer-session") } + : {}), + }; + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId, now }), + { + id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), + type: "provider-thread.updated", + threadId, + driver: CODEX_DRIVER, + occurredAt: now, + payload: stored, + }, + ], + }); + const before = (yield* projectionStore.getThreadRecords(threadId, ["providerThreads"])) + .providerThreads[0]!; + const runtime = yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + const subscription = yield* runtime.subscribeEvents!; + const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); + assert.isDefined(queue); + yield* Queue.offer(queue!, { + type: "provider_thread.updated", + driver: CODEX_DRIVER, + providerThread: { + ...providerThread, + nativeMetadata: { title: "stale title", acpInstructions: null }, + }, + }); + yield* subscription.events.pipe(Stream.runHead); + const projection = yield* projectionStore.getThreadRecords(threadId, ["providerThreads"]); + const updated = projection.providerThreads.find( + (candidate) => candidate.id === providerThread.id, + ); + assert.deepEqual( + updated, + owner === "current" + ? { + ...before, + nativeMetadata: { ...before.nativeMetadata, acpInstructions: null }, + } + : before, + ); + }).pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 }))); + }), +); + it.effect("ProviderSessionManagerV2 persists session-scoped runtime requests without a run", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 00b42d510bd91..3b3eaf1519a3a 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -1876,6 +1876,48 @@ export const layerWithOptions = ( ), ); + const persistInstructionInvalidation = Effect.fn( + "ProviderSessionManager.persistInstructionInvalidation", + )(function* (entry: LiveSessionEntry, offered: OrchestrationV2ProviderThread) { + const threadId = offered.appThreadId; + if (threadId === null || offered.nativeMetadata?.acpInstructions !== null) return; + // Idle updates outlive run subscribers; invalidate only the receipt, not newer thread state. + yield* Effect.gen(function* () { + const current = (yield* Ref.get(sessions)).get( + sessionKey(entry.runtime.providerSessionId), + ); + if (current?.runtime !== entry.runtime || !current.attachedThreadIds.has(threadId)) + return true; + const projection = yield* projectionStore.getThreadRecords(threadId, ["providerThreads"]); + const stored = projection.providerThreads.find( + (candidate) => candidate.id === offered.id, + ); + if ( + stored === undefined || + stored.providerSessionId !== entry.runtime.providerSessionId || + stored.lastRunOrdinal !== offered.lastRunOrdinal || + stored.nativeThreadRef?.nativeId !== offered.nativeThreadRef?.nativeId || + stored.nativeMetadata?.acpInstructions == null + ) + return true; + const events = yield* providerEventIngestor.normalize({ + providerSessionId: entry.runtime.providerSessionId, + providerInstanceId: entry.runtime.instanceId, + threadId, + event: { + type: "provider_thread.updated", + driver: entry.runtime.driver, + providerThread: { + ...stored, + nativeMetadata: { ...stored.nativeMetadata, acpInstructions: null }, + }, + }, + }); + return (yield* eventSink.writeIfProviderThreadUnchanged({ expected: stored, events })) + .committed; + }).pipe(Effect.repeat({ until: (committed) => committed })); + }); + const startEventPump = (entry: LiveSessionEntry) => { let stoppedByProvider = false; return entry.runtime.events.pipe( @@ -1917,6 +1959,18 @@ export const layerWithOptions = ( ), Effect.andThen( Effect.gen(function* () { + if (event.type === "provider_thread.updated") { + yield* persistInstructionInvalidation(entry, event.providerThread).pipe( + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver: entry.runtime.driver, + providerSessionId: entry.runtime.providerSessionId, + cause, + }), + ), + ); + } if ( event.type === "provider_thread.updated" && event.providerThread.status === "error" diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 05f555a765905..49dcb26fcd156 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -1933,13 +1933,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const emitProviderEvent = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); - const recordInstructions = Effect.fnUntraced(function* ( + const invalidateInstructions = Effect.fn("AcpAdapterV2.invalidateInstructions")(function* ( context: Pick, - receipt: NonNullable< - OrchestrationV2ProviderThreadNativeMetadata["acpInstructions"] - > | null, ) { - const metadata = { ...context.nativeMetadata, acpInstructions: receipt }; + const metadata = { ...context.nativeMetadata, acpInstructions: null }; context.nativeMetadata = metadata; yield* Ref.update(nativeMetadataBySessionId, (current) => new Map(current).set(context.nativeThreadId, metadata), @@ -3686,7 +3683,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( : { beforeTokenCount: context.contextUsage.usedTokens }), }; if (status === "completed") { - yield* recordInstructions(context, null); + yield* invalidateInstructions(context); context.contextUsage = null; yield* Ref.update(contextUsageBySessionId, (current) => { const updated = new Map(current); @@ -4521,16 +4518,13 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const known = (yield* Ref.get(providerThreadByNativeSessionId)).get( notification.sessionId, ); - yield* recordInstructions( - { - nativeThreadId: notification.sessionId, - nativeMetadata: - (yield* Ref.get(nativeMetadataBySessionId)).get(notification.sessionId) ?? - known?.nativeMetadata ?? - null, - }, - null, - ); + yield* invalidateInstructions({ + nativeThreadId: notification.sessionId, + nativeMetadata: + (yield* Ref.get(nativeMetadataBySessionId)).get(notification.sessionId) ?? + known?.nativeMetadata ?? + null, + }); yield* Ref.update(promptInstructionStates, (current) => { const updated = new Map(current); updated.delete(notification.sessionId); @@ -6982,10 +6976,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( (yield* Ref.get(compactionMarkers)).get(context.nativeThreadId) === delivery.compactionMarker ) { - yield* recordInstructions(context, { - nativeSessionId: context.nativeThreadId, - digest: delivery.digest, - }); + context.nativeMetadata = { + ...context.nativeMetadata, + acpInstructions: { nativeSessionId: context.nativeThreadId, digest: delivery.digest }, + }; yield* Ref.update(promptInstructionStates, (current) => new Map(current).set(context.nativeThreadId, delivery.state), ); @@ -7002,7 +6996,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( settledStatus === "completed" ) { const nativeItemId = `${context.nativeTurnId}:compaction`; - yield* recordInstructions(context, null); + yield* invalidateInstructions(context); yield* Ref.update(compactionMarkers, (current) => { const updated = new Map(current); updated.set(context.nativeThreadId, Symbol()); @@ -7043,6 +7037,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( threadId: context.input.threadId, providerTurn: turn, }); + const nativeMetadata = context.nativeMetadata; + if (nativeMetadata !== null) { + yield* Ref.update(nativeMetadataBySessionId, (current) => + new Map(current).set(context.nativeThreadId, nativeMetadata), + ); + } const updatedProviderThread: OrchestrationV2ProviderThread = { ...context.input.providerThread, providerSessionId: input.providerSessionId, @@ -7610,7 +7610,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( } const promptGeneration = yield* Ref.get(runtimeCallbackGeneration); if (promptParts?.instructionState !== undefined) { - yield* recordInstructions(context, null); + yield* invalidateInstructions(context); yield* Ref.update(promptInstructionStates, (current) => { const updated = new Map(current); updated.delete(requestedSessionId); @@ -7890,6 +7890,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( Effect.gen(function* () { yield* awaitRuntimeTeardown(); const sessionId = yield* nativeThreadId(driver, threadInput.providerThread); + yield* Ref.update(providerThreadByNativeSessionId, (current) => + new Map(current).set(sessionId, { + ...threadInput.providerThread, + providerSessionId: input.providerSessionId, + }), + ); const previousItemIdentityVersion = itemIdentityVersion; useProviderThreadIdentity(threadInput.providerThread); const restorePreviousItemIdentity = Effect.sync(() => { @@ -7919,12 +7925,27 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* Ref.set(activeInteractionMode, nextRuntimePolicy.interactionMode); } const now = yield* DateTime.now; - return { + const rememberedMetadata = (yield* Ref.get(nativeMetadataBySessionId)).get( + sessionId, + ); + const providerThread = { ...threadInput.providerThread, providerSessionId: input.providerSessionId, + ...(rememberedMetadata === undefined + ? {} + : { + nativeMetadata: { + ...threadInput.providerThread.nativeMetadata, + ...rememberedMetadata, + }, + }), status: "idle" as const, updatedAt: now, }; + yield* Ref.update(providerThreadByNativeSessionId, (current) => + new Map(current).set(sessionId, providerThread), + ); + return providerThread; }), ); }, From 30702d9c346d3aeea7d7488809ac182be99167d5 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 02:58:44 +0000 Subject: [PATCH 06/14] fix(acp): retain invalidations through run finalization --- .../ProviderSessionManager.test.ts | 28 ++++++- .../ProviderSessionManager.ts | 81 ++++++++++++------- 2 files changed, 79 insertions(+), 30 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 033a809348ae3..0af28e6de56ce 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -2900,7 +2900,7 @@ it.effect("ProviderSessionManagerV2 settles a request the event pump persists du }), ); -it.effect.each(["current", "newer-run", "newer-session"] as const)( +it.effect.each(["current", "queued-completion", "newer-run", "newer-session"] as const)( "ProviderSessionManagerV2 persists instruction invalidation only for its owner (%s)", (owner) => Effect.gen(function* () { @@ -2966,14 +2966,36 @@ it.effect.each(["current", "newer-run", "newer-session"] as const)( nativeMetadata: { title: "stale title", acpInstructions: null }, }, }); - yield* subscription.events.pipe(Stream.runHead); + const completionEventId = yield* idAllocator.allocate.event({ + threadId, + providerSessionId, + }); + yield* subscription.events.pipe( + Stream.tap(() => + owner === "queued-completion" + ? eventSink.write({ + events: [ + { + id: completionEventId, + type: "provider-thread.updated", + threadId, + driver: CODEX_DRIVER, + occurredAt: now, + payload: before, + }, + ], + }) + : Effect.void, + ), + Stream.runHead, + ); const projection = yield* projectionStore.getThreadRecords(threadId, ["providerThreads"]); const updated = projection.providerThreads.find( (candidate) => candidate.id === providerThread.id, ); assert.deepEqual( updated, - owner === "current" + owner === "current" || owner === "queued-completion" ? { ...before, nativeMetadata: { ...before.nativeMetadata, acpInstructions: null }, diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 3b3eaf1519a3a..8892cc1aa4ca3 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -199,6 +199,7 @@ interface LiveSessionEntry { ReadonlyMap> >; readonly requestEventPermit: Semaphore.Semaphore; + readonly instructionInvalidations: Map; readonly scope: Scope.Closeable; readonly idleGeneration: number; readonly busyTurns: ReadonlySet; @@ -1626,6 +1627,7 @@ export const layerWithOptions = ( const makeEventSubscription = Effect.fn("ProviderSessionManager.makeEventSubscription")( function* ( + providerSessionId: ProviderSessionId, subscribers: Ref.Ref< ReadonlyMap> >, @@ -1637,20 +1639,29 @@ export const layerWithOptions = ( updated.set(subscriberId, queue); return updated; }); - const close = Ref.modify(subscribers, (current) => { - if (!current.has(subscriberId)) { - return [false, current] as const; - } - const updated = new Map(current); - updated.delete(subscriberId); - return [true, updated] as const; - }).pipe( - Effect.flatMap((removed) => - removed - ? Queue.clear(queue).pipe(Effect.andThen(Queue.end(queue)), Effect.asVoid) - : Effect.void, - ), - ); + const close = Effect.gen(function* () { + const entry = (yield* Ref.get(sessions)).get(sessionKey(providerSessionId)); + const release = Effect.gen(function* () { + const removed = yield* Ref.modify(subscribers, (current) => { + if (!current.has(subscriberId)) return [false, current] as const; + const updated = new Map(current); + updated.delete(subscriberId); + return [true, updated] as const; + }); + if (!removed) return; + yield* Queue.clear(queue); + yield* Queue.end(queue); + // Completion can restore its cached receipt after an idle invalidation was persisted. + if (entry !== undefined) { + yield* Effect.forEach( + entry.instructionInvalidations.values(), + (offered) => persistInstructionInvalidation(entry, offered), + { discard: true }, + ); + } + }); + yield* entry === undefined ? release : entry.requestEventPermit.withPermits(1)(release); + }).pipe(Effect.orDie); const events = Stream.fromQueue(queue).pipe( Stream.mapEffect((signal) => signal.type === "event" @@ -1670,7 +1681,7 @@ export const layerWithOptions = ( >, ): ProviderAdapter.ProviderAdapterV2SessionRuntime => { const providerSessionId = runtime.providerSessionId; - const subscribeEvents = makeEventSubscription(eventSubscribers); + const subscribeEvents = makeEventSubscription(providerSessionId, eventSubscribers); const turnMetrics = (operation: string, model?: string) => withMetrics({ counter: providerTurnsTotal, @@ -1959,18 +1970,6 @@ export const layerWithOptions = ( ), Effect.andThen( Effect.gen(function* () { - if (event.type === "provider_thread.updated") { - yield* persistInstructionInvalidation(entry, event.providerThread).pipe( - Effect.mapError( - (cause) => - new ProviderAdapter.ProviderAdapterEventStreamError({ - driver: entry.runtime.driver, - providerSessionId: entry.runtime.providerSessionId, - cause, - }), - ), - ); - } if ( event.type === "provider_thread.updated" && event.providerThread.status === "error" @@ -1990,6 +1989,33 @@ export const layerWithOptions = ( }); }); } + if ( + event.type === "provider_thread.updated" && + event.providerThread.nativeMetadata?.acpInstructions !== undefined + ) { + yield* Effect.gen(function* () { + if (event.providerThread.nativeMetadata?.acpInstructions === null) { + entry.instructionInvalidations.set( + event.providerThread.id, + event.providerThread, + ); + } else { + entry.instructionInvalidations.delete(event.providerThread.id); + } + yield* persistInstructionInvalidation(entry, event.providerThread).pipe( + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver: entry.runtime.driver, + providerSessionId: entry.runtime.providerSessionId, + cause, + }), + ), + ); + yield* publishToSubscribers(entry.eventSubscribers, { type: "event", event }); + }).pipe(entry.requestEventPermit.withPermits(1)); + return; + } // Some providers can block before a run subscriber exists // (project trust, login, or session-switch hooks). Persist // their runless request artifacts directly so the normal T3 @@ -2232,6 +2258,7 @@ export const layerWithOptions = ( exposedRuntime, eventSubscribers, requestEventPermit: yield* Semaphore.make(1), + instructionInvalidations: new Map(), scope: sessionScope, idleGeneration: 0, busyTurns: new Set(), From 767503edd43c0218f885da94742c3a8c95592b32 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 03:03:15 +0000 Subject: [PATCH 07/14] fix(acp): acknowledge instructions after cancelled prompts --- .../Adapters/AcpAdapterV2.test.ts | 20 +++++++++++++++++++ packages/provider-acp/src/server/adapter.ts | 16 ++++++++++++++- 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 0f8d7c3f2efcc..2f8853c881cea 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -1142,6 +1142,26 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.live( + "acknowledges reinjected instructions on a cancelled prompt without a durable receipt", + () => + Effect.gen(function* () { + const runTurn = yield* makeInstructionCompactionHarness({ + T3_ACP_VIBE_RETRY_OUTCOME: "cancelled", + }); + yield* runTurn("Start work.", "cancelled"); + assertReinjectedInstructions( + yield* runTurn("Continue after compact.", "cancelled", "compact"), + "Continue after compact.", + ); + assert.equal((yield* runTurn("Continue work.", "cancelled"))[0]?.text, "Continue work."); + assertReinjectedInstructions( + yield* runTurn("Continue after reopen.", "cancelled", true), + "Continue after reopen.", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.live("records instructions before ACP work-state completion ends the run", () => Effect.gen(function* () { const runTurn = yield* makeInstructionCompactionHarness( diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 49dcb26fcd156..cc5dde991cefb 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -1031,7 +1031,7 @@ interface ActiveAcpTurn { readonly instructionDelivery?: { readonly state: T3AcpInstructionState; readonly digest: string; - readonly compactionMarker: symbol | undefined; + compactionMarker: symbol | undefined; }; readonly input: ProviderAdapter.ProviderAdapterV2TurnInput; readonly providerTurnId: OrchestrationV2ProviderTurn["id"]; @@ -7624,6 +7624,20 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( return new Map(current).set(requestedSessionId, promptParts.instructionState); }), ), + Effect.tap(() => + Ref.update(compactionMarkers, (current) => { + const delivery = context.instructionDelivery; + if ( + delivery?.compactionMarker === undefined || + current.get(requestedSessionId) !== delivery.compactionMarker + ) + return current; + delivery.compactionMarker = undefined; + const updated = new Map(current); + updated.delete(requestedSessionId); + return updated; + }), + ), // Wire settlement precedes the completion callback's permit request so // settled-soft classification can observe the native return even when // the completion fiber has not yet set promptSettled under the permit. From cb0b4371bb7c93867dc731188639ce09a51d6553 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 03:12:48 +0000 Subject: [PATCH 08/14] fix(acp): preserve invalidation cleanup after session release --- .../ProviderSessionManager.test.ts | 50 +++++++++++++------ .../ProviderSessionManager.ts | 15 +++--- 2 files changed, 40 insertions(+), 25 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 0af28e6de56ce..5571059c2c480 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -2900,7 +2900,14 @@ it.effect("ProviderSessionManagerV2 settles a request the event pump persists du }), ); -it.effect.each(["current", "queued-completion", "newer-run", "newer-session"] as const)( +it.effect.each([ + "current", + "queued-completion", + "released-completion", + "repeated-close", + "newer-run", + "newer-session", +] as const)( "ProviderSessionManagerV2 persists instruction invalidation only for its owner (%s)", (owner) => Effect.gen(function* () { @@ -2970,32 +2977,43 @@ it.effect.each(["current", "queued-completion", "newer-run", "newer-session"] as threadId, providerSessionId, }); + const restoreCompletion = eventSink.write({ + events: [ + { + id: completionEventId, + type: "provider-thread.updated", + threadId, + driver: CODEX_DRIVER, + occurredAt: now, + payload: before, + }, + ], + }); yield* subscription.events.pipe( Stream.tap(() => - owner === "queued-completion" - ? eventSink.write({ - events: [ - { - id: completionEventId, - type: "provider-thread.updated", - threadId, - driver: CODEX_DRIVER, - occurredAt: now, - payload: before, - }, - ], - }) - : Effect.void, + Effect.gen(function* () { + if (owner === "released-completion") + yield* manager.release({ providerSessionId, reason: "manual_shutdown" }); + if (owner === "queued-completion" || owner === "released-completion") + yield* restoreCompletion; + }), ), Stream.runHead, ); + if (owner === "repeated-close") { + yield* restoreCompletion; + yield* subscription.close; + } const projection = yield* projectionStore.getThreadRecords(threadId, ["providerThreads"]); const updated = projection.providerThreads.find( (candidate) => candidate.id === providerThread.id, ); assert.deepEqual( updated, - owner === "current" || owner === "queued-completion" + owner === "current" || + owner === "queued-completion" || + owner === "released-completion" || + owner === "repeated-close" ? { ...before, nativeMetadata: { ...before.nativeMetadata, acpInstructions: null }, diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 8892cc1aa4ca3..4468c534718cd 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -1639,8 +1639,8 @@ export const layerWithOptions = ( updated.set(subscriberId, queue); return updated; }); + const entry = (yield* Ref.get(sessions)).get(sessionKey(providerSessionId)); const close = Effect.gen(function* () { - const entry = (yield* Ref.get(sessions)).get(sessionKey(providerSessionId)); const release = Effect.gen(function* () { const removed = yield* Ref.modify(subscribers, (current) => { if (!current.has(subscriberId)) return [false, current] as const; @@ -1648,9 +1648,10 @@ export const layerWithOptions = ( updated.delete(subscriberId); return [true, updated] as const; }); - if (!removed) return; - yield* Queue.clear(queue); - yield* Queue.end(queue); + if (removed) { + yield* Queue.clear(queue); + yield* Queue.end(queue); + } // Completion can restore its cached receipt after an idle invalidation was persisted. if (entry !== undefined) { yield* Effect.forEach( @@ -1894,11 +1895,7 @@ export const layerWithOptions = ( if (threadId === null || offered.nativeMetadata?.acpInstructions !== null) return; // Idle updates outlive run subscribers; invalidate only the receipt, not newer thread state. yield* Effect.gen(function* () { - const current = (yield* Ref.get(sessions)).get( - sessionKey(entry.runtime.providerSessionId), - ); - if (current?.runtime !== entry.runtime || !current.attachedThreadIds.has(threadId)) - return true; + if (!entry.attachedThreadIds.has(threadId)) return true; const projection = yield* projectionStore.getThreadRecords(threadId, ["providerThreads"]); const stored = projection.providerThreads.find( (candidate) => candidate.id === offered.id, From 473afc2a9ae92961ef022d6627b2359fcab572b5 Mon Sep 17 00:00:00 2001 From: Nathan Shan Date: Mon, 26 Oct 2026 11:56:04 +0800 Subject: [PATCH 09/14] =?UTF-8?q?fix(acp):=20=E6=81=A2=E5=A4=8D=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D=E6=97=B6=E4=BF=9D=E7=95=99=E5=AE=8C=E6=95=B4=E5=8E=9F?= =?UTF-8?q?=E7=94=9F=E5=85=83=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 恢复会话时用已持久化元数据初始化缓存，同时保留较新的进程内失效记录。 补充原生标题更新后重开的提示词回归；修复前重复注入指令，修复后保持裸提示词。75 项适配器测试及 check:changed 的 136 项测试通过。 --- .../Adapters/AcpAdapterV2.test.ts | 35 +++++++++++++++++-- packages/provider-acp/src/server/adapter.ts | 6 ++++ 2 files changed, 39 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 2f8853c881cea..30e1ead2eaa8f 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -392,7 +392,12 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( return Effect.fnUntraced(function* ( messageText: string, expectedStatus = "completed", - reopen: boolean | "rollback" | "compact" | "resume-compact-reopen" = false, + reopen: + | boolean + | "rollback" + | "compact" + | "resume-compact-reopen" + | "resume-info-reopen" = false, interactionMode: "default" | "plan" = "default", ) { const turnRuntimePolicy = { ...runtimePolicy, interactionMode }; @@ -432,7 +437,23 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( runtimePolicy: turnRuntimePolicy, }); }); - if (reopen === true || reopen === "resume-compact-reopen") yield* resume(); + if (reopen === true || reopen === "resume-compact-reopen" || reopen === "resume-info-reopen") + yield* resume(); + if (reopen === "resume-info-reopen") { + if (sessionUpdate === undefined) return yield* Effect.die("Missing session handler"); + const sessionId = providerThread.nativeThreadRef?.nativeId; + if (sessionId == null) return yield* Effect.die("Missing native session"); + yield* sessionUpdate({ + sessionId, + update: { sessionUpdate: "session_info_update", title: "Updated native title" }, + }); + const updated = yield* session.runtime.events.pipe( + Stream.filter((event) => event.type === "provider_thread.updated"), + Stream.runHead, + ); + providerThread = Option.getOrThrow(updated).providerThread; + yield* resume(); + } if (reopen === "resume-compact-reopen") { if (sessionUpdate === undefined) return yield* Effect.die("Missing session handler"); const sessionId = providerThread.nativeThreadRef?.nativeId; @@ -496,6 +517,16 @@ function assertReinjectedInstructions( } describe("AcpAdapterV2", () => { + it.live("retains delivered instructions across a resumed session info update", () => + Effect.gen(function* () { + const runTurn = yield* makeInstructionCompactionHarness({}); + yield* runTurn("Start work."); + assert.equal( + (yield* runTurn("Continue work.", "completed", "resume-info-reopen"))[0]?.text, + "Continue work.", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); it.live.each(["rollback", "compact"] as const)( "reinjects instructions after idle %s with no continuation consumer", (transition) => diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index cc5dde991cefb..200e2da670a6a 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -7904,6 +7904,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( Effect.gen(function* () { yield* awaitRuntimeTeardown(); const sessionId = yield* nativeThreadId(driver, threadInput.providerThread); + yield* Ref.update(nativeMetadataBySessionId, (current) => + new Map(current).set(sessionId, { + ...threadInput.providerThread.nativeMetadata, + ...current.get(sessionId), + }), + ); yield* Ref.update(providerThreadByNativeSessionId, (current) => new Map(current).set(sessionId, { ...threadInput.providerThread, From 0d78d832d6e71c3f7c1048e61af70231a9a3bde0 Mon Sep 17 00:00:00 2001 From: Nathan Shan Date: Wed, 28 Oct 2026 18:02:37 +0800 Subject: [PATCH 10/14] =?UTF-8?q?fix(server):=20=E8=AE=A9=E6=8C=87?= =?UTF-8?q?=E4=BB=A4=E5=A4=B1=E6=95=88=E9=87=8D=E8=AF=95=E9=87=8A=E6=94=BE?= =?UTF-8?q?=E4=BC=9A=E8=AF=9D=E6=B8=85=E7=90=86=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将比较写入的锁限制在单次尝试内，重试之间让会话释放和订阅清理继续执行。 保留失效记录与所有权检查，避免并发更新覆盖新状态或因重试上限丢失失效写入。 新增连续四次写入冲突的回归验证；变更范围检查与 141 个测试全部通过。 --- .../ProviderSessionManager.test.ts | 50 +++++++++++++++++-- .../ProviderSessionManager.ts | 49 ++++++++++-------- 2 files changed, 75 insertions(+), 24 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index b8d5361908bdf..327892307e048 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -2905,6 +2905,7 @@ it.effect.each([ "queued-completion", "released-completion", "repeated-close", + "contended-close", "newer-run", "newer-session", ] as const)( @@ -2912,6 +2913,7 @@ it.effect.each([ (owner) => Effect.gen(function* () { const state = yield* Ref.make(emptyState); + const beforeUnchangedWrite = yield* Ref.make>(Effect.void); yield* Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; const idAllocator = yield* IdAllocator.IdAllocatorV2; @@ -2989,6 +2991,37 @@ it.effect.each([ }, ], }); + let conflicts = 0; + let release: Fiber.Fiber | undefined; + const raceWithCleanup: Effect.Effect = Effect.gen(function* () { + conflicts++; + if (conflicts === 1) { + release = yield* manager + .release({ providerSessionId, reason: "manual_shutdown" }) + .pipe(Effect.orDie, Effect.forkChild({ startImmediately: true })); + } else { + assert.isFalse( + yield* manager.isLive(providerSessionId), + "release must enter between retries", + ); + } + yield* eventSink.write({ + events: [ + { + id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), + type: "provider-thread.updated", + threadId, + driver: CODEX_DRIVER, + occurredAt: now, + payload: { + ...before, + nativeMetadata: { ...before.nativeMetadata, title: `conflict ${conflicts}` }, + }, + }, + ], + }); + if (conflicts < 4) yield* Ref.set(beforeUnchangedWrite, raceWithCleanup); + }).pipe(Effect.orDie); yield* subscription.events.pipe( Stream.tap(() => Effect.gen(function* () { @@ -2996,10 +3029,16 @@ it.effect.each([ yield* manager.release({ providerSessionId, reason: "manual_shutdown" }); if (owner === "queued-completion" || owner === "released-completion") yield* restoreCompletion; + if (owner === "contended-close") { + yield* restoreCompletion; + yield* Ref.set(beforeUnchangedWrite, raceWithCleanup); + } }), ), Stream.runHead, ); + if (release !== undefined) yield* Fiber.join(release); + if (owner === "contended-close") assert.equal(conflicts, 4); if (owner === "repeated-close") { yield* restoreCompletion; yield* subscription.close; @@ -3013,14 +3052,19 @@ it.effect.each([ owner === "current" || owner === "queued-completion" || owner === "released-completion" || - owner === "repeated-close" + owner === "repeated-close" || + owner === "contended-close" ? { ...before, - nativeMetadata: { ...before.nativeMetadata, acpInstructions: null }, + nativeMetadata: { + ...before.nativeMetadata, + ...(owner === "contended-close" ? { title: "conflict 4" } : {}), + acpInstructions: null, + }, } : before, ); - }).pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 }))); + }).pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000, beforeUnchangedWrite }))); }), ); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 8747843077fdc..c71a89aafd8a4 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -1632,16 +1632,17 @@ export const layerWithOptions = ( yield* Queue.clear(queue); yield* Queue.end(queue); } - // Completion can restore its cached receipt after an idle invalidation was persisted. - if (entry !== undefined) { - yield* Effect.forEach( - entry.instructionInvalidations.values(), - (offered) => persistInstructionInvalidation(entry, offered), - { discard: true }, - ); - } }); yield* entry === undefined ? release : entry.requestEventPermit.withPermits(1)(release); + // Completion can restore its cached receipt after an idle invalidation was persisted. + // Retry outside the cleanup lock so release can stop a competing runtime writer. + if (entry !== undefined) { + yield* Effect.forEach( + [...entry.instructionInvalidations.values()], + (offered) => persistInstructionInvalidation(entry, offered), + { discard: true }, + ); + } }).pipe(Effect.orDie); const events = Stream.fromQueue(queue).pipe( Stream.mapEffect((signal) => @@ -1875,6 +1876,7 @@ export const layerWithOptions = ( if (threadId === null || offered.nativeMetadata?.acpInstructions !== null) return; // Idle updates outlive run subscribers; invalidate only the receipt, not newer thread state. yield* Effect.gen(function* () { + if (entry.instructionInvalidations.get(offered.id) !== offered) return true; if (!entry.attachedThreadIds.has(threadId)) return true; const projection = yield* projectionStore.getThreadRecords(threadId, ["providerThreads"]); const stored = projection.providerThreads.find( @@ -1903,7 +1905,12 @@ export const layerWithOptions = ( }); return (yield* eventSink.writeIfProviderThreadUnchanged({ expected: stored, events })) .committed; - }).pipe(Effect.repeat({ until: (committed) => committed })); + }).pipe( + // Each comparison owns the permit; retries must let cleanup and newer receipts through. + entry.requestEventPermit.withPermits(1), + Effect.tap((committed) => (committed ? Effect.void : Effect.yieldNow)), + Effect.repeat({ until: (committed) => committed }), + ); }); const startEventPump = (entry: LiveSessionEntry) => { @@ -1970,7 +1977,7 @@ export const layerWithOptions = ( event.type === "provider_thread.updated" && event.providerThread.nativeMetadata?.acpInstructions !== undefined ) { - yield* Effect.gen(function* () { + yield* Effect.sync(() => { if (event.providerThread.nativeMetadata?.acpInstructions === null) { entry.instructionInvalidations.set( event.providerThread.id, @@ -1979,18 +1986,18 @@ export const layerWithOptions = ( } else { entry.instructionInvalidations.delete(event.providerThread.id); } - yield* persistInstructionInvalidation(entry, event.providerThread).pipe( - Effect.mapError( - (cause) => - new ProviderAdapter.ProviderAdapterEventStreamError({ - driver: entry.runtime.driver, - providerSessionId: entry.runtime.providerSessionId, - cause, - }), - ), - ); - yield* publishToSubscribers(entry.eventSubscribers, { type: "event", event }); }).pipe(entry.requestEventPermit.withPermits(1)); + yield* persistInstructionInvalidation(entry, event.providerThread).pipe( + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver: entry.runtime.driver, + providerSessionId: entry.runtime.providerSessionId, + cause, + }), + ), + ); + yield* publishToSubscribers(entry.eventSubscribers, { type: "event", event }); return; } // Some providers can block before a run subscriber exists From 27752f22a53b44dc397c38649ff1fa6bafb250b8 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sat, 31 Oct 2026 06:09:02 +0000 Subject: [PATCH 11/14] fix(acp): invalidate instructions after deferred finalization --- .../Adapters/AcpAdapterV2.test.ts | 27 +++++++++++++++++-- packages/provider-acp/src/server/adapter.ts | 2 ++ 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 3c5aae65fb9b9..8c5b3bddb75d0 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -6093,6 +6093,7 @@ describe("AcpAdapterV2", () => { const secondChildSessionId = "mock-child-session-finalize-window-second"; let firstSubagentStatus: "running" | "completed" = "running"; let secondSubagentStatus: "waiting" | "completed" = "waiting"; + let backgroundTurn = false; type RuntimeService = AcpSessionRuntime.AcpSessionRuntime["Service"]; let sessionUpdateHandler: | Parameters[0] @@ -6110,6 +6111,7 @@ describe("AcpAdapterV2", () => { deferFinalizeForBackgroundWork: true, enablePostSettleContinuation: true, extractSubagentUpdate: (toolCall) => { + if (!backgroundTurn) return undefined; if (toolCall.toolCallId === "tool-call-generic-1") { return { nativeTaskId: "task-finalize-window-first", @@ -6178,7 +6180,7 @@ describe("AcpAdapterV2", () => { Stream.runForEach((event) => Queue.offer(events, event)), Effect.forkScoped, ); - const providerThread = yield* runtime.ensureThread({ + let providerThread = yield* runtime.ensureThread({ threadId, modelSelection, runtimePolicy, @@ -6187,6 +6189,16 @@ describe("AcpAdapterV2", () => { yield* runtime.startTurn( makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now }), ); + while (true) { + const event = yield* Queue.take(events); + if (event.type === "provider_thread.updated") providerThread = event.providerThread; + if (event.type === "turn.terminal") break; + } + assert.isNotNull(providerThread.nativeMetadata?.acpInstructions); + backgroundTurn = true; + yield* runtime.startTurn( + makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now, ordinal: 2 }), + ); yield* Deferred.await(promptSettled); yield* TestClock.adjust("1 second"); assert.isDefined(sessionUpdateHandler, "session update handler must be wired"); @@ -6237,20 +6249,30 @@ describe("AcpAdapterV2", () => { rawOutput: { content: "SECOND_DONE" }, }, }); + const compactFiber = yield* sessionUpdateHandler!({ + sessionId: "mock-session-1", + update: { + sessionUpdate: "compaction_update", + compactionId: "finalization-window", + status: "completed", + }, + }).pipe(Effect.forkScoped({ startImmediately: true })); yield* Deferred.succeed(releaseFinalizationClockRead, undefined); yield* Fiber.join(adjustFiber); + yield* Fiber.join(compactFiber); yield* Effect.yieldNow; yield* Effect.yieldNow; const providerTurnId = idAllocator.derive.providerTurn({ driver: ACP_TEST_DRIVER, - nativeTurnId: acpScopedNativeId(instanceId, "mock-session-1:turn:1"), + nativeTurnId: acpScopedNativeId(instanceId, "mock-session-1:turn:2"), }); let completedSubagentUpdates = 0; let rootTerminalStatus: string | null = null; let polled = yield* Queue.poll(events); while (Option.isSome(polled)) { const event = polled.value; + if (event.type === "provider_thread.updated") providerThread = event.providerThread; if ( event.type === "turn_item.updated" && event.turnItem.type === "subagent" && @@ -6265,6 +6287,7 @@ describe("AcpAdapterV2", () => { polled = yield* Queue.poll(events); } assert.equal(rootTerminalStatus, "completed"); + assert.isNull(providerThread.nativeMetadata?.acpInstructions); assert.equal(completedSubagentUpdates, 1); assert.isFalse( yield* hasPendingBackgroundWork, diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 06cdbd8f678fa..d1c67d946dccf 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -4615,6 +4615,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( update.sessionUpdate === "compaction_update" && update.status === "completed" ) { + // Deferred finalization can still publish a receipt after marking itself finalized. + if (context !== null) yield* Deferred.await(context.completed); const known = (yield* Ref.get(providerThreadByNativeSessionId)).get( notification.sessionId, ); From a4e10f9a7757523a119ee7ad7ac378f8c2dc40d5 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 2 Nov 2026 23:45:09 +0000 Subject: [PATCH 12/14] fix(acp): preserve invalidation ownership during cleanup --- .../ProviderSessionManager.test.ts | 31 ++++++++++++++++--- .../ProviderSessionManager.ts | 16 ++++++++-- 2 files changed, 40 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index b0db0fda1898e..45b9f4cb2a985 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -144,7 +144,9 @@ const layerPausingAttachEventSink = (pause: { }), ).pipe(Layer.provide(layerTestEventSink)); -const layerBeforeUnchangedWriteEventSink = (before: Ref.Ref>) => +const layerBeforeUnchangedWriteEventSink = ( + before: Ref.Ref>, +) => Layer.effect( EventSink.EventSinkV2, Effect.gen(function* () { @@ -476,7 +478,7 @@ function layerTest(input: { readonly failReleaseEventWrites?: boolean; readonly flakyReleaseWrites?: FlakyReleaseWrites; readonly pauseAttachWrite?: Parameters[0]; - readonly beforeUnchangedWrite?: Ref.Ref>; + readonly beforeUnchangedWrite?: Ref.Ref>; readonly pauseResolve?: { readonly armed: Ref.Ref; readonly paused: Deferred.Deferred; @@ -2516,7 +2518,9 @@ it.effect("ProviderSessionManagerV2 keeps the roster of a runtime whose tasks ca it.effect("ProviderSessionManagerV2 keeps a provider update that lands while release clears", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); - const beforeUnchangedWrite = yield* Ref.make>(Effect.void); + const beforeUnchangedWrite = yield* Ref.make< + Effect.Effect + >(Effect.void); const effect = Effect.gen(function* () { const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; const { providerSessionId, providerThreads, writeProviderThread, rosters } = @@ -3084,6 +3088,8 @@ it.effect.each([ "released-completion", "repeated-close", "contended-close", + "detached-close", + "failed-close", "newer-run", "newer-session", ] as const)( @@ -3091,7 +3097,9 @@ it.effect.each([ (owner) => Effect.gen(function* () { const state = yield* Ref.make(emptyState); - const beforeUnchangedWrite = yield* Ref.make>(Effect.void); + const beforeUnchangedWrite = yield* Ref.make< + Effect.Effect + >(Effect.void); yield* Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; const idAllocator = yield* IdAllocator.IdAllocatorV2; @@ -3207,6 +3215,17 @@ it.effect.each([ yield* manager.release({ providerSessionId, reason: "manual_shutdown" }); if (owner === "queued-completion" || owner === "released-completion") yield* restoreCompletion; + if (owner === "detached-close") { + yield* restoreCompletion; + yield* manager.detach({ providerSessionId, threadId }); + } + if (owner === "failed-close") { + yield* restoreCompletion; + yield* Ref.set( + beforeUnchangedWrite, + Effect.fail(new EventSink.EventSinkWriteError({ eventCount: 1 })), + ); + } if (owner === "contended-close") { yield* restoreCompletion; yield* Ref.set(beforeUnchangedWrite, raceWithCleanup); @@ -3217,6 +3236,7 @@ it.effect.each([ ); if (release !== undefined) yield* Fiber.join(release); if (owner === "contended-close") assert.equal(conflicts, 4); + if (owner === "failed-close") yield* subscription.close; if (owner === "repeated-close") { yield* restoreCompletion; yield* subscription.close; @@ -3231,7 +3251,8 @@ it.effect.each([ owner === "queued-completion" || owner === "released-completion" || owner === "repeated-close" || - owner === "contended-close" + owner === "contended-close" || + owner === "failed-close" ? { ...before, nativeMetadata: { diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index f07d810d77521..b515487e98820 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -1686,7 +1686,15 @@ export const layerWithOptions = ( if (entry !== undefined) { yield* Effect.forEach( [...entry.instructionInvalidations.values()], - (offered) => persistInstructionInvalidation(entry, offered), + (offered) => + persistInstructionInvalidation(entry, offered).pipe( + Effect.catch((cause) => + Effect.logWarning( + "orchestration-v2.provider-session.instruction-invalidation-failed", + { providerSessionId, providerThreadId: offered.id, cause }, + ), + ), + ), { discard: true }, ); } @@ -1961,7 +1969,11 @@ export const layerWithOptions = ( // Idle updates outlive run subscribers; invalidate only the receipt, not newer thread state. yield* Effect.gen(function* () { if (entry.instructionInvalidations.get(offered.id) !== offered) return true; - if (!entry.attachedThreadIds.has(threadId)) return true; + const current = (yield* Ref.get(sessions)).get( + sessionKey(entry.runtime.providerSessionId), + ); + if (current !== undefined && current.runtime !== entry.runtime) return true; + if (!(current?.attachedThreadIds ?? entry.attachedThreadIds).has(threadId)) return true; const projection = yield* projectionStore.getThreadRecords(threadId, ["providerThreads"]); const stored = projection.providerThreads.find( (candidate) => candidate.id === offered.id, From e10de58ad81c6aab4bc8613c86a0d59cf63614e6 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 2 Nov 2026 23:57:52 +0000 Subject: [PATCH 13/14] test(acp): expect retained instructions in kiro resume replay --- .../fixtures/provider_thread_resume/kiro_transcript.ndjson | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/kiro_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/kiro_transcript.ndjson index 2f5d9833e73c7..3fc95702aa121 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/kiro_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/kiro_transcript.ndjson @@ -44,7 +44,7 @@ {"type":"emit_inbound","label":"notification:session/update:available_commands_update","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","update":{"sessionUpdate":"available_commands_update","availableCommands":[{"name":"architecture-selection","description":"Steering: architecture-selection","input":{"hint":"optional context"},"_meta":{"kiro":{"type":"steering","contextQuery":"global:architecture-selection","commandId":"architecture-selection","resource":{"resourceType":"steering","source":{"origin":"bundled"}},"originalName":"architecture-selection"}}},{"name":"quick-spec","description":"Steering: quick-spec","input":{"hint":"optional context"},"_meta":{"kiro":{"type":"steering","contextQuery":"global:quick-spec","commandId":"quick-spec","resource":{"resourceType":"steering","source":{"origin":"bundled"}},"originalName":"quick-spec"}}},{"name":"bug-fix","description":"Steering: bug-fix","input":{"hint":"optional context"},"_meta":{"kiro":{"type":"steering","contextQuery":"global:bug-fix","commandId":"bug-fix","resource":{"resourceType":"steering","source":{"origin":"bundled"}},"originalName":"bug-fix"}}},{"name":"general-task-execution","description":"General-purpose sub-agent with access to all tools for executing arbitrary tasks","input":{"hint":"task to delegate"},"_meta":{"kiro":{"type":"custom-agent","commandId":"general-task-execution","resource":{"resourceType":"agent","source":{"origin":"bundled"}},"originalName":"general-task-execution"}}},{"name":"context-gatherer","description":"Investigates the codebase to answer a specific question about it. Deeply analyzes existing code by tracing execution paths, mapping architecture and abstraction layers, understanding patterns, and documenting dependencies, returning a thorough, direct written analysis that explains how the relevant code works and points to the specific files and lines involved. Use it to understand code before you change it: before designing or implementing a change to code you don't already fully understand, start with context-gatherer so your edits are grounded in how the code actually works. When a task requires understanding code you don't already know, dispatch context-gatherer to do that exploration before running your own searches or reads. Rules: 1. Provide a clear, specific prompt that includes your intent: a detailed task description of the context you need and what you're trying to accomplish. 2. The subagent explores the codebase and returns an analysis of your query 3. Trust the returned results. You may still inspect additional code yourself when you need context beyond what was returned. 4. Reuse existing context-gatherer output for related follow-up questions; do not re-invoke on the same question with different phrasing. 5. For multiple distinct questions, dispatch several context-gatherers in parallel, each targeting a different aspect of the codebase.","input":{"hint":"task to delegate"},"_meta":{"kiro":{"type":"custom-agent","commandId":"context-gatherer","resource":{"resourceType":"agent","source":{"origin":"bundled"}},"originalName":"context-gatherer"}}}]}}}} {"type":"emit_inbound","label":"response:session/set_config_option","frame":{"kind":"response","method":"session/set_config_option","result":{"configOptions":[{"type":"select","id":"mode","name":"Mode","category":"mode","currentValue":"vibe","options":[{"value":"vibe","name":"Default","description":"General coding assistance","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"spec","name":"Spec","description":"Structured feature development","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"quick-spec","name":"Quick Spec","description":"Fast spec workflow: clarify, then auto-generate requirements, design, and tasks","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"bug-fix","name":"Bug Fix","description":"Structured bug-fix workflow: investigate, diagnose, and resolve bugs","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"plan","name":"Plan","description":"Plan-only mode that helps break ideas down into an implementation plan without making any changes","_meta":{"kiro":{"welcomeMessage":"Transform any idea into fully working code. What do you want to build today?","source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"autonomous","name":"Autonomous","description":"Autonomous agent execution","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"semantic_reviewer","name":"semantic_reviewer","description":"Behavioral, narrative code review of a local or PR diff. Writes a review document and a verdict.","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}}]},{"type":"select","id":"memoryReflection","name":"Memory reflection","description":"Allow this session to reflect on new memory","currentValue":"on","options":[{"value":"on","name":"On"},{"value":"off","name":"Off"}]},{"type":"select","id":"autopilot","name":"Autopilot","currentValue":"on","options":[{"value":"on","name":"Autopilot","description":"Agent executes tools without confirmation"},{"value":"off","name":"Supervised","description":"Agent asks for approval before file changes"}]},{"type":"select","id":"contentCollection","name":"Content Collection","currentValue":"enabled","options":[{"value":"enabled","name":"Enabled","description":"Allow content to be used for service improvement"},{"value":"disabled","name":"Disabled","description":"Opt out of content collection"}]}]}}} {"type":"emit_inbound","label":"notification:session/update:config_option_update","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","update":{"sessionUpdate":"config_option_update","configOptions":[{"type":"select","id":"mode","name":"Mode","category":"mode","currentValue":"vibe","options":[{"value":"vibe","name":"Default","description":"General coding assistance","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"spec","name":"Spec","description":"Structured feature development","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"quick-spec","name":"Quick Spec","description":"Fast spec workflow: clarify, then auto-generate requirements, design, and tasks","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"bug-fix","name":"Bug Fix","description":"Structured bug-fix workflow: investigate, diagnose, and resolve bugs","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"plan","name":"Plan","description":"Plan-only mode that helps break ideas down into an implementation plan without making any changes","_meta":{"kiro":{"welcomeMessage":"Transform any idea into fully working code. What do you want to build today?","source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"autonomous","name":"Autonomous","description":"Autonomous agent execution","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}},{"value":"semantic_reviewer","name":"semantic_reviewer","description":"Behavioral, narrative code review of a local or PR diff. Writes a review document and a verdict.","_meta":{"kiro":{"source":"bundled","resource":{"resourceType":"agent","source":{"origin":"bundled"}}}}}]},{"type":"select","id":"model","name":"Model","category":"model","currentValue":"claude-haiku-4.5","options":[{"value":"auto","name":"Auto","description":"Models chosen by task for optimal usage and consistent quality","_meta":{"kiro":{"rateMultiplier":1,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"claude-opus-5.5","name":"Claude Opus 5.5","description":"Experimental preview of Claude Opus 5.5 model with 1M context window","_meta":{"kiro":{"rateMultiplier":2,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","xhigh","max"],"defaultEffortLevel":"medium","thinkingToggleable":false}}},{"value":"claude-sonnet-5.5","name":"Claude Sonnet 5.5","description":"Claude Sonnet 5.5 model with 1M context window","_meta":{"kiro":{"rateMultiplier":1.3,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":false}}},{"value":"claude-opus-5","name":"Claude Opus 5","description":"Claude Opus 5 model with 1M context window","_meta":{"kiro":{"rateMultiplier":2.2,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":true,"defaultThinkingEnabled":true}}},{"value":"claude-sonnet-5","name":"Claude Sonnet 5","description":"Claude Sonnet 5 model with 1M context window","_meta":{"kiro":{"rateMultiplier":1.3,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":true,"defaultThinkingEnabled":true}}},{"value":"claude-opus-4.8","name":"Claude Opus 4.8","description":"Claude Opus 4.8 model with 1M context window","_meta":{"kiro":{"rateMultiplier":2.2,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":true,"defaultThinkingEnabled":true}}},{"value":"gpt-5.6-sol","name":"GPT 5.6 Sol","description":"Experimental preview of OpenAI GPT 5.6 Sol with 1M context window","_meta":{"kiro":{"rateMultiplier":4.4,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"reasoning","effortLevels":["none","low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":false}}},{"value":"gpt-5.6-terra","name":"GPT 5.6 Terra","description":"Experimental preview of OpenAI GPT 5.6 Terra with 1M context window","_meta":{"kiro":{"rateMultiplier":2.2,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"reasoning","effortLevels":["none","low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":false}}},{"value":"gpt-5.6-luna","name":"GPT 5.6 Luna","description":"Experimental preview of OpenAI GPT 5.6 Luna with 1M context window","_meta":{"kiro":{"rateMultiplier":0.6,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"reasoning","effortLevels":["none","low","medium","high","xhigh","max"],"defaultEffortLevel":"high","thinkingToggleable":false}}},{"value":"claude-opus-4.7","name":"Claude Opus 4.7","description":"Claude Opus 4.7 model with 1M context window","_meta":{"kiro":{"rateMultiplier":2.2,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","xhigh","max"],"defaultEffortLevel":"xhigh","thinkingToggleable":true,"defaultThinkingEnabled":true}}},{"value":"claude-opus-4.6","name":"Claude Opus 4.6","description":"Claude Opus 4.6 model with 1M context window","_meta":{"kiro":{"rateMultiplier":2.2,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","max"],"defaultEffortLevel":"high","thinkingToggleable":true,"defaultThinkingEnabled":true}}},{"value":"claude-sonnet-4.6","name":"Claude Sonnet 4.6","description":"Claude Sonnet 4.6 model with 1M context window","_meta":{"kiro":{"rateMultiplier":1.3,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"output_config","effortLevels":["low","medium","high","max"],"defaultEffortLevel":"high","thinkingToggleable":true,"defaultThinkingEnabled":true}}},{"value":"claude-opus-4.5","name":"Claude Opus 4.5","description":"Claude Opus 4.5 model","_meta":{"kiro":{"rateMultiplier":2.2,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"claude-sonnet-4.5","name":"Claude Sonnet 4.5","description":"Claude Sonnet 4.5 model","_meta":{"kiro":{"rateMultiplier":1.3,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"claude-haiku-4.5","name":"Claude Haiku 4.5","description":"The latest Claude Haiku model","_meta":{"kiro":{"rateMultiplier":0.4,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"deepseek-3.2","name":"Deepseek v3.2","description":"Experimental preview of DeepSeek V3.2","_meta":{"kiro":{"rateMultiplier":0.25,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"minimax-m2.5","name":"MiniMax M2.5","description":"MiniMax M2.5 model","_meta":{"kiro":{"rateMultiplier":0.25,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"minimax-m2.1","name":"MiniMax M2.1","description":"Experimental preview of MiniMax M2.1","_meta":{"kiro":{"rateMultiplier":0.15,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"glm-5","name":"GLM 5","description":"GLM-5 model","_meta":{"kiro":{"rateMultiplier":0.5,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"qwen3-coder-next","name":"Qwen3 Coder Next","description":"Experimental preview of Qwen3 Coder Next","_meta":{"kiro":{"rateMultiplier":0.05,"rateUnit":"Credit","hasEffort":false,"thinkingToggleable":false}}},{"value":"kimi-k3","name":"Kimi K3","description":"Experimental preview of Kimi K3. Kimi K3 is a Kiro Excluded Model. For more info, see documentation.","_meta":{"kiro":{"rateMultiplier":2.4,"rateUnit":"Credit","hasEffort":true,"effortSchemaPath":"reasoning","effortLevels":["low","high","max"],"defaultEffortLevel":"max","thinkingToggleable":false}}}]},{"type":"select","id":"memoryReflection","name":"Memory reflection","description":"Allow this session to reflect on new memory","currentValue":"on","options":[{"value":"on","name":"On"},{"value":"off","name":"Off"}]},{"type":"select","id":"autopilot","name":"Autopilot","currentValue":"on","options":[{"value":"on","name":"Autopilot","description":"Agent executes tools without confirmation"},{"value":"off","name":"Supervised","description":"Agent asks for approval before file changes"}]},{"type":"select","id":"contentCollection","name":"Content Collection","currentValue":"enabled","options":[{"value":"enabled","name":"Enabled","description":"Allow content to be used for service improvement"},{"value":"disabled","name":"Disabled","description":"Opt out of content collection"}]}]}}}} -{"type":"expect_outbound","label":"request:session/prompt","frame":{"kind":"request","method":"session/prompt","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","prompt":[{"type":"text","text":"\n\n\n\n\nUsing the conversation history available in this resumed thread, first repeat the exact final answer you gave in the previous turn. Then on a new line write exactly: provider thread resume fixture second turn complete\n"}]}}} +{"type":"expect_outbound","label":"request:session/prompt","frame":{"kind":"request","method":"session/prompt","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","prompt":[{"type":"text","text":"Using the conversation history available in this resumed thread, first repeat the exact final answer you gave in the previous turn. Then on a new line write exactly: provider thread resume fixture second turn complete"}]}}} {"type":"emit_inbound","label":"notification:session/update:session_info_update","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","update":{"sessionUpdate":"session_info_update","_meta":{"kiro":{"userMessageId":"00000000-0000-4000-9000-000000000006","kind":"user_message_id_assigned"}}}}}} {"type":"emit_inbound","label":"notification:session/update:session_info_update","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","update":{"sessionUpdate":"session_info_update","_meta":{"kiro":{"focus":{"status":"in_progress"},"kind":"focus_update","status":"in_progress","activity":{"turnActive":true,"runningWorkflows":0,"pausedWorkflows":0}}}}}}} {"type":"emit_inbound","label":"notification:session/update:session_info_update","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"sess_00000000-0000-4000-8000-000000000001","update":{"sessionUpdate":"session_info_update","_meta":{"kiro":{"turnStart":true,"kind":"turn_start","messageId":"00000000-0000-4000-9000-000000000007-turn-start"}}}}}} From 755344d0af52eb6570c308e02b3131c6db7a59ab Mon Sep 17 00:00:00 2001 From: Nathan Shan Date: Sat, 7 Nov 2026 21:33:57 +0800 Subject: [PATCH 14/14] =?UTF-8?q?fix(acp):=20=E6=8C=81=E4=B9=85=E5=8C=96?= =?UTF-8?q?=E5=A4=B1=E6=95=88=E5=A4=B1=E8=B4=A5=E6=97=B6=E4=BF=9D=E7=95=99?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E4=B8=AD=E7=9A=84=E4=BC=9A=E8=AF=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 将指令失效写入失败记录为警告，保留待处理状态供订阅清理重试。 - 扩展现有所有权回归，确认事件继续投递且会话不被关闭。 - 适配主线 Digest 服务及驱动依赖，完成变更范围检查和 201 项测试。 --- apps/server/scripts/acp-mock-agent.ts | 3 ++- .../ProviderSessionManager.test.ts | 22 ++++++++++++++++++- .../ProviderSessionManager.ts | 12 +++++----- .../src/server/adapter.ts | 2 ++ packages/provider-acp/src/server/adapter.ts | 4 +++- packages/provider-fx/src/server/adapter.ts | 2 ++ packages/provider-grok/src/server/adapter.ts | 2 ++ packages/provider-kiro/src/server/adapter.ts | 2 ++ 8 files changed, 41 insertions(+), 8 deletions(-) diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 44901a1760f2b..e7d85f8b39799 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -64,6 +64,7 @@ const failLoadSessionAfterConfigReplay = process.env.T3_ACP_FAIL_LOAD_SESSION_AFTER_CONFIG_REPLAY === "1"; const emitLoadReplay = process.env.T3_ACP_EMIT_LOAD_REPLAY === "1"; const failPrompt = process.env.T3_ACP_FAIL_PROMPT === "1"; +const echoAcceptedPrompt = process.env.T3_ACP_ECHO_ACCEPTED_PROMPT === "1"; // JSON for the `data` of an "Internal error" response to `session/new`, as // Antigravity sends when its runtime cannot reach its local harness. const newSessionErrorData = @@ -829,7 +830,7 @@ const program = Effect.gen(function* () { yield* agent.handlePrompt((request) => Effect.gen(function* () { const requestedSessionId = String(request.sessionId ?? sessionId); - if (process.env.T3_ACP_ECHO_ACCEPTED_PROMPT === "1") { + if (echoAcceptedPrompt) { yield* agent.client.sessionUpdate({ sessionId: requestedSessionId, update: { diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 6e16c3560f541..5aced3f7facda 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -3467,6 +3467,7 @@ it.effect.each([ "contended-close", "detached-close", "failed-close", + "failed-event", "newer-run", "newer-session", ] as const)( @@ -3530,6 +3531,12 @@ it.effect.each([ const subscription = yield* runtime.subscribeEvents!; const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); assert.isDefined(queue); + if (owner === "failed-event") { + yield* Ref.set( + beforeUnchangedWrite, + Effect.fail(new EventSink.EventSinkWriteError({ eventCount: 1 })), + ); + } yield* Queue.offer(queue!, { type: "provider_thread.updated", driver: CODEX_DRIVER, @@ -3588,6 +3595,14 @@ it.effect.each([ yield* subscription.events.pipe( Stream.tap(() => Effect.gen(function* () { + if (owner === "failed-event") { + assert.isTrue(yield* manager.isLive(providerSessionId)); + assert.deepEqual( + (yield* projectionStore.getThreadRecords(threadId, ["providerThreads"])) + .providerThreads[0]?.nativeMetadata?.acpInstructions, + before.nativeMetadata?.acpInstructions, + ); + } if (owner === "released-completion") yield* manager.release({ providerSessionId, reason: "manual_shutdown" }); if (owner === "queued-completion" || owner === "released-completion") @@ -3612,6 +3627,10 @@ it.effect.each([ Stream.runHead, ); if (release !== undefined) yield* Fiber.join(release); + if (owner === "failed-event") { + assert.isTrue(yield* manager.isLive(providerSessionId)); + assert.equal((yield* Ref.get(state)).closeCount, 0); + } if (owner === "contended-close") assert.equal(conflicts, 4); if (owner === "failed-close") yield* subscription.close; if (owner === "repeated-close") { @@ -3629,7 +3648,8 @@ it.effect.each([ owner === "released-completion" || owner === "repeated-close" || owner === "contended-close" || - owner === "failed-close" + owner === "failed-close" || + owner === "failed-event" ? { ...before, nativeMetadata: { diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index d207374f1cc41..989c37e857ffd 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -2106,13 +2106,15 @@ export const layerWithOptions = ( } }).pipe(entry.requestEventPermit.withPermits(1)); yield* persistInstructionInvalidation(entry, event.providerThread).pipe( - Effect.mapError( - (cause) => - new ProviderAdapter.ProviderAdapterEventStreamError({ - driver: entry.runtime.driver, + Effect.catch((cause) => + Effect.logWarning( + "orchestration-v2.provider-session.instruction-invalidation-failed", + { providerSessionId: entry.runtime.providerSessionId, + providerThreadId: event.providerThread.id, cause, - }), + }, + ), ), ); yield* publishToSubscribers(entry.eventSubscribers, { type: "event", event }); diff --git a/packages/provider-acp-registry/src/server/adapter.ts b/packages/provider-acp-registry/src/server/adapter.ts index 1b8b4240fd00e..506b02e9f11d6 100644 --- a/packages/provider-acp-registry/src/server/adapter.ts +++ b/packages/provider-acp-registry/src/server/adapter.ts @@ -1,3 +1,4 @@ +import type * as Digest from "effect/crypto/Digest"; import { extractClaudeSubagentUpdate, extractClaudeSubagentSessionEvent, @@ -309,6 +310,7 @@ export const makeAcpRegistryAdapterV2 = Effect.fn("makeAcpRegistryAdapterV2")(fu }); export type AcpRegistryAdapterV2DriverEnv = + | Digest.Digest | ChildProcessSpawner.ChildProcessSpawner | SecureRandom.SecureRandom | FileSystem.FileSystem diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 9e9eb25526ea4..7849a3e59813e 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -47,6 +47,7 @@ import { import { FILE_HEADERS_ONLY, formatPatch, structuredPatch } from "diff"; import * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; +import * as Digest from "effect/crypto/Digest"; import * as SecureRandom from "effect/crypto/SecureRandom"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; @@ -1684,6 +1685,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( options: AcpAdapterV2Options, ) { const random = yield* SecureRandom.SecureRandom; + const crypto = yield* Digest.Digest; const fileSystem = yield* FileSystem.FileSystem; const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const agentScope = yield* AgentScope; @@ -8006,7 +8008,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( } satisfies T3AcpInstructionState; const instructionDigest = yield* crypto .digest( - "SHA-256", + Digest.sha256, new TextEncoder().encode( t3AcpPromptWithInstructions({ prompt: "", state: instructionState }), ), diff --git a/packages/provider-fx/src/server/adapter.ts b/packages/provider-fx/src/server/adapter.ts index b48293f4fb986..05fbad54a52a3 100644 --- a/packages/provider-fx/src/server/adapter.ts +++ b/packages/provider-fx/src/server/adapter.ts @@ -1,3 +1,4 @@ +import type * as Digest from "effect/crypto/Digest"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { resolveSelfInvocation, type SelfInvocation } from "@t3tools/shared/nodeRuntime"; import { type OrchestrationV2ProviderCapabilities } from "@t3tools/contracts"; @@ -205,6 +206,7 @@ export const makeFxAdapterV2 = Effect.fn("makeFxAdapterV2")(function* ( }); export type FxAdapterV2DriverEnv = + | Digest.Digest | ChildProcessSpawner.ChildProcessSpawner | SecureRandom.SecureRandom | FileSystem.FileSystem diff --git a/packages/provider-grok/src/server/adapter.ts b/packages/provider-grok/src/server/adapter.ts index 9ed1368f16984..6912930216faf 100644 --- a/packages/provider-grok/src/server/adapter.ts +++ b/packages/provider-grok/src/server/adapter.ts @@ -1,3 +1,4 @@ +import type * as Digest from "effect/crypto/Digest"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import { XAiPromptFailureText, @@ -357,6 +358,7 @@ export const makeGrokAdapterV2 = Effect.fn("makeGrokAdapterV2")(function* ( }); export type GrokAdapterV2DriverEnv = + | Digest.Digest | ChildProcessSpawner.ChildProcessSpawner | SecureRandom.SecureRandom | FileSystem.FileSystem diff --git a/packages/provider-kiro/src/server/adapter.ts b/packages/provider-kiro/src/server/adapter.ts index 6815b60e8f227..8f64cfdda5f73 100644 --- a/packages/provider-kiro/src/server/adapter.ts +++ b/packages/provider-kiro/src/server/adapter.ts @@ -1,3 +1,4 @@ +import type * as Digest from "effect/crypto/Digest"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { resolveSelfInvocation, type SelfInvocation } from "@t3tools/shared/nodeRuntime"; import { @@ -589,6 +590,7 @@ export const makeKiroAdapterV2 = Effect.fn("makeKiroAdapterV2")(function* ( }); export type KiroAdapterV2DriverEnv = + | Digest.Digest | ChildProcessSpawner.ChildProcessSpawner | SecureRandom.SecureRandom | FileSystem.FileSystem