fix(server): a silent ACP agent no longer hangs a turn forever

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;
       }
 

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<PromptStreamActivity>) =>
+  Effect.gen(function* () {
+    const lastActivityAtMillis = yield* Clock.currentTimeMillis;
+    return yield* Ref.make<PromptStreamActivity>({
+      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)));
+    }),
+  );
+});

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<PromptStreamActivity>;
+  readonly stallAfter: Duration.Duration;
+}): Effect.Effect<number, never> =>
+  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 {

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)),
+  );
+});