diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index f63a220fb5191..b382af3031e83 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -8,7 +8,11 @@ import { } from "@t3tools/provider-acp-registry/testing"; import { AcpRegistrySettings } from "@t3tools/provider-acp-registry/settings"; import { AcpRegistryCatalog } from "@t3tools/provider-acp-registry/server/AcpRegistrySupport"; -import { decodeAcpReplayTranscript, makeAcpReplayRuntime } from "./AcpAdapterV2.testkit.ts"; +import { + decodeAcpReplayTranscript, + makeAcpReplayCompletenessAssertion, + makeAcpReplayRuntime, +} from "./AcpAdapterV2.testkit.ts"; // @effect-diagnostics nodeBuiltinImport:off import * as NodeFS from "node:fs"; import * as NodePath from "node:path"; @@ -2411,6 +2415,180 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + // regression: #27646 an agent that leaves its mode itself must not be forced back, but picking + // the old mode again must reach it, even when a v1 agent reports only the mode update. + it.live("reports a mode the agent left itself so picking it again is sent", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-acp-mode-only-" }); + const statusPath = path.join(directory, "status.json"); + const sessionId = "mode-only-session"; + const modeOption = (currentValue: string) => ({ + id: "mode", + name: "Mode", + category: "mode", + type: "select", + currentValue, + options: [ + { value: "ask", name: "Ask" }, + { value: "architect", name: "Architect" }, + ], + }); + const setArchitect = [ + { + type: "expect_outbound" as const, + frame: { + kind: "request", + method: "session/set_config_option", + params: { sessionId, configId: "mode", value: "architect" }, + }, + }, + { + type: "emit_inbound" as const, + frame: { + kind: "response", + method: "session/set_config_option", + result: { configOptions: [modeOption("architect")] }, + }, + }, + ]; + const prompt = (updates: ReadonlyArray) => [ + { + type: "expect_outbound" as const, + frame: { kind: "request", method: "session/prompt", params: "" }, + }, + ...updates.map((update) => ({ + type: "emit_inbound" as const, + frame: { kind: "notification", method: "session/update", params: { sessionId, update } }, + })), + { + type: "emit_inbound" as const, + frame: { kind: "response", method: "session/prompt", result: { stopReason: "end_turn" } }, + }, + ]; + const transcript = yield* decodeAcpReplayTranscript( + { + provider: ACP_TEST_DRIVER, + protocol: "acp.ndjson-jsonrpc", + version: "1", + scenario: "mode-only-update", + entries: [ + { + type: "expect_outbound", + frame: { kind: "request", method: "initialize", params: "" }, + }, + { + type: "emit_inbound", + frame: { + kind: "response", + method: "initialize", + result: { protocolVersion: 1, agentCapabilities: {}, authMethods: [] }, + }, + }, + { + type: "expect_outbound", + frame: { kind: "request", method: "session/new", params: "" }, + }, + { + type: "emit_inbound", + frame: { + kind: "response", + method: "session/new", + result: { sessionId, configOptions: [modeOption("ask")] }, + }, + }, + ...setArchitect, + ...prompt([{ sessionUpdate: "current_mode_update", currentModeId: "ask" }]), + ...prompt([]), + ...setArchitect, + ...prompt([]), + ], + }, + ACP_TEST_DRIVER, + ); + const instanceId = ProviderInstanceId.make("acp-test-mode-only"); + const threadId = ThreadId.make("thread-acp-mode-only"); + const adapter = yield* makeAcpAdapterV2({ + instanceId, + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + makeRuntime: makeAcpReplayRuntime({ + transcript, + statusPath, + fileSystem, + scriptPath: yield* path.fromFileUrl( + new URL("../../../scripts/acp-replay-agent.ts", import.meta.url), + ), + childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner, + }), + }, + selfInvocation: yield* resolveSelfInvocation(), + }); + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: directory, + }); + const architect = { + instanceId, + model: "default", + options: [{ id: "mode", value: "architect" }], + } as const; + const ask = { ...architect, options: [{ id: "mode", value: "ask" }] }; + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-acp-mode-only"), + modelSelection: architect, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection: architect, + runtimePolicy, + }); + const now = yield* DateTime.now; + const reported: Array = []; + // The thread follows the agent to ask; only the user's renewed pick sets architect again. + for (const [ordinal, modelSelection] of [ + [1, architect], + [2, ask], + [3, architect], + ] as const) { + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now, + ordinal, + modelSelection, + }), + ); + yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runForEach((event) => + Effect.sync(() => { + if (event.type === "thread_model_selection.updated") { + reported.push({ + threadId: event.threadId, + instanceId: event.instanceId, + changes: event.changes, + }); + } + }), + ), + ); + } + assert.deepStrictEqual(reported, [ + { threadId, instanceId, changes: [{ id: "mode", from: "architect", to: "ask" }] }, + ]); + yield* makeAcpReplayCompletenessAssertion(fileSystem, statusPath, transcript); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + // regression: #21499 turns blocked only on T3 thread waits must show as waiting it.effect.each(["agent_message_chunk"] as const)( "keeps named ACP %s open while deriving thread wait execution state", diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index 520fea7c0575b..c90e901c35a79 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -2144,6 +2144,60 @@ layer("ProviderEventIngestorV2", (it) => { }), ); + it.effect("applies an agent's mode change without undoing newer thread settings", () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const eventSink = yield* EventSink.EventSinkV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const rootEvent = yield* threadCreatedEvent(now); + if (rootEvent.type !== "thread.created") throw new Error("Expected thread fixture"); + // Saved mid-turn: a newer model and effort the reporting turn never saw. + const saved = { + ...modelSelection, + model: "gpt-6.1-sol", + options: [ + { id: "effort", value: "high" }, + { id: "mode", value: "plan" }, + ], + }; + yield* eventSink.write({ + events: [{ ...rootEvent, payload: { ...rootEvent.payload, modelSelection: saved } }], + }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId: rootEvent.threadId, + }); + const ingest = (instanceId: ProviderInstanceId, from: string) => + ingestor.ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId: rootEvent.threadId, + event: { + type: "thread_model_selection.updated", + driver: CODEX_DRIVER, + threadId: rootEvent.threadId, + instanceId, + changes: [{ id: "mode", from, to: "default" }], + }, + }); + assert.deepEqual( + (yield* ingest(modelSelection.instanceId, "plan")).map((stored) => stored.event.type), + ["thread.model-selection-updated"], + ); + assert.deepEqual((yield* projectionStore.getThread(rootEvent.threadId)).modelSelection, { + ...saved, + options: [ + { id: "effort", value: "high" }, + { id: "mode", value: "default" }, + ], + }); + assert.deepEqual(yield* ingest(modelSelection.instanceId, "plan"), []); + assert.deepEqual(yield* ingest(ProviderInstanceId.make("other-instance"), "default"), []); + }), + ); + it.effect( "recovers a missing Claude child once and preserves saved metadata on later recovery", () => diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 623fddb491108..70002b7b743af 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -493,6 +493,37 @@ export const layer: Layer.Layer< }), ]; } + case "thread_model_selection.updated": { + const { instanceId, changes } = input.event; + const thread = yield* projections.getThread(input.event.threadId).pipe( + Effect.catchTags({ + ProjectionStoreThreadNotFoundError: () => Effect.succeed(null), + }), + ); + const current = thread?.modelSelection; + if (thread === null || current?.instanceId !== instanceId || !current.options) { + return []; + } + let changed = false; + const options = current.options.map((option) => { + const change = changes.find( + (candidate) => candidate.id === option.id && candidate.from === option.value, + ); + if (change === undefined) return option; + changed = true; + return { ...option, value: change.to }; + }); + if (!changed) return []; + const now = yield* DateTime.now; + return [ + yield* makeDomainEvent(input, { + type: "thread.model-selection-updated", + threadId: thread.id, + payload: { ...thread, modelSelection: { ...current, options }, updatedAt: now }, + occurredAt: now, + }), + ]; + } case "provider_session.updated": return [ yield* makeDomainEvent(input, { @@ -849,7 +880,9 @@ export const layer: Layer.Layer< (effect) => input.event.type === "app_thread.created" ? threadCommands.withLock(input.event.appThread.id, effect) - : effect, + : input.event.type === "thread_model_selection.updated" + ? threadCommands.withLock(input.event.threadId, effect) + : effect, ), ), }); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index e981ab7362d2d..64f18e1da551b 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -408,6 +408,8 @@ export function routeProviderEvent( }, ]; } + case "thread_model_selection.updated": + return [event.threadId === input.threadId, state]; case "provider_thread.updated": { const belongs = state.ownedProviderThreadIds.has(event.providerThread.id) || diff --git a/packages/provider-acp/src/server/AcpSessionRuntime.ts b/packages/provider-acp/src/server/AcpSessionRuntime.ts index 514ac91164153..068a63d300c0f 100644 --- a/packages/provider-acp/src/server/AcpSessionRuntime.ts +++ b/packages/provider-acp/src/server/AcpSessionRuntime.ts @@ -2798,10 +2798,25 @@ const handleSessionUpdate = ); } const parsed = parseSessionUpdateEvent(params); - if (parsed.modeId) { + const modeId = parsed.modeId; + if (modeId) { yield* Ref.update(modeStateRef, (current) => - current === undefined ? current : updateModeState(current, parsed.modeId!), + current === undefined ? current : updateModeState(current, modeId), ); + // A mode-only update would otherwise leave the mode option showing the old mode. + const configOptions = yield* SubscriptionRef.get(configOptionsRef); + const isStaleMode = (option: (typeof configOptions)[number]) => + option.category === "mode" && option.type === "select" && option.currentValue !== modeId; + if (configOptions.some(isStaleMode)) { + yield* SubscriptionRef.set( + configOptionsRef, + configOptions.map((option) => + option.category === "mode" && option.type === "select" + ? { ...option, currentValue: modeId } + : option, + ), + ); + } } for (const event of parsed.events) { if (event._tag === "ToolCallUpdated") { diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index a2c2c429fa88c..1607ae0d8a091 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -1932,6 +1932,24 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( }); const activeSelection = yield* Ref.make(null); const activeInteractionMode = yield* Ref.make(null); + // Mode values configureSession left the agent in; a later difference is the agent's own change. + const configuredModes = yield* Ref.make>(new Map()); + const readModes = Effect.gen(function* () { + const modes = new Map(); + for (const option of yield* runtime.getConfigOptions) { + if ( + option.type === "select" && + (option.category === "mode" || option.category === "collaboration_mode") + ) { + modes.set(option.id, option.currentValue); + } + } + const modeId = (yield* runtime.getModeState)?.currentModeId; + if (modeId !== undefined && !modes.has(ACP_SESSION_MODE_OPTION_ID)) { + modes.set(ACP_SESSION_MODE_OPTION_ID, modeId); + } + return modes; + }); const promptInstructionStates = yield* Ref.make(new Map()); // Unique markers keep delayed prompt completions from clearing later compactions. const compactionMarkers = yield* Ref.make(new Map()); @@ -7466,6 +7484,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( } } yield* validateModelSelection(modelSelection); + yield* Ref.set(configuredModes, yield* readModes); yield* ( flavor.onSessionConfigurationUpdate?.( yield* runtime.getConfigOptions, @@ -7603,6 +7622,39 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ); }); + // Re-applying the thread's mode would undo a deliberate exit, such as leaving plan mode once + // the plan was approved. Report it instead, so picking the old mode again is a real change. + const reportAgentModeChanges = Effect.fnUntraced(function* (threadId: ThreadId) { + const selection = yield* Ref.get(activeSelection); + if (selection === null || selection.options === undefined) return; + const configured = yield* Ref.get(configuredModes); + const live = yield* readModes; + const changes: Array<{ id: string; from: string; to: string }> = []; + const options = selection.options.map((option) => { + const value = live.get(option.id); + if ( + value === undefined || + typeof option.value !== "string" || + value === configured.get(option.id) || + value === option.value + ) { + return option; + } + changes.push({ id: option.id, from: option.value, to: value }); + return { ...option, value }; + }); + if (changes.length === 0) return; + yield* Ref.set(activeSelection, { ...selection, options }); + yield* Ref.set(configuredModes, live); + yield* emitProviderEvent({ + type: "thread_model_selection.updated", + driver, + threadId, + instanceId: selection.instanceId, + changes, + }); + }); + const finalizeTurn = Effect.fnUntraced(function* ( context: ActiveAcpTurn, status: "completed" | "interrupted" | "failed" | "cancelled", @@ -7679,6 +7731,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( }, }); } + yield* reportAgentModeChanges(context.input.threadId); const turn = providerTurnPayload(context, settledStatus, now); yield* Ref.update(providerTurns, (current) => { const updated = new Map(current); diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index e925f10c8f154..589905cf9b17c 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -102,6 +102,17 @@ export const ProviderAdapterV2Event = Schema.Union([ sessionScoped: Schema.optional(Schema.Boolean), retainedNativeTurnIds: Schema.optional(Schema.Array(Schema.String)), }), + /** The agent changed selected options itself, such as leaving plan mode once its plan ran. */ + Schema.Struct({ + type: Schema.Literal("thread_model_selection.updated"), + driver: ProviderDriverKind, + threadId: ThreadId, + instanceId: ProviderInstanceId, + /** Applies only where the thread still holds `from`, so newer user picks win. */ + changes: Schema.Array( + Schema.Struct({ id: Schema.String, from: Schema.String, to: Schema.String }), + ), + }), Schema.Struct({ type: Schema.Literal("provider_turn.updated"), driver: ProviderDriverKind,