diff --git a/.changeset/deferred-channel-deliveries.md b/.changeset/deferred-channel-deliveries.md new file mode 100644 index 000000000..c5cd5281a --- /dev/null +++ b/.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. diff --git a/packages/eve/src/channel/adapter.ts b/packages/eve/src/channel/adapter.ts index e74219cfa..06274b229 100644 --- a/packages/eve/src/channel/adapter.ts +++ b/packages/eve/src/channel/adapter.ts @@ -100,6 +100,24 @@ export type { FetchFileResult }; export type ChannelInstrumentationMetadata = Readonly>; +/** 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 | undefined, ) => ChannelInstrumentationMetadata; @@ -138,7 +156,10 @@ export type ChannelAdapter = 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; + deliver?( + payload: DeliverPayload, + ctx: TCtx, + ): ChannelDeliveryResult | Promise; /** * Optional factory that builds the adapter context for this adapter. diff --git a/packages/eve/src/context/keys.ts b/packages/eve/src/context/keys.ts index a198e1db6..cd756d21c 100644 --- a/packages/eve/src/context/keys.ts +++ b/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("eve.capabili */ export const SessionCallbackKey = new ContextKey("eve.sessionCallback"); +/** Framework-owned channel inputs deferred until a later delivery dispatches. */ +export const DeferredChannelInputsKey = new ContextKey( + "eve.deferredChannelInputs", +); + // --------------------------------------------------------------------------- // Derived keys — reconstructed by providers each step, never serialized. // --------------------------------------------------------------------------- diff --git a/packages/eve/src/execution/channel-delivery-decision.test.ts b/packages/eve/src/execution/channel-delivery-decision.test.ts new file mode 100644 index 000000000..3d216f58a --- /dev/null +++ b/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" }); + }); +}); diff --git a/packages/eve/src/execution/channel-delivery-decision.ts b/packages/eve/src/execution/channel-delivery-decision.ts new file mode 100644 index 000000000..93cd61113 --- /dev/null +++ b/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"; +} diff --git a/packages/eve/src/execution/workflow-steps.test.ts b/packages/eve/src/execution/workflow-steps.test.ts index ed804a817..9d9117041 100644 --- a/packages/eve/src/execution/workflow-steps.test.ts +++ b/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 { return { agent: { modelReference: { id: "test" }, system: "", tools: [] }, @@ -162,12 +178,14 @@ function createStubSession(overrides: Partial = {}): HarnessSess }; } -function createSerializedContext(): Record { +function createSerializedContext( + adapter: ChannelAdapter = threadContextAdapter, +): Record { 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 { 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 => { + 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(); diff --git a/packages/eve/src/execution/workflow-steps.ts b/packages/eve/src/execution/workflow-steps.ts index 2930ecb84..cadd58211 100644 --- a/packages/eve/src/execution/workflow-steps.ts +++ b/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 { 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, diff --git a/packages/eve/src/public/channels/slack/slackChannel.test.ts b/packages/eve/src/public/channels/slack/slackChannel.test.ts index a2e7f07d7..e8e000464 100644 --- a/packages/eve/src/public/channels/slack/slackChannel.test.ts +++ b/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 () => { diff --git a/packages/eve/src/public/channels/slack/slackChannel.ts b/packages/eve/src/public/channels/slack/slackChannel.ts index 247916bed..aff0965fc 100644 --- a/packages/eve/src/public/channels/slack/slackChannel.ts +++ b/packages/eve/src/public/channels/slack/slackChannel.ts @@ -835,6 +835,8 @@ export function slackChannel( ); } if (stateful && "state" in reply) adapterCtx.state.threadReplyState = reply.state; + const input = defaultDeliverResult(payload); + if (input === undefined) return undefined; if (!reply.respond) { log.info("Slack thread message skipped by thread reply policy", { channelId: delivery.message.channelId, @@ -844,7 +846,24 @@ export function slackChannel( threadTs: delivery.message.threadTs, }); } - return reply.respond ? defaultDeliverResult(payload) : undefined; + if (reply.respond) return { action: "dispatch", input }; + + // Slack's fetched thread history is model context preceding the current + // message. Store skipped attributed text the same way so a later + // dispatch preserves the triggering message as the turn input. + const deferredInput = + typeof input.message === "string" + ? { + ...input, + context: [...(input.context ?? []), input.message], + message: undefined, + } + : input; + return { + action: "defer", + input: deferredInput, + reason: reply.reason ?? "thread-reply-policy", + }; }, fetchFile: slackFetchFile, metadata(state): SlackInstrumentationMetadata {