fix(eve): workflow lifecycle - replace polling with acknowledgements

.changeset/known-session-inboxes.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+Persist the receiving session's inbox address and wire version with local subagent input requests. Replies can resume the original child directly without reading hook metadata, including after the child's continuation address changes.

.changeset/quiet-hook-waits.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+Replace polling during subagent and task startup, session reset, and task cancellation with workflow ownership acknowledgements and streamed lifecycle signals. Reset completion requires a session started on this version; older pinned sessions can time out waiting for the new cleanup signal.

docs/concepts/sessions-runs-and-streaming.md

@@ -226,6 +226,8 @@ Clear removes model-message history in place, including static and dynamic user-
 
 Reset terminally retires the exact session ID. A reset ID never becomes a new session; create another session explicitly for a fresh conversation. Compact, clear, and reset return `"no_active_session"` when the target is already inactive.
 
+Reset waits up to 30 seconds for the session driver to acknowledge that it released its command hooks. Sessions pinned to deployments that predate lifecycle acknowledgements can reset successfully but still return a timeout. Retry reset after that timeout; `"no_active_session"` confirms the old session has retired.
+
 ## Reconnect and rewind
 
 The stream is durable. Every event is recorded before a step completes, so consumers can reconnect from their cursor when an HTTP connection ends. A nonnegative `startIndex` is an absolute event count: use it to pick up where you dropped off or pass `0` to rewind to the start.

e2e/fixtures/agent-cancellation/evals/cancellation/reset-channel-session.eval.ts

