feat(eve): sketch deferred channel deliveries

.changeset/deferred-channel-deliveries.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+Added an internal channel delivery decision contract that can defer normalized input until a later delivery dispatches, enabling threaded channels to preserve skipped messages across durable session boundaries.

packages/eve/src/channel/adapter.ts

@@ -100,6 +100,24 @@ export type { FetchFileResult };
 
 export type ChannelInstrumentationMetadata = Readonly<Record<string, unknown>>;
 
+/** Framework-owned control result from a channel's durable delivery hook. */
+export type ChannelDeliveryDecision =
+  | {
+      readonly action: "defer";
+      readonly input: StepInput;
+      readonly reason?: string;
+    }
+  | {
+      readonly action: "drop";
+      readonly reason?: string;
+    }
+  | {
+      readonly action: "dispatch";
+      readonly input: StepInput;
+    };
+
+export type ChannelDeliveryResult = StepInput | ChannelDeliveryDecision | void;
+
 export type ChannelInstrumentationMetadataProjector = (
   state: Record<string, unknown> | undefined,
 ) => ChannelInstrumentationMetadata;
@@ -138,7 +156,10 @@ export type ChannelAdapter<TCtx extends ChannelAdapterContext<any> = ChannelAdap
    * Return a {@link StepInput} to override the input the harness sees, or
    * return void to use the default payload projection.
    */
-  deliver?(payload: DeliverPayload, ctx: TCtx): StepInput | void | Promise<StepInput | void>;
+  deliver?(
+    payload: DeliverPayload,
+    ctx: TCtx,
+  ): ChannelDeliveryResult | Promise<ChannelDeliveryResult>;
 
   /**
    * Optional factory that builds the adapter context for this adapter.

packages/eve/src/context/keys.ts

@@ -19,6 +19,7 @@ import { ContextKey } from "#context/key.js";
 import type { SandboxAccess } from "#sandbox/state.js";
 import type { RunMode } from "#shared/run-mode.js";
 import type { RuntimeModelReference } from "#runtime/agent/bootstrap.js";
+import type { StepInput } from "#harness/types.js";
 
 // Re-export so consumers don't need a direct channel/ import.
 export type { SessionAuthContext, SessionParent, SessionTurn } from "#channel/types.js";
@@ -81,6 +82,11 @@ export const CapabilitiesKey = new ContextKey<SessionCapabilities>("eve.capabili
  */
 export const SessionCallbackKey = new ContextKey<SessionCallback>("eve.sessionCallback");
 
+/** Framework-owned channel inputs deferred until a later delivery dispatches. */
+export const DeferredChannelInputsKey = new ContextKey<readonly StepInput[]>(
+  "eve.deferredChannelInputs",
+);
+
 // ---------------------------------------------------------------------------
 // Derived keys — reconstructed by providers each step, never serialized.
 // ---------------------------------------------------------------------------

packages/eve/src/execution/channel-delivery-decision.test.ts

@@ -0,0 +1,25 @@
+import { describe, expect, it } from "vitest";
+
+import { normalizeChannelDeliveryDecision } from "#execution/channel-delivery-decision.js";
+
+describe("normalizeChannelDeliveryDecision", () => {
+  it("preserves explicit decisions", () => {
+    const decision = {
+      action: "defer" as const,
+      input: { message: "later" },
+      reason: "classifier",
+    };
+    expect(normalizeChannelDeliveryDecision(decision)).toBe(decision);
+  });
+
+  it("normalizes StepInput as dispatch", () => {
+    expect(normalizeChannelDeliveryDecision({ message: "now" })).toEqual({
+      action: "dispatch",
+      input: { message: "now" },
+    });
+  });
+
+  it("normalizes void as drop", () => {
+    expect(normalizeChannelDeliveryDecision(undefined)).toEqual({ action: "drop" });
+  });
+});

packages/eve/src/execution/channel-delivery-decision.ts

