From f45d207e9aadcb20e767451643883bba8d350855 Mon Sep 17 00:00:00 2001 From: "vercel-gh-bot-3[bot]" <282332853+vercel-gh-bot-3[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 16:34:42 +0000 Subject: [PATCH] fix(eve): preserve stale proxied subagent responses --- .../preserve-stale-subagent-responses.md | 5 + .../src/execution/subagent-hitl-proxy.test.ts | 2 +- .../eve/src/execution/subagent-hitl-proxy.ts | 11 +- .../eve/src/execution/turn-workflow.test.ts | 2 + packages/eve/src/execution/turn-workflow.ts | 3 + .../eve/src/execution/workflow-entry.test.ts | 17 ++- packages/eve/src/execution/workflow-entry.ts | 22 +++- .../eve/src/execution/workflow-steps.test.ts | 119 ++++++++++++++++++ packages/eve/src/execution/workflow-steps.ts | 41 +++++- 9 files changed, 205 insertions(+), 17 deletions(-) create mode 100644 .changeset/preserve-stale-subagent-responses.md diff --git a/.changeset/preserve-stale-subagent-responses.md b/.changeset/preserve-stale-subagent-responses.md new file mode 100644 index 000000000..bd5bd6f0e --- /dev/null +++ b/.changeset/preserve-stale-subagent-responses.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Preserve late responses to completed subagent input prompts as parent follow-up input instead of failing the parent session. diff --git a/packages/eve/src/execution/subagent-hitl-proxy.test.ts b/packages/eve/src/execution/subagent-hitl-proxy.test.ts index c6af68016..87feec680 100644 --- a/packages/eve/src/execution/subagent-hitl-proxy.test.ts +++ b/packages/eve/src/execution/subagent-hitl-proxy.test.ts @@ -102,9 +102,9 @@ describe("routeDeliverPayload", () => { expect(routed.forChildren).toEqual([ { childContinuationToken: "child-a", + parentAction: { kind: "cancel-turn" }, payload: { inputResponses: [{ optionId: "stop", requestId: "req-limit" }] }, }, ]); - expect(routed.parentAction).toEqual({ kind: "cancel-turn" }); }); }); diff --git a/packages/eve/src/execution/subagent-hitl-proxy.ts b/packages/eve/src/execution/subagent-hitl-proxy.ts index ef411516a..24a7d925f 100644 --- a/packages/eve/src/execution/subagent-hitl-proxy.ts +++ b/packages/eve/src/execution/subagent-hitl-proxy.ts @@ -73,10 +73,10 @@ export async function emitProxiedInputRequest(input: { export interface RoutedDeliverPayload { readonly forChildren: readonly { readonly childContinuationToken: string; + readonly parentAction?: { readonly kind: "cancel-turn" }; readonly payload: { readonly inputResponses: readonly InputResponse[] }; }[]; readonly forSelf: DeliverPayload | undefined; - readonly parentAction: { readonly kind: "cancel-turn" } | undefined; } /** Splits a deliver payload into parent-local and proxied-child buckets. */ @@ -89,7 +89,7 @@ export function routeDeliverPayload(input: { const responsesByChild = new Map(); const unroutedResponses: InputResponse[] = []; - let parentAction: RoutedDeliverPayload["parentAction"]; + const childCancellationRequests = new Set(); for (const response of inputResponses) { const route = entries.get(response.requestId); @@ -100,7 +100,7 @@ export function routeDeliverPayload(input: { } if (route.kind === "session-limit" && response.optionId === SESSION_LIMIT_STOP_OPTION_ID) { - parentAction = { kind: "cancel-turn" }; + childCancellationRequests.add(route.childContinuationToken); } const existing = responsesByChild.get(route.childContinuationToken); @@ -115,6 +115,9 @@ export function routeDeliverPayload(input: { const forChildren: RoutedDeliverPayload["forChildren"] = [...responsesByChild.entries()].map( ([childContinuationToken, responses]) => ({ childContinuationToken, + ...(childCancellationRequests.has(childContinuationToken) + ? { parentAction: { kind: "cancel-turn" } as const } + : {}), payload: { inputResponses: responses }, }), ); @@ -138,5 +141,5 @@ export function routeDeliverPayload(input: { const forSelf = Object.keys(remainder).length > 0 ? (remainder as DeliverPayload) : undefined; - return { forChildren, forSelf, parentAction }; + return { forChildren, forSelf }; } diff --git a/packages/eve/src/execution/turn-workflow.test.ts b/packages/eve/src/execution/turn-workflow.test.ts index 22cd1140d..fe14f86f8 100644 --- a/packages/eve/src/execution/turn-workflow.test.ts +++ b/packages/eve/src/execution/turn-workflow.test.ts @@ -865,6 +865,7 @@ describe("turnWorkflow", () => { }); vi.mocked(routeDeliverToChildren).mockResolvedValue({ kind: "cancel-turn", + remainder: { message: "late answer" }, }); vi.mocked(turnStep).mockResolvedValueOnce({ action: "park", @@ -894,6 +895,7 @@ describe("turnWorkflow", () => { serializedContext: { state: "proxied" }, sessionState: proxyState, }, + bufferedDeliveries: [{ kind: "deliver", payloads: [{ message: "late answer" }] }], kind: "turn-result", }); }); diff --git a/packages/eve/src/execution/turn-workflow.ts b/packages/eve/src/execution/turn-workflow.ts index a72af6afb..83c253a80 100644 --- a/packages/eve/src/execution/turn-workflow.ts +++ b/packages/eve/src/execution/turn-workflow.ts @@ -371,6 +371,9 @@ async function waitForRuntimeActionResults(input: { sessionState: input.cursor.sessionState, }); if (routed.kind === "cancel-turn") { + if (routed.remainder !== undefined) { + input.bufferedDeliveries.push({ ...value.delivery, payloads: [routed.remainder] }); + } return routed.kind; } if (routed.remainder !== undefined) { diff --git a/packages/eve/src/execution/workflow-entry.test.ts b/packages/eve/src/execution/workflow-entry.test.ts index d4c626884..996b5b980 100644 --- a/packages/eve/src/execution/workflow-entry.test.ts +++ b/packages/eve/src/execution/workflow-entry.test.ts @@ -57,6 +57,10 @@ vi.mock("./route-child-delivery.js", () => ({ })), })); +vi.mock("./cancel-descendant-turns-step.js", () => ({ + cancelDescendantTurnsStep: vi.fn().mockResolvedValue(undefined), +})); + vi.mock("./delegated-parent-notification.js", () => ({ notifyDelegatedParentStep: vi.fn().mockResolvedValue(undefined), })); @@ -705,6 +709,10 @@ describe("workflowEntry", () => { serializedContext: { "eve.sessionId": "wrun_test_123", settled: true }, sessionState: settledState, }); + vi.mocked(routeDeliverToChildren).mockResolvedValueOnce({ + kind: "cancel-turn", + remainder: { message: "late answer" }, + }); installHookMocks({ deliveryHooks: [ { @@ -732,12 +740,19 @@ describe("workflowEntry", () => { }); expect(result).toEqual({ output: "ok" }); - expect(settleCancelledTurnStep).toHaveBeenCalledExactlyOnceWith({ + expect(settleCancelledTurnStep).toHaveBeenCalledTimes(2); + expect(settleCancelledTurnStep).toHaveBeenNthCalledWith(1, { parentWritable: expect.any(WritableStream), serializedContext: { "eve.sessionId": "wrun_test_123" }, sessionState, }); + expect(settleCancelledTurnStep).toHaveBeenNthCalledWith(2, { + parentWritable: expect.any(WritableStream), + serializedContext: { "eve.sessionId": "wrun_test_123", settled: true }, + sessionState: settledState, + }); expect(vi.mocked(dispatchTurnStep).mock.calls[1]?.[0]).toMatchObject({ + delivery: { kind: "deliver", payloads: [{ message: "late answer" }] }, serializedContext: { settled: true }, sessionState: settledState, }); diff --git a/packages/eve/src/execution/workflow-entry.ts b/packages/eve/src/execution/workflow-entry.ts index 9f19f8700..ca132d7db 100644 --- a/packages/eve/src/execution/workflow-entry.ts +++ b/packages/eve/src/execution/workflow-entry.ts @@ -370,11 +370,23 @@ async function runDriverLoop(input: { serializedContext: action.serializedContext, sessionState: action.sessionState, }); - action = { - ...action, - serializedContext: settled.serializedContext, - sessionState: settled.sessionState, - }; + action = + routed.remainder === undefined + ? { + ...action, + serializedContext: settled.serializedContext, + sessionState: settled.sessionState, + } + : await runTurn({ + delivery: { + auth: nextDeliver.auth, + kind: "deliver", + payloads: [routed.remainder], + requestId: nextDeliver.requestId, + }, + serializedContext: settled.serializedContext, + sessionState: settled.sessionState, + }); continue; } diff --git a/packages/eve/src/execution/workflow-steps.test.ts b/packages/eve/src/execution/workflow-steps.test.ts index 24a7e021e..74a753e26 100644 --- a/packages/eve/src/execution/workflow-steps.test.ts +++ b/packages/eve/src/execution/workflow-steps.test.ts @@ -17,6 +17,7 @@ import { import { BundleKey, ChannelKey } from "#runtime/sessions/runtime-context-keys.js"; import { serializeContext } from "#context/serialize.js"; import { setPendingRuntimeActionBatch } from "#harness/runtime-actions.js"; +import { upsertProxyInputRequests } from "#harness/proxy-input-requests.js"; import { getAgentHandleStore } from "#harness/handles/store.js"; import { requestTurnSleep } from "#harness/turn-sleep.js"; import { getPendingAuthorization, setPendingAuthorization } from "#harness/authorization.js"; @@ -40,6 +41,7 @@ import { emitTerminalSessionFailureStep } from "#execution/terminal-session-fail import { dispatchTurnStep, resolveEffectiveOutputSchema, + routeProxiedDeliverStep, turnStep, } from "#execution/workflow-steps.js"; import { @@ -47,6 +49,7 @@ import { turnWorkflowReference, workflowEntryReference, } from "#execution/workflow-runtime.js"; +import { resumeHook } from "#internal/workflow/runtime.js"; vi.mock("./durable-session-store.js", async (importOriginal) => { const actual = await importOriginal(); @@ -194,6 +197,7 @@ function createSerializedContext(): Record { afterEach(() => { getRunMock.mockReset(); startMock.mockReset(); + vi.mocked(resumeHook).mockReset(); workflowWritesByNamespace.clear(); vi.unstubAllGlobals(); vi.unstubAllEnvs(); @@ -312,6 +316,121 @@ describe("dispatchTurnStep", () => { }); }); +describe("routeProxiedDeliverStep", () => { + it("preserves a response when its proxied child hook is already disposed", async () => { + const childContinuationToken = "subagent:parent:call-1"; + const requestId = "question-1"; + const session = upsertProxyInputRequests({ + entries: [[requestId, { childContinuationToken, kind: "question" }]], + forChildContinuationToken: childContinuationToken, + session: createStubSession(), + }); + installSessionStoreMocks([session]); + const { HookNotFoundError } = await import("#compiled/@workflow/errors/index.js"); + vi.mocked(resumeHook).mockRejectedValueOnce(new HookNotFoundError(childContinuationToken)); + + await expect( + routeProxiedDeliverStep({ + parentWritable: createTestWritable(), + payload: { inputResponses: [{ optionId: "candidate", requestId }] }, + sessionState: createStubSessionState({ hasProxyInputRequests: true }), + }), + ).resolves.toEqual({ + kind: "continue", + remainder: { inputResponses: [{ optionId: "candidate", requestId }] }, + }); + }); + + it("only removes responses accepted by live children", async () => { + const liveChild = "subagent:parent:call-live"; + const staleChild = "subagent:parent:call-stale"; + const session = upsertProxyInputRequests({ + entries: [["limit-stale", { childContinuationToken: staleChild, kind: "session-limit" }]], + forChildContinuationToken: staleChild, + session: upsertProxyInputRequests({ + entries: [["question-live", { childContinuationToken: liveChild, kind: "question" }]], + forChildContinuationToken: liveChild, + session: createStubSession(), + }), + }); + installSessionStoreMocks([session]); + const { HookNotFoundError } = await import("#compiled/@workflow/errors/index.js"); + vi.mocked(resumeHook) + .mockResolvedValueOnce(undefined as never) + .mockRejectedValueOnce(new HookNotFoundError(staleChild)); + + await expect( + routeProxiedDeliverStep({ + parentWritable: createTestWritable(), + payload: { + inputResponses: [ + { optionId: "candidate", requestId: "question-live" }, + { optionId: "stop", requestId: "limit-stale" }, + ], + }, + sessionState: createStubSessionState({ hasProxyInputRequests: true }), + }), + ).resolves.toEqual({ + kind: "continue", + remainder: { inputResponses: [{ optionId: "stop", requestId: "limit-stale" }] }, + }); + }); + + it("preserves stale responses while cancelling for a live child Stop", async () => { + const liveChild = "subagent:parent:call-live"; + const staleChild = "subagent:parent:call-stale"; + const session = upsertProxyInputRequests({ + entries: [["question-stale", { childContinuationToken: staleChild, kind: "question" }]], + forChildContinuationToken: staleChild, + session: upsertProxyInputRequests({ + entries: [["limit-live", { childContinuationToken: liveChild, kind: "session-limit" }]], + forChildContinuationToken: liveChild, + session: createStubSession(), + }), + }); + installSessionStoreMocks([session]); + const { HookNotFoundError } = await import("#compiled/@workflow/errors/index.js"); + vi.mocked(resumeHook) + .mockResolvedValueOnce(undefined as never) + .mockRejectedValueOnce(new HookNotFoundError(staleChild)); + + await expect( + routeProxiedDeliverStep({ + parentWritable: createTestWritable(), + payload: { + inputResponses: [ + { optionId: "stop", requestId: "limit-live" }, + { optionId: "candidate", requestId: "question-stale" }, + ], + }, + sessionState: createStubSessionState({ hasProxyInputRequests: true }), + }), + ).resolves.toEqual({ + kind: "cancel-turn", + remainder: { inputResponses: [{ optionId: "candidate", requestId: "question-stale" }] }, + }); + }); + + it("propagates child delivery failures other than a missing hook", async () => { + const childContinuationToken = "subagent:parent:call-1"; + const session = upsertProxyInputRequests({ + entries: [["question-1", { childContinuationToken, kind: "question" }]], + forChildContinuationToken: childContinuationToken, + session: createStubSession(), + }); + installSessionStoreMocks([session]); + vi.mocked(resumeHook).mockRejectedValueOnce(new Error("delivery failed")); + + await expect( + routeProxiedDeliverStep({ + parentWritable: createTestWritable(), + payload: { inputResponses: [{ optionId: "candidate", requestId: "question-1" }] }, + sessionState: createStubSessionState({ hasProxyInputRequests: true }), + }), + ).rejects.toThrow("delivery failed"); + }); +}); + describe("dispatchRuntimeActionsStep", () => { it("preserves a started local child when a later start fails", async () => { vi.stubEnv("VERCEL_ENV", "production"); diff --git a/packages/eve/src/execution/workflow-steps.ts b/packages/eve/src/execution/workflow-steps.ts index 0d0d38a71..2221efebb 100644 --- a/packages/eve/src/execution/workflow-steps.ts +++ b/packages/eve/src/execution/workflow-steps.ts @@ -1,3 +1,5 @@ +import { HookNotFoundError } from "#compiled/@workflow/errors/index.js"; + import { buildAdapterContext } from "#channel/adapter-context.js"; import { callAdapterEventHandler, defaultDeliverResult } from "#channel/adapter.js"; import type { DeliverPayload, SessionAuthContext } from "#channel/types.js"; @@ -41,6 +43,7 @@ import { getTurnUsageState, toUsage } from "#harness/turn-tag-state.js"; import type { TokenUsage } from "#shared/token-usage.js"; import type { JsonObject } from "#shared/json.js"; import type { RunMode } from "#shared/run-mode.js"; +import type { InputResponse } from "#runtime/input/types.js"; import { getRuntimeActionRequestKey } from "#runtime/actions/keys.js"; import { createAuthorizationCompletedEvent, @@ -575,6 +578,7 @@ export function resolveEffectiveOutputSchema(input: { export type RoutedDeliverResult = | { readonly kind: "cancel-turn"; + readonly remainder: DeliverPayload | undefined; } | { readonly kind: "continue"; @@ -601,15 +605,40 @@ export async function routeProxiedDeliverStep(input: { state: durableSession.state, }); + const deliveredResponses = new Set(); + let parentAction: { readonly kind: "cancel-turn" } | undefined; + for (const forChild of routed.forChildren) { - await resumeHook(forChild.childContinuationToken, { - auth: input.auth, - kind: "deliver", - payloads: [forChild.payload], - }); + try { + await resumeHook(forChild.childContinuationToken, { + auth: input.auth, + kind: "deliver", + payloads: [forChild.payload], + }); + } catch (error) { + if (HookNotFoundError.is(error)) { + continue; + } + throw error; + } + + for (const response of forChild.payload.inputResponses) { + deliveredResponses.add(response); + } + parentAction ??= forChild.parentAction; } - return routed.parentAction ?? { kind: "continue", remainder: routed.forSelf }; + const undeliveredResponses = input.payload.inputResponses?.filter( + (response) => !deliveredResponses.has(response), + ); + const remainder = + undeliveredResponses === undefined || undeliveredResponses.length === 0 + ? routed.forSelf + : { ...routed.forSelf, inputResponses: undeliveredResponses }; + + return parentAction === undefined + ? { kind: "continue", remainder } + : { ...parentAction, remainder }; } /** Starts a per-turn child workflow for the current driver session. */