diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index c3352c5a3..bd04767ec 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,7 @@ import { createScheduledWorkflowRoutes, renderWorkflowSourceTree, WORKFLOW_SOURCE_ENTRY, + ensureRunSession, } from "@corbits/workflows"; import { createSidecarProvisioner as createE2BSidecarProvisioner, @@ -219,6 +221,7 @@ import { createWorkflowAllocationService, createWorkflowDispatchService, DEFAULT_ASSET_REF, + resolveRoutableAddress, WorkflowProvisioningError, type AgentRepoStore, type EventCollectorRegistry, @@ -713,6 +716,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 +758,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 +901,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 +928,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, @@ -3069,6 +3126,7 @@ export async function createHub(config: HubConfig) { repoStore: agentRepoStore.repoStore, workflowAllocationService, sessionService, + eventCollectors, }, runnerInput, ), diff --git a/apps/hub/src/mailbox-persist.ts b/apps/hub/src/mailbox-persist.ts index 827dad17e..f8aa0e2f0 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,8 @@ import { type ChatStore, type RoomMessageStore, } from "@corbits/chat"; +import { ensureRunSession, type EventCollectorPort } from "@corbits/workflows"; +import { reportError } from "@corbits/error-sink"; /** * `@corbits/mailbox` does not export `ResolveMailboxRefs` itself (only the @@ -57,6 +60,48 @@ 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( + db: DB["db"], + eventCollectorsRef: { current?: Pick }, + upstream: (args: MailboxPersistArgs) => Promise, +): (args: MailboxPersistArgs) => Promise { + return async (args) => { + try { + const eventCollectors = eventCollectorsRef.current; + const sender = await resolveRoutableAddress(db, args.senderAddress); + if (sender !== undefined && eventCollectors !== undefined) { + await ensureRunSession({ db, eventCollectors, runId: sender.id }); + } + } catch (err) { + reportError(err, { + operation: "hub.mailboxPersist.ensureRunSession", + extra: { senderAddress: args.senderAddress }, + }); + } + 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 f3084ea66..1851527d4 100644 --- a/packages/agent-directory/src/one-shot-prompt.ts +++ b/packages/agent-directory/src/one-shot-prompt.ts @@ -11,8 +11,8 @@ import { endAgentSessionForRun, readDefinitionProjection, readFoldedBody, - recordAgentSessionForRun, WORKFLOW_SOURCE_ENTRY, + type EventCollectorPort, } from "@corbits/workflows"; import { listVisibleOfferings, type DB } from "@intx/db"; import { tenant as tenantTable, workflowDefinition } from "@intx/db/schema"; @@ -57,6 +57,7 @@ export type OneShotRunnerDeps = { "prepareProvisionedDeployment" >; readonly sessionService: Pick; + readonly eventCollectors: EventCollectorPort; /** * Test seam only. Production never sets these; they default to * Interchange `prepareProvisionedDeployment` and `sendUserMessage`. @@ -164,10 +165,10 @@ async function provisionOnAsset( ? { toolPackagePins: input.foldedBody.toolPackagePins } : {}), }); - // Not `recordAgentSessionForRun` here: a freshly provisioned run's + // Not `ensureRunSession` 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. + // reconciles one onto it, and the hub's own mail/dispatch seams + // ensure the session lazily once that happens (CL-7480). return { runId: prepared.anchorRunId, address: prepared.deploymentAddress, @@ -267,7 +268,11 @@ export async function runOneShotPrompt( clearTimeout(timer); unsubscribe(); try { - await endAgentSessionForRun(deps.db, launched.runId); + await endAgentSessionForRun( + deps.db, + launched.runId, + deps.eventCollectors, + ); } catch (err) { reportError(err, { operation: "agent-directory.one-shot.end-session", @@ -316,21 +321,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, - }); - } 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 5a432c5d7..a546564e7 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -16,7 +16,8 @@ import { and, desc, eq } from "drizzle-orm"; import { authoredDefinitionCandidates, DefinitionProjectionMissingError, - endAgentSessionForPrincipal, + endAgentSessionForRun, + ensureRunSession, readFoldedBody, resolveNewestProjectedDefinition, WORKFLOW_SOURCE_ENTRY, @@ -343,11 +344,12 @@ export function createHubChatPlatform( ? { toolPackagePins: input.foldedBody.toolPackagePins } : {}), }); - // Not `recordAgentSessionForRun` here: a freshly provisioned run's + // Not `ensureRunSession` 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. + // would only ever no-op. The hub's own mail/dispatch seams ensure + // the session lazily instead, once the run's first turn has + // actually reconciled a principal onto it (CL-7480). return { runId: prepared.anchorRunId, address: prepared.deploymentAddress, @@ -476,9 +478,7 @@ export function createHubChatPlatform( foldedBody: binding.foldedBody, }); - if (run.principalId !== null) { - await endAgentSessionForPrincipal(deps.db, run.principalId); - } + await endAgentSessionForRun(deps.db, run.id, deps.eventCollectors); await repointBinding( deps.db, @@ -1314,6 +1314,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/send-run-mail.ts b/packages/chat/src/send-run-mail.ts index e79a51e8f..139dfd98a 100644 --- a/packages/chat/src/send-run-mail.ts +++ b/packages/chat/src/send-run-mail.ts @@ -1,7 +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 } 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"; @@ -102,24 +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, - }); - } 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/webhook-triggers/src/launch.ts b/packages/webhook-triggers/src/launch.ts index 6fda59774..4a0331aca 100644 --- a/packages/webhook-triggers/src/launch.ts +++ b/packages/webhook-triggers/src/launch.ts @@ -17,7 +17,6 @@ import { deliverWhenRoutable, readDefinitionProjection, readFoldedBody, - recordAgentSessionForRun, WORKFLOW_SOURCE_ENTRY, } from "@corbits/workflows"; import { listVisibleOfferings, type DB } from "@intx/db"; @@ -238,25 +237,11 @@ 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, - }); - } catch (error) { - reportError(error, { - operation: "webhookTriggers.launch.recordAgentSession", - tenantId: trigger.tenantId, - agentId: prepared.deploymentAddress, - extra: { instanceId: prepared.anchorRunId }, - }); - } + // 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 { instanceId: prepared.anchorRunId, diff --git a/packages/webhook-triggers/test/launch.test.ts b/packages/webhook-triggers/test/launch.test.ts index 177a5090b..fa7559086 100644 --- a/packages/webhook-triggers/test/launch.test.ts +++ b/packages/webhook-triggers/test/launch.test.ts @@ -88,25 +88,7 @@ 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", - }), - }, }, - insert: (_table: unknown) => ({ - values: (_values: unknown) => ({ - onConflictDoNothing: async (_opts: unknown) => {}, - }), - }), }; } diff --git a/packages/workflows/src/index.ts b/packages/workflows/src/index.ts index 7eac44dbb..74828b7b2 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, endAgentSessionForPrincipal, endAgentSessionForRun, - type RecordAgentSessionParams, + type EventCollectorPort, } from "./launch/agent-session"; export { deliverWhenRoutable, diff --git a/packages/workflows/src/launch/agent-session.ts b/packages/workflows/src/launch/agent-session.ts index f95c53960..0e2cbff99 100644 --- a/packages/workflows/src/launch/agent-session.ts +++ b/packages/workflows/src/launch/agent-session.ts @@ -13,50 +13,82 @@ // 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. +// +// So `ensureRunSession` (CL-7480) is lazy and idempotent rather than +// called once right after a send: every seam that might be the run's +// first mail-routable moment (a hub mailbox persist, a session +// orchestrator dispatch with no live collector yet) calls it before +// doing its own work. A run with no principal yet is a normal, expected +// state, not a bug — this returns `null` rather than throwing. 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 + * `eventCollectors`) — never a second registry construction. `create` + * 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; `has` lets a caller check whether a collector already exists + * before creating a second one. + */ +export type EventCollectorPort = Pick< + EventCollectorRegistry, + "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). + * Idempotently makes a run's session and event collector exist, the + * instant that becomes possible: the run's `tenantId`, `definitionId` + * (the `agent_id` FK), and `principalId` are all read fresh off the + * `workflow_run` row, never guessed by the caller. Returns `null` (and + * writes nothing) when the row has no principal yet — an un-anchored + * run is not a bug, just not ensurable yet. Returns the run's + * `sessionId` once the session exists (freshly inserted or already + * there — a second call for the same run is a no-op, not a conflict), + * with the run's event collector created too if none was live for its + * address. */ -export async function recordAgentSessionForRun( - db: DB["db"], - params: RecordAgentSessionParams, -): Promise { +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: - "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; + return null; + } + + const launchSpecRow = await db.query.workflowRunLaunchSpec.findFirst({ + where: eq(workflowRunLaunchSpec.anchorRunId, runId), + }); + if (launchSpecRow === undefined) { + return null; } + const sessionId = launchSpecRow.sessionId; const now = new Date(); await db .insert(agentSession) .values({ - id: params.sessionId, + id: sessionId, tenantId: runRow.tenantId, agentId: runRow.definitionId, principalId: runRow.principalId, @@ -65,7 +97,16 @@ export async function recordAgentSessionForRun( updatedAt: now, }) .onConflictDoNothing({ target: agentSession.id }); - return true; + + if (runRow.address !== null && !eventCollectors.has(runRow.address)) { + eventCollectors.create( + runRow.address, + runRow.tenantId, + sessionId, + runRow.id, + ); + } + return sessionId; } /** @@ -87,14 +128,19 @@ export async function endAgentSessionForPrincipal( /** * Same as `endAgentSessionForPrincipal`, keyed by the run's own id - * instead — for a caller (e.g. a one-shot prompt's teardown) that only - * ever held the run id, never minted or looked up its principal. - * No-ops for a run row already gone, or one never triggered (no - * principal, hence no session was ever recorded for it). + * instead — for a caller (e.g. a one-shot prompt's teardown, or a chat + * relaunch replacing a terminal run) that only ever held the run id, + * never minted or looked up its principal. Also abandons the outgoing + * run's event collector, through the same shared + * `EventCollectorPort` every launcher threads — one mechanism for + * ending a run's session. No-ops for a run row already gone, or one + * never triggered (no principal, hence no session was ever recorded + * for it). */ export async function endAgentSessionForRun( db: DB["db"], anchorRunId: string, + eventCollectors: Pick, ): Promise { const runRow = await db.query.workflowRun.findFirst({ where: eq(workflowRun.id, anchorRunId), @@ -103,4 +149,7 @@ export async function endAgentSessionForRun( return; } await endAgentSessionForPrincipal(db, runRow.principalId); + if (runRow.address !== null) { + eventCollectors.abandon(runRow.address); + } } diff --git a/packages/workflows/test/agent-session.test.ts b/packages/workflows/test/agent-session.test.ts index a7446adda..de783406f 100644 --- a/packages/workflows/test/agent-session.test.ts +++ b/packages/workflows/test/agent-session.test.ts @@ -2,11 +2,11 @@ // 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-7477/CL-7480: nothing writes `agent_session` for a native launch +// until something calls `ensureRunSession` — this proves it writes a +// row `resolveRunSessionId` (vendor's mail-routing lookup) actually +// reads back, only once the run is anchored with a principal, and stays +// idempotent (one insert, one collector) across repeated calls. import { afterAll, beforeAll, expect, test } from "bun:test"; import { eq } from "drizzle-orm"; @@ -15,7 +15,7 @@ import { generateId } from "@intx/hub-common"; import { resolveRunSessionId } from "@intx/hub-sessions"; import { dbGate } from "../../../scripts/e2e/db-gate"; -import { recordAgentSessionForRun } from "../src/launch/agent-session"; +import { ensureRunSession } from "../src/launch/agent-session"; function dbConfigFromUrl(databaseUrl: string) { const url = new URL(databaseUrl); @@ -31,14 +31,16 @@ function dbConfigFromUrl(databaseUrl: string) { const databaseUrl = process.env["DATABASE_URL"]; const describeIfDb = dbGate(databaseUrl, import.meta.path); -describeIfDb("recordAgentSessionForRun", () => { +describeIfDb("ensureRunSession", () => { let db: DB; const tenantId = generateId("tenant"); const principalId = generateId("principal"); const definitionId = generateId("workflowDefinition"); - const anchorRunId = generateId("workflowRun"); + const anchoredRunId = generateId("workflowRun"); + const unanchoredRunId = generateId("workflowRun"); const sessionId = generateId("session"); + const domain = `agent-session-${tenantId}.localhost`; beforeAll(async () => { if (databaseUrl === undefined) return; @@ -48,7 +50,7 @@ describeIfDb("recordAgentSessionForRun", () => { id: tenantId, name: "Agent Session Test Tenant", slug: `agent-session-${tenantId}`, - domain: `agent-session-${tenantId}.localhost`, + domain, parentId: null, config: null, }); @@ -57,7 +59,7 @@ describeIfDb("recordAgentSessionForRun", () => { id: principalId, tenantId, kind: "workflow", - refId: anchorRunId, + refId: anchoredRunId, status: "active", }); @@ -69,13 +71,38 @@ describeIfDb("recordAgentSessionForRun", () => { name: "agent-session-test-definition", }); + // Anchored: has a principal, so `ensureRunSession` can insert its + // `agent_session` row. await db.db.insert(schema.workflowRun).values({ - id: anchorRunId, + id: anchoredRunId, definitionId, - anchorRunId, + anchorRunId: anchoredRunId, tenantId, principalId, - address: `${anchorRunId}@agent-session-${tenantId}.localhost`, + address: `${anchoredRunId}@${domain}`, + status: "running", + }); + await db.db.insert(schema.workflowRunLaunchSpec).values({ + anchorRunId: anchoredRunId, + sessionId, + deploymentDomain: domain, + sourceAuthorityPrincipalId: principalId, + frozenApprovalBundle: {}, + sourceOfferingIds: [], + defaultSourceOfferingId: "off_test", + deployContent: { systemPrompt: "" }, + }); + + // Unanchored: `prepareProvisionedDeployment` minted the row, but no + // trigger has reconciled a principal onto it yet — the normal state + // for a freshly provisioned invite. + await db.db.insert(schema.workflowRun).values({ + id: unanchoredRunId, + definitionId, + anchorRunId: unanchoredRunId, + tenantId, + principalId: null, + address: `${unanchoredRunId}@${domain}`, status: "running", }); }); @@ -85,9 +112,15 @@ 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, anchoredRunId)); + await db.db + .delete(schema.workflowRun) + .where(eq(schema.workflowRun.id, anchoredRunId)); await db.db .delete(schema.workflowRun) - .where(eq(schema.workflowRun.id, anchorRunId)); + .where(eq(schema.workflowRun.id, unanchoredRunId)); await db.db .delete(schema.workflowDefinition) .where(eq(schema.workflowDefinition.id, definitionId)); @@ -97,8 +130,54 @@ describeIfDb("recordAgentSessionForRun", () => { 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 }); + test("an unanchored run returns null and writes nothing", async () => { + const createCalls: unknown[] = []; + const resolved = await ensureRunSession({ + db: db.db, + eventCollectors: { + create: (...args: unknown[]) => { + createCalls.push(args); + }, + has: () => false, + }, + runId: unanchoredRunId, + }); + + expect(resolved).toBeNull(); + expect(createCalls).toEqual([]); + const sessionRow = await db.db.query.agentSession.findFirst({ + where: eq(schema.agentSession.agentId, definitionId), + }); + expect(sessionRow).toBeUndefined(); + }); + + test("an anchored run inserts its session and creates its collector once across two calls", async () => { + const createCalls: unknown[] = []; + let hasCollector = false; + const eventCollectors = { + create: (...args: unknown[]) => { + createCalls.push(args); + hasCollector = true; + }, + has: () => hasCollector, + }; + + const first = await ensureRunSession({ + db: db.db, + eventCollectors, + runId: anchoredRunId, + }); + const second = await ensureRunSession({ + db: db.db, + eventCollectors, + runId: anchoredRunId, + }); + + expect(first).toBe(sessionId); + expect(second).toBe(sessionId); + expect(createCalls).toEqual([ + [`${anchoredRunId}@${domain}`, tenantId, sessionId, anchoredRunId], + ]); const resolved = await resolveRunSessionId(db.db, principalId); expect(resolved).toBe(sessionId);