diff --git a/packages/coding-agent/changelog.d/sdk-unowned-lifecycle-receipts.md b/packages/coding-agent/changelog.d/sdk-unowned-lifecycle-receipts.md new file mode 100644 index 00000000000..baaf51dc861 --- /dev/null +++ b/packages/coding-agent/changelog.d/sdk-unowned-lifecycle-receipts.md @@ -0,0 +1,3 @@ +### Fixed +- Keep session-initiated lifecycle boundaries separate from queued SDK turns so a delayed predecessor completion cannot publish the successor's receipt with the wrong final answer. Preserve attached invocations and existing deadline, abort, and retired-owner isolation. +- Preserve a single public lifecycle boundary across accepted retries that suppress their predecessor's end, including tokenless interactive retries, so reopening the session retains terminal delivery and idle checkpoints. diff --git a/packages/coding-agent/src/extensibility/extensions/types.ts b/packages/coding-agent/src/extensibility/extensions/types.ts index e8bbd42b21f..e9a0757474f 100644 --- a/packages/coding-agent/src/extensibility/extensions/types.ts +++ b/packages/coding-agent/src/extensibility/extensions/types.ts @@ -764,6 +764,8 @@ export interface AgentStartEvent extends SharedAgentStartEvent { sdkRunToken?: string; /** Complete set of SDK queue owners consumed together by this run. */ sdkRunTokens?: string[]; + /** Internal public lifecycle boundary; accepted retries with a suppressed predecessor end retain it. */ + lifecycleScope?: AttemptScope; } /** Fired when an agent run fails before emitting agent_end. The error is the diff --git a/packages/coding-agent/src/sdk/host/session-runtime.test.ts b/packages/coding-agent/src/sdk/host/session-runtime.test.ts index 54483df144e..62181e6d394 100644 --- a/packages/coding-agent/src/sdk/host/session-runtime.test.ts +++ b/packages/coding-agent/src/sdk/host/session-runtime.test.ts @@ -5,6 +5,7 @@ import { chmod, mkdir, mkdtemp, readdir, readFile, rm, writeFile } from "node:fs import * as os from "node:os"; import * as path from "node:path"; import { Agent, markNonDispatchedToolEvent, ThinkingLevel } from "@gajae-code/agent-core"; +import { createAttemptMinter } from "@gajae-code/agent-core/attempt-scope"; import type { UserMessage } from "@gajae-code/ai/core"; import { createMockModel } from "@gajae-code/ai/providers/mock"; import { logger } from "@gajae-code/utils"; @@ -5139,6 +5140,54 @@ test("SDK turn.steer preserves its expected run token and propagates a stale-run }); describe("post-acceptance invocation terminalization", () => { + test("preserves external prompt correlation across a retryable provider error", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "gjc-retryable-provider-correlation-")); + let harness: InvocationHarness | undefined; + try { + harness = await invocationHarness("retryable-provider-correlation", cwd, { + sendUserMessage: async (_content, options) => { + await options?.onPreflightAcceptCommit?.(); + await neverSettlingPromise(); + }, + }); + const accepted = await harness.control("turn.prompt", { + text: "recover after a provider overload", + clientRef: "retryable-provider-correlation", + }); + expect(accepted.ok).toBe(true); + const correlation = { + commandId: accepted.result?.commandId, + turnId: accepted.result?.turnId, + }; + const retryScope = {}; + + for (let attempt = 0; attempt < 3; attempt++) { + await harness.emit("agent_start", { lifecycleScope: retryScope }); + } + await harness.emit("agent_end", { + messages: [{ role: "assistant", stopReason: "stop", content: [{ type: "text", text: "recovered" }] }], + }); + + const terminal = await settledStatus(harness, "turn.result", { kind: "prompt", ...correlation }); + expect(terminal).toMatchObject({ + status: "terminal_ok", + commandId: correlation.commandId, + turnId: correlation.turnId, + outcome: { kind: "stopped", reason: "end_turn" }, + }); + expect(harness.broadcasts.filter(frame => frame.kind === "agent_end").at(-1)).toMatchObject({ + payload: { + commandId: correlation.commandId, + turnId: correlation.turnId, + outcome: { kind: "stopped", reason: "end_turn" }, + }, + }); + } finally { + await harness?.stop(); + await rm(cwd, { recursive: true, force: true }); + } + }); + test.each([ { status: 402, code: "provider_http_402" }, { status: 429, code: "provider_http_429" }, @@ -5786,6 +5835,85 @@ describe("post-acceptance invocation terminalization", () => { } }); + test("tokenless retry boundaries release replacement terminal publication and replay", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "sdk-tokenless-retry-reopen-")); + const sessionId = "tokenless-retry-reopen"; + const harness = await invocationHarness(sessionId, cwd, {}); + const scopes = createAttemptMinter(); + const activity = spyOn(SessionSdkSessionRuntime.prototype, "reportActivity"); + try { + // Accepted ordinary retries suppress predecessor ends. Three attempts + // therefore share one public lifecycle boundary, not three unmatched ends. + const lifecycleScope = scopes.mint("main"); + for (let attempt = 0; attempt < 3; attempt++) { + await harness.emit("agent_start", { lifecycleScope }); + } + await harness.emit("agent_end"); + for (let turn = 0; turn < 2; turn++) { + await harness.emit("agent_start", { lifecycleScope: scopes.mint("main") }); + await harness.emit("agent_end"); + } + await harness.switchSession("different-session"); + await harness.switchSession(sessionId); + const priorEnds = harness.broadcasts.filter(frame => frame.kind === "agent_end").length; + await harness.emit("agent_start", { lifecycleScope: scopes.mint("main") }); + await harness.emit("agent_end"); + expect(harness.broadcasts.filter(frame => frame.kind === "agent_end")).toHaveLength(priorEnds + 1); + expect(activity.mock.calls.at(-1)?.[0]).toBe("idle"); + expect(await harness.requestOnSession(sessionId, { type: "event_replay", sinceSeq: 0 })).toMatchObject({ + type: "event_replay_result", + events: expect.arrayContaining([expect.objectContaining({ kind: "agent_end" })]), + }); + } finally { + activity.mockRestore(); + await harness.stop(); + await rm(cwd, { recursive: true, force: true }); + } + }); + + test("tokenless retry preserves attached owners alongside newly promoted SDK work", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "sdk-tokenless-retry-owners-")); + const promotions: Array<(promotion: { startsOwnRun: boolean }) => void> = []; + const harness = await invocationHarness("tokenless-retry-owners", cwd, { + sendUserMessage: async (_content, options) => { + await options?.onPreflightAcceptCommit?.(); + if (options?.onQueuedPromoted) promotions.push(options.onQueuedPromoted); + options?.onPreflightAccepted?.(); + }, + }); + try { + const lifecycleScope = createAttemptMinter().mint("main"); + await harness.emit("agent_start", { lifecycleScope }); + const attached = await harness.control("turn.follow_up", { text: "attached work" }); + const queued = await harness.control("turn.follow_up", { text: "promoted work" }); + expect(attached.ok).toBe(true); + expect(queued.ok).toBe(true); + expect(promotions).toHaveLength(2); + promotions[0]!({ startsOwnRun: false }); + promotions[1]!({ startsOwnRun: true }); + await harness.emit("agent_start", { lifecycleScope }); + await harness.emit("agent_end", { messages: [{ role: "assistant", content: "shared retry final" }] }); + for (const accepted of [attached, queued]) { + expect( + await harness.query("turn.result", { + kind: "prompt", + commandId: accepted.result?.commandId, + turnId: accepted.result?.turnId, + }), + ).toMatchObject({ + result: { + status: "terminal_ok", + receiptState: "present", + content: { text: "shared retry final" }, + }, + }); + } + } finally { + await harness.stop(); + await rm(cwd, { recursive: true, force: true }); + } + }); + test("a prompt killed by a provider stream interrupt reports a terminal failed status", async () => { const cwd = await mkdtemp(path.join(os.tmpdir(), "gjc-terminalize-prompt-")); try { @@ -5851,6 +5979,129 @@ describe("post-acceptance invocation terminalization", () => { await rm(cwd, { recursive: true, force: true }); } }); + test("unowned predecessor end cannot publish a queued successor receipt", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "sdk-unowned-predecessor-")); + const promotions: Array<(promotion: { startsOwnRun: boolean }) => void> = []; + const harness = await invocationHarness("unowned-predecessor", cwd, { + sendUserMessage: async (_content, options) => { + await options?.onPreflightAcceptCommit?.(); + if (options?.onQueuedPromoted) promotions.push(options.onQueuedPromoted); + options?.onPreflightAccepted?.(); + }, + }); + try { + await harness.emit("agent_start"); + const accepted = await harness.control("turn.follow_up", { text: "queued verification" }); + expect(accepted.ok).toBe(true); + expect(promotions).toHaveLength(1); + promotions[0]!({ startsOwnRun: true }); + await harness.emit("agent_start"); + const selector = { + kind: "prompt", + commandId: accepted.result?.commandId, + turnId: accepted.result?.turnId, + }; + await harness.emit("agent_end", { messages: [{ role: "assistant", content: "predecessor final" }] }); + expect(await harness.query("turn.result", selector)).toMatchObject({ + result: { status: "in_flight", receiptState: "absent" }, + }); + await harness.emit("agent_end", { messages: [{ role: "assistant", content: "verification final" }] }); + expect(await harness.query("turn.result", selector)).toMatchObject({ + result: { + status: "terminal_ok", + turnId: accepted.result?.turnId, + receiptState: "present", + content: { text: "verification final" }, + }, + }); + } finally { + await harness.stop(); + await rm(cwd, { recursive: true, force: true }); + } + }); + + test("unowned run stays paired after predecessor SDK references retire", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "sdk-unowned-retirement-")); + const promotions: Array<(promotion: { startsOwnRun: boolean }) => void> = []; + const harness = await invocationHarness("unowned-retirement", cwd, { + sendUserMessage: async (_content, options) => { + await options?.onPreflightAcceptCommit?.(); + if (options?.onQueuedPromoted) promotions.push(options.onQueuedPromoted); + options?.onPreflightAccepted?.(); + }, + }); + try { + const first = await harness.control("turn.follow_up", { text: "first" }); + expect(promotions).toHaveLength(1); + promotions[0]!({ startsOwnRun: true }); + await harness.emit("agent_start"); + await harness.emit("agent_start"); + await harness.emit("agent_end", { messages: [{ role: "assistant", content: "first final" }] }); + expect( + await harness.query("turn.result", { + kind: "prompt", + commandId: first.result?.commandId, + turnId: first.result?.turnId, + }), + ).toMatchObject({ result: { status: "terminal_ok", content: { text: "first final" } } }); + const next = await harness.control("turn.follow_up", { text: "next" }); + expect(promotions).toHaveLength(2); + promotions[1]!({ startsOwnRun: true }); + await harness.emit("agent_start"); + const selector = { kind: "prompt", commandId: next.result?.commandId, turnId: next.result?.turnId }; + await harness.emit("agent_end", { messages: [{ role: "assistant", content: "unowned final" }] }); + expect(await harness.query("turn.result", selector)).toMatchObject({ + result: { status: "in_flight", receiptState: "absent" }, + }); + await harness.emit("agent_end", { messages: [{ role: "assistant", content: "next final" }] }); + expect(await harness.query("turn.result", selector)).toMatchObject({ + result: { status: "terminal_ok", content: { text: "next final" } }, + }); + } finally { + await harness.stop(); + await rm(cwd, { recursive: true, force: true }); + } + }); + + test("unowned run retains attached invocation across predecessor completion", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "sdk-unowned-attached-")); + const promotions: Array<(promotion: { startsOwnRun: boolean }) => void> = []; + const harness = await invocationHarness("unowned-attached", cwd, { + sendUserMessage: async (_content, options) => { + await options?.onPreflightAcceptCommit?.(); + if (options?.onQueuedPromoted) promotions.push(options.onQueuedPromoted); + options?.onPreflightAccepted?.(); + }, + }); + try { + await harness.control("turn.follow_up", { text: "predecessor" }); + expect(promotions).toHaveLength(1); + promotions[0]!({ startsOwnRun: true }); + await harness.emit("agent_start"); + await harness.emit("agent_start"); + const attached = await harness.control("turn.follow_up", { text: "attach to current unowned run" }); + expect(attached.ok).toBe(true); + expect(promotions).toHaveLength(2); + promotions[1]!({ startsOwnRun: false }); + const selector = { + kind: "prompt", + commandId: attached.result?.commandId, + turnId: attached.result?.turnId, + }; + await harness.emit("agent_end", { messages: [{ role: "assistant", content: "predecessor final" }] }); + expect(await harness.query("turn.result", selector)).toMatchObject({ + result: { status: expect.stringMatching(/accepted|in_flight/), receiptState: "absent" }, + }); + await harness.emit("agent_end", { messages: [{ role: "assistant", content: "attached final" }] }); + expect(await harness.query("turn.result", selector)).toMatchObject({ + result: { status: "terminal_ok", content: { text: "attached final" } }, + }); + } finally { + await harness.stop(); + await rm(cwd, { recursive: true, force: true }); + } + }); + test("immediate prompt after abort ack is not terminalized by the aborted turn's delayed agent_end", async () => { const cwd = await mkdtemp(path.join(os.tmpdir(), "gjc-abort-immediate-prompt-")); try { diff --git a/packages/coding-agent/src/sdk/host/session-runtime.ts b/packages/coding-agent/src/sdk/host/session-runtime.ts index f4465c3a626..e314d1161ae 100644 --- a/packages/coding-agent/src/sdk/host/session-runtime.ts +++ b/packages/coding-agent/src/sdk/host/session-runtime.ts @@ -5,6 +5,7 @@ import * as fs from "node:fs/promises"; import * as path from "node:path"; import { promisify } from "node:util"; import { isContinuingMidRunMaintenanceOutcome, isNonDispatchedToolEvent, ThinkingLevel } from "@gajae-code/agent-core"; +import type { AttemptScope } from "@gajae-code/agent-core/attempt-scope"; import type { Api, ImageContent, Model, ProviderDiagnostic } from "@gajae-code/ai/core"; import { logger } from "@gajae-code/utils"; @@ -4842,6 +4843,8 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre }>; openLifecycleBatches: Array<{ epoch: number; + /** Session-initiated starts still reserve their own unmatched end. */ + unownedStart: boolean; invocations: Array<{ kind: InvocationKind; correlation: InvocationCorrelation; @@ -4897,6 +4900,7 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre string, { state: RuntimeState; batch?: LifecycleBatch; correlationKey?: string } >(); + const lifecycleScopeOwners = new WeakMap(); const lifecycleCorrelationKey = (correlation: InvocationCorrelation): string => `${correlation.commandId}:${correlation.turnId}`; const sessionIdentityForContext = (ctx: ExtensionContext): string | undefined => { @@ -4993,8 +4997,10 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre entry => lifecycleCorrelationKey(entry.correlation) !== target, ); } + // Retiring an SDK correlation must not discard an unrelated session- + // initiated start. Its delayed end still belongs to that empty batch. owner.openLifecycleBatches = owner.openLifecycleBatches.filter( - batch => batch.invocations.length > 0 || batch.attachedInvocations.length > 0, + batch => batch.unownedStart || batch.invocations.length > 0 || batch.attachedInvocations.length > 0, ); owner.drainedInvocations = owner.drainedInvocations?.filter( entry => lifecycleCorrelationKey(entry.correlation) !== target, @@ -5273,32 +5279,24 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre stopReason?: AgentEndEvent["stopReason"], startToken?: string, startTokens?: string[], + startLifecycleScope?: AttemptScope, ): Promise => { const current = lifecycleOwner?.state ?? active; if (!current) return; const failureBatch = lifecycleOwner?.batch; - const adoptLifecycleBatch = ( - batch: - | Array<{ - kind: InvocationKind; - correlation: InvocationCorrelation; - connectionId: string | undefined; - }> - | undefined, - ): void => { - if (!batch || batch.length === 0) { - current.activeInvocation = undefined; - current.drainedInvocations = undefined; - current.attachedInvocations = undefined; - if (current === active) activePromptOwnerHolder.connectionIds = undefined; - return; - } - current.activeInvocation = batch[0]; - current.drainedInvocations = batch.map(({ kind, correlation }) => ({ kind, correlation })); - current.attachedInvocations = undefined; + const adoptLifecycleBatch = (batch: LifecycleBatch | undefined): void => { + current.activeInvocation = batch?.invocations[0]; + current.drainedInvocations = batch + ? [...batch.invocations, ...batch.attachedInvocations].map(({ kind, correlation }) => ({ + kind, + correlation, + })) + : undefined; + current.attachedInvocations = batch?.attachedInvocations.slice(); const owners = new Set(); - for (const entry of batch) if (entry.connectionId !== undefined) owners.add(entry.connectionId); - if (current === active) activePromptOwnerHolder.connectionIds = owners; + for (const entry of batch?.invocations ?? []) + if (entry.connectionId !== undefined) owners.add(entry.connectionId); + if (current === active) activePromptOwnerHolder.connectionIds = owners.size > 0 ? owners : undefined; }; let transitions: Array<{ kind: InvocationKind; correlation: InvocationCorrelation }> = []; if (type === "agent_failed") { @@ -5332,13 +5330,18 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre ); transitions = [...fallback, ...attached].map(({ kind, correlation }) => ({ kind, correlation })); } else if (type === "agent_start") { - current.lifecycleEpoch = ++nextLifecycleEpoch; + const scopeOwner = startLifecycleScope ? lifecycleScopeOwners.get(startLifecycleScope) : undefined; + const scopeBatch = + scopeOwner?.state === current && current.openLifecycleBatches.includes(scopeOwner.batch) + ? scopeOwner.batch + : undefined; + current.lifecycleEpoch = scopeBatch?.epoch ?? ++nextLifecycleEpoch; if (current === active) activePromptOwnerHolder.lifecycleEpoch = current.lifecycleEpoch; // Mark lifecycle active even when the drain is empty: a monitor/cron // run started by the session has no SDK pending entry but is still a // real active run that later in-run promotions must attach to instead - // of falling back to pending (review P1). Empty drains leave the - // previous SDK owner untouched. + // of falling back to pending (review P1). Only an explicitly bound + // continuation retains a previous public lifecycle boundary. current.lifecycleActive = true; // Drain EVERY entry admitted for this run: a continuation may promote // several follow-ups (each with its own requester correlation) into one @@ -5359,40 +5362,47 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre typeof startToken === "string" && startToken.length > 0 ? lifecycleRunOwners.get(startToken)?.batch : undefined; - if (drained.length > 0) { - const batch: LifecycleBatch = { + const continuationBatch = + scopeBatch ?? + (drained.length === 0 && existingTokenBatch && current.openLifecycleBatches.includes(existingTokenBatch) + ? existingTokenBatch + : undefined); + let batch: LifecycleBatch; + if (!continuationBatch) { + // Every new run owns an end, including interactive/monitor runs + // without SDK invocations. Their delayed ends must never consume + // the next SDK run's terminal receipt. + batch = { epoch: current.lifecycleEpoch, + unownedStart: drained.length === 0, invocations: drained, attachedInvocations: [], - startPublished: false, + startPublished: drained.length === 0, heldContentDropped: false, heldContent: [], }; current.openLifecycleBatches.push(batch); - for (const entry of drained) { - lifecycleRunOwners.set(entry.sdkRunToken, { - state: current, - batch, - correlationKey: lifecycleCorrelationKey(entry.correlation), - }); - } - adoptLifecycleBatch(drained); - for (const entry of drained) { - if (entry.kind !== "prompt") continue; - if (current.deadlineManager.isExpiring(entry.correlation)) - current.deadlineManager.captureExpiringRun(entry.correlation); - else current.deadlineManager.onRunStarted(entry.correlation); - } - } else if (existingTokenBatch && current.openLifecycleBatches.includes(existingTokenBatch)) { - // A continuation within the same SDK-owned run carries the same token - // but has no newly pending invocation. Preserve the established owner so - // later chunks and the final terminal remain directed to its submitters. - adoptLifecycleBatch(existingTokenBatch.invocations); } else { - // An empty drain is an agent-initiated successor run. Clear the - // predecessor's SDK owner and active invocation so abort ownership and - // tool-progress renewal cannot leak into the successor. - adoptLifecycleBatch(undefined); + // The producer consumed the predecessor end, or this is the same + // token-owned run. Preserve its attached owners and single final end. + batch = continuationBatch; + batch.invocations.push(...drained); + if (drained.length > 0) batch.startPublished = false; + } + if (startLifecycleScope) lifecycleScopeOwners.set(startLifecycleScope, { state: current, batch }); + for (const entry of drained) { + lifecycleRunOwners.set(entry.sdkRunToken, { + state: current, + batch, + correlationKey: lifecycleCorrelationKey(entry.correlation), + }); + } + adoptLifecycleBatch(batch); + for (const entry of drained) { + if (entry.kind !== "prompt") continue; + if (current.deadlineManager.isExpiring(entry.correlation)) + current.deadlineManager.captureExpiringRun(entry.correlation); + else current.deadlineManager.onRunStarted(entry.correlation); } transitions = drained.map(({ kind, correlation }) => ({ kind, correlation })); } else { @@ -5468,7 +5478,8 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre transitions.map(({ correlation }) => `${correlation.commandId}:${correlation.turnId}`), ); if ( - !ended.invocations.some(({ correlation }) => + (ended.invocations.length > 0 || ended.attachedInvocations.length > 0) && + ![...ended.invocations, ...ended.attachedInvocations].some(({ correlation }) => transitionKeys.has(`${correlation.commandId}:${correlation.turnId}`), ) ) @@ -5515,7 +5526,7 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre ); } current.openLifecycleBatches = current.openLifecycleBatches.filter( - batch => batch.invocations.length > 0, + batch => batch.unownedStart || batch.invocations.length > 0 || batch.attachedInvocations.length > 0, ); current.drainedInvocations = current.drainedInvocations?.filter( entry => @@ -5526,7 +5537,7 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre current.activeInvocation?.correlation.commandId === invocation.correlation.commandId && current.activeInvocation.correlation.turnId === invocation.correlation.turnId ) - adoptLifecycleBatch(current.openLifecycleBatches[0]?.invocations); + adoptLifecycleBatch(current.openLifecycleBatches[0]); if (current.openLifecycleBatches.length === 0) current.lifecycleActive = false; } catch (error) { if (controller.signal.aborted) { @@ -5949,7 +5960,7 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre !current.deadlineManager.shouldDeferTerminalTransition(invocation.correlation) ) current.deadlineManager.clear(invocation.correlation); - adoptLifecycleBatch(current.openLifecycleBatches[0]?.invocations); + adoptLifecycleBatch(current.openLifecycleBatches[0]); if (current.openLifecycleBatches.length === 0) current.lifecycleActive = false; // Resolve EVERY concurrent waiter for the aborted turn: the turn emits // exactly one agent_end, and each admitted abort of it must observe the @@ -5983,6 +5994,7 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre undefined, typeof event.sdkRunToken === "string" ? event.sdkRunToken : undefined, event.sdkRunTokens, + event.lifecycleScope, ), owner, ).catch(error => { @@ -6586,26 +6598,7 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre true, ); } - const openLifecycleBatches: Array<{ - epoch: number; - invocations: Array<{ - kind: InvocationKind; - correlation: InvocationCorrelation; - connectionId: string | undefined; - sdkRunToken: string; - }>; - attachedInvocations: Array<{ - kind: InvocationKind; - correlation: InvocationCorrelation; - connectionId: string | undefined; - }>; - startPublished: boolean; - heldContentDropped: boolean; - heldContent: Array<{ - event: AgentSessionEvent; - recipients: Array<{ correlation: InvocationCorrelation; connectionId: string | undefined }>; - }>; - }> = []; + const openLifecycleBatches: LifecycleBatch[] = []; const configRevision = { current: 0 }; let acceptingGateResolutions = true; const inFlightGateResolutions = new Set>(); @@ -6893,7 +6886,7 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre entry.correlation.turnId === current.activeInvocation?.correlation.turnId, ), ) - : undefined; + : current.openLifecycleBatches.at(-1); const alreadyAttached = current.drainedInvocations.some( entry => entry.correlation.commandId === correlation.commandId && diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index 0741c5cacd1..75fb0717a02 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -135,8 +135,8 @@ import { classifyFallbackTrigger, EMPTY_RESPONSE_PROVIDER_CODE, type FallbackAttemptToken, - PROVIDER_PROTOCOL_MISMATCH_ERROR_CODE, type FallbackTriggerClass, + PROVIDER_PROTOCOL_MISMATCH_ERROR_CODE, SERVER_OVERLOADED_PROVIDER_CODE, STREAM_FIRST_EVENT_TIMEOUT_PROVIDER_CODE, } from "@gajae-code/ai/utils/fallback-transport"; @@ -409,11 +409,7 @@ import type { NotificationSessionController } from "../sdk/bus/session-control"; import { buildSyntheticModelId, syntheticNamespaceCollision } from "../sdk/model-profile-model"; import { sanitizePromptFailure } from "../sdk/prompt-failure"; import type { SecretObfuscator } from "../secrets/obfuscator"; -import { - formatNoCredentialOnboardingError, - formatNoModelOnboardingError, - NoModelSelectedError, -} from "../setup/model-onboarding-guidance"; +import { formatNoCredentialOnboardingError, NoModelSelectedError } from "../setup/model-onboarding-guidance"; import { isCanonicalGjcWorkflowSkill, isWorkflowContinuationInert, @@ -3801,6 +3797,7 @@ export class AgentSession { #skipPostPromptRecoveryWaitByAttemptScope = new WeakSet(); #sdkRunTokensByAttemptScope = new WeakMap(); #sdkRunCohortsByAttemptScope = new WeakMap(); + #lifecycleScopesByAttemptScope = new WeakMap(); #activeSdkRunToken: string | undefined; #activeAttemptScope: AttemptScope | undefined; #attemptAuthority!: AttemptScopeAuthority; @@ -4074,11 +4071,7 @@ export class AgentSession { readonly rawSseDebugBuffer: RawSseDebugBuffer; #acquirePowerAssertion(): void { - if ( - process.platform !== "darwin" && - process.platform !== "linux" && - process.platform !== "win32" - ) { + if (process.platform !== "darwin" && process.platform !== "linux" && process.platform !== "win32") { return; } if (this.#powerAssertion || this.#powerAssertionLoad) return; @@ -5499,12 +5492,7 @@ export class AgentSession { controller !== undefined && controller.chain.entries.length > 1; if (shouldCapture && modelKey) this.#managedFallbackActiveCredentialRows.delete(modelKey); - if ( - override && - model && - storageProvider === override.storageProvider && - modelKey === override.modelKey - ) { + if (override && model && storageProvider === override.storageProvider && modelKey === override.modelKey) { this.#managedFallbackNextCredentialOverride = undefined; const apiKey = await this.#modelRegistry.getApiKey(model, this.credentialSessionId, { credentialSelector: { kind: "id", value: String(override.rowId) }, @@ -5514,10 +5502,7 @@ export class AgentSession { return apiKey; } // The preselected credential is no longer available; try to resolve another one of the same kind. - const fallbackApiKey = await this.#resolveManagedFallbackCredentialRow( - model, - override.credentialKind, - ); + const fallbackApiKey = await this.#resolveManagedFallbackCredentialRow(model, override.credentialKind); if (fallbackApiKey !== undefined) { if (shouldCapture && model) this.#captureManagedFallbackActiveCredential(model); return fallbackApiKey; @@ -5525,8 +5510,7 @@ export class AgentSession { throw new Error("Selected managed fallback credential is no longer available"); } const apiKey = await invokeOriginalGetApiKey(provider); - if (shouldCapture && model && isAuthenticated(apiKey)) - this.#captureManagedFallbackActiveCredential(model); + if (shouldCapture && model && isAuthenticated(apiKey)) this.#captureManagedFallbackActiveCredential(model); return apiKey; }; this.agent.getApiKey = managedGetApiKey; @@ -5832,9 +5816,8 @@ export class AgentSession { (largestPhase, phase) => Math.max( largestPhase, - phase.tasks.filter( - task => task.status === "pending" || task.status === "in_progress", - ).length, + phase.tasks.filter(task => task.status === "pending" || task.status === "in_progress") + .length, ), 0, ); @@ -8691,6 +8674,14 @@ export class AgentSession { sdkRunToken, consumedSdkRunTokens.length > 0 ? consumedSdkRunTokens : undefined, ); + const predecessorScope = + predecessorAgentEnd && this.#agentEventAdmission.get(predecessorAgentEnd)?.scope; + if (predecessorScope) { + this.#lifecycleScopesByAttemptScope.set( + handle.scope, + this.#lifecycleScopesByAttemptScope.get(predecessorScope) ?? predecessorScope, + ); + } options?.onRunAccepted?.(handle); // Keep the queued token available through the acceptance callback; // SDK follow-up owners bind it to the new attempt scope there. @@ -9913,7 +9904,12 @@ export class AgentSession { type: "agent_start", ...(sdkRunToken ? { sdkRunToken } : {}), ...(deliveryScope - ? { sdkRunTokens: this.#sdkRunCohortsByAttemptScope.get(deliveryScope as AttemptScope) } + ? { + sdkRunTokens: this.#sdkRunCohortsByAttemptScope.get(deliveryScope as AttemptScope), + lifecycleScope: + this.#lifecycleScopesByAttemptScope.get(deliveryScope as AttemptScope) ?? + (deliveryScope as AttemptScope), + } : {}), }, undefined, @@ -12362,7 +12358,9 @@ export class AgentSession { */ get messages(): AgentMessage[] { const messages = this.agent.state.messages; - return messages.some(message => message.role === "custom" && this.#isEphemeralCustomMessageType(message.customType)) + return messages.some( + message => message.role === "custom" && this.#isEphemeralCustomMessageType(message.customType), + ) ? this.#withoutEphemeralCustomMessages(messages) : messages; } @@ -14716,10 +14714,7 @@ export class AgentSession { // prompt still in preflight, aborting) has no run to steer. Ordinary steers // become a sequential follow-up owned by the next turn; SDK turn-bound // steers fail closed rather than being delivered to a successor run. - if ( - options?.expectedSdkRunToken !== undefined && - this.#activeSdkRunToken !== options.expectedSdkRunToken - ) + if (options?.expectedSdkRunToken !== undefined && this.#activeSdkRunToken !== options.expectedSdkRunToken) throw Object.assign(new Error("The expected SDK run is not active."), { code: "turn_not_active" }); const admission = this.agent.steer(message, options?.forceOneAtATime ? { forceOneAtATime: true } : undefined); if (!admission.admitted) { @@ -21132,10 +21127,7 @@ export class AgentSession { const isContextPromoted = this.#temporaryProviderSessionScopes.some( scope => scope.token.reason === "context-promotion", ); - if ( - !isContextPromoted || - !isDefaultAutoThresholdCeilingApplied(contextWindow, settings) - ) { + if (!isContextPromoted || !isDefaultAutoThresholdCeilingApplied(contextWindow, settings)) { return settings; } @@ -23040,7 +23032,8 @@ export class AgentSession { isAuthenticated(apiKey) && authStorage.getSessionCredentialRowId(model.provider, this.credentialSessionId) === credential.id && authStorage.getSessionCredentialType(model.provider, this.credentialSessionId) === credentialKind && - currentCredential && !currentCredential.disabled + currentCredential && + !currentCredential.disabled ) return apiKey; } @@ -23957,9 +23950,7 @@ export class AgentSession { !authStorage.isCredentialAvailable(resolvedModel.provider, credential.id) ) continue; - let credentialKeyResult: - | { kind: "resolved"; value: string | undefined } - | { kind: "aborted" }; + let credentialKeyResult: { kind: "resolved"; value: string | undefined } | { kind: "aborted" }; try { credentialKeyResult = await awaitWithCancellation( this.#modelRegistry.getApiKey(resolvedModel, this.credentialSessionId, { @@ -24057,7 +24048,11 @@ export class AgentSession { !transitionStillOwned() ) return await rollbackCancelled(); - if (selectedCredentialRowId !== undefined && selectedCredentialKind !== undefined && this.#restoreManagedFallbackGetApiKey) { + if ( + selectedCredentialRowId !== undefined && + selectedCredentialKind !== undefined && + this.#restoreManagedFallbackGetApiKey + ) { this.#managedFallbackNextCredentialOverride = { modelKey: canonicalModelKey, storageProvider: resolveOAuthStorageProvider(resolvedModel.provider), @@ -24490,10 +24485,7 @@ export class AgentSession { // proper session termination handling. if (managedFallback && isStatuslessTypedOverloadFacts(transportFailure)) { return managedOutcome - ? this.#managedFallbackExhaustionDecision( - message, - message.errorMessage || "Model fallback attempt failed", - ) + ? this.#managedFallbackExhaustionDecision(message, message.errorMessage || "Model fallback attempt failed") : false; } if (!trigger) { @@ -24708,10 +24700,7 @@ export class AgentSession { if (outcome === "exhausted") { if (managedFallback) { let errorMessage = this.#fallbackExhaustionError(controller); - if ( - !providerRetryCeilingReached && - (trigger.class === "quota" || trigger.class === "rate_limit") - ) { + if (!providerRetryCeilingReached && (trigger.class === "quota" || trigger.class === "rate_limit")) { if (!assistantMessageHasVisibleOrToolContent(message)) { const mark = quotaCredentialMark ?? diff --git a/packages/coding-agent/test/agent-session-retry-busy-recovery.test.ts b/packages/coding-agent/test/agent-session-retry-busy-recovery.test.ts index f19c62a71c1..ebaf6358da5 100644 --- a/packages/coding-agent/test/agent-session-retry-busy-recovery.test.ts +++ b/packages/coding-agent/test/agent-session-retry-busy-recovery.test.ts @@ -5,6 +5,7 @@ import { type AssistantMessage, getBundledModel, type ToolCall } from "@gajae-co import { createMockModel } from "@gajae-code/ai/providers/mock"; import { ModelRegistry } from "@gajae-code/coding-agent/config/model-registry"; import { Settings } from "@gajae-code/coding-agent/config/settings"; +import type { AgentStartEvent, ExtensionEvent } from "@gajae-code/coding-agent/extensibility/extensions/types"; import { AgentSession, type AgentSessionEvent } from "@gajae-code/coding-agent/session/agent-session"; import { AuthStorage } from "@gajae-code/coding-agent/session/auth-storage"; import { SessionManager } from "@gajae-code/coding-agent/session/session-manager"; @@ -378,8 +379,10 @@ describe("AgentSession auto-retry busy recovery", () => { if (!model) throw new Error("Expected bundled Anthropic test model to exist"); const mock = createMockModel({ responses: [ + { throw: "503 service unavailable: overloaded_error retry-after-ms=5" }, { throw: "503 service unavailable: overloaded_error retry-after-ms=5" }, { content: ["retry successor succeeded"] }, + { content: ["independent prompt succeeded"] }, ], }); const agent = new Agent({ @@ -393,7 +396,21 @@ describe("AgentSession auto-retry busy recovery", () => { "retry.maxDelayMs": 5_000, }); settings.setModelRole("default", `${model.provider}/${model.id}`); - session = new AgentSession({ agent, sessionManager: SessionManager.inMemory(), settings, modelRegistry }); + const lifecycleStarts: AgentStartEvent[] = []; + const extensionRunner = { + emitBeforeAgentStart: async () => undefined, + hasHandlers: () => false, + emit: async (event: ExtensionEvent) => { + if (event.type === "agent_start") lifecycleStarts.push(event); + }, + } as never; + session = new AgentSession({ + agent, + sessionManager: SessionManager.inMemory(), + settings, + modelRegistry, + extensionRunner, + }); const handles: string[] = []; const terminalEvents: AgentSessionEvent[] = []; @@ -405,15 +422,31 @@ describe("AgentSession auto-retry busy recovery", () => { await session.prompt("retry with an owned successor"); await session.waitForIdle(); - expect(handles).toHaveLength(2); + expect(handles).toHaveLength(3); expect(handles[0]).not.toBe(handles[1]); + expect(handles[1]).not.toBe(handles[2]); expect(await agent.resourceLedger.waitForSettlement(handles[0]!, { graceMs: 100 })).toEqual({ status: "settled", }); expect(await agent.resourceLedger.waitForSettlement(handles[1]!, { graceMs: 100 })).toEqual({ status: "settled", }); + expect(await agent.resourceLedger.waitForSettlement(handles[2]!, { graceMs: 100 })).toEqual({ + status: "settled", + }); expect(terminalEvents).toHaveLength(1); + expect(lifecycleStarts).toHaveLength(3); + expect(lifecycleStarts[0]?.sdkRunToken).toBeUndefined(); + expect(lifecycleStarts[0]?.lifecycleScope).toBeDefined(); + expect(lifecycleStarts[1]?.lifecycleScope).toBe(lifecycleStarts[0]?.lifecycleScope); + expect(lifecycleStarts[2]?.lifecycleScope).toBe(lifecycleStarts[0]?.lifecycleScope); + + await session.prompt("independent prompt after retry"); + await session.waitForIdle(); + expect(lifecycleStarts).toHaveLength(4); + expect(lifecycleStarts[3]?.lifecycleScope).toBeDefined(); + expect(lifecycleStarts[3]?.lifecycleScope).not.toBe(lifecycleStarts[0]?.lifecycleScope); + expect(terminalEvents).toHaveLength(2); }); it("does not wedge when an auto-retry recovers on a turn ending with a successful yield", async () => { const model = getBundledModel("anthropic", "claude-sonnet-4-5");