Seed: redeploy a workflow orphaned by a stack restart, don't skip it

docs/seed-reconciliation.md

@@ -99,6 +99,31 @@ publish from stranding a usable-looking-but-empty asset:
   this history is the same operation as seeding one for the first
   time: re-run `workbench seed`.
 
+## Workflow deployments (`workbench seed`)
+
+`ensureDeployment` (`packages/hub-client/src/seed.ts`) treats a
+workflow's `workflow_run` deployment row as seed-owned state, but the
+row's `status` column is not the whole story: the hub only routes mail
+to a deployment through an in-memory table (`sidecarRouter`'s
+`addressIndex`, `vendor/intx/hub-sessions/src/ws/sidecar-handler.ts`)
+that binds an agent address to whichever sidecar socket most recently
+proved ownership of it. That table lives in the hub process, not the
+database — a hub or sidecar restart empties it, while the persisted row
+still reads `deployed`.
+
+Before skipping a `deployed`/`pending` row as "already deployed",
+`ensureDeployment` checks `GET
+/api/tenants/:tenantId/workflows/runs/:runId/health` (a live read of
+`sidecarRouter.getRoutableAddresses()`, not the stored status) and
+skips only when `liveness` answers `"ok"`. A row whose sidecar is gone
+is stale, not deployed: seed logs it as stale and pushes a fresh
+deployment, which mints a new `workflow_run` (new anchor run id, new
+agent address) on whichever sidecar is currently connected. The stale
+row is left in place rather than rebound — a sidecar carries no durable
+state of its own, so handing an old run's identity to a new sidecar
+process would silently pretend session state survived that never did.
+A genuine redeploy is the only honest repair.
+
 ## Env provider credentials (hub boot)
 
 `apps/hub/src/env-credential-plant.ts` delegates to

packages/hub-client/src/seed.ts

@@ -19,6 +19,7 @@ import {
   ModelProviderResponse,
   ModelResponse,
   ProviderResponse,
+  WorkflowRunHealth,
   paginatedSchema,
   Capability,
 } from "@intx/types";
@@ -634,6 +635,43 @@ async function listRunIds(
   return parseAs(WorkflowRunListResponse, runs.data, "runs response").runIds;
 }
 
