diff --git a/apps/hub/src/crypto-provider-cache-wiring.test.ts b/apps/hub/src/crypto-provider-cache-wiring.test.ts new file mode 100644 index 000000000..a973a1fd2 --- /dev/null +++ b/apps/hub/src/crypto-provider-cache-wiring.test.ts @@ -0,0 +1,75 @@ +// Pins CL-7284: createHub constructs one CryptoProviderCache and hands +// that same reference to every mail sender. A second constructor would +// still typecheck; this scan is the pin. crypto-cache.test.ts covers the +// Map; platform-adapter.test.ts covers chat injection only. +import { describe, expect, test } from "bun:test"; +import { readdirSync, readFileSync } from "node:fs"; +import path from "node:path"; + +const HUB_DIR = path.join(import.meta.dir, ".."); +const HUB_INDEX = path.join(import.meta.dir, "index.ts"); + +function productionTsFilesUnder(dir: string): string[] { + return readdirSync(dir, { withFileTypes: true }).flatMap((entry) => { + const full = path.join(dir, entry.name); + if (entry.isDirectory()) { + if (entry.name === "node_modules") return []; + return productionTsFilesUnder(full); + } + return entry.name.endsWith(".ts") && !entry.name.endsWith(".test.ts") + ? [full] + : []; + }); +} + +/** The full `callee(...)` text of the first call in `source`. */ +function firstCall(source: string, callee: string): string { + const token = `${callee}(`; + const start = source.indexOf(token); + if (start < 0) { + throw new Error( + `expected ${callee}(...) in ${path.relative(HUB_DIR, HUB_INDEX)}`, + ); + } + let depth = 1; + let i = start + token.length; + while (i < source.length && depth > 0) { + const ch = source[i]; + if (ch === "(") depth += 1; + else if (ch === ")") depth -= 1; + i += 1; + } + return source.slice(start, i); +} + +describe("hub crypto-provider cache wiring", () => { + test("createHub constructs one cache and passes it to every mail sender", () => { + const sites = productionTsFilesUnder(HUB_DIR).flatMap((file) => { + const matches = readFileSync(file, "utf8").match( + /createCryptoProviderCache\s*\(/g, + ); + return matches === null ? [] : matches.map(() => file); + }); + expect(sites).toEqual([HUB_INDEX]); + + const source = readFileSync(HUB_INDEX, "utf8"); + const assigned = + /const\s+(\w+)\s*=\s*createCryptoProviderCache\s*\(\s*\)/.exec( + source, + )?.[1]; + if (assigned === undefined) { + throw new Error( + "createHub must assign createCryptoProviderCache() to a const", + ); + } + + expect(firstCall(source, "createHubChatPlatform")).toContain(assigned); + expect(firstCall(source, "launchWebhookTrigger")).toContain( + `cryptoProviderCache: ${assigned}`, + ); + expect(firstCall(source, "createHubRoutineLauncher")).toContain( + `cryptoProviderCache: ${assigned}`, + ); + expect(firstCall(source, "runOneShotFoldedPrompt")).toContain(assigned); + }); +}); diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index baacce2fc..c9fec70fa 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -1329,6 +1329,17 @@ export async function createHub(config: HubConfig) { // notices has to be constructed first, since the sweep that triggers // most of them hangs off it. const relaunchNoticeRef: RelaunchNoticePort = {}; + // One CryptoProviderCache for the whole hub process (CL-7284). Chat + // sendMail keys by workbench id; webhook, routine, and agent-definition + // drafting first-turn mail key by the launched run's instance id. New + // workbenches and run ids are `run_` (`generateId("workflowRun")`); + // older workbenches are `ins_` (`generateId("instance")`). They share a + // string shape — a second cache for the same id would mint a different + // signing key. generateId uniqueness keeps distinct entities from + // colliding; sharing the cache keeps the same entity from rotating keys + // across consumers. TTL-bounded by `createCryptoProviderCache` itself + // (CL-7223). + const cryptoProviders = createCryptoProviderCache(); const chatPlatform = createHubChatPlatform({ db, sessionService, @@ -1339,6 +1350,7 @@ export async function createHub(config: HubConfig) { toolGrantsForPins, mcpCredentialBindingsFor, pinnedPackageCredentialBindingsFor, + cryptoProviders, // Chat residents are undeployed on idle again (see the comment above // this function): `chatIdleReapMs` (env-overridable via // `WORKBENCH_CHAT_IDLE_REAP_MS`, default 30 minutes) is @@ -2155,12 +2167,6 @@ export async function createHub(config: HubConfig) { // lives in our own schema rather than as any change to Interchange's // `grant` table. const repoReviewLeaseStore = createDrizzleRepoReviewLeaseStore(db); - // Shared by every folded-run first-turn mail send below (webhook - // triggers and routines alike) — a `CryptoProviderCache` is keyed by - // instance id, which is globally unique across this hub regardless of - // which caller minted the run, so one cache serves both without - // collision risk. - const foldedRunCryptoProviders = createCryptoProviderCache(); app.route( `${TENANT_PREFIX}/webhook-triggers`, createWebhookTriggerRoutes({ @@ -2196,7 +2202,7 @@ export async function createHub(config: HubConfig) { toolGrantsForPins, mcpCredentialBindingsFor, pinnedPackageCredentialBindingsFor, - cryptoProviderCache: foldedRunCryptoProviders, + cryptoProviderCache: cryptoProviders, launchMode: AGENT_SECTION_MODE, persistLaunch: workbenchLaunchPersistExtra, recordLaunchSources: ({ instanceId, sourcesDigest }) => @@ -2908,7 +2914,7 @@ export async function createHub(config: HubConfig) { toolGrantsForPins, mcpCredentialBindingsFor, pinnedPackageCredentialBindingsFor, - cryptoProviderCache: foldedRunCryptoProviders, + cryptoProviderCache: cryptoProviders, joinDeliveryWorkbench: (input) => joinRunParticipant({ store: chatStore }, input), }); @@ -3213,12 +3219,6 @@ export async function createHub(config: HubConfig) { listModels: listMyraModels, }; - // A separate `CryptoProviderCache` from `foldedRunCryptoProviders` - // above: an agent-definition drafting one-shot run's instance id has - // nothing to do with a folded run's, so a separate cache keeps them - // from ever contending over the same key space. - const agentDefinitionDraftingCryptoProviders = createCryptoProviderCache(); - // The create-agent panel's "Describe" step (CL-6074): a real one-shot // Myra call that proposes a starting system prompt/tool pins/skills // from a name + plain-language purpose, offering her the same @@ -3242,7 +3242,7 @@ export async function createHub(config: HubConfig) { { foldedRuns: oneShotFoldedRunsDeps, events: sidecarRouter.events, - cryptoProviders: agentDefinitionDraftingCryptoProviders, + cryptoProviders, undeploy: (address, reason) => sidecarRouter.sendAgentUndeploy(address, reason), }, diff --git a/apps/hub/src/routine-launcher.ts b/apps/hub/src/routine-launcher.ts index 2b2096220..cd2aed23d 100644 --- a/apps/hub/src/routine-launcher.ts +++ b/apps/hub/src/routine-launcher.ts @@ -267,6 +267,9 @@ export function createHubRoutineLauncher( const renderedContent = renderRoutineInput(input.input); const content = renderedContent === "" ? "Run this routine now." : renderedContent; + // Keyed by the launched run's instance id (`generateId("workflowRun")`), + // the same string shape chat uses for workbench ids — this adapter + // takes the host's process-wide cache rather than minting its own. const cryptoProvider = await deps.cryptoProviderCache.get(instanceId); const result = await sendFoldedMailWithRetry(deps, { tenantId: input.tenantId, diff --git a/packages/chat/README.md b/packages/chat/README.md index 8c75097c1..3fb9b017c 100644 --- a/packages/chat/README.md +++ b/packages/chat/README.md @@ -29,7 +29,10 @@ primitives, `@intx/authz` for grant checks, `@intx/crypto` for signing, folded, single-agent run whose mailbox is the workbench's shared timeline. - **`src/platform-adapter.ts`** / **`src/platform-port.ts`** — the `ChatPlatform` port this package needs from its host, and the hub-side - implementation composed from `@corbits/folded-runs`. + implementation composed from `@corbits/folded-runs`. The adapter does + not mint its own signing-key cache: the host injects one + `CryptoProviderCache` so chat sendMail shares keys with every other + folded-mail sender in the process. - **`src/chat-orchestrator.ts`** — turns invited-agent replies and approval-gate events into workbench messages and in-chat approve blocks. - **`src/workbench-service.ts`** — workbench-level orchestration above the diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index e59b1db7e..20b65dbc1 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -14,7 +14,7 @@ import { } from "@corbits/agent-lifecycle"; import { authoredDefinitionCandidates, - createCryptoProviderCache, + type CryptoProviderCache, DefinitionProjectionMissingError, domainOf, inferenceSourcesDigest, @@ -102,6 +102,15 @@ export type CreateHubChatPlatformDeps = { * the lifecycle construction below) has no signal at all. */ eventCollectors: EventCollectorRegistry; + /** + * Signing-key cache for outbound folded mail. The host constructs one + * process-wide instance (CL-7284) and passes it here so a workbench id + * looked up from chat cannot mint a different key than the same id + * looked up from webhook, routine, or one-shot drafting mail. This + * adapter keys `get` by `workbenchId` (`generateId("workflowRun")`, or + * older `generateId("instance")`). + */ + cryptoProviders: CryptoProviderCache; /** * Opt-in idle-sleep for every launched instance: absent here, the adapter keeps today's * behavior exactly (nothing ever sleeps, no interval runs). When @@ -223,8 +232,10 @@ export type HubChatPlatform = ChatPlatform & { /** * Composes the `ChatPlatform` port over the hub's real session - * services and `@corbits/folded-runs`. One crypto provider is minted - * per workbench and cached for the adapter's lifetime. + * services and `@corbits/folded-runs`. Outbound mail is signed with + * a `CryptoProvider` from `deps.cryptoProviders`, keyed by workbench + * id — the host owns the cache so every mail sender in the process + * shares it. */ export function createHubChatPlatform( deps: CreateHubChatPlatformDeps, @@ -250,7 +261,7 @@ export function createHubChatPlatform( : {}), }; - const cryptoProviders = createCryptoProviderCache(); + const cryptoProviders = deps.cryptoProviders; const wakeLogger = getLogger(["chat", "wake"]); // Built from `@corbits/agent-lifecycle` — the idle-sleep sweep and @@ -1228,6 +1239,10 @@ export function createHubChatPlatform( "sendMail requires either principalId or fromWorkbenchId", ); } + // Keyed by workbench id (`generateId("workflowRun")` / older + // `generateId("instance")`), not the invited agent's instance id. + // Those two id families share a prefix, so a second cache in this + // process would mint a different signing key for the same string. const cryptoProvider = await cryptoProviders.get(input.workbenchId); const attachments = input.content.attachments?.map( diff --git a/packages/chat/test/platform-adapter.test.ts b/packages/chat/test/platform-adapter.test.ts index aa635e31d..4ad0ffc7c 100644 --- a/packages/chat/test/platform-adapter.test.ts +++ b/packages/chat/test/platform-adapter.test.ts @@ -23,31 +23,34 @@ // exercised without a real Postgres. import { describe, expect, mock, test } from "bun:test"; -import type { FoldedRunsDeps } from "@corbits/folded-runs"; -import { - agentSession, - asset, - sessionAsset, - sessionMail, - workflowDefinition, - workflowDefinitionVersion, - workflowRun, -} from "@intx/db/schema"; -import { workbenchLaunch } from "../src/schema"; +import { IDLE_HIBERNATE_UNDEPLOY_REASON } from "@corbits/agent-lifecycle"; +import { AGENT_RUNTIME_SECTION_ID } from "@corbits/agent-runtime"; import { + createCryptoProviderCache, foldedRun, inferenceSourcesDigest, DefinitionProjectionMissingError, + type CryptoProviderCache, + type FoldedRunsDeps, } from "@corbits/folded-runs"; -import { IDLE_HIBERNATE_UNDEPLOY_REASON } from "@corbits/agent-lifecycle"; -import { AGENT_RUNTIME_SECTION_ID } from "@corbits/agent-runtime"; import { parseWorkflowSourceEntry, WORKFLOW_SOURCE_ENTRY_PATH, } from "@corbits/workflows"; +import type { DefinitionSourceResolution } from "@intx/hub-api"; +import { + agentSession, + asset, + sessionAsset, + sessionMail, + workflowDefinition, + workflowDefinitionVersion, + workflowRun, +} from "@intx/db/schema"; import { SessionLaunchError } from "@intx/hub-sessions"; import type { EventCollectorRegistry, SidecarRouter } from "@intx/hub-sessions"; -import type { DefinitionSourceResolution } from "@intx/hub-api"; +import { workbenchLaunch } from "../src/schema"; +import type { CreateHubChatPlatformDeps } from "../src/platform-adapter"; import { MODEL_UNAVAILABLE_CONSUMER_MESSAGE } from "../src/model-unavailable"; const actualHubApi = await import("@intx/hub-api"); @@ -103,7 +106,19 @@ mock.module("@intx/db", () => ({ listVisibleOfferings: async () => [], })); -const { createHubChatPlatform } = await import("../src/platform-adapter"); +const { createHubChatPlatform: buildHubChatPlatform } = + await import("../src/platform-adapter"); + +function createHubChatPlatform( + deps: Omit & { + cryptoProviders?: CryptoProviderCache; + }, +) { + return buildHubChatPlatform({ + ...deps, + cryptoProviders: deps.cryptoProviders ?? createCryptoProviderCache(), + }); +} type SelectChain = PromiseLike & { where(...args: unknown[]): SelectChain; @@ -991,6 +1006,76 @@ describe("createHubChatPlatform", () => { ); }); + test("sendMail signs with the injected crypto cache keyed by workbenchId", async () => { + resolveDefinitionSourcesResult = { + ok: true, + sources: [ + { + id: "off_1", + provider: "anthropic", + baseURL: "https://inference.invalid", + apiKey: "placeholder", + model: "claude-sonnet-5", + }, + ], + defaultSource: "off_1", + }; + + const db = createFakeDb({ + assetRow: { + tenantId: "ten_1", + creatorPrincipalId: "prin_creator", + name: "workbench-1", + displayName: null, + }, + definitionId: "wfd_workbench1", + workflowRunRow: { + id: "ins_workbench1", + address: "ins_workbench1@ten1.workbench.test", + principalId: "prin_run1", + }, + }); + db.inserted.push({ + table: agentSession, + values: { id: "ses_run1", principalId: "prin_run1" }, + }); + + const sessionService = createFakeSessionService(); + const getKeys: string[] = []; + const injectedProvider = { + getPublicKey: () => new Uint8Array([7, 2, 8, 4]), + }; + const cryptoProviders: CryptoProviderCache = { + get: async (key) => { + getKeys.push(key); + return injectedProvider as never; + }, + }; + + const platform = createHubChatPlatform({ + toolGrantsForPins: () => [], + db: db as never, + sessionService, + assetService: createFakeAssetService(), + sidecarRouter: createFakeSidecarRouter(), + eventCollectors: createFakeEventCollectors(), + cryptoProviders, + }); + + await platform.sendMail({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + principalId: "prin_sender", + content: { content: "hello workbench" }, + }); + + expect(getKeys).toEqual(["ins_workbench1"]); + const call = sessionService.sendUserMessageCalls[0] as { + cryptoProvider: unknown; + }; + expect(call.cryptoProvider).toBe(injectedProvider); + }); + test("sendMail rejects within the mail-delivery deadline instead of hanging forever when delivery never settles (CL-6644)", async () => { resolveDefinitionSourcesResult = { ok: true, diff --git a/packages/chat/test/relaunch-close.test.ts b/packages/chat/test/relaunch-close.test.ts index 1054f79d9..e3a6a7b1e 100644 --- a/packages/chat/test/relaunch-close.test.ts +++ b/packages/chat/test/relaunch-close.test.ts @@ -12,6 +12,7 @@ // session really is a different session — which is the only reason // this test can fail. import { describe, expect, test } from "bun:test"; +import { createCryptoProviderCache } from "@corbits/folded-runs"; import { agentSession } from "@intx/db/schema"; import { workbenchLaunch } from "../src/schema"; import { createHubChatPlatform } from "../src/platform-adapter"; @@ -115,6 +116,7 @@ function createPlatform(db: never) { assetService: {} as never, sidecarRouter: { getRoutableAddresses: () => [] } as never, eventCollectors: {} as never, + cryptoProviders: createCryptoProviderCache(), }); } diff --git a/packages/folded-run-one-shot/src/one-shot-reply.ts b/packages/folded-run-one-shot/src/one-shot-reply.ts index b4f6ceab0..789276d8b 100644 --- a/packages/folded-run-one-shot/src/one-shot-reply.ts +++ b/packages/folded-run-one-shot/src/one-shot-reply.ts @@ -238,6 +238,10 @@ export async function runOneShotFoldedPrompt( void (async () => { try { + // Keyed by the launched run's instance id (`generateId("workflowRun")`), + // the same string shape chat uses for workbench ids — the host + // injects its process-wide cache rather than this runner minting + // one of its own. const cryptoProvider = await deps.cryptoProviders.get(instanceId); const sent = await sendMail(deps.foldedRuns, { tenantId: input.tenantId, diff --git a/packages/folded-runs/README.md b/packages/folded-runs/README.md index 744cee711..12477dbd2 100644 --- a/packages/folded-runs/README.md +++ b/packages/folded-runs/README.md @@ -25,6 +25,10 @@ or a host-specific package such as `@corbits/chat`. 7-day idle ttl refreshed on every access (CL-7223): a key going momentarily unreachable (idle sleep, a sweep) does not evict it, so only a key nobody has asked for in a week is treated as gone for good. + Independent cache instances mint independent keys for the same string, + so a host constructs one process-wide cache and passes it to every + mail sender (chat, webhook, routine, one-shot) rather than constructing + one per consumer (CL-7284). - **Run lookups** (`./src/runs.ts`) — resolving a run by id or address, and bridging a run's principal to its live session via the shared-principal bridge. diff --git a/packages/folded-runs/src/crypto-cache.test.ts b/packages/folded-runs/src/crypto-cache.test.ts index 5bce413b7..c96ec11e7 100644 --- a/packages/folded-runs/src/crypto-cache.test.ts +++ b/packages/folded-runs/src/crypto-cache.test.ts @@ -61,4 +61,28 @@ describe("createCryptoProviderCache", () => { expect(a).not.toBe(b); expect(a.getPublicKey()).not.toEqual(b.getPublicKey()); }); + + test("independent caches mint different providers for the same key", async () => { + const first = createCryptoProviderCache(); + const second = createCryptoProviderCache(); + + const a = await first.get("run_same"); + const b = await second.get("run_same"); + + expect(a).not.toBe(b); + expect(a.getPublicKey()).not.toEqual(b.getPublicKey()); + }); + + test("one shared cache returns the same provider to every consumer of a key", async () => { + const cache = createCryptoProviderCache(); + + const chat = await cache.get("run_same"); + const webhook = await cache.get("run_same"); + const routine = await cache.get("run_same"); + const drafting = await cache.get("run_same"); + + expect(webhook).toBe(chat); + expect(routine).toBe(chat); + expect(drafting).toBe(chat); + }); }); diff --git a/packages/folded-runs/src/crypto-cache.ts b/packages/folded-runs/src/crypto-cache.ts index c8fef7937..d0a546168 100644 --- a/packages/folded-runs/src/crypto-cache.ts +++ b/packages/folded-runs/src/crypto-cache.ts @@ -2,7 +2,9 @@ // key stays active — mirroring the per-instance signing-key cache the // platform's own mail route keeps. A caller picks its own key (a // workbench id, an instance id, ...); this module knows nothing about -// what the key means. +// what the key means. Independent instances mint independent keys for +// the same string, so a host that has several mail senders constructs +// one cache and passes it to each of them. import { createEd25519Crypto, generateKeyPair } from "@intx/crypto"; import { createExpiringMap } from "@corbits/collections"; import type { CryptoProvider } from "@intx/types/runtime"; diff --git a/packages/webhook-triggers/src/launch.ts b/packages/webhook-triggers/src/launch.ts index 2edb06c62..4fd29d365 100644 --- a/packages/webhook-triggers/src/launch.ts +++ b/packages/webhook-triggers/src/launch.ts @@ -164,6 +164,9 @@ export async function launchWebhookTrigger( }); const content = renderInputTemplate(trigger.inputTemplate, payload); + // Keyed by the launched run's instance id (`generateId("workflowRun")`), + // the same string shape chat uses for workbench ids — share the host's + // process-wide cache so those lookups cannot mint different keys. const cryptoProvider = await deps.cryptoProviderCache.get(instanceId); const result = await sendFoldedMailWithRetry(deps, { tenantId: trigger.tenantId,