diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index b7c5c5147..745ac214f 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -16,8 +16,10 @@ import { and, desc, eq } from "drizzle-orm"; import { authoredDefinitionCandidates, DefinitionProjectionMissingError, + deliverWhenRoutable, endAgentSessionForRun, ensureRunSession, + isAgentUnreachableError, readFoldedBody, recordAgentSessionAtProvision, resolveNewestProjectedDefinition, @@ -139,13 +141,6 @@ export type CreateHubChatPlatformDeps = { * `ensureAwake` to redeploy a non-routable target before sending. */ lifecycle?: { idleSleepMs: number; sweepIntervalMs?: number }; - /** - * `sendRunMailWithReclaimRetry`'s backoff between retries of a - * mail send that failed with "agent is unreachable" (see - * RECLAIM_RETRY_DELAYS_MS below). Injectable so tests exercise the - * backoff in milliseconds instead of the production ~8s budget. - */ - reclaimRetryDelaysMs?: readonly number[]; /** * Per-attempt wall-clock bound on `sendRunMail` inside * `sendRunMailWithReclaimRetry` (CL-6644). Defaults to @@ -153,6 +148,16 @@ export type CreateHubChatPlatformDeps = { * in milliseconds instead of the production ~30s budget. */ mailDeliveryTimeoutMs?: number; + /** + * How long `sendRunMailWithReclaimRetry` waits for a freshly + * woken/relaunched run's address to show up in the hub's routing + * table before giving up (see `deliverWhenRoutable` in + * `@corbits/workflows`). Defaults to `DEFAULT_ROUTABLE_DEADLINE_MS` + * — the same boot-and-register budget the webhook trigger's ingress + * delivery waits on (CL-7476). Injectable so tests exercise the wait + * in milliseconds instead of the production ~20s budget. + */ + routableWaitDeadlineMs?: number; /** * The invite-launch model fallback (see `./inference-preferences.ts`'s * `createWorkbenchHostInferencePreferencesResolver`): a @@ -409,35 +414,17 @@ export function createHubChatPlatform( const lifecycle = deps.lifecycle !== undefined ? buildLifecycle(deps.lifecycle) : undefined; - // Bounded backoff for a wake racing the sidecar's own post-restart - // reclaim, and separately for a mail delivery racing the same - // window: 250ms, 500ms, 1s, 2s, 4s — a ~7.75s budget, long enough - // for a normal reconnect challenge to settle without leaving a - // sender stuck for much longer than that. - const RECLAIM_RETRY_DELAYS_MS = deps.reclaimRetryDelaysMs ?? [ - 250, 500, 1000, 2000, 4000, - ]; - - // CL-6644: the reclaim-retry loop's own delays only run between - // attempts that already failed LOUD with "agent is unreachable" -- - // they bound nothing about an attempt that instead stalls forever - // (a sidecar ack that never comes, a wedged promise anywhere in - // `sendRunMail`'s call chain) without ever throwing. Each - // attempt gets the same kind of wall-clock bound `wakeByAddressBounded` - // already puts on the wake itself, so that hang becomes a rejection + // CL-6644: a per-attempt bound on `sendRunMail` inside + // `sendRunMailWithReclaimRetry` -- a send that stalls forever (a + // sidecar ack that never comes, a wedged promise anywhere in + // `sendRunMail`'s call chain) without ever throwing gets the same + // kind of wall-clock bound `wakeByAddressBounded` already puts on + // the wake itself, so that hang becomes a rejection // `dispatchTurnBatch`'s catch can report and notify on, instead of a // promise nothing ever settles. const MAIL_DELIVERY_TIMEOUT_MS = deps.mailDeliveryTimeoutMs ?? DEFAULT_WAKE_TIMEOUT_MS; - function sleep(ms: number): Promise { - return new Promise((resolveSleep) => setTimeout(resolveSleep, ms)); - } - - function isAgentUnreachable(err: unknown): boolean { - return err instanceof Error && err.message.includes("agent is unreachable"); - } - function isRoutable(address: string): boolean { return deps.sidecarRouter.getRoutableAddresses().includes(address); } @@ -977,54 +964,89 @@ export function createHubChatPlatform( } /** - * `sendRunMail` delivers synchronously against the sidecar's - * current routable set — the same in-memory index `isRoutable` reads - * — so a send that lands in the same post-restart reclaim window - * `wakeByAddress` above tolerates can still fail with "agent is - * unreachable" even right after a successful (or no-op) wake. Each - * retry forces a fresh wake first: if the earlier reclaim tore the - * agent down, this becomes the genuine redeploy that recovers it; if - * the reclaim is still in flight, the wake itself waits it out (or - * redeploys once its own budget is exhausted) and the delay gives it - * more time regardless. Exhausting every delay means the condition - * is not transient and the caller gets a clean `AgentUnreachableError` - * rather than an unhandled 500. + * `sendRunMail` delivers synchronously against the sidecar's current + * routable set — the same in-memory index `isRoutable` reads — so a + * send lands in "agent is unreachable" for two different reasons + * that both need the same treatment: a run whose sidecar just + * finished a provisioned deploy and hasn't registered yet (CL-7476 — + * `deliverWhenRoutable` from `@corbits/workflows` is the exact same + * wait the webhook trigger's ingress delivery already uses for this), + * and a run that this loop's own wake tore down and is mid-reclaim. + * `deliverWhenRoutable` sends once, and — only on an unreachable + * failure — polls the routing table for `current.agentAddress` up to + * its shared deadline (`DEFAULT_ROUTABLE_DEADLINE_MS`, the same + * boot-and-register budget the webhook path waits on) before sending + * exactly once more. + * + * CL-7486: a wake this loop runs on the first failure can itself + * relaunch the run — `wakeByAddress`'s dead-run branch repoints the + * room's stable participant at a fresh address without telling this + * loop. Retrying `sendRunMail` against the address it started with + * would then keep hammering a run that will never become routable + * again. `reresolve` re-reads the live address (and its session) off + * the stable participant id every time `send` runs — including the + * one `deliverWhenRoutable` runs again after the routing-table wait — + * so this loop always chases whichever run is actually current, even + * one relaunched again while it was waiting. */ async function sendRunMailWithReclaimRetry( params: SendRunMailParams, + reresolve: () => Promise<{ agentAddress: string; sessionId: string }>, ): Promise>> { - let loggedRetryStart = false; - for (let attempt = 0; ; attempt++) { - try { - // CL-7193: `sendRunMail` does a DB write plus a sidecar - // delivery that shouldn't be half-cancelled, and has no signal - // to accept regardless — the signal parameter is unused here, - // same abandon-on-timeout behavior as before. - return await withTimeout( - () => sendRunMail(runMailDeps, params), - MAIL_DELIVERY_TIMEOUT_MS, - `mail to ${params.agentAddress} did not settle within ${String(MAIL_DELIVERY_TIMEOUT_MS)}ms`, - ); - } catch (err) { - const delay = RECLAIM_RETRY_DELAYS_MS[attempt]; - if (!isAgentUnreachable(err) || delay === undefined) { - if (loggedRetryStart) { - wakeLogger.info`mail to ${params.agentAddress} exhausted every reclaim retry; giving up`; - } - if (isAgentUnreachable(err)) { - throw new AgentUnreachableError(params.agentAddress, { - cause: err, - }); + let current = params; + let wakedOnFailure = false; + async function sendCurrent() { + // CL-7193: `sendRunMail` does a DB write plus a sidecar delivery + // that shouldn't be half-cancelled, and has no signal to accept + // regardless — the signal parameter is unused here, same + // abandon-on-timeout behavior as before. + return withTimeout( + () => sendRunMail(runMailDeps, current), + MAIL_DELIVERY_TIMEOUT_MS, + `mail to ${current.agentAddress} did not settle within ${String(MAIL_DELIVERY_TIMEOUT_MS)}ms`, + ); + } + try { + return await deliverWhenRoutable({ + send: async () => { + try { + return await sendCurrent(); + } catch (err) { + if (!isAgentUnreachableError(err) || wakedOnFailure) throw err; + // Wake exactly once, the moment the first attempt reports + // unreachable: if the run had already died, this is the + // genuine redeploy that recovers it; if a deploy is merely + // still booting, `wakeByAddress`/`lifecycle.ensureAwake` + // no-op past the routability check and this becomes a + // cheap reconcile. Re-resolving only after the wake means a + // relaunch it just performed is exactly what the fresh read + // picks up, rather than the address that was current a + // moment before. + wakeLogger.info`mail to ${current.agentAddress} hit "agent is unreachable"; waiting for it to become routable`; + wakedOnFailure = true; + await wakeByAddressBounded(current.agentAddress); + const fresh = await reresolve(); + current = { + ...current, + agentAddress: fresh.agentAddress, + sessionId: fresh.sessionId, + }; + throw err; } - throw err; - } - if (!loggedRetryStart) { - wakeLogger.info`mail to ${params.agentAddress} hit "agent is unreachable"; retrying with backoff while the post-restart reclaim settles`; - loggedRetryStart = true; - } - await sleep(delay); - await wakeByAddressBounded(params.agentAddress); + }, + isRoutable: () => isRoutable(current.agentAddress), + ...(deps.routableWaitDeadlineMs !== undefined + ? { deadlineMs: deps.routableWaitDeadlineMs } + : {}), + }); + } catch (err) { + if (isAgentUnreachableError(err)) { + wakeLogger.info`mail to ${current.agentAddress} exhausted the routable-wait budget; giving up`; + throw new AgentUnreachableError(current.agentAddress, { + cause: err, + }); } + throw err; } } @@ -1399,7 +1421,30 @@ export function createHubChatPlatform( ? { references: input.content.references } : {}), }; - const sent = await sendRunMailWithReclaimRetry(withThreading); + const sent = await sendRunMailWithReclaimRetry( + withThreading, + async () => { + const relive = await requireLive(input.workbenchId, input.tenantId); + try { + await ensureRunSession({ + db: deps.db, + eventCollectors: deps.eventCollectors, + runId: relive.run.id, + }); + } catch (err) { + reportError(err, { + operation: "chat.sendMail.ensureRunSession", + tenantId: input.tenantId, + agentId: relive.binding.liveAddress, + extra: { runId: relive.run.id }, + }); + } + return { + agentAddress: relive.binding.liveAddress, + sessionId: await resolveRunSessionIdOrThrow(deps.db, relive.run), + }; + }, + ); lifecycle?.recordActivity(deliveryAddress); diff --git a/packages/chat/test/platform-adapter.test.ts b/packages/chat/test/platform-adapter.test.ts index 120538c92..c9583847f 100644 --- a/packages/chat/test/platform-adapter.test.ts +++ b/packages/chat/test/platform-adapter.test.ts @@ -2726,6 +2726,7 @@ describe("createHubChatPlatform", () => { // in flight when the concurrent `ensureAwake` call joins it. let deployCallCount = 0; const gatedDeployReleases: (() => void)[] = []; + const sidecarRouter = createFakeSidecarRouter({ routableAddresses: [] }); const allocationService = createFakeWorkflowAllocationService(); const originalPrepare = allocationService.prepareProvisionedDeployment.bind(allocationService); @@ -2738,6 +2739,14 @@ describe("createHubChatPlatform", () => { await new Promise((resolve) => { gatedDeployReleases.push(resolve); }); + // Models the sidecar registering once the reclaim retry's own + // redeploy actually completes — the routable-wait + // `sendRunMailWithReclaimRetry` now does (CL-7488) needs this + // to see the retried send through. The cold wake ahead of the + // first send attempt deliberately stays unrouted so that + // attempt still finds `wakeByAddress`'s own routability check + // false and takes the reclaim-retry path this test exercises. + sidecarRouter.routableAddresses.push(result.deploymentAddress); } return result; }; @@ -2746,10 +2755,10 @@ describe("createHubChatPlatform", () => { toolGrantsForPins: () => [], db: db as never, sessionService, - sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), + sidecarRouter, eventCollectors: createFakeEventCollectors(), lifecycle: { idleSleepMs: 60_000 }, - reclaimRetryDelaysMs: [1], + routableWaitDeadlineMs: 5_000, mailDeliveryTimeoutMs: 5_000, workflowAllocationService: allocationService, }); @@ -2781,6 +2790,156 @@ describe("createHubChatPlatform", () => { }); }); + // CL-7486: `sendRunMailWithReclaimRetry` used to retry every attempt + // against the address it started with, even after a wake in between + // attempts relaunched the run. A run already one relaunch generation + // in (`instanceId` stable, `currentRunId` already repointed once) that + // dies again while this loop is retrying reproduces the production + // failure exactly: a second relaunch repoints `currentRunId` again, + // and the address this loop still holds is now neither the stable + // `instanceId` nor the current `currentRunId` — invisible to + // `readBindingByAddressAnyTenant`, so the next wake for it throws "No + // workbench_launch binding" instead of finding the run that replaced + // it. + describe("sendRunMailWithReclaimRetry re-resolves the live address", () => { + test("a relaunch between retries is chased to its new address, not retried against the one that died", 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", + }; + + const staleAddress = "run_gen1@ten1.workbench.test"; + const db = createFakeDb({ + assetRow: { + tenantId: "ten_1", + creatorPrincipalId: "prin_creator", + name: "ins_room1", + displayName: null, + }, + definitionId: "wfd_room1", + workflowRunRow: { + id: "run_gen1", + address: staleAddress, + principalId: "prin_room1", + definitionId: "wfd_room1", + // Already dead by the time the reclaim retry's own wake reads + // it — the run died again while this loop was mid-retry, not + // at the top of `sendMail` (which would have relaunched it + // before ever reaching this retry loop at all). + status: "failed", + }, + workflowDefinitionRow: { + id: "wfd_room1", + tenantId: "ten_1", + status: "deployed", + origin: "authored", + assetId: "asst_room1", + }, + workbenchLaunchRow: { + tenantId: "ten_1", + // `instanceId` is the room's own stable id, already distinct + // from `currentRunId` — this participant has been relaunched + // once before this test even starts. + instanceId: "ins_room1", + currentRunId: "run_gen1", + foldedBody: { + systemPrompt: "host prompt", + model: "claude-sonnet-5", + toolPackagePins: [], + grantRequirements: [], + credentialBindings: [], + }, + }, + }); + db.inserted.push({ + table: agentSession, + values: { id: "ses_run1", principalId: "prin_room1" }, + }); + + const sessionService = createFakeSessionService(); + sessionService.sendUserMessage = async (params: unknown) => { + sessionService.sendUserMessageCalls.push(params); + const { agentAddress } = params as { agentAddress: string }; + // The dead run never becomes reachable again — only a send to + // whatever address the relaunch actually produced can succeed. + if (agentAddress === staleAddress) { + throw new Error("agent is unreachable"); + } + return new TextEncoder().encode("raw-mime-bytes"); + }; + + // Routable so `sendMail`'s own opening wake-before-send gate skips + // waking it — the relaunch this test cares about must happen + // inside the reclaim-retry loop below, not before it. + const sidecarRouter = createFakeSidecarRouter({ + routableAddresses: [staleAddress], + }); + + // Models the sidecar registering the relaunch's fresh address the + // moment its deploy actually completes — `sendRunMailWithReclaimRetry` + // now waits on this routing-table transition (CL-7488) instead of a + // fixed backoff, so the retried send needs it to flip before it can + // see the run through. + const allocationService = createFakeWorkflowAllocationService(); + const originalPrepare = + allocationService.prepareProvisionedDeployment.bind(allocationService); + allocationService.prepareProvisionedDeployment = async ( + params: Parameters[0], + ) => { + const result = await originalPrepare(params); + sidecarRouter.routableAddresses.push(result.deploymentAddress); + return result; + }; + + const platform = createPlatform({ + toolGrantsForPins: () => [], + db: db as never, + sessionService, + sidecarRouter, + eventCollectors: createFakeEventCollectors(), + workflowAllocationService: allocationService, + routableWaitDeadlineMs: 5_000, + mailDeliveryTimeoutMs: 5_000, + }); + + const sent = await platform.sendMail({ + tenantId: "ten_1", + workbenchId: "ins_room1", + principalId: "prin_sender", + content: { content: "hello" }, + }); + + expect(sent.id).toBeTruthy(); + expect(sessionService.sendUserMessageCalls).toHaveLength(2); + const [first, second] = sessionService.sendUserMessageCalls as { + agentAddress: string; + }[]; + expect(first?.agentAddress).toBe(staleAddress); + // The retry that follows the relaunch targets the run that + // replaced `run_gen1`, never the dead address the loop started + // with. + expect(second?.agentAddress).not.toBe(staleAddress); + + const repointed = db.updated.at(-1)?.values as { + currentRunId: string; + priorRunIds: string[]; + }; + const secondAddress = second?.agentAddress ?? ""; + expect(repointed.currentRunId).toBe(secondAddress.split("@")[0] ?? ""); + expect(repointed.priorRunIds).toEqual(["run_gen1"]); + }); + }); + // Proves the actual lever an edited system prompt reaches a running // instance through: `wakeByAddress` (exercised via `sendMail`'s // wake-on-send path above) replays `workbench_launch.foldedBody` diff --git a/packages/tool-registry-publish/src/publish.ts b/packages/tool-registry-publish/src/publish.ts index 301bdd6c6..d6eebbd36 100644 --- a/packages/tool-registry-publish/src/publish.ts +++ b/packages/tool-registry-publish/src/publish.ts @@ -110,7 +110,19 @@ async function ensureRegistryAsset( if (created.status === 201) { return parseSchema(AssetResponse, created.data, "create asset response").id; } - if (created.status !== 409) { + // CL-7486: two overlapping seed runs for the same tenant (the root + // tenant's own boot-time seed racing a second caller seeding that + // same tenant) can both pass the `listRegistryAsset` check above + // before either has created the row, then both post-create — the + // loser hits the `asset_tenant_kind_name` unique constraint. The + // substrate answers that as `409` in the common case; asserted here + // by name too, in case a raw constraint violation surfaces as a + // generic 500 instead, so the loser still resolves the winner's row + // rather than throwing on a race this ensure is meant to tolerate. + const isNameCollision = + created.status === 409 || + JSON.stringify(created.data).includes("asset_tenant_kind_name"); + if (!isNameCollision) { throw new Error( `publishCorbitsToolsRegistry: failed to create the ${CORBITS_TOOLS_REGISTRY} asset: ${String(created.status)}`, ); diff --git a/packages/workflows/src/deliver-when-routable.ts b/packages/workflows/src/deliver-when-routable.ts index a48e99403..cee443ed9 100644 --- a/packages/workflows/src/deliver-when-routable.ts +++ b/packages/workflows/src/deliver-when-routable.ts @@ -3,9 +3,9 @@ // sidecar has finished booting and registering with the hub, so the send // fails with "agent is unreachable" even though nothing is actually wrong // — the address just isn't routable yet. The webhook trigger's ingress -// delivery goes through this helper; chat keeps its own existing -// wake-and-retry loop (`sendRunMailWithReclaimRetry`) instead. -const DEFAULT_DEADLINE_MS = 20_000; +// delivery goes through this helper; chat's `sendRunMailWithReclaimRetry` +// (CL-7488) shares this same budget rather than declaring its own. +export const DEFAULT_ROUTABLE_DEADLINE_MS = 20_000; const DEFAULT_POLL_INTERVAL_MS = 250; export function isAgentUnreachableError(err: unknown): boolean { @@ -48,7 +48,8 @@ export async function deliverWhenRoutable( if (!isUnreachable(err)) { throw err; } - const deadline = Date.now() + (opts.deadlineMs ?? DEFAULT_DEADLINE_MS); + const deadline = + Date.now() + (opts.deadlineMs ?? DEFAULT_ROUTABLE_DEADLINE_MS); const pollIntervalMs = opts.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS; const sleep = opts.sleep ?? defaultSleep; while (!opts.isRoutable()) { diff --git a/packages/workflows/src/index.ts b/packages/workflows/src/index.ts index eefa55a85..499914e5e 100644 --- a/packages/workflows/src/index.ts +++ b/packages/workflows/src/index.ts @@ -65,6 +65,7 @@ export { export { deliverWhenRoutable, isAgentUnreachableError, + DEFAULT_ROUTABLE_DEADLINE_MS, type DeliverWhenRoutableOptions, } from "./deliver-when-routable"; export { diff --git a/packages/workflows/src/launch/agent-session.ts b/packages/workflows/src/launch/agent-session.ts index f73c2a0d0..e4c972ef4 100644 --- a/packages/workflows/src/launch/agent-session.ts +++ b/packages/workflows/src/launch/agent-session.ts @@ -33,9 +33,17 @@ // 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. +// `ensureRunSession` re-keys onto the run's own principal once one is +// anchored, moving `principal_id` without ever touching the session id. +// +// That eager write only ever runs for a Workbench launcher, though — a +// run deployed straight through Interchange's own +// `POST /api/tenants/:id/workflows/deployments` route (the e2e suites, +// or any other API client) never passes through one, so no +// `agent_session` row exists for it until something creates one. +// CL-7489 makes `ensureRunSession` a true upsert: by the time +// persist/dispatch call it the run is anchored, so a missing row is +// created rather than treated as a bug. import { eq } from "drizzle-orm"; import type { DB } from "@intx/db"; import { @@ -118,18 +126,24 @@ export async function recordAgentSessionAtProvision(params: { } /** - * 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). + * True upsert of a run's `agent_session`, keyed by the run's own + * principal once Interchange anchors one (`anchorWithPrincipal`, the + * run's first trigger). Before that, `recordAgentSessionAtProvision` + * records the row under the deploying principal for a Workbench-launched + * run — but a run deployed straight through Interchange's own + * `POST /api/tenants/:id/workflows/deployments` route (the e2e suites, + * or any other API client) never passes through a Workbench launcher, so + * that eager write never runs for it, and no `agent_session` row exists + * yet when this seam is first reached. By the time persist/dispatch call + * this, though, the run is anchored (`runRow.principalId` is set), so + * everything needed to create the row is on hand: this inserts it rather + * than treating the missing row as a bug. The launch-spec row is still + * the source of truth for which session id 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"]; @@ -159,11 +173,20 @@ export async function ensureRunSession(params: { 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) { + const now = new Date(); + await db + .insert(agentSession) + .values({ + id: sessionId, + tenantId: runRow.tenantId, + agentId: runRow.definitionId, + principalId: runRow.principalId, + status: "active", + createdAt: now, + updatedAt: now, + }) + .onConflictDoNothing({ target: agentSession.id }); + } else if (sessionRow.principalId !== runRow.principalId) { await db .update(agentSession) .set({ principalId: runRow.principalId, updatedAt: new Date() }) diff --git a/packages/workflows/test/agent-session.test.ts b/packages/workflows/test/agent-session.test.ts index 6fe8dcfb6..7837ec3cb 100644 --- a/packages/workflows/test/agent-session.test.ts +++ b/packages/workflows/test/agent-session.test.ts @@ -7,10 +7,13 @@ // 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 +// `ensureRunSession` is a true upsert: 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 when a row +// already exists, that it creates the row from scratch for a run +// deployed straight through Interchange's own deployments route (no +// Workbench launcher ever ran `recordAgentSessionAtProvision` for it), +// 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"; @@ -48,7 +51,10 @@ describeIfDb("recordAgentSessionAtProvision / ensureRunSession", () => { const definitionId = generateId("workflowDefinition"); const provisionedRunId = generateId("workflowRun"); const noLaunchSpecRunId = generateId("workflowRun"); + const apiDeployedRunId = generateId("workflowRun"); + const apiDeployedPrincipalId = generateId("principal"); const sessionId = generateId("session"); + const apiDeployedSessionId = generateId("session"); const domain = `agent-session-${tenantId}.localhost`; beforeAll(async () => { @@ -79,6 +85,13 @@ describeIfDb("recordAgentSessionAtProvision / ensureRunSession", () => { refId: provisionedRunId, status: "active", }, + { + id: apiDeployedPrincipalId, + tenantId, + kind: "workflow", + refId: apiDeployedRunId, + status: "active", + }, ]); await db.db.insert(schema.workflowDefinition).values({ @@ -124,6 +137,32 @@ describeIfDb("recordAgentSessionAtProvision / ensureRunSession", () => { address: `${noLaunchSpecRunId}@${domain}`, status: "deployed", }); + + // Deployed through Interchange's own deployments route rather than a + // Workbench launcher: Interchange itself writes the launch spec (it + // mints the session id), but nothing ever calls + // `recordAgentSessionAtProvision` for this path, so no `agent_session` + // row exists yet. Already anchored with its own principal by the time + // `ensureRunSession` is reached, same as at persist/dispatch. + await db.db.insert(schema.workflowRun).values({ + id: apiDeployedRunId, + definitionId, + anchorRunId: apiDeployedRunId, + tenantId, + principalId: apiDeployedPrincipalId, + address: `${apiDeployedRunId}@${domain}`, + status: "running", + }); + await db.db.insert(schema.workflowRunLaunchSpec).values({ + anchorRunId: apiDeployedRunId, + sessionId: apiDeployedSessionId, + deploymentDomain: domain, + sourceAuthorityPrincipalId: deployingPrincipalId, + frozenApprovalBundle: {}, + sourceOfferingIds: [], + defaultSourceOfferingId: "off_test", + deployContent: { systemPrompt: "" }, + }); }); afterAll(async () => { @@ -140,12 +179,24 @@ describeIfDb("recordAgentSessionAtProvision / ensureRunSession", () => { await db.db .delete(schema.workflowRun) .where(eq(schema.workflowRun.id, noLaunchSpecRunId)); + await db.db + .delete(schema.agentSession) + .where(eq(schema.agentSession.id, apiDeployedSessionId)); + await db.db + .delete(schema.workflowRunLaunchSpec) + .where(eq(schema.workflowRunLaunchSpec.anchorRunId, apiDeployedRunId)); + await db.db + .delete(schema.workflowRun) + .where(eq(schema.workflowRun.id, apiDeployedRunId)); await db.db .delete(schema.workflowDefinition) .where(eq(schema.workflowDefinition.id, definitionId)); await db.db .delete(schema.principal) .where(eq(schema.principal.id, runPrincipalId)); + await db.db + .delete(schema.principal) + .where(eq(schema.principal.id, apiDeployedPrincipalId)); await db.db .delete(schema.principal) .where(eq(schema.principal.id, deployingPrincipalId)); @@ -215,6 +266,40 @@ describeIfDb("recordAgentSessionAtProvision / ensureRunSession", () => { expect(viaVendorLookup).toBe(sessionId); }); + test("creates the session and collector on first ensure for an API-deployed run", async () => { + const createCalls: unknown[] = []; + let hasCollector = false; + const eventCollectors = { + create: (...args: unknown[]) => { + createCalls.push(args); + hasCollector = true; + }, + has: () => hasCollector, + }; + + const resolved = await ensureRunSession({ + db: db.db, + eventCollectors, + runId: apiDeployedRunId, + }); + + expect(resolved).toBe(apiDeployedSessionId); + expect(createCalls).toEqual([ + [ + `${apiDeployedRunId}@${domain}`, + tenantId, + apiDeployedSessionId, + apiDeployedRunId, + ], + ]); + + const sessionRow = await db.db.query.agentSession.findFirst({ + where: eq(schema.agentSession.id, apiDeployedSessionId), + }); + expect(sessionRow?.principalId).toBe(apiDeployedPrincipalId); + expect(sessionRow?.agentId).toBe(definitionId); + }); + test("throws for a run with no launch spec row", async () => { await expect( ensureRunSession({