@@ -37,7 +37,7 @@ async function postJson<T>(target: EveEvalTargetHandle, path: string, body: unkn
  * for a different workflow session.
  */
 export default defineEval({
-  tags: ["real-model"],
+  tags: ["session-inbox"],
   description: "Reset a custom-channel session without creating a model turn for /new.",
   timeoutMs: 240_000,
 

e2e/fixtures/agent-subagents-hitl/evals/hitl.eval.ts

@@ -16,7 +16,7 @@ type SessionCursor = Pick<
  * Parking is server-side.
  */
 export default defineEval({
-  tags: ["real-model"],
+  tags: ["session-inbox"],
   description: "Subagent tool approval proxied through the parent session.",
   timeoutMs: 90_000,
 

packages/eve/src/channel/types.ts

@@ -341,6 +341,7 @@ export interface SubagentInputRequestHookPayload {
   readonly callId: string;
   readonly childContinuationToken: string;
   readonly childSessionId: string;
+  readonly childSessionInbox?: import("#execution/wire/session-inbox-contract.js").SessionInboxAddress;
   readonly event: SubagentInputRequestEvent;
   readonly kind: "subagent-input-request";
   readonly subagentName: string;

packages/eve/src/context/keys.ts

@@ -18,6 +18,10 @@ import type {
   SessionTurn,
 } from "#channel/types.js";
 import { ContextKey } from "#context/key.js";
+import {
+  SESSION_INBOX_CONTEXT_KEY,
+  type SessionInboxAddress,
+} from "#execution/wire/session-inbox-contract.js";
 import { SESSION_CALLBACK_CONTEXT_KEY_NAME } from "#context/key-names.js";
 import type { InstrumentationChannelDeliveryRef } from "#instrumentation/lifecycle.js";
 import type { HandleEventFn } from "#harness/types.js";
@@ -70,6 +74,7 @@ export interface Session {
 export const AuthKey = new ContextKey<SessionAuthContext | null>("eve.auth");
 export const InitiatorAuthKey = new ContextKey<SessionAuthContext | null>("eve.initiatorAuth");
 export const SessionIdKey = new ContextKey<string>("eve.sessionId");
+export const SessionInboxKey = new ContextKey<SessionInboxAddress>(SESSION_INBOX_CONTEXT_KEY);
 export const ContinuationTokenKey = new ContextKey<string>("eve.continuationToken");
 export const ChannelRequestIdKey = new ContextKey<string>("eve.channelRequestId");
 /** Parent-verified local client provenance, valid only for the current dev host secret. */

packages/eve/src/execution/claim-session-inbox.ts

@@ -0,0 +1,30 @@
+import { isHookConflictError } from "#execution/hook-ownership.js";
+import type { SessionCommandInbox } from "#execution/session-command-inbox.js";
+import { sessionCommandHookToken } from "#execution/session-command-token.js";
+import { publishWorkflowOwnershipStep } from "#execution/workflow-lifecycle-step.js";
+
+/** Claims the session's inboxes and acknowledges the winner before session initialization settles. */
+export async function claimSessionInbox(input: {
+  readonly acknowledgeOwnership?: boolean;
+  readonly commandInbox: SessionCommandInbox;
+  readonly continuationToken: string;
+  readonly sessionId: string;
+}): Promise<PromiseSettledResult<void>> {
+  const [stableClaim, authorizationClaim, continuationClaim] = await Promise.allSettled([
+    input.commandInbox.claimStable(sessionCommandHookToken(input.sessionId)),
+    input.commandInbox.claimAuthorization(`${input.sessionId}:auth`),
+    input.commandInbox.rekeyContinuation(input.continuationToken),
+  ]);
+  if (stableClaim.status === "rejected") throw stableClaim.reason;
+  if (authorizationClaim.status === "rejected") throw authorizationClaim.reason;
+  if (input.acknowledgeOwnership === true) {
+    let runId = input.sessionId;
+    if (continuationClaim.status === "rejected") {
+      const error = continuationClaim.reason;
+      if (!isHookConflictError(error) || typeof error.conflictingRunId !== "string") throw error;
+      runId = error.conflictingRunId;
+    }
+    await publishWorkflowOwnershipStep({ runId });
+  }
+  return continuationClaim;
+}

packages/eve/src/execution/proxied-deliver-step.ts

@@ -1,3 +1,4 @@
+import type { SessionInboxAddress } from "#execution/wire/session-inbox-contract.js";
 import type { DeliverHookPayload, DeliverPayload, SessionAuthContext } from "#channel/types.js";
 import { coalesceDeliverPayloads } from "#execution/deliver-payloads.js";
 import {
@@ -46,6 +47,7 @@ type LegacyRoutedDeliverResult =
 interface ChildBucket {
   readonly answerHook?: AnswerHookRoute;
   readonly childContinuationToken: string;
+  readonly childSessionInbox?: SessionInboxAddress;
   readonly childResponseUrl?: string;
   readonly metadata: NonNullable<DeliverHookPayload["deliveryMetadata"]>[number][];
   readonly payloads: DeliverPayload[];
@@ -108,17 +110,16 @@ export async function routeProxiedDeliverStep(
     if (routed.forSelf !== undefined) parentPayloads.set(sourcePayloadIndex, routed.forSelf);
 
     for (const [childIndex, forChild] of routed.forChildren.entries()) {
-      const key =
-        forChild.taskId === undefined
-          ? forChild.childContinuationToken
-          : [
-              forChild.childContinuationToken,
-              forChild.childResponseUrl ?? "",
-              forChild.taskId,
-            ].join("\0");
+      const key = [
+        forChild.childContinuationToken,
+        forChild.childSessionInbox?.sessionId ?? "",
+        forChild.childResponseUrl ?? "",
+        forChild.taskId ?? "",
+      ].join("\0");
       const child = children.get(key) ?? {
         answerHook: forChild.answerHook,
         childContinuationToken: forChild.childContinuationToken,
+        childSessionInbox: forChild.childSessionInbox,
         childResponseUrl: forChild.childResponseUrl,
         metadata: [],
         payloads: [],
@@ -158,6 +159,7 @@ export async function routeProxiedDeliverStep(
         payload: {
           auth: sourceDelivery.auth,
           childContinuationToken: child.childContinuationToken,
+          childSessionInbox: child.childSessionInbox,
           childResponseUrl: child.childResponseUrl,
           inputResponses: coalesceDeliverPayloads(child.payloads).inputResponses ?? [],
           kind: "input-response",
@@ -188,7 +190,10 @@ export async function routeProxiedDeliverStep(
       deliveryMetadata: child.metadata.length === 0 ? undefined : child.metadata,
       payloads: child.payloads,
     };
-    await resumeSessionInbox(child.childContinuationToken, childDelivery);
+    await resumeSessionInbox(
+      child.childSessionInbox ?? child.childContinuationToken,
+      childDelivery,
+    );
     // Successfully forwarded request IDs are retired so later deliveries
     // cannot route through stale entries.
     durableSession = retireProxyInputRequests(durableSession, child.retireRequestIds);

packages/eve/src/execution/tasks/child/steps.test.ts

@@ -166,6 +166,21 @@ describe("deliverTaskInputResponsesStep", () => {
     });
   });
 
+  it("uses the persisted child address after its continuation alias changes", async () => {
+    const childSessionInbox = { sessionId: "original-child", version: 1 };
+    await deliverTaskInputResponsesStep({
+      answer: { ...answer, childSessionInbox },
+      requestIds: ["req-1"],
+    });
+
+    expect(resumeSessionInbox).toHaveBeenCalledWith(
+      childSessionInbox,
+      expect.objectContaining({
+        payload: { inputResponses: [{ optionId: "approve", requestId: "req-1" }] },
+      }),
+    );
+  });
+
   it("posts a remote child answer to its narrowed task-input route", async () => {
     const fetchMock = vi.fn().mockResolvedValue(new Response(null, { status: 202 }));
     vi.stubGlobal("fetch", fetchMock);

packages/eve/src/execution/tasks/child/steps.ts

@@ -329,7 +329,10 @@ export async function deliverTaskInputResponsesStep(input: {
         command.payload.inputResponses,
       );
     } else {
-      await resumeSessionInbox(input.answer.childContinuationToken, command);
+      await resumeSessionInbox(
+        input.answer.childSessionInbox ?? input.answer.childContinuationToken,
+        command,
+      );
     }
     return "delivered";
   } catch (error) {

packages/eve/src/execution/tasks/child/workflow.test.ts

@@ -9,6 +9,9 @@ const mocks = vi.hoisted(() => ({
   appendTaskViewStep: vi.fn(),
   cancelWorkflowToolRunStep: vi.fn(),
   claimHookOwnership: vi.fn(),
+  isHookConflictError: vi.fn(),
+  publishWorkflowOwnershipStep: vi.fn(),
+  publishWorkflowCleanupStep: vi.fn(),
   createChannelReader: vi.fn((channel: string) => ({ channel, iterator: [][Symbol.iterator]() })),
   createHook: vi.fn(() => ({ token: "task-token" })),
   deliverTaskInputResponsesStep: vi.fn(),
@@ -37,11 +40,16 @@ const mocks = vi.hoisted(() => ({
 vi.mock("#compiled/@workflow/core/index.js", async (importOriginal) => ({
   ...(await importOriginal<typeof import("#compiled/@workflow/core/index.js")>()),
   createHook: mocks.createHook,
+  getWorkflowMetadata: () => ({ workflowRunId: "task-run" }),
+}));
+vi.mock("#execution/workflow-lifecycle-step.js", () => ({
+  publishWorkflowOwnershipStep: mocks.publishWorkflowOwnershipStep,
+  publishWorkflowCleanupStep: mocks.publishWorkflowCleanupStep,
 }));
 vi.mock("#execution/hook-ownership.js", () => ({
   claimHookOwnership: mocks.claimHookOwnership,
   disposeHook: mocks.disposeHook,
-  isHookConflictError: () => false,
+  isHookConflictError: mocks.isHookConflictError,
 }));
 vi.mock("#execution/tasks/child/steps.js", () => ({
   appendTaskProgressStep: mocks.appendTaskProgressStep,
@@ -117,6 +125,37 @@ describe("taskRunWorkflow", () => {
     });
   });
 
+  it("acknowledges a duplicate's actual owner without executing its body", async () => {
+    const conflict = { name: "HookConflictError", conflictingRunId: "winning-run" };
+    mocks.claimHookOwnership.mockRejectedValue(conflict);
+    mocks.isHookConflictError.mockReturnValue(true);
+
+    await taskRunWorkflow({
+      initialView,
+      parentContinuationToken: "parent-token",
+      taskInboxToken: "task-token",
+    });
+
+    expect(mocks.publishWorkflowOwnershipStep).toHaveBeenCalledWith({ runId: "winning-run" });
+    expect(mocks.appendTaskViewStep).not.toHaveBeenCalled();
+    expect(mocks.executeWorkflowBody).not.toHaveBeenCalled();
+    expect(mocks.publishWorkflowCleanupStep).not.toHaveBeenCalled();
+  });
+
+  it("does not acknowledge a failed hook claim as successful startup", async () => {
+    const failure = new Error("claim failed");
+    mocks.claimHookOwnership.mockRejectedValue(failure);
+
+    await expect(
+      taskRunWorkflow({
+        initialView,
+        parentContinuationToken: "parent-token",
+        taskInboxToken: "task-token",
+      }),
+    ).rejects.toBe(failure);
+    expect(mocks.publishWorkflowOwnershipStep).not.toHaveBeenCalled();
+  });
+
   it("delivers an authored message queued before completion and dispatch acknowledgement", async () => {
     const message = {
       callId: "call-1",
@@ -148,6 +187,10 @@ describe("taskRunWorkflow", () => {
     expect(mocks.wakeTaskMessageParentStep.mock.invocationCallOrder[0]).toBeLessThan(
       mocks.wakeTaskParentStep.mock.invocationCallOrder[0]!,
     );
+    expect(mocks.publishWorkflowCleanupStep).toHaveBeenCalledOnce();
+    expect(mocks.disposeHook.mock.invocationCallOrder[0]).toBeLessThan(
+      mocks.publishWorkflowCleanupStep.mock.invocationCallOrder[0]!,
+    );
   });
 
   it("buffers agent requests until task dispatch is acknowledged", async () => {

packages/eve/src/execution/tasks/child/workflow.ts

@@ -1,4 +1,8 @@
-import { createHook } from "#compiled/@workflow/core/index.js";
+import { createHook, getWorkflowMetadata } from "#compiled/@workflow/core/index.js";
+import {
+  publishWorkflowCleanupStep,
+  publishWorkflowOwnershipStep,
+} from "#execution/workflow-lifecycle-step.js";
 
 import type { ActivityObserverConfig } from "#channel/types.js";
 import { claimHookOwnership, disposeHook, isHookConflictError } from "#execution/hook-ownership.js";
@@ -107,10 +111,14 @@ export async function taskRunWorkflow(input: TaskRunWorkflowInput): Promise<void
       await claimHookOwnership(commands);
       ownsHook = true;
     } catch (error) {
-      if (isHookConflictError(error)) return;
+      if (isHookConflictError(error) && typeof error.conflictingRunId === "string") {
+        await publishWorkflowOwnershipStep({ runId: error.conflictingRunId });
+        return;
+      }
       throw error;
     }
 
+    await publishWorkflowOwnershipStep({ runId: getWorkflowMetadata().workflowRunId });
     await appendTaskViewStep({ activityObserver: input.activityObserver, view });
     while (true) {
       // Hook persistence does not mean the owner has consumed every report yet.
@@ -177,6 +185,7 @@ export async function taskRunWorkflow(input: TaskRunWorkflowInput): Promise<void
     if (ownsHook) {
       await workflowToolRunChannels.dispose();
       await disposeHook(commands);
+      await publishWorkflowCleanupStep();
     }
   }
 

packages/eve/src/execution/tasks/parent/delegate.ts

@@ -10,7 +10,6 @@ import {
   sendTaskCommand,
   sendTaskCommandToOwner,
   startTaskRun,
-  waitForTaskCommandOwner,
 } from "#execution/tasks/parent/run-parent.js";
 import { sessionCommandHookToken } from "#execution/session-command-token.js";
 import type { JsonValue } from "#shared/json.js";
@@ -68,13 +67,12 @@ export async function beginBackgroundTask(input: {
   readonly session: HarnessSession;
 }): Promise<BackgroundTask> {
   const task = prepareBackgroundTask(input);
-  await startTaskRun({
+  const owner = await startTaskRun({
     activityObserver: input.activityObserver,
     taskInboxToken: task.taskInboxToken,
     initialView: { metadata: task.metadata, status: "working", taskId: task.taskId },
     parentContinuationToken: sessionCommandHookToken(input.session.sessionId),
   });
-  const owner = await waitForTaskCommandOwner({ taskInboxToken: task.taskInboxToken });
   return { ...task, taskRunId: owner.runId };
 }
 
@@ -92,7 +90,6 @@ export async function acknowledgeDelegatedTasksStep(input: {
     const owner = await sendTaskCommandToOwner({
       command: { kind: "ready" },
       taskInboxToken: task.taskInboxToken,
-      retryUnreachable: { attempts: 20, delayMs: 250 },
     });
     if (owner !== undefined) continue;
     const view = await readLatestTaskView({ taskRunId: task.taskRunId });
@@ -109,6 +106,5 @@ export async function rejectDelegatedDispatch(input: {
   await sendTaskCommand({
     command: { data: input.error, kind: "reject-dispatch" },
     taskInboxToken: input.task.taskInboxToken,
-    retryUnreachable: { attempts: 20, delayMs: 250 },
   });
 }

packages/eve/src/execution/tasks/parent/dispatch.test.ts

@@ -1,4 +1,5 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { WORKFLOW_CLEANUP_STREAM_NAMESPACE } from "#execution/workflow-lifecycle-contract.js";
 
 import { cancelOwnedTask, executeTaskControlAction } from "#execution/tasks/parent/dispatch.js";
 import { readLatestTaskView, sendTaskCommand } from "#execution/tasks/parent/run-parent.js";
@@ -34,7 +35,21 @@ describe("task cancellation", () => {
     vi.resetAllMocks();
     vi.useFakeTimers();
     vi.mocked(sendTaskCommand).mockResolvedValue("delivered");
-    getRun.mockReturnValue({ status: Promise.resolve("completed") });
+    getRun.mockReturnValue({
+      getReadable: (options: { namespace?: string }) =>
+        new ReadableStream({
+          async start(controller) {
+            if (options.namespace === WORKFLOW_CLEANUP_STREAM_NAMESPACE) {
+              controller.enqueue({ released: true });
+            } else {
+              controller.enqueue(
+                await vi.mocked(readLatestTaskView)({ taskRunId: entry.taskRunId }),
+              );
+            }
+            controller.close();
+          },
+        }),
+    });
   });
 
   afterEach(() => {
@@ -82,7 +97,18 @@ describe("task cancellation", () => {
       status: "cancelled",
       taskId: entry.taskId,
     });
-    getRun.mockReturnValue({ status: Promise.resolve("running") });
+    getRun.mockReturnValue({
+      getReadable: (options: { namespace?: string }) =>
+        new ReadableStream({
+          async start(controller) {
+            if (options.namespace !== WORKFLOW_CLEANUP_STREAM_NAMESPACE) {
+              controller.enqueue(
+                await vi.mocked(readLatestTaskView)({ taskRunId: entry.taskRunId }),
+              );
+            }
+          },
+        }),
+    });
 
     const cancelled = cancelOwnedTask({ entry });
     await vi.runAllTimersAsync();