From a13648f086eb6aed13da3723f85976334b58e324 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 18 Aug 2026 15:16:28 -0400 Subject: [PATCH 1/3] fix(server): a silent ACP agent no longer hangs a turn forever session/prompt has no deadline, so an agent whose upstream connection dies mid-turn leaves the turn running with nothing behind it. The working indicator counts from turn start, and the session reaper deliberately skips threads with an active turn, so nothing else could notice or end it. Signed-off-by: Yordis Prieto --- .../src/provider/acp/AcpRuntimeModel.test.ts | 89 ++++++++++++++- .../src/provider/acp/AcpRuntimeModel.ts | 39 +++++++ .../src/provider/acp/AcpSessionRuntime.ts | 107 +++++++++++++++--- docs/internals/providers.md | 20 ++++ 4 files changed, 240 insertions(+), 15 deletions(-) diff --git a/apps/server/src/provider/acp/AcpRuntimeModel.test.ts b/apps/server/src/provider/acp/AcpRuntimeModel.test.ts index 7682c5f5f9cb..e77ff3625264 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))); + }).pipe(Effect.provide(TestClock.layer())), + ); + + 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); + }).pipe(Effect.provide(TestClock.layer())), + ); + + 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); + }).pipe(Effect.provide(TestClock.layer())), + ); + + 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))); + }).pipe(Effect.provide(TestClock.layer())), + ); +}); 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.ts b/apps/server/src/provider/acp/AcpSessionRuntime.ts index 33edf0947735..be77f3ecdb40 100644 --- a/apps/server/src/provider/acp/AcpSessionRuntime.ts +++ b/apps/server/src/provider/acp/AcpSessionRuntime.ts @@ -29,7 +29,9 @@ import { parseSessionModeState, parseSessionUpdateEvent, sessionUpdateIsReplay, + waitForPromptStreamStall, waitForSessionLoadReplayIdle, + type PromptStreamActivity, type SessionLoadGate, type AcpParsedSessionEvent, type AcpSessionModeState, @@ -49,6 +51,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 +69,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 +306,48 @@ 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), + })); + }); + + /** + * Wraps a handler registration so serving the agent counts as liveness. Without + * this a long command would look identical to a dead stream. + */ + const trackClientRequest = + ( + register: ( + handler: (request: Request) => Effect.Effect, + ) => Effect.Effect, + ) => + (handler: (request: Request) => Effect.Effect) => + register((request) => + adjustInFlightClientRequests(1).pipe( + Effect.flatMap(() => handler(request)), + Effect.ensuring(adjustInFlightClientRequests(-1)), + ), + ); const logRequest = (event: AcpSessionRequestLogEvent) => options.requestLogger ? options.requestLogger(event) : Effect.void; @@ -371,6 +420,7 @@ export const make = ( yield* acp.handleSessionUpdate((notification) => Effect.gen(function* () { + yield* touchPromptStreamActivity; const gate = yield* Ref.get(sessionLoadGateRef); if (Option.isSome(gate) && gate.value.active) { const lastActivityAtMillis = yield* Clock.currentTimeMillis; @@ -692,15 +742,15 @@ 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, @@ -734,20 +784,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..3f7902109016 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -75,6 +75,25 @@ 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 inbound traffic plus outstanding work. Any `session/update` bumps the last-activity +stamp, and every client-side handler the agent can call (`fs/read_text_file`, `terminal/create`, +`terminal/wait_for_exit`, permission requests, and the rest) is counted while it runs. Silence alone +is not a stall: an agent blocked on a twenty-minute `terminal/wait_for_exit` is waiting on us, and +the in-flight count keeps the watchdog quiet. A stall needs both no 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 +107,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 From 8ab0bd6e1f7a51e9ccb74b6c13ffb92814e7310b Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 18 Aug 2026 15:47:44 -0400 Subject: [PATCH 2/3] fix(server): a parked extension question no longer reads as a dead stream The stall watchdog only counted the built-in ACP handlers, so an agent waiting on cursor/ask_question or x.ai/ask_user_question looked identical to one that had died. A user who took ten minutes to answer would have had their live turn cancelled out from under them. Signed-off-by: Yordis Prieto --- .../src/provider/acp/AcpRuntimeModel.test.ts | 8 +-- .../provider/acp/AcpSessionRuntime.test.ts | 70 +++++++++++++++++++ .../src/provider/acp/AcpSessionRuntime.ts | 52 ++++++++++---- docs/internals/providers.md | 13 ++-- 4 files changed, 121 insertions(+), 22 deletions(-) create mode 100644 apps/server/src/provider/acp/AcpSessionRuntime.test.ts diff --git a/apps/server/src/provider/acp/AcpRuntimeModel.test.ts b/apps/server/src/provider/acp/AcpRuntimeModel.test.ts index e77ff3625264..bbe82bf2cb9e 100644 --- a/apps/server/src/provider/acp/AcpRuntimeModel.test.ts +++ b/apps/server/src/provider/acp/AcpRuntimeModel.test.ts @@ -407,7 +407,7 @@ describe("waitForPromptStreamStall", () => { const idleMillis = yield* Fiber.join(stall); expect(idleMillis).toBeGreaterThanOrEqual(Duration.toMillis(Duration.minutes(10))); - }).pipe(Effect.provide(TestClock.layer())), + }), ); it.effect("stays quiet while a client request is still in flight", () => @@ -422,7 +422,7 @@ describe("waitForPromptStreamStall", () => { expect(stall.pollUnsafe()).toBeUndefined(); yield* Fiber.interrupt(stall); - }).pipe(Effect.provide(TestClock.layer())), + }), ); it.effect("stays quiet while the agent keeps streaming", () => @@ -440,7 +440,7 @@ describe("waitForPromptStreamStall", () => { expect(stall.pollUnsafe()).toBeUndefined(); yield* Fiber.interrupt(stall); - }).pipe(Effect.provide(TestClock.layer())), + }), ); it.effect("reports a stall once the last client request settles", () => @@ -459,6 +459,6 @@ describe("waitForPromptStreamStall", () => { const idleMillis = yield* Fiber.join(stall); expect(idleMillis).toBeGreaterThanOrEqual(Duration.toMillis(Duration.minutes(10))); - }).pipe(Effect.provide(TestClock.layer())), + }), ); }); 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..cdc395dadbe8 --- /dev/null +++ b/apps/server/src/provider/acp/AcpSessionRuntime.test.ts @@ -0,0 +1,70 @@ +// @effect-diagnostics nodeBuiltinImport:off +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)), + ); +}); diff --git a/apps/server/src/provider/acp/AcpSessionRuntime.ts b/apps/server/src/provider/acp/AcpSessionRuntime.ts index be77f3ecdb40..78793f44ff4d 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"; @@ -332,9 +333,17 @@ export const make = ( }); /** - * Wraps a handler registration so serving the agent counts as liveness. Without - * this a long command would look identical to a dead stream. + * 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: ( @@ -342,12 +351,7 @@ export const make = ( ) => Effect.Effect, ) => (handler: (request: Request) => Effect.Effect) => - register((request) => - adjustInFlightClientRequests(1).pipe( - Effect.flatMap(() => handler(request)), - Effect.ensuring(adjustInFlightClientRequests(-1)), - ), - ); + register((request) => countClientRequest(handler(request))); const logRequest = (event: AcpSessionRequestLogEvent) => options.requestLogger ? options.requestLogger(event) : Effect.void; @@ -753,10 +757,34 @@ export const make = ( 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* () { diff --git a/docs/internals/providers.md b/docs/internals/providers.md index 3f7902109016..1f253775afed 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -82,12 +82,13 @@ nothing in the protocol says how long that takes. An agent whose upstream connec 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 inbound traffic plus outstanding work. Any `session/update` bumps the last-activity -stamp, and every client-side handler the agent can call (`fs/read_text_file`, `terminal/create`, -`terminal/wait_for_exit`, permission requests, and the rest) is counted while it runs. Silence alone -is not a stall: an agent blocked on a twenty-minute `terminal/wait_for_exit` is waiting on us, and -the in-flight count keeps the watchdog quiet. A stall needs both no traffic and nothing of ours -outstanding, for `promptStallTimeout` (ten minutes by default). +Liveness is inbound traffic plus outstanding work. Any inbound notification bumps the last-activity +stamp, and every request the agent can make of us is counted while it runs. That includes 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`. Silence alone is +never a stall, because in both cases the agent is waiting on us. A stall needs no 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 From 84f95bb3293c42e675240435a0b3f72af75d6c2e Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 18 Aug 2026 16:02:03 -0400 Subject: [PATCH 3/3] fix(server): a chattering child session no longer proves the root prompt is alive A delegating agent can keep a wedged root prompt looking healthy forever, which is exactly the hang the watchdog exists to catch. Signed-off-by: Yordis Prieto --- apps/server/scripts/acp-mock-agent.ts | 19 +++++++++ .../provider/acp/AcpSessionRuntime.test.ts | 40 +++++++++++++++++++ .../src/provider/acp/AcpSessionRuntime.ts | 4 +- docs/internals/providers.md | 17 +++++--- 4 files changed, 73 insertions(+), 7 deletions(-) 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/AcpSessionRuntime.test.ts b/apps/server/src/provider/acp/AcpSessionRuntime.test.ts index cdc395dadbe8..b2e82456b593 100644 --- a/apps/server/src/provider/acp/AcpSessionRuntime.test.ts +++ b/apps/server/src/provider/acp/AcpSessionRuntime.test.ts @@ -1,4 +1,6 @@ // @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"; @@ -67,4 +69,42 @@ describe("AcpSessionRuntime prompt stall detection", () => { 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 78793f44ff4d..f70183caa630 100644 --- a/apps/server/src/provider/acp/AcpSessionRuntime.ts +++ b/apps/server/src/provider/acp/AcpSessionRuntime.ts @@ -424,7 +424,6 @@ export const make = ( yield* acp.handleSessionUpdate((notification) => Effect.gen(function* () { - yield* touchPromptStreamActivity; const gate = yield* Ref.get(sessionLoadGateRef); if (Option.isSome(gate) && gate.value.active) { const lastActivityAtMillis = yield* Clock.currentTimeMillis; @@ -449,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, diff --git a/docs/internals/providers.md b/docs/internals/providers.md index 1f253775afed..f36e28b2c15e 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -82,13 +82,18 @@ nothing in the protocol says how long that takes. An agent whose upstream connec 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 inbound traffic plus outstanding work. Any inbound notification bumps the last-activity -stamp, and every request the agent can make of us is counted while it runs. That includes the -extension requests, which is the case worth stating: `cursor/ask_question` and +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`. Silence alone is -never a stall, because in both cases the agent is waiting on us. A stall needs no traffic _and_ -nothing of ours outstanding, for `promptStallTimeout` (ten minutes by default). +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