fix(eve): prioritize current hook deliveries over stale backlog

.changeset/fresh-session-deliveries-lead.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+Deliveries received on the current session hook now run before older deliveries captured from a replaced hook. This prevents a re-opened session request from being preceded by stale buffered input from a dropped connection.

packages/eve/src/execution/delivery-buffers.ts

@@ -0,0 +1,46 @@
+import type { DeliverHookPayload } from "#channel/types.js";
+
+export interface DeliveryBuffers {
+  readonly currentHookDeliveries: DeliverHookPayload[];
+  readonly replacedHookDeliveries: DeliverHookPayload[];
+  readonly turnDeliveries: DeliverHookPayload[];
+}
+
+export type HookDeliveryBuffer = "current-hook" | "replaced-hook";
+
+export function createDeliveryBuffers(): DeliveryBuffers {
+  return {
+    currentHookDeliveries: [],
+    replacedHookDeliveries: [],
+    turnDeliveries: [],
+  };
+}
+
+export function bufferHookDelivery(
+  buffers: DeliveryBuffers,
+  delivery: DeliverHookPayload,
+  source: HookDeliveryBuffer,
+): void {
+  const target =
+    source === "current-hook" ? buffers.currentHookDeliveries : buffers.replacedHookDeliveries;
+  target.push(delivery);
+}
+
+export function prependTurnDeliveries(
+  buffers: DeliveryBuffers,
+  deliveries: readonly DeliverHookPayload[] | undefined,
+): void {
+  if (deliveries !== undefined) buffers.turnDeliveries.unshift(...deliveries);
+}
+
+export function takeBufferedDelivery(buffers: DeliveryBuffers): DeliverHookPayload | undefined {
+  return (
+    buffers.turnDeliveries.shift() ??
+    buffers.currentHookDeliveries.shift() ??
+    buffers.replacedHookDeliveries.shift()
+  );
+}
+
+export function takePriorityDelivery(buffers: DeliveryBuffers): DeliverHookPayload | undefined {
+  return buffers.turnDeliveries.shift() ?? buffers.currentHookDeliveries.shift();
+}

packages/eve/src/execution/session-delivery-hook.integration.test.ts

@@ -1,4 +1,5 @@
 import { describe, expect, it } from "vitest";
+import { randomUUID } from "node:crypto";
 
 import { sessionDeliveryHookWorkflow } from "#internal/testing/session-delivery-hook-workflow.js";
 import { waitForHook } from "#internal/testing/workflow-test-helpers.js";
