diff --git a/.changeset/fresh-session-deliveries-lead.md b/.changeset/fresh-session-deliveries-lead.md new file mode 100644 index 000000000..623699b25 --- /dev/null +++ b/.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. diff --git a/packages/eve/src/execution/delivery-buffers.ts b/packages/eve/src/execution/delivery-buffers.ts new file mode 100644 index 000000000..0666be926 --- /dev/null +++ b/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(); +} diff --git a/packages/eve/src/execution/session-delivery-hook.integration.test.ts b/packages/eve/src/execution/session-delivery-hook.integration.test.ts index 055bc4528..6e527fe9c 100644 --- a/packages/eve/src/execution/session-delivery-hook.integration.test.ts +++ b/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); diff --git a/packages/eve/src/execution/session-delivery-hook.test.ts b/packages/eve/src/execution/session-delivery-hook.test.ts index d3337c139..1d13954af 100644 --- a/packages/eve/src/execution/session-delivery-hook.test.ts +++ b/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(); diff --git a/packages/eve/src/execution/session-delivery-hook.ts b/packages/eve/src/execution/session-delivery-hook.ts index 6af0d2970..8a6ec7687 100644 --- a/packages/eve/src/execution/session-delivery-hook.ts +++ b/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; next(): Promise>; rekey(token: string): Promise; } @@ -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 => { // 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 { + 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 { @@ -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(); diff --git a/packages/eve/src/execution/turn-control-receiver.test.ts b/packages/eve/src/execution/turn-control-receiver.test.ts index 3e587039f..8ca40cd2c 100644 --- a/packages/eve/src/execution/turn-control-receiver.test.ts +++ b/packages/eve/src/execution/turn-control-receiver.test.ts @@ -1,6 +1,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { DeliverHookPayload, HookPayload } from "#channel/types.js"; +import { createDeliveryBuffers, type DeliveryBuffers } from "#execution/delivery-buffers.js"; import type { DurableSessionState } from "#execution/durable-session-store.js"; import { forwardTurnDeliveryStep } from "#execution/forward-turn-delivery-step.js"; import type { SessionDeliveryHook } from "#execution/session-delivery-hook.js"; @@ -30,16 +31,17 @@ describe("TurnControlReceiver", () => { { kind: "turn-delivery-accepted", requestId: "req-1" }, parkResult(), ]); - const bufferedDeliveries: DeliverHookPayload[] = [delivery]; + const deliveryBuffers = createDeliveryBuffers(); + deliveryBuffers.turnDeliveries.push(delivery); - const action = await runReceiver(bufferedDeliveries); + const action = await runReceiver(deliveryBuffers); expect(forwardTurnDeliveryStep).toHaveBeenCalledWith({ inboxToken: "turn-inbox", payload: { delivery, kind: "driver-delivery", requestId: "req-1" }, }); expect(action.kind).toBe("park"); - expect(bufferedDeliveries).toEqual([]); + expect(deliveryBuffers.turnDeliveries).toEqual([]); }); it("re-buffers the outstanding delivery when the turn cancels its request", async () => { @@ -49,13 +51,14 @@ describe("TurnControlReceiver", () => { { kind: "turn-delivery-cancelled", requestId: "req-1" }, parkResult(), ]); - const bufferedDeliveries: DeliverHookPayload[] = [delivery]; + const deliveryBuffers = createDeliveryBuffers(); + deliveryBuffers.turnDeliveries.push(delivery); - const action = await runReceiver(bufferedDeliveries); + const action = await runReceiver(deliveryBuffers); expect(forwardTurnDeliveryStep).toHaveBeenCalledOnce(); expect(action.kind).toBe("park"); - expect(bufferedDeliveries).toEqual([delivery]); + expect(deliveryBuffers.turnDeliveries).toEqual([delivery]); }); it("re-buffers an unresolved forwarded delivery when the turn terminates", async () => { @@ -67,41 +70,71 @@ describe("TurnControlReceiver", () => { kind: "turn-result", }, ]); - const bufferedDeliveries: DeliverHookPayload[] = [delivery]; + const deliveryBuffers = createDeliveryBuffers(); + deliveryBuffers.turnDeliveries.push(delivery); - const action = await runReceiver(bufferedDeliveries); + const action = await runReceiver(deliveryBuffers); expect(action).toMatchObject({ kind: "done", output: "bye" }); - expect(bufferedDeliveries).toEqual([delivery]); + expect(deliveryBuffers.turnDeliveries).toEqual([delivery]); }); it("hands the turn's remainders back ahead of existing buffered deliveries", async () => { const earlier: DeliverHookPayload = { kind: "deliver", payloads: [{ message: "earlier" }] }; const handBack: DeliverHookPayload = { kind: "deliver", payloads: [{ message: "from-turn" }] }; installControlHook([{ ...parkResult(), bufferedDeliveries: [handBack] }]); - const bufferedDeliveries: DeliverHookPayload[] = [earlier]; + const deliveryBuffers = createDeliveryBuffers(); + deliveryBuffers.turnDeliveries.push(earlier); - await runReceiver(bufferedDeliveries); + await runReceiver(deliveryBuffers); - expect(bufferedDeliveries.map((item) => item.payloads[0]?.message)).toEqual([ + expect(deliveryBuffers.turnDeliveries.map((item) => item.payloads[0]?.message)).toEqual([ "from-turn", "earlier", ]); }); + it("drains current-hook deliveries before serving replaced-hook backlog", async () => { + const stale: DeliverHookPayload = { kind: "deliver", payloads: [{ message: "stale" }] }; + const fresh: DeliverHookPayload = { kind: "deliver", payloads: [{ message: "fresh" }] }; + installControlHook([ + deliveryRequest("req-1"), + { kind: "turn-delivery-accepted", requestId: "req-1" }, + parkResult(), + ]); + const deliveryBuffers = createDeliveryBuffers(); + deliveryBuffers.replacedHookDeliveries.push(stale); + const deliveryHook = createDeliveryHook({ + drainReady: vi.fn(async () => { + if (deliveryBuffers.currentHookDeliveries.length === 0) { + deliveryBuffers.currentHookDeliveries.push(fresh); + } + }), + }); + + await runReceiver(deliveryBuffers, deliveryHook); + + expect(forwardTurnDeliveryStep).toHaveBeenCalledWith({ + inboxToken: "turn-inbox", + payload: { delivery: fresh, kind: "driver-delivery", requestId: "req-1" }, + }); + expect(deliveryBuffers.replacedHookDeliveries).toEqual([stale]); + }); + it("rethrows a rebuilt error when the turn reports a failure", async () => { installControlHook([{ error: { message: "boom", name: "TurnError" }, kind: "turn-error" }]); - await expect(runReceiver([])).rejects.toThrow("boom"); + await expect(runReceiver(createDeliveryBuffers())).rejects.toThrow("boom"); }); }); function runReceiver( - bufferedDeliveries: DeliverHookPayload[], + deliveryBuffers: DeliveryBuffers, + deliveryHook: SessionDeliveryHook = createDeliveryHook(), ): ReturnType { const receiver = new TurnControlReceiver({ - bufferedDeliveries, - deliveryHook: createDeliveryHook(), + deliveryBuffers, + deliveryHook, token: "turn-control", }); return receiver.waitForAction().finally(() => receiver.dispose()); @@ -125,7 +158,9 @@ function parkResult(): Extract = {}): SessionDeliveryHook { return { + bufferNext: vi.fn(async (): Promise<"buffered" | "closed"> => "buffered"), consumeNext: vi.fn(), + drainReady: vi.fn(async () => {}), next: vi.fn(() => new Promise>(() => {})), rekey: vi.fn(), ...overrides, diff --git a/packages/eve/src/execution/turn-control-receiver.ts b/packages/eve/src/execution/turn-control-receiver.ts index 7c8aef2e2..b85971059 100644 --- a/packages/eve/src/execution/turn-control-receiver.ts +++ b/packages/eve/src/execution/turn-control-receiver.ts @@ -1,6 +1,12 @@ import { createHook, type Hook } from "#compiled/@workflow/core/index.js"; import type { DeliverHookPayload } from "#channel/types.js"; +import { + prependTurnDeliveries, + takeBufferedDelivery, + takePriorityDelivery, + type DeliveryBuffers, +} from "#execution/delivery-buffers.js"; import type { TurnControlPayload } from "#execution/turn-control-protocol.js"; import { forwardTurnDeliveryStep } from "#execution/forward-turn-delivery-step.js"; import { closeHookIterator, disposeHook } from "#execution/hook-ownership.js"; @@ -12,18 +18,19 @@ type DeliveryRequest = Extract; private readonly controlIterator: AsyncIterator; private readonly deliveryHook: SessionDeliveryHook; private pendingControl: Promise> | null = null; + private pendingDelivery: ReturnType | null = null; constructor(input: { - readonly bufferedDeliveries: DeliverHookPayload[]; + readonly deliveryBuffers: DeliveryBuffers; readonly deliveryHook: SessionDeliveryHook; readonly token: string; }) { - this.bufferedDeliveries = input.bufferedDeliveries; + this.deliveryBuffers = input.deliveryBuffers; this.control = createHook({ token: input.token }); this.controlIterator = this.control[Symbol.asyncIterator](); this.deliveryHook = input.deliveryHook; @@ -60,9 +67,7 @@ export class TurnControlReceiver { private bufferTurnDeliveries( payload: Extract, ): void { - if (payload.bufferedDeliveries !== undefined) { - this.bufferedDeliveries.unshift(...payload.bufferedDeliveries); - } + prependTurnDeliveries(this.deliveryBuffers, payload.bufferedDeliveries); } private consumeControl(): void { @@ -74,6 +79,15 @@ export class TurnControlReceiver { return this.pendingControl; } + private consumeDelivery(): void { + this.pendingDelivery = null; + } + + private getDeliveryPromise(): ReturnType { + this.pendingDelivery ??= this.deliveryHook.bufferNext(); + return this.pendingDelivery; + } + private async nextControl( onClosed: string, ): Promise< @@ -105,11 +119,16 @@ export class TurnControlReceiver { ): Promise { await this.deliveryHook.rekey(request.continuationToken); - let delivery = this.bufferedDeliveries.shift(); + let delivery = takePriorityDelivery(this.deliveryBuffers); + if (delivery === undefined && this.deliveryBuffers.replacedHookDeliveries.length > 0) { + await this.deliveryHook.drainReady(); + delivery = takeBufferedDelivery(this.deliveryBuffers); + } + while (delivery === undefined) { const winner = await Promise.race([ this.getControlPromise().then((value) => ({ kind: "control" as const, value })), - this.deliveryHook.next().then((value) => ({ kind: "delivery" as const, value })), + this.getDeliveryPromise().then((value) => ({ kind: "delivery" as const, value })), ]); if (winner.kind === "control") { @@ -132,13 +151,13 @@ export class TurnControlReceiver { continue; } - if (winner.value.done) { + this.consumeDelivery(); + if (winner.value === "closed") { throw new Error("Session delivery hook closed during a turn delivery request."); } - this.deliveryHook.consumeNext(); - if (winner.value.value.kind !== "deliver") continue; - delivery = winner.value.value; + await this.deliveryHook.drainReady(); + delivery = takeBufferedDelivery(this.deliveryBuffers); } // Forwarding is provisional until the turn acknowledges it. If the inbox is @@ -181,12 +200,12 @@ export class TurnControlReceiver { } if (payload.kind === "turn-delivery-cancelled" && payload.requestId === requestId) { - this.bufferedDeliveries.unshift(outstanding); + prependTurnDeliveries(this.deliveryBuffers, [outstanding]); return undefined; } if (payload.kind === "turn-result") { - this.bufferedDeliveries.unshift(outstanding); + prependTurnDeliveries(this.deliveryBuffers, [outstanding]); } const terminal = this.readTerminalControl(payload); diff --git a/packages/eve/src/execution/turn-dispatch.test.ts b/packages/eve/src/execution/turn-dispatch.test.ts index 45c6bea15..a2a9ac824 100644 --- a/packages/eve/src/execution/turn-dispatch.test.ts +++ b/packages/eve/src/execution/turn-dispatch.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 type { DurableSessionState } from "#execution/durable-session-store.js"; import { forwardTurnDeliveryStep } from "#execution/forward-turn-delivery-step.js"; import type { SessionDeliveryHook } from "#execution/session-delivery-hook.js"; @@ -40,8 +41,8 @@ describe("dispatchAndAwaitTurn", () => { const deliveryHook = createDeliveryHook({ rekey: rekeyHook }); await dispatchAndAwaitTurn({ - bufferedDeliveries: [], controlToken: "turn-control", + deliveryBuffers: createDeliveryBuffers(), delivery: { kind: "deliver", payloads: [{ message: "start" }] }, deliveryHook, mode: "conversation", @@ -73,8 +74,8 @@ describe("dispatchAndAwaitTurn", () => { const deliveryHook = createDeliveryHook({ rekey: rekeyHook }); await dispatchAndAwaitTurn({ - bufferedDeliveries: [], controlToken: "turn-control", + deliveryBuffers: createDeliveryBuffers(), delivery: { kind: "deliver", payloads: [{ message: "start" }] }, deliveryHook, mode: "conversation", @@ -127,16 +128,19 @@ describe("dispatchAndAwaitTurn", () => { }), ); - const bufferedDeliveries: DeliverHookPayload[] = []; + const deliveryBuffers = createDeliveryBuffers(); await dispatchAndAwaitTurn({ - bufferedDeliveries, controlToken: "turn-control", + deliveryBuffers, delivery: { kind: "deliver", payloads: [{ message: "start" }] }, deliveryHook: createDeliveryHook({ - next: async () => ({ - done: false, - value: { kind: "deliver", payloads: [{ message: "later delivery" }] }, - }), + bufferNext: async (): Promise<"buffered" | "closed"> => { + deliveryBuffers.currentHookDeliveries.push({ + kind: "deliver", + payloads: [{ message: "later delivery" }], + }); + return "buffered"; + }, }), mode: "conversation", parentWritable: new WritableStream(), @@ -144,7 +148,7 @@ describe("dispatchAndAwaitTurn", () => { sessionState: state, }); - expect(bufferedDeliveries.map((item) => item.payloads[0]?.message)).toEqual([ + expect(deliveryBuffers.turnDeliveries.map((item) => item.payloads[0]?.message)).toEqual([ "earlier remainder", "later delivery", ]); @@ -169,10 +173,11 @@ describe("dispatchAndAwaitTurn", () => { }, ]); - const bufferedDeliveries: DeliverHookPayload[] = [delivery]; + const deliveryBuffers = createDeliveryBuffers(); + deliveryBuffers.turnDeliveries.push(delivery); const action = await dispatchAndAwaitTurn({ - bufferedDeliveries, controlToken: "turn-control", + deliveryBuffers, delivery: { kind: "deliver", payloads: [{ message: "start" }] }, deliveryHook: createDeliveryHook(), mode: "conversation", @@ -183,13 +188,15 @@ describe("dispatchAndAwaitTurn", () => { expect(forwardTurnDeliveryStep).toHaveBeenCalledOnce(); expect(action.kind).toBe("park"); - expect(bufferedDeliveries).toEqual([delivery]); + expect(deliveryBuffers.turnDeliveries).toEqual([delivery]); }); }); function createDeliveryHook(overrides: Partial = {}): SessionDeliveryHook { return { + bufferNext: vi.fn(async (): Promise<"buffered" | "closed"> => "buffered"), consumeNext: vi.fn(), + drainReady: vi.fn(async () => {}), next: vi.fn(() => new Promise>(() => {})), rekey: vi.fn(), ...overrides, diff --git a/packages/eve/src/execution/turn-dispatch.ts b/packages/eve/src/execution/turn-dispatch.ts index b2aa35d59..5adcf8c55 100644 --- a/packages/eve/src/execution/turn-dispatch.ts +++ b/packages/eve/src/execution/turn-dispatch.ts @@ -1,4 +1,5 @@ -import type { DeliverHookPayload, HookPayload, SessionCapabilities } from "#channel/types.js"; +import type { HookPayload, SessionCapabilities } from "#channel/types.js"; +import type { DeliveryBuffers } from "#execution/delivery-buffers.js"; import { TurnControlReceiver } from "#execution/turn-control-receiver.js"; import type { DurableSessionState } from "#execution/durable-session-store.js"; import type { NextDriverAction } from "#execution/next-driver-action.js"; @@ -8,9 +9,9 @@ import type { RunMode } from "#shared/run-mode.js"; /** Dispatches one turn and services its private-inbox control protocol until it terminates. */ export async function dispatchAndAwaitTurn(input: { - readonly bufferedDeliveries: DeliverHookPayload[]; readonly capabilities?: SessionCapabilities; readonly controlToken: string; + readonly deliveryBuffers: DeliveryBuffers; readonly delivery: HookPayload; readonly deliveryHook: SessionDeliveryHook; readonly mode: RunMode; @@ -19,7 +20,7 @@ export async function dispatchAndAwaitTurn(input: { readonly sessionState: DurableSessionState; }): Promise { const control = new TurnControlReceiver({ - bufferedDeliveries: input.bufferedDeliveries, + deliveryBuffers: input.deliveryBuffers, deliveryHook: input.deliveryHook, token: input.controlToken, }); diff --git a/packages/eve/src/execution/workflow-entry.test.ts b/packages/eve/src/execution/workflow-entry.test.ts index 25fbe2518..4b019d0e5 100644 --- a/packages/eve/src/execution/workflow-entry.test.ts +++ b/packages/eve/src/execution/workflow-entry.test.ts @@ -599,6 +599,70 @@ describe("workflowEntry", () => { ); }); + it("dispatches current-hook deliveries before deliveries captured from a replaced hook", async () => { + const baseSessionState = createBaseSessionState({ continuationToken: "slack:C01:" }); + const rekeyedSessionState: DurableSessionState = { + ...baseSessionState, + continuationToken: "slack:C01:1800000000.123456", + }; + + vi.mocked(createSessionStep).mockResolvedValue( + createSessionStepResultForMock(baseSessionState), + ); + + installHookMocks({ + deliveryHooks: [ + { + token: "slack:C01:", + values: [ + { + kind: "deliver", + payloads: [{ message: "trigger rekey" }], + }, + { + kind: "deliver", + payloads: [{ message: "stale old hook" }], + }, + ], + }, + { + token: "slack:C01:1800000000.123456", + values: [ + { + kind: "deliver", + payloads: [{ message: "fresh reopened request" }], + }, + ], + }, + ], + turnControls: [ + turnResult({ action: "park", sessionState: baseSessionState }), + turnResult({ action: "park", sessionState: rekeyedSessionState }), + turnResult({ action: "done", output: "ok", sessionState: rekeyedSessionState }), + ], + }); + + const result = await workflowEntry({ + input: { message: "hello" }, + serializedContext: createSerializedContext({ + "eve.channel": { kind: "slack", state: {} }, + "eve.continuationToken": "slack:C01:", + }), + }); + + expect(result).toEqual({ output: "ok" }); + expect(vi.mocked(dispatchTurnStep).mock.calls[1]?.[0].delivery).toEqual({ + kind: "deliver", + payloads: [{ message: "trigger rekey" }], + }); + expect(vi.mocked(dispatchTurnStep).mock.calls[2]?.[0].delivery).toEqual({ + auth: undefined, + kind: "deliver", + payloads: [{ message: "fresh reopened request" }], + requestId: undefined, + }); + }); + it("disposes the workflow hook after the loop exits", async () => { const sessionState = createBaseSessionState(); vi.mocked(createSessionStep).mockResolvedValue(createSessionStepResultForMock(sessionState)); @@ -742,27 +806,40 @@ function createMockHook(input: { const getConflict = input.getConflict ?? vi.fn(async () => null); const symbolDispose = input.symbolDispose ?? vi.fn(); const iteratorReturn = input.return; + const next = + input.next ?? + async function next(): Promise> { + const value = values.shift(); + if (value === undefined) { + return { done: true, value: undefined }; + } + return { done: false, value }; + }; - return Object.assign(Promise.resolve(undefined), { + return { [Symbol.asyncIterator]() { return { - next: - input.next ?? - async function next(): Promise> { - const value = values.shift(); - if (value === undefined) { - return { done: true, value: undefined }; - } - return { done: false, value }; - }, + next, return: iteratorReturn, }; }, [Symbol.dispose]: symbolDispose, dispose, getConflict, + // eslint-disable-next-line unicorn/no-thenable -- a Workflow Hook is itself a thenable + then( + onFulfilled?: ((value: T) => TResult1 | PromiseLike) | null, + onRejected?: ((reason: unknown) => TResult2 | PromiseLike) | null, + ): Promise { + const value = values.shift(); + const settled = + value === undefined + ? Promise.reject(Object.assign(new Error("hook disposed"), { name: "HookNotFoundError" })) + : Promise.resolve(value); + return settled.then(onFulfilled, onRejected); + }, token: input.token, - }); + }; } function createIteratorReturn(): () => Promise> { diff --git a/packages/eve/src/execution/workflow-entry.ts b/packages/eve/src/execution/workflow-entry.ts index 7b10b55e6..f2b644fe2 100644 --- a/packages/eve/src/execution/workflow-entry.ts +++ b/packages/eve/src/execution/workflow-entry.ts @@ -12,6 +12,7 @@ import { readChannelRequestId, readRootSessionId } from "#execution/eve-workflow import type { RunMode } from "#shared/run-mode.js"; import type { RuntimeCompiledArtifactsSource } from "#runtime/compiled-artifacts-source.js"; import { notifyDelegatedParentStep } from "#execution/delegated-parent-notification.js"; +import { createDeliveryBuffers, type DeliveryBuffers } from "#execution/delivery-buffers.js"; import { createDelegatedSubagentErrorResult, createDelegatedSubagentSuccessResult, @@ -164,8 +165,8 @@ async function runDriverLoop(input: { const nextTurnControlToken = (): string => `${input.sessionState.sessionId}:turn-control:${String(turnDispatchIndex++)}`; - const bufferedDeliveries: DeliverHookPayload[] = []; - const deliveryHook = createSessionDeliveryHook(bufferedDeliveries); + const deliveryBuffers = createDeliveryBuffers(); + const deliveryHook = createSessionDeliveryHook(deliveryBuffers); try { if (input.sessionState.continuationToken) { @@ -173,9 +174,9 @@ async function runDriverLoop(input: { } let action: NextDriverAction = await dispatchAndAwaitTurn({ - bufferedDeliveries, capabilities: input.capabilities, controlToken: nextTurnControlToken(), + deliveryBuffers, delivery: input.initialInput, deliveryHook, mode: input.mode, @@ -225,9 +226,9 @@ async function runDriverLoop(input: { } action = await dispatchAndAwaitTurn({ - bufferedDeliveries, capabilities: input.capabilities, controlToken: nextTurnControlToken(), + deliveryBuffers, delivery: { kind: "deliver", payloads: allPayloads, @@ -242,7 +243,7 @@ async function runDriverLoop(input: { } const nextDeliver = await waitForNextDeliver({ - bufferedDeliveries, + deliveryBuffers, deliveryHook, }); @@ -263,9 +264,9 @@ async function runDriverLoop(input: { } action = await dispatchAndAwaitTurn({ - bufferedDeliveries, capabilities: input.capabilities, controlToken: nextTurnControlToken(), + deliveryBuffers, delivery: { auth: nextDeliver.auth, kind: "deliver", @@ -311,54 +312,34 @@ async function finalizeDone(input: { } async function waitForNextDeliver(input: { - readonly bufferedDeliveries: DeliverHookPayload[]; + readonly deliveryBuffers: DeliveryBuffers; readonly deliveryHook: SessionDeliveryHook; }): Promise { - if (input.bufferedDeliveries.length > 0) { - return coalesceDeliveries(input.bufferedDeliveries.splice(0)); - } - while (true) { - const first = await input.deliveryHook.next(); - input.deliveryHook.consumeNext(); - - if (first.done) { - return null; - } - - if (first.value.kind !== "deliver") { - continue; - } - - let coalesced = first.value; - - while (true) { - const ready = await takeReadyPayload(input.deliveryHook.next()); - - if (ready === NO_READY_MESSAGE) { - break; - } + await input.deliveryHook.drainReady(); + const buffered = takeBufferedDeliveryBatch(input.deliveryBuffers); + if (buffered !== undefined) return buffered; - input.deliveryHook.consumeNext(); - - if (ready.done) { - break; - } - - if (ready.value.kind !== "deliver") { - continue; - } + const result = await input.deliveryHook.bufferNext(); + if (result === "closed") return null; + } +} - coalesced = coalesceDeliveries([coalesced, ready.value]); - } +function takeBufferedDeliveryBatch(buffers: DeliveryBuffers): DeliverHookPayload | undefined { + const turnDeliveries = buffers.turnDeliveries.splice(0); + if (turnDeliveries.length > 0) { + return coalesceDeliveries(turnDeliveries); + } - return coalesced; + const currentHookDeliveries = buffers.currentHookDeliveries.splice(0); + if (currentHookDeliveries.length > 0) { + return coalesceDeliveries(currentHookDeliveries); } -} -const NO_READY_MESSAGE = Symbol("no-ready-message"); + const replacedHookDeliveries = buffers.replacedHookDeliveries.splice(0); + if (replacedHookDeliveries.length > 0) { + return coalesceDeliveries(replacedHookDeliveries); + } -async function takeReadyPayload(promise: Promise): Promise { - await Promise.resolve(); - return await Promise.race([promise, Promise.resolve(NO_READY_MESSAGE)]); + return undefined; } diff --git a/packages/eve/src/internal/testing/session-delivery-hook-workflow.ts b/packages/eve/src/internal/testing/session-delivery-hook-workflow.ts index 7808b3990..ec0faea1c 100644 --- a/packages/eve/src/internal/testing/session-delivery-hook-workflow.ts +++ b/packages/eve/src/internal/testing/session-delivery-hook-workflow.ts @@ -1,4 +1,5 @@ import type { DeliverHookPayload, HookPayload } from "#channel/types.js"; +import { createDeliveryBuffers, takeBufferedDelivery } from "#execution/delivery-buffers.js"; import { createSessionDeliveryHook, type SessionDeliveryHook, @@ -10,8 +11,8 @@ export async function sessionDeliveryHookWorkflow(input: { }): Promise { "use workflow"; - const bufferedDeliveries: DeliverHookPayload[] = []; - const deliveryHook = createSessionDeliveryHook(bufferedDeliveries); + const deliveryBuffers = createDeliveryBuffers(); + const deliveryHook = createSessionDeliveryHook(deliveryBuffers); try { await deliveryHook.rekey(input.token); @@ -22,7 +23,7 @@ export async function sessionDeliveryHookWorkflow(input: { collectMessages(await pendingDelivery, deliveryHook, messages); while (messages.length < 2) { - const buffered = bufferedDeliveries.shift(); + const buffered = takeBufferedDelivery(deliveryBuffers); if (buffered !== undefined) { appendMessages(buffered, messages); continue;