@@ -0,0 +1,23 @@
+import type { ChannelDeliveryDecision, ChannelDeliveryResult } from "#channel/adapter.js";
+import type { StepInput } from "#harness/types.js";
+
+/** Internal normalized result from one channel adapter delivery. */
+export type NormalizedChannelDeliveryDecision =
+  | { readonly action: "defer"; readonly input: StepInput; readonly reason?: string }
+  | { readonly action: "drop"; readonly reason?: string }
+  | { readonly action: "dispatch"; readonly input: StepInput };
+
+/** Normalizes the legacy StepInput/void adapter contract into explicit control actions. */
+export function normalizeChannelDeliveryDecision(
+  result: ChannelDeliveryResult,
+): NormalizedChannelDeliveryDecision {
+  if (result === undefined || result === null) return { action: "drop" };
+  if (isChannelDeliveryDecision(result)) return result;
+  return { action: "dispatch", input: result };
+}
+
+function isChannelDeliveryDecision(value: unknown): value is ChannelDeliveryDecision {
+  if (typeof value !== "object" || value === null || !("action" in value)) return false;
+  const action = (value as { readonly action?: unknown }).action;
+  return action === "defer" || action === "dispatch" || action === "drop";
+}

packages/eve/src/execution/workflow-steps.test.ts

@@ -19,7 +19,7 @@ import {
   setPendingRuntimeActionBatch,
 } from "#harness/runtime-actions.js";
 import { getPendingAuthorization, setPendingAuthorization } from "#harness/authorization.js";
-import type { HarnessSession, StepResult } from "#harness/types.js";
+import type { HarnessSession, StepInput, StepResult } from "#harness/types.js";
 import { createEmptyHookRegistry } from "#runtime/hooks/registry.js";
 import { getCompiledRuntimeAgentBundle } from "#runtime/sessions/compiled-agent-cache.js";
 import {
@@ -151,6 +151,22 @@ const threadContextAdapter: ChannelAdapter = {
   },
 };
 
+const deferredDeliveryAdapter: ChannelAdapter = {
+  kind: "deferred-delivery",
+  deliver(payload: DeliverPayload) {
+    const message = String(payload.message ?? "");
+    if (message.startsWith("defer:")) {
+      return {
+        action: "defer",
+        input: { context: [message.slice(6)] },
+        reason: "test-defer",
+      };
+    }
+    if (message === "drop") return { action: "drop", reason: "test-drop" };
+    return { action: "dispatch", input: { message } };
+  },
+};
+
 function createStubSession(overrides: Partial<HarnessSession> = {}): HarnessSession {
   return {
     agent: { modelReference: { id: "test" }, system: "", tools: [] },
@@ -162,12 +178,14 @@ function createStubSession(overrides: Partial<HarnessSession> = {}): HarnessSess
   };
 }
 
