CL-6365: close proof 4's remainders — sweep, notice, old attachments, audit

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<typeof setTimeout> | 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

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.

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,
 };

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<LiveAgent["run"][]> {
+  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<void> {
+  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<LiveAgent[]> {
+  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;
 }

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,

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;
+    `,
+  },
 ];
 
 /**