diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 79657fd6a901..90fe72c33b2c 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -22,6 +22,7 @@ const emitXAiAskUserQuestion = process.env.T3_ACP_EMIT_XAI_ASK_USER_QUESTION === const emitXAiPromptCompleteThenHang = process.env.T3_ACP_EMIT_XAI_PROMPT_COMPLETE_THEN_HANG === "1"; const emitForeignSessionUpdates = process.env.T3_ACP_EMIT_FOREIGN_SESSION_UPDATES === "1"; const hangPromptForever = process.env.T3_ACP_HANG_PROMPT_FOREVER === "1"; +const emitChildUpdatesWhileHanging = process.env.T3_ACP_EMIT_CHILD_UPDATES_WHILE_HANGING === "1"; const hangFirstPromptForever = process.env.T3_ACP_HANG_FIRST_PROMPT_FOREVER === "1"; const emitLateUpdateAfterCancel = process.env.T3_ACP_EMIT_LATE_UPDATE_AFTER_CANCEL === "1"; const omitXAiPromptCompleteStopReason = @@ -525,6 +526,24 @@ const program = Effect.gen(function* () { } if (hangPromptForever || (hangFirstPromptForever && promptCount === 1)) { + if (emitChildUpdatesWhileHanging) { + // A live child session on a dead root prompt: traffic on the pipe, but + // nothing that says this prompt is still going. + yield* Effect.forkChild( + Effect.gen(function* () { + while (true) { + yield* Effect.sleep("250 millis"); + writeJsonRpcNotification("session/update", { + sessionId: "mock-child-session-1", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "child still talking" }, + }, + }); + } + }), + ); + } return yield* Effect.never; } diff --git a/apps/server/src/provider/acp/AcpRuntimeModel.test.ts b/apps/server/src/provider/acp/AcpRuntimeModel.test.ts index 7682c5f5f9cb..bbe82bf2cb9e 100644 --- a/apps/server/src/provider/acp/AcpRuntimeModel.test.ts +++ b/apps/server/src/provider/acp/AcpRuntimeModel.test.ts @@ -1,5 +1,12 @@ -import { describe, expect, it } from "vite-plus/test"; +import { describe, expect } from "vite-plus/test"; +import { it } from "@effect/vitest"; +import * as Clock from "effect/Clock"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Ref from "effect/Ref"; +import * as TestClock from "effect/testing/TestClock"; import type * as EffectAcpSchema from "effect-acp/schema"; import { @@ -10,8 +17,20 @@ import { parseSessionUpdateEvent, sessionUpdateIsReplay, syntheticLoadSessionResponseFromInitialize, + waitForPromptStreamStall, + type PromptStreamActivity, } from "./AcpRuntimeModel.ts"; +const makeActivityRef = (overrides?: Partial) => + Effect.gen(function* () { + const lastActivityAtMillis = yield* Clock.currentTimeMillis; + return yield* Ref.make({ + lastActivityAtMillis, + inFlightClientRequests: 0, + ...overrides, + }); + }); + describe("AcpRuntimeModel", () => { it("parses session mode state from typed ACP session setup responses", () => { const modeState = parseSessionModeState({ @@ -375,3 +394,71 @@ describe("AcpRuntimeModel", () => { }); }); }); + +describe("waitForPromptStreamStall", () => { + it.effect("reports the idle duration once the agent goes silent", () => + Effect.gen(function* () { + const activityRef = yield* makeActivityRef(); + const stall = yield* Effect.forkChild( + waitForPromptStreamStall({ activityRef, stallAfter: Duration.minutes(10) }), + ); + + yield* TestClock.adjust(Duration.minutes(11)); + + const idleMillis = yield* Fiber.join(stall); + expect(idleMillis).toBeGreaterThanOrEqual(Duration.toMillis(Duration.minutes(10))); + }), + ); + + it.effect("stays quiet while a client request is still in flight", () => + Effect.gen(function* () { + // A slow build behind terminal/wait_for_exit: the agent is waiting on us. + const activityRef = yield* makeActivityRef({ inFlightClientRequests: 1 }); + const stall = yield* Effect.forkChild( + waitForPromptStreamStall({ activityRef, stallAfter: Duration.minutes(10) }), + ); + + yield* TestClock.adjust(Duration.minutes(45)); + + expect(stall.pollUnsafe()).toBeUndefined(); + yield* Fiber.interrupt(stall); + }), + ); + + it.effect("stays quiet while the agent keeps streaming", () => + Effect.gen(function* () { + const activityRef = yield* makeActivityRef(); + const stall = yield* Effect.forkChild( + waitForPromptStreamStall({ activityRef, stallAfter: Duration.minutes(10) }), + ); + + for (let tick = 0; tick < 6; tick += 1) { + yield* TestClock.adjust(Duration.minutes(9)); + const lastActivityAtMillis = yield* Clock.currentTimeMillis; + yield* Ref.update(activityRef, (activity) => ({ ...activity, lastActivityAtMillis })); + } + + expect(stall.pollUnsafe()).toBeUndefined(); + yield* Fiber.interrupt(stall); + }), + ); + + it.effect("reports a stall once the last client request settles", () => + Effect.gen(function* () { + const activityRef = yield* makeActivityRef({ inFlightClientRequests: 1 }); + const stall = yield* Effect.forkChild( + waitForPromptStreamStall({ activityRef, stallAfter: Duration.minutes(10) }), + ); + + yield* TestClock.adjust(Duration.minutes(30)); + expect(stall.pollUnsafe()).toBeUndefined(); + + const lastActivityAtMillis = yield* Clock.currentTimeMillis; + yield* Ref.set(activityRef, { lastActivityAtMillis, inFlightClientRequests: 0 }); + yield* TestClock.adjust(Duration.minutes(11)); + + const idleMillis = yield* Fiber.join(stall); + expect(idleMillis).toBeGreaterThanOrEqual(Duration.toMillis(Duration.minutes(10))); + }), + ); +}); diff --git a/apps/server/src/provider/acp/AcpRuntimeModel.ts b/apps/server/src/provider/acp/AcpRuntimeModel.ts index e6bfc127e6e9..240822be275d 100644 --- a/apps/server/src/provider/acp/AcpRuntimeModel.ts +++ b/apps/server/src/provider/acp/AcpRuntimeModel.ts @@ -487,6 +487,45 @@ export const waitForSessionLoadReplayIdle = (input: { } }); +/** + * Liveness of the agent->client half of a `session/prompt`. `lastActivityAtMillis` + * is bumped by any inbound traffic; `inFlightClientRequests` counts the agent's + * requests we are still serving. + */ +export interface PromptStreamActivity { + readonly lastActivityAtMillis: number; + readonly inFlightClientRequests: number; +} + +/** + * Resolves with the observed idle duration once a prompt has gone silent for + * `stallAfter`. Used to bound `session/prompt`, which otherwise waits forever on + * an agent that stopped answering. + */ +export const waitForPromptStreamStall = (input: { + readonly activityRef: Ref.Ref; + readonly stallAfter: Duration.Duration; +}): Effect.Effect => + Effect.gen(function* () { + const pollInterval = Duration.seconds(1); + const stallAfterMillis = Duration.toMillis(input.stallAfter); + while (true) { + yield* Effect.sleep(pollInterval); + const activity = yield* Ref.get(input.activityRef); + // A request we have not answered yet means the agent is waiting on us, not + // the other way around. A slow build behind terminal/wait_for_exit is the + // common case, and it is not a stall. + if (activity.inFlightClientRequests > 0) { + continue; + } + const nowMillis = yield* Clock.currentTimeMillis; + const idleMillis = nowMillis - activity.lastActivityAtMillis; + if (idleMillis >= stallAfterMillis) { + return idleMillis; + } + } + }); + export function syntheticLoadSessionResponseFromInitialize( initializeResult: EffectAcpSchema.InitializeResponse, ): EffectAcpSchema.LoadSessionResponse { diff --git a/apps/server/src/provider/acp/AcpSessionRuntime.test.ts b/apps/server/src/provider/acp/AcpSessionRuntime.test.ts new file mode 100644 index 000000000000..b2e82456b593 --- /dev/null +++ b/apps/server/src/provider/acp/AcpSessionRuntime.test.ts @@ -0,0 +1,110 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeFS from "node:fs"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; +import * as NodeURL from "node:url"; + +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { it } from "@effect/vitest"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; +import { describe, expect } from "vite-plus/test"; + +import * as AcpSessionRuntime from "./AcpSessionRuntime.ts"; + +const __dirname = NodePath.dirname(NodeURL.fileURLToPath(import.meta.url)); +const mockAgentPath = NodePath.join(__dirname, "../../../scripts/acp-mock-agent.ts"); + +// These drive a real subprocess on the live clock, so `it.live` rather than +// `it.effect`. The watchdog polls once a second, so a stall needs at least two +// polls to be observed: the timeout sits above one poll and the handler holds +// the request across several. +const testStallTimeout = Duration.millis(1500); +const handlerHoldDuration = Duration.seconds(4); + +const makeRuntime = (env: NodeJS.ProcessEnv) => + AcpSessionRuntime.make({ + spawn: { + command: process.execPath, + args: [mockAgentPath], + env, + }, + cwd: process.cwd(), + clientInfo: { name: "t3-test", version: "0.0.0" }, + authMethodId: "test", + promptStallTimeout: testStallTimeout, + }); + +describe("AcpSessionRuntime prompt stall detection", () => { + it.live("keeps waiting while an extension question is parked on the user", () => + Effect.gen(function* () { + const runtime = yield* makeRuntime({ T3_ACP_EMIT_ASK_QUESTION: "1" }); + // A real user takes as long as they take. The agent is not silent because + // it died, it is silent because it is waiting on this answer. + yield* runtime.handleExtRequest("cursor/ask_question", Schema.Unknown, () => + Effect.sleep(handlerHoldDuration).pipe( + Effect.as({ answers: [{ id: "scope", value: "workspace" }] }), + ), + ); + yield* runtime.start(); + + const result = yield* runtime.prompt({ prompt: [{ type: "text", text: "hi" }] }); + + expect(result).toMatchObject({ stopReason: "end_turn" }); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.live("fails the prompt once the agent goes silent with nothing outstanding", () => + Effect.gen(function* () { + const runtime = yield* makeRuntime({ T3_ACP_HANG_PROMPT_FOREVER: "1" }); + yield* runtime.start(); + + const error = yield* runtime + .prompt({ prompt: [{ type: "text", text: "hi" }] }) + .pipe(Effect.flip); + + expect(error._tag).toBe("AcpTransportError"); + expect(error).toMatchObject({ method: "session/prompt" }); + expect(String((error as { detail?: string }).detail)).toContain("stalled"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.live("does not accept a chattering child session as proof this prompt is alive", () => + Effect.gen(function* () { + const runtime = yield* makeRuntime({ + T3_ACP_HANG_PROMPT_FOREVER: "1", + T3_ACP_EMIT_CHILD_UPDATES_WHILE_HANGING: "1", + }); + yield* runtime.start(); + + const error = yield* runtime + .prompt({ prompt: [{ type: "text", text: "hi" }] }) + .pipe(Effect.flip); + + expect(error._tag).toBe("AcpTransportError"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.live("cancels the wedged prompt so the agent can release it", () => + Effect.gen(function* () { + const requestLogPath = NodePath.join( + yield* Effect.sync(() => NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-acp-"))), + "requests.ndjson", + ); + const runtime = yield* makeRuntime({ + T3_ACP_HANG_PROMPT_FOREVER: "1", + T3_ACP_REQUEST_LOG_PATH: requestLogPath, + }); + yield* runtime.start(); + + yield* runtime.prompt({ prompt: [{ type: "text", text: "hi" }] }).pipe(Effect.flip); + // The notification is fired off as the prompt fails, so give the write a + // moment to land before reading the agent's view of what it received. + yield* Effect.sleep(Duration.millis(250)); + + const received = yield* Effect.sync(() => NodeFS.readFileSync(requestLogPath, "utf8")); + expect(received).toContain("session/cancel"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); +}); diff --git a/apps/server/src/provider/acp/AcpSessionRuntime.ts b/apps/server/src/provider/acp/AcpSessionRuntime.ts index 33edf0947735..f70183caa630 100644 --- a/apps/server/src/provider/acp/AcpSessionRuntime.ts +++ b/apps/server/src/provider/acp/AcpSessionRuntime.ts @@ -10,6 +10,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; @@ -29,7 +30,9 @@ import { parseSessionModeState, parseSessionUpdateEvent, sessionUpdateIsReplay, + waitForPromptStreamStall, waitForSessionLoadReplayIdle, + type PromptStreamActivity, type SessionLoadGate, type AcpParsedSessionEvent, type AcpSessionModeState, @@ -49,6 +52,10 @@ export type AcpSessionRuntimeEvent = AcpParsedSessionEvent | AcpSessionEventStre const defaultSessionLoadTimeout = Duration.seconds(90); const defaultSessionLoadReplayIdleGap = Duration.seconds(2); +// Healthy turns stream something every few seconds. Ten minutes of total +// silence, with nothing of ours outstanding, is a dead stream rather than a slow +// one. The threshold is generous so long reasoning is never mistaken for it. +const defaultPromptStallTimeout = Duration.minutes(10); export interface AcpSpawnInput { readonly command: string; @@ -63,6 +70,7 @@ export interface AcpSessionRuntimeOptions { readonly resumeSessionId?: string; readonly sessionLoadTimeout?: Duration.Input; readonly sessionLoadReplayIdleGap?: Duration.Input; + readonly promptStallTimeout?: Duration.Input; readonly clientCapabilities?: EffectAcpSchema.InitializeRequest["clientCapabilities"]; readonly clientInfo: { readonly name: string; @@ -299,6 +307,51 @@ export const make = ( Option.Option> >(Option.none()); const sessionLoadGateRef = yield* Ref.make>(Option.none()); + const promptStreamActivityRef = yield* Ref.make({ + lastActivityAtMillis: yield* Clock.currentTimeMillis, + inFlightClientRequests: 0, + }); + const promptStallTimeout = Duration.fromInputUnsafe( + options.promptStallTimeout ?? defaultPromptStallTimeout, + ); + + const touchPromptStreamActivity = Effect.gen(function* () { + const nowMillis = yield* Clock.currentTimeMillis; + yield* Ref.update(promptStreamActivityRef, (activity) => ({ + ...activity, + lastActivityAtMillis: nowMillis, + })); + }); + + const adjustInFlightClientRequests = (delta: number) => + Effect.gen(function* () { + const nowMillis = yield* Clock.currentTimeMillis; + yield* Ref.update(promptStreamActivityRef, (activity) => ({ + lastActivityAtMillis: nowMillis, + inFlightClientRequests: Math.max(0, activity.inFlightClientRequests + delta), + })); + }); + + /** + * Holds an agent-initiated request open for as long as it runs, so time spent + * serving the agent never reads as a dead stream. The long ones wait on a + * human: permission prompts and extension questions such as + * `cursor/ask_question`. + */ + const countClientRequest = (effect: Effect.Effect) => + adjustInFlightClientRequests(1).pipe( + Effect.flatMap(() => effect), + Effect.ensuring(adjustInFlightClientRequests(-1)), + ); + + const trackClientRequest = + ( + register: ( + handler: (request: Request) => Effect.Effect, + ) => Effect.Effect, + ) => + (handler: (request: Request) => Effect.Effect) => + register((request) => countClientRequest(handler(request))); const logRequest = (event: AcpSessionRequestLogEvent) => options.requestLogger ? options.requestLogger(event) : Effect.void; @@ -395,6 +448,9 @@ export const make = ( ) { return; } + // Only our own session counts as our liveness. A child session chattering + // on the same pipe says nothing about whether this prompt is still alive. + yield* touchPromptStreamActivity; yield* handleSessionUpdate({ queue: eventQueue, modeStateRef, @@ -692,21 +748,45 @@ export const make = ( }); return { - handleRequestPermission: acp.handleRequestPermission, - handleElicitation: acp.handleElicitation, - handleReadTextFile: acp.handleReadTextFile, - handleWriteTextFile: acp.handleWriteTextFile, - handleCreateTerminal: acp.handleCreateTerminal, - handleTerminalOutput: acp.handleTerminalOutput, - handleTerminalWaitForExit: acp.handleTerminalWaitForExit, - handleTerminalKill: acp.handleTerminalKill, - handleTerminalRelease: acp.handleTerminalRelease, + handleRequestPermission: trackClientRequest(acp.handleRequestPermission), + handleElicitation: trackClientRequest(acp.handleElicitation), + handleReadTextFile: trackClientRequest(acp.handleReadTextFile), + handleWriteTextFile: trackClientRequest(acp.handleWriteTextFile), + handleCreateTerminal: trackClientRequest(acp.handleCreateTerminal), + handleTerminalOutput: trackClientRequest(acp.handleTerminalOutput), + handleTerminalWaitForExit: trackClientRequest(acp.handleTerminalWaitForExit), + handleTerminalKill: trackClientRequest(acp.handleTerminalKill), + handleTerminalRelease: trackClientRequest(acp.handleTerminalRelease), handleSessionUpdate: acp.handleSessionUpdate, handleElicitationComplete: acp.handleElicitationComplete, - handleUnknownExtRequest: acp.handleUnknownExtRequest, - handleUnknownExtNotification: acp.handleUnknownExtNotification, - handleExtRequest: acp.handleExtRequest, - handleExtNotification: acp.handleExtNotification, + handleUnknownExtRequest: ( + handler: ( + method: string, + params: unknown, + ) => Effect.Effect, + ) => + acp.handleUnknownExtRequest((method, params) => + countClientRequest(handler(method, params)), + ), + handleUnknownExtNotification: ( + handler: (method: string, params: unknown) => Effect.Effect, + ) => + acp.handleUnknownExtNotification((method, params) => + touchPromptStreamActivity.pipe(Effect.flatMap(() => handler(method, params))), + ), + handleExtRequest: ( + method: string, + payload: Schema.Codec, + handler: (payload: A) => Effect.Effect, + ) => acp.handleExtRequest(method, payload, (parsed) => countClientRequest(handler(parsed))), + handleExtNotification: ( + method: string, + payload: Schema.Codec, + handler: (payload: A) => Effect.Effect, + ) => + acp.handleExtNotification(method, payload, (parsed) => + touchPromptStreamActivity.pipe(Effect.flatMap(() => handler(parsed))), + ), start: () => start, getEvents: () => Stream.fromQueue(eventQueue), drainEvents: Effect.gen(function* () { @@ -734,20 +814,49 @@ export const make = ( const cancelledResponse = { stopReason: "cancelled", } satisfies EffectAcpSchema.PromptResponse; + yield* touchPromptStreamActivity; const promptRpcFiber = yield* runLoggedRequest( "session/prompt", requestPayload, acp.agent.prompt(requestPayload), ).pipe(Effect.forkIn(runtimeScope)); yield* Ref.set(activePromptFiberRef, Option.some(promptRpcFiber)); - return yield* Fiber.join(promptRpcFiber).pipe( - Effect.catchCause((cause) => - Cause.hasInterruptsOnly(cause) - ? Effect.succeed(cancelledResponse) - : Effect.failCause(cause), + const stallFiber = yield* waitForPromptStreamStall({ + activityRef: promptStreamActivityRef, + stallAfter: promptStallTimeout, + }).pipe(Effect.forkIn(runtimeScope)); + return yield* Effect.raceFirst( + Fiber.join(promptRpcFiber).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.succeed(cancelledResponse) + : Effect.failCause(cause), + ), + ), + Fiber.join(stallFiber).pipe( + Effect.flatMap((idleMillis) => + Effect.gen(function* () { + // The agent process normally outlives the wedged request, so + // cancelling gives it a chance to drop the dead prompt and stay + // usable for the next turn instead of being respawned. + yield* acp.agent + .cancel({ sessionId: started.sessionId }) + .pipe(Effect.ignore, Effect.forkIn(runtimeScope)); + return yield* new EffectAcpErrors.AcpTransportError({ + operation: "call-rpc", + method: "session/prompt", + detail: `session/prompt stalled: the agent sent nothing for ${Math.round( + idleMillis / 1000, + )}s with no client request outstanding`, + cause: undefined, + }); + }), + ), ), + ).pipe( Effect.ensuring( Effect.gen(function* () { + yield* Fiber.interrupt(stallFiber).pipe(Effect.ignore); yield* Fiber.interrupt(promptRpcFiber).pipe(Effect.ignore); yield* Ref.set(activePromptFiberRef, Option.none()); }), diff --git a/docs/internals/providers.md b/docs/internals/providers.md index a309d70f03de..f36e28b2c15e 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -75,6 +75,31 @@ spills the whole accumulated text as one delta. The buffer also flushes at inter when a request opens (approval) or user input is requested, via `flushBufferedAssistantMessagesForTurn`. +### Stalled prompt detection + +`session/prompt` is a long-lived request: ACP agents answer it only once the whole turn is done, and +nothing in the protocol says how long that takes. An agent whose upstream connection dies mid-turn +never answers and never errors, so [`AcpSessionRuntime`][acpruntime] races the RPC against a +liveness watchdog and fails the turn instead of waiting forever. + +Liveness is traffic for this session plus outstanding work, and both halves are load-bearing. + +Scoping matters because one runtime projects one root session: a child session chattering on the +same pipe says nothing about whether the root prompt is alive, so only updates that pass the +root-session check refresh the stamp. Counting outstanding requests matters because silence is +often our fault, not the agent's. Every request the agent makes of us is held open while it runs, +including the extension requests, which is the case worth stating: `cursor/ask_question` and +`x.ai/ask_user_question` park on a human, and a user who takes fifteen minutes to answer must not +look like a dead agent. The same holds for a twenty-minute `terminal/wait_for_exit`. + +A stall therefore needs no root-session traffic _and_ nothing of ours outstanding, for +`promptStallTimeout` (ten minutes by default). + +On a stall the runtime sends `session/cancel` so the agent can release the dead prompt and stay +usable, then fails with an `AcpTransportError`. `ProviderCommandReactor` turns that into a thread +session error with a `provider.turn.start.failed` activity and clears `activeTurnId`, so the working +indicator stops and the reason is visible in the timeline. + [drivers]: ../../apps/server/src/provider/builtInDrivers.ts [codex]: ../../apps/server/src/provider/Drivers/CodexDriver.ts [claude]: ../../apps/server/src/provider/Drivers/ClaudeDriver.ts @@ -88,5 +113,6 @@ when a request opens (approval) or user input is requested, via [contracts]: ../../packages/contracts/src/orchestration.ts [worker]: ../../packages/shared/src/DrainableWorker.ts [ingest]: ../../apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +[acpruntime]: ../../apps/server/src/provider/acp/AcpSessionRuntime.ts [cmd]: ../../apps/server/src/orchestration/Layers/ProviderCommandReactor.ts [checkpoint]: ../../apps/server/src/orchestration/Layers/CheckpointReactor.ts