diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index b7c5c51477..745ac214f3 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 120538c926..c9583847ff 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 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/workflows/src/deliver-when-routable.ts b/packages/workflows/src/deliver-when-routable.ts index a48e994038..cee443ed92 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 eefa55a850..499914e5ee 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 {