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.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/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() 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({