diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index b7c5c51477..d32fb65d15 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -989,10 +989,24 @@ export function createHubChatPlatform( * more time regardless. Exhausting every delay means the condition * is not transient and the caller gets a clean `AgentUnreachableError` * rather than an unhandled 500. + * + * CL-7486: a wake this loop runs between retries 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, + * and the wake ahead of the NEXT retry would fail outright (no + * `workbench_launch` row still names that address at all) instead of + * the "unreachable" this loop knows how to ride out. `reresolve` + * re-reads the live address (and its session) off the stable + * participant id before every wake/retry, so this loop always chases + * whichever run is actually current. */ async function sendRunMailWithReclaimRetry( params: SendRunMailParams, + reresolve: () => Promise<{ agentAddress: string; sessionId: string }>, ): Promise>> { + let current = params; let loggedRetryStart = false; for (let attempt = 0; ; attempt++) { try { @@ -1001,29 +1015,44 @@ export function createHubChatPlatform( // to accept regardless — the signal parameter is unused here, // same abandon-on-timeout behavior as before. return await withTimeout( - () => sendRunMail(runMailDeps, params), + () => sendRunMail(runMailDeps, current), MAIL_DELIVERY_TIMEOUT_MS, - `mail to ${params.agentAddress} did not settle within ${String(MAIL_DELIVERY_TIMEOUT_MS)}ms`, + `mail to ${current.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`; + wakeLogger.info`mail to ${current.agentAddress} exhausted every reclaim retry; giving up`; } if (isAgentUnreachable(err)) { - throw new AgentUnreachableError(params.agentAddress, { + throw new AgentUnreachableError(current.agentAddress, { cause: err, }); } throw err; } if (!loggedRetryStart) { - wakeLogger.info`mail to ${params.agentAddress} hit "agent is unreachable"; retrying with backoff while the post-restart reclaim settles`; + wakeLogger.info`mail to ${current.agentAddress} hit "agent is unreachable"; retrying with backoff while the post-restart reclaim settles`; loggedRetryStart = true; } await sleep(delay); - await wakeByAddressBounded(params.agentAddress); + // Wake whatever this loop still believes is live — `wakeByAddress` + // itself tolerates an address one relaunch behind the room's + // current one (its lookup matches either the stable participant + // id or the live run id) — and only THEN re-resolve, since the + // wake just run is exactly the kind of relaunch that can leave + // `current.agentAddress` pointing at a run that no longer exists + // at all. Re-resolving before the wake instead would still hand + // the next retry whatever address was current a moment ago, + // stale again the instant this wake relaunches it. + await wakeByAddressBounded(current.agentAddress); + const fresh = await reresolve(); + current = { + ...current, + agentAddress: fresh.agentAddress, + sessionId: fresh.sessionId, + }; } } } @@ -1399,7 +1428,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 120538c926..89ad655d58 100644 --- a/packages/chat/test/platform-adapter.test.ts +++ b/packages/chat/test/platform-adapter.test.ts @@ -2781,6 +2781,139 @@ 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], + }); + + const platform = createPlatform({ + toolGrantsForPins: () => [], + db: db as never, + sessionService, + sidecarRouter, + eventCollectors: createFakeEventCollectors(), + reclaimRetryDelaysMs: [1], + 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 301bdd6c64..d6eebbd36d 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/webhook-triggers/test/launch.test.ts b/packages/webhook-triggers/test/launch.test.ts index dad4716158..e671da0093 100644 --- a/packages/webhook-triggers/test/launch.test.ts +++ b/packages/webhook-triggers/test/launch.test.ts @@ -88,35 +88,28 @@ 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, - address: INTERCHANGE_ADDRESS, - principalId: null, + principalId: "prn_interchange", }), }, }, - insert: () => ({ - values: () => ({ - onConflictDoNothing: async () => undefined, + insert: (_table: unknown) => ({ + values: (_values: unknown) => ({ + onConflictDoNothing: async (_opts: unknown) => {}, }), }), }; } -let eventCollectorCreateCalls: unknown[] = []; - -function createFakeEventCollectors() { - return { - create: (...args: unknown[]) => { - eventCollectorCreateCalls.push(args); - }, - has: () => false, - }; -} - const TRIGGER = { id: "wht_1", tenantId: "ten_1", @@ -204,7 +197,10 @@ function baseDeps() { }, }, isRoutable: () => isRoutableForTest, - eventCollectors: createFakeEventCollectors(), + eventCollectors: { + create: () => undefined, + has: () => false, + }, }; } @@ -216,7 +212,6 @@ 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 eefa55a850..64bfbdff98 100644 --- a/packages/workflows/src/index.ts +++ b/packages/workflows/src/index.ts @@ -56,8 +56,8 @@ export { } from "./schedule/cron"; export { scheduleCronFromProjection } from "./schedule/from-projection"; export { - ensureRunSession, recordAgentSessionAtProvision, + ensureRunSession, endAgentSessionForPrincipal, endAgentSessionForRun, type EventCollectorPort,