@@ -9,7 +10,7 @@ describe("session delivery hook integration", () => {
     ["old then replacement", ["old", "replacement"] as const],
     ["replacement then old", ["replacement", "old"] as const],
   ])("preserves deliveries committed %s during rekey", async (_label, order) => {
-    const suffix = order.join("-");
+    const suffix = `${order.join("-")}-${randomUUID()}`;
     const oldToken = `http:session-delivery-hook:${suffix}:old`;
     const replacementToken = `http:session-delivery-hook:${suffix}:replacement`;
     const disposal = await pauseHookDisposal(oldToken);

packages/eve/src/execution/session-delivery-hook.test.ts

@@ -1,6 +1,7 @@
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
 import type { DeliverHookPayload, HookPayload } from "#channel/types.js";
+import { createDeliveryBuffers } from "#execution/delivery-buffers.js";
 import { createSessionDeliveryHook } from "#execution/session-delivery-hook.js";
 
 const createHookMock = vi.fn();
@@ -23,7 +24,7 @@ describe("createSessionDeliveryHook", () => {
       token: "replacement",
     });
     installHooks(oldHook, replacementHook);
-    const deliveryHook = createSessionDeliveryHook([]);
+    const deliveryHook = createSessionDeliveryHook(createDeliveryBuffers());
 
     await deliveryHook.rekey("old");
     const pending = deliveryHook.next();
@@ -42,8 +43,9 @@ describe("createSessionDeliveryHook", () => {
     expect(replacementHook.dispose).toHaveBeenCalledOnce();
   });
 
-  it("appends drained rekey deliveries after existing buffered deliveries", async () => {
-    const bufferedDeliveries = [deliveryPayload("existing")];
+  it("classifies drained rekey deliveries by hook ownership", async () => {
+    const deliveryBuffers = createDeliveryBuffers();
+    deliveryBuffers.turnDeliveries.push(deliveryPayload("existing"));
     const oldHook = createMockHook({
       reads: [Promise.resolve(delivery("old"))],
       token: "old",
@@ -53,17 +55,15 @@ describe("createSessionDeliveryHook", () => {
       token: "replacement",
     });
     installHooks(oldHook, replacementHook);
-    const deliveryHook = createSessionDeliveryHook(bufferedDeliveries);
+    const deliveryHook = createSessionDeliveryHook(deliveryBuffers);
 
     await deliveryHook.rekey("old");
     await deliveryHook.rekey("replacement");
     await deliveryHook.dispose();
 
-    expect(bufferedDeliveries).toEqual([
-      deliveryPayload("existing"),
-      deliveryPayload("old"),
-      deliveryPayload("replacement"),
-    ]);
+    expect(deliveryBuffers.turnDeliveries).toEqual([deliveryPayload("existing")]);
+    expect(deliveryBuffers.replacedHookDeliveries).toEqual([deliveryPayload("old")]);
+    expect(deliveryBuffers.currentHookDeliveries).toEqual([deliveryPayload("replacement")]);
   });
 
   it("disposes a conflicting candidate without releasing the active hook", async () => {
@@ -73,7 +73,7 @@ describe("createSessionDeliveryHook", () => {
       token: "candidate",
     });
     installHooks(oldHook, candidateHook);
-    const deliveryHook = createSessionDeliveryHook([]);
+    const deliveryHook = createSessionDeliveryHook(createDeliveryBuffers());
 
     await deliveryHook.rekey("old");
     await expect(deliveryHook.rekey("candidate")).rejects.toMatchObject({
@@ -88,6 +88,32 @@ describe("createSessionDeliveryHook", () => {
     expect(oldHook.dispose).toHaveBeenCalledOnce();
   });
 
+  it("keeps the active hook current when a candidate claim fails", async () => {
+    const deliveryBuffers = createDeliveryBuffers();
+    const oldHook = createMockHook({
+      reads: [Promise.resolve(delivery("still-active"))],
+      token: "old",
+    });
+    const candidateHook = createMockHook({
+      conflict: { runId: "wrun_owner" },
+      token: "candidate",
+    });
+    installHooks(oldHook, candidateHook);
+    const deliveryHook = createSessionDeliveryHook(deliveryBuffers);
+
+    await deliveryHook.rekey("old");
+    await expect(deliveryHook.rekey("candidate")).rejects.toMatchObject({
+      name: "HookConflictError",
+      token: "candidate",
+    });
+    await deliveryHook.drainReady();
+
+    expect(deliveryBuffers.currentHookDeliveries).toEqual([deliveryPayload("still-active")]);
+    expect(deliveryBuffers.replacedHookDeliveries).toEqual([]);
+
+    await deliveryHook.dispose();
+  });
+
   it("preserves an old-hook disposal failure while cleaning the candidate", async () => {
     const failure = new Error("old hook disposal failed");
     const oldHook = createMockHook({
@@ -98,7 +124,7 @@ describe("createSessionDeliveryHook", () => {
     });
     const candidateHook = createMockHook({ token: "candidate" });
     installHooks(oldHook, candidateHook);
-    const deliveryHook = createSessionDeliveryHook([]);
+    const deliveryHook = createSessionDeliveryHook(createDeliveryBuffers());
 
     await deliveryHook.rekey("old");
     await expect(deliveryHook.rekey("candidate")).rejects.toBe(failure);
@@ -112,13 +138,13 @@ describe("createSessionDeliveryHook", () => {
     // the armed iterator read). Re-arming the retired hook awaits the exhausted
     // cursor, which rejects; the delivery must surface exactly once and the
     // rejection must not throw or resurrect it across later drain cycles.
-    const bufferedDeliveries: DeliverHookPayload[] = [];
+    const deliveryBuffers = createDeliveryBuffers();
     installHooks(
       createCursorHook({ committed: [deliveryPayload("old")], token: "old" }),
       createCursorHook({ committed: [], token: "replacement" }),
       createCursorHook({ committed: [], token: "third" }),
     );
-    const deliveryHook = createSessionDeliveryHook(bufferedDeliveries);
+    const deliveryHook = createSessionDeliveryHook(deliveryBuffers);
 
     await deliveryHook.rekey("old");
     await deliveryHook.rekey("replacement");
@@ -127,13 +153,14 @@ describe("createSessionDeliveryHook", () => {
     await deliveryHook.rekey("third");
     await deliveryHook.dispose();
 
-    expect(bufferedDeliveries).toEqual([deliveryPayload("old")]);
+    expect(deliveryBuffers.replacedHookDeliveries).toEqual([deliveryPayload("old")]);
+    expect(deliveryBuffers.currentHookDeliveries).toEqual([]);
   });
 
   it("disposes without closing or settling a pending read", async () => {
     const hook = createMockHook({ token: "active" });
     installHooks(hook);
-    const deliveryHook = createSessionDeliveryHook([]);
+    const deliveryHook = createSessionDeliveryHook(createDeliveryBuffers());
 
     await deliveryHook.rekey("active");
     void deliveryHook.next();

packages/eve/src/execution/session-delivery-hook.ts

@@ -1,6 +1,7 @@
 import { createHook, type Hook } from "#compiled/@workflow/core/index.js";
 
-import type { DeliverHookPayload, HookPayload } from "#channel/types.js";
+import type { HookPayload } from "#channel/types.js";
+import { bufferHookDelivery, type DeliveryBuffers } from "#execution/delivery-buffers.js";
 import { claimHookOwnership, disposeHook } from "#execution/hook-ownership.js";
 
 interface HookRead {
@@ -17,11 +18,14 @@ interface SessionDeliveryHookState {
   pending: boolean;
   retired: boolean;
   resolved?: HookRead;
+  superseded: boolean;
 }
 
 /** Reads and rekeys the public delivery hook for one session driver. */
 export interface SessionDeliveryHook {
+  bufferNext(): Promise<"buffered" | "closed">;
   consumeNext(): void;
+  drainReady(): Promise<void>;
   next(): Promise<IteratorResult<HookPayload>>;
   rekey(token: string): Promise<void>;
 }
@@ -40,7 +44,7 @@ export interface SessionDeliveryHookHandle extends SessionDeliveryHook {
  * to `hook_disposed` and receives `HookNotFoundError` from the Workflow SDK.
  */
 export function createSessionDeliveryHook(
-  bufferedDeliveries: DeliverHookPayload[],
+  deliveryBuffers: DeliveryBuffers,
 ): SessionDeliveryHookHandle {
   let active: SessionDeliveryHookState | undefined;
   const retired: SessionDeliveryHookState[] = [];
@@ -98,6 +102,34 @@ export function createSessionDeliveryHook(
     if (state.resolved !== undefined) enqueue(state.resolved);
   };
 
+  const consumeRead = (read: HookRead): void => {
+    read.state.pending = false;
+    read.state.resolved = undefined;
+
+    if (read.result.done) {
+      read.state.closed = true;
+    } else if (read.result.value.kind === "deliver") {
+      bufferHookDelivery(
+        deliveryBuffers,
+        read.result.value,
+        read.state.superseded ? "replaced-hook" : "current-hook",
+      );
+    }
+
+    arm(read.state);
+  };
+
+  const consumeOffered = (): HookRead => {
+    if (offeredRead === undefined) {
+      throw new Error("Cannot consume a public delivery before it resolves.");
+    }
+
+    const read = offeredRead;
+    offeredRead = undefined;
+    offered = null;
+    return read;
+  };
+
   const drainReady = async (): Promise<void> => {
     // A caller may already be racing this read against turn control. Leave
     // that promise intact: the losing race ignores it, and the next call to
@@ -108,31 +140,35 @@ export function createSessionDeliveryHook(
 
     while (ready.length > 0) {
       const read = ready.shift()!;
-      read.state.pending = false;
-      read.state.resolved = undefined;
-
-      if (read.result.done) {
-        read.state.closed = true;
-      } else if (read.result.value.kind === "deliver") {
-        bufferedDeliveries.push(read.result.value);
-      }
-
-      arm(read.state);
+      consumeRead(read);
       await Promise.resolve();
     }
   };
 
   return {
+    async bufferNext(): Promise<"buffered" | "closed"> {
+      await this.next();
+      const read = consumeOffered();
+      consumeRead(read);
+      return read.result.done ? "closed" : "buffered";
+    },
+
     consumeNext(): void {
-      if (offeredRead === undefined) {
-        throw new Error("Cannot consume a public delivery before it resolves.");
+      const read = consumeOffered();
+      read.state.pending = false;
+      read.state.resolved = undefined;
+      if (read.result.done) read.state.closed = true;
+    },
+
+    async drainReady(): Promise<void> {
+      if (active === undefined) {
+        throw new Error("Cannot read deliveries before a continuation token is available.");
       }
 
-      offeredRead.state.pending = false;
-      offeredRead.state.resolved = undefined;
-      if (offeredRead.result.done) offeredRead.state.closed = true;
-      offeredRead = undefined;
-      offered = null;
+      arm(active);
+      for (const state of retired) arm(state);
+
+      await drainReady();
     },
 
     async dispose(): Promise<void> {
@@ -187,6 +223,7 @@ export function createSessionDeliveryHook(
         iterator: candidateHook[Symbol.asyncIterator](),
         pending: false,
         retired: false,
+        superseded: false,
       };
 
       if (active === undefined) {
@@ -200,6 +237,7 @@ export function createSessionDeliveryHook(
       arm(previous);
       arm(candidate);
       await claimHookOwnership(candidate.hook);
+      previous.superseded = true;
       enable(candidate);
       await drainReady();