fix(agents): keep stalled turns visible after Responses streams settle (#120426)

* fix(agents): keep stalled turns visible after Responses streams settle

A bare-continue turn in live QA (goal-followthrough-live, gpt-5.4 via
openai-responses) completed its SSE requests and then produced no terminal
result, delivery, timeout, or error for 6+ minutes until SIGTERM. Root cause:
turn liveness is enforced only while awaiting provider stream events (llm-idle
watchdog); queued subscription handlers are fire-and-forget during the turn and
finalize joined them unbounded and un-abortable, backed only by the 48h default
run budget. One hung delivery handler silently dead-ended the whole turn.

- Bound the pending-events join in attempt-stream-finalize with a 120s liveness
  deadline plus the run-abort signal; on expiry, warn with the runId and proceed
  to settlement so the run always yields a visible terminal outcome.
- Responses transports now report every SSE event via notifyLlmRequestActivity
  (parity with completions/anthropic), so bookkeeping-only events keep the idle
  watchdog quiet instead of counting as network silence.

* fix(agents): bound the settlement block-reply flush with the shared liveness join

ClawSweeper P1 on #120426: after the finalize-phase join times out, settlement
still awaited onBlockReplyFlush on the same wedged delivery chain (unbounded on
the supported blockReplyTimeoutMs: 0 path). Generalize the bounded join into
joinWithRunLivenessDeadline in run/abortable.ts (owner of abort/liveness
racing) and use it for both the pending-events join and the settle flush;
timeout and abort resolve with a recorded warning so the turn always reaches a
visible terminal outcome. New coverage: helper tests (hang, abort, rejection)
and a real-settle-path regression holding the flush past the deadline.

packages/ai/src/transports/openai-responses-stream-activity.test.ts

@@ -0,0 +1,77 @@
+// Responses streams must report every SSE event as request activity so the
+// embedded-runner idle watchdog stays quiet while bookkeeping-only events
+// (in_progress, *.done echoes) arrive, matching the completions and anthropic
+// transports.
+import { describe, expect, it, vi } from "vitest";
+import type { AssistantMessage, Model } from "../types.js";
+import { onLlmRequestActivity } from "../utils/llm-request-activity.js";
+import {
+  processResponsesStream,
+  type OpenAIResponsesStreamEvent,
+} from "./openai-responses-stream-internal.js";
+
+const model = {
+  id: "gpt-5.6-luna",
+  name: "GPT-5.6 Luna",
+  api: "openai-responses",
+  provider: "openai",
+  baseUrl: "https://api.openai.com/v1",
+  reasoning: true,
+  input: ["text"],
+  cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+  contextWindow: 200_000,
+  maxTokens: 8192,
+} satisfies Model<"openai-responses">;
+
+function createOutput(): AssistantMessage {
+  return {
+    role: "assistant",
+    content: [],
+    api: model.api,
+    provider: model.provider,
+    model: model.id,
+    usage: {
+      input: 0,
+      output: 0,
+      cacheRead: 0,
+      cacheWrite: 0,
+      totalTokens: 0,
+      cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+    },
+    stopReason: "stop",
+    timestamp: 0,
+  };
+}
+
+async function* eventStream(
+  events: readonly Record<string, unknown>[],
+): AsyncGenerator<OpenAIResponsesStreamEvent> {
+  for (const event of events) {
+    yield event as OpenAIResponsesStreamEvent;
+  }
+}
+
+describe("processResponsesStream request activity", () => {
+  it("notifies request activity for every SSE event, including ignored ones", async () => {
+    const abortController = new AbortController();
+    const onActivity = vi.fn();
+    const unsubscribe = onLlmRequestActivity(abortController.signal, onActivity);
+    try {
+      const events: Record<string, unknown>[] = [
+        { type: "response.created", response: { id: "resp_activity" } },
+        // Ignored bookkeeping event: no consumer-visible event is pushed.
+        { type: "response.in_progress", response: { id: "resp_activity" } },
+        {
+          type: "response.completed",
+          response: { id: "resp_activity", status: "completed", output: [] },
+        },
+      ];
+      await processResponsesStream(eventStream(events), createOutput(), { push: () => {} }, model, {
+        signal: abortController.signal,
+      });
+      expect(onActivity).toHaveBeenCalledTimes(events.length);
+    } finally {
+      unsubscribe();
+    }
+  });
+});

packages/ai/src/transports/openai-responses-stream-internal.ts

@@ -20,6 +20,7 @@ import {
 } from "../providers/openai-responses-tool-call-tracker.js";
 import type { Api, AssistantMessage, Model, TextContent, ToolCall, Usage } from "../types.js";
 import { parseStreamingJson } from "../utils/json-parse.js";
+import { notifyLlmRequestActivity } from "../utils/llm-request-activity.js";
 import {
   type FirstStreamEventInternalOptions,
   withFirstStreamEventTimeout,
@@ -287,6 +288,10 @@ export async function processResponsesStream<TApi extends Api>(
   );
   try {
     for await (const event of guardedStream) {
+      // Bookkeeping-only SSE events (in_progress, *.done echoes) are still
+      // provider progress; keep the idle watchdog alive without exposing them,
+      // matching the completions and anthropic transports.
+      notifyLlmRequestActivity(options?.signal);
       if (event.type === "response.created") {
         output.responseId = event.response.id;
       } else if (event.type === "response.output_item.added") {

src/agents/embedded-agent-runner/run/abortable.test.ts

@@ -1,6 +1,10 @@
 // Coverage for abort-aware promise wrapping in embedded attempts.
-import { describe, expect, it } from "vitest";
-import { abortable } from "./abortable.js";
+import { describe, expect, it, vi } from "vitest";
+import {
+  abortable,
+  joinWithRunLivenessDeadline,
+  RUN_LIVENESS_JOIN_TIMEOUT_MS,
+} from "./abortable.js";
 
 describe("abortable", () => {
   it("rejects with AbortError when signal aborts before inner settles", async () => {
@@ -30,3 +34,52 @@ describe("abortable", () => {
     await expect(abortable(ac.signal, Promise.resolve(42))).resolves.toBe(42);
   });
 });
+
+describe("joinWithRunLivenessDeadline", () => {
+  it("resolves when the joined work settles, without firing onTimeout", async () => {
+    const ac = new AbortController();
+    const onTimeout = vi.fn();
+    await joinWithRunLivenessDeadline({
+      joinWork: () => Promise.resolve(),
+      runAbortSignal: ac.signal,
+      onTimeout,
+    });
+    expect(onTimeout).not.toHaveBeenCalled();
+  });
+
+  it("resolves at the liveness deadline when the joined work hangs", async () => {
+    vi.useFakeTimers();
+    try {
+      const ac = new AbortController();
+      const onTimeout = vi.fn();
+      const join = joinWithRunLivenessDeadline({
+        joinWork: () => new Promise<never>(() => {}),
+        runAbortSignal: ac.signal,
+        onTimeout,
+      });
+      await vi.advanceTimersByTimeAsync(RUN_LIVENESS_JOIN_TIMEOUT_MS);
+      await join;
+      expect(onTimeout).toHaveBeenCalledOnce();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("resolves immediately on an aborted run signal and treats rejection as settled", async () => {
+    const aborted = new AbortController();
+    aborted.abort();
+    const onTimeout = vi.fn();
+    await joinWithRunLivenessDeadline({
+      joinWork: () => new Promise<never>(() => {}),
+      runAbortSignal: aborted.signal,
+      onTimeout,
+    });
+    const ac = new AbortController();
+    await joinWithRunLivenessDeadline({
+      joinWork: () => Promise.reject(new Error("delivery chain error already logged")),
+      runAbortSignal: ac.signal,
+      onTimeout,
+    });
+    expect(onTimeout).not.toHaveBeenCalled();
+  });
+});

src/agents/embedded-agent-runner/run/abortable.ts

@@ -31,6 +31,60 @@ function makeAbortError(signal: AbortSignal): Error {
   return tagAsAbortableWrapper(err);
 }
 
+// Post-turn joins (pending subscription handlers, block-reply flush) ride
+// delivery chains that can wedge; the default run budget is 48h, so an
+// unbounded await there dead-ends the turn with no visible outcome. 120s
+// matches the cloud llm-idle class: anything quiet longer is a stuck lane,
+// not legitimate delivery work.
+export const RUN_LIVENESS_JOIN_TIMEOUT_MS = 120_000;
+
+/**
+ * Awaits post-turn work that must never dead-end the run: races the joined
+ * promise against the run-abort signal and a liveness deadline. Timeout and
+ * abort RESOLVE (timeout after `onTimeout`) instead of rejecting so settlement
+ * still produces a visible terminal outcome; rejections also resolve because
+ * the joined chains own their error logging.
+ */
+export function joinWithRunLivenessDeadline(input: {
+  joinWork: () => Promise<void> | void;
+  runAbortSignal: AbortSignal;
+  timeoutMs?: number;
+  onTimeout: () => void;
+}): Promise<void> {
+  return new Promise<void>((resolve) => {
+    let settled = false;
+    const finish = (reason: "settled" | "timeout" | "abort") => {
+      if (settled) {
+        return;
+      }
+      settled = true;
+      clearTimeout(timer);
+      input.runAbortSignal.removeEventListener("abort", onAbort);
+      if (reason === "timeout") {
+        input.onTimeout();
+      }
+      resolve();
+    };
+    const onAbort = () => finish("abort");
+    const timer = setTimeout(
+      () => finish("timeout"),
+      input.timeoutMs ?? RUN_LIVENESS_JOIN_TIMEOUT_MS,
+    );
+    timer.unref?.();
+    if (input.runAbortSignal.aborted) {
+      finish("abort");
+      return;
+    }
+    input.runAbortSignal.addEventListener("abort", onAbort, { once: true });
+    Promise.resolve()
+      .then(() => input.joinWork())
+      .then(
+        () => finish("settled"),
+        () => finish("settled"),
+      );
+  });
+}
+
 /**
  * Races a promise against an AbortSignal while preserving normal promise
  * settlement. Abort wins immediately and rejected non-Error payloads are

src/agents/embedded-agent-runner/run/attempt-stream-finalize.test.ts

@@ -246,6 +246,86 @@ describe("finalizeEmbeddedAttemptStreamPhase", () => {
     );
   });
 
+  it("proceeds to settlement when pending subscription events never settle", async () => {
+    vi.useFakeTimers();
+    try {
+      const fixture = createFixture();
+      // A hung delivery handler must not dead-end the turn until the run budget.
+      fixture.input.waitForPendingEvents = vi.fn(() => new Promise<never>(() => {}));
+      const settledStream = {
+        promptError: null,
+        promptErrorSource: null,
+        timedOutDuringCompaction: false,
+        compactionOccurredThisAttempt: false,
+        messagesSnapshot: [],
+        sessionIdUsed: "session-1",
+        lastAssistant: undefined,
+        currentAttemptAssistant: undefined,
+        currentAttemptCompletedAssistant: undefined,
+        attemptUsage: undefined,
+        cacheBreak: null,
+        lastCallUsage: undefined,
+        promptCache: undefined,
+      };
+      mocks.settleStream.mockResolvedValue(settledStream);
+      mocks.completeAfterTurn.mockResolvedValue({
+        sessionIdUsed: "session-1",
+        sessionFileUsed: "session.jsonl",
+      });
+
+      const finalize = finalizeEmbeddedAttemptStreamPhase(fixture.input);
+      await vi.advanceTimersByTimeAsync(119_999);
+      expect(mocks.settleStream).not.toHaveBeenCalled();
+      await vi.advanceTimersByTimeAsync(1);
+      await expect(finalize).resolves.toEqual({
+        sessionIdUsed: "session-1",
+        sessionFileUsed: "session.jsonl",
+      });
+      expect(mocks.settleStream).toHaveBeenCalledOnce();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("skips the pending-events join once the run abort signal fires", async () => {
+    const abortController = new AbortController();
+    abortController.abort(new Error("operator cancel"));
+    const fixture = createFixture();
+    fixture.input.settle.runAbortSignal = abortController.signal;
+    fixture.input.waitForPendingEvents = vi.fn(() => new Promise<never>(() => {}));
+    fixture.input.settle.readLifecycleState = () => ({
+      aborted: true,
+      timedOut: false,
+      timedOutDuringCompaction: false,
+    });
+    const settledStream = {
+      promptError: null,
+      promptErrorSource: null,
+      timedOutDuringCompaction: false,
+      compactionOccurredThisAttempt: false,
+      messagesSnapshot: [],
+      sessionIdUsed: "session-1",
+      lastAssistant: undefined,
+      currentAttemptAssistant: undefined,
+      currentAttemptCompletedAssistant: undefined,
+      attemptUsage: undefined,
+      cacheBreak: null,
+      lastCallUsage: undefined,
+      promptCache: undefined,
+    };
+    mocks.settleStream.mockResolvedValue(settledStream);
+    mocks.completeAfterTurn.mockResolvedValue({
+      sessionIdUsed: "session-1",
+      sessionFileUsed: "session.jsonl",
+    });
+
+    await expect(finalizeEmbeddedAttemptStreamPhase(fixture.input)).resolves.toEqual({
+      sessionIdUsed: "session-1",
+      sessionFileUsed: "session.jsonl",
+    });
+    expect(mocks.settleStream).toHaveBeenCalledOnce();
+  });
+
   it("settles an aborted run when prompt release returns its recorded cancellation reason", async () => {
     const cancellationReason = new Error("cancelled by operator");
     const fixture = createFixture({

src/agents/embedded-agent-runner/run/attempt-stream-finalize.ts

@@ -1,5 +1,7 @@
 /** Settles the provider stream and completes the post-turn lifecycle phase. */
 import { isRunnerAbortError } from "../abort.js";
+import { log } from "../logger.js";
+import { joinWithRunLivenessDeadline, RUN_LIVENESS_JOIN_TIMEOUT_MS } from "./abortable.js";
 import { completeEmbeddedAttemptAfterTurn } from "./attempt-after-turn.js";
 import { settleEmbeddedAttemptStream } from "./attempt-stream-settle.js";
 
@@ -17,6 +19,13 @@ type SharedPhaseInputKeys =
   | "sessionLockController"
   | "withOwnedSessionWriteLock";
 
+// Queued subscription handlers (block-reply delivery, tool events) are
+// fire-and-forget during the turn; the pending-events join below is the only
+// place the run waits for them. One hung handler (e.g. a stuck delivery
+// dispatch lane) must not dead-end the turn until the run budget — 48h by
+// default — so the join is bounded and settlement proceeds with a recorded
+// warning instead of producing no visible outcome at all.
+
 export async function finalizeEmbeddedAttemptStreamPhase(input: {
   attempt: StreamSettleInput["attempt"];
   activeSession: StreamSettleInput["activeSession"];
@@ -44,7 +53,16 @@ export async function finalizeEmbeddedAttemptStreamPhase(input: {
 }): Promise<{ sessionIdUsed: string; sessionFileUsed?: string }> {
   const { activeSession, sessionManager, sessionLockController, withOwnedSessionWriteLock } = input;
 
-  await input.waitForPendingEvents();
+  await joinWithRunLivenessDeadline({
+    joinWork: input.waitForPendingEvents,
+    runAbortSignal: input.settle.runAbortSignal,
+    onTimeout: () => {
+      log.warn(
+        `pending subscription events did not settle within ${RUN_LIVENESS_JOIN_TIMEOUT_MS}ms; ` +
+          `proceeding to stream settlement: runId=${input.attempt.runId}`,
+      );
+    },
+  });
   const beforeAgentFinalizeRevisionReason = input.getBeforeAgentFinalizeRevisionReason();
   const beforeAgentFinalizeRevisionEntryId = input.getBeforeAgentFinalizeRevisionEntryId();
   let rewoundBeforeAgentFinalizeRevision = false;

src/agents/embedded-agent-runner/run/attempt-stream-settle.test.ts

@@ -0,0 +1,102 @@
+// Settlement liveness: a wedged block-reply flush must not park the turn.
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { SessionManager } from "../../sessions/index.js";
+import { RUN_LIVENESS_JOIN_TIMEOUT_MS } from "./abortable.js";
+import { settleEmbeddedAttemptStream } from "./attempt-stream-settle.js";
+
+type SettleInput = Parameters<typeof settleEmbeddedAttemptStream>[0];
+
+function createSettleFixture(overrides?: Partial<SettleInput>): SettleInput {
+  const sessionManager = SessionManager.inMemory();
+  return {
+    attempt: {
+      runId: "run-settle-1",
+      sessionId: "sess-settle-1",
+      sessionKey: "agent:main:test",
+      provider: "openai",
+      modelId: "gpt-5.6-luna",
+      model: { api: "openai-responses" },
+      config: {},
+      promptCacheKey: undefined,
+    },
+    activeSession: {
+      sessionId: "sess-settle-1",
+      isCompacting: false,
+      isStreaming: false,
+      messages: [],
+    },
+    sessionManager,
+    sessionLockController: {},
+    withOwnedSessionWriteLock: async (operation: () => unknown) => await operation(),
+    subscription: {
+      toolMetas: [],
+      waitForCompactionRetry: async () => {},
+      isCompactionInFlight: () => false,
+      getCompactionCount: () => 0,
+      getCurrentAttemptAssistant: () => undefined,
+      getUsageTotals: () => undefined,
+      getLastAssistantUsage: () => undefined,
+    },
+    state: {
+      promptError: null,
+      promptErrorSource: null,
+      yieldAborted: false,
+      sessionIdUsed: "sess-settle-1",
+    },
+    readLifecycleState: () => ({
+      aborted: false,
+      timedOut: false,
+      timedOutDuringCompaction: false,
+    }),
+    markTimedOutDuringCompaction: vi.fn(),
+    runAbortDeadlineAtMs: Date.now() + 600_000,
+    runAbortSignal: new AbortController().signal,
+    isProbeSession: true,
+    abortable: async <T>(promise: Promise<T>) => await promise,
+    prePromptMessageCount: 0,
+    toolSearchTargetTranscriptProjections: [],
+    cache: {
+      observabilityEnabled: false,
+      changesForTurn: null,
+      retention: undefined,
+    },
+    shouldFlushForContextEngine: false,
+    ...overrides,
+  } as unknown as SettleInput;
+}
+
+describe("settleEmbeddedAttemptStream liveness", () => {
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("settles past a block-reply flush that never resolves", async () => {
+    vi.useFakeTimers();
+    // A wedged delivery lane (including the supported blockReplyTimeoutMs: 0
+    // path) previously parked settlement until the 48h run budget.
+    const input = createSettleFixture({
+      onBlockReplyFlush: () => new Promise<never>(() => {}),
+    } as Partial<SettleInput>);
+
+    const settle = settleEmbeddedAttemptStream(input);
+    let settled = false;
+    void settle.then(() => {
+      settled = true;
+    });
+    await vi.advanceTimersByTimeAsync(RUN_LIVENESS_JOIN_TIMEOUT_MS - 1);
+    expect(settled).toBe(false);
+    await vi.advanceTimersByTimeAsync(1);
+    const result = await settle;
+    expect(result.sessionIdUsed).toBe("sess-settle-1");
+  });
+
+  it("settles normally when the flush resolves", async () => {
+    const flushed = vi.fn(async () => {});
+    const input = createSettleFixture({
+      onBlockReplyFlush: flushed,
+    } as Partial<SettleInput>);
+    const result = await settleEmbeddedAttemptStream(input);
+    expect(flushed).toHaveBeenCalledWith({ reason: "pre_compaction", attemptAccepted: false });
+    expect(result.sessionIdUsed).toBe("sess-settle-1");
+  });
+});

src/agents/embedded-agent-runner/run/attempt-stream-settle.ts

@@ -17,6 +17,7 @@ import {
   type PromptCacheBreak,
   type PromptCacheChange,
 } from "../prompt-cache-observability.js";
+import { joinWithRunLivenessDeadline, RUN_LIVENESS_JOIN_TIMEOUT_MS } from "./abortable.js";
 import {
   flushSessionManagerTranscript,
   normalizeCompactionRecoveryTranscriptTail,
@@ -191,7 +192,19 @@ export async function settleEmbeddedAttemptStream(input: {
         !input.readLifecycleState().timedOut &&
         !state.yieldAborted &&
         currentAssistant?.stopReason === "stop";
-      await input.onBlockReplyFlush({ reason: "pre_compaction", attemptAccepted });
+      // The flush rides the same delivery chain the finalize-phase join just
+      // bounded; a wedged lane (including the supported blockReplyTimeoutMs: 0
+      // path) must not park settlement until the 48h run budget either.
+      await joinWithRunLivenessDeadline({
+        joinWork: () => input.onBlockReplyFlush?.({ reason: "pre_compaction", attemptAccepted }),
+        runAbortSignal: input.runAbortSignal,
+        onTimeout: () => {
+          log.warn(
+            `block-reply flush did not settle within ${RUN_LIVENESS_JOIN_TIMEOUT_MS}ms; ` +
+              `proceeding with settlement: runId=${attempt.runId}`,
+          );
+        },
+      });
     }
 
     const compactionRetryWait = state.yieldAborted