-function createSerializedContext(): Record<string, unknown> {
+function createSerializedContext(
+  adapter: ChannelAdapter = threadContextAdapter,
+): Record<string, unknown> {
   const ctx = new ContextContainer();
   ctx.set(AuthKey, null);
   ctx.set(BundleKey, {
     adapterRegistry: {
-      adaptersByKind: new Map([[threadContextAdapter.kind, threadContextAdapter]]),
+      adaptersByKind: new Map([[adapter.kind, adapter]]),
     },
     compiledArtifactsSource: {} as never,
     graph: {
@@ -183,7 +201,7 @@ function createSerializedContext(): Record<string, unknown> {
     toolRegistry: {},
     turnAgent: TestTurnAgent,
   } as never);
-  ctx.set(ChannelKey, threadContextAdapter);
+  ctx.set(ChannelKey, adapter);
   ctx.set(ContinuationTokenKey, "http:thread-context");
   ctx.set(ModeKey, "conversation");
   ctx.set(SessionIdKey, "session-1");
@@ -778,6 +796,82 @@ describe("turnStep", () => {
     });
   });
 
+  it("persists deferred input across steps, leaves it on drop, then forwards and clears it", async () => {
+    const seen: StepInput[] = [];
+    const session = createStubSession();
+    installSessionStoreMocks([session]);
+    vi.mocked(getCompiledRuntimeAgentBundle).mockResolvedValue({
+      adapterRegistry: {
+        adaptersByKind: new Map([[deferredDeliveryAdapter.kind, deferredDeliveryAdapter]]),
+      },
+      compiledArtifactsSource: {} as never,
+      graph: {
+        nodesByNodeId: new Map(),
+        root: { sandboxRegistry: { sandbox: null }, turnAgent: TestTurnAgent },
+      },
+      moduleMap: { nodes: {} },
+      hookRegistry: createEmptyHookRegistry(),
+      resolvedAgent: { config: {} },
+      subagentRegistry: {},
+      toolRegistry: {},
+      turnAgent: TestTurnAgent,
+    } as never);
+    vi.mocked(createExecutionNodeStep).mockImplementation(() => {
+      return async (session, input): Promise<StepResult> => {
+        seen.push(input ?? {});
+        return { next: null, session };
+      };
+    });
+
+    const info = vi.spyOn(console, "log").mockImplementation(() => {});
+    const first = await turnStep({
+      input: {
+        kind: "deliver",
+        payloads: Array.from({ length: 13 }, (_, index) => ({ message: `defer:${index}` })),
+      },
+      parentWritable: createTestWritable(),
+      serializedContext: createSerializedContext(deferredDeliveryAdapter),
+      sessionState: createStubSessionState(),
+    });
+    expect(first.action).toBe("park");
+    expect(seen).toHaveLength(0);
+    expect(first.serializedContext["eve.deferredChannelInputs"]).toEqual(
+      Array.from({ length: 12 }, (_, index) => ({ context: [String(index + 1)] })),
+    );
+    expect(info).toHaveBeenCalledWith(
+      "[eve:execution.workflow-steps] channel delivery deferred without starting a turn",
+      expect.objectContaining({ bufferedInputs: 12, droppedInputs: 1 }),
+    );
+
+    const second = await turnStep({
+      input: { kind: "deliver", payloads: [{ message: "drop" }] },
+      parentWritable: createTestWritable(),
+      serializedContext: first.serializedContext,
+      sessionState: first.sessionState,
+    });
+    expect(second.action).toBe("park");
+    expect(seen).toHaveLength(0);
+
+    const third = await turnStep({
+      input: { kind: "deliver", payloads: [{ message: "now" }] },
+      parentWritable: createTestWritable(),
+      serializedContext: second.serializedContext,
+      sessionState: second.sessionState,
+    });
+    expect(third.action).toBe("park");
+    expect(seen).toEqual([
+      {
+        context: Array.from({ length: 12 }, (_, index) => String(index + 1)),
+        message: "now",
+      },
+    ]);
+    expect(third.serializedContext["eve.deferredChannelInputs"]).toEqual([]);
+    expect(info).toHaveBeenCalledWith(
+      "[eve:execution.workflow-steps] channel delivery dispatching with deferred inputs",
+      expect.objectContaining({ forwardedInputs: 12 }),
+    );
+  });
+
   it("persists onDeliver context into the next durable step", async () => {
     const seenMessages: string[] = [];
     const session = createStubSession();

packages/eve/src/execution/workflow-steps.ts

@@ -12,6 +12,7 @@ import {
 import {
   AuthKey,
   CapabilitiesKey,
+  DeferredChannelInputsKey,
   ModeKey,
   SessionDynamicToolRuntimeRevisionKey,
 } from "#context/keys.js";
@@ -25,6 +26,7 @@ import {
   throwIfTurnAborted,
 } from "#harness/turn-cancellation.js";
 import { setChannelContext } from "#execution/channel-context.js";
+import { normalizeChannelDeliveryDecision } from "#execution/channel-delivery-decision.js";
 import { hasPendingInputBatch } from "#harness/input-requests.js";
 import { coalesceTurnInputs } from "#harness/messages.js";
 import {
@@ -73,6 +75,7 @@ import { reconcileSessionContinuationToken } from "#execution/reconcile-session-
 import { hydrateDurableSession, refreshSessionFromTurnAgent } from "#execution/session.js";
 import { buildTurnAttributes, readRootSessionId } from "#execution/eve-workflow-attributes.js";
 import { normalizeEveAttributes } from "#runtime/attributes/normalize.js";
+import { createLogger } from "#internal/logging.js";
 import { resolveRuntimeCompiledArtifactsVersionedCacheKey } from "#runtime/cache-key.js";
 import {
   createWorkflowRuntime,
@@ -127,6 +130,9 @@ export type DurableStepResult =
 
 export type { TurnStepInput };
 
+const log = createLogger("execution.workflow-steps");
+const MAX_DEFERRED_CHANNEL_INPUTS = 12;
+
 /**
  * Runs one atomic harness step inside a durable `"use step"` boundary.
  */
@@ -230,15 +236,45 @@ export async function turnStep(rawInput: TurnStepInput): Promise<DurableStepResu
     const beforeDeliverySession = initialSession;
     const scoped = await withContextScope(ctx, initialSession, async (enrichedSession) => {
       const results: StepInput[] = [];
+      let deferred = [...(ctx.get(DeferredChannelInputsKey) ?? [])];
       for (const payload of delivery.payloads) {
         const result = adapter.deliver
           ? await adapter.deliver(payload, adapterCtx)
           : defaultDeliverResult(payload);
-
-        if (result !== undefined && result !== null) {
-          results.push(result);
+        const decision = normalizeChannelDeliveryDecision(result);
+
+        if (decision.action === "defer") {
+          deferred.push(decision.input);
+          const dropped = Math.max(0, deferred.length - MAX_DEFERRED_CHANNEL_INPUTS);
+          if (dropped > 0) deferred = deferred.slice(-MAX_DEFERRED_CHANNEL_INPUTS);
+          log.info("channel delivery deferred without starting a turn", {
+            adapterKind: adapter.kind,
+            bufferedInputs: deferred.length,
+            droppedInputs: dropped,
+            reason: decision.reason,
+            sessionId: initialSession.sessionId,
+          });
+          continue;
         }
+        if (decision.action === "drop") {
+          log.debug("channel delivery dropped without starting a turn", {
+            adapterKind: adapter.kind,
+            bufferedInputs: deferred.length,
+            reason: decision.reason,
+            sessionId: initialSession.sessionId,
+          });
+          continue;
+        }
+
+        log.info("channel delivery dispatching with deferred inputs", {
+          adapterKind: adapter.kind,
+          forwardedInputs: deferred.length,
+          sessionId: initialSession.sessionId,
+        });
+        results.push(...deferred, decision.input);
+        deferred = [];
       }
+      ctx.set(DeferredChannelInputsKey, deferred);
       return {
         result: results.length === 0 ? undefined : results.reduce(coalesceTurnInputs),
         session: enrichedSession,

packages/eve/src/public/channels/slack/slackChannel.test.ts

@@ -1603,7 +1603,11 @@ describe("slackChannel() onMessage", () => {
       ),
     );
 
-    expect(result).toBeUndefined();
+    expect(result).toEqual({
+      action: "defer",
+      input: expect.objectContaining({ context: ["attributed message"], message: undefined }),
+      reason: "human-handoff",
+    });
     expect(adapterCtx.state.threadReplyState).toBe("ignoringThread");
     expect(log).toHaveBeenCalledWith(
       "[eve:slack.channel] Slack thread message skipped by thread reply policy",
@@ -1657,7 +1661,11 @@ describe("slackChannel() onMessage", () => {
       ),
     );
 
-    expect(result).toBeUndefined();
+    expect(result).toEqual({
+      action: "defer",
+      input: expect.objectContaining({ context: ["attributed message"], message: undefined }),
+      reason: "thread-reply-policy",
+    });
     expect(adapterCtx.state.threadReplyState).toBeUndefined();
     expect(onReply).toHaveBeenCalledWith(
       message,
@@ -1702,7 +1710,11 @@ describe("slackChannel() onMessage", () => {
       ),
     );
 
-    expect(result).toBeUndefined();
+    expect(result).toEqual({
+      action: "defer",
+      input: expect.objectContaining({ context: ["attributed message"], message: undefined }),
+      reason: "thread-reply-policy",
+    });
     expect(adapterCtx.state.threadReplyState).toBe("followingThread");
   });
 
@@ -1818,7 +1830,10 @@ describe("slackChannel() onMessage", () => {
       ),
     );
 
-    expect(result).toEqual({ message: "attributed message" });
+    expect(result).toEqual({
+      action: "dispatch",
+      input: expect.objectContaining({ message: "attributed message" }),
+    });
   });
 
   it("passes messages from other bots to onMessage", async () => {