diff --git a/.changeset/calm-hooks-read.md b/.changeset/calm-hooks-read.md new file mode 100644 index 0000000000..1ab6fa8b5e --- /dev/null +++ b/.changeset/calm-hooks-read.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Avoid decrypting hook metadata when resolving session ownership or waiting for inbox registration and release. This removes unnecessary encryption-key work from channel routing, subagent startup, and session reset. diff --git a/packages/eve/src/execution/continuation-conflict-step.test.ts b/packages/eve/src/execution/continuation-conflict-step.test.ts index 52d5d83098..1b64fc119a 100644 --- a/packages/eve/src/execution/continuation-conflict-step.test.ts +++ b/packages/eve/src/execution/continuation-conflict-step.test.ts @@ -5,8 +5,9 @@ import { settleContinuationConflictStep } from "#execution/continuation-conflict import { sessionCommandHookToken } from "#execution/session-command-token.js"; const cancelRunMock = vi.fn(); -const getHookByTokenMock = vi.fn(); +const getRawHookByTokenMock = vi.fn(); const getWorldMock = vi.fn(); +const world = { hooks: { getByToken: getRawHookByTokenMock } }; const resumeSessionInboxMock = vi.fn(); vi.mock("#execution/wire/session-inbox-resume.js", () => ({ @@ -15,7 +16,7 @@ vi.mock("#execution/wire/session-inbox-resume.js", () => ({ vi.mock("#internal/workflow/runtime.js", () => ({ cancelRun: (...args: unknown[]) => cancelRunMock(...args), - getHookByToken: (...args: unknown[]) => getHookByTokenMock(...args), + getRawHookByToken: (...args: unknown[]) => getRawHookByTokenMock(...args), getWorld: (...args: unknown[]) => getWorldMock(...args), })); @@ -29,7 +30,7 @@ const command = { describe("settleContinuationConflictStep", () => { beforeEach(() => { vi.clearAllMocks(); - getWorldMock.mockResolvedValue("world"); + getWorldMock.mockResolvedValue(world); resumeSessionInboxMock.mockResolvedValue({ runId: "wrun_owner" }); cancelRunMock.mockResolvedValue(undefined); }); @@ -43,7 +44,7 @@ describe("settleContinuationConflictStep", () => { }); expect(resumeSessionInboxMock).toHaveBeenCalledWith("slack:C1:T1", command); - expect(cancelRunMock).toHaveBeenCalledWith("world", "wrun_collector", { + expect(cancelRunMock).toHaveBeenCalledWith(world, "wrun_collector", { cancelReason: "Session candidate did not acquire continuation ownership", }); }); @@ -71,15 +72,15 @@ describe("settleContinuationConflictStep", () => { resumeSessionInboxMock .mockRejectedValueOnce(new HookNotFoundError("slack:C1:T1")) .mockResolvedValueOnce({ runId: "wrun_owner" }); - getHookByTokenMock.mockResolvedValue({ runId: "wrun_owner" }); + getRawHookByTokenMock.mockResolvedValue({ runId: "wrun_owner" }); await settleContinuationConflictStep({ command, continuationToken: "slack:C1:T1", }); - expect(getHookByTokenMock).toHaveBeenCalledOnce(); - expect(getHookByTokenMock).toHaveBeenCalledWith("slack:C1:T1"); + expect(getRawHookByTokenMock).toHaveBeenCalledOnce(); + expect(getRawHookByTokenMock).toHaveBeenCalledWith("slack:C1:T1"); expect(resumeSessionInboxMock).toHaveBeenNthCalledWith( 2, sessionCommandHookToken("wrun_owner"), @@ -89,7 +90,7 @@ describe("settleContinuationConflictStep", () => { it("identifies a legacy delivery whose owner can no longer be resolved", async () => { resumeSessionInboxMock.mockRejectedValue(new HookNotFoundError("slack:C1:T1")); - getHookByTokenMock.mockRejectedValue(new HookNotFoundError("slack:C1:T1")); + getRawHookByTokenMock.mockRejectedValue(new HookNotFoundError("slack:C1:T1")); await expect( settleContinuationConflictStep({ diff --git a/packages/eve/src/execution/continuation-conflict-step.ts b/packages/eve/src/execution/continuation-conflict-step.ts index 2751fbd23f..408fe3a008 100644 --- a/packages/eve/src/execution/continuation-conflict-step.ts +++ b/packages/eve/src/execution/continuation-conflict-step.ts @@ -8,7 +8,7 @@ import { import type { SessionCommand } from "#channel/types.js"; import { sessionCommandHookToken } from "#execution/session-command-token.js"; import { resumeSessionInbox } from "#execution/wire/session-inbox-resume.js"; -import { cancelRun, getHookByToken, getWorld } from "#internal/workflow/runtime.js"; +import { cancelRun, getRawHookByToken, getWorld } from "#internal/workflow/runtime.js"; import { walkCauseChain } from "#shared/errors.js"; /** Settles side effects owned by a session candidate that lost its continuation claim. */ @@ -57,7 +57,7 @@ async function resolveOwnerSessionId(input: { if (input.ownerSessionId !== undefined) return input.ownerSessionId; try { - const hook = await getHookByToken(input.continuationToken); + const hook = await getRawHookByToken(input.continuationToken); if (typeof hook.runId === "string" && hook.runId.length > 0) return hook.runId; } catch (error) { if (!HookNotFoundError.is(error)) throw error; diff --git a/packages/eve/src/execution/wire/session-inbox-resume.test.ts b/packages/eve/src/execution/wire/session-inbox-resume.test.ts index 254c804f63..8eb0aaaded 100644 --- a/packages/eve/src/execution/wire/session-inbox-resume.test.ts +++ b/packages/eve/src/execution/wire/session-inbox-resume.test.ts @@ -9,15 +9,18 @@ import { } from "#execution/wire/session-inbox-resume.js"; const getHookByTokenMock = vi.fn(); +const getRawHookByTokenMock = vi.fn(); const resumeHookMock = vi.fn(); vi.mock("#compiled/@workflow/core/runtime.js", () => ({ getHookByToken: (...args: unknown[]) => getHookByTokenMock(...args), + getWorld: async () => ({ hooks: { getByToken: getRawHookByTokenMock } }), resumeHook: (...args: unknown[]) => resumeHookMock(...args), })); afterEach(() => { getHookByTokenMock.mockReset(); + getRawHookByTokenMock.mockReset(); resumeHookMock.mockReset(); }); @@ -45,17 +48,27 @@ describe("session inbox target resolution", () => { }); it("selects raw send for a markerless continuation owned by the stable-inbox cohort", async () => { - getHookByTokenMock.mockResolvedValue( + getRawHookByTokenMock.mockResolvedValue( sessionHook("session-1", sessionCommandHookToken("session-1")), ); await expect( resolveSessionInboxWireTarget(sessionHook("session-1", "continuation-1")), ).resolves.toEqual({ variant: "send", version: 0 }); + expect(getHookByTokenMock).not.toHaveBeenCalled(); + }); + + it("rejects a stable inbox belonging to a different run without hydrating its metadata", async () => { + getRawHookByTokenMock.mockResolvedValue({ runId: "different-run" }); + + await expect( + resolveSessionInboxWireTarget(sessionHook("session-1", "continuation-1")), + ).rejects.toThrow(/belongs to run "different-run", expected "session-1"/); + expect(getHookByTokenMock).not.toHaveBeenCalled(); }); it("selects deliver for a markerless continuation without a stable inbox", async () => { - getHookByTokenMock.mockRejectedValue( + getRawHookByTokenMock.mockRejectedValue( new HookNotFoundError(sessionCommandHookToken("session-1")), ); diff --git a/packages/eve/src/execution/wire/session-inbox-resume.ts b/packages/eve/src/execution/wire/session-inbox-resume.ts index d49702e625..60bad8335c 100644 --- a/packages/eve/src/execution/wire/session-inbox-resume.ts +++ b/packages/eve/src/execution/wire/session-inbox-resume.ts @@ -16,7 +16,7 @@ import { type SessionInboxWireTarget, } from "#execution/wire/session-inbox-contract.js"; import { sessionInboxWire } from "#execution/wire/session-inbox-encoder.js"; -import { getHookByToken, resumeHook } from "#internal/workflow/runtime.js"; +import { getHookByToken, getRawHookByToken, resumeHook } from "#internal/workflow/runtime.js"; import { isObject } from "#shared/guards.js"; type ResumedSessionInboxHook = Awaited>; @@ -66,7 +66,7 @@ export async function resolveSessionInboxWireTarget( if (hook.token === stableToken) return { variant: "send", version: 0 }; try { - const stableHook = await getHookByToken(stableToken); + const stableHook = await getRawHookByToken(stableToken); if (stableHook.runId !== hook.runId) { throw new SessionInboxWireError( `Stable session inbox ${JSON.stringify(stableToken)} belongs to run ${JSON.stringify(stableHook.runId)}, expected ${JSON.stringify(hook.runId)}.`, diff --git a/packages/eve/src/execution/workflow-runtime.test.ts b/packages/eve/src/execution/workflow-runtime.test.ts index d661975c18..45f619f457 100644 --- a/packages/eve/src/execution/workflow-runtime.test.ts +++ b/packages/eve/src/execution/workflow-runtime.test.ts @@ -13,6 +13,7 @@ import { ChannelRequestIdKey, ActivityObserverKey } from "#context/keys.js"; import { resolveInstalledPackageInfo } from "#internal/application/package.js"; import { createWorkflowRuntime, + waitForCommandHookOwner, activityCollectorWorkflowReference, sessionTimeoutWorkflowReference, startWorkflowOnCurrentDeployment, @@ -28,6 +29,8 @@ import { getCompiledRuntimeAgentBundle } from "#runtime/sessions/compiled-agent- import { markAgentTraceContext } from "#tracing/agent-trace-context.js"; const getHookByTokenMock = vi.fn(); +const getRawHookByTokenMock = vi.fn(); +const world = { hooks: { getByToken: getRawHookByTokenMock } }; const getRunMock = vi.fn(); const getWorldMock = vi.fn(); const resumeHookMock = vi.fn(); @@ -50,11 +53,13 @@ vi.mock("#runtime/sessions/compiled-agent-cache.js", () => ({ beforeEach(() => { cancelRunMock.mockResolvedValue(undefined); getHookByTokenMock.mockImplementation(async (token: string) => currentSessionHook(token)); - getWorldMock.mockResolvedValue("world"); + getWorldMock.mockResolvedValue(world); + getRawHookByTokenMock.mockImplementation(async (token: string) => currentSessionHook(token)); }); afterEach(() => { getHookByTokenMock.mockReset(); + getRawHookByTokenMock.mockReset(); getRunMock.mockReset(); getWorldMock.mockReset(); resumeHookMock.mockReset(); @@ -291,9 +296,9 @@ describe("createWorkflowRuntime command dispatch", () => { it("waits for reset to release the stable command inbox", async () => { const { HookNotFoundError } = await import("#compiled/@workflow/errors/index.js"); resumeHookMock.mockResolvedValue({ runId: "session-1" }); - getHookByTokenMock - .mockResolvedValueOnce(currentSessionHook("eve:token")) - .mockRejectedValue(new HookNotFoundError(sessionCommandHookToken("session-1"))); + getRawHookByTokenMock.mockRejectedValue( + new HookNotFoundError(sessionCommandHookToken("session-1")), + ); await expect( buildRuntime().dispatchContinuation({ @@ -306,7 +311,7 @@ describe("createWorkflowRuntime command dispatch", () => { reason: "User requested /new", version: 1, }); - expect(getHookByTokenMock).toHaveBeenCalledWith(sessionCommandHookToken("session-1")); + expect(getRawHookByTokenMock).toHaveBeenCalledWith(sessionCommandHookToken("session-1")); }); }); @@ -324,29 +329,52 @@ describe("createWorkflowRuntime#resolveContinuation", () => { } it("returns the owning session id from the hook lookup", async () => { - getHookByTokenMock.mockResolvedValue({ runId: "owner-session" }); + getRawHookByTokenMock.mockResolvedValue({ runId: "owner-session" }); await expect(buildRuntime().resolveContinuation("test:token")).resolves.toEqual({ sessionId: "owner-session", }); - expect(getHookByTokenMock).toHaveBeenCalledWith("test:token"); + expect(getRawHookByTokenMock).toHaveBeenCalledWith("test:token"); + expect(getHookByTokenMock).not.toHaveBeenCalled(); }); it("returns undefined for an unknown token", async () => { const { HookNotFoundError } = await import("#compiled/@workflow/errors/index.js"); - getHookByTokenMock.mockRejectedValue(new HookNotFoundError("test:token")); + getRawHookByTokenMock.mockRejectedValue(new HookNotFoundError("test:token")); await expect(buildRuntime().resolveContinuation("test:token")).resolves.toBeUndefined(); }); it("rethrows unexpected lookup failures", async () => { const failure = new Error("transient backing-store outage"); - getHookByTokenMock.mockRejectedValue(failure); + getRawHookByTokenMock.mockRejectedValue(failure); await expect(buildRuntime().resolveContinuation("test:token")).rejects.toBe(failure); }); }); +describe("waitForCommandHookOwner", () => { + it("resolves the winning run without hydrating hook metadata", async () => { + getRawHookByTokenMock.mockResolvedValue({ + get metadata() { + throw new Error("Ownership must not read encrypted metadata."); + }, + runId: "winning-run", + }); + + await expect(waitForCommandHookOwner("task:token")).resolves.toEqual({ runId: "winning-run" }); + expect(getHookByTokenMock).not.toHaveBeenCalled(); + }); + + it("does not turn storage failures into missing ownership", async () => { + const failure = new Error("backing store unavailable"); + getRawHookByTokenMock.mockRejectedValue(failure); + + await expect(waitForCommandHookOwner("task:token")).rejects.toBe(failure); + expect(getRawHookByTokenMock).toHaveBeenCalledOnce(); + }); +}); + describe("createWorkflowRuntime#createSession", () => { const adapter: ChannelAdapter = { kind: "http" }; @@ -632,7 +660,7 @@ describe("createWorkflowRuntime#createSession", () => { }), ).rejects.toBe(failure); - expect(cancelRunMock).toHaveBeenCalledWith("world", "collector-run", { + expect(cancelRunMock).toHaveBeenCalledWith(world, "collector-run", { cancelReason: "Root session creation did not complete", }); }); diff --git a/packages/eve/src/execution/workflow-runtime.ts b/packages/eve/src/execution/workflow-runtime.ts index 2a86252659..493017d0de 100644 --- a/packages/eve/src/execution/workflow-runtime.ts +++ b/packages/eve/src/execution/workflow-runtime.ts @@ -32,7 +32,7 @@ import { resolveInstalledPackageInfo } from "#internal/application/package.js"; import { createLogger, logError } from "#internal/logging.js"; import { cancelRun, - getHookByToken, + getRawHookByToken, getRun, getWorld, start, @@ -305,7 +305,7 @@ export function createWorkflowRuntime(config: { continuationToken: string, ): Promise<{ sessionId: string } | undefined> { try { - const hook = await getHookByToken(continuationToken); + const hook = await getRawHookByToken(continuationToken); return { sessionId: hook.runId }; } catch (error) { if (HookNotFoundError.is(error)) { @@ -423,7 +423,7 @@ export async function waitForCommandHookOwner(token: string): Promise= deadline) throw error; await new Promise((resolve) => setTimeout(resolve, 20)); @@ -435,7 +435,7 @@ async function waitForCommandHookRelease(token: string, sessionId: string): Prom const deadline = Date.now() + COMMAND_HOOK_READY_TIMEOUT_MS; while (true) { try { - const owner = normalizeWorkflowHook(await getHookByToken(token)); + const owner = normalizeWorkflowHook(await getRawHookByToken(token)); if (owner.runId !== sessionId) return; } catch (error) { if (HookNotFoundError.is(error)) return; diff --git a/packages/eve/src/internal/workflow/runtime.ts b/packages/eve/src/internal/workflow/runtime.ts index 649eb02d70..b3e2e55a60 100644 --- a/packages/eve/src/internal/workflow/runtime.ts +++ b/packages/eve/src/internal/workflow/runtime.ts @@ -1,4 +1,5 @@ import * as workflowRuntime from "#compiled/@workflow/core/runtime.js"; +import type { Hook } from "#compiled/@workflow/world/index.js"; export * from "#compiled/@workflow/core/runtime.js"; export type { @@ -7,6 +8,14 @@ export type { WorkflowMetadata, } from "#compiled/@workflow/core/runtime/start.js"; +/** + * Reads a hook without decrypting metadata or resolving an encryption key. + * Metadata is excluded from the return type; use `getHookByToken` to read it. + */ +export async function getRawHookByToken(token: string): Promise> { + return await (await workflowRuntime.getWorld()).hooks.getByToken(token); +} + /** Installs a World across source and vendored Workflow package identities. */ export function setWorld(world: unknown): void { workflowRuntime.setWorld(world as Parameters[0]);