diff --git a/packages/chat/src/agent-binding.test.ts b/packages/chat/src/agent-binding.test.ts index f11d61a60f..e9967fb04f 100644 --- a/packages/chat/src/agent-binding.test.ts +++ b/packages/chat/src/agent-binding.test.ts @@ -186,12 +186,29 @@ describe("isBeyondWake", () => { ).toBe(false); }); - test("a completed run is beyond waking — wake is a fresh provision", async () => { + test("a completed run that took a turn is beyond waking — wake is a fresh provision", async () => { expect( await isBeyondWake(fakeDb([]), { id: "run_done", status: "completed", + principalId: "prin_1", }), ).toBe(true); }); + + // CL-7490: a deployment `markTerminal`'d before its first trigger ever + // landed settles "completed" too (see `workflow-run-store.ts`'s + // `markTerminal`), and the store cannot otherwise tell that apart from a + // run still mid-boot. Treating it as beyond waking here is exactly the + // bug that made a booting run's status read as terminal and relaunched + // it out from under its own in-flight sidecar registration. + test("a completed run that never took a turn is not beyond waking — the store cannot tell it apart from still booting", async () => { + expect( + await isBeyondWake(fakeDb([]), { + id: "run_never_triggered", + status: "completed", + principalId: null, + }), + ).toBe(false); + }); }); diff --git a/packages/chat/src/agent-binding.ts b/packages/chat/src/agent-binding.ts index 47b4d0d714..dd61aa6a21 100644 --- a/packages/chat/src/agent-binding.ts +++ b/packages/chat/src/agent-binding.ts @@ -314,27 +314,46 @@ export async function readPriorRuns( } /** - * The statuses a `workflow_run` can hold that mean "this run will never - * accept mail again". Wake of a reaped or lost instance is a fresh - * `prepareProvisionedDeployment`, not a reuse of this row. + * The statuses a `workflow_run` can hold that are terminal regardless of + * whether the run ever took a turn. `"completed"` is deliberately absent + * here — a deployment `markTerminal`'d before its first trigger settles + * "completed" too (see `workflow-run-store.ts`'s `markTerminal`), and that + * is a run that is still booting, not one that will "never accept mail + * again". `isBeyondWake` below folds `"completed"` back in only once it + * can tell the two apart. */ -const TERMINAL_RUN_STATUSES: ReadonlySet = new Set([ +const UNCONDITIONALLY_TERMINAL_RUN_STATUSES: ReadonlySet = new Set([ "failed", "cancelled", "canceled", - "completed", ]); /** - * Whether this run is dead: terminal in `workflow_run.status`. A run - * this returns true for cannot be woken — it can only be relaunched as - * a fresh provisioned deployment. + * Whether this run is dead: terminal in `workflow_run.status`, in a way + * that actually means "this run will never accept mail again" rather than + * "this deployment never got as far as its first trigger". A run this + * returns true for cannot be woken — it can only be relaunched as a fresh + * provisioned deployment. A run this returns false for — `"deployed"` or + * `"running"`, or `"completed"` with no principal ever anchored to it — is + * still live or still booting: waking it means waiting for it to become + * routable (`deliverWhenRoutable`), never relaunching it (CL-7490). + * + * `principalId` is the signal for "ever took a turn": `anchorWithPrincipal` + * sets it in the same update that flips `"deployed"` -> `"running"` on a + * deployment's first trigger, and a child run is born with one already. A + * `"completed"` run with no `principalId` can only be a deployment torn + * down before that first trigger ever landed — the store cannot tell that + * apart from a mid-boot deployment misreporting itself terminal, so it is + * treated as still booting rather than relaunched out from under a caller + * that is only waiting on it. */ export async function isBeyondWake( _db: DB["db"], - run: { id: string; status: string }, + run: { id: string; status: string; principalId?: string | null }, ): Promise { - return TERMINAL_RUN_STATUSES.has(run.status); + if (UNCONDITIONALLY_TERMINAL_RUN_STATUSES.has(run.status)) return true; + if (run.status !== "completed") return false; + return run.principalId !== null && run.principalId !== undefined; } /** diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index 745ac214f3..5e8f971812 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -733,35 +733,18 @@ export function createHubChatPlatform( return; } - if (live.run.definitionId === null) { - throw new Error( - `Cannot wake "${binding.stableId}": its run ${live.run.id} names no definition`, - ); - } - const definitionAssetId = await resolveDefinitionAssetId( - live.run.definitionId, - ); - if (definitionAssetId === undefined) { - throw new Error( - `Cannot wake "${binding.stableId}": definition ${live.run.definitionId} has no workflow asset`, - ); - } - const prepared = await provisionOnAsset({ - tenantId: binding.tenantId, - deploymentDomain: domainOf(binding.roomAddress), - sourceAuthorityPrincipalId: - await sourceAuthorityForAsset(definitionAssetId), - definitionAssetId, - foldedBody: binding.foldedBody, - }); - await repointBinding( - deps.db, - binding, - prepared.runId, - prepared.sourcesDigest, - ); - lifecycle?.untrack(binding.liveAddress); - lifecycle?.track(prepared.address); + // CL-7490: not terminal, and not (yet) routable. A provisioned + // anchor stays "deployed" until its first trigger — this is + // exactly what a run mid-boot looks like from here, and relaunching + // it out from under its own in-flight sidecar registration is what + // supersedes the allocation and restarts the boot forever (the + // "Deployed agent" line never appears; a fresh sidecar/allocation + // shows up every few seconds instead). There is nothing this + // function can safely do for a live-but-not-yet-routable run: the + // caller's own routable wait (`deliverWhenRoutable`, wired through + // `sendRunMailWithReclaimRetry`) is what actually resolves this, by + // polling the routing table until the boot finishes or its budget + // runs out. A no-op here is the correct "nothing more to do". } catch (error) { throw wrapWakeInferenceError(error); } diff --git a/packages/chat/test/platform-adapter.test.ts b/packages/chat/test/platform-adapter.test.ts index c9583847ff..47ff59d594 100644 --- a/packages/chat/test/platform-adapter.test.ts +++ b/packages/chat/test/platform-adapter.test.ts @@ -870,10 +870,15 @@ describe("createHubChatPlatform", () => { displayName: null, }, definitionId: "wfd_workbench1", + // CL-7490: `wakeByAddress` only ever redeploys a genuinely dead + // run now — a merely not-yet-routable one is booting and is + // waited on, never redeployed — so this run must be terminal for + // the deploy failure below to actually be reached. workflowRunRow: { id: "ins_workbench1", address: "ins_workbench1@ten1.workbench.test", principalId: "prin_run1", + status: "failed", }, workbenchLaunchRow: { tenantId: "ten_1", @@ -947,10 +952,13 @@ describe("createHubChatPlatform", () => { displayName: null, }, definitionId: "wfd_workbench1", + // CL-7490: same reasoning as above — the run must be terminal for + // `wakeByAddress` to attempt a redeploy at all. workflowRunRow: { id: "ins_workbench1", address: "ins_workbench1@ten1.workbench.test", principalId: "prin_run1", + status: "failed", }, workbenchLaunchRow: { tenantId: "ten_1", @@ -2028,7 +2036,12 @@ describe("createHubChatPlatform", () => { // when `deps.lifecycle` is configured, and that `sendMail` actually // redeploys a non-routable target before sending. describe("lifecycle wiring", () => { - test("sendMail wakes a non-routable workbench by redeploying before sending, then sends", async () => { + // CL-7490: a provisioned anchor stays "deployed" until its first + // trigger, so a run mid-boot looks exactly like this — live, not yet + // routable. `wakeByAddress` must not relaunch it; the run is waited + // on (`deliverWhenRoutable`, via `sendRunMailWithReclaimRetry`) until + // the sidecar finishes registering, then delivered once. + test("sendMail waits on a deployed, not-yet-routable workbench and delivers once it becomes routable, without relaunching it", async () => { resolveDefinitionSourcesResult = { ok: true, materials: [], @@ -2044,6 +2057,7 @@ describe("createHubChatPlatform", () => { defaultSource: "off_1", }; + const address = "ins_workbench1@ten1.workbench.test"; const db = createFakeDb({ assetRow: { tenantId: "ten_1", @@ -2054,9 +2068,10 @@ describe("createHubChatPlatform", () => { definitionId: "wfd_workbench1", workflowRunRow: { id: "ins_workbench1", - address: "ins_workbench1@ten1.workbench.test", - principalId: "prin_run1", + address, + principalId: null, definitionId: "wfd_workbench1", + status: "deployed", }, workflowDefinitionRow: { id: "wfd_workbench1", @@ -2083,9 +2098,22 @@ describe("createHubChatPlatform", () => { }); const sessionService = createFakeSessionService(); - // Not in the sidecar's routable set: this workbench is asleep (or - // never came back after a restart) when the send arrives. + let sendAttempts = 0; + sessionService.sendUserMessage = async (params: unknown) => { + sessionService.sendUserMessageCalls.push(params); + sendAttempts += 1; + // The first attempt lands while the sidecar is still booting. + if (sendAttempts === 1) { + throw new Error("agent is unreachable"); + } + return new TextEncoder().encode("raw-mime-bytes"); + }; + // Not routable yet — the sidecar is still finishing its boot; + // it registers a moment later, on its own, with no redeploy. const sidecarRouter = createFakeSidecarRouter({ routableAddresses: [] }); + setTimeout(() => { + sidecarRouter.routableAddresses.push(address); + }, 10); const eventCollectors = createFakeEventCollectors(); const platform = createPlatform({ @@ -2105,8 +2133,10 @@ describe("createHubChatPlatform", () => { }); expect(sent.id).toBeTruthy(); - expect(lastAllocationService.prepareCalls).toHaveLength(1); - expect(sessionService.sendUserMessageCalls).toHaveLength(1); + // No relaunch: the booting run kept its own run id and address — + // `wakeByAddress` never called `prepareProvisionedDeployment`. + expect(lastAllocationService.prepareCalls).toHaveLength(0); + expect(sessionService.sendUserMessageCalls).toHaveLength(2); }); // CL-6267: the sidecar's own park/wake handler now owns respawning @@ -2507,7 +2537,13 @@ describe("createHubChatPlatform", () => { expect(lastAllocationService.prepareCalls).toHaveLength(0); }); - test("redeploys a non-routable address when lifecycle is configured", async () => { + // CL-7490: `ensureAwake` (via `wakeByAddress`) must not redeploy a + // merely not-yet-routable run — only a genuinely terminal one. A + // `deployed`/`running` run that has not yet registered is left alone + // here (no unit test needed for that: `wakeByAddress` becomes a + // pure no-op, nothing to assert beyond "did not call prepare", which + // the failed-run tests below cover by contrast). + test("relaunches a failed, non-routable address exactly once when lifecycle is configured", async () => { resolveDefinitionSourcesResult = { ok: true, materials: [], @@ -2535,6 +2571,7 @@ describe("createHubChatPlatform", () => { id: "ins_workbench1", address, principalId: "prin_run1", + status: "failed", }, workbenchLaunchRow: { tenantId: "ten_1", @@ -2568,7 +2605,7 @@ describe("createHubChatPlatform", () => { expect(lastAllocationService.prepareCalls).toHaveLength(1); }); - test("redeploys a non-routable address when lifecycle is not configured", async () => { + test("relaunches a failed, non-routable address exactly once when lifecycle is not configured", async () => { resolveDefinitionSourcesResult = { ok: true, materials: [], @@ -2596,6 +2633,7 @@ describe("createHubChatPlatform", () => { id: "ins_workbench1", address, principalId: "prin_run1", + status: "failed", }, workbenchLaunchRow: { tenantId: "ten_1", @@ -2652,15 +2690,30 @@ describe("createHubChatPlatform", () => { }); }); - // CL-7214: `sendRunMailWithReclaimRetry`'s reclaim-retry loop used - // to call `wakeByAddress` directly, bypassing `lifecycle.ensureAwake`'s - // per-address coalescing entirely. Proves that a reclaim-retry wake and - // an independent, concurrent `ensureAwake` call for the same address - // now coalesce onto the same in-flight wake — never dispatching a - // second, concurrent `wakeByAddress` that would race the first on the - // same `session_asset` primary key and git ref. - describe("wakeByAddressBounded reclaim-retry coalescing", () => { - test("a reclaim-retry wake and a concurrent ensureAwake call for the same address never redeploy it twice", async () => { + // CL-7214 proved `sendRunMailWithReclaimRetry`'s reclaim-retry wake and + // an independent, concurrent `ensureAwake` call coalesced onto the same + // in-flight redeploy rather than racing two `prepareProvisionedDeployment` + // calls on the same `session_asset` primary key and git ref. That + // redeploy is no longer wakeByAddress's behavior for a live, merely + // not-yet-routable run (CL-7490) — the only thing left to redeploy is a + // genuinely terminal run, and the coalescing that guards a concurrent + // redeploy of it is `@corbits/agent-lifecycle`'s `pendingWakes` map, + // proven directly in that package's own test suite (see the note above + // `describe("lifecycle wiring", ...)`), not re-proven here. + + // 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: [], @@ -2675,23 +2728,41 @@ describe("createHubChatPlatform", () => { ], defaultSource: "off_1", }; - const address = "ins_workbench1@ten1.workbench.test"; + + const staleAddress = "run_gen1@ten1.workbench.test"; const db = createFakeDb({ assetRow: { tenantId: "ten_1", creatorPrincipalId: "prin_creator", - name: "workbench-1", + name: "ins_room1", displayName: null, }, - definitionId: "wfd_workbench1", + definitionId: "wfd_room1", workflowRunRow: { - id: "ins_workbench1", - address, - principalId: "prin_run1", + 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: "ins_workbench1", + // `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", @@ -2703,30 +2774,33 @@ describe("createHubChatPlatform", () => { }); db.inserted.push({ table: agentSession, - values: { id: "ses_run1", principalId: "prin_run1" }, + values: { id: "ses_run1", principalId: "prin_room1" }, }); const sessionService = createFakeSessionService(); - let sendAttempts = 0; sessionService.sendUserMessage = async (params: unknown) => { sessionService.sendUserMessageCalls.push(params); - sendAttempts += 1; - // The first delivery attempt fails as "agent is unreachable", - // forcing `sendFoldedMailWithReclaimRetry` down its reclaim - // path — the call site that used to bypass coalescing. - if (sendAttempts === 1) { + 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"); }; - // The first provision (the cold wake before the first send attempt) - // resolves immediately; every prepare after it is held open until - // the test releases it, so the reclaim-retry's redeploy is still - // in flight when the concurrent `ensureAwake` call joins it. - let deployCallCount = 0; - const gatedDeployReleases: (() => void)[] = []; - const sidecarRouter = createFakeSidecarRouter({ routableAddresses: [] }); + // 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); @@ -2734,20 +2808,7 @@ describe("createHubChatPlatform", () => { params: Parameters[0], ) => { const result = await originalPrepare(params); - deployCallCount += 1; - if (deployCallCount >= 2) { - 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); - } + sidecarRouter.routableAddresses.push(result.deploymentAddress); return result; }; @@ -2757,36 +2818,36 @@ describe("createHubChatPlatform", () => { sessionService, sidecarRouter, eventCollectors: createFakeEventCollectors(), - lifecycle: { idleSleepMs: 60_000 }, + workflowAllocationService: allocationService, routableWaitDeadlineMs: 5_000, mailDeliveryTimeoutMs: 5_000, - workflowAllocationService: allocationService, }); - const sendMailPromise = platform.sendMail({ + const sent = await platform.sendMail({ tenantId: "ten_1", - workbenchId: "ins_workbench1", + workbenchId: "ins_room1", principalId: "prin_sender", - content: { content: "hello workbench" }, + content: { content: "hello" }, }); - // Let the cold wake (prepare #1) resolve, the first send attempt - // fail, and the reclaim retry's own wake (prepare #2) start and - // gate. - await new Promise((resolve) => setTimeout(resolve, 50)); - expect(deployCallCount).toBe(2); - - // A second, independent caller asks for the same address while - // prepare #2 is still in flight — the concurrent-wake race CL-7214 - // closes. - const ensureAwakePromise = platform.ensureAwake(address); - await new Promise((resolve) => setTimeout(resolve, 50)); - expect(deployCallCount).toBe(2); - - gatedDeployReleases.forEach((release) => release()); - await Promise.all([sendMailPromise, ensureAwakePromise]); + 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); - expect(deployCallCount).toBe(2); + 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"]); }); }); @@ -2954,7 +3015,7 @@ describe("createHubChatPlatform", () => { model: "claude-sonnet-5", }); - function buildRefreshableDb() { + function buildRefreshableDb(workflowRunStatus?: string) { return createFakeDb({ assetRow: { tenantId: "ten_1", @@ -2968,6 +3029,9 @@ describe("createHubChatPlatform", () => { address: "agent1@ten1.workbench.test", principalId: "prin_agent1", definitionId: "wfd_agent1", + ...(workflowRunStatus !== undefined + ? { status: workflowRunStatus } + : {}), }, workflowDefinitionRow: { id: "wfd_agent1", @@ -3032,7 +3096,12 @@ describe("createHubChatPlatform", () => { defaultSource: "off_1", }; - const db = buildRefreshableDb(); + // CL-7490: `wakeByAddress` no longer redeploys a merely + // not-yet-routable run — only a genuinely dead one, so this run + // must actually be terminal (not just offline) for `sendMail`'s + // wake-on-send path to be the one that carries the refreshed + // system prompt into a fresh deploy. + const db = buildRefreshableDb("failed"); db.inserted.push({ table: agentSession, values: { id: "ses_agent1", principalId: "prin_agent1" },