From 8be511985539ac2070b090cfca4113391c7c3525 Mon Sep 17 00:00:00 2001 From: Rui Conti Date: Fri, 24 Jul 2026 20:41:33 -0400 Subject: [PATCH] feat(eve): remote session-callback terminal envelope (#1170) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Session callbacks now carry an event envelope discriminated by event.status. The terminal lane (status: "termination", session.completed/session.failed) is the durable control path: the callee posts one terminal callback, the caller's callback route projects it and resumes the parked parent turn exactly once, never dropped. "notification", "working", and "input_required" are reserved on the wire vocabulary and rejected by the terminal route — the notification relay lane lands separately (it has different delivery semantics: fire-once, drop-ok, rendering-only). Also drops a model-supplied empty outputSchema in remote dispatch, which otherwise forced the callee into structured-output completion and failed the call. parseSessionCallback is renamed to parseCallbackMetadata to distinguish the create-session callback metadata from the new callback events. The callback wire shape changed: caller and callee must both run this version. Signed-off-by: Rui Conti --- .changeset/remote-empty-output-schema.md | 5 + .changeset/remote-terminal-envelope.md | 5 + docs/guides/remote-agents.md | 6 + .../eve/src/channel/session-callback.test.ts | 24 ++-- packages/eve/src/channel/session-callback.ts | 118 +++++++++++++++++- ...ti-agent-callback-routing.scenario.test.ts | 6 +- .../execution/remote-agent-dispatch.test.ts | 66 ++++++++++ .../src/execution/remote-agent-dispatch.ts | 13 +- .../src/execution/session-callback-post.ts | 33 +++++ .../execution/session-callback-step.test.ts | 44 ++++--- .../src/execution/session-callback-step.ts | 87 +++++++------ packages/eve/src/public/channels/eve.ts | 4 +- .../runtime/session-callback-route.test.ts | 112 +++++++++++++---- .../eve/src/runtime/session-callback-route.ts | 108 +++++++++------- 14 files changed, 484 insertions(+), 147 deletions(-) create mode 100644 .changeset/remote-empty-output-schema.md create mode 100644 .changeset/remote-terminal-envelope.md create mode 100644 packages/eve/src/execution/session-callback-post.ts diff --git a/.changeset/remote-empty-output-schema.md b/.changeset/remote-empty-output-schema.md new file mode 100644 index 000000000..4e390b7a3 --- /dev/null +++ b/.changeset/remote-empty-output-schema.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Remote-agent dispatch now drops a model-supplied empty `outputSchema: {}` instead of forwarding it, matching local delegation. An empty object previously forced the remote session into structured-output task completion, which failed the call with `OUTPUT_SCHEMA_NOT_FULFILLED` even though the callee produced a correct prose result. diff --git a/.changeset/remote-terminal-envelope.md b/.changeset/remote-terminal-envelope.md new file mode 100644 index 000000000..3ed5a7a4e --- /dev/null +++ b/.changeset/remote-terminal-envelope.md @@ -0,0 +1,5 @@ +--- +"eve": minor +--- + +Remote-agent session callbacks now carry an event envelope discriminated by `event.status`. The terminal callback (`status: "termination"`, `session.completed`/`session.failed`) resumes the parked parent turn as before; `"notification"`, `"working"`, and `"input_required"` are reserved on the wire and rejected by the terminal route for now. The callback wire shape changed: caller and callee deployments must both run this version. diff --git a/docs/guides/remote-agents.md b/docs/guides/remote-agents.md index 2d8e103c5..19d315fda 100644 --- a/docs/guides/remote-agents.md +++ b/docs/guides/remote-agents.md @@ -70,6 +70,12 @@ A local subagent runs inline. A remote one runs in its own deployment, so dispat The parent stream carries the same `subagent.called`, `action.result`, and `subagent.completed` events as local delegation. For a remote call, `subagent.called.data.remote.url` records the target. +Each callback POST carries one event, classed by `event.status`: + +- `"termination"` — the remote completed or failed. Exactly one termination event ends every remote call; this is the callback that resumes the parked parent with the result. It is the durable control lane: delivery resumes the parent turn exactly once and is never dropped. + +`"notification"` (informational events the parent surfaces without resolving the call), `"working"`, and `"input_required"` are reserved statuses on the wire vocabulary; the terminal callback route rejects them today. The `"notification"` relay lane is added separately. + Cancelling the parent while a remote call is active sends an authenticated `POST /eve/v1/session/:childSessionId/cancel` to the remote and waits for that request to be accepted before the parent settles. eve resolves the remote's `headers` and `auth` again for every cancellation attempt, so rotating credentials work the same way as they do for session creation. Cancellation always uses the standard eve cancel path on `url`, even when `path` customizes only the create-session endpoint. The remote child reports `turn.cancelled` → `session.waiting` on its own stream; an older or unreachable remote is logged but cannot turn the parent's cancellation into a failure. Both failure paths surface to the parent as a failed tool result, so the caller can explain or recover within the same session. A failed _start_ returns the error inline. A remote that starts and then fails posts a terminal failure callback, which the parent receives as an errored subagent result carrying the remote's error (or `REMOTE_AGENT_FAILED` when none is supplied). Terminal callback delivery runs as a durable step on the underlying workflow engine (see [Execution model & durability](../concepts/execution-model-and-durability)). A failed callback POST is rethrown rather than marking the task complete, so the engine retries it. diff --git a/packages/eve/src/channel/session-callback.test.ts b/packages/eve/src/channel/session-callback.test.ts index e08f98d6f..89583f5fd 100644 --- a/packages/eve/src/channel/session-callback.test.ts +++ b/packages/eve/src/channel/session-callback.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from "vitest"; -import { parseSessionCallback } from "#channel/session-callback.js"; +import { parseCallbackMetadata } from "#channel/session-callback.js"; function createCallback(url: string, token = "tok123"): Record { return { @@ -11,16 +11,16 @@ function createCallback(url: string, token = "tok123"): Record }; } -describe("parseSessionCallback callback-URL token extraction", () => { +describe("parseCallbackMetadata callback-URL token extraction", () => { it("accepts a callback URL mounted at the origin root", () => { expect( - parseSessionCallback(createCallback("https://agent.example.com/eve/v1/callback/tok123")), + parseCallbackMetadata(createCallback("https://agent.example.com/eve/v1/callback/tok123")), ).toMatchObject({ ok: true }); }); it("accepts a callback URL mounted behind a public route prefix", () => { expect( - parseSessionCallback( + parseCallbackMetadata( createCallback("https://agent.example.com/eve/agents/support/eve/v1/callback/tok123"), ), ).toMatchObject({ ok: true }); @@ -28,7 +28,7 @@ describe("parseSessionCallback callback-URL token extraction", () => { it("reads the token from the last callback route segment", () => { expect( - parseSessionCallback( + parseCallbackMetadata( createCallback("https://agent.example.com/eve/v1/callback/x/eve/v1/callback/tok123"), ), ).toMatchObject({ ok: true }); @@ -36,7 +36,7 @@ describe("parseSessionCallback callback-URL token extraction", () => { it("decodes the encoded token segment before comparing", () => { expect( - parseSessionCallback( + parseCallbackMetadata( createCallback( "https://agent.example.com/eve/agents/support/eve/v1/callback/eve%3Aparent-token", "eve:parent-token", @@ -47,7 +47,7 @@ describe("parseSessionCallback callback-URL token extraction", () => { it("rejects a URL without the callback route", () => { expect( - parseSessionCallback( + parseCallbackMetadata( createCallback("https://agent.example.com/eve/agents/support/eve/v1/session"), ), ).toMatchObject({ @@ -58,7 +58,7 @@ describe("parseSessionCallback callback-URL token extraction", () => { it("rejects a URL whose token segment does not match the token field", () => { expect( - parseSessionCallback( + parseCallbackMetadata( createCallback("https://agent.example.com/eve/agents/support/eve/v1/callback/other-token"), ), ).toMatchObject({ ok: false }); @@ -66,19 +66,21 @@ describe("parseSessionCallback callback-URL token extraction", () => { it("rejects an empty token segment", () => { expect( - parseSessionCallback(createCallback("https://agent.example.com/eve/v1/callback/")), + parseCallbackMetadata(createCallback("https://agent.example.com/eve/v1/callback/")), ).toMatchObject({ ok: false }); }); it("rejects extra path segments after the token", () => { expect( - parseSessionCallback(createCallback("https://agent.example.com/eve/v1/callback/tok123/more")), + parseCallbackMetadata( + createCallback("https://agent.example.com/eve/v1/callback/tok123/more"), + ), ).toMatchObject({ ok: false }); }); it("rejects a token segment with invalid percent-encoding", () => { expect( - parseSessionCallback(createCallback("https://agent.example.com/eve/v1/callback/%E0%A4%A")), + parseCallbackMetadata(createCallback("https://agent.example.com/eve/v1/callback/%E0%A4%A")), ).toMatchObject({ ok: false }); }); }); diff --git a/packages/eve/src/channel/session-callback.ts b/packages/eve/src/channel/session-callback.ts index 27c4430ae..78b67f99c 100644 --- a/packages/eve/src/channel/session-callback.ts +++ b/packages/eve/src/channel/session-callback.ts @@ -1,8 +1,122 @@ import { z } from "#compiled/zod/index.js"; -import type { SessionCallback } from "#channel/types.js"; +import type { SessionCallback, SubagentAuthorizationEvent } from "#channel/types.js"; import { createEveCallbackRoutePath } from "#protocol/routes.js"; +import type { JsonValue } from "#shared/json.js"; import { isReservedIpAddress } from "#shared/network-address.js"; +import type { TokenUsage } from "#shared/token-usage.js"; + +/** + * Status classes a session callback event can carry. + * + * The callee POSTs one callback per event to the caller's callback URL; + * `status` is the coarse class the caller routes on. `"working"` + * (progress) and `"input_required"` (proxied HITL) are declared so the + * vocabulary is stable, but are not yet emitted or accepted. + */ +export type SessionCallbackEventStatus = + | "input_required" + | "notification" + | "termination" + | "working"; + +/** + * Terminal callback event: the callee session completed or failed. + * Exactly one termination event ends every callback stream. + * + * `usage` — the callee session's token totals — rides along on completed + * events so the caller can attribute the callee's spend. Failed events + * never carry usage. + */ +export type SessionCallbackTerminationEvent = + | { + readonly kind: "session.completed"; + readonly output: JsonValue; + readonly status: "termination"; + readonly usage?: TokenUsage; + } + | { + readonly error: JsonValue; + readonly kind: "session.failed"; + readonly status: "termination"; + }; + +/** + * Informational callback event: a stream event the caller should surface + * without resolving or blocking the pending call. Carries the callee's + * authorization lifecycle events — the callback-URL analog of the local + * subagent adapter's `subagent-authorization-event` forwarding. + */ +export type SessionCallbackNotificationEvent = SubagentAuthorizationEvent & { + readonly status: "notification"; +}; + +/** + * One event on the session callback wire contract, discriminated by + * {@link SessionCallbackEventStatus}. + */ +export type SessionCallbackEvent = + | SessionCallbackNotificationEvent + | SessionCallbackTerminationEvent; + +/** + * Body of one callback POST from a callee session to its caller's + * callback URL. `callId` and `subagentName` correlate the event to the + * pending caller tool call; `sessionId` is the callee session. + */ +export interface SessionCallbackPayload { + readonly callId: string; + readonly event: SessionCallbackEvent; + readonly sessionId: string; + readonly subagentName: string; +} + +const authorizationChallengeSchema = z.object({ + displayName: z.string().optional(), + expiresAt: z.string().optional(), + instructions: z.string().optional(), + url: z.string().optional(), + userCode: z.string().optional(), +}); + +const authorizationRequiredNotificationSchema = z.object({ + data: z.object({ + authorization: authorizationChallengeSchema.optional(), + description: z.string(), + name: z.string(), + sequence: z.number(), + stepIndex: z.number(), + turnId: z.string(), + webhookUrl: z.string().optional(), + }), + status: z.literal("notification"), + type: z.literal("authorization.required"), +}); + +const authorizationCompletedNotificationSchema = z.object({ + data: z.object({ + authorization: authorizationChallengeSchema.optional(), + name: z.string(), + outcome: z.enum(["authorized", "declined", "failed", "timed-out"]), + reason: z.string().optional(), + sequence: z.number(), + stepIndex: z.number(), + turnId: z.string(), + }), + status: z.literal("notification"), + type: z.literal("authorization.completed"), +}); + +/** + * Schema for one `status: "notification"` callback event. + * + * The event arrives from a remote callee that may run a different eve + * version and is re-emitted on the caller's stream, so it is validated + * to the exact shapes the caller knows how to re-emit; unknown keys are + * stripped rather than passed through. + */ +export const sessionCallbackNotificationEventSchema: z.ZodType = + z.union([authorizationRequiredNotificationSchema, authorizationCompletedNotificationSchema]); export type SessionCallbackParseResult = | { @@ -56,7 +170,7 @@ const sessionCallbackSchema = z } }); -export function parseSessionCallback(value: unknown): SessionCallbackParseResult { +export function parseCallbackMetadata(value: unknown): SessionCallbackParseResult { const parsed = sessionCallbackSchema.safeParse(value); if (parsed.success) { return { callback: parsed.data, ok: true }; diff --git a/packages/eve/src/execution/multi-agent-callback-routing.scenario.test.ts b/packages/eve/src/execution/multi-agent-callback-routing.scenario.test.ts index 6a6c74e6b..1a3319a6d 100644 --- a/packages/eve/src/execution/multi-agent-callback-routing.scenario.test.ts +++ b/packages/eve/src/execution/multi-agent-callback-routing.scenario.test.ts @@ -6,7 +6,7 @@ import { join } from "node:path"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; -import { parseSessionCallback } from "#channel/session-callback.js"; +import { parseCallbackMetadata } from "#channel/session-callback.js"; import { ContextContainer, contextStorage } from "#context/container.js"; import { SessionCallbackKey, SessionIdKey } from "#context/keys.js"; import { fireSessionCallbackStep } from "#execution/session-callback-step.js"; @@ -35,7 +35,7 @@ import type { HarnessSession } from "#harness/types.js"; * remote-subagent session callback (`startRemoteAgentSession`) and the * connection hook URL (`getHookUrl`); * 3. the remote side validates the callback metadata with the real - * create-session parser (`parseSessionCallback`) and posts the terminal + * create-session parser (`parseCallbackMetadata`) and posts the terminal * result with the real durable step (`fireSessionCallbackStep`); * 4. each parent service only serves its own prefix-stripped `/eve/v1/*` * callback routes — anything else 404s, matching production. @@ -313,7 +313,7 @@ describe("multi-agent callback routing", () => { // Remote-side create-session validation (public/channels/eve.ts): // the prefixed URL must parse, or the remote rejects with HTTP 400. - expect(parseSessionCallback(callback)).toMatchObject({ ok: true }); + expect(parseCallbackMetadata(callback)).toMatchObject({ ok: true }); // Remote-side terminal POST (the step that failed with HTTP 404 in // production) must reach the parent service through the deployment diff --git a/packages/eve/src/execution/remote-agent-dispatch.test.ts b/packages/eve/src/execution/remote-agent-dispatch.test.ts index 5c0a268ca..9813d7fa5 100644 --- a/packages/eve/src/execution/remote-agent-dispatch.test.ts +++ b/packages/eve/src/execution/remote-agent-dispatch.test.ts @@ -108,6 +108,72 @@ describe("startRemoteAgentSession", () => { expect(body.mode).toBe("task"); }); + it("drops a model-supplied empty outputSchema instead of forwarding it", async () => { + const fetchMock = vi.fn().mockResolvedValue( + new Response(JSON.stringify({ ok: true, sessionId: "remote-session" }), { + headers: { "x-eve-session-id": "remote-session-header" }, + status: 202, + }), + ); + vi.stubGlobal("fetch", fetchMock); + + await startRemoteAgentSession({ + action: { + ...createAction(), + input: { message: "find the marker", outputSchema: {} }, + }, + callbackBaseUrl: "https://caller.example.com", + remote: createRemoteAgent(), + session: { + agent: { modelReference: { id: "mock/test" }, system: "", tools: [] }, + compaction: { recentWindowSize: 10, threshold: 100000 }, + continuationToken: "eve:parent-token", + history: [], + sessionId: "parent-session", + state: {}, + }, + }); + + const body = JSON.parse(fetchMock.mock.calls[0]?.[1]?.body as string); + expect("outputSchema" in body).toBe(false); + }); + + it("falls back to the declared outputSchema when the model sends an empty one", async () => { + const fetchMock = vi.fn().mockResolvedValue( + new Response(JSON.stringify({ ok: true, sessionId: "remote-session" }), { + headers: { "x-eve-session-id": "remote-session-header" }, + status: 202, + }), + ); + vi.stubGlobal("fetch", fetchMock); + + const outputSchema = { + properties: { answer: { type: "string" } }, + required: ["answer"], + type: "object", + } as const; + + await startRemoteAgentSession({ + action: { + ...createAction(), + input: { message: "find the marker", outputSchema: {} }, + }, + callbackBaseUrl: "https://caller.example.com", + remote: { ...createRemoteAgent(), outputSchema }, + session: { + agent: { modelReference: { id: "mock/test" }, system: "", tools: [] }, + compaction: { recentWindowSize: 10, threshold: 100000 }, + continuationToken: "eve:parent-token", + history: [], + sessionId: "parent-session", + state: {}, + }, + }); + + const body = JSON.parse(fetchMock.mock.calls[0]?.[1]?.body as string); + expect(body.outputSchema).toEqual(outputSchema); + }); + it("targets an active turn inbox when a callback token is supplied", async () => { const fetchMock = vi .fn() diff --git a/packages/eve/src/execution/remote-agent-dispatch.ts b/packages/eve/src/execution/remote-agent-dispatch.ts index 160aa36c2..469af7fe8 100644 --- a/packages/eve/src/execution/remote-agent-dispatch.ts +++ b/packages/eve/src/execution/remote-agent-dispatch.ts @@ -1,4 +1,5 @@ import { EVE_SESSION_ID_HEADER } from "#protocol/message.js"; +import { isJsonObjectValue, type JsonValue } from "#shared/json.js"; import { CancelTurnResponseSchema } from "#protocol/cancel-turn.js"; import { createEveCallbackRoutePath, createEveCancelTurnRoutePath } from "#protocol/routes.js"; import type { CancelTurnResult } from "#channel/types.js"; @@ -49,7 +50,7 @@ export async function startRemoteAgentSession(input: { message: formatRemoteAgentCallInputMessage({ action: input.action, remote: input.remote }), mode: "task", outputSchema: - (input.action.input.outputSchema as object | undefined) ?? input.remote.outputSchema, + requestedRemoteOutputSchema(input.action.input.outputSchema) ?? input.remote.outputSchema, }), headers: { "content-type": "application/json", @@ -185,6 +186,16 @@ function formatRemoteAgentCallInputMessage(input: { }).message; } +/** + * A model-supplied per-call schema only counts when it is a non-empty JSON + * object — models routinely emit `outputSchema: {}` for the optional param. + * Mirrors the local-delegation filter in `subagent-tool.ts`, so an empty + * object never forces the remote into structured-output task completion. + */ +function requestedRemoteOutputSchema(value: JsonValue | undefined): object | undefined { + return isJsonObjectValue(value) && Object.keys(value).length > 0 ? value : undefined; +} + function trimTrailingSlash(value: string): string { return value.endsWith("/") ? value.slice(0, -1) : value; } diff --git a/packages/eve/src/execution/session-callback-post.ts b/packages/eve/src/execution/session-callback-post.ts new file mode 100644 index 000000000..4dde903de --- /dev/null +++ b/packages/eve/src/execution/session-callback-post.ts @@ -0,0 +1,33 @@ +import type { SessionCallbackPayload } from "#channel/session-callback.js"; + +const SESSION_CALLBACK_TIMEOUT_MS = 30_000; + +/** + * POSTs one {@link SessionCallbackPayload} to a caller's callback URL. + * + * Shared by the terminal callback step and notification forwarding so + * every callback POST carries the same transport guards. Throws on any + * non-2xx response; callers decide whether delivery is best-effort + * (notifications) or part of the durable result path (termination). + */ +export async function postSessionCallback(input: { + readonly payload: SessionCallbackPayload; + readonly url: string; +}): Promise { + const response = await fetch(input.url, { + body: JSON.stringify(input.payload), + headers: { + "content-type": "application/json", + }, + method: "POST", + // Do not follow redirects: a validated callback host could otherwise + // 3xx-bounce the framework to an internal/metadata address after the + // path/token check has already passed. + redirect: "error", + signal: AbortSignal.timeout(SESSION_CALLBACK_TIMEOUT_MS), + }); + + if (!response.ok) { + throw new Error(`Session callback failed with HTTP ${response.status}.`); + } +} diff --git a/packages/eve/src/execution/session-callback-step.test.ts b/packages/eve/src/execution/session-callback-step.test.ts index 7ed5eb7cd..a2fe2f14f 100644 --- a/packages/eve/src/execution/session-callback-step.test.ts +++ b/packages/eve/src/execution/session-callback-step.test.ts @@ -53,8 +53,11 @@ describe("fireSessionCallbackStep", () => { expect(fetchMock).toHaveBeenCalledWith("https://caller.example.com/eve/v1/callback/tok123", { body: JSON.stringify({ callId: "call-1", - kind: "session.completed", - output: "done", + event: { + kind: "session.completed", + output: "done", + status: "termination", + }, sessionId: "remote-session", subagentName: "research", }), @@ -80,8 +83,11 @@ describe("fireSessionCallbackStep", () => { expect(fetchMock).toHaveBeenCalledWith("https://caller.example.com/eve/v1/callback/tok123", { body: JSON.stringify({ callId: "call-1", - kind: "session.completed", - output: "", + event: { + kind: "session.completed", + output: "", + status: "termination", + }, sessionId: "remote-session", subagentName: "research", }), @@ -108,11 +114,14 @@ describe("fireSessionCallbackStep", () => { expect(fetchMock).toHaveBeenCalledWith("https://caller.example.com/eve/v1/callback/tok123", { body: JSON.stringify({ callId: "call-1", - kind: "session.completed", - output: "done", + event: { + kind: "session.completed", + output: "done", + status: "termination", + usage: USAGE, + }, sessionId: "remote-session", subagentName: "research", - usage: USAGE, }), headers: { "content-type": "application/json", @@ -134,7 +143,7 @@ describe("fireSessionCallbackStep", () => { status: "completed", }); - expect(parsePostedBody(fetchMock).usage).toBeUndefined(); + expect(parsePostedBody(fetchMock).event.usage).toBeUndefined(); expect(errorSpy).not.toHaveBeenCalled(); }); @@ -149,7 +158,7 @@ describe("fireSessionCallbackStep", () => { usage: USAGE, }); - expect(parsePostedBody(fetchMock).usage).toBeUndefined(); + expect(parsePostedBody(fetchMock).event.usage).toBeUndefined(); }); it("posts the failed callback with the normalized error message", async () => { @@ -165,11 +174,14 @@ describe("fireSessionCallbackStep", () => { expect(fetchMock).toHaveBeenCalledWith("https://caller.example.com/eve/v1/callback/tok123", { body: JSON.stringify({ callId: "call-1", - error: { - code: "SESSION_FAILED", - message: "remote exploded", + event: { + error: { + code: "SESSION_FAILED", + message: "remote exploded", + }, + kind: "session.failed", + status: "termination", }, - kind: "session.failed", sessionId: "remote-session", subagentName: "research", }), @@ -298,12 +310,14 @@ describe("fireSessionCallbackStep", () => { }); }); -function parsePostedBody(fetchMock: ReturnType): { usage?: unknown } { +function parsePostedBody(fetchMock: ReturnType): { + event: { usage?: unknown }; +} { const call = fetchMock.mock.calls[0]; if (call === undefined) { throw new Error("expected fetch to have been called"); } - return JSON.parse((call[1] as { body: string }).body) as { usage?: unknown }; + return JSON.parse((call[1] as { body: string }).body) as { event: { usage?: unknown } }; } function createSerializedContext(): Record { diff --git a/packages/eve/src/execution/session-callback-step.ts b/packages/eve/src/execution/session-callback-step.ts index f74b72245..b5143aaa4 100644 --- a/packages/eve/src/execution/session-callback-step.ts +++ b/packages/eve/src/execution/session-callback-step.ts @@ -1,11 +1,16 @@ import type { SessionCallback } from "#channel/types.js"; -import { parseSessionCallback } from "#channel/session-callback.js"; +import type { + SessionCallbackPayload, + SessionCallbackTerminationEvent, +} from "#channel/session-callback.js"; +import { parseCallbackMetadata } from "#channel/session-callback.js"; import { SessionCallbackKey } from "#context/keys.js"; +import { postSessionCallback } from "#execution/session-callback-post.js"; import { createLogger } from "#internal/logging.js"; import { toErrorMessage } from "#shared/errors.js"; +import { parseJsonValue } from "#shared/json.js"; import type { TokenUsage } from "#shared/token-usage.js"; -const SESSION_CALLBACK_TIMEOUT_MS = 30_000; const log = createLogger("execution.session-callback"); /** @@ -39,41 +44,10 @@ export async function fireSessionCallbackStep(input: { try { const callback = parseSerializedSessionCallback(value); - const body = - input.status === "completed" - ? buildCompletedCallbackBody({ - callback, - output: input.output, - sessionId, - usage: input.usage, - }) - : { - callId: callback.callId, - error: { - code: "SESSION_FAILED", - message: toErrorMessage(input.error), - }, - kind: "session.failed" as const, - sessionId, - subagentName: callback.subagentName, - }; - - const response = await fetch(callback.url, { - body: JSON.stringify(body), - headers: { - "content-type": "application/json", - }, - method: "POST", - // Do not follow redirects: a validated callback host could otherwise - // 3xx-bounce the framework to an internal/metadata address after the - // path/token check has already passed. - redirect: "error", - signal: AbortSignal.timeout(SESSION_CALLBACK_TIMEOUT_MS), + await postSessionCallback({ + payload: toPayload({ callback, input, sessionId }), + url: callback.url, }); - - if (!response.ok) { - throw new Error(`Session callback failed with HTTP ${response.status}.`); - } } catch (error) { log.error("failed to post session callback", { error, @@ -83,24 +57,49 @@ export async function fireSessionCallbackStep(input: { } } -function buildCompletedCallbackBody(input: { +function toPayload(args: { readonly callback: SessionCallback; - readonly output: unknown; + readonly input: { + readonly error?: unknown; + readonly output?: unknown; + readonly status: "completed" | "failed"; + readonly usage?: TokenUsage; + }; readonly sessionId: string; +}): SessionCallbackPayload { + const { callback, input, sessionId } = args; + return { + callId: callback.callId, + event: + input.status === "completed" + ? buildCompletedTerminationEvent({ output: input.output, usage: input.usage }) + : { + error: { + code: "SESSION_FAILED", + message: toErrorMessage(input.error), + }, + kind: "session.failed", + status: "termination", + }, + sessionId, + subagentName: callback.subagentName, + }; +} + +function buildCompletedTerminationEvent(input: { + readonly output: unknown; readonly usage: TokenUsage | undefined; -}): Record { +}): SessionCallbackTerminationEvent { const base = { - callId: input.callback.callId, kind: "session.completed" as const, - output: input.output ?? "", - sessionId: input.sessionId, - subagentName: input.callback.subagentName, + output: parseJsonValue(input.output ?? ""), + status: "termination" as const, }; return input.usage === undefined ? base : { ...base, usage: input.usage }; } function parseSerializedSessionCallback(value: unknown): SessionCallback { - const parsed = parseSessionCallback(value); + const parsed = parseCallbackMetadata(value); if (!parsed.ok) { throw new Error("Serialized session callback is invalid.", { cause: parsed.cause, diff --git a/packages/eve/src/public/channels/eve.ts b/packages/eve/src/public/channels/eve.ts index 2d4825e6a..d7d512c74 100644 --- a/packages/eve/src/public/channels/eve.ts +++ b/packages/eve/src/public/channels/eve.ts @@ -2,7 +2,7 @@ import { type FilePart, type TextPart, type UserContent } from "ai"; import type { CancelTurnResult, SessionAuthContext, SessionCallback } from "#channel/types.js"; import type { CancelTurnResponse } from "#protocol/cancel-turn.js"; -import { parseSessionCallback } from "#channel/session-callback.js"; +import { parseCallbackMetadata } from "#channel/session-callback.js"; import { hasInternalRefScheme } from "#internal/attachments/url-refs.js"; import { createLogger, logError } from "#internal/logging.js"; import { @@ -672,7 +672,7 @@ function parseOutputSchemaField(value: unknown): JsonObject | Response | undefin function parseCallbackField(value: unknown): SessionCallback | Response | undefined { if (value === undefined) return undefined; - const parsed = parseSessionCallback(value); + const parsed = parseCallbackMetadata(value); if (parsed.ok) return parsed.callback; return Response.json({ error: parsed.message, ok: false }, { status: 400 }); diff --git a/packages/eve/src/runtime/session-callback-route.test.ts b/packages/eve/src/runtime/session-callback-route.test.ts index 8c4ac568e..4e22667b2 100644 --- a/packages/eve/src/runtime/session-callback-route.test.ts +++ b/packages/eve/src/runtime/session-callback-route.test.ts @@ -43,8 +43,11 @@ describe("session callback route", () => { new Request("https://app.example.com/eve/v1/callback/tok123", { body: JSON.stringify({ callId: "call-1", - kind: "session.completed", - output: "done", + event: { + kind: "session.completed", + output: "done", + status: "termination", + }, sessionId: "remote-session", subagentName: "research", }), @@ -67,6 +70,62 @@ describe("session callback route", () => { }); }); + it("resumes a failed remote-agent result with the reported error", async () => { + resumeHookMock.mockResolvedValue(undefined); + + const response = await handleSessionCallbackRequest( + new Request("https://app.example.com/eve/v1/callback/tok123", { + body: JSON.stringify({ + callId: "call-1", + event: { + error: { code: "SESSION_FAILED", message: "remote exploded" }, + kind: "session.failed", + status: "termination", + }, + sessionId: "remote-session", + subagentName: "research", + }), + method: "POST", + }), + createRouteContext({ token: "tok123" }), + ); + + expect(response.status).toBe(202); + expect(resumeHookMock).toHaveBeenCalledWith("tok123", { + kind: "runtime-action-result", + results: [ + { + callId: "call-1", + isError: true, + kind: "subagent-result", + output: { code: "SESSION_FAILED", message: "remote exploded" }, + subagentName: "research", + }, + ], + }); + }); + + it.each(["notification", "working", "input_required", "unknown-status"])( + "rejects the non-terminal callback event status %s on the durable lane", + async (status) => { + const response = await handleSessionCallbackRequest( + new Request("https://app.example.com/eve/v1/callback/tok123", { + body: JSON.stringify({ + callId: "call-1", + event: { status }, + sessionId: "remote-session", + subagentName: "research", + }), + method: "POST", + }), + createRouteContext({ token: "tok123" }), + ); + + expect(response.status).toBe(400); + expect(resumeHookMock).not.toHaveBeenCalled(); + }, + ); + it("projects reported usage onto the resumed result", async () => { resumeHookMock.mockResolvedValue(undefined); @@ -74,11 +133,14 @@ describe("session callback route", () => { new Request("https://app.example.com/eve/v1/callback/tok123", { body: JSON.stringify({ callId: "call-1", - kind: "session.completed", - output: "done", + event: { + kind: "session.completed", + output: "done", + status: "termination", + usage: { cacheReadTokens: 10, cacheWriteTokens: 5, inputTokens: 100, outputTokens: 50 }, + }, sessionId: "remote-session", subagentName: "research", - usage: { cacheReadTokens: 10, cacheWriteTokens: 5, inputTokens: 100, outputTokens: 50 }, }), method: "POST", }), @@ -107,17 +169,20 @@ describe("session callback route", () => { new Request("https://app.example.com/eve/v1/callback/tok123", { body: JSON.stringify({ callId: "call-1", - kind: "session.completed", - output: "done", + event: { + kind: "session.completed", + output: "done", + status: "termination", + usage: { + cacheReadTokens: 10, + cacheWriteTokens: 5, + inputTokens: 100, + outputTokens: 50, + reasoningOutputTokens: 7, + }, + }, sessionId: "remote-session", subagentName: "research", - usage: { - cacheReadTokens: 10, - cacheWriteTokens: 5, - inputTokens: 100, - outputTokens: 50, - reasoningOutputTokens: 7, - }, }), method: "POST", }), @@ -143,16 +208,19 @@ describe("session callback route", () => { new Request("https://app.example.com/eve/v1/callback/tok123", { body: JSON.stringify({ callId: "call-1", - kind: "session.completed", - output: "done", + event: { + kind: "session.completed", + output: "done", + status: "termination", + usage: { + cacheReadTokens: 10, + cacheWriteTokens: 5, + inputTokens: "lots", + outputTokens: 50, + }, + }, sessionId: "remote-session", subagentName: "research", - usage: { - cacheReadTokens: 10, - cacheWriteTokens: 5, - inputTokens: "lots", - outputTokens: 50, - }, }), method: "POST", }), diff --git a/packages/eve/src/runtime/session-callback-route.ts b/packages/eve/src/runtime/session-callback-route.ts index 499c88445..9db167c4f 100644 --- a/packages/eve/src/runtime/session-callback-route.ts +++ b/packages/eve/src/runtime/session-callback-route.ts @@ -1,32 +1,19 @@ import { resumeHook } from "#internal/workflow/runtime.js"; import { EVE_CALLBACK_ROUTE_PATTERN } from "#protocol/routes.js"; +import type { + SessionCallbackPayload, + SessionCallbackTerminationEvent, +} from "#channel/session-callback.js"; +import type { HookPayload } from "#channel/types.js"; import type { ChannelMethod, RouteContext } from "#public/definitions/channel.js"; import type { ResolvedChannelDefinition } from "#runtime/types.js"; import type { RuntimeSubagentResultActionResult } from "#runtime/actions/types.js"; -import type { JsonValue } from "#shared/json.js"; import { tokenUsageSchema, type TokenUsage } from "#shared/token-usage.js"; export const HTTP_SESSION_CALLBACK_CHANNEL_NAME_PREFIX = "eve/v1/callback"; const HANDLED_METHODS: readonly ChannelMethod[] = ["POST"]; -type SessionTerminalCallbackPayload = - | { - readonly callId: string; - readonly kind: "session.completed"; - readonly output: string; - readonly sessionId: string; - readonly subagentName: string; - readonly usage?: TokenUsage; - } - | { - readonly callId: string; - readonly error: JsonValue; - readonly kind: "session.failed"; - readonly sessionId: string; - readonly subagentName: string; - }; - export function getSessionCallbackChannelDefinitions(): readonly ResolvedChannelDefinition[] { return HANDLED_METHODS.map((method) => buildCallbackChannelDefinition(method)); } @@ -68,16 +55,13 @@ export async function handleSessionCallbackRequest( return Response.json({ error: "Invalid JSON body.", ok: false }, { status: 400 }); } - const result = projectSessionCallbackResult(body); - if (result instanceof Response) { - return result; + const hookPayload = projectSessionCallbackHookPayload(body); + if (hookPayload instanceof Response) { + return hookPayload; } try { - await resumeHook(token, { - kind: "runtime-action-result", - results: [result], - }); + await resumeHook(token, hookPayload); } catch { return Response.json({ error: "Session callback not pending.", ok: false }, { status: 404 }); } @@ -85,45 +69,75 @@ export async function handleSessionCallbackRequest( return Response.json({ ok: true }, { status: 202 }); } -function projectSessionCallbackResult( - value: unknown, -): RuntimeSubagentResultActionResult | Response { +function projectSessionCallbackHookPayload(value: unknown): HookPayload | Response { if (value === null || typeof value !== "object") { return Response.json({ error: "Expected a JSON object.", ok: false }, { status: 400 }); } - const payload = value as Partial; + const payload = value as Partial; if (typeof payload.callId !== "string" || payload.callId.length === 0) { return Response.json({ error: "Missing callback callId.", ok: false }, { status: 400 }); } if (typeof payload.subagentName !== "string" || payload.subagentName.length === 0) { return Response.json({ error: "Missing callback subagentName.", ok: false }, { status: 400 }); } + const event = payload.event; + if (event === null || typeof event !== "object") { + return Response.json({ error: "Missing callback event.", ok: false }, { status: 400 }); + } - if (payload.kind === "session.completed") { - const base: RuntimeSubagentResultActionResult = { + if (event.status === "termination") { + return resultTermination({ callId: payload.callId, - kind: "subagent-result", - output: payload.output ?? "", + event, subagentName: payload.subagentName, + }); + } + + // "notification" (and the reserved "working"/"input_required") are the + // relay lane, delivered separately from this durable terminal path. + return Response.json({ error: "Unsupported callback event status.", ok: false }, { status: 400 }); +} + +function resultTermination(input: { + readonly callId: string; + readonly event: Partial; + readonly subagentName: string; +}): HookPayload | Response { + const event = input.event; + + if (event.kind === "session.completed") { + const base: RuntimeSubagentResultActionResult = { + callId: input.callId, + kind: "subagent-result", + output: event.output ?? "", + subagentName: input.subagentName, + }; + const usage = parseCallbackUsage((event as { usage?: unknown }).usage); + return { + kind: "runtime-action-result", + results: [usage === undefined ? base : { ...base, usage }], }; - const usage = parseCallbackUsage((payload as { usage?: unknown }).usage); - return usage === undefined ? base : { ...base, usage }; } - if (payload.kind === "session.failed") { + if (event.kind === "session.failed") { return { - callId: payload.callId, - isError: true, - kind: "subagent-result", - output: - payload.error === undefined - ? { - code: "REMOTE_AGENT_FAILED", - message: "Remote agent failed.", - } - : payload.error, - subagentName: payload.subagentName, + kind: "runtime-action-result", + results: [ + { + callId: input.callId, + isError: true, + kind: "subagent-result", + output: + event.error === undefined + ? { + code: "REMOTE_AGENT_FAILED", + message: "Remote agent failed.", + } + : event.error, + subagentName: input.subagentName, + }, + ], }; }