+/**
+ * Whether a "deployed" deployment's run is actually routable right now.
+ * `GET .../runs/:runId/health` reads `sidecarRouter.getRoutableAddresses()`
+ * — the hub's in-memory table binding an agent address to the specific
+ * connected sidecar socket that owns it — so this is a live check, not a
+ * read of the persisted `workflow_run.status` column the caller already
+ * has. That column survives a hub/sidecar restart; the routing table
+ * does not, so a "deployed" row can answer `false` here forever until
+ * something redeploys it. 404 (run never existed) and 410 (run stopped)
+ * both count as not routable: either way, nothing this deployment id
+ * names can be reused.
+ */
+async function isDeploymentRoutable(
+  api: ApiCall,
+  cookies: string[],
+  tenantId: string,
+  deploymentId: string,
+): Promise<boolean> {
+  const health = await api(
+    "GET",
+    `/api/tenants/${tenantId}/workflows/runs/${deploymentId}/health`,
+    undefined,
+    cookies,
+  );
+  if (health.status === 404 || health.status === 410) return false;
+  if (health.status !== 200) {
+    throw new CliError(
+      `the hub answered deployment ${deploymentId}'s health check with status ${health.status}: ${JSON.stringify(health.data)}`,
+      "check the hub logs for the underlying failure, then re-run: workbench seed",
+    );
+  }
+  return (
+    parseAs(WorkflowRunHealth, health.data, "run health response").liveness ===
+    "ok"
+  );
+}
+
 async function ensureDeployment(
   api: ApiCall,
   cookies: string[],
@@ -662,10 +700,23 @@ async function ensureDeployment(
       d.definitionAssetId === args.assetId && isLiveDeploymentStatus(d.status),
   );
   if (active) {
+    if (await isDeploymentRoutable(api, cookies, args.tenantId, active.id)) {
+      log(
+        `workflow ${args.assetName} already deployed as ${active.id} (skipped)`,
+      );
+      return active.id;
+    }
+    // The DB row survives a stack restart; the in-memory sidecar
+    // routing table that binds an address to a live process does not.
+    // Restart the hub and sidecar and every previously "deployed"
+    // workflow_run still reads "deployed" while nothing routes its
+    // address. Skipping here would just move the same 409
+    // `confirmDeploymentAnswers` hits below one step earlier — redeploy
+    // fresh instead of trusting a status column that outlived the
+    // process it described.
     log(
-      `workflow ${args.assetName} already deployed as ${active.id} (skipped)`,
+      `workflow ${args.assetName}'s deployment ${active.id} is stale (its sidecar is gone); redeploying`,
     );
-    return active.id;
   }
 
   const deployed = await api(

packages/hub-client/test/seed.test.ts

@@ -546,6 +546,14 @@ describe("seedTenant", () => {
           status: 200,
           data: [deploymentRow("dep_1", "ast_1", "deployed")],
         };
+      if (
+        method === "GET" &&
+        path === `/api/tenants/${TENANT_ID}/workflows/runs/dep_1/health`
+      )
+        return {
+          status: 200,
+          data: { liveness: "ok", readiness: "ok", lastCheckedAt: null },
+        };
       if (
         method === "GET" &&
         path === `/api/tenants/${TENANT_ID}/workflows/dep_1/runs`
@@ -592,6 +600,107 @@ describe("seedTenant", () => {
     expect(output).toContain("confirmed workflow echo: run run_2 started");
   });
 
+  test("a deployment orphaned by a stack restart is redeployed, not skipped", async () => {
+    const { lines, log } = collector();
+    const push: WorkflowPusher = async () => ({
+      outcome: "unchanged" as const,
+      commitSha: "b".repeat(40),
+    });
+    let runsCalls = 0;
+    const handler: FakeHandler = (method, path) => {
+      const base = baseRoutes(method, path);
+      if (base) return base;
+      if (method === "POST" && path === `/api/tenants/${TENANT_ID}/assets`)
+        return { status: 409, data: { error: "name taken" } };
+      if (
+        method === "GET" &&
+        path ===
+          `/api/tenants/${TENANT_ID}/assets?kind=workflow&inherited=false`
+      )
+        return {
+          status: 200,
+          data: [
+            {
+              ...assetRow("ast_1", "echo"),
+              origin: { tenantId: TENANT_ID, direct: true },
+            },
+          ],
+        };
+      if (
+        method === "GET" &&
+        path === `/api/tenants/${TENANT_ID}/workflows/deployments`
+      )
+        return {
+          status: 200,
+          // dep_1 is a survivor from before the stack restarted: its
+          // workflow_run row still reads "deployed", but no sidecar
+          // owns its address anymore.
+          data: [deploymentRow("dep_1", "ast_1", "deployed")],
+        };
+      if (
+        method === "GET" &&
+        path === `/api/tenants/${TENANT_ID}/workflows/runs/dep_1/health`
+      )
+        return {
+          status: 200,
+          data: {
+            liveness: "unhealthy",
+            readiness: "not_ready",
+            lastCheckedAt: null,
+          },
+        };
+      if (
+        method === "POST" &&
+        path === `/api/tenants/${TENANT_ID}/workflows/deployments`
+      )
+        return {
+          status: 201,
+          data: deploymentRow("dep_2", "ast_1", "deployed"),
+        };
+      if (
+        method === "GET" &&
+        path === `/api/tenants/${TENANT_ID}/workflows/dep_2/runs`
+      ) {
+        runsCalls += 1;
+        return {
+          status: 200,
+          data: { runIds: runsCalls === 1 ? [] : ["run_1"] },
+        };
+      }
+      if (
+        method === "POST" &&
+        path === `/api/tenants/${TENANT_ID}/workflows/dep_2/mail`
+      )
+        return {
+          status: 202,
+          data: {
+            runId: "dep_2",
+            address: `ins_dep_2@${TENANT_DOMAIN}`,
+            messageId: "<m3@workbench.localhost>",
+          },
+        };
+      return undefined;
+    };
+
+    const echoOnly = DEFAULT_WORKFLOWS.filter((w) => w.assetName === "echo");
+    await seedTenant(
+      args({
+        api: fakeAPI(handler),
+        pushWorkflow: push,
+        log,
+        workflows: echoOnly,
+      }),
+    );
+
+    const output = lines.join("\n");
+    expect(output).toContain(
+      "workflow echo's deployment dep_1 is stale (its sidecar is gone); redeploying",
+    );
+    expect(output).toContain("deployed workflow echo as dep_2");
+    expect(output).toContain("confirmed workflow echo: run run_1 started");
+    expect(output).not.toContain("not routable");
+  });
+
   test("an unreachable deployment address names the sidecar as the fix", async () => {
     const handler: FakeHandler = (method, path) => {
       const base = baseRoutes(method, path);

packages/onboarding/test/complete-credential.test.ts

@@ -1024,6 +1024,17 @@ describe("completeCredentialSetup", () => {
           cookies: [],
         };
       }
+      if (
+        method === "GET" &&
+        path.startsWith(`/api/tenants/${TENANT_ID}/workflows/runs/`) &&
+        path.endsWith("/health")
+      ) {
+        return {
+          status: 200,
+          data: { liveness: "ok", readiness: "ok", lastCheckedAt: null },
+          cookies: [],
+        };
+      }
       if (method === "POST" && path === `/api/tenants/${TENANT_ID}/providers`) {
         const name = (body as { name: string }).name;
         const existing = providers.find((p) => p.name === name);
@@ -1776,6 +1787,17 @@ describe("ensureSeeded (the slow half)", () => {
           cookies: [],
         };
       }
+      if (
+        method === "GET" &&
+        path.startsWith(`/api/tenants/${TENANT_ID}/workflows/runs/`) &&
+        path.endsWith("/health")
+      ) {
+        return {
+          status: 200,
+          data: { liveness: "ok", readiness: "ok", lastCheckedAt: null },
+          cookies: [],
+        };
+      }
       if (
         method === "POST" &&
         path === `/api/tenants/${TENANT_ID}/workflows/deployments`