diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index 1faa70f3a7..fbf44a094d 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -168,6 +168,7 @@ import { import { createHubMailboxAuthorizeSender, createHubMailboxResolveRefs, + createHubPersistMailWithSessionEnsure, } from "./mailbox-persist"; import { createCommandRegistry, @@ -196,6 +197,8 @@ import { createScheduledWorkflowRoutes, renderWorkflowSourceTree, WORKFLOW_SOURCE_ENTRY, + ensureRunSession, + recordAgentSessionAtProvision, } from "@corbits/workflows"; import { createSidecarProvisioner as createE2BSidecarProvisioner, @@ -219,6 +222,7 @@ import { createWorkflowAllocationService, createWorkflowDispatchService, DEFAULT_ASSET_REF, + resolveRoutableAddress, WorkflowProvisioningError, type AgentRepoStore, type EventCollectorRegistry, @@ -713,6 +717,15 @@ export async function createHub(config: HubConfig) { args: Parameters[0], ) => Promise; } = {}; + // Forward reference: `eventCollectors` (the wrapped + // `EventCollectorRegistry`) isn't constructed until later in this + // composition, but `createHubPersistMailWithSessionEnsure` below needs + // it. Set once `eventCollectors` exists; every persistMail call before + // that point (there are none — the server isn't serving requests yet) + // would just no-op. + const eventCollectorsRef: { + current?: Pick; + } = {}; // CL-6499 (native multi-step routines): materializes a mail-triggered // run's authorization grants from its deploy-approved snapshot, so // ANY plain mail delivered to a workflow deployment's address — not @@ -746,7 +759,11 @@ export async function createHub(config: HubConfig) { // workbench ref is present before the post-commit bus event fires -- // no out-of-band UPDATE, no polling read. persistMail: createMailboxPersist(mailboxDb, { - upstream: baseLookups.persistMail, + upstream: createHubPersistMailWithSessionEnsure( + db, + eventCollectorsRef, + baseLookups.persistMail, + ), authorizeSender: createHubMailboxAuthorizeSender(db), bus: mailboxBus, resolveRefs: createHubMailboxResolveRefs(chatStore, roomMessages), @@ -885,6 +902,17 @@ export async function createHub(config: HubConfig) { baseEventCollectors.create(agentAddress, tenantId, sessionId, runId); }, dispatch(agentAddress, event) { + // CL-7480: a run's turn events can start arriving before anything + // else ever recorded its session — the first inbound trigger that + // just reconciled its principal races this same dispatch. No live + // collector for the address is exactly that case: ensure the + // session (and, inside it, the collector) before delegating, + // rather than silently dropping the event for a collector that + // never gets created. + if (!baseEventCollectors.has(agentAddress)) { + void ensureCollectorThenDispatch(agentAddress, event); + return; + } turnLatency.onEvent(agentAddress, event); baseEventCollectors.dispatch(agentAddress, event); // Mirrors the registry's own `isTerminal` check (event-collector-registry.ts) @@ -901,6 +929,36 @@ export async function createHub(config: HubConfig) { baseEventCollectors.abandon(agentAddress); }, }; + eventCollectorsRef.current = eventCollectors; + // Named so `dispatch`'s no-collector branch above reads clearly; not + // inlined there because it needs to be `async` and `dispatch` itself + // must stay synchronous (the `EventCollectorRegistry` contract). + async function ensureCollectorThenDispatch( + agentAddress: string, + event: Parameters[1], + ): Promise { + try { + const run = await resolveRoutableAddress(db, agentAddress); + if (run !== undefined) { + await ensureRunSession({ + db, + eventCollectors: baseEventCollectors, + runId: run.id, + }); + } + } catch (err) { + reportError(err, { + operation: "hub.eventCollectors.ensureRunSession", + extra: { agentAddress }, + }); + } + turnLatency.onEvent(agentAddress, event); + baseEventCollectors.dispatch(agentAddress, event); + const isTerminal = + event.type === "reactor.done" || + (event.type === "reactor.error" && event.data.fatal); + if (isTerminal) turnLatency.onSessionEnd(agentAddress); + } createHubSessionOrchestrator({ events: sidecarRouter.events, router: sidecarRouter, @@ -2017,15 +2075,7 @@ export async function createHub(config: HubConfig) { } try { - // Not `recordAgentSessionForRun` here: this mints a deployment - // with no message sent against it — a `workflow_run` born this - // way carries no principal until its first trigger (a call this - // deployer never makes) reconciles one onto it, so the call - // would only ever no-op. That trigger runs entirely through - // Interchange's own native route, with no Workbench-owned hook - // to record the session afterward (CL-7477 tracks the chat, - // webhook, and one-shot-prompt launch paths, which all send a - // message of their own and record there instead). + const sessionId = generateId("session"); const prepared = await workflowAllocationService.prepareProvisionedDeployment({ tenantId, @@ -2038,12 +2088,26 @@ export async function createHub(config: HubConfig) { }, entry, definitionAssetId: assetId, - sessionId: generateId("session"), + sessionId, sourceAuthorityPrincipalId: principalId, sourceOfferingIds, defaultSourceOfferingId, deployContent: { systemPrompt: "" }, }); + // The eager record every native launcher makes right after + // `prepareProvisionedDeployment` returns (CL-7481): this + // deployment mints no opening message of its own, but the + // trigger that eventually reconciles a principal onto it runs + // entirely through Interchange's own native route, with no + // Workbench-owned hook downstream to record the session + // afterward — so this is the only chance to record it at all. + await recordAgentSessionAtProvision({ + db, + eventCollectors, + runId: prepared.anchorRunId, + sessionId, + sourceAuthorityPrincipalId: principalId, + }); return { deploymentId: prepared.anchorRunId, definitionAssetId: assetId, @@ -2269,10 +2333,10 @@ export async function createHub(config: HubConfig) { repoStore: agentRepoStore.repoStore, workflowAllocationService, credentialCipher, + eventCollectors, isRoutable: (address) => sidecarRouter.getRoutableAddresses().includes(address), cryptoProviderCache: cryptoProviders, - eventCollectors, persistLaunch: async (input) => { await workbenchLaunchPersistExtra(input)(db); }, diff --git a/apps/hub/src/mailbox-persist.ts b/apps/hub/src/mailbox-persist.ts index 827dad17e1..6d37382a6f 100644 --- a/apps/hub/src/mailbox-persist.ts +++ b/apps/hub/src/mailbox-persist.ts @@ -16,6 +16,7 @@ import { tenant as tenantTable } from "@intx/db/schema"; import type { AuthorizeMailboxSender, CreateMailboxPersistOpts, + MailboxPersistArgs, MailboxRef, } from "@corbits/mailbox"; import { resolveRoutableAddress } from "@intx/hub-sessions"; @@ -24,6 +25,11 @@ import { type ChatStore, type RoomMessageStore, } from "@corbits/chat"; +import { ensureRunSession, type EventCollectorPort } from "@corbits/workflows"; +import { reportError } from "@corbits/error-sink"; +import { getLogger } from "@intx/log"; + +const logger = getLogger(["hub", "mailbox-persist"]); /** * `@corbits/mailbox` does not export `ResolveMailboxRefs` itself (only the @@ -57,6 +63,68 @@ type ResolveMailboxRefs = NonNullable< * fact from two directions: one extra indexed lookup by address per frame, * not per recipient. */ +/** + * Wraps the hub's own `persistMail` (vendor's `baseLookups.persistMail`) + * so a run's `agent_session` and event collector exist before that write + * runs — CL-7480: the first inbound mail on a freshly triggered run is + * often the earliest point the run is mail-routable at all, since + * `workflow_run.principal_id` only reconciles onto the trigger that just + * fired. Resolves the frame's own sender address to its run + * (`resolveRoutableAddress`, the same resolver `persistMail` itself + * uses) rather than assuming the caller already knows it; a sender that + * is not a live run resolves to `undefined` and this is a no-op past + * that point. Best-effort: `ensureRunSession` failing must not swallow + * the mail upstream is about to persist regardless, so this only + * degrades to "no session yet" rather than ever throwing past itself. + * + * `eventCollectorsRef` is a forward reference (the same pattern this + * file's other composition-order refs use, e.g. `apps/hub/src/index.ts`'s + * `grantAllowanceGateRef`): the wrapped `eventCollectors` registry isn't + * constructed until after `lookups` is, so this only reads `.current` + * inside the returned closure, never at wiring time. + */ +export function createHubPersistMailWithSessionEnsure< + R extends readonly unknown[], +>( + db: DB["db"], + eventCollectorsRef: { current?: Pick }, + upstream: (args: MailboxPersistArgs) => Promise, +): (args: MailboxPersistArgs) => Promise { + return async (args) => { + const sender = await resolveRoutableAddress(db, args.senderAddress); + try { + const eventCollectors = eventCollectorsRef.current; + if (sender !== undefined && eventCollectors !== undefined) { + await ensureRunSession({ db, eventCollectors, runId: sender.id }); + } + } catch (err) { + reportError(err, { + operation: "hub.mailboxPersist.ensureRunSession", + extra: { senderAddress: args.senderAddress }, + }); + } + // The vendored `session_mail` ledger (`upstream`, `baseLookups.persistMail`) + // keys on the sender run's principal, which does not exist until the + // run's first trigger anchors one — a run's own first outbound mail can + // reach this wrapper before that happens. `@corbits/mailbox`'s own write + // (this function's caller, `createMailboxPersist`) is keyed on the run + // address instead and runs regardless, so it is already the durable + // record for this frame; skip the vendored delegate rather than let it + // throw "no session for address" past the caller. + if (sender !== undefined && sender.sessionId === null) { + logger.debug( + "skipping vendored persistMail for run {runId}: no session yet", + { runId: sender.id, senderAddress: args.senderAddress }, + ); + // No `session_mail` rows were written for this frame: an empty + // result, not a thrown error, so a caller iterating this like + // `handleMailPersist` does simply emits no `mail.persisted` events. + return [] as unknown as R; + } + return upstream(args); + }; +} + export function createHubMailboxAuthorizeSender( db: DB["db"], ): AuthorizeMailboxSender { diff --git a/packages/agent-directory/src/one-shot-prompt.ts b/packages/agent-directory/src/one-shot-prompt.ts index 36efb6022e..b03351e45a 100644 --- a/packages/agent-directory/src/one-shot-prompt.ts +++ b/packages/agent-directory/src/one-shot-prompt.ts @@ -11,7 +11,7 @@ import { endAgentSessionForRun, readDefinitionProjection, readFoldedBody, - recordAgentSessionForRun, + recordAgentSessionAtProvision, WORKFLOW_SOURCE_ENTRY, type EventCollectorPort, } from "@corbits/workflows"; @@ -166,10 +166,13 @@ async function provisionOnAsset( ? { toolPackagePins: input.foldedBody.toolPackagePins } : {}), }); - // Not `recordAgentSessionForRun` here: a freshly provisioned run's - // `workflow_run.principal_id` is still null until its first trigger - // reconciles one onto it — `runOneShotPrompt` records the session - // right after the opening prompt actually sends. + await recordAgentSessionAtProvision({ + db: deps.db, + eventCollectors: deps.eventCollectors, + runId: prepared.anchorRunId, + sessionId, + sourceAuthorityPrincipalId: input.principalId, + }); return { runId: prepared.anchorRunId, address: prepared.deploymentAddress, @@ -322,25 +325,11 @@ export async function runOneShotPrompt( cryptoProvider, }); } - // The send above just drove the run's trigger path, the only - // thing that ever reconciles a principal onto a freshly - // provisioned `workflow_run` (CL-7477) — recording only now is - // what makes this ever find one. - try { - await recordAgentSessionForRun( - deps.db, - { - sessionId: launched.sessionId, - anchorRunId: launched.runId, - }, - deps.eventCollectors, - ); - } catch (err) { - reportError(err, { - operation: "agent-directory.one-shot.recordAgentSession", - extra: { address: launched.address }, - }); - } + // A run's session and event collector are ensured lazily now, at + // the hub seams that actually see the run become mail-routable + // (`apps/hub/src/mailbox-persist.ts`, the wrapped + // `eventCollectors` dispatch in `apps/hub/src/index.ts`) — + // CL-7480. Nothing here needs to record it. } catch (cause) { reportError(cause, { operation: "agent-directory.one-shot.send", diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index 8b820493d7..b7c5c51477 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -17,7 +17,9 @@ import { authoredDefinitionCandidates, DefinitionProjectionMissingError, endAgentSessionForRun, + ensureRunSession, readFoldedBody, + recordAgentSessionAtProvision, resolveNewestProjectedDefinition, WORKFLOW_SOURCE_ENTRY, } from "@corbits/workflows"; @@ -258,7 +260,6 @@ export function createHubChatPlatform( db: deps.db, sessionService: deps.sessionService, sidecarRouter: deps.sidecarRouter, - eventCollectors: deps.eventCollectors, }; function offeringDigest(sourceOfferingIds: readonly string[]): string { @@ -344,11 +345,13 @@ export function createHubChatPlatform( ? { toolPackagePins: input.foldedBody.toolPackagePins } : {}), }); - // Not `recordAgentSessionForRun` here: a freshly provisioned run's - // `workflow_run.principal_id` is still null (an invite sits - // un-triggered until someone actually writes into it), so this - // would only ever no-op. `sendRunMail` records it instead, once the - // run's first turn has actually reconciled a principal onto it. + await recordAgentSessionAtProvision({ + db: deps.db, + eventCollectors: deps.eventCollectors, + runId: prepared.anchorRunId, + sessionId, + sourceAuthorityPrincipalId: input.sourceAuthorityPrincipalId, + }); return { runId: prepared.anchorRunId, address: prepared.deploymentAddress, @@ -1313,6 +1316,25 @@ export function createHubChatPlatform( // idle sweep the moment they see traffic. lifecycle?.track(deliveryAddress); + // CL-7480: a run just woken/reconciled above can be routable + // before anything ever recorded its session — waking it is not + // itself a mail or dispatch seam, so nothing else along this path + // would ensure one. Best-effort: an ensure failure here still + // falls through to `resolveRunSessionIdOrThrow`'s own error. + try { + await ensureRunSession({ + db: deps.db, + eventCollectors: deps.eventCollectors, + runId: delivery.run.id, + }); + } catch (err) { + reportError(err, { + operation: "chat.sendMail.ensureRunSession", + tenantId: input.tenantId, + agentId: deliveryAddress, + extra: { runId: delivery.run.id }, + }); + } const sessionId = await resolveRunSessionIdOrThrow(deps.db, delivery.run); const domain = domainOf(deliveryAddress); // `fromWorkbenchId` names the room a dispatch speaks for. A room diff --git a/packages/chat/src/run-session.ts b/packages/chat/src/run-session.ts index 79ffea89c3..54a366a33c 100644 --- a/packages/chat/src/run-session.ts +++ b/packages/chat/src/run-session.ts @@ -1,19 +1,26 @@ -// Resolves a provisioned run's session id via Interchange's -// shared-principal bridge rather than any name derived from the run id. +// Resolves a provisioned run's session id off its own launch spec +// (`workflow_run_launch_spec.session_id`, written once at provision by +// every native launcher's `recordAgentSessionAtProvision` — CL-7481) +// rather than Interchange's principal-keyed `resolveRunSessionId`. A +// run's session id is fixed at provision time and never changes; its +// principal is not assigned until the run's first trigger anchors one, +// so a principal-keyed lookup returns nothing for exactly the run states +// (freshly provisioned, un-anchored) chat most needs to resolve. +import { eq } from "drizzle-orm"; import type { DB } from "@intx/db"; -import { resolveRunSessionId } from "@intx/hub-sessions"; +import { workflowRunLaunchSpec } from "@intx/db/schema"; export async function resolveRunSessionIdOrThrow( db: DB["db"], - run: { principalId: string | null }, + run: { id: string }, ): Promise { - const sessionId = await resolveRunSessionId(db, run.principalId, { - includeEnded: true, + const launchSpecRow = await db.query.workflowRunLaunchSpec.findFirst({ + where: eq(workflowRunLaunchSpec.anchorRunId, run.id), }); - if (sessionId === null) { + if (launchSpecRow === undefined) { throw new Error( - "no agent_session found for this run's principal; provision may not have completed", + `no workflow_run_launch_spec for run "${run.id}"; provision may not have completed`, ); } - return sessionId; + return launchSpecRow.sessionId; } diff --git a/packages/chat/src/send-run-mail.ts b/packages/chat/src/send-run-mail.ts index b570a22d20..139dfd98a4 100644 --- a/packages/chat/src/send-run-mail.ts +++ b/packages/chat/src/send-run-mail.ts @@ -1,10 +1,5 @@ // Signs one inbound message via Interchange `sendUserMessage` and // records it on session_mail so the run's own live subscribers see it. -import { - recordAgentSessionForRun, - type EventCollectorPort, -} from "@corbits/workflows"; -import { reportError } from "@corbits/error-sink"; import { sessionMail } from "@intx/db/schema"; import type { DB } from "@intx/db"; import type { SessionService, SidecarRouter } from "@intx/hub-sessions"; @@ -14,7 +9,6 @@ export type RunMailDeps = { db: DB["db"]; sessionService: Pick; sidecarRouter: Pick; - eventCollectors: Pick; }; export type SendRunMailParams = { @@ -106,28 +100,10 @@ export async function sendRunMail( const mailId = crypto.randomUUID(); const now = new Date(); const rawMIME = await deliverRunMailMIME(deps, params, mailId, now); - // `sendUserMessage` just drove this run's trigger path, which is the - // only thing that ever reconciles a principal onto a freshly - // provisioned `workflow_run` (CL-7477) — so the run is mail-routable - // now if it never was before. Best-effort and idempotent past the - // first successful call; a failure here must not swallow the mail - // that already sent. - try { - await recordAgentSessionForRun( - deps.db, - { - sessionId: params.sessionId, - address: params.agentAddress, - }, - deps.eventCollectors, - ); - } catch (err) { - reportError(err, { - operation: "chat.sendRunMail.recordAgentSession", - tenantId: params.tenantId, - agentId: params.agentAddress, - extra: { sessionId: params.sessionId }, - }); - } + // A run's session and event collector are ensured lazily now, at the + // hub seams that actually see the run become mail-routable + // (`apps/hub/src/mailbox-persist.ts`, the wrapped `eventCollectors` + // dispatch in `apps/hub/src/index.ts`) — CL-7480. Nothing here needs + // to record it. return recordRunMail(deps, params, mailId, now, rawMIME); } diff --git a/packages/chat/test/platform-adapter.test.ts b/packages/chat/test/platform-adapter.test.ts index 29281d3308..120538c926 100644 --- a/packages/chat/test/platform-adapter.test.ts +++ b/packages/chat/test/platform-adapter.test.ts @@ -343,6 +343,18 @@ function createFakeDb(opts: { * production code asks for, in any order. */ wireProjectionsByDefinitionId?: Record | undefined; + /** + * The `workflow_run_launch_spec` row `resolveRunSessionIdOrThrow` + * (CL-7481) reads by run id — the session id is fixed at provision + * time, unlike the principal it used to be looked up by. Explicit + * only where a test cares about an un-anchored run's session + * resolving before its principal ever exists; every other test here + * predates CL-7481 and seeds an `agentSession` row directly, so the + * fallback below derives one from that instead of touching every + * call site. + */ + workflowRunLaunchSpecRow?: + { anchorRunId: string; sessionId: string } | undefined; }) { const inserted: { table: unknown; values: unknown }[] = []; const updated: { table: unknown; values: unknown }[] = []; @@ -427,13 +439,63 @@ function createFakeDb(opts: { status: "pending", }; } - return inserted.findLast((row) => row.table === workflowRun) - ?.values as typeof opts.workflowRunRow | undefined; + const insertedRow = inserted.findLast( + (row) => row.table === workflowRun, + )?.values as typeof opts.workflowRunRow | undefined; + if (insertedRow !== undefined) return insertedRow; + // A run this fixture never configured: the freshly minted + // `anchorRunId` a fake `prepareProvisionedDeployment` just + // returned, which never wrote a `workflowRun` row into this + // fake db (Interchange's own provisioning is entirely mocked + // out here). `recordAgentSessionAtProvision` (CL-7481) reads + // the run row it just "provisioned" immediately after — this + // synthesizes the generic shape production code would see, + // so that read succeeds without every test wiring one up. + if (id === undefined) return undefined; + const domain = + opts.tenantRow?.domain ?? + opts.workflowRunRow?.address?.split("@")[1] ?? + "ten1.workbench.test"; + return { + id, + tenantId: opts.tenantRow?.id ?? "ten_1", + definitionId: opts.definitionId, + address: `${id}@${domain}`, + principalId: null, + }; }, }, sessionMail: { findFirst: async () => opts.sessionMailRow, }, + agentSession: { + findFirst: async (query?: { where?: unknown }) => { + const [id] = + query?.where !== undefined ? boundStringValues(query.where) : []; + return inserted.findLast( + (row) => + row.table === agentSession && + (id === undefined || (row.values as { id?: string }).id === id), + )?.values as { id: string; principalId: string } | undefined; + }, + }, + workflowRunLaunchSpec: { + findFirst: async (query?: { where?: unknown }) => { + const [id] = + query?.where !== undefined ? boundStringValues(query.where) : []; + if (opts.workflowRunLaunchSpecRow !== undefined) { + return id === undefined || + id === opts.workflowRunLaunchSpecRow.anchorRunId + ? opts.workflowRunLaunchSpecRow + : undefined; + } + const seededSession = inserted.find( + (row) => row.table === agentSession, + )?.values as { id: string } | undefined; + if (seededSession === undefined) return undefined; + return { anchorRunId: id, sessionId: seededSession.id }; + }, + }, workflowDefinition: { findFirst: async () => opts.workflowDefinitionRow ?? @@ -1020,6 +1082,71 @@ describe("createHubChatPlatform", () => { ); }); + test("sendMail on a freshly provisioned, un-anchored run resolves the launch-spec session id (CL-7481)", async () => { + resolveDefinitionSourcesResult = { + ok: true, + materials: [], + sources: [ + { + id: "off_1", + provider: "anthropic", + baseURL: "https://inference.invalid", + credentialId: "cred_placeholder", + model: "claude-sonnet-5", + }, + ], + defaultSource: "off_1", + }; + + // No principal on the run yet — an invite sits un-triggered until + // its first message. Before CL-7481, `resolveRunSessionIdOrThrow` + // looked the session up by this (null) principal and found nothing; + // now it reads the launch spec `recordAgentSessionAtProvision` + // wrote at provision time, keyed by the run's own id instead. + 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: null, + }, + workflowRunLaunchSpecRow: { + anchorRunId: "ins_workbench1", + sessionId: "ses_unanchored", + }, + }); + + const sessionService = createFakeSessionService(); + const sidecarRouter = createFakeSidecarRouter(); + + const platform = createPlatform({ + toolGrantsForPins: () => [], + db: db as never, + sessionService, + sidecarRouter, + eventCollectors: createFakeEventCollectors(), + }); + + const sent = await platform.sendMail({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + principalId: "prin_sender", + content: { content: "hello workbench" }, + }); + + expect(sent.id).toBeTruthy(); + const call = sessionService.sendUserMessageCalls[0] as { + sessionId: string; + }; + expect(call.sessionId).toBe("ses_unanchored"); + }); + test("sendMail signs with the injected crypto cache keyed by workbenchId", async () => { resolveDefinitionSourcesResult = { ok: true, diff --git a/packages/chat/test/relaunch-close.test.ts b/packages/chat/test/relaunch-close.test.ts index 4e85d0626b..35f2c0627d 100644 --- a/packages/chat/test/relaunch-close.test.ts +++ b/packages/chat/test/relaunch-close.test.ts @@ -78,6 +78,22 @@ function createFakeDb(opts: { ); }, }, + // CL-7481: `resolveRunSessionIdOrThrow` reads the launch spec by + // run id instead of walking through the run's principal — the + // session id it returns still matches this fixture's + // `ses_` convention, since every run here is already + // "anchored" with a fixed principal from the moment it's defined. + workflowRunLaunchSpec: { + findFirst: async ({ where }: { where: unknown }) => { + const [anchorRunId] = comparedValues(where); + const run = opts.runs.find((row) => row.id === anchorRunId); + if (run === undefined) return undefined; + return { + anchorRunId, + sessionId: `ses_${run.principalId ?? ""}`, + }; + }, + }, }, select: () => ({ from: (table: unknown) => ({ diff --git a/packages/webhook-triggers/src/launch.ts b/packages/webhook-triggers/src/launch.ts index 22e686d40e..d215dfb1a2 100644 --- a/packages/webhook-triggers/src/launch.ts +++ b/packages/webhook-triggers/src/launch.ts @@ -17,7 +17,7 @@ import { deliverWhenRoutable, readDefinitionProjection, readFoldedBody, - recordAgentSessionForRun, + recordAgentSessionAtProvision, WORKFLOW_SOURCE_ENTRY, type EventCollectorPort, } from "@corbits/workflows"; @@ -49,6 +49,13 @@ export type LaunchWebhookTriggerDeps = { "prepareProvisionedDeployment" >; sessionService: Pick; + /** + * The same wrapped `EventCollectorRegistry` every native launcher + * threads through (`apps/hub/src/index.ts`'s `eventCollectors`) — this + * launcher records the run's `agent_session` and event collector the + * instant `prepareProvisionedDeployment` returns (CL-7481). + */ + eventCollectors: Pick; /** * Reads the hub's live sidecar routing table. A freshly provisioned * run's sidecar takes several seconds to boot and register (CL-7476) @@ -59,7 +66,6 @@ export type LaunchWebhookTriggerDeps = { */ isRoutable: (address: string) => boolean; cryptoProviderCache: CryptoProviderCache; - eventCollectors: Pick; /** * Host-supplied cipher. Interchange decrypts bindings inside * `prepareProvisionedDeployment`; this port still owns the field so @@ -199,6 +205,14 @@ export async function launchWebhookTrigger( : {}), }); + await recordAgentSessionAtProvision({ + db: deps.db, + eventCollectors: deps.eventCollectors, + runId: prepared.anchorRunId, + sessionId, + sourceAuthorityPrincipalId: trigger.createdBy, + }); + await deps.persistLaunch({ tenantId: trigger.tenantId, instanceId: prepared.anchorRunId, @@ -240,30 +254,6 @@ export async function launchWebhookTrigger( }); } - // The delivery attempt above is what drove the run's trigger path, - // the only thing that ever reconciles a principal onto a freshly - // provisioned `workflow_run` (CL-7477) — recording only now, after - // that attempt, is what makes this call ever find one. Best-effort: - // a run that never became routable stays without a session until its - // first successful send. - try { - await recordAgentSessionForRun( - deps.db, - { - sessionId, - anchorRunId: prepared.anchorRunId, - }, - deps.eventCollectors, - ); - } catch (error) { - reportError(error, { - operation: "webhookTriggers.launch.recordAgentSession", - tenantId: trigger.tenantId, - agentId: prepared.deploymentAddress, - extra: { instanceId: prepared.anchorRunId }, - }); - } - return { instanceId: prepared.anchorRunId, triggerAddress: prepared.deploymentAddress, diff --git a/packages/webhook-triggers/test/launch.test.ts b/packages/webhook-triggers/test/launch.test.ts index 01b3b6785c..dad4716158 100644 --- a/packages/webhook-triggers/test/launch.test.ts +++ b/packages/webhook-triggers/test/launch.test.ts @@ -88,28 +88,35 @@ function createFakeDb() { query: { workflowDefinition: { findFirst: async () => DEFINITION_ROW }, tenant: { findFirst: async () => TENANT_ROW }, - // `recordAgentSessionForRun` reads the just-provisioned run back by - // id — production never inserts this row itself (Interchange's - // `prepareProvisionedDeployment` mints the anchor row - // transactionally before returning it), so this models the same - // row for `INTERCHANGE_RUN_ID`. workflowRun: { findFirst: async () => ({ id: INTERCHANGE_RUN_ID, tenantId: TENANT_ROW.id, definitionId: DEFINITION_ROW.id, - principalId: "prn_interchange", + address: INTERCHANGE_ADDRESS, + principalId: null, }), }, }, - insert: (_table: unknown) => ({ - values: (_values: unknown) => ({ - onConflictDoNothing: async (_opts: unknown) => {}, + insert: () => ({ + values: () => ({ + onConflictDoNothing: async () => undefined, }), }), }; } +let eventCollectorCreateCalls: unknown[] = []; + +function createFakeEventCollectors() { + return { + create: (...args: unknown[]) => { + eventCollectorCreateCalls.push(args); + }, + has: () => false, + }; +} + const TRIGGER = { id: "wht_1", tenantId: "ten_1", @@ -197,9 +204,7 @@ function baseDeps() { }, }, isRoutable: () => isRoutableForTest, - eventCollectors: { - create: () => undefined, - }, + eventCollectors: createFakeEventCollectors(), }; } @@ -211,6 +216,7 @@ function resetLaunchSpies() { sendUserMessageCalls = []; cryptoGetKeys = []; reportErrorCalls = []; + eventCollectorCreateCalls = []; visibleOfferings = [...DEFAULT_VISIBLE_OFFERINGS]; frozenProjection = INERT_PROJECTION; sendUserMessageImpl = async () => new Uint8Array([1]); diff --git a/packages/workflows/src/index.ts b/packages/workflows/src/index.ts index 244d0478ab..eefa55a850 100644 --- a/packages/workflows/src/index.ts +++ b/packages/workflows/src/index.ts @@ -56,10 +56,10 @@ export { } from "./schedule/cron"; export { scheduleCronFromProjection } from "./schedule/from-projection"; export { - recordAgentSessionForRun, + ensureRunSession, + recordAgentSessionAtProvision, endAgentSessionForPrincipal, endAgentSessionForRun, - type RecordAgentSessionParams, type EventCollectorPort, } from "./launch/agent-session"; export { diff --git a/packages/workflows/src/launch/agent-session.ts b/packages/workflows/src/launch/agent-session.ts index 58ec8fd409..f73c2a0d03 100644 --- a/packages/workflows/src/launch/agent-session.ts +++ b/packages/workflows/src/launch/agent-session.ts @@ -13,23 +13,38 @@ // keys on) is still null the instant `prepareProvisionedDeployment` // returns — a provisioned anchor is born "deployed" with no principal, // and only the run's first trigger reconciles one onto it (Interchange's -// `anchorWithPrincipal`, `vendor/intx/db/src/workflow-run-store.ts`), -// through the exact same trigger path every `sendUserMessage` drives. So -// this cannot run right after provisioning (an invite sits un-triggered -// until someone actually writes into it) — call it after a send that -// just delivered the run's first turn instead. A run with no principal -// yet is a normal, expected state, not a bug: this returns `false` -// rather than throwing, and idempotent past its first successful call. +// `anchorWithPrincipal`, `vendor/intx/db/src/workflow-run-store.ts`). +// Pre-creating the run's principal to dodge this is not an option either: +// Interchange's own grant materialization +// (`vendor/intx/hub-api/src/run-grant-materialization.ts`) inserts the +// principal row `onConflictDoNothing` and treats a conflict as "grants +// already committed", throwing rather than re-materializing — so nothing +// upstream of that first trigger may write the principal first. +// +// CL-7480 made `ensureRunSession` lazy: called from every seam that +// might be a run's first mail-routable moment, no-opping until a +// principal existed to key the row on. That left a real gap — a run's +// very first outbound message can itself be that seam, both for chat's +// `sendMail` (which looks the session up by run principal) and for the +// vendored `persistMail` a heartbeat run's own first outbound mail hits +// — and both landed on an un-anchored run with nothing recorded yet. +// CL-7481 fixes this at the root: `recordAgentSessionAtProvision` writes +// the row immediately after `prepareProvisionedDeployment` returns, +// keyed on the deploying principal (`sourceAuthorityPrincipalId`) since +// the run principal does not exist yet — every other field the launcher +// already knows or can read straight off the fresh run row. +// `ensureRunSession` is now purely a re-key: once the run's first +// trigger anchors a real principal onto it, this moves `principal_id` +// onto that principal without ever touching the session id. import { eq } from "drizzle-orm"; import type { DB } from "@intx/db"; -import { agentSession, workflowRun } from "@intx/db/schema"; +import { + agentSession, + workflowRun, + workflowRunLaunchSpec, +} from "@intx/db/schema"; import type { EventCollectorRegistry } from "@intx/hub-sessions"; -export type RecordAgentSessionParams = { - /** The id `prepareProvisionedDeployment` was called with. */ - readonly sessionId: string; -} & ({ readonly anchorRunId: string } | { readonly address: string }); - /** * The one port every launcher threads the same wrapped * `EventCollectorRegistry` through (`apps/hub/src/index.ts`'s @@ -37,64 +52,134 @@ export type RecordAgentSessionParams = { * is what actually makes `inference_turn`/`turn_part` rows exist for a * run (CL-7479: nothing else ever called it once `folded-runs` was * deleted); `abandon` tears the collector down when a run's session - * ends. + * ends; `has` lets a caller check whether a collector already exists + * before creating a second one. */ export type EventCollectorPort = Pick< EventCollectorRegistry, - "create" | "abandon" + "create" | "abandon" | "has" >; /** - * Inserts the `agent_session` row a run needs to be mail-routable: the - * run's `tenantId`, `definitionId` (the `agent_id` FK — the column keeps - * its old name but targets `workflow_definition` now), and `principalId` - * are all read fresh off the `workflow_run` row, never guessed by the - * caller. Returns `false` (and writes nothing) when the row has no - * principal yet, `true` once the session is recorded (or already was — - * a second call for the same `sessionId` is a no-op, not a conflict). + * The one shared write every native launcher calls right after its own + * `prepareProvisionedDeployment` returns (CL-7481) — a run row exists + * the instant that call resolves, so the session it will use is known + * then too: the launch-spec `sessionId` it just minted, `tenantId` and + * `definitionId` read fresh off the new `workflow_run` row (Interchange + * resolves the asset id the caller passed to a real `workflow_definition` + * row internally; callers never see that id themselves), and the + * deploying principal (`sourceAuthorityPrincipalId`) standing in for the + * run's own principal, which does not exist yet. `onConflictDoNothing` + * makes a second call for the same run id a no-op. Every launcher threads + * the same wrapped `EventCollectorRegistry` through here so the run's + * `inference_turn`/`turn_part` rows have somewhere to land from its very + * first turn. */ -export async function recordAgentSessionForRun( - db: DB["db"], - params: RecordAgentSessionParams, - eventCollectors: Pick, -): Promise { +export async function recordAgentSessionAtProvision(params: { + readonly db: DB["db"]; + readonly eventCollectors: Pick; + readonly runId: string; + readonly sessionId: string; + readonly sourceAuthorityPrincipalId: string; +}): Promise { + const { db, eventCollectors, runId, sessionId, sourceAuthorityPrincipalId } = + params; const runRow = await db.query.workflowRun.findFirst({ - where: - "anchorRunId" in params - ? eq(workflowRun.id, params.anchorRunId) - : eq(workflowRun.address, params.address), + where: eq(workflowRun.id, runId), }); - if (runRow === undefined || runRow.principalId === null) { - return false; + if (runRow === undefined) { + throw new Error( + `recordAgentSessionAtProvision: no workflow_run "${runId}" — prepareProvisionedDeployment must have already returned`, + ); } const now = new Date(); await db .insert(agentSession) .values({ - id: params.sessionId, + id: sessionId, tenantId: runRow.tenantId, agentId: runRow.definitionId, - principalId: runRow.principalId, + principalId: sourceAuthorityPrincipalId, status: "active", createdAt: now, updatedAt: now, }) .onConflictDoNothing({ target: agentSession.id }); - const agentAddress = "address" in params ? params.address : runRow.address; - if (agentAddress !== null) { - // Idempotent across wakes: a collector already live for this - // address is replaced, matching the registry's own contract - // (`event-collector-registry.ts`'s `create`). + if (runRow.address !== null && !eventCollectors.has(runRow.address)) { + eventCollectors.create( + runRow.address, + runRow.tenantId, + sessionId, + runRow.id, + ); + } +} + +/** + * Re-keys a run's `agent_session` onto the run's own principal once + * Interchange anchors one (`anchorWithPrincipal`, the run's first + * trigger) — before that, the session row is recorded under the + * deploying principal by `recordAgentSessionAtProvision`, since the run + * principal does not exist yet. The session id never changes; only + * `principal_id` moves, and only once (a session already re-keyed to the + * current run principal is left alone). The launch-spec row is the + * source of truth for which session belongs to this run — its absence + * means a launcher provisioned this run without going through the + * shared path, which is a bug, not a state to paper over. Also creates + * the run's event collector if none is live for its address (a fresh + * process has no in-memory collectors regardless of what is on disk). + */ +export async function ensureRunSession(params: { + readonly db: DB["db"]; + readonly eventCollectors: Pick; + readonly runId: string; +}): Promise { + const { db, eventCollectors, runId } = params; + const runRow = await db.query.workflowRun.findFirst({ + where: eq(workflowRun.id, runId), + }); + if (runRow === undefined) { + return null; + } + + const launchSpecRow = await db.query.workflowRunLaunchSpec.findFirst({ + where: eq(workflowRunLaunchSpec.anchorRunId, runId), + }); + if (launchSpecRow === undefined) { + throw new Error( + `ensureRunSession: no workflow_run_launch_spec for run "${runId}" — every launcher records one at provision time`, + ); + } + const sessionId = launchSpecRow.sessionId; + + if (runRow.principalId !== null) { + const sessionRow = await db.query.agentSession.findFirst({ + where: eq(agentSession.id, sessionId), + }); + if (sessionRow === undefined) { + throw new Error( + `ensureRunSession: no agent_session "${sessionId}" for run "${runId}" — recordAgentSessionAtProvision should have created it at launch`, + ); + } + if (sessionRow.principalId !== runRow.principalId) { + await db + .update(agentSession) + .set({ principalId: runRow.principalId, updatedAt: new Date() }) + .where(eq(agentSession.id, sessionId)); + } + } + + if (runRow.address !== null && !eventCollectors.has(runRow.address)) { eventCollectors.create( - agentAddress, + runRow.address, runRow.tenantId, - params.sessionId, + sessionId, runRow.id, ); } - return true; + return sessionId; } /** diff --git a/packages/workflows/test/agent-session.test.ts b/packages/workflows/test/agent-session.test.ts index 3ae50e221e..6fe8dcfb64 100644 --- a/packages/workflows/test/agent-session.test.ts +++ b/packages/workflows/test/agent-session.test.ts @@ -2,11 +2,16 @@ // repo's convention for tests that talk to a real Postgres (see // `packages/access-tools/test/routes.test.ts`). // -// CL-7477: before this fix, nothing wrote `agent_session` for a native -// launch, so `resolveRunSessionId` (vendor's mail-routing lookup) could -// never find a session for a run's principal and every launch's first -// turn failed to persist mail. This proves `recordAgentSessionForRun` -// writes a row `resolveRunSessionId` actually reads back. +// CL-7481: `recordAgentSessionAtProvision` is the eager write every +// native launcher makes right after `prepareProvisionedDeployment` +// returns — this proves it lands the `agent_session` row under the +// *deploying* principal (the run's own principal does not exist yet) +// and creates the run's event collector exactly once, idempotently. +// `ensureRunSession` is now purely a re-key: once the run's first +// trigger anchors a real principal onto it, this proves the session +// moves onto that principal without ever touching the session id, and +// that a run with no launch spec row is a bug this throws on rather +// than papering over. import { afterAll, beforeAll, expect, test } from "bun:test"; import { eq } from "drizzle-orm"; @@ -16,8 +21,8 @@ import { resolveRunSessionId } from "@intx/hub-sessions"; import { dbGate } from "../../../scripts/e2e/db-gate"; import { - endAgentSessionForRun, - recordAgentSessionForRun, + ensureRunSession, + recordAgentSessionAtProvision, } from "../src/launch/agent-session"; function dbConfigFromUrl(databaseUrl: string) { @@ -34,14 +39,17 @@ function dbConfigFromUrl(databaseUrl: string) { const databaseUrl = process.env["DATABASE_URL"]; const describeIfDb = dbGate(databaseUrl, import.meta.path); -describeIfDb("recordAgentSessionForRun", () => { +describeIfDb("recordAgentSessionAtProvision / ensureRunSession", () => { let db: DB; const tenantId = generateId("tenant"); - const principalId = generateId("principal"); + const deployingPrincipalId = generateId("principal"); + const runPrincipalId = generateId("principal"); const definitionId = generateId("workflowDefinition"); - const anchorRunId = generateId("workflowRun"); + const provisionedRunId = generateId("workflowRun"); + const noLaunchSpecRunId = generateId("workflowRun"); const sessionId = generateId("session"); + const domain = `agent-session-${tenantId}.localhost`; beforeAll(async () => { if (databaseUrl === undefined) return; @@ -51,35 +59,70 @@ describeIfDb("recordAgentSessionForRun", () => { id: tenantId, name: "Agent Session Test Tenant", slug: `agent-session-${tenantId}`, - domain: `agent-session-${tenantId}.localhost`, + domain, parentId: null, config: null, }); - await db.db.insert(schema.principal).values({ - id: principalId, - tenantId, - kind: "workflow", - refId: anchorRunId, - status: "active", - }); + await db.db.insert(schema.principal).values([ + { + id: deployingPrincipalId, + tenantId, + kind: "user", + refId: "usr_test", + status: "active", + }, + { + id: runPrincipalId, + tenantId, + kind: "workflow", + refId: provisionedRunId, + status: "active", + }, + ]); await db.db.insert(schema.workflowDefinition).values({ id: definitionId, tenantId, - creatorPrincipalId: principalId, + creatorPrincipalId: deployingPrincipalId, assetId: null, name: "agent-session-test-definition", }); + // Born the instant `prepareProvisionedDeployment` returns: a run row + // exists, its own principal is still null, and its launch spec names + // the session `recordAgentSessionAtProvision` will record. + await db.db.insert(schema.workflowRun).values({ + id: provisionedRunId, + definitionId, + anchorRunId: provisionedRunId, + tenantId, + principalId: null, + address: `${provisionedRunId}@${domain}`, + status: "deployed", + }); + await db.db.insert(schema.workflowRunLaunchSpec).values({ + anchorRunId: provisionedRunId, + sessionId, + deploymentDomain: domain, + sourceAuthorityPrincipalId: deployingPrincipalId, + frozenApprovalBundle: {}, + sourceOfferingIds: [], + defaultSourceOfferingId: "off_test", + deployContent: { systemPrompt: "" }, + }); + + // A run with no launch spec row at all: a launcher that skipped the + // shared provisioning path — a bug `ensureRunSession` must surface, + // not silently no-op past. await db.db.insert(schema.workflowRun).values({ - id: anchorRunId, + id: noLaunchSpecRunId, definitionId, - anchorRunId, + anchorRunId: noLaunchSpecRunId, tenantId, - principalId, - address: `${anchorRunId}@agent-session-${tenantId}.localhost`, - status: "running", + principalId: null, + address: `${noLaunchSpecRunId}@${domain}`, + status: "deployed", }); }); @@ -88,68 +131,97 @@ describeIfDb("recordAgentSessionForRun", () => { await db.db .delete(schema.agentSession) .where(eq(schema.agentSession.id, sessionId)); + await db.db + .delete(schema.workflowRunLaunchSpec) + .where(eq(schema.workflowRunLaunchSpec.anchorRunId, provisionedRunId)); + await db.db + .delete(schema.workflowRun) + .where(eq(schema.workflowRun.id, provisionedRunId)); await db.db .delete(schema.workflowRun) - .where(eq(schema.workflowRun.id, anchorRunId)); + .where(eq(schema.workflowRun.id, noLaunchSpecRunId)); await db.db .delete(schema.workflowDefinition) .where(eq(schema.workflowDefinition.id, definitionId)); await db.db .delete(schema.principal) - .where(eq(schema.principal.id, principalId)); + .where(eq(schema.principal.id, runPrincipalId)); + await db.db + .delete(schema.principal) + .where(eq(schema.principal.id, deployingPrincipalId)); await db.db.delete(schema.tenant).where(eq(schema.tenant.id, tenantId)); }); - test("writes an agent_session resolveRunSessionId reads back", async () => { - await recordAgentSessionForRun( - db.db, - { sessionId, anchorRunId }, - { create: () => undefined }, - ); - - const resolved = await resolveRunSessionId(db.db, principalId); - expect(resolved).toBe(sessionId); - }); - - test("creates the run's event collector with its address/tenant/session/run id", async () => { + test("records the session under the deploying principal and creates the collector once", async () => { const createCalls: unknown[] = []; - const otherSessionId = generateId("session"); - - await recordAgentSessionForRun( - db.db, - { sessionId: otherSessionId, anchorRunId }, - { - create: (...args: unknown[]) => { - createCalls.push(args); - }, + let hasCollector = false; + const eventCollectors = { + create: (...args: unknown[]) => { + createCalls.push(args); + hasCollector = true; }, - ); + has: () => hasCollector, + }; + + await recordAgentSessionAtProvision({ + db: db.db, + eventCollectors, + runId: provisionedRunId, + sessionId, + sourceAuthorityPrincipalId: deployingPrincipalId, + }); + // A second call for the same run is a no-op, not a conflict, and + // must not create a second collector. + await recordAgentSessionAtProvision({ + db: db.db, + eventCollectors, + runId: provisionedRunId, + sessionId, + sourceAuthorityPrincipalId: deployingPrincipalId, + }); expect(createCalls).toEqual([ - [ - `${anchorRunId}@agent-session-${tenantId}.localhost`, - tenantId, - otherSessionId, - anchorRunId, - ], + [`${provisionedRunId}@${domain}`, tenantId, sessionId, provisionedRunId], ]); - await db.db - .delete(schema.agentSession) - .where(eq(schema.agentSession.id, otherSessionId)); + const sessionRow = await db.db.query.agentSession.findFirst({ + where: eq(schema.agentSession.id, sessionId), + }); + expect(sessionRow?.principalId).toBe(deployingPrincipalId); + expect(sessionRow?.agentId).toBe(definitionId); }); - test("abandons the run's event collector when the session ends", async () => { - const abandonCalls: string[] = []; + test("re-keys the session onto the run principal once Interchange anchors it", async () => { + await db.db + .update(schema.workflowRun) + .set({ principalId: runPrincipalId, status: "running" }) + .where(eq(schema.workflowRun.id, provisionedRunId)); + + const resolved = await ensureRunSession({ + db: db.db, + eventCollectors: { create: () => {}, has: () => true }, + runId: provisionedRunId, + }); - await endAgentSessionForRun(db.db, anchorRunId, { - abandon: (address: string) => { - abandonCalls.push(address); - }, + expect(resolved).toBe(sessionId); + const sessionRow = await db.db.query.agentSession.findFirst({ + where: eq(schema.agentSession.id, sessionId), }); + expect(sessionRow?.principalId).toBe(runPrincipalId); - expect(abandonCalls).toEqual([ - `${anchorRunId}@agent-session-${tenantId}.localhost`, - ]); + // Interchange's own mail-routing lookup now finds the session by the + // run's own principal, exactly as it would for a folded run. + const viaVendorLookup = await resolveRunSessionId(db.db, runPrincipalId); + expect(viaVendorLookup).toBe(sessionId); + }); + + test("throws for a run with no launch spec row", async () => { + await expect( + ensureRunSession({ + db: db.db, + eventCollectors: { create: () => {}, has: () => false }, + runId: noLaunchSpecRunId, + }), + ).rejects.toThrow(/no workflow_run_launch_spec/); }); });