diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index 8f7f6df6..29f993dd 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -52,6 +52,7 @@ import { isWorkbenchHostDefinitionName } from "./workbench-host-naming"; import type { EventCollectorRegistry, SidecarRouter } from "@intx/hub-sessions"; import type { InferencePreference } from "@intx/agent"; import { formatRunAddress } from "@intx/types"; +import type { FoldedBody } from "@intx/workflow-deploy"; import { AgentUnreachableError, type ChatWorkbenchEvent, @@ -383,14 +384,112 @@ export function createHubChatPlatform( } /** - * Brings the run behind `address` back to routable, whichever kind of - * "not routable" it is in. `address` may be either side of the - * mapping — the stable address the room holds, or the live deployment - * address the sidecar reports. + * A JSON encoding with every object's keys sorted, recursively — + * order-independent, so two objects with the same key/value pairs + * built by different code paths (a fresh `readFoldedBody()` call vs. + * whatever `workbench_launch.foldedBody` already holds) always + * encode identically regardless of construction order. + */ + function canonicalJSON(value: unknown): string { + return JSON.stringify(value, function replacer(_key, val) { + if (val === null || typeof val !== "object" || Array.isArray(val)) { + return val; + } + const sorted: Record = {}; + for (const k of Object.keys(val as Record).sort()) { + sorted[k] = (val as Record)[k]; + } + return sorted; + }); + } + + /** + * Whether two folded bodies are the same deployable content. + * + * Deliberately NOT a wire-hash comparison. The first version of this + * check compared CL-6452's per-deploy clone's `wireHash` against the + * asset's current hub-authored `wireHash` — but that clone's hash is + * unique to the run BY DESIGN (`launch.ts`'s `markRunDeployClone`: + * "a folded run's deployed bytes carry per-run values (`wf_`, + * the run's trigger address), so their wire hash is unique to the + * run"). Comparing it to the authored hash therefore reads every + * live run as "drifted" on every single send, regardless of whether + * anything actually changed — the PR #298 regression that broke + * three chat e2e tests by relaunching a healthy run mid-flight. + * `foldedBody` itself carries no per-run values, so comparing its + * content directly is the correct, stable signal. + * + * A second, subtler regression on the way to this version: a raw + * `JSON.stringify(a) === JSON.stringify(b)` comparison is NOT the + * same as content equality — a freshly-built `readFoldedBody()` + * result and the object already stored on `workbench_launch` can + * hold identical key/value pairs in different insertion order (e.g. + * `model` last vs. first), which `JSON.stringify` renders as + * different strings. Confirmed live against the real echo-agent e2e + * fixture: `fresh`/`current` were byte-for-byte the same data, + * reordered, and every send relaunched a perfectly healthy run. + * `canonicalJSON` above sorts keys at every level before comparing, + * so construction order can never manufacture a false drift signal. + */ + function foldedBodyContentEquals(a: FoldedBody, b: FoldedBody): boolean { + return canonicalJSON(a) === canonicalJSON(b); + } + + /** + * Whether a routable run's deployed content has drifted from its + * definition's current hub-authored projection, and the folded body + * it should redeploy with if so. + * + * CL-6588: a launch (and every wake/relaunch since) renders a run's + * `foldedBody` once and never re-reads the definition's asset again + * on its own — `refreshAgentInstanceFromDefinition` above is the + * existing, explicitly-triggered lever for that, fired only when a + * human saves an edit. This is the same recompute, fired automatically + * ahead of a send instead of waiting for someone to click refresh, so + * a definition that changed for a reason the room's occupants never + * caused (a platform code fix, a redeployed default agent package) + * still reaches an already-launched instance. + */ + async function resolveDriftedFoldedBody( + tenantId: string, + run: LiveAgent["run"], + currentFoldedBody: FoldedBody, + ) { + if (run.definitionId === null) return undefined; + const definitionRow = await deps.db.query.workflowDefinition.findFirst({ + where: and( + eq(workflowDefinition.id, run.definitionId), + eq(workflowDefinition.tenantId, tenantId), + ), + }); + if (definitionRow === undefined || definitionRow.assetId === null) { + return undefined; + } + const { row: authoredRow, projection } = + await resolveAuthoredProjectedDefinition(tenantId, { + assetId: definitionRow.assetId, + name: definitionRow.name, + }); + const freshFoldedBody = readFoldedBody( + projection, + authoredRow.grantRequirements, + ); + if (foldedBodyContentEquals(freshFoldedBody, currentFoldedBody)) { + return undefined; + } + return freshFoldedBody; + } + + /** + * Brings the run behind `address` back to routable — or, now, current + * — whichever kind of "not serving what it should" it is in. + * `address` may be either side of the mapping — the stable address + * the room holds, or the live deployment address the sidecar reports. * * CL-6267: the sidecar's own park/wake handler owns respawning a * parked-but-still-announced deployment the moment mail routes to it, - * so a routable address is never deployed or undeployed here. + * so a routable address is never deployed or undeployed here for + * routability alone. * * CL-6365: a run that is unroutable because it DIED — the hub's own * `workflow_run.status` is terminal and it is not merely a folded run @@ -398,6 +497,13 @@ export function createHubChatPlatform( * durable event log already carries the terminal event, so * redeploying it would come straight back as `workflow_run_terminal` * and the message would be dropped in silence. That case relaunches. + * + * CL-6588: a run that is very much alive and routable can still be + * serving stale bytes — the pattern behind every "fixed but still + * broken for yesterday's signup" bug tonight. `relaunchTerminalRun` + * already mints a fresh run and repoints the room's stable address at + * it without moving the room; a drifted-but-alive run gets exactly + * that treatment, just triggered by content drift instead of death. */ async function wakeByAddress(address: string): Promise { const binding = await readBindingByAddress(deps.db, address); @@ -417,7 +523,10 @@ export function createHubChatPlatform( await relaunchTerminalRun(live); return; } - if (isRoutable(live.run.address)) return; + if (isRoutable(live.run.address)) { + await reconcileDriftedRun(address); + return; + } const wakeParams = { tenantId: binding.tenantId, @@ -432,6 +541,50 @@ export function createHubChatPlatform( }); } + /** + * Redeploys `address`'s run in place if its deployed definition has + * drifted from the current authored projection, otherwise no-ops. + * Never throws for an address this adapter cannot resolve to a live, + * routable, non-terminal run — `sendMail` calls this unconditionally + * after its own wake gate (see the CL-6588 note there), so a send + * that never needed waking must not gain a new failure mode from a + * check that used to never run for it. + */ + async function reconcileDriftedRun(address: string): Promise { + const binding = await readBindingByAddress(deps.db, address); + if (binding === undefined) return; + const live = await resolveLiveAgent(deps.db, binding); + if (live === undefined || live.run.address === null) return; + if (await isBeyondWake(deps.db, live.run)) return; + if (!isRoutable(live.run.address)) return; + // Best-effort: a staleness check that cannot resolve the current + // authored projection (e.g. `DefinitionProjectionMissingError` for + // a pre-cutover definition with no frozen wire projection at all) + // must never take the send down with it — the run is exactly as + // routable as it was before this check ran. Log and proceed as + // "nothing to reconcile", the same posture `sweepTerminalRuns` + // takes for a relaunch failure. + let driftedFoldedBody: Awaited>; + try { + driftedFoldedBody = await resolveDriftedFoldedBody( + binding.tenantId, + live.run, + binding.foldedBody, + ); + } catch (cause: unknown) { + wakeLogger.error`drift check for ${binding.roomAddress} (run ${live.run.id}) failed, leaving it as-is: ${ + cause instanceof Error ? cause.message : String(cause) + }`; + return; + } + if (driftedFoldedBody === undefined) return; + wakeLogger.info`relaunching ${binding.roomAddress}: run ${live.run.id} is routable but its deployed definition has drifted from the current authored projection; minting a fresh run`; + await relaunchTerminalRun({ + binding: { ...binding, foldedBody: driftedFoldedBody }, + run: live.run, + }); + } + /** * The live run mail must actually be delivered to for a stable * participant id — not the room's own address, once anything has been @@ -754,6 +907,13 @@ export function createHubChatPlatform( } else if (!isRoutable(liveAddress)) { await wakeByAddress(liveAddress); } + // CL-6588: `lifecycle.ensureAwake` returns immediately for an + // address that is already routable — routability is the only + // thing it checks — so an already-live-but-stale run would never + // reach `wakeByAddress`'s drift check above through that branch. + // Run it unconditionally so every send through this choke point + // — not only the ones that needed waking — reconciles staleness. + await reconcileDriftedRun(liveAddress); const delivery = await requireLive(input.workbenchId); const deliveryAddress = delivery.binding.liveAddress; // Tracking here (not only at launch) brings instances that were @@ -891,9 +1051,16 @@ export function createHubChatPlatform( } if (lifecycle !== undefined) { await lifecycle.ensureAwake(binding.liveAddress); + // CL-6588: see the matching note in `sendMail` — routability + // alone is what `lifecycle.ensureAwake` checks, so an + // already-routable-but-stale run needs this run unconditionally. + await reconcileDriftedRun(binding.liveAddress); + return; + } + if (isRoutable(binding.liveAddress)) { + await reconcileDriftedRun(binding.liveAddress); return; } - if (isRoutable(binding.liveAddress)) return; await wakeByAddress(binding.liveAddress); }, }; diff --git a/packages/chat/test/platform-adapter.test.ts b/packages/chat/test/platform-adapter.test.ts index 47ba651a..c15e6993 100644 --- a/packages/chat/test/platform-adapter.test.ts +++ b/packages/chat/test/platform-adapter.test.ts @@ -183,6 +183,7 @@ function createFakeDb(opts: { name?: string; origin?: "authored" | "run"; grantRequirements?: unknown; + wireHash?: string | null; } | undefined; workflowDefinitionRows?: @@ -195,6 +196,7 @@ function createFakeDb(opts: { assetId?: string | null; origin?: "authored" | "run"; grantRequirements?: unknown; + wireHash?: string | null; }[] | undefined; tenantRow?: { id: string; domain: string } | undefined; @@ -2527,6 +2529,283 @@ describe("createHubChatPlatform", () => { }); }); +// CL-6588: a launch renders `workflow_run.definitionId`/`workbench_launch`'s +// `foldedBody` once, and neither a wake nor a relaunch has ever re-read the +// definition's asset on its own -- only an explicit +// `refreshAgentInstanceFromDefinition` call (a human saving settings) did. +// A run that is routable but was deployed from a definition that has since +// changed for a reason nobody in the room caused (a platform code fix, a +// redeployed default agent package) stayed silently wrong forever. These +// prove the automatic reconciliation added ahead of `wakeByAddress`'s +// already-routable return and `sendMail`'s choke point. +describe("createHubChatPlatform stale-definition reconciliation", () => { + const STALE_SYSTEM_PROMPT = + "the openai adapter: invalid quirks: default must be removed"; + const FIXED_SYSTEM_PROMPT = "I am working in this workbench."; + + // CL-6452: every deploy freezes a per-run clone of the agent's + // definition under a wire hash that bakes in per-run values + // (`wf_`, the run's own trigger address) — so the clone's + // hash is unique to the run BY DESIGN, even when its content is + // byte-identical to what's authored today. The fixture's clone row + // deliberately carries no `wireHash` field at all (undefined), and + // the tests below prove staleness is decided on CONTENT + // (`foldedBody`), never on that per-run-unique hash. + function createDriftFixture(opts: { + deployedSystemPrompt: string; + authoredSystemPrompt: string; + routable: boolean; + }) { + const db = createFakeDb({ + assetRow: { + tenantId: "ten_1", + creatorPrincipalId: "prin_creator", + name: "workbench-1", + displayName: null, + }, + definitionId: "wfd_unused", + workflowRunRow: { + id: "run_stale", + address: "run_stale@ten1.workbench.test", + principalId: "prin_room1", + definitionId: "wfd_run_clone", + status: "running", + }, + workflowDefinitionRow: { + id: "wfd_run_clone", + tenantId: "ten_1", + status: "deployed", + assetId: "asst_myra", + name: "myra", + origin: "run", + }, + workflowDefinitionRows: [ + { + id: "wfd_run_clone", + tenantId: "ten_1", + status: "deployed", + name: "myra", + assetId: "asst_myra", + origin: "run", + }, + { + id: "wfd_myra_authored", + tenantId: "ten_1", + status: "deployed", + name: "myra", + assetId: "asst_myra", + origin: "authored", + }, + ], + workbenchLaunchRow: { + tenantId: "ten_1", + instanceId: "run_stale", + currentRunId: "run_stale", + foldedBody: { + systemPrompt: opts.deployedSystemPrompt, + toolPackagePins: [], + grantRequirements: [], + credentialBindings: [], + model: null, + }, + }, + wireProjectionsByDefinitionId: { + // `model: null` matches `foldedBody.model` above -- `inertProjection` + // defaults `model` to `"claude-sonnet-5"`, which would otherwise + // read as content drift on its own and mask what these tests + // are actually proving (system-prompt equality vs. difference). + wfd_run_clone: inertProjection({ + id: "wfd_run_clone", + systemPrompt: opts.deployedSystemPrompt, + model: null, + }), + wfd_myra_authored: inertProjection({ + id: "wfd_myra_authored", + systemPrompt: opts.authoredSystemPrompt, + model: null, + }), + }, + }); + const sidecarRouter = createFakeSidecarRouter({ + routableAddresses: opts.routable ? ["run_stale@ten1.workbench.test"] : [], + }); + const platform = createHubChatPlatform({ + toolGrantsForPins: () => [], + db: db as never, + sessionService: createFakeSessionService(), + assetService: createFakeAssetService(), + sidecarRouter, + eventCollectors: createFakeEventCollectors(), + }); + return { db, platform }; + } + + test("a routable run whose deployed content differs from the current authored content is relaunched, not served as-is", async () => { + const { db, platform } = createDriftFixture({ + deployedSystemPrompt: STALE_SYSTEM_PROMPT, + authoredSystemPrompt: FIXED_SYSTEM_PROMPT, + routable: true, + }); + + await platform.ensureAwake("run_stale@ten1.workbench.test"); + + const repointed = db.updated.find((row) => row.table === workbenchLaunch) + ?.values as { currentRunId: string } | undefined; + expect(repointed?.currentRunId).toBeDefined(); + expect(repointed?.currentRunId).not.toBe("run_stale"); + }); + + // The exact regression this test guards against: PR #298's first cut + // compared the run's own clone's wire hash (always unique per run) + // against the authored row's hash, so this fixture -- content + // identical, hash necessarily different -- read as "drifted" on + // every single call and relaunched a perfectly healthy run on every + // wake/send, breaking three chat e2e tests that watched a run stay + // alive across a turn. + test("a routable run whose deployed content matches the current authored content is left alone, even though its per-run clone's wire hash is necessarily unrelated to the authored row's", async () => { + const { db, platform } = createDriftFixture({ + deployedSystemPrompt: FIXED_SYSTEM_PROMPT, + authoredSystemPrompt: FIXED_SYSTEM_PROMPT, + routable: true, + }); + + await platform.ensureAwake("run_stale@ten1.workbench.test"); + + expect( + db.updated.find((row) => row.table === workbenchLaunch), + ).toBeUndefined(); + }); + + test("sendMail redeploys an already-routable-but-drifted target before delivering — lifecycle.ensureAwake's routability check alone would have missed it", async () => { + resolveDefinitionSourcesResult = { + ok: true, + sources: [ + { + id: "off_1", + provider: "anthropic", + baseURL: "https://inference.invalid", + apiKey: "placeholder", + model: "claude-sonnet-5", + }, + ], + defaultSource: "off_1", + }; + const { db } = createDriftFixture({ + deployedSystemPrompt: STALE_SYSTEM_PROMPT, + authoredSystemPrompt: FIXED_SYSTEM_PROMPT, + routable: true, + }); + db.inserted.push({ + table: agentSession, + values: { id: "ses_stale", principalId: "prin_room1" }, + }); + const sessionService = createFakeSessionService(); + const platform = createHubChatPlatform({ + toolGrantsForPins: () => [], + db: db as never, + sessionService, + assetService: createFakeAssetService(), + sidecarRouter: createFakeSidecarRouter({ + routableAddresses: ["run_stale@ten1.workbench.test"], + }), + eventCollectors: createFakeEventCollectors(), + // Configured: `sendMail` takes the `lifecycle.ensureAwake` branch, + // whose own routability check alone would never have caught this. + lifecycle: { idleSleepMs: 60_000 }, + }); + + await platform.sendMail({ + tenantId: "ten_1", + workbenchId: "run_stale", + principalId: "prin_sender", + content: { content: "hello" }, + }); + + expect(sessionService.adoptedDeployCalls).toHaveLength(1); + const deployed = sessionService.adoptedDeployCalls[0] as { + config: { systemPrompt: string }; + }; + expect(deployed.config.systemPrompt).toBe(FIXED_SYSTEM_PROMPT); + }); + + // The coordinator's explicit ask: "unknown" (no authored sibling this + // adapter can resolve at all -- e.g. a standalone/section-mode run + // whose asset carries no hub-authored candidate) must mean leave it + // alone, never treat as drifted. + test("a run whose asset has no resolvable authored sibling is left alone, not blocked or relaunched", async () => { + const db = createFakeDb({ + assetRow: { + tenantId: "ten_1", + creatorPrincipalId: "prin_creator", + name: "workbench-1", + displayName: null, + }, + definitionId: "wfd_unused", + workflowRunRow: { + id: "run_standalone", + address: "run_standalone@ten1.workbench.test", + principalId: "prin_room1", + definitionId: "wfd_standalone_clone", + status: "running", + }, + workflowDefinitionRow: { + id: "wfd_standalone_clone", + tenantId: "ten_1", + status: "deployed", + assetId: "asst_standalone", + name: "standalone-agent", + origin: "run", + }, + // No "authored" sibling at all -- `resolveAuthoredProjectedDefinition` + // finds no candidate and raises `DefinitionProjectionMissingError`. + workflowDefinitionRows: [ + { + id: "wfd_standalone_clone", + tenantId: "ten_1", + status: "deployed", + name: "standalone-agent", + assetId: "asst_standalone", + origin: "run", + }, + ], + workbenchLaunchRow: { + tenantId: "ten_1", + instanceId: "run_standalone", + currentRunId: "run_standalone", + foldedBody: { + systemPrompt: "be helpful", + toolPackagePins: [], + grantRequirements: [], + credentialBindings: [], + model: null, + }, + }, + wireProjectionsByDefinitionId: { + wfd_standalone_clone: inertProjection({ + id: "wfd_standalone_clone", + systemPrompt: "be helpful", + }), + }, + }); + const platform = createHubChatPlatform({ + toolGrantsForPins: () => [], + db: db as never, + sessionService: createFakeSessionService(), + assetService: createFakeAssetService(), + sidecarRouter: createFakeSidecarRouter({ + routableAddresses: ["run_standalone@ten1.workbench.test"], + }), + eventCollectors: createFakeEventCollectors(), + }); + + await platform.ensureAwake("run_standalone@ten1.workbench.test"); + + expect( + db.updated.find((row) => row.table === workbenchLaunch), + ).toBeUndefined(); + }); +}); + // CL-6365: the send-triggered relaunch only fires when somebody writes // into the room. A room whose agent died in a crash has nobody writing // into it — that is the whole failure — so the sweep is what makes the