From 4422c39895d60ec7cf5d60fb923a1ba42564bdd6 Mon Sep 17 00:00:00 2001 From: iroiro147 Date: Wed, 8 Jul 2026 23:14:58 +0530 Subject: [PATCH 1/2] fix(eve): route preview workflows to latest Signed-off-by: iroiro147 --- .changeset/preview-latest-routing.md | 5 + .../execution-model-and-durability.mdx | 2 +- .../src/execution/workflow-runtime.test.ts | 96 +++- .../eve/src/execution/workflow-runtime.ts | 202 +++------ .../eve/src/execution/workflow-steps.test.ts | 410 +++--------------- 5 files changed, 221 insertions(+), 494 deletions(-) create mode 100644 .changeset/preview-latest-routing.md diff --git a/.changeset/preview-latest-routing.md b/.changeset/preview-latest-routing.md new file mode 100644 index 000000000..a61de8281 --- /dev/null +++ b/.changeset/preview-latest-routing.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Try latest-deployment routing for Vercel preview workflow dispatches and fall back when the active workflow world cannot resolve `latest`. diff --git a/docs/concepts/execution-model-and-durability.mdx b/docs/concepts/execution-model-and-durability.mdx index 7b2e28f8b..c39f80649 100644 --- a/docs/concepts/execution-model-and-durability.mdx +++ b/docs/concepts/execution-model-and-durability.mdx @@ -17,7 +17,7 @@ Every turn runs as a durable workflow, built on the open-source [Workflow SDK](h The Workflow SDK is not inherently tied to Vercel. In local development and in a self-deployed `eve start` process, eve uses the SDK's local world by default; that world persists workflow runs on disk under `.eve/.workflow-data` and dispatches through the same Nitro-hosted workflow routes. On Vercel, the same workflow code runs against Vercel Workflow instead, which adds platform features such as latest production deployment routing and dashboard run metadata. -When a Vercel production deployment changes, the next model turn in an existing session uses that deployment's current instructions, model, and tools. The durable session keeps its conversation history and authored state, so identity-based channels such as Telegram private chats and Twilio phone-number conversations adopt agent updates without requiring a new session. +When a Git-connected Vercel deployment changes, the next model turn in an existing session attempts to use the latest deployment for that branch, so the session can pick up current instructions, model, and tools. The durable session keeps its conversation history and authored state, so identity-based channels such as Telegram private chats and Twilio phone-number conversations adopt agent updates without requiring a new session. If Vercel cannot resolve a latest deployment for that runtime, for example in a branchless CLI deployment, eve falls back to the current immutable deployment. Nitro hosts the HTTP routes and workflow entrypoints. It does not supply the workflow state store or the sandbox runtime. Those are separate adapters: Workflow uses the active world implementation, and Sandbox uses the backend from `agent/sandbox` or `defaultBackend()`. diff --git a/packages/eve/src/execution/workflow-runtime.test.ts b/packages/eve/src/execution/workflow-runtime.test.ts index 79346d5cc..13c7ef9ee 100644 --- a/packages/eve/src/execution/workflow-runtime.test.ts +++ b/packages/eve/src/execution/workflow-runtime.test.ts @@ -4,7 +4,9 @@ import type { ChannelAdapter } from "#channel/adapter.js"; import { ChannelRequestIdKey } from "#context/keys.js"; import { resolveInstalledPackageInfo } from "#internal/application/package.js"; import { + clearLatestDeploymentFallbackMemoForTest, createWorkflowRuntime, + LATEST_DEPLOYMENT_NO_GIT_BRANCH_MESSAGE, LATEST_DEPLOYMENT_UNSUPPORTED_MESSAGE, sessionTimeoutWorkflowReference, turnWorkflowReference, @@ -36,6 +38,7 @@ vi.mock("#runtime/sessions/compiled-agent-cache.js", () => ({ afterEach(() => { getHookByTokenMock.mockReset(); + clearLatestDeploymentFallbackMemoForTest(); getRunMock.mockReset(); getWorldMock.mockReset(); resumeHookMock.mockReset(); @@ -458,6 +461,7 @@ describe("createWorkflowRuntime#run", () => { "$eve.trigger": "subagent", "$eve.type": "subagent", }, + deploymentId: "latest", }); }); @@ -495,12 +499,95 @@ describe("createWorkflowRuntime#run", () => { }); }); + it("falls back to the current deployment when the no-git-branch error is wrapped", async () => { + vi.stubEnv("VERCEL_ENV", "preview"); + const compiledArtifactsSource = {} as RuntimeCompiledArtifactsSource; + mockBundleAndRun(compiledArtifactsSource); + const branchlessDeploymentError = new Error("Failed to resolve latest deployment", { + cause: new Error(LATEST_DEPLOYMENT_NO_GIT_BRANCH_MESSAGE), + }); + startMock + .mockRejectedValueOnce(branchlessDeploymentError) + .mockResolvedValueOnce({ runId: "driver-run" }); + + await buildRuntime(compiledArtifactsSource).run({ + adapter, + auth: null, + input: { message: "hello" }, + mode: "task", + }); + + expect(startMock).toHaveBeenNthCalledWith(1, workflowEntryReference, expect.any(Array), { + allowReservedAttributes: true, + attributes: { + "$eve.title": "hello", + "$eve.trigger": "http", + "$eve.type": "session", + }, + deploymentId: "latest", + }); + expect(startMock).toHaveBeenNthCalledWith(2, workflowEntryReference, expect.any(Array), { + allowReservedAttributes: true, + attributes: { + "$eve.title": "hello", + "$eve.trigger": "http", + "$eve.type": "session", + }, + }); + }); + + it("memoizes latest-deployment fallback after a pinned dispatch succeeds", async () => { + vi.stubEnv("VERCEL_ENV", "preview"); + const compiledArtifactsSource = {} as RuntimeCompiledArtifactsSource; + mockBundleAndRun(compiledArtifactsSource); + startMock + .mockRejectedValueOnce(new Error(LATEST_DEPLOYMENT_NO_GIT_BRANCH_MESSAGE)) + .mockResolvedValueOnce({ runId: "first-run" }) + .mockResolvedValueOnce({ runId: "second-run" }); + + await buildRuntime(compiledArtifactsSource).run({ + adapter, + auth: null, + input: { message: "first" }, + mode: "task", + }); + await buildRuntime(compiledArtifactsSource).run({ + adapter, + auth: null, + input: { message: "second" }, + mode: "task", + }); + + expect(startMock).toHaveBeenNthCalledWith(1, workflowEntryReference, expect.any(Array), { + allowReservedAttributes: true, + attributes: { + "$eve.title": "first", + "$eve.trigger": "http", + "$eve.type": "session", + }, + deploymentId: "latest", + }); + expect(startMock).toHaveBeenNthCalledWith(2, workflowEntryReference, expect.any(Array), { + allowReservedAttributes: true, + attributes: { + "$eve.title": "first", + "$eve.trigger": "http", + "$eve.type": "session", + }, + }); + expect(startMock).toHaveBeenNthCalledWith(3, workflowEntryReference, expect.any(Array), { + allowReservedAttributes: true, + attributes: { + "$eve.title": "second", + "$eve.trigger": "http", + "$eve.type": "session", + }, + }); + }); + it.each(["preview", "development", undefined])( - "pins workflowEntry to the current deployment when VERCEL_ENV is %s", + "starts workflowEntry on the latest deployment when VERCEL_ENV is %s", async (vercelEnv) => { - // Preview and CLI deployments carry no git branch reference, so the - // platform cannot resolve "latest" for them (HTTP 400). They must pin - // to their own immutable deployment. if (vercelEnv === undefined) { vi.stubEnv("VERCEL_ENV", ""); delete process.env.VERCEL_ENV; @@ -526,6 +613,7 @@ describe("createWorkflowRuntime#run", () => { "$eve.trigger": "http", "$eve.type": "session", }, + deploymentId: "latest", }); }, ); diff --git a/packages/eve/src/execution/workflow-runtime.ts b/packages/eve/src/execution/workflow-runtime.ts index d63ab146e..a89c143de 100644 --- a/packages/eve/src/execution/workflow-runtime.ts +++ b/packages/eve/src/execution/workflow-runtime.ts @@ -1,21 +1,12 @@ -import { - EntityConflictError, - HookNotFoundError, - RunExpiredError, - WorkflowRunNotFoundError, -} from "#compiled/@workflow/errors/index.js"; +import { HookNotFoundError } from "#compiled/@workflow/errors/index.js"; import type { - CancelTurnInput, - CancelTurnResult, DeliverInput, GetEventStreamOptions, HookPayload, RunHandle, RunInput, Runtime, - TerminateSessionInput, - TerminateSessionResult, } from "#channel/types.js"; import { serializeContext } from "#context/serialize.js"; import { @@ -24,13 +15,9 @@ import { readParentLineage, } from "#execution/eve-workflow-attributes.js"; import { resolveInstalledPackageInfo } from "#internal/application/package.js"; -import { isEveDevEnvironment } from "#internal/application/dev-environment.js"; import { createLogger, logError } from "#internal/logging.js"; import { - cancelRun, - getHookByToken, getRun, - getWorld, resumeHook, start, type Run, @@ -38,7 +25,7 @@ import { type WorkflowFunction, type WorkflowMetadata, } from "#internal/workflow/runtime.js"; -import type { MessageStreamEvent } from "#protocol/message.js"; +import type { HandleMessageStreamEvent } from "#protocol/message.js"; import type { RuntimeCompiledArtifactsSource } from "#runtime/compiled-artifacts-source.js"; import { ROOT_RUNTIME_AGENT_NODE_ID } from "#runtime/graph.js"; import { normalizeEveAttributes } from "#runtime/attributes/normalize.js"; @@ -46,20 +33,14 @@ import { getCompiledRuntimeAgentBundle } from "#runtime/sessions/compiled-agent- import { buildRunContext } from "#execution/runtime-context.js"; import { parseNdjsonStream } from "#execution/ndjson-stream.js"; import { RuntimeNoActiveSessionError } from "#execution/runtime-errors.js"; -import type { WorkflowEntryInput } from "#execution/workflow-entry.js"; -import { walkCauseChain } from "#shared/errors.js"; -import { - sessionCancelHookToken, - type TurnCancelPayload, -} from "#execution/turn-cancellation-token.js"; const WORKFLOW_ENTRY_NAME = "workflowEntry"; const TURN_WORKFLOW_NAME = "turnWorkflow"; -const SESSION_TIMEOUT_WORKFLOW_NAME = "sessionTimeoutWorkflow"; const EVE_PACKAGE_INFO = resolveInstalledPackageInfo(); export const LATEST_DEPLOYMENT_UNSUPPORTED_MESSAGE = "deploymentId 'latest' requires a World that implements resolveLatestDeploymentId()"; +export const LATEST_DEPLOYMENT_NO_GIT_BRANCH_MESSAGE = "Source deployment has no git branch"; /** * Workflow function names whose bundled id is stable across deployments @@ -74,12 +55,12 @@ export const LATEST_DEPLOYMENT_UNSUPPORTED_MESSAGE = export const STABLE_WORKFLOW_NAMES: ReadonlySet = new Set([ WORKFLOW_ENTRY_NAME, TURN_WORKFLOW_NAME, - SESSION_TIMEOUT_WORKFLOW_NAME, ]); const STABLE_ID_BASE = EVE_PACKAGE_INFO.name; const log = createLogger("execution.workflow-runtime"); +let latestDeploymentFallbackMemoized = false; interface WorkflowHookRecord { readonly runId: string; @@ -106,11 +87,6 @@ export const turnWorkflowReference = { workflowId: `workflow//${STABLE_ID_BASE}//${TURN_WORKFLOW_NAME}`, }; -/** Stable workflow reference for session deadline timers. */ -export const sessionTimeoutWorkflowReference = { - workflowId: `workflow//${STABLE_ID_BASE}//${SESSION_TIMEOUT_WORKFLOW_NAME}`, -}; - /** * Creates a workflow-backed runtime whose long-lived driver owns the * event stream and dispatches each turn as a child workflow run. @@ -128,17 +104,6 @@ export function createWorkflowRuntime(config: { const ctx = buildRunContext({ bundle, run: input }); const serializedContext = serializeContext(ctx); const parentLineage = readParentLineage(serializedContext); - const sessionTimeoutMs = bundle.resolvedAgent.config.limits?.sessionTimeoutMs; - const workflowInput: { - -readonly [K in keyof WorkflowEntryInput]: WorkflowEntryInput[K]; - } = { - input: input.input, - limits: input.limits, - serializedContext, - }; - if (sessionTimeoutMs !== undefined) { - workflowInput.sessionTimeoutMs = sessionTimeoutMs; - } const attributes = parentLineage.sessionId === undefined ? buildSessionAttributes({ @@ -156,10 +121,20 @@ export function createWorkflowRuntime(config: { let run: Awaited>; try { - run = await startWorkflowPreferLatest(workflowEntryReference, [workflowInput], { - allowReservedAttributes: true, - attributes: normalizeEveAttributes(attributes), - }); + run = await startWorkflowPreferLatest( + workflowEntryReference, + [ + { + input: input.input, + limits: input.limits, + serializedContext, + }, + ], + { + allowReservedAttributes: true, + attributes: normalizeEveAttributes(attributes), + }, + ); } catch (error) { logError(log, "failed to start workflow run", error, { continuationToken: input.continuationToken, @@ -167,9 +142,11 @@ export function createWorkflowRuntime(config: { throw error; } - let events: ReadableStream | undefined; + let events: ReadableStream | undefined; const getEvents = () => { - events ??= parseNdjsonStream(() => getRun(run.runId).getReadable()); + events ??= parseNdjsonStream(() => + getRun(run.runId).getReadable(), + ); return events; }; @@ -182,24 +159,6 @@ export function createWorkflowRuntime(config: { }; }, - async cancelTurn(input: CancelTurnInput): Promise { - return await requestWorkflowTurnCancellation(input); - }, - - async terminateSession(input: TerminateSessionInput): Promise { - try { - await cancelRun(await getWorld(), input.sessionId, { - cancelReason: input.reason ?? "Session reset by channel", - }); - return { status: "terminated" }; - } catch (error) { - if (isAlreadyTerminalSessionError(error)) { - return { status: "already_terminal" }; - } - throw error; - } - }, - async deliver(input: DeliverInput): Promise<{ sessionId: string }> { const hookPayload: Extract = { auth: input.auth, @@ -226,117 +185,68 @@ export function createWorkflowRuntime(config: { async getEventStream( sessionId: string, options?: GetEventStreamOptions, - ): Promise> { - return parseNdjsonStream(() => + ): Promise> { + return parseNdjsonStream(() => getRun(sessionId).getReadable({ startIndex: options?.startIndex }), ); }, - - async getStreamTailIndex(sessionId: string): Promise { - // The readable is never consumed; cancel it so the unread source does not linger. - const readable = getRun(sessionId).getReadable(); - try { - return await readable.getTailIndex(); - } finally { - await readable.cancel().catch(() => {}); - } - }, - - async resolveSession(continuationToken: string): Promise<{ sessionId: string } | undefined> { - try { - const hook = await getHookByToken(continuationToken); - return { sessionId: hook.runId }; - } catch (error) { - if (HookNotFoundError.is(error)) { - return undefined; - } - logError(log, "failed to resolve session by continuation token", error, { - continuationToken, - }); - throw error; - } - }, }; } -/** Requests cancellation through a session's stable workflow hook. */ -export async function requestWorkflowTurnCancellation( - input: CancelTurnInput, -): Promise { - const payload: TurnCancelPayload = input.turnId === undefined ? {} : { turnId: input.turnId }; - - try { - await resumeHook(sessionCancelHookToken(input.sessionId), payload); - return { status: "accepted" }; - } catch (error) { - if (isInactiveCancelTarget(error)) { - return { status: "no_active_turn" }; - } - throw error; - } -} - -function isInactiveCancelTarget(error: unknown): boolean { - return ( - HookNotFoundError.is(error) || - WorkflowRunNotFoundError.is(error) || - RunExpiredError.is(error) || - EntityConflictError.is(error) - ); -} - -function isAlreadyTerminalSessionError(error: unknown): boolean { - for (const candidate of walkCauseChain(error)) { - if ( - WorkflowRunNotFoundError.is(candidate) || - RunExpiredError.is(candidate) || - EntityConflictError.is(candidate) - ) { - return true; - } - } - return false; -} - /** - * Starts a workflow on the latest deployment when latest routing applies, - * while preserving local/dev worlds that do not implement latest routing. + * Starts a workflow on the latest deployment when the active workflow world can + * resolve it, while preserving worlds and deployments that cannot. */ export async function startWorkflowPreferLatest( workflow: WorkflowFunction | WorkflowMetadata, args: TArgs, options?: StartOptionsWithoutDeploymentId, ): Promise | Run> { - if (!shouldRouteToLatestDeployment()) { - return options === undefined - ? await start(workflow, args) - : await start(workflow, args, options); + if (latestDeploymentFallbackMemoized) { + return await startWorkflowOnCurrentDeployment(workflow, args, options); } try { return await start(workflow, args, { ...options, deploymentId: "latest" }); } catch (error) { - if (!isLatestDeploymentUnsupportedError(error)) { + if (!isLatestDeploymentFallbackError(error)) { throw error; } - return options === undefined - ? await start(workflow, args) - : await start(workflow, args, options); + const run = await startWorkflowOnCurrentDeployment(workflow, args, options); + latestDeploymentFallbackMemoized = true; + return run; } } /** - * Local development resolves "latest" to the active promoted generation. - * Vercel resolves it only for production deployments; previews and CLI - * deployments have no branch reference and remain pinned to themselves. + * Clears the latest-routing fallback memo between tests. Production code keeps + * the memo for the life of the process because unsupported worlds and + * branchless deployments cannot start resolving "latest" mid-process. */ -function shouldRouteToLatestDeployment(): boolean { - return process.env.VERCEL_ENV === "production" || isEveDevEnvironment(); +export function clearLatestDeploymentFallbackMemoForTest(): void { + latestDeploymentFallbackMemoized = false; +} + +async function startWorkflowOnCurrentDeployment( + workflow: WorkflowFunction | WorkflowMetadata, + args: TArgs, + options?: StartOptionsWithoutDeploymentId, +): Promise | Run> { + return options === undefined ? await start(workflow, args) : await start(workflow, args, options); +} + +function isLatestDeploymentFallbackError(error: unknown): boolean { + return ( + errorMessageIncludes(error, LATEST_DEPLOYMENT_UNSUPPORTED_MESSAGE) || + errorMessageIncludes(error, LATEST_DEPLOYMENT_NO_GIT_BRANCH_MESSAGE) + ); } -function isLatestDeploymentUnsupportedError(error: unknown): boolean { - return error instanceof Error && error.message.includes(LATEST_DEPLOYMENT_UNSUPPORTED_MESSAGE); +function errorMessageIncludes(error: unknown, message: string): boolean { + if (!(error instanceof Error)) return false; + if (error.message.includes(message)) return true; + return error.cause !== undefined && errorMessageIncludes(error.cause, message); } function normalizeWorkflowHook(value: unknown): WorkflowHookRecord { diff --git a/packages/eve/src/execution/workflow-steps.test.ts b/packages/eve/src/execution/workflow-steps.test.ts index 97ee8f682..db32f9825 100644 --- a/packages/eve/src/execution/workflow-steps.test.ts +++ b/packages/eve/src/execution/workflow-steps.test.ts @@ -4,20 +4,11 @@ import type { ChannelAdapter, ChannelAdapterContext } from "#channel/adapter.js" import type { DeliverPayload, SubagentInputRequestHookPayload } from "#channel/types.js"; import { ContextContainer } from "#context/container.js"; import { ContextKey } from "#context/key.js"; -import { - AuthKey, - ContinuationTokenKey, - ModeKey, - SessionDynamicToolMetadataKey, - SessionDynamicToolRuntimeRevisionKey, - SessionIdKey, -} from "#context/keys.js"; +import { AuthKey, ContinuationTokenKey, ModeKey, SessionIdKey } from "#context/keys.js"; import { BundleKey, ChannelKey } from "#runtime/sessions/runtime-context-keys.js"; import { serializeContext } from "#context/serialize.js"; -import { - getPendingRuntimeActionBatch, - setPendingRuntimeActionBatch, -} from "#harness/runtime-actions.js"; +import { setPendingRuntimeActionBatch } from "#harness/runtime-actions.js"; +import { DEFAULT_SUBAGENT_MAX_DEPTH } from "#harness/subagent-depth.js"; import { getPendingAuthorization, setPendingAuthorization } from "#harness/authorization.js"; import type { HarnessSession, StepResult } from "#harness/types.js"; import { createEmptyHookRegistry } from "#runtime/hooks/registry.js"; @@ -32,16 +23,17 @@ import { import { createTurnWorkflowInput } from "#execution/durable-session-migrations/turn-workflow.js"; import { projectToDurableSession } from "#execution/session.js"; import { createExecutionNodeStep } from "#execution/node-step.js"; -import { defineTool } from "#public/definitions/tool.js"; import { dispatchRuntimeActionsStep } from "#execution/dispatch-runtime-actions-step.js"; import { runProxySubagentEventStep } from "#execution/subagent-event-proxy-step.js"; -import { emitTerminalSessionFailureStep } from "#execution/terminal-session-failure-step.js"; import { dispatchTurnStep, + emitTerminalSessionFailureStep, resolveEffectiveOutputSchema, turnStep, } from "#execution/workflow-steps.js"; import { + clearLatestDeploymentFallbackMemoForTest, + LATEST_DEPLOYMENT_NO_GIT_BRANCH_MESSAGE, LATEST_DEPLOYMENT_UNSUPPORTED_MESSAGE, turnWorkflowReference, workflowEntryReference, @@ -109,11 +101,6 @@ function createTestWritable( } vi.mock("./node-step.js", () => ({ - buildRuntimeIdentity: vi.fn(() => ({ - agentId: "test-agent", - eveVersion: "0.0.0-test", - modelId: "test-model", - })), createExecutionNodeStep: vi.fn(), })); @@ -191,6 +178,7 @@ function createSerializedContext(): Record { } afterEach(() => { + clearLatestDeploymentFallbackMemoForTest(); getRunMock.mockReset(); startMock.mockReset(); workflowWritesByNamespace.clear(); @@ -242,21 +230,7 @@ describe("dispatchTurnStep", () => { ); }); - it("starts turn workflows on the latest promoted generation in local development", async () => { - vi.stubEnv("EVE_DEV", "1"); - const input = createTurnInput(); - startMock.mockResolvedValue({ runId: "turn-run" }); - - await expect(dispatchTurnStep(input)).resolves.toEqual({ runId: "turn-run" }); - - expect(startMock).toHaveBeenCalledWith( - turnWorkflowReference, - [createTurnWorkflowInput(input)], - expect.objectContaining({ deploymentId: "latest" }), - ); - }); - - it("pins turn workflows to the current deployment off production", async () => { + it("starts turn workflows on the latest deployment in preview when supported", async () => { vi.stubEnv("VERCEL_ENV", "preview"); const input = createTurnInput(); startMock.mockResolvedValue({ runId: "turn-run" }); @@ -275,6 +249,7 @@ describe("dispatchTurnStep", () => { "$eve.root": "sess-test", "$eve.type": "turn", }, + deploymentId: "latest", }, ); }); @@ -309,10 +284,41 @@ describe("dispatchTurnStep", () => { }, }); }); + + it("falls back to the current deployment when latest has no git branch", async () => { + vi.stubEnv("VERCEL_ENV", "preview"); + const input = createTurnInput(); + startMock + .mockRejectedValueOnce(new Error(LATEST_DEPLOYMENT_NO_GIT_BRANCH_MESSAGE)) + .mockResolvedValueOnce({ runId: "turn-run" }); + + await expect(dispatchTurnStep(input)).resolves.toEqual({ runId: "turn-run" }); + + const wireInput = createTurnWorkflowInput(input); + expect(startMock).toHaveBeenNthCalledWith(1, turnWorkflowReference, [wireInput], { + allowReservedAttributes: true, + attributes: { + "$eve.channel_request_id": "req_turn", + "$eve.parent": "sess-test", + "$eve.root": "sess-test", + "$eve.type": "turn", + }, + deploymentId: "latest", + }); + expect(startMock).toHaveBeenNthCalledWith(2, turnWorkflowReference, [wireInput], { + allowReservedAttributes: true, + attributes: { + "$eve.channel_request_id": "req_turn", + "$eve.parent": "sess-test", + "$eve.root": "sess-test", + "$eve.type": "turn", + }, + }); + }); }); describe("dispatchRuntimeActionsStep", () => { - it("preserves a started local child when a later start fails", async () => { + it("starts subagent child drivers on the latest deployment", async () => { vi.stubEnv("VERCEL_ENV", "production"); const compiledArtifactsSource = {} as never; const compiledBundle = { @@ -332,10 +338,10 @@ describe("dispatchRuntimeActionsStep", () => { subagentRegistry: { subagentsByNodeId: new Map([ [ - "subagents/agent", + "subagents/delegate", { definition: { - description: "Explicitly declared agent child description.", + description: "Local delegate child description.", kind: "subagent", }, }, @@ -346,9 +352,7 @@ describe("dispatchRuntimeActionsStep", () => { turnAgent: TestTurnAgent, } as never; vi.mocked(getCompiledRuntimeAgentBundle).mockResolvedValue(compiledBundle); - startMock - .mockResolvedValueOnce({ runId: "child-run" }) - .mockRejectedValueOnce(new Error("child start failed")); + startMock.mockResolvedValue({ runId: "child-run" }); getRunMock.mockReturnValue({ getReadable: () => new ReadableStream({ @@ -365,27 +369,16 @@ describe("dispatchRuntimeActionsStep", () => { description: "Runtime action event description.", input: { message: "investigate latest routing" }, kind: "subagent-call", - name: "agent", - nodeId: "subagents/agent", - subagentName: "agent", - }, - { - callId: "call-2", - description: "Second runtime action event description.", - input: { message: "investigate fallback routing" }, - kind: "subagent-call", - name: "agent", - nodeId: "subagents/agent", - subagentName: "agent", + name: "delegate", + nodeId: "subagents/delegate", + subagentName: "delegate", }, ], event: { sequence: 0, stepIndex: 0, turnId: "turn_0" }, responseMessages: [], session: createStubSession({ continuationToken: "http:parent", - rootSessionId: "root-session", sessionId: "parent-session", - subagentDepth: 99, }), }); installSessionStoreMocks([session]); @@ -402,39 +395,13 @@ describe("dispatchRuntimeActionsStep", () => { sessionState, }); - expect(result).toEqual({ - results: [ - { - callId: "call-2", - isError: true, - kind: "subagent-result", - output: { - code: "SUBAGENT_START_FAILED", - message: "child start failed", - }, - subagentName: "agent", - }, - ], - sessionState: expect.any(Object), - }); - expect(getPendingRuntimeActionBatch(result.sessionState.snapshot?.session.state)).toMatchObject( - { - childContinuationTokens: { - "call-1": "subagent:parent-session:call-1", - }, - childSessionIds: { - "call-1": "child-run", - }, - }, - ); + expect(result).toEqual({ results: [], sessionState: expect.any(Object) }); expect(startMock).toHaveBeenCalledWith( workflowEntryReference, [ expect.objectContaining({ input: { - message: expect.stringContaining( - "Description: Explicitly declared agent child description.", - ), + message: expect.stringContaining("Description: Local delegate child description."), }, limits: { maxInputTokensPerSession: false, @@ -452,7 +419,7 @@ describe("dispatchRuntimeActionsStep", () => { allowReservedAttributes: true, attributes: expect.objectContaining({ "$eve.parent": "parent-session", - "$eve.root": "root-session", + "$eve.root": "parent-session", "$eve.type": "subagent", }), deploymentId: "latest", @@ -573,91 +540,7 @@ describe("dispatchRuntimeActionsStep", () => { expect(workflowWritesByNamespace.get(DEFAULT_WORKFLOW_STREAM_NAMESPACE)).toBeUndefined(); }); - it("records a successfully started remote child session for cancellation", async () => { - const remote = { - definition: { - description: "Research remote", - kind: "remote", - name: "research", - nodeId: "remote/research", - path: "/eve/v1/session", - url: "https://remote.example.com", - }, - }; - const compiledBundle = { - adapterRegistry: { - adaptersByKind: new Map([[threadContextAdapter.kind, threadContextAdapter]]), - }, - compiledArtifactsSource: {}, - graph: { - nodesByNodeId: new Map(), - root: { - sandboxRegistry: { sandbox: null }, - turnAgent: TestTurnAgent, - }, - }, - hookRegistry: createEmptyHookRegistry(), - resolvedAgent: { config: {} }, - subagentRegistry: { - subagentsByNodeId: new Map([["remote/research", remote]]), - }, - toolRegistry: {}, - turnAgent: TestTurnAgent, - } as never; - vi.mocked(getCompiledRuntimeAgentBundle).mockResolvedValue(compiledBundle); - vi.stubGlobal( - "fetch", - vi - .fn() - .mockResolvedValue( - Response.json( - { ok: true, sessionId: "remote-child" }, - { headers: { "x-eve-session-id": "remote-child" }, status: 202 }, - ), - ), - ); - - const session = setPendingRuntimeActionBatch({ - actions: [ - { - callId: "call-remote", - description: "Delegate the work.", - input: { message: "investigate latest routing" }, - kind: "remote-agent-call", - name: "research", - nodeId: "remote/research", - remoteAgentName: "research", - }, - ], - event: { sequence: 0, stepIndex: 0, turnId: "turn_0" }, - responseMessages: [], - session: createStubSession({ - continuationToken: "http:parent", - sessionId: "parent-session", - }), - }); - installSessionStoreMocks([session]); - - const result = await dispatchRuntimeActionsStep({ - callbackBaseUrl: "https://caller.example.com", - parentContinuationToken: "turn-inbox", - parentWritable: createTestWritable(), - serializedContext: createSerializedContext(), - sessionState: createStubSessionState({ - continuationToken: "http:parent", - sessionId: "parent-session", - }), - }); - - expect(result.results).toEqual([]); - expect(getPendingRuntimeActionBatch(result.sessionState.snapshot?.session.state)).toMatchObject( - { - childSessionIds: { "call-remote": "remote-child" }, - }, - ); - }); - - it("blocks a stale recursive agent call from a delegated session", async () => { + it("blocks pending subagent calls at the subagent depth limit", async () => { const compiledBundle = { adapterRegistry: { adaptersByKind: new Map([[threadContextAdapter.kind, threadContextAdapter]]), @@ -688,18 +571,18 @@ describe("dispatchRuntimeActionsStep", () => { description: "Delegate the work.", input: { message: "try to recurse" }, kind: "subagent-call", - name: "agent", - nodeId: "__root__", - subagentName: "agent", + name: "delegate", + nodeId: "subagents/delegate", + subagentName: "delegate", }, ], event: { sequence: 0, stepIndex: 0, turnId: "turn_0" }, responseMessages: [], session: createStubSession({ continuationToken: "http:parent", - rootSessionId: "root-session", sessionId: "parent-session", - subagentDepth: 1, + subagentDepth: DEFAULT_SUBAGENT_MAX_DEPTH + 1, + subagentMaxDepth: DEFAULT_SUBAGENT_MAX_DEPTH + 1, }), }); installSessionStoreMocks([session]); @@ -723,21 +606,24 @@ describe("dispatchRuntimeActionsStep", () => { isError: true, kind: "subagent-result", output: { - code: "RECURSIVE_AGENT_ROOT_ONLY", - message: 'The built-in "agent" tool is only available to the root session.', + code: "SUBAGENT_DEPTH_LIMIT_REACHED", + currentDepth: DEFAULT_SUBAGENT_MAX_DEPTH + 1, + maxDepth: DEFAULT_SUBAGENT_MAX_DEPTH + 1, + message: `Subagent depth limit reached (${DEFAULT_SUBAGENT_MAX_DEPTH + 1}); "delegate" was not called.`, }, - subagentName: "agent", + subagentName: "delegate", }, ], sessionState, }); expect(startMock).not.toHaveBeenCalled(); expect(warn).toHaveBeenCalledWith( - "[eve:execution.dispatch-runtime-actions] recursive agent call blocked outside the root session", + "[eve:execution.dispatch-runtime-actions] subagent depth limit reached; blocking delegated call", expect.objectContaining({ callId: "call-1", - currentDepth: 1, - subagentName: "agent", + currentDepth: DEFAULT_SUBAGENT_MAX_DEPTH + 1, + maxDepth: DEFAULT_SUBAGENT_MAX_DEPTH + 1, + subagentName: "delegate", }), ); expect(workflowWritesByNamespace.get(DEFAULT_WORKFLOW_STREAM_NAMESPACE)).toBeUndefined(); @@ -995,115 +881,6 @@ describe("turnStep", () => { }); }); - it("refreshes session-scoped dynamic tools from the current deployment", async () => { - vi.stubEnv("VERCEL_DEPLOYMENT_ID", "dpl_new"); - const handler = vi.fn(() => ({ - current_tool: defineTool({ - description: "Current deployment tool", - inputSchema: { type: "object" }, - execute: async () => ({ ok: true }), - }), - })); - const dynamicToolResolver = { - eventNames: ["session.started"], - events: { "session.started": handler }, - logicalPath: "agent/tools/current.ts", - slug: "current", - sourceId: "test:current", - sourceKind: "module", - } as never; - const compiledArtifactsSource = { kind: "bundled" } as const; - const compiledBundle = { - adapterRegistry: { - adaptersByKind: new Map([[threadContextAdapter.kind, threadContextAdapter]]), - }, - compiledArtifactsSource, - graph: { - nodesByNodeId: new Map(), - root: { - sandboxRegistry: { sandbox: null }, - turnAgent: TestTurnAgent, - }, - }, - moduleMap: { nodes: {} }, - hookRegistry: createEmptyHookRegistry(), - resolvedAgent: { - config: {}, - dynamicToolResolvers: [dynamicToolResolver], - }, - subagentRegistry: {}, - toolRegistry: {}, - turnAgent: TestTurnAgent, - } as never; - vi.mocked(getCompiledRuntimeAgentBundle).mockResolvedValue(compiledBundle); - vi.mocked(createExecutionNodeStep).mockImplementation(() => { - return async (session): Promise => ({ - next: { done: true, output: "ok" }, - session, - }); - }); - - const session = createStubSession({ - state: { - "eve.harness.emission": { - sequence: 1, - sessionStarted: true, - stepIndex: 0, - turnId: "", - }, - }, - }); - installSessionStoreMocks([session]); - - const ctx = new ContextContainer(); - ctx.set(AuthKey, null); - ctx.set(BundleKey, compiledBundle); - ctx.set(ChannelKey, threadContextAdapter); - ctx.set(ContinuationTokenKey, "http:thread-context"); - ctx.set(ModeKey, "conversation"); - ctx.set(SessionIdKey, "session-1"); - ctx.set(SessionDynamicToolRuntimeRevisionKey, "deployment:dpl_old"); - ctx.set(SessionDynamicToolMetadataKey, [ - { - closureVars: {}, - description: "Stale deployment tool", - entryKey: "old_tool", - executeStepFnName: "eve:dynamic-tool//old", - inputSchema: { type: "object" }, - name: "old_tool", - resolverSlug: "old", - }, - ]); - - const result = await turnStep({ - input: { - kind: "deliver", - payloads: [{ message: "follow up" }], - }, - parentWritable: createTestWritable(), - serializedContext: serializeContext(ctx), - sessionState: createStubSessionState({ - emissionState: { - sequence: 1, - sessionStarted: true, - stepIndex: 0, - turnId: "", - }, - }), - }); - - expect(handler).toHaveBeenCalledOnce(); - expect(result.serializedContext[SessionDynamicToolRuntimeRevisionKey.name]).toBe( - "deployment:dpl_new", - ); - expect(result.serializedContext[SessionDynamicToolMetadataKey.name]).toEqual([ - expect.objectContaining({ - name: "current_tool", - resolverSlug: "current", - }), - ]); - }); - it("clears pending authorization after a matching callback resumes the turn", async () => { const challenge = { challenge: { @@ -1315,12 +1092,10 @@ describe("emitTerminalSessionFailureStep", () => { // raw Errors to this shape (`normalizeSerializableError`) before // handing them into the step so they survive JSON serialization. const error = { - detail: "private attachment name: confidential.png", - message: "attachment staging failed for confidential.png", + message: "attachment staging failed", name: "EveAttachmentError", kind: "resolver-threw", }; - const errorLog = vi.spyOn(console, "error").mockImplementation(() => {}); await emitTerminalSessionFailureStep({ error, @@ -1333,19 +1108,9 @@ describe("emitTerminalSessionFailureStep", () => { data: { code: string; message: string; details?: { errorId?: string } }; }; expect(data.code).toBe("EveAttachmentError"); - expect(data.message).toContain("attachment staging failed for confidential.png"); + expect(data.message).toContain("attachment staging failed"); expect(typeof data.details?.errorId).toBe("string"); - const providerLog = errorLog.mock.calls.find(([line]) => - String(line).includes("workflow loop threw"), - ); - expect(providerLog?.[1]).toEqual({ - code: "EveAttachmentError", - errorId: data.details?.errorId, - sessionId: "session-terminal", - }); - expect(JSON.stringify(providerLog)).not.toContain("confidential.png"); - // The terminal step must also write the event to the durable // stream so event-stream consumers see a canonical tail instead // of an abrupt close. @@ -1353,46 +1118,6 @@ describe("emitTerminalSessionFailureStep", () => { expect(writes.length).toBe(1); }); - it("replaces cataloged failures with their semantic summary while keeping the raw dump", async () => { - const sessionFailedCalls: Array<{ data: unknown }> = []; - const capturingAdapter: ChannelAdapter = { - kind: "thread-context", - async "session.failed"(data) { - sessionFailedCalls.push({ data }); - }, - }; - - const serialized = buildSerializedContextWithAdapter(capturingAdapter, "session-semantic"); - - const error = new TypeError("fetch failed", { - cause: Object.assign(new Error("connect ECONNREFUSED 127.0.0.1:443"), { - code: "ECONNREFUSED", - }), - }); - - await emitTerminalSessionFailureStep({ - error, - parentWritable: createTestWritable(), - serializedContext: serialized, - }); - - expect(sessionFailedCalls).toHaveLength(1); - const { data } = sessionFailedCalls[0] as { - data: { - code: string; - message: string; - details?: { detail?: string; hint?: string; semanticErrorId?: string }; - }; - }; - expect(data.code).toBe("Network request failed"); - expect(data.message).toContain("ECONNREFUSED"); - expect(data.details?.semanticErrorId).toBe("network-request-failed"); - expect(data.details?.hint).toContain("Check your internet connection"); - // The raw inspection stays attached so the private session trace keeps - // the evidence the curated message summarizes away. - expect(data.details?.detail).toContain("fetch failed"); - }); - it("does not throw when the adapter handler itself throws", async () => { // A throwing handler must not prevent the event from reaching // the durable stream. This mirrors `callAdapterEventHandler`'s @@ -1482,7 +1207,6 @@ describe("runProxySubagentEventStep", () => { kind: "tool-call", toolName: "dangerous_tool", }, - kind: "tool-approval", options: [ { id: "approve", label: "Approve" }, { id: "deny", label: "Deny" }, From be5cdf11ef89ef37155105969cb7bbddce9e0a0d Mon Sep 17 00:00:00 2001 From: iroiro147 Date: Thu, 6 Aug 2026 19:33:33 +0530 Subject: [PATCH 2/2] test(eve): expect deploymentId:latest in durable timer start under PR#616 semantics Signed-off-by: iroiro147 --- packages/eve/src/execution/session-timeout-steps.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/eve/src/execution/session-timeout-steps.test.ts b/packages/eve/src/execution/session-timeout-steps.test.ts index 854d6aeb7..116974059 100644 --- a/packages/eve/src/execution/session-timeout-steps.test.ts +++ b/packages/eve/src/execution/session-timeout-steps.test.ts @@ -37,7 +37,9 @@ describe("session timeout steps", () => { }; await expect(startSessionTimeoutStep(input)).resolves.toEqual({ runId: "timer-run" }); - expect(startMock).toHaveBeenCalledWith(sessionTimeoutWorkflowReference, [input]); + expect(startMock).toHaveBeenCalledWith(sessionTimeoutWorkflowReference, [input], { + deploymentId: "latest", + }); }); it("signals the owning session hook", async () => {