diff --git a/packages/ai/src/transports/openai-responses-stream-activity.test.ts b/packages/ai/src/transports/openai-responses-stream-activity.test.ts new file mode 100644 index 000000000000..55a2336df6a1 --- /dev/null +++ b/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[], +): AsyncGenerator { + 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[] = [ + { 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(); + } + }); +}); diff --git a/packages/ai/src/transports/openai-responses-stream-internal.ts b/packages/ai/src/transports/openai-responses-stream-internal.ts index 8694586e2654..6a0494a78f61 100644 --- a/packages/ai/src/transports/openai-responses-stream-internal.ts +++ b/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( ); 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") { diff --git a/src/agents/embedded-agent-runner/run/abortable.test.ts b/src/agents/embedded-agent-runner/run/abortable.test.ts index cba9cfc9f3df..2c6e78bb2495 100644 --- a/src/agents/embedded-agent-runner/run/abortable.test.ts +++ b/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(() => {}), + 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(() => {}), + 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(); + }); +}); diff --git a/src/agents/embedded-agent-runner/run/abortable.ts b/src/agents/embedded-agent-runner/run/abortable.ts index 5088811d9596..c48b07429817 100644 --- a/src/agents/embedded-agent-runner/run/abortable.ts +++ b/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; + runAbortSignal: AbortSignal; + timeoutMs?: number; + onTimeout: () => void; +}): Promise { + return new Promise((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 diff --git a/src/agents/embedded-agent-runner/run/attempt-stream-finalize.test.ts b/src/agents/embedded-agent-runner/run/attempt-stream-finalize.test.ts index 6704f48cfdbd..acee5e676f15 100644 --- a/src/agents/embedded-agent-runner/run/attempt-stream-finalize.test.ts +++ b/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(() => {})); + 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(() => {})); + 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({ diff --git a/src/agents/embedded-agent-runner/run/attempt-stream-finalize.ts b/src/agents/embedded-agent-runner/run/attempt-stream-finalize.ts index 9834229de69b..e4aa58da67f6 100644 --- a/src/agents/embedded-agent-runner/run/attempt-stream-finalize.ts +++ b/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; diff --git a/src/agents/embedded-agent-runner/run/attempt-stream-settle.test.ts b/src/agents/embedded-agent-runner/run/attempt-stream-settle.test.ts new file mode 100644 index 000000000000..0aaacb5f8203 --- /dev/null +++ b/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[0]; + +function createSettleFixture(overrides?: Partial): 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 (promise: Promise) => 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(() => {}), + } as Partial); + + 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); + const result = await settleEmbeddedAttemptStream(input); + expect(flushed).toHaveBeenCalledWith({ reason: "pre_compaction", attemptAccepted: false }); + expect(result.sessionIdUsed).toBe("sess-settle-1"); + }); +}); diff --git a/src/agents/embedded-agent-runner/run/attempt-stream-settle.ts b/src/agents/embedded-agent-runner/run/attempt-stream-settle.ts index 2541fddd5089..0fa66b06c2a4 100644 --- a/src/agents/embedded-agent-runner/run/attempt-stream-settle.ts +++ b/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