diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index c2f258d36..911828592 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -76,6 +76,7 @@ import { createDrizzleWriteClaimStore, createHubChatPlatform, createNoopInferenceRoutes, + createRelaunchNoticePoster, findExistingAgentChat, createWorkflowParticipantRoutes, isWorkbenchHostDefinitionName, @@ -85,6 +86,7 @@ import { startWorkflowCommand, sendWorkbenchMessage, } from "@corbits/chat"; +import type { RelaunchNoticePort } from "@corbits/chat"; import type { FinalizedTurnToolCall } from "@corbits/turn-artifacts"; import { createCryptoProviderCache, @@ -1042,6 +1044,12 @@ export async function createHub(config: HubConfig) { createWorkbenchHostInferencePreferencesResolver((tenantId) => listDefaultInferencePreferences(db, tenantId), ); + // Where a relaunch announces itself in the room (see `@corbits/chat`'s + // `relaunch-notice.ts`). Armed further down, once the room-message + // store the poster writes through exists — the platform that fires + // notices has to be constructed first, since the sweep that triggers + // most of them hangs off it. + const relaunchNoticeRef: RelaunchNoticePort = {}; const chatPlatform = createHubChatPlatform({ db, sessionService, @@ -1067,6 +1075,7 @@ export async function createHub(config: HubConfig) { // tenant-catalog default, instead of 409ing `not_launchable`. workbenchHostInferencePreferences: workbenchHostInferencePreferencesResolver, + relaunchNotice: relaunchNoticeRef, }); wireMailRedelivery({ sidecarRouter, chatPlatform }); // The one SSE subscriber registry for this process's workbench events @@ -1093,6 +1102,11 @@ export async function createHub(config: HubConfig) { // The room timeline store (CL-6327): a workbench's own messages, held // as workbench data rather than platform mail. const roomMessages = createDrizzleRoomMessageStore(db); + relaunchNoticeRef.current = createRelaunchNoticePoster({ + store: chatStore, + roomMessages, + publish: workbenchSubscribers.publish, + }); // Built once, beside the platform, for the process's lifetime: turns // an invited agent's `connector.reply` events into workbench messages, // and a gate-blocked run's approval park into an in-chat approve @@ -1120,6 +1134,53 @@ export async function createHub(config: HubConfig) { chatOrchestratorDeps.memory = memoryHandle.memory; } const chatOrchestrator = createChatOrchestrator(chatOrchestratorDeps); + // A room participant that died with its sidecar is otherwise silently + // dead until somebody writes into it, and the turn the crash + // interrupted never surfaces at all — the run that died never sends + // the `message.run.ended` the orchestrator's turn-drop notice hangs + // off. The sweep finds those runs and relaunches each one, posting + // its notice. + // + // A series of passes rather than one, because "this run is dead" is + // not knowable at the instant the execution plane comes back: the + // terminal event is committed to the run's durable log by the dying + // sidecar and reaches `workflow_run.status` only once the restarted + // sidecar has packed it back to the hub, seconds later. The series is + // bounded and re-armed by a sidecar disconnect, which is the one + // event that can newly orphan a room. + const relaunchSweepLog = getLogger(["chat", "relaunch-sweep"]); + const RELAUNCH_SWEEP_DELAYS_MS = [0, 2_000, 5_000, 15_000, 45_000]; + // Bumped by every reschedule so a pass still in flight from the + // previous series retires instead of continuing beside the new one. + let relaunchSweepSeries = 0; + let relaunchSweepTimer: ReturnType | undefined; + function runNextRelaunchSweepPass(series: number, pass: number): void { + const delay = RELAUNCH_SWEEP_DELAYS_MS[pass]; + if (delay === undefined || series !== relaunchSweepSeries) return; + const timer = setTimeout(() => { + void chatPlatform + .sweepTerminalRuns() + .catch((cause: unknown) => { + relaunchSweepLog.error`relaunch sweep pass failed: ${ + cause instanceof Error ? cause.message : String(cause) + }`; + }) + .finally(() => { + runNextRelaunchSweepPass(series, pass + 1); + }); + }, delay); + timer.unref?.(); + relaunchSweepTimer = timer; + } + function scheduleRelaunchSweep(): void { + clearTimeout(relaunchSweepTimer); + relaunchSweepSeries += 1; + runNextRelaunchSweepPass(relaunchSweepSeries, 0); + } + scheduleRelaunchSweep(); + sidecarRouter.events.on("sidecar.disconnect", () => { + scheduleRelaunchSweep(); + }); // Now that `chatStore`/`chatPlatform` exist, arm the finalized-turn // artifact-delivery ref declared beside `eventCollectors` above. // `memory` (absent when the plane isn't mounted) lets this handler diff --git a/docs/revendor-inventory.md b/docs/revendor-inventory.md index 912ba1f58..5a012763b 100644 --- a/docs/revendor-inventory.md +++ b/docs/revendor-inventory.md @@ -1040,23 +1040,87 @@ 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. +### Closing it: the sweep, the notice, and the old attachments + +Three things stood between the relaunch and a reader actually +experiencing it. + +**The sweep.** The relaunch was send-triggered — it fired on the next +`sendMail` through the wake choke point — so a room whose agent died in +a crash stayed silently dead until somebody wrote into it, which is +precisely what nobody does when the room looks broken. +`createHubChatPlatform.sweepTerminalRuns` scans `workbench_launch` +(bounded at 100 rows), keeps the participants `isBeyondWake` says are +terminal-and-not-merely-parked, and relaunches each through the same +path a send would; every relaunch and every failure is logged under +`chat·wake`, and one failure never aborts the pass. + +The hub runs it at boot and re-arms it on `sidecar.disconnect`, as a +short bounded series of passes rather than a single one. That is not +belt-and-braces: "this run is dead" is not knowable at the instant the +execution plane comes back. The dying sidecar commits the terminal event +to the run's own durable log, and `workflow_run.status` only follows once +the RESTARTED sidecar has packed that log back to the hub — seconds +after the reconnect. A single pass at boot would look at a run still +marked `running` and find nothing. + +**The notice.** `chat-orchestrator.ts`'s turn-drop notice hangs off +`message.run.ended`, which a killed sidecar never sends — so the +interrupted turn had no way to surface at all. `relaunch-notice.ts` is +the other half: `createHubChatPlatform` takes a notice port (a ref, since +the adapter is built before the room-message store it posts through), the +hub arms it beside `roomMessages`, and every relaunch — swept or +send-triggered — posts into each room the replaced participant belongs +to, under the STABLE address the room has always known it by, so the +notice reads as the same teammate rather than a stranger. The wording is +cause-aware (crashed / stopped / ended) and stays in the reader's +language: the turn didn't finish, it's back now, say it again. + +**Pre-relaunch attachments.** `fetchBlob` read through the live run's +session, and a folded run's mail session is resolved from its PRINCIPAL +— a fresh run has a fresh principal, so every attachment sent before the +crash became unreachable. `workbench_launch.prior_run_ids` records each +run as it is retired (capped at 20), and `fetchBlob` walks those +sessions newest-first after the live one. Room messages were never +affected: they are `chat.workbench_messages` rows, not mail. + +**The audit assertion.** Proof 4 now ends by reading the replaced run's +events back through the ordinary run routes AFTER its replacement is +already answering. That is the fresh-run ruling's whole justification, +and it is now a hop rather than an argument. + +### And what re-running proof 4 found: the detection signal does not fire + +Proof 4 was re-run on a real stack (scratch database, real signup, real +Ollama, both shapes). Proofs 1, 2, 2b and 3 are green; proof 4's restart +and boot restore are green at 9.2s. Hop 3 — "the turn the kill +interrupted surfaces visibly" — is still red, and the sweep is not why. +Nothing in the sweep's log fired at all, because it found nothing to +sweep: + +``` +run_10dd09d1… failed <- the SECTION deployment +run_4de07624… running <- the folded CHAT run, after the mid-turn kill +``` + +The claim recorded above — that both top-level runs end up +`workflow_run.status = 'failed'` — holds for the section deployment +only. The folded chat run's terminal event goes to its durable log +(which is why its supervisor rejects the next message) while its +`workflow_run` row stays `running` forever. `isBeyondWake` reads that +row, so for the shape hop 3 measures it answers "still alive" and +neither the sweep nor the send-triggered relaunch ever fires. The room +keeps the reader's message with nothing after it — the exact silent +drop the hop asserts against. + +So the remaining gap is the DETECTION signal, not the relaunch: the hub +needs to learn a folded run's lifecycle from the same durable log the +supervisor reads (`readWorkflowRunLifecycle`) rather than from +`workflow_run.status`, either by reconciling the row when a restarted +sidecar packs a terminal log back, or by having `isBeyondWake` consult +the log directly. Everything downstream of that signal — sweep, notice, +repoint, attachment history, audit read — is built and unit-proven. + +The section shape stays red for its own separate reason: a plain +workflow deployment has no room, so nothing maps a stable id onto a +fresh run for it. That is out of CL-6365's scope, which is the room. diff --git a/packages/chat/src/agent-binding.test.ts b/packages/chat/src/agent-binding.test.ts index 11b37d205..821938dbb 100644 --- a/packages/chat/src/agent-binding.test.ts +++ b/packages/chat/src/agent-binding.test.ts @@ -21,6 +21,7 @@ type LaunchRow = { tenantId: string; instanceId: string; currentRunId: string; + priorRunIds: string[]; foldedBody: unknown; noopInference: boolean; }; @@ -69,6 +70,7 @@ const relaunched: LaunchRow = { tenantId: "ten_1", instanceId: "run_original", currentRunId: "run_fresh", + priorRunIds: ["run_original"], foldedBody: FOLDED_BODY, noopInference: false, }; diff --git a/packages/chat/src/agent-binding.ts b/packages/chat/src/agent-binding.ts index 9ca65329c..a871bca8f 100644 --- a/packages/chat/src/agent-binding.ts +++ b/packages/chat/src/agent-binding.ts @@ -40,10 +40,33 @@ export interface AgentBinding { readonly roomAddress: string; readonly currentRunId: string; readonly liveAddress: string; + /** Every run this participant used to be, oldest first. */ + readonly priorRunIds: readonly string[]; readonly foldedBody: FoldedBody; readonly noopInference: boolean; } +/** + * How far back `prior_run_ids` remembers. Long enough that a room can + * survive a bad afternoon and still hand back an attachment from + * before it, short enough that the column never becomes an unbounded + * append log on a row read on every single message. + */ +const PRIOR_RUN_HISTORY_LIMIT = 20; + +const PriorRunIdsSchema = type("string[]"); + +function priorRunIdsFrom(row: LaunchRow): readonly string[] { + const parsed = PriorRunIdsSchema(row.priorRunIds); + if (parsed instanceof type.errors) { + throw new Error( + `workbench_launch row for "${row.instanceId}" carries an invalid ` + + `prior-run history: ${parsed.summary}`, + ); + } + return parsed; +} + /** The live `workflow_run` row behind a binding, plus the binding itself. */ export interface LiveAgent { readonly binding: AgentBinding; @@ -73,6 +96,7 @@ function bindingFrom(row: LaunchRow, domain: string): AgentBinding { roomAddress: formatRunAddress(row.instanceId, domain), currentRunId: row.currentRunId, liveAddress: formatRunAddress(row.currentRunId, domain), + priorRunIds: priorRunIdsFrom(row), foldedBody: parsed, noopInference: row.noopInference, }; @@ -166,6 +190,25 @@ export async function resolveLiveByStableId( return { binding: bindingFrom(row, requireDomain(run.address)), run }; } +/** + * The runs this participant used to be, newest first — the order + * `fetchBlob` walks them in, so the most recently retired session is + * tried before older ones. A retired run whose row has since been + * deleted is skipped rather than raising: history that no longer + * exists is not an error, it is just history that cannot answer. + */ +export async function readPriorRuns( + db: DB["db"], + binding: AgentBinding, +): Promise { + const runs: LiveAgent["run"][] = []; + for (const runId of [...binding.priorRunIds].reverse()) { + const run = await readRun(db, runId); + if (run !== undefined) runs.push(run); + } + return runs; +} + /** * 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 @@ -196,18 +239,45 @@ export async function isBeyondWake( } /** - * 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. + * Re-points a stable participant at a freshly launched run, retiring + * the run it was pointing at into `priorRunIds`. 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, + binding: AgentBinding, newRunId: string, ): Promise { + const history = [...binding.priorRunIds, binding.currentRunId].slice( + -PRIOR_RUN_HISTORY_LIMIT, + ); await db .update(workbenchLaunch) - .set({ currentRunId: newRunId }) - .where(eq(workbenchLaunch.instanceId, stableId)); + .set({ currentRunId: newRunId, priorRunIds: history }) + .where(eq(workbenchLaunch.instanceId, binding.stableId)); +} + +/** + * Every participant whose current run is beyond waking, for the boot + * sweep that relaunches them (`platform-adapter.ts`'s + * `sweepTerminalRuns`). Bounded by `limit` rather than streaming the + * whole table: a sweep is a best-effort recovery pass at start-up, not + * a migration, and an unbounded one on a large tenant would turn every + * boot into a deploy storm. + */ +export async function listLaunchesBeyondWake( + db: DB["db"], + limit: number, +): Promise { + const rows = await db.select().from(workbenchLaunch).limit(limit); + const dead: LiveAgent[] = []; + for (const row of rows) { + const run = await readRun(db, row.currentRunId); + if (run === undefined || run.address === null) continue; + if (!(await isBeyondWake(db, run))) continue; + dead.push({ binding: bindingFrom(row, requireDomain(run.address)), run }); + } + return dead; } diff --git a/packages/chat/src/index.ts b/packages/chat/src/index.ts index 146bdb450..a748b32c3 100644 --- a/packages/chat/src/index.ts +++ b/packages/chat/src/index.ts @@ -261,6 +261,12 @@ export type { HubChatPlatform, } from "./platform-adapter"; +export { + createRelaunchNoticePoster, + relaunchNoticeText, +} from "./relaunch-notice"; +export type { RelaunchNotice, RelaunchNoticePort } from "./relaunch-notice"; + export { createDrizzleRoomMessageStore, createInMemoryRoomMessageStore, diff --git a/packages/chat/src/migrations.ts b/packages/chat/src/migrations.ts index 80df627f2..a8fad3b84 100644 --- a/packages/chat/src/migrations.ts +++ b/packages/chat/src/migrations.ts @@ -361,6 +361,13 @@ export const chatMigrations: readonly ChatMigration[] = [ ON "chat"."workbench_launch" ("current_run_id"); `, }, + { + name: "0021_workbench_launch_prior_runs", + sql: ` + ALTER TABLE "chat"."workbench_launch" + ADD COLUMN IF NOT EXISTS "prior_run_ids" jsonb NOT NULL DEFAULT '[]'::jsonb; + `, + }, ]; /** diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index 57700c221..f908a2f96 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -29,13 +29,16 @@ import { } from "@corbits/folded-runs"; import { isBeyondWake, + listLaunchesBeyondWake, readBindingByAddress, + readPriorRuns, repointBinding, resolveLiveAgent, resolveLiveByStableId, type AgentBinding, type LiveAgent, } from "./agent-binding"; +import type { RelaunchNoticePort } from "./relaunch-notice"; import type { FoldedBody } from "@intx/workflow-deploy"; import type { DB } from "@intx/db"; import { @@ -145,6 +148,13 @@ export type CreateHubChatPlatformDeps = { workbenchHostInferencePreferences?: ( tenantId: string, ) => Promise; + /** + * Where a relaunch announces itself in the room — see + * `./relaunch-notice.ts` for why this is a ref the host arms later + * rather than a callback passed in here. Absent (or never armed), a + * relaunch happens silently. + */ + relaunchNotice?: RelaunchNoticePort; }; // Workbench-host asset naming lives in `./workbench-host-naming` — a @@ -161,6 +171,15 @@ export type CreateHubChatPlatformDeps = { const NOOP_INFERENCE_SOURCE_ID = "noop"; const NOOP_INFERENCE_MODEL_FALLBACK = "noop"; +/** + * How many launch rows one relaunch sweep will look at. A sweep is a + * best-effort recovery pass, and every relaunch it performs is a real + * sidecar deploy — an unbounded one would turn a boot after a bad night + * into a deploy storm. Rooms past the bound still recover the moment + * somebody writes into them, through the same send-triggered path. + */ +const RELAUNCH_SWEEP_LIMIT = 100; + /** * The `SourcesOverride` every workbench-HOST launch and wake pins * instead of resolving against the tenant catalog (see @@ -231,6 +250,14 @@ export type HubChatPlatform = ChatPlatform & { * "not mine to wake", not a bug. */ ensureAwake(address: string): Promise; + /** + * Relaunches every room participant whose run is beyond waking, and + * posts each one's notice. The host runs this at boot and again + * whenever the execution plane has come back, since a run that died + * with its sidecar is only discoverable once the hub has ingested + * that run's terminal event. + */ + sweepTerminalRuns(): Promise<{ scanned: number; relaunched: number }>; }; /** @@ -397,12 +424,57 @@ export function createHubChatPlatform( // 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); + await repointBinding(deps.db, binding, newRunId); lifecycle?.untrack(binding.liveAddress); lifecycle?.track(newAddress); + + // The turn that died with the old run never sent + // `message.run.ended`, so the orchestrator's turn-drop notice can + // never fire for it. This is the only thing that tells the reader + // their message was not silently swallowed. + deps.relaunchNotice?.current?.({ + tenantId: binding.tenantId, + roomAddress: binding.roomAddress, + deadRunId: run.id, + deadRunStatus: run.status, + newRunId, + }); return newAddress; } + /** + * Replaces every participant whose run died while nothing was + * watching. A send-triggered relaunch only fires when somebody writes + * into the room; a room whose agent died in a crash would otherwise + * stay silently dead until then, with the interrupted turn never + * surfacing at all. + * + * Best-effort and bounded: one failed relaunch is logged and the + * sweep moves on, because a boot that aborts on the first + * unrelaunchable room leaves every room after it dead too. + */ + async function sweepTerminalRuns(): Promise<{ + scanned: number; + relaunched: number; + }> { + const dead = await listLaunchesBeyondWake(deps.db, RELAUNCH_SWEEP_LIMIT); + let relaunched = 0; + for (const live of dead) { + try { + await relaunchTerminalRun(live); + relaunched += 1; + } catch (cause: unknown) { + wakeLogger.error`relaunch sweep: could not relaunch ${live.binding.roomAddress} (run ${live.run.id} is ${live.run.status}): ${ + cause instanceof Error ? cause.message : String(cause) + }`; + } + } + if (dead.length > 0) { + wakeLogger.info`relaunch sweep: ${String(relaunched)} of ${String(dead.length)} dead room participants relaunched`; + } + return { scanned: dead.length, relaunched }; + } + /** * Brings the run behind `address` back to routable, whichever kind of * "not routable" it is in. `address` may be either side of the @@ -466,6 +538,25 @@ export function createHubChatPlatform( return live; } + /** + * The mail sessions this participant used to hold, newest first. A + * retired run whose principal never got a session (a launch that + * rolled back before one existed) has nothing to contribute and is + * skipped — that is history with no mail in it, not a failure to + * read the blob the caller asked for. + */ + async function retiredSessionIds(binding: AgentBinding): Promise { + const sessionIds: string[] = []; + for (const run of await readPriorRuns(deps.db, binding)) { + try { + sessionIds.push(await resolveFoldedRunSessionId(deps.db, run)); + } catch { + continue; + } + } + return sessionIds; + } + /** * `sendFoldedMail` delivers synchronously against the sidecar's * current routable set — the same in-memory index `isRoutable` reads @@ -828,12 +919,6 @@ export function createHubChatPlatform( }, async fetchBlob(workbenchId, blobId): Promise { - // 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 requireLive(workbenchId); - const sessionId = await resolveFoldedRunSessionId(deps.db, run); - const match = /^blob_(.+?)_(\d[\d.]*)$/.exec(blobId); if (match === null) { throw new Error(`Invalid blob id "${blobId}"`); @@ -842,16 +927,33 @@ export function createHubChatPlatform( if (mailId === undefined || partPath === undefined) { throw new Error(`Invalid blob id "${blobId}"`); } - const mailRow = await deps.db.query.sessionMail.findFirst({ - where: and( - eq(sessionMail.id, mailId), - eq(sessionMail.sessionId, sessionId), - ), - }); - if (mailRow === undefined) { - throw new Error(`No mail "${mailId}" for blob "${blobId}"`); + + // Blobs are only readable when the mail row lives on a session + // this participant has actually held. Looking up by mail id alone + // let any authenticated caller read another tenant's attachment + // by guessing a blob id. + // + // "Held", not "holds": a relaunch mints a fresh run with a fresh + // principal, and a folded run's mail session hangs off its + // principal — so an attachment sent before the crash lives on a + // session the live run has never seen. Walking the retired runs + // newest-first is what keeps yesterday's attachment downloadable + // after today's relaunch. + const { binding, run } = await requireLive(workbenchId); + const liveSessionId = await resolveFoldedRunSessionId(deps.db, run); + const priorSessionIds = await retiredSessionIds(binding); + for (const sessionId of [liveSessionId, ...priorSessionIds]) { + const mailRow = await deps.db.query.sessionMail.findFirst({ + where: and( + eq(sessionMail.id, mailId), + eq(sessionMail.sessionId, sessionId), + ), + }); + if (mailRow !== undefined) { + return extractPartByPath(mailRow.raw, partPath); + } } - return extractPartByPath(mailRow.raw, partPath); + throw new Error(`No mail "${mailId}" for blob "${blobId}"`); }, subscribeToWorkbench( @@ -904,5 +1006,6 @@ export function createHubChatPlatform( return Object.assign(platform, { recordActivity: (address: string) => lifecycle?.recordActivity(address), + sweepTerminalRuns, }); } diff --git a/packages/chat/src/relaunch-notice.ts b/packages/chat/src/relaunch-notice.ts new file mode 100644 index 000000000..56063513f --- /dev/null +++ b/packages/chat/src/relaunch-notice.ts @@ -0,0 +1,103 @@ +// What the room is told when the teammate it was talking to had to be +// replaced (see `agent-binding.ts` for why a relaunch mints a fresh run +// rather than resurrecting the dead one). +// +// A relaunch is invisible by construction: the run that died mid-turn +// never sends the `message.run.ended` event `chat-orchestrator.ts`'s +// turn-drop notice hangs off, so without this the reader's message was +// accepted and then silently swallowed. The notice closes that hole +// from the other side — it is posted by whoever performs the relaunch, +// whether the boot sweep found the dead run or the next send did. +// +// The port is a ref rather than a plain callback because the platform +// adapter is constructed before the room-message store it needs to +// post through; `apps/hub` arms it once both exist. Unarmed, every +// relaunch is silent — which is exactly the behavior before this +// existed, not a fallback beside a live path. + +import { getLogger } from "@intx/log"; +import { localPartOf } from "./agent-address"; +import { parseParticipants } from "./participants"; +import { postRoomMessage, type RoomMessageStore } from "./room-messages"; +import type { ChatStore } from "./store"; +import type { WorkbenchSubscriberRegistry } from "./workbench-events"; + +const log = getLogger(["chat", "relaunch-notice"]); + +/** The relaunch a room is being told about. */ +export type RelaunchNotice = { + readonly tenantId: string; + /** The stable participant address the room knows this agent by. */ + readonly roomAddress: string; + readonly deadRunId: string; + /** The dead run's own terminal `workflow_run.status`. */ + readonly deadRunStatus: string; + readonly newRunId: string; +}; + +export type RelaunchNoticePort = { + current?: (notice: RelaunchNotice) => void; +}; + +/** + * The line the agent says in its own voice, in the reader's language + * rather than the system's — never "run", "terminal", or "relaunch". + * Cause-aware, because the three ways a turn can be lost read + * differently to the person who was waiting on it: a crash cut the + * answer off, a cancel stopped it, and anything else simply ended + * before it was done. + */ +export function relaunchNoticeText(deadRunStatus: string): string { + const cause = + deadRunStatus === "failed" + ? "I got cut off partway through that last one and never finished it." + : deadRunStatus === "cancelled" || deadRunStatus === "canceled" + ? "That last one was stopped before I finished it." + : "I shut down before I finished that last one."; + return `${cause} I'm back now — send it again and I'll pick it up.`; +} + +/** + * Posts each relaunch notice into every room the replaced participant + * belongs to, in that participant's own voice and under the stable + * address the room has always known it by — so the notice lands in the + * same conversation thread as the message it is apologizing for, not + * as a message from a stranger. + * + * Returns the synchronous port shape `createHubChatPlatform` calls: + * a relaunch must never be held up (or undone) by a timeline write, so + * a failed post is logged and the fresh run still goes on serving. + */ +export function createRelaunchNoticePoster(deps: { + readonly store: Pick; + readonly roomMessages: RoomMessageStore; + readonly publish: WorkbenchSubscriberRegistry["publish"]; +}): (notice: RelaunchNotice) => void { + async function post(notice: RelaunchNotice): Promise { + const workbenches = await deps.store.listWorkbenchSettings(notice.tenantId); + const rooms = workbenches.filter((workbench) => + parseParticipants(workbench.settings["chat/participants"]).some( + (participant) => participant.address === notice.roomAddress, + ), + ); + for (const room of rooms) { + await postRoomMessage(deps, { + tenantId: notice.tenantId, + workbenchId: room.workbenchId, + sender: { name: null, address: notice.roomAddress }, + parts: [ + { kind: "text", text: relaunchNoticeText(notice.deadRunStatus) }, + ], + runId: localPartOf(notice.roomAddress), + }); + } + } + + return (notice) => { + void post(notice).catch((cause: unknown) => { + log.error`failed to post ${notice.roomAddress}'s relaunch notice (run ${notice.deadRunId} -> ${notice.newRunId}): ${ + cause instanceof Error ? cause.message : String(cause) + }`; + }); + }; +} diff --git a/packages/chat/src/schema.ts b/packages/chat/src/schema.ts index 7b19aca5b..23e42f68f 100644 --- a/packages/chat/src/schema.ts +++ b/packages/chat/src/schema.ts @@ -107,6 +107,15 @@ export const workbenchLaunch = chatSchema.table("workbench_launch", { * Unique: one run backs at most one room participant. */ currentRunId: text("current_run_id").notNull().unique(), + /** + * Every run that used to be `currentRunId`, oldest first. A relaunch + * mints a fresh run with its own principal, and a folded run's mail + * session is resolved from its principal — so an attachment a reader + * uploaded before the crash lives on a session the live run cannot + * see. This is the trail `fetchBlob` walks back through so a blob + * posted to the room yesterday is still downloadable today. + */ + priorRunIds: jsonb("prior_run_ids").notNull().default([]), 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 93d08b438..83c1867d3 100644 --- a/packages/chat/test/chat-orchestrator.test.ts +++ b/packages/chat/test/chat-orchestrator.test.ts @@ -129,6 +129,7 @@ function launchRowFor(runId: string, tenantId: string) { tenantId, instanceId: runId, currentRunId: runId, + priorRunIds: [], foldedBody: { systemPrompt: "be helpful", toolPackagePins: [], diff --git a/packages/chat/test/platform-adapter.test.ts b/packages/chat/test/platform-adapter.test.ts index 02972440c..f06c99498 100644 --- a/packages/chat/test/platform-adapter.test.ts +++ b/packages/chat/test/platform-adapter.test.ts @@ -305,6 +305,7 @@ function createFakeDb(opts: { tenantId: "ten_1", instanceId: opts.workflowRunRow.id, currentRunId: opts.workflowRunRow.id, + priorRunIds: [], foldedBody: { systemPrompt: "be helpful", toolPackagePins: [], @@ -319,12 +320,14 @@ function createFakeDb(opts: { const withCurrent = row as { instanceId: string; currentRunId?: string; + priorRunIds?: string[]; }; return selectChain([ { ...withCurrent, currentRunId: withCurrent.currentRunId ?? withCurrent.instanceId, + priorRunIds: withCurrent.priorRunIds ?? [], }, ]); } @@ -2524,3 +2527,122 @@ describe("createHubChatPlatform", () => { }); }); }); + +// CL-6365: the send-triggered relaunch only fires when somebody writes +// into the room. A room whose agent died in a crash has nobody writing +// into it — that is the whole failure — so the sweep is what makes the +// interrupted turn surface at all. +describe("createHubChatPlatform relaunch sweep", () => { + const DEAD_ROOM_FOLDED_BODY = { + systemPrompt: "be helpful", + toolPackagePins: [], + grantRequirements: [], + credentialBindings: [], + model: null, + }; + + function createSweepFixture(opts: { runStatus: string; parked: boolean }) { + const db = createFakeDb({ + assetRow: { + tenantId: "ten_1", + creatorPrincipalId: "prin_creator", + name: "workbench-1", + displayName: null, + }, + definitionId: "wfd_room1", + workflowRunRow: { + id: "run_dead", + address: "run_dead@ten1.workbench.test", + principalId: "prin_room1", + definitionId: "wfd_room1", + status: opts.runStatus, + }, + foldedRunMarker: opts.parked, + workbenchLaunchRow: { + tenantId: "ten_1", + instanceId: "ins_room1", + currentRunId: "run_dead", + foldedBody: DEAD_ROOM_FOLDED_BODY, + noopInference: true, + }, + }); + const sessionService = createFakeSessionService(); + const notices: unknown[] = []; + const platform = createHubChatPlatform({ + toolGrantsForPins: () => [], + db: db as never, + noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", + sessionService, + assetService: createFakeAssetService(), + // Routable, and dead anyway: that combination is exactly what the + // wake path cannot fix — boot restore re-announced the address, so + // nothing looks broken until the next message is dropped. + sidecarRouter: createFakeSidecarRouter({ + routableAddresses: ["run_dead@ten1.workbench.test"], + }), + eventCollectors: createFakeEventCollectors(), + relaunchNotice: { current: (notice) => notices.push(notice) }, + }); + return { db, platform, sessionService, notices }; + } + + test("relaunches a routable-but-dead participant and tells the room", async () => { + const { db, platform, sessionService, notices } = createSweepFixture({ + runStatus: "failed", + parked: false, + }); + + const swept = await platform.sweepTerminalRuns(); + + expect(swept).toEqual({ scanned: 1, relaunched: 1 }); + expect(sessionService.adoptedDeployCalls).toHaveLength(1); + + // The fresh run keeps neither the dead run's id nor its address — + // the platform derives one from the other — while the room's own + // stable id never moves. + const repointed = db.updated.at(-1)?.values as { + currentRunId: string; + priorRunIds: string[]; + }; + expect(repointed.currentRunId).not.toBe("run_dead"); + expect(repointed.priorRunIds).toEqual(["run_dead"]); + + expect(notices).toEqual([ + { + tenantId: "ten_1", + roomAddress: "ins_room1@ten1.workbench.test", + deadRunId: "run_dead", + deadRunStatus: "failed", + newRunId: repointed.currentRunId, + }, + ]); + }); + + test("leaves a folded run merely parked between messages alone", async () => { + const { platform, sessionService, notices } = createSweepFixture({ + runStatus: "completed", + parked: true, + }); + + expect(await platform.sweepTerminalRuns()).toEqual({ + scanned: 0, + relaunched: 0, + }); + expect(sessionService.adoptedDeployCalls).toHaveLength(0); + expect(notices).toEqual([]); + }); + + test("leaves a running participant alone", async () => { + const { platform, sessionService, notices } = createSweepFixture({ + runStatus: "running", + parked: false, + }); + + expect(await platform.sweepTerminalRuns()).toEqual({ + scanned: 0, + relaunched: 0, + }); + expect(sessionService.adoptedDeployCalls).toHaveLength(0); + expect(notices).toEqual([]); + }); +}); diff --git a/packages/chat/test/relaunch-close.test.ts b/packages/chat/test/relaunch-close.test.ts new file mode 100644 index 000000000..fd71dd97a --- /dev/null +++ b/packages/chat/test/relaunch-close.test.ts @@ -0,0 +1,307 @@ +// The two halves of a relaunch a reader actually experiences: the +// notice that tells them the turn they were waiting on was lost, and +// an attachment they sent BEFORE the crash still opening afterwards. +// +// Both hang off the same fact — a relaunch mints a fresh run with a +// fresh principal, and a folded run's mail session hangs off its +// principal — so the live run cannot see the retired run's mail at all +// unless something walks back through the history. +// +// The fake below serves `agent_session` by the one-session-per-run- +// principal invariant `resolveRunSessionId` reads, so a retired run's +// session really is a different session — which is the only reason +// this test can fail. +import { describe, expect, test } from "bun:test"; +import { agentSession } from "@intx/db/schema"; +import { workbenchLaunch } from "../src/schema"; +import { createHubChatPlatform } from "../src/platform-adapter"; +import { + createRelaunchNoticePoster, + relaunchNoticeText, +} from "../src/relaunch-notice"; + +const FOLDED_BODY = { + systemPrompt: "be helpful", + toolPackagePins: [], + grantRequirements: [], + credentialBindings: [], + model: null, +}; + +/** Every string a drizzle `eq`/`and` predicate compares against, in order. */ +function comparedValues(node: unknown): string[] { + const chunks = (node as { queryChunks?: unknown[] }).queryChunks; + if (chunks === undefined) { + const value = (node as { value?: unknown }).value; + return typeof value === "string" ? [value] : []; + } + return chunks.flatMap(comparedValues); +} + +type RunRow = { + id: string; + tenantId: string; + definitionId: string | null; + principalId: string | null; + address: string | null; + status: string; +}; + +type MailRow = { id: string; sessionId: string; raw: Uint8Array }; + +function createFakeDb(opts: { + launch: { + tenantId: string; + instanceId: string; + currentRunId: string; + priorRunIds: string[]; + foldedBody: unknown; + noopInference: boolean; + }; + runs: RunRow[]; + mail: MailRow[]; +}) { + return { + query: { + workflowRun: { + findFirst: async ({ where }: { where: unknown }) => { + const [id] = comparedValues(where); + return opts.runs.find((run) => run.id === id); + }, + }, + sessionMail: { + findFirst: async ({ where }: { where: unknown }) => { + const [id, sessionId] = comparedValues(where); + return opts.mail.find( + (row) => row.id === id && row.sessionId === sessionId, + ); + }, + }, + }, + select: () => ({ + from: (table: unknown) => ({ + where: (predicate: unknown) => { + const chain = { + orderBy: () => chain, + limit: async () => { + const [value] = comparedValues(predicate); + // One session per run principal — the invariant + // `resolveRunSessionId` reads, which is exactly what makes + // a fresh run's session a DIFFERENT session. + if (table === agentSession) { + return opts.runs.some((run) => run.principalId === value) + ? [{ id: `ses_${value ?? ""}` }] + : []; + } + if (table !== workbenchLaunch) return []; + return value === opts.launch.instanceId || + value === opts.launch.currentRunId + ? [opts.launch] + : []; + }, + }; + return chain; + }, + }), + }), + } as never; +} + +function createPlatform(db: never) { + return createHubChatPlatform({ + db, + toolGrantsForPins: () => [], + noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", + sessionService: {} as never, + assetService: {} as never, + sidecarRouter: { getRoutableAddresses: () => [] } as never, + eventCollectors: {} as never, + }); +} + +const RAW_MAIL = new TextEncoder().encode( + [ + "Content-Type: multipart/mixed; boundary=b", + "", + "--b", + "Content-Type: text/plain", + "", + "the receipt you asked for", + "--b--", + "", + ].join("\r\n"), +); + +describe("fetchBlob across a relaunch", () => { + const RELAUNCHED_LAUNCH = { + tenantId: "ten_1", + instanceId: "ins_room1", + currentRunId: "run_fresh", + priorRunIds: ["run_dead"], + foldedBody: FOLDED_BODY, + noopInference: false, + }; + const RUNS: RunRow[] = [ + { + id: "run_fresh", + tenantId: "ten_1", + definitionId: "wfd_1", + principalId: "prin_fresh", + address: "run_fresh@ten1.workbench.test", + status: "running", + }, + { + id: "run_dead", + tenantId: "ten_1", + definitionId: "wfd_1", + principalId: "prin_dead", + address: "run_dead@ten1.workbench.test", + status: "failed", + }, + ]; + + test("reads an attachment sent before the crash, off the retired run's session", async () => { + const db = createFakeDb({ + launch: RELAUNCHED_LAUNCH, + runs: RUNS, + mail: [{ id: "mail_1", sessionId: "ses_prin_dead", raw: RAW_MAIL }], + }); + + const body = await createPlatform(db).fetchBlob( + "ins_room1", + "blob_mail_1_1", + ); + + expect(new TextDecoder().decode(body as Uint8Array)).toContain( + "the receipt you asked for", + ); + }); + + test("a blob on no session this participant ever held is still refused", async () => { + const db = createFakeDb({ + launch: RELAUNCHED_LAUNCH, + runs: RUNS, + mail: [{ id: "mail_1", sessionId: "ses_prin_stranger", raw: RAW_MAIL }], + }); + + await expect( + createPlatform(db).fetchBlob("ins_room1", "blob_mail_1_1"), + ).rejects.toThrow('No mail "mail_1"'); + }); + + test("a retired run whose row is gone is skipped, not fatal", async () => { + const db = createFakeDb({ + launch: { + ...RELAUNCHED_LAUNCH, + priorRunIds: ["run_reaped", "run_dead"], + }, + runs: RUNS, + mail: [{ id: "mail_1", sessionId: "ses_prin_dead", raw: RAW_MAIL }], + }); + + const body = await createPlatform(db).fetchBlob( + "ins_room1", + "blob_mail_1_1", + ); + + expect(new TextDecoder().decode(body as Uint8Array)).toContain( + "the receipt you asked for", + ); + }); +}); + +describe("relaunchNoticeText", () => { + test("names the cause the reader experienced, never the machinery", () => { + const crashed = relaunchNoticeText("failed"); + const cancelled = relaunchNoticeText("cancelled"); + const ended = relaunchNoticeText("completed"); + + expect(crashed).not.toBe(cancelled); + expect(cancelled).not.toBe(ended); + for (const notice of [crashed, cancelled, ended]) { + expect(notice).toContain("I'm back now"); + expect(notice.toLowerCase()).not.toContain("run"); + expect(notice.toLowerCase()).not.toContain("terminal"); + expect(notice.toLowerCase()).not.toContain("relaunch"); + } + }); +}); + +describe("createRelaunchNoticePoster", () => { + function createRoom(participantAddresses: string[][]) { + const posted: { + workbenchId: string; + senderAddress: string; + text: string; + }[] = []; + const store = { + listWorkbenchSettings: async () => + participantAddresses.map((addresses, index) => ({ + workbenchId: `wb_${String(index)}`, + settings: { + "chat/participants": addresses.map((address) => ({ + address, + handle: address.split("@")[0], + name: null, + kind: "agent", + })), + }, + })), + }; + const roomMessages = { + insertMessage: async (input: { + id: string; + workbenchId: string; + sender: { address: string }; + parts: { kind: string; text: string }[]; + }) => { + posted.push({ + workbenchId: input.workbenchId, + senderAddress: input.sender.address, + text: input.parts[0]?.text ?? "", + }); + return { + ...input, + createdAt: "2026-08-20T00:00:00.000Z", + threadId: null, + }; + }, + }; + return { posted, store, roomMessages }; + } + + test("posts into every room the replaced participant belongs to, in its own voice", async () => { + const room = createRoom([ + ["ins_room1@ten1.workbench.test"], + ["ins_other@ten1.workbench.test"], + ["ins_room1@ten1.workbench.test", "ins_other@ten1.workbench.test"], + ]); + const poster = createRelaunchNoticePoster({ + store: room.store as never, + roomMessages: room.roomMessages as never, + publish: () => undefined, + }); + + poster({ + tenantId: "ten_1", + roomAddress: "ins_room1@ten1.workbench.test", + deadRunId: "run_dead", + deadRunStatus: "failed", + newRunId: "run_fresh", + }); + await Bun.sleep(5); + + expect(room.posted).toEqual([ + { + workbenchId: "wb_0", + senderAddress: "ins_room1@ten1.workbench.test", + text: relaunchNoticeText("failed"), + }, + { + workbenchId: "wb_2", + senderAddress: "ins_room1@ten1.workbench.test", + text: relaunchNoticeText("failed"), + }, + ]); + }); +}); diff --git a/scripts/e2e/cl-6324-launch-proof.ts b/scripts/e2e/cl-6324-launch-proof.ts index 3cb6291da..355b83c86 100644 --- a/scripts/e2e/cl-6324-launch-proof.ts +++ b/scripts/e2e/cl-6324-launch-proof.ts @@ -1251,6 +1251,26 @@ async function main(): Promise { sendAndAwaitReply("Are you still there? One sentence.", "proof 4", 2), ); + // The audit trail is the whole reason a relaunch mints a fresh run + // instead of reclaiming the dead one's address: the run that died + // mid-turn keeps its own durable log, under its own address, readable + // through the ordinary run routes — after the run that replaced it is + // already answering (the hop above). + await hop("PROOF 4 — the replaced run's log is still readable", async () => { + const events = await readRunEvents(); + if (events.length === 0) { + throw new Error( + `the replaced run ${agentRunId}'s durable event log is unreadable ` + + `after its replacement went live: the relaunch reclaimed the ` + + `audit trail it exists to preserve`, + ); + } + console.log( + ` TRANSCRIPT — replaced run ${agentRunId} still readable: ` + + `${String(events.length)} events, last ${events.at(-1)?.type ?? "?"}`, + ); + }); + console.log("\n=== TIMINGS ==="); for (const t of timings) { console.log(` ${t.label}: ${(t.ms / 1000).toFixed(1)}s`);