From 84ba3854f3c5d25628d2e4f56eeb2f4bdda4d143 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Thu, 20 Aug 2026 03:36:05 -0700 Subject: [PATCH 1/2] Add tests for the address-to-current-run mapping Covers both directions of the mapping a relaunch needs: the room's own address resolving to whichever run is live now, and a live deployment address resolving back to the participant the room has been addressing all along. Also separates a run that is beyond waking (terminal, its durable log already sealed) from a folded run merely parked between messages, which still wakes. The two existing chat fakes gain the launch row every chat run has in production, since that row is now the mapping every lookup goes through. --- packages/chat/src/agent-binding.test.ts | 152 +++++++++++++++++++ packages/chat/test/chat-orchestrator.test.ts | 44 ++++++ packages/chat/test/platform-adapter.test.ts | 63 +++++++- 3 files changed, 252 insertions(+), 7 deletions(-) create mode 100644 packages/chat/src/agent-binding.test.ts diff --git a/packages/chat/src/agent-binding.test.ts b/packages/chat/src/agent-binding.test.ts new file mode 100644 index 000000000..11b37d205 --- /dev/null +++ b/packages/chat/src/agent-binding.test.ts @@ -0,0 +1,152 @@ +// The address→run mapping's own behavior: that a room address and a +// live deployment address both resolve to the same participant once +// they have come apart, and that "this run is dead" is told apart from +// "this folded run is parked between messages". +import { describe, expect, test } from "bun:test"; +import { + isBeyondWake, + readBindingByAddress, + resolveRoomAddress, +} from "./agent-binding"; + +const FOLDED_BODY = { + systemPrompt: "be helpful", + toolPackagePins: [], + grantRequirements: [], + credentialBindings: [], + model: null, +}; + +type LaunchRow = { + tenantId: string; + instanceId: string; + currentRunId: string; + foldedBody: unknown; + noopInference: boolean; +}; + +/** + * Honours the `where` filter, unlike this package's older fakes: the + * whole point of these cases is which COLUMN a lookup matched on, so a + * filter-ignoring double would pass them vacuously. `readLaunchRow` + * builds its filter with drizzle's `eq`, whose serialized form carries + * the compared value in `queryChunks`; matching on the value alone is + * enough here because no scenario has one id appearing in two columns + * of different rows. + */ +function fakeDb(rows: LaunchRow[], foldedRunMarkerIds: string[] = []) { + function matchingValue(where: unknown): string | undefined { + const chunks = (where as { queryChunks?: unknown[] }).queryChunks ?? []; + for (const chunk of chunks) { + const value = (chunk as { value?: unknown }).value; + if (typeof value === "string") return value; + } + return undefined; + } + return { + select: (columns?: Record) => ({ + from: () => ({ + where: (predicate: unknown) => ({ + limit: async () => { + const value = matchingValue(predicate); + if (columns !== undefined) { + // `isFoldedRunSettled`'s marker probe. + return foldedRunMarkerIds.includes(value ?? "") + ? [{ id: value }] + : []; + } + return rows.filter( + (row) => row.instanceId === value || row.currentRunId === value, + ); + }, + }), + }), + }), + } as never; +} + +const relaunched: LaunchRow = { + tenantId: "ten_1", + instanceId: "run_original", + currentRunId: "run_fresh", + foldedBody: FOLDED_BODY, + noopInference: false, +}; + +describe("readBindingByAddress", () => { + test("resolves the room's own address to the run that is live now", async () => { + const binding = await readBindingByAddress( + fakeDb([relaunched]), + "run_original@acme.example", + ); + expect(binding?.stableId).toBe("run_original"); + expect(binding?.currentRunId).toBe("run_fresh"); + expect(binding?.roomAddress).toBe("run_original@acme.example"); + expect(binding?.liveAddress).toBe("run_fresh@acme.example"); + }); + + test("resolves the live deployment address back to the same participant", async () => { + // This is the inbound half: a relaunched run announces itself under + // an address the room has never seen, and its reply still has to + // land in the room that has been addressing it as `run_original`. + const binding = await readBindingByAddress( + fakeDb([relaunched]), + "run_fresh@acme.example", + ); + expect(binding?.roomAddress).toBe("run_original@acme.example"); + }); + + test("is undefined for an address this package never launched", async () => { + expect( + await readBindingByAddress(fakeDb([relaunched]), "echo_1@acme.example"), + ).toBeUndefined(); + }); +}); + +describe("resolveRoomAddress", () => { + test("leaves a non-participant address alone", async () => { + expect( + await resolveRoomAddress(fakeDb([relaunched]), "echo_1@acme.example"), + ).toBe("echo_1@acme.example"); + }); +}); + +describe("isBeyondWake", () => { + test("a failed run is beyond waking — its durable log is already terminal", async () => { + expect( + await isBeyondWake(fakeDb([], ["run_fresh"]), { + id: "run_fresh", + status: "failed", + }), + ).toBe(true); + }); + + test("a running run is not", async () => { + expect( + await isBeyondWake(fakeDb([], ["run_fresh"]), { + id: "run_fresh", + status: "running", + }), + ).toBe(false); + }); + + test("a folded run parked between messages is not — that one wakes", async () => { + expect( + await isBeyondWake(fakeDb([], ["run_fresh"]), { + id: "run_fresh", + status: "completed", + }), + ).toBe(false); + }); + + test("a plain deployment's genuine completion IS beyond waking", async () => { + // "completed" with no `folded_run` marker is a one-shot deployment + // that is done forever, not an idle conversational run. + expect( + await isBeyondWake(fakeDb([], []), { + id: "run_oneshot", + status: "completed", + }), + ).toBe(true); + }); +}); diff --git a/packages/chat/test/chat-orchestrator.test.ts b/packages/chat/test/chat-orchestrator.test.ts index fd8750be9..93d08b438 100644 --- a/packages/chat/test/chat-orchestrator.test.ts +++ b/packages/chat/test/chat-orchestrator.test.ts @@ -119,6 +119,28 @@ function approvalRow(overrides?: { } as const; } +// The `workbench_launch` mapping row `readBindingByAddress` reads to +// turn an event-stream address into the room's own participant address +// (see `../src/agent-binding.ts`). Every scenario here predates any +// relaunch, so the stable id and the current run id are the same value +// — which is exactly the identity mapping a room starts life with. +function launchRowFor(runId: string, tenantId: string) { + return { + tenantId, + instanceId: runId, + currentRunId: runId, + foldedBody: { + systemPrompt: "be helpful", + toolPackagePins: [], + grantRequirements: [], + credentialBindings: [], + model: null, + }, + noopInference: false, + createdAt: new Date("2026-08-08T09:00:00.000Z"), + }; +} + // The real `findFoldedRunByAddress` (exercised, not mocked, so this // file never risks poisoning `@corbits/folded-runs`'s module namespace // for `platform-adapter.test.ts` when the whole package's suite runs @@ -142,6 +164,14 @@ function createFakeDb(run?: { : { ...run, principalId: run.principalId ?? null }, }, }, + select: () => ({ + from: () => ({ + where: () => ({ + limit: async () => + run === undefined ? [] : [launchRowFor(run.id, run.tenantId)], + }), + }), + }), }; } @@ -294,6 +324,7 @@ describe("createChatOrchestrator", () => { { id: "ins_echo1", tenantId: "ten_1" }, ]; let dbCallIndex = 0; + let launchCallIndex = 0; const db = { query: { workflowRun: { @@ -306,6 +337,19 @@ describe("createChatOrchestrator", () => { }, }, }, + select: () => ({ + from: () => ({ + where: () => ({ + limit: async () => { + const run = runsByCallOrder[launchCallIndex]; + launchCallIndex += 1; + return run === undefined + ? [] + : [launchRowFor(run.id, run.tenantId)]; + }, + }), + }), + }), }; const orchestrator = createChatOrchestrator({ diff --git a/packages/chat/test/platform-adapter.test.ts b/packages/chat/test/platform-adapter.test.ts index 304d8426e..02972440c 100644 --- a/packages/chat/test/platform-adapter.test.ts +++ b/packages/chat/test/platform-adapter.test.ts @@ -170,6 +170,13 @@ function createFakeDb(opts: { | { tenantId: string; instanceId: string; + /** + * The run the stable `instanceId` currently resolves to (see + * `../src/agent-binding.ts`). Defaults to `instanceId` — the + * identity mapping every room starts life with, before any + * relaunch has re-pointed it. + */ + currentRunId?: string; foldedBody: unknown; noopInference?: boolean; } @@ -285,13 +292,41 @@ function createFakeDb(opts: { const insertedLaunch = inserted.findLast( (row) => row.table === workbenchLaunch, )?.values; - return selectChain( - opts.workbenchLaunchRow !== undefined - ? [opts.workbenchLaunchRow] - : insertedLaunch !== undefined - ? [insertedLaunch] - : [], - ); + // Every run this package launches has a launch row, and + // that row is now the address→run mapping every lookup + // goes through — so a scenario that configures a run but + // no launch row gets the identity mapping for it rather + // than a hole no production run could be in. + const row = + opts.workbenchLaunchRow ?? + insertedLaunch ?? + (opts.workflowRunRow !== undefined + ? { + tenantId: "ten_1", + instanceId: opts.workflowRunRow.id, + currentRunId: opts.workflowRunRow.id, + foldedBody: { + systemPrompt: "be helpful", + toolPackagePins: [], + grantRequirements: [], + credentialBindings: [], + model: null, + }, + noopInference: false, + } + : undefined); + if (row === undefined) return selectChain([]); + const withCurrent = row as { + instanceId: string; + currentRunId?: string; + }; + return selectChain([ + { + ...withCurrent, + currentRunId: + withCurrent.currentRunId ?? withCurrent.instanceId, + }, + ]); } if (table === agentSession) { // `resolveRunSessionId` selects `{ id }` filtered by @@ -2192,6 +2227,20 @@ describe("createHubChatPlatform", () => { displayName: null, }, definitionId: "wfd_workbench1", + // `ensureAwake` resolves the LIVE address through the mapping + // before it asks whether anything is routable, so even the + // no-op path needs the participant's binding to exist. + workbenchLaunchRow: { + tenantId: "ten_1", + instanceId: "ins_workbench1", + foldedBody: { + systemPrompt: "be helpful", + toolPackagePins: [], + grantRequirements: [], + credentialBindings: [], + model: null, + }, + }, }); const sessionService = createFakeSessionService(); const platform = createHubChatPlatform({ From 1446b067a42367a54e2b194f535833bc47d3805f Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Thu, 20 Aug 2026 03:36:07 -0700 Subject: [PATCH 2/2] Relaunch: a dead run is replaced, not resurrected MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A run that dies mid-turn commits its terminal event to the durable log before exiting, so waking its address again comes straight back as workflow_run_terminal and the next message is dropped in silence. The fix is a fresh run — and never reclaiming the dead one's log, which is the audit trail this shape exists to keep. The platform fuses a run's id to its address in three independent places, so a fresh run necessarily carries a fresh address. That is only survivable because the room stops being the run: chat.workbench_launch now maps a stable participant id (the address the room uses forever) to the current run id behind it, re-pointed on every relaunch. agent-binding.ts owns both directions — outbound sends resolve the live address, inbound events resolve back to the room address that participant records, mention handles, and posted messages all carry. Detection is workflow_run.status plus folded-runs' isFoldedRunSettled, which had no caller until now: a terminal status that is not a parked folded run means relaunch rather than wake. The relaunch is send-triggered; a boot-time sweep and an in-room notice are still open. See docs/revendor-inventory.md. --- docs/revendor-inventory.md | 63 ++++++ packages/chat/src/agent-binding.ts | 213 ++++++++++++++++++ packages/chat/src/chat-orchestrator.ts | 50 +++-- packages/chat/src/migrations.ts | 24 +- packages/chat/src/platform-adapter.ts | 299 +++++++++++++++++-------- packages/chat/src/schema.ts | 18 ++ 6 files changed, 554 insertions(+), 113 deletions(-) create mode 100644 packages/chat/src/agent-binding.ts diff --git a/docs/revendor-inventory.md b/docs/revendor-inventory.md index 8b5d18a6d..912ba1f58 100644 --- a/docs/revendor-inventory.md +++ b/docs/revendor-inventory.md @@ -997,3 +997,66 @@ The hub-side detection and the relaunch itself are not implemented here. What is on the record is a deterministic red: the proof now fails at "PROOF 4 — the section survives the restart and runs its next occurrence" with `409 workflow_run_terminal`, every time. + +### The ruling: relaunch, and un-fuse the room from the run + +The open design decision above is settled in favour of the FRESH run. +The durable log is never reclaimed and never erased: it is the audit +trail, and a resurrection that overwrote it would trade the one thing +this shape is better at for a shortcut. A relaunch mints a new run, +which the platform gives a new address and therefore a new event-log +repo (`/workflow-runs//runs//`), +leaving the dead run's log intact and readable through the ordinary run +routes. + +That is only possible because the room stops being the run. Three +independent points in the platform fuse a run's id to its address — +`deriveWorkflowRunId(address)` is the address's local part, the +code-sourced deploy front refuses an `(anchorRunId, agentAddress)` pair +that does not name the same run +(`vendor/intx/hub-sessions/src/session-service.ts`'s coherence guard), +and `receiveWorkflowRunPack` marks runs terminal by an id read out of +the address-derived repo — so a fresh run CANNOT keep the old address. +Nothing can be done about that from here, and nothing needs to be: the +address only has to be stable for the ROOM, not for the sidecar. + +`chat.workbench_launch` is where the two come apart. `instance_id` is +now the stable participant id the room addresses forever — the +workbench id for a host, the first-mint id for an invited agent — and +`current_run_id` names the run executing behind it, re-pointed on every +relaunch. `packages/chat/src/agent-binding.ts` owns both directions: +outbound (`sendMail`, `ensureAwake`, `subscribeToWorkbench`) resolves +the stable id to the live address, and inbound +(`chat-orchestrator.ts`'s `resolveMemberWorkbenches`) resolves a live +address the sidecar reported back to the room address the participant +records, the mention handles, and every already-posted message's +`sender_address` carry. Nothing else in the room moves, because none of +the room's tables were ever keyed on the run — only on the id that is +now the stable one. + +Detection is `workflow_run.status` plus `@corbits/folded-runs`' +previously orphaned `isFoldedRunSettled`: a terminal status that is NOT +a folded run parked between messages is a run beyond waking, and +`wakeByAddress` relaunches it instead of redeploying an address whose +log already says terminal. + +### What is still open + +- **The relaunch is send-triggered.** It fires on the next `sendMail` + through the wake choke point. Nothing sweeps for terminal runs at + boot, so a room whose agent died stays silently dead until somebody + writes into it. Proof 4's "the turn the kill interrupted surfaces + visibly" hop asserts a notice with no new message sent, and that hop + needs the boot-time sweep, not this. +- **No relaunch notice is posted.** The turn-drop notice in + `chat-orchestrator.ts` fires on `message.run.ended`, which a killed + sidecar never sends. A relaunch should announce itself in the room in + the agent's own voice; the seam for it is a notice port on + `createHubChatPlatform`, wired in the hub beside `roomMessages`. +- **Pre-relaunch attachments.** `fetchBlob` reads through the LIVE + run's session, so mail attachments written under a previous run's + session are not reachable after a relaunch. Room messages themselves + are unaffected (they are `chat.workbench_messages` rows, not mail). +- **The e2e proof has not been re-run on this branch.** Proof 4's + deterministic red is unchanged; the green half is asserted only by + the unit suites so far. diff --git a/packages/chat/src/agent-binding.ts b/packages/chat/src/agent-binding.ts new file mode 100644 index 000000000..9ca65329c --- /dev/null +++ b/packages/chat/src/agent-binding.ts @@ -0,0 +1,213 @@ +// The address→current-run mapping, and the ruling it encodes: a room's +// participant is NOT a run. +// +// The platform fuses a run's identity to its mail address by +// construction — `deriveWorkflowRunId(address)` is the address's local +// part, the deploy front refuses an `(anchorRunId, agentAddress)` pair +// that does not name the same run, and the sidecar keys a run's durable +// event log at `workflow-runs//runs/`. So a +// genuinely fresh run always carries a fresh address, and a dead run's +// address can never be reused without inheriting its terminal log. +// +// A chat room cannot afford that. The room is data — its timeline, +// settings, threads, and participant records all key off ONE stable id +// — and a run that dies mid-turn must be replaceable without the room +// moving. `workbench_launch` is where the two identities come apart: +// `instanceId` is the stable id the room addresses forever, and +// `currentRunId` is the run executing behind it right now, re-pointed +// by every relaunch. The old run's terminal log is never reclaimed or +// erased; it stays readable through the platform's own run routes, +// which is the whole audit-trail argument for relaunching rather than +// resurrecting. +import { eq } from "drizzle-orm"; +import { type } from "arktype"; +import type { DB } from "@intx/db"; +import { workflowRun } from "@intx/db/schema"; +import { formatRunAddress } from "@intx/types"; +import { FoldedBodySchema, isFoldedRunSettled } from "@corbits/folded-runs"; +import type { FoldedBody } from "@intx/workflow-deploy"; +import { workbenchLaunch } from "./schema"; +import { domainOf, localPartOf } from "./agent-address"; + +/** + * The mapping row, parsed. `stableId`/`roomAddress` are what the room + * knows; `currentRunId`/`liveAddress` are what the sidecar knows. Only + * the second pair moves. + */ +export interface AgentBinding { + readonly tenantId: string; + readonly stableId: string; + readonly roomAddress: string; + readonly currentRunId: string; + readonly liveAddress: string; + readonly foldedBody: FoldedBody; + readonly noopInference: boolean; +} + +/** The live `workflow_run` row behind a binding, plus the binding itself. */ +export interface LiveAgent { + readonly binding: AgentBinding; + readonly run: { + readonly id: string; + readonly tenantId: string; + readonly definitionId: string | null; + readonly principalId: string | null; + readonly address: string | null; + readonly status: string; + }; +} + +type LaunchRow = typeof workbenchLaunch.$inferSelect; + +function bindingFrom(row: LaunchRow, domain: string): AgentBinding { + const parsed = FoldedBodySchema(row.foldedBody); + if (parsed instanceof type.errors) { + throw new Error( + `workbench_launch row for "${row.instanceId}" carries an invalid ` + + `folded body: ${parsed.summary}`, + ); + } + return { + tenantId: row.tenantId, + stableId: row.instanceId, + roomAddress: formatRunAddress(row.instanceId, domain), + currentRunId: row.currentRunId, + liveAddress: formatRunAddress(row.currentRunId, domain), + foldedBody: parsed, + noopInference: row.noopInference, + }; +} + +function requireDomain(address: string): string { + const domain = domainOf(address); + if (domain === undefined || domain.length === 0) { + throw new Error(`malformed agent address, missing "@": ${address}`); + } + return domain; +} + +async function readLaunchRow( + db: DB["db"], + column: "instanceId" | "currentRunId", + value: string, +): Promise { + const rows = await db + .select() + .from(workbenchLaunch) + .where(eq(workbenchLaunch[column], value)) + .limit(1); + return rows[0]; +} + +/** + * The binding an address names, whichever side of the mapping it is on: + * the stable room address the participant records hold, or the live + * deployment address the sidecar's event stream reports. Both resolve to + * the same binding, which is exactly what lets an inbound reply from a + * relaunched run still find the room that has been addressing it under + * its original name all along. + */ +export async function readBindingByAddress( + db: DB["db"], + address: string, +): Promise { + const domain = requireDomain(address); + const localPart = localPartOf(address); + const byStableId = await readLaunchRow(db, "instanceId", localPart); + if (byStableId !== undefined) return bindingFrom(byStableId, domain); + const byRunId = await readLaunchRow(db, "currentRunId", localPart); + return byRunId === undefined ? undefined : bindingFrom(byRunId, domain); +} + +/** + * The address the ROOM knows this agent by, for an address the sidecar + * reported. Returns the input unchanged when it names no launch this + * package owns — an echo instance on the shared event stream is not + * this mapping's business. + */ +export async function resolveRoomAddress( + db: DB["db"], + liveAddress: string, +): Promise { + const binding = await readBindingByAddress(db, liveAddress); + return binding?.roomAddress ?? liveAddress; +} + +async function readRun( + db: DB["db"], + runId: string, +): Promise { + return db.query.workflowRun.findFirst({ where: eq(workflowRun.id, runId) }); +} + +/** The binding plus its live run row, or `undefined` when either is missing. */ +export async function resolveLiveAgent( + db: DB["db"], + binding: AgentBinding, +): Promise { + const run = await readRun(db, binding.currentRunId); + return run === undefined ? undefined : { binding, run }; +} + +/** + * The live run behind a stable participant id, with the binding that + * names it. The mail domain is read off the live run's own address + * rather than taken from a caller — a stable id alone does not carry + * one, and the run is the only row that does. + */ +export async function resolveLiveByStableId( + db: DB["db"], + stableId: string, +): Promise { + const row = await readLaunchRow(db, "instanceId", stableId); + if (row === undefined) return undefined; + const run = await readRun(db, row.currentRunId); + if (run === undefined || run.address === null) return undefined; + return { binding: bindingFrom(row, requireDomain(run.address)), run }; +} + +/** + * The statuses a `workflow_run` can hold that mean "this run will never + * accept mail again". A folded run's own idle settle lands on + * "completed" too (see `@corbits/folded-runs`' `isFoldedRunSettled`), + * and that one is ordinary — it wakes. + */ +const TERMINAL_RUN_STATUSES: ReadonlySet = new Set([ + "failed", + "cancelled", + "canceled", + "completed", +]); + +/** + * Whether this run is routable-but-dead: terminal in the hub's own + * `workflow_run.status`, and not merely a folded run parked between + * messages. A run this returns true for cannot be woken — its durable + * event log already carries a terminal event, so redeploying the same + * address would come straight back as `workflow_run_terminal` — it can + * only be RELAUNCHED as a fresh run. + */ +export async function isBeyondWake( + db: DB["db"], + run: { id: string; status: string }, +): Promise { + if (!TERMINAL_RUN_STATUSES.has(run.status)) return false; + return !(await isFoldedRunSettled(db, run)); +} + +/** + * Re-points a stable participant at a freshly launched run. Written + * after the new run has actually deployed, never before: a repoint that + * outlives a failed launch would leave the room addressing a run that + * was rolled back. + */ +export async function repointBinding( + db: DB["db"], + stableId: string, + newRunId: string, +): Promise { + await db + .update(workbenchLaunch) + .set({ currentRunId: newRunId }) + .where(eq(workbenchLaunch.instanceId, stableId)); +} diff --git a/packages/chat/src/chat-orchestrator.ts b/packages/chat/src/chat-orchestrator.ts index 12517d7c4..cf29ff0f7 100644 --- a/packages/chat/src/chat-orchestrator.ts +++ b/packages/chat/src/chat-orchestrator.ts @@ -23,7 +23,6 @@ import { headlineFor } from "@corbits/approvals"; import { connectorReplyContent, - findFoldedRunByAddress, messageRunEnded, messageRunStarted, } from "@corbits/folded-runs"; @@ -46,6 +45,7 @@ import { encodeParts } from "./codec"; import type { ConnectedProviderLister } from "./inference-preferences"; import { mentionedParticipants } from "./mentions"; import { localPartOf } from "./agent-address"; +import { readBindingByAddress, resolveLiveAgent } from "./agent-binding"; import { parseParticipants, type ParticipantRecord } from "./participants"; import type { ChatPlatform } from "./platform-port"; import { postRoomMessage, type RoomMessageStore } from "./room-messages"; @@ -199,6 +199,17 @@ async function resolveMemberWorkbenches( * than guessing an owner. */ principalId: string | null; + /** + * The address the ROOM knows this agent by — its stable + * participant address, which is what the participant records, the + * mention handles, and every already-posted message's + * `senderAddress` carry. Not necessarily the address the event + * arrived on: a relaunched run announces itself under a fresh + * address (the platform derives one from the other), and the + * room must keep attributing its replies to the same teammate it + * has been talking to all along. See `./agent-binding.ts`. + */ + roomAddress: string; workbenchIds: string[]; /** * Each member workbench's own participant records, keyed by @@ -210,26 +221,33 @@ async function resolveMemberWorkbenches( } | undefined > { - const run = await findFoldedRunByAddress(deps.db, agentAddress); - if (run === undefined) { + // Resolved through the address→run mapping, not by matching the + // event's address against `workflow_run.address` directly: after a + // relaunch the two differ, and only the mapping knows that the run + // announcing itself under a fresh address is the same room teammate. + const binding = await readBindingByAddress(deps.db, agentAddress); + if (binding === undefined) { // Not every agent address on the event stream belongs to a chat // workbench (an echo instance, say) — an address this package's own // launch machinery never produced is silently not this // orchestrator's concern. return undefined; } + const live = await resolveLiveAgent(deps.db, binding); + if (live === undefined) return undefined; - const workbenches = await deps.store.listWorkbenchSettings(run.tenantId); + const workbenches = await deps.store.listWorkbenchSettings(binding.tenantId); const memberWorkbenches = workbenches.filter((workbench) => parseParticipants(workbench.settings["chat/participants"]).some( - (participant) => participant.address === agentAddress, + (participant) => participant.address === binding.roomAddress, ), ); if (memberWorkbenches.length === 0) return undefined; return { - tenantId: run.tenantId, - principalId: run.principalId, + tenantId: binding.tenantId, + principalId: live.run.principalId, + roomAddress: binding.roomAddress, workbenchIds: memberWorkbenches.map((workbench) => workbench.workbenchId), participantsByWorkbenchId: new Map( memberWorkbenches.map((workbench) => [ @@ -303,15 +321,15 @@ async function postReply( const posted = await postRoomMessage(deps, { tenantId: resolved.tenantId, workbenchId, - sender: { name: null, address: agentAddress }, + sender: { name: null, address: resolved.roomAddress }, parts: [{ kind: "text", text: content }], - runId: localPartOf(agentAddress), + runId: localPartOf(resolved.roomAddress), }); await threadDelegatedReply( deps, pendingDelegationThreads, - agentAddress, + resolved.roomAddress, workbenchId, posted.id, ); @@ -325,7 +343,9 @@ async function postReply( const mentioned = mentionedParticipants( [{ kind: "text", text: content }], participants, - ).filter((address) => localPartOf(address) !== localPartOf(agentAddress)); + ).filter( + (address) => localPartOf(address) !== localPartOf(resolved.roomAddress), + ); for (const recipient of mentioned) { await deps.platform.sendMail({ tenantId: resolved.tenantId, @@ -383,9 +403,9 @@ async function postApproveBlock( await postRoomMessage(deps, { tenantId: resolved.tenantId, workbenchId, - sender: { name: null, address: agentAddress }, + sender: { name: null, address: resolved.roomAddress }, parts: [{ kind: "block", block: { type: "approve", data } }], - runId: localPartOf(agentAddress), + runId: localPartOf(resolved.roomAddress), }); } } @@ -439,9 +459,9 @@ async function postFinalizedTurnArtifacts( await postRoomMessage(deps, { tenantId: resolved.tenantId, workbenchId, - sender: { name: null, address: agentAddress }, + sender: { name: null, address: resolved.roomAddress }, parts, - runId: localPartOf(agentAddress), + runId: localPartOf(resolved.roomAddress), }); } catch (error) { await deps.claims.release(claim); diff --git a/packages/chat/src/migrations.ts b/packages/chat/src/migrations.ts index 031cf4805..80df627f2 100644 --- a/packages/chat/src/migrations.ts +++ b/packages/chat/src/migrations.ts @@ -339,6 +339,28 @@ export const chatMigrations: readonly ChatMigration[] = [ DELETE FROM "chat"."workbench_threads" WHERE "kind" = 'reply'; `, }, + { + // Un-fuses the room's identity from the run's. `instance_id` stops + // meaning "the run" and starts meaning "the stable participant the + // room addresses"; `current_run_id` names the run actually + // executing behind it, re-pointed on every relaunch. Existing rows + // are the identity mapping they have always implied. + name: "0020_workbench_launch_current_run", + sql: ` + ALTER TABLE "chat"."workbench_launch" + ADD COLUMN IF NOT EXISTS "current_run_id" text; + + UPDATE "chat"."workbench_launch" + SET "current_run_id" = "instance_id" + WHERE "current_run_id" IS NULL; + + ALTER TABLE "chat"."workbench_launch" + ALTER COLUMN "current_run_id" SET NOT NULL; + + CREATE UNIQUE INDEX IF NOT EXISTS "workbench_launch_current_run_idx" + ON "chat"."workbench_launch" ("current_run_id"); + `, + }, ]; /** @@ -366,7 +388,7 @@ export async function listWorkbenchLaunchFoldedRunIds( const sql = postgres(databaseUrl, { max: 1, onnotice: () => undefined }); try { const rows = await sql.unsafe( - `SELECT "instance_id" AS "id", "tenant_id" AS "tenantId" FROM "chat"."workbench_launch"`, + `SELECT "current_run_id" AS "id", "tenant_id" AS "tenantId" FROM "chat"."workbench_launch"`, ); return rows.map((row) => ({ id: String(row["id"]), diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index ceac12e86..57700c221 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -12,8 +12,8 @@ import { createAgentLifecycle } from "@corbits/agent-lifecycle"; import { createCryptoProviderCache, domainOf, - findFoldedRunByAddress, findFoldedRunById, + launchFoldedRun, mintFoldedRun, readDefinitionProjection, readFoldedBody, @@ -22,12 +22,20 @@ import { resolveNewestProjectedDefinition, sendFoldedMail, wakeFoldedRun, - FoldedBodySchema, type FoldedRunMode, type FoldedRunsDeps, type SendFoldedMailParams, type SourcesOverride, } from "@corbits/folded-runs"; +import { + isBeyondWake, + readBindingByAddress, + repointBinding, + resolveLiveAgent, + resolveLiveByStableId, + type AgentBinding, + type LiveAgent, +} from "./agent-binding"; import type { FoldedBody } from "@intx/workflow-deploy"; import type { DB } from "@intx/db"; import { @@ -305,72 +313,157 @@ export function createHubChatPlatform( return deps.sidecarRouter.getRoutableAddresses().includes(address); } - // CL-6267: the sidecar's own park/wake handler now owns respawning a - // parked-but-still-announced deployment the moment mail routes to it - // — a routable address is never deployed or undeployed here, this - // just proceeds so `sendFoldedMail` can deliver straight to it. - // Redeploying over it would only trip the sidecar's "already - // deployed" bookkeeping for a resident it never actually stopped. - // Only a genuinely unroutable/unannounced address gets a real - // deploy, and a rejection from that deploy propagates honestly. - async function wakeByAddress(address: string): Promise { - if (isRoutable(address)) return; + /** + * A definition that declares no model of its own resolves the same + * catalog default at every deploy that `launchInvite` used to resolve + * at launch time — every deploy of such a run now goes through a wake + * or a relaunch (launches mint only), and a slept one always did. + */ + async function resolveFallbackModel( + binding: AgentBinding, + ): Promise { + if (binding.noopInference || binding.foldedBody.model !== null) { + return undefined; + } + const preferences = + (await deps.workbenchHostInferencePreferences?.(binding.tenantId)) ?? []; + return preferences[0]?.model; + } + + /** + * The per-deploy pins a binding carries, identical for a wake and a + * relaunch: a workbench host pins the noop inference source and the + * literal-input step mode, an invited agent resolves the tenant + * catalog with a fallback model when its definition declares none. + */ + async function deployShapeFor( + binding: AgentBinding, + ): Promise< + | { sources: SourcesOverride; mode: FoldedRunMode } + | { fallbackModel?: string } + > { + if (binding.noopInference) { + return { + sources: noopSourcesOverride( + deps.noopInferenceBaseUrl, + binding.foldedBody, + ), + mode: WORKBENCH_HOST_MODE, + }; + } + const fallbackModel = await resolveFallbackModel(binding); + return fallbackModel !== undefined ? { fallbackModel } : {}; + } - const run = await findFoldedRunByAddress(deps.db, address); - if (run === undefined || run.address === null) { - throw new Error(`No run found for address "${address}"`); + /** + * Replaces a run that died terminally with a genuinely fresh one. + * + * The dead run's durable event log is never reclaimed or erased — it + * stays on disk under its own address and readable through the + * platform's run routes, which is the audit trail a resurrection + * would have destroyed. What moves is the mapping: a new run id, a + * new address (the platform derives one from the other, so a fresh + * run cannot keep the old address), a new anchor row, and a new + * event log; `repointBinding` then swings the room's stable + * participant onto it. The room itself — timeline, settings, + * threads, participant records — never moves, because none of it was + * ever keyed on the run. + */ + async function relaunchTerminalRun(live: LiveAgent): Promise { + const { binding, run } = live; + if (run.definitionId === null) { + throw new Error( + `Cannot relaunch "${binding.stableId}": its run ${run.id} names no definition`, + ); } - const launchRows = await deps.db - .select() - .from(workbenchLaunch) - .where(eq(workbenchLaunch.instanceId, run.id)) - .limit(1); - const launchRow = launchRows[0]; - if (launchRow === undefined) { + const newRunId = generateId("workflowRun"); + const newAddress = formatRunAddress( + newRunId, + domainOf(binding.roomAddress), + ); + wakeLogger.info`relaunching ${binding.roomAddress}: run ${run.id} is terminal (${run.status}); minting fresh run ${newRunId}`; + + await launchFoldedRun(foldedRunsDeps, { + tenantId: binding.tenantId, + instanceId: newRunId, + triggerAddress: newAddress, + definitionId: run.definitionId, + foldedBody: binding.foldedBody, + launchLabel: "the relaunched instance", + ...(await deployShapeFor(binding)), + }); + + // After the deploy, never inside its transaction: a repoint that + // outlived a failed launch would leave the room addressing a run + // `launchFoldedRun` had already rolled back, and the next message + // would resolve nothing at all. + await repointBinding(deps.db, binding.stableId, newRunId); + lifecycle?.untrack(binding.liveAddress); + lifecycle?.track(newAddress); + return newAddress; + } + + /** + * 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. + * + * 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. + * + * 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 + * parked between messages — cannot be woken at all. Its address's + * 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. + */ + async function wakeByAddress(address: string): Promise { + const binding = await readBindingByAddress(deps.db, address); + if (binding === undefined) { throw new Error( - `No workbench_launch row for instance "${run.id}"; instances ` + + `No workbench_launch binding for address "${address}"; instances ` + `launched before launch-body persistence existed cannot be woken`, ); } - const parsedFoldedBody = FoldedBodySchema(launchRow.foldedBody); - if (parsedFoldedBody instanceof type.errors) { + const live = await resolveLiveAgent(deps.db, binding); + if (live === undefined || live.run.address === null) { throw new Error( - `workbench_launch row for instance "${run.id}" carries an invalid folded body: ${parsedFoldedBody.summary}`, + `No run found for address "${address}" (binding names run "${binding.currentRunId}")`, ); } - // A definition that declares no model of its own resolves the same - // catalog default here that `launchInvite` used to resolve at - // launch time — every deploy of such a run now goes through this - // wake path (launches mint only), and a slept one always did. - const fallbackModel = - !launchRow.noopInference && parsedFoldedBody.model === null - ? ((await deps.workbenchHostInferencePreferences?.( - launchRow.tenantId, - )) ?? [])[0]?.model - : undefined; + if (await isBeyondWake(deps.db, live.run)) { + await relaunchTerminalRun(live); + return; + } + if (isRoutable(live.run.address)) return; + const wakeParams = { - tenantId: launchRow.tenantId, - instanceId: run.id, - triggerAddress: run.address, - principalId: run.principalId, - foldedBody: parsedFoldedBody, + tenantId: binding.tenantId, + instanceId: live.run.id, + triggerAddress: live.run.address, + principalId: live.run.principalId, + foldedBody: binding.foldedBody, }; - await wakeFoldedRun( - foldedRunsDeps, - launchRow.noopInference - ? { - ...wakeParams, - sources: noopSourcesOverride( - deps.noopInferenceBaseUrl, - parsedFoldedBody, - ), - mode: WORKBENCH_HOST_MODE, - } - : { - ...wakeParams, - ...(fallbackModel !== undefined ? { fallbackModel } : {}), - }, - ); + await wakeFoldedRun(foldedRunsDeps, { + ...wakeParams, + ...(await deployShapeFor(binding)), + }); + } + + /** + * The live run mail must actually be delivered to for a stable + * participant id — not the room's own address, once anything has been + * relaunched. + */ + async function requireLive(stableId: string): Promise { + const live = await resolveLiveByStableId(deps.db, stableId); + if (live === undefined) { + throw new Error(`No live workbench run for "${stableId}"`); + } + return live; } /** @@ -468,6 +561,7 @@ export function createHubChatPlatform( await tx.insert(workbenchLaunch).values({ tenantId: input.tenantId, instanceId: input.workbenchId, + currentRunId: input.workbenchId, foldedBody, createdAt: new Date(), noopInference: true, @@ -572,6 +666,7 @@ export function createHubChatPlatform( await tx.insert(workbenchLaunch).values({ tenantId: input.tenantId, instanceId, + currentRunId: instanceId, foldedBody, createdAt: new Date(), noopInference: false, @@ -603,8 +698,10 @@ export function createHubChatPlatform( }, async resolveDefinitionIdByAddress(address): Promise { - const run = await findFoldedRunByAddress(deps.db, address); - return run?.definitionId ?? undefined; + const binding = await readBindingByAddress(deps.db, address); + if (binding === undefined) return undefined; + const live = await resolveLiveAgent(deps.db, binding); + return live?.run.definitionId ?? undefined; }, async refreshAgentInstanceFromDefinition( @@ -612,12 +709,15 @@ export function createHubChatPlatform( _workbenchId, address, ): Promise { - const run = await findFoldedRunByAddress(deps.db, address); - if (run === undefined || run.definitionId === null) return; + const binding = await readBindingByAddress(deps.db, address); + if (binding === undefined) return; + const live = await resolveLiveAgent(deps.db, binding); + const definitionId = live?.run.definitionId; + if (definitionId === undefined || definitionId === null) return; const definitionRow = await deps.db.query.workflowDefinition.findFirst({ where: and( - eq(workflowDefinition.id, run.definitionId), + eq(workflowDefinition.id, definitionId), eq(workflowDefinition.tenantId, tenantId), ), }); @@ -634,17 +734,15 @@ export function createHubChatPlatform( await deps.db .update(workbenchLaunch) .set({ foldedBody }) - .where(eq(workbenchLaunch.instanceId, run.id)); + .where(eq(workbenchLaunch.instanceId, binding.stableId)); }, async sendMail(input): Promise { - const run = await findFoldedRunById(deps.db, input.workbenchId); - if (run === undefined) { - throw new Error(`No workbench run for "${input.workbenchId}"`); - } - if (run.address === null) { - throw new Error(`Workbench run "${input.workbenchId}" has no address`); - } + // The stable id names the room's participant; the run it resolves + // to is whichever one is alive right now, which is a different + // run (and a different address) after every relaunch. + const { binding, run } = await requireLive(input.workbenchId); + const liveAddress = binding.liveAddress; // Wake before send: a sleeping instance (the lifecycle package's // own sweep) or one that never came back up after a stack @@ -661,30 +759,32 @@ export function createHubChatPlatform( // undeploys anything for a routable address, it just proceeds to // send. Only a genuinely unroutable address gets an explicit // wake here. + // + // CL-6365: a relaunch can happen inside this wake, minting a + // fresh run at a fresh address. Re-resolve afterwards so the + // send that follows targets the run that is actually alive, not + // the one that just died. if (lifecycle !== undefined) { - await lifecycle.ensureAwake(run.address); - } else if (!isRoutable(run.address)) { - await wakeByAddress(run.address); + await lifecycle.ensureAwake(liveAddress); + } else if (!isRoutable(liveAddress)) { + await wakeByAddress(liveAddress); } + const delivery = await requireLive(input.workbenchId); + const deliveryAddress = delivery.binding.liveAddress; // Tracking here (not only at launch) brings instances that were // already resident before this hub process started — restored by // a sidecar reconnect, launched by an earlier run — under the // idle sweep the moment they see traffic. - lifecycle?.track(run.address); + lifecycle?.track(deliveryAddress); - const sessionId = await resolveFoldedRunSessionId(deps.db, run); - const domain = domainOf(run.address); + const sessionId = await resolveFoldedRunSessionId(deps.db, delivery.run); + const domain = domainOf(deliveryAddress); let from: string; let originAddress: string | undefined; if (input.fromWorkbenchId !== undefined) { - const origin = await findFoldedRunById(deps.db, input.fromWorkbenchId); - if (origin?.address == null) { - throw new Error( - `Origin workbench "${input.fromWorkbenchId}" has no address`, - ); - } - from = origin.address; - originAddress = origin.address; + const origin = await requireLive(input.fromWorkbenchId); + from = origin.binding.liveAddress; + originAddress = origin.binding.liveAddress; } else if (input.principalId !== undefined) { from = `${input.principalId}@${domain}`; } else { @@ -705,7 +805,7 @@ export function createHubChatPlatform( const sendMailBase = { tenantId: input.tenantId, sessionId, - agentAddress: run.address, + agentAddress: deliveryAddress, from, domain, content: input.content.content, @@ -721,7 +821,7 @@ export function createHubChatPlatform( : withAttachments, ); - lifecycle?.recordActivity(run.address); + lifecycle?.recordActivity(deliveryAddress); if (originAddress !== undefined) lifecycle?.recordActivity(originAddress); return sent; @@ -731,10 +831,7 @@ export function createHubChatPlatform( // Blobs are only readable when the mail row lives on this workbench's // session. Looking up by mail id alone let any authenticated caller // read another tenant's attachment by guessing a blob id. - const run = await findFoldedRunById(deps.db, workbenchId); - if (run === undefined) { - throw new Error(`No workbench run for "${workbenchId}"`); - } + const { run } = await requireLive(workbenchId); const sessionId = await resolveFoldedRunSessionId(deps.db, run); const match = /^blob_(.+?)_(\d[\d.]*)$/.exec(blobId); @@ -764,11 +861,11 @@ export function createHubChatPlatform( let cancelled = false; let unsubscribeAgent: (() => void) | undefined; - void findFoldedRunById(deps.db, workbenchId) - .then((run) => { - if (cancelled || run === undefined || run.address === null) return; + void resolveLiveByStableId(deps.db, workbenchId) + .then((live) => { + if (cancelled || live === undefined) return; unsubscribeAgent = deps.sidecarRouter.subscribeAgent( - run.address, + live.binding.liveAddress, (event) => { onEvent({ type: "chat.agent", data: event }); }, @@ -788,12 +885,20 @@ export function createHubChatPlatform( }, async ensureAwake(address: string): Promise { + // The caller may hold either side of the mapping (the hub's + // undelivered-mail handler holds whatever the envelope named), so + // the lifecycle is driven on the LIVE address it resolves to — + // that is the only address the sidecar ever announces. + const binding = await readBindingByAddress(deps.db, address); + if (binding === undefined) { + throw new Error(`No workbench_launch binding for address "${address}"`); + } if (lifecycle !== undefined) { - await lifecycle.ensureAwake(address); + await lifecycle.ensureAwake(binding.liveAddress); return; } - if (isRoutable(address)) return; - await wakeByAddress(address); + if (isRoutable(binding.liveAddress)) return; + await wakeByAddress(binding.liveAddress); }, }; diff --git a/packages/chat/src/schema.ts b/packages/chat/src/schema.ts index a54fd6760..7b19aca5b 100644 --- a/packages/chat/src/schema.ts +++ b/packages/chat/src/schema.ts @@ -85,10 +85,28 @@ export const workbenchReadState = chatSchema.table( * definition exists nowhere else (its workflow asset is never pushed * a workflow.json), so this row is the single wake-time source for * both launch kinds. + * + * This row is also the address→run mapping the whole room depends on. + * `instanceId` is the STABLE participant id — the room's own workbench + * id for a host, the id an invited agent was first minted under — and + * `formatRunAddress(instanceId, domain)` is the address the room + * addresses this agent by forever (participant records, message + * `senderAddress`, mention handles). `currentRunId` is the run actually + * executing behind it. The two are equal until the first relaunch; + * after a run dies terminally (a mid-turn crash), `currentRunId` is + * re-pointed at a FRESH run with a fresh id, a fresh address, and a + * fresh durable event log, while the stable id — and therefore the + * room, its timeline, its settings, and every participant record — + * does not move. See `./agent-binding.ts`. */ export const workbenchLaunch = chatSchema.table("workbench_launch", { tenantId: text("tenant_id").notNull(), instanceId: text("instance_id").primaryKey(), + /** + * The live `workflow_run.id` this stable id currently resolves to. + * Unique: one run backs at most one room participant. + */ + currentRunId: text("current_run_id").notNull().unique(), foldedBody: jsonb("folded_body").notNull(), createdAt: timestamp("created_at", { withTimezone: true }) .notNull()