fix(eve): preserve stale proxied subagent responses

.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.

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

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<string, InputResponse[]>();
   const unroutedResponses: InputResponse[] = [];
-  let parentAction: RoutedDeliverPayload["parentAction"];
+  const childCancellationRequests = new Set<string>();
 
   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 };
 }

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

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) {

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

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

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,13 +41,15 @@ import { emitTerminalSessionFailureStep } from "#execution/terminal-session-fail
 import {
   dispatchTurnStep,
   resolveEffectiveOutputSchema,
+  routeProxiedDeliverStep,
   turnStep,
 } from "#execution/workflow-steps.js";
 import {
   LATEST_DEPLOYMENT_UNSUPPORTED_MESSAGE,
   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<typeof import("./durable-session-store.js")>();
@@ -194,6 +197,7 @@ function createSerializedContext(): Record<string, unknown> {
 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");

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<InputResponse>();
+  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. */