CL-6365: the terminal signal was never delivered — hold boot-restore pushes until the link is routable

apps/hub/src/index.ts

@@ -42,6 +42,7 @@ import {
 } from "@intx/hub-api";
 
 import {
+  agentDefinitionSourceTree,
   buildAgentDefinitionWorkflow,
   createAgentDefinitionRoutes,
   createDefinitionAssetHistory,
@@ -75,6 +76,7 @@ import {
   createDrizzleWriteClaimStore,
   createHubChatPlatform,
   createNoopInferenceRoutes,
+  createRelaunchNoticePoster,
   findExistingAgentChat,
   createWorkflowParticipantRoutes,
   isWorkbenchHostDefinitionName,
@@ -84,6 +86,7 @@ import {
   startWorkflowCommand,
   sendWorkbenchMessage,
 } from "@corbits/chat";
+import type { RelaunchNoticePort } from "@corbits/chat";
 import type { FinalizedTurnToolCall } from "@corbits/turn-artifacts";
 import {
   createCryptoProviderCache,
@@ -1041,14 +1044,19 @@ 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,
     assetService,
     sidecarRouter,
     eventCollectors,
     credentialCipher,
-    hubPublicKey,
     toolGrantsForPins,
     mcpCredentialBindingsFor,
     noopInferenceBaseUrl: `${config.baseUrl}/api/chat/noop-inference`,
@@ -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
@@ -1650,7 +1711,6 @@ export async function createHub(config: HubConfig) {
             assetService,
             sidecarRouter,
             eventCollectors,
-            hubPublicKey,
             toolGrantsForPins,
             mcpCredentialBindingsFor,
             cryptoProviderCache: foldedRunCryptoProviders,
@@ -2198,7 +2258,6 @@ export async function createHub(config: HubConfig) {
     sidecarRouter,
     eventCollectors,
     credentialCipher,
-    hubPublicKey,
     toolGrantsForPins,
     mcpCredentialBindingsFor,
     cryptoProviderCache: foldedRunCryptoProviders,
@@ -2571,13 +2630,6 @@ export async function createHub(config: HubConfig) {
     listModels: listMyraModels,
   };
 
-  // Mirrors `@corbits/agent-directory`'s own private
-  // `AGENT_DEFINITION_ASSET_PATH` constant (not exported — the route
-  // module keeps it internal), kept in lockstep by convention since
-  // this is the same asset-tree contract `ensureWorkflowDefinitionForAsset`
-  // reads back from.
-  const PLANNER_AGENT_DEFINITION_ASSET_PATH = "workflow.json";
-
   /**
    * Wraps the same sequence `@corbits/agent-directory`'s `POST /`
    * handler runs (`buildAgentDefinitionWorkflow` → `reindexPinnedSkills`
@@ -2659,9 +2711,7 @@ export async function createHub(config: HubConfig) {
       ref: DEFAULT_ASSET_REF,
       principal: { kind: "hub" },
       tree: {
-        files: {
-          [PLANNER_AGENT_DEFINITION_ASSET_PATH]: workflowJson,
-        },
+        files: agentDefinitionSourceTree({ handle, workflowJson }),
         message: `Define agent ${input.name}`,
       },
     });

apps/hub/src/routine-launcher.test.ts

@@ -5,7 +5,7 @@
 // goes through — and that a delivery failure past launch never un-does
 // or hides the already-real run. `@corbits/folded-runs` is real here
 // except for `launchFoldedRun`/`sendFoldedMailWithRetry`/
-// `readDefinitionJSON`, which would otherwise need a real tenant catalog
+// `readDefinitionProjection`, which would otherwise need a real tenant catalog
 // and asset store — the same "swap the one export that needs a join"
 // approach `packages/folded-runs/test/launch.test.ts` and
 // `packages/webhook-triggers/test/launch.test.ts` use.
@@ -33,7 +33,7 @@ let sendFoldedMailWithRetryResult: unknown = {
 
 mock.module("@corbits/folded-runs", () => ({
   ...actualFoldedRuns,
-  readDefinitionJSON: async () => ({ __fake: true }),
+  readDefinitionProjection: async () => ({ __fake: true }),
   readFoldedBody: () => FOLDED_BODY,
   launchFoldedRun: async (...args: unknown[]) => {
     launchFoldedRunCalls.push(args);
@@ -114,7 +114,6 @@ function buildLauncher(overrides: { definition?: unknown } = {}) {
     sessionService: {} as never,
     assetService: {} as never,
     sidecarRouter: {} as never,
-    hubPublicKey: "hub-key",
     toolGrantsForPins: () => [],
     eventCollectors: {} as never,
     cryptoProviderCache: { get: async () => ({}) as never },
@@ -232,7 +231,6 @@ describe("createHubRoutineLauncher — delivery workbench", () => {
       sessionService: {} as never,
       assetService: {} as never,
       sidecarRouter: {} as never,
-      hubPublicKey: "hub-key",
       toolGrantsForPins: () => [],
       eventCollectors: {} as never,
       cryptoProviderCache: { get: async () => ({}) as never },
@@ -259,7 +257,6 @@ describe("createHubRoutineLauncher — recurring-task bridge", () => {
       sessionService: {} as never,
       assetService: {} as never,
       sidecarRouter: {} as never,
-      hubPublicKey: "hub-key",
       toolGrantsForPins: () => [],
       eventCollectors: {} as never,
       cryptoProviderCache: { get: async () => ({}) as never },
@@ -297,7 +294,6 @@ describe("createHubRoutineLauncher — recurring-task bridge", () => {
       sessionService: {} as never,
       assetService: {} as never,
       sidecarRouter: {} as never,
-      hubPublicKey: "hub-key",
       toolGrantsForPins: () => [],
       eventCollectors: {} as never,
       cryptoProviderCache: { get: async () => ({}) as never },
@@ -324,7 +320,6 @@ describe("createHubRoutineLauncher — recurring-task bridge", () => {
       sessionService: {} as never,
       assetService: {} as never,
       sidecarRouter: {} as never,
-      hubPublicKey: "hub-key",
       toolGrantsForPins: () => [],
       eventCollectors: {} as never,
       cryptoProviderCache: { get: async () => ({}) as never },
@@ -356,7 +351,6 @@ describe("createHubRoutineLauncher — recurring-task bridge", () => {
       sessionService: {} as never,
       assetService: {} as never,
       sidecarRouter: {} as never,
-      hubPublicKey: "hub-key",
       toolGrantsForPins: () => [],
       eventCollectors: {} as never,
       cryptoProviderCache: { get: async () => ({}) as never },
@@ -378,7 +372,6 @@ describe("createHubRoutineLauncher — recurring-task bridge", () => {
       sessionService: {} as never,
       assetService: {} as never,
       sidecarRouter: {} as never,
-      hubPublicKey: "hub-key",
       toolGrantsForPins: () => [],
       eventCollectors: {} as never,
       cryptoProviderCache: { get: async () => ({}) as never },

apps/hub/src/routine-launcher.ts

@@ -48,7 +48,7 @@ import { tenant as tenantTable, workflowDefinition } from "@intx/db/schema";
 import {
   domainOf,
   launchFoldedRun,
-  readDefinitionJSON,
+  readDefinitionProjection,
   readFoldedBody,
   sendFoldedMailWithRetry,
   type CryptoProviderCache,
@@ -162,11 +162,11 @@ export function createHubRoutineLauncher(
         throw new Error(`no tenant "${input.tenantId}"`);
       }
 
-      const definitionJSON = await readDefinitionJSON(
-        deps.assetService,
-        definitionRow.assetId,
+      const projection = await readDefinitionProjection(deps.db, definitionRow);
+      const foldedBody = readFoldedBody(
+        projection,
+        definitionRow.grantRequirements,
       );
-      const foldedBody = readFoldedBody(definitionJSON);
 
       const instanceId = generateId("workflowRun");
       const triggerAddress = formatRunAddress(instanceId, tenantRow.domain);

apps/sidecar/bin/workflow-probe-child

@@ -0,0 +1,21 @@
+#!/usr/bin/env bun
+// One-shot workflow-probe child. The sidecar host spawns this by path
+// (`Bun.spawn([binaryPath])`).
+//
+// The child reads the materialized package dir and IPC anchors from its
+// fresh env, evaluates the workflow entry behind the airlock, and ships
+// one HMAC-signed result frame on stdout. An evaluation failure is shipped
+// as an `ok: false` frame (handled inside the runner), so a throw reaching
+// here is a pre-evaluation defect (bad env, unwritable stdout) that exits
+// non-zero -- the host then reaps and answers `workflow.probe.error`.
+import { runWorkflowProbeChildFromProcessEnv } from "../src/workflow-probe-handler";
+
+try {
+  await runWorkflowProbeChildFromProcessEnv();
+  process.exit(0);
+} catch (err) {
+  process.stderr.write(
+    `workflow probe child failed: ${err instanceof Error ? err.message : String(err)}\n`,
+  );
+  process.exit(1);
+}

apps/sidecar/src/boot-restore-push-hold.test.ts

@@ -0,0 +1,50 @@
+import { describe, expect, test } from "bun:test";
+
+import { createBootRestorePushHold } from "./boot-restore-push-hold";
+
+function recordingStore() {
+  const held: string[] = [];
+  return {
+    held,
+    markAddressUnroutable(agentAddress: string) {
+      held.push(agentAddress);
+    },
+  };
+}
+
+describe("createBootRestorePushHold", () => {
+  test("holds every address a boot restore registers", () => {
+    const store = recordingStore();
+    const hold = createBootRestorePushHold(store);
+
+    hold.begin();
+    hold.onDeploymentRegistered("run_a@bench.localhost");
+    hold.onDeploymentRegistered("run_b@bench.localhost");
+    hold.end();
+
+    expect(store.held).toEqual([
+      "run_a@bench.localhost",
+      "run_b@bench.localhost",
+    ]);
+  });
+
+  test("does not hold a deployment that arrives over a live link", () => {
+    const store = recordingStore();
+    const hold = createBootRestorePushHold(store);
+
+    hold.begin();
+    hold.end();
+    hold.onDeploymentRegistered("run_c@bench.localhost");
+
+    expect(store.held).toEqual([]);
+  });
+
+  test("holds nothing before the restore is armed", () => {
+    const store = recordingStore();
+    const hold = createBootRestorePushHold(store);
+
+    hold.onDeploymentRegistered("run_d@bench.localhost");
+
+    expect(store.held).toEqual([]);
+  });
+});

apps/sidecar/src/boot-restore-push-hold.ts

@@ -0,0 +1,43 @@
+// Boot restore spawns each restored deployment's supervisor BEFORE
+// `hubLink.connect()`, and a supervisor that finds a run whose step died
+// mid-invocation commits that run's `StepFailed`/`RunFailed` immediately.
+// Those commits schedule a workflow-run pack push into a link that does not
+// exist yet, so the push fails "Connection lost" and its only recovery is the
+// post-challenge re-drive, which re-ships a slot whose error has already
+// latched. When the rejection lands AFTER the challenge fired, nothing
+// re-arms the slot: the terminal event stays on sidecar disk, the hub's copy
+// of the durable log stops at the last pre-crash event, and
+// `workflow_run.status` stays "running" for a run that will never accept mail
+// again.
+//
+// Holding every address registered during boot restore until the reconnect
+// challenge proves it routable removes the race rather than recovering from
+// it — the same block the link already applies across a mid-life disconnect.
+import type { WorkflowRunPackPushingRepoStore } from "./workflow-run-pack-client";
+
+export interface BootRestorePushHold {
+  /** Arm the hold for the duration of `restoreWorkflowDeployments`. */
+  begin(): void;
+  /** Disarm it; deployments registered afterwards arrive over a live link. */
+  end(): void;
+  /** Called for every deployment registration the deploy router makes. */
+  onDeploymentRegistered(agentAddress: string): void;
+}
+
+export function createBootRestorePushHold(
+  store: Pick<WorkflowRunPackPushingRepoStore, "markAddressUnroutable">,
+): BootRestorePushHold {
+  let restoring = false;
+  return {
+    begin() {
+      restoring = true;
+    },
+    end() {
+      restoring = false;
+    },
+    onDeploymentRegistered(agentAddress) {
+      if (!restoring) return;
+      store.markAddressUnroutable(agentAddress);
+    },
+  };
+}