Chat: wake resolves the freshly invited run's binding; seed writers share one asset ensure (CL-7486)

packages/chat/src/platform-adapter.ts

@@ -989,10 +989,24 @@ export function createHubChatPlatform(
    * more time regardless. Exhausting every delay means the condition
    * is not transient and the caller gets a clean `AgentUnreachableError`
    * rather than an unhandled 500.
+   *
+   * CL-7486: a wake this loop runs between retries can itself relaunch
+   * the run — `wakeByAddress`'s dead-run branch repoints the room's
+   * stable participant at a fresh address without telling this loop.
+   * Retrying `sendRunMail` against the address it started with would
+   * then keep hammering a run that will never become routable again,
+   * and the wake ahead of the NEXT retry would fail outright (no
+   * `workbench_launch` row still names that address at all) instead of
+   * the "unreachable" this loop knows how to ride out. `reresolve`
+   * re-reads the live address (and its session) off the stable
+   * participant id before every wake/retry, so this loop always chases
+   * whichever run is actually current.
    */
   async function sendRunMailWithReclaimRetry(
     params: SendRunMailParams,
+    reresolve: () => Promise<{ agentAddress: string; sessionId: string }>,
   ): Promise<Awaited<ReturnType<typeof sendRunMail>>> {
+    let current = params;
     let loggedRetryStart = false;
     for (let attempt = 0; ; attempt++) {
       try {
@@ -1001,29 +1015,44 @@ export function createHubChatPlatform(
         // to accept regardless — the signal parameter is unused here,
         // same abandon-on-timeout behavior as before.
         return await withTimeout(
-          () => sendRunMail(runMailDeps, params),
+          () => sendRunMail(runMailDeps, current),
           MAIL_DELIVERY_TIMEOUT_MS,
-          `mail to ${params.agentAddress} did not settle within ${String(MAIL_DELIVERY_TIMEOUT_MS)}ms`,
+          `mail to ${current.agentAddress} did not settle within ${String(MAIL_DELIVERY_TIMEOUT_MS)}ms`,
         );
       } catch (err) {
         const delay = RECLAIM_RETRY_DELAYS_MS[attempt];
         if (!isAgentUnreachable(err) || delay === undefined) {
           if (loggedRetryStart) {
-            wakeLogger.info`mail to ${params.agentAddress} exhausted every reclaim retry; giving up`;
+            wakeLogger.info`mail to ${current.agentAddress} exhausted every reclaim retry; giving up`;
           }
           if (isAgentUnreachable(err)) {
-            throw new AgentUnreachableError(params.agentAddress, {
+            throw new AgentUnreachableError(current.agentAddress, {
               cause: err,
             });
           }
           throw err;
         }
         if (!loggedRetryStart) {
-          wakeLogger.info`mail to ${params.agentAddress} hit "agent is unreachable"; retrying with backoff while the post-restart reclaim settles`;
+          wakeLogger.info`mail to ${current.agentAddress} hit "agent is unreachable"; retrying with backoff while the post-restart reclaim settles`;
           loggedRetryStart = true;
         }
         await sleep(delay);
-        await wakeByAddressBounded(params.agentAddress);
+        // Wake whatever this loop still believes is live — `wakeByAddress`
+        // itself tolerates an address one relaunch behind the room's
+        // current one (its lookup matches either the stable participant
+        // id or the live run id) — and only THEN re-resolve, since the
+        // wake just run is exactly the kind of relaunch that can leave
+        // `current.agentAddress` pointing at a run that no longer exists
+        // at all. Re-resolving before the wake instead would still hand
+        // the next retry whatever address was current a moment ago,
+        // stale again the instant this wake relaunches it.
+        await wakeByAddressBounded(current.agentAddress);
+        const fresh = await reresolve();
+        current = {
+          ...current,
+          agentAddress: fresh.agentAddress,
+          sessionId: fresh.sessionId,
+        };
       }
     }
   }
@@ -1399,7 +1428,30 @@ export function createHubChatPlatform(
           ? { references: input.content.references }
           : {}),
       };
-      const sent = await sendRunMailWithReclaimRetry(withThreading);
+      const sent = await sendRunMailWithReclaimRetry(
+        withThreading,
+        async () => {
+          const relive = await requireLive(input.workbenchId, input.tenantId);
+          try {
+            await ensureRunSession({
+              db: deps.db,
+              eventCollectors: deps.eventCollectors,
+              runId: relive.run.id,
+            });
+          } catch (err) {
+            reportError(err, {
+              operation: "chat.sendMail.ensureRunSession",
+              tenantId: input.tenantId,
+              agentId: relive.binding.liveAddress,
+              extra: { runId: relive.run.id },
+            });
+          }
+          return {
+            agentAddress: relive.binding.liveAddress,
+            sessionId: await resolveRunSessionIdOrThrow(deps.db, relive.run),
+          };
+        },
+      );
 
       lifecycle?.recordActivity(deliveryAddress);
 

packages/chat/test/platform-adapter.test.ts

@@ -2781,6 +2781,139 @@ describe("createHubChatPlatform", () => {
     });
   });
 
+  // CL-7486: `sendRunMailWithReclaimRetry` used to retry every attempt
+  // against the address it started with, even after a wake in between
+  // attempts relaunched the run. A run already one relaunch generation
+  // in (`instanceId` stable, `currentRunId` already repointed once) that
+  // dies again while this loop is retrying reproduces the production
+  // failure exactly: a second relaunch repoints `currentRunId` again,
+  // and the address this loop still holds is now neither the stable
+  // `instanceId` nor the current `currentRunId` — invisible to
+  // `readBindingByAddressAnyTenant`, so the next wake for it throws "No
+  // workbench_launch binding" instead of finding the run that replaced
+  // it.
+  describe("sendRunMailWithReclaimRetry re-resolves the live address", () => {
+    test("a relaunch between retries is chased to its new address, not retried against the one that died", async () => {
+      resolveDefinitionSourcesResult = {
+        ok: true,
+        materials: [],
+        sources: [
+          {
+            id: "off_1",
+            provider: "anthropic",
+            baseURL: "https://inference.invalid",
+            credentialId: "cred_placeholder",
+            model: "claude-sonnet-5",
+          },
+        ],
+        defaultSource: "off_1",
+      };
+
+      const staleAddress = "run_gen1@ten1.workbench.test";
+      const db = createFakeDb({
+        assetRow: {
+          tenantId: "ten_1",
+          creatorPrincipalId: "prin_creator",
+          name: "ins_room1",
+          displayName: null,
+        },
+        definitionId: "wfd_room1",
+        workflowRunRow: {
+          id: "run_gen1",
+          address: staleAddress,
+          principalId: "prin_room1",
+          definitionId: "wfd_room1",
+          // Already dead by the time the reclaim retry's own wake reads
+          // it — the run died again while this loop was mid-retry, not
+          // at the top of `sendMail` (which would have relaunched it
+          // before ever reaching this retry loop at all).
+          status: "failed",
+        },
+        workflowDefinitionRow: {
+          id: "wfd_room1",
+          tenantId: "ten_1",
+          status: "deployed",
+          origin: "authored",
+          assetId: "asst_room1",
+        },
+        workbenchLaunchRow: {
+          tenantId: "ten_1",
+          // `instanceId` is the room's own stable id, already distinct
+          // from `currentRunId` — this participant has been relaunched
+          // once before this test even starts.
+          instanceId: "ins_room1",
+          currentRunId: "run_gen1",
+          foldedBody: {
+            systemPrompt: "host prompt",
+            model: "claude-sonnet-5",
+            toolPackagePins: [],
+            grantRequirements: [],
+            credentialBindings: [],
+          },
+        },
+      });
+      db.inserted.push({
+        table: agentSession,
+        values: { id: "ses_run1", principalId: "prin_room1" },
+      });
+
+      const sessionService = createFakeSessionService();
+      sessionService.sendUserMessage = async (params: unknown) => {
+        sessionService.sendUserMessageCalls.push(params);
+        const { agentAddress } = params as { agentAddress: string };
+        // The dead run never becomes reachable again — only a send to
+        // whatever address the relaunch actually produced can succeed.
+        if (agentAddress === staleAddress) {
+          throw new Error("agent is unreachable");
+        }
+        return new TextEncoder().encode("raw-mime-bytes");
+      };
+
+      // Routable so `sendMail`'s own opening wake-before-send gate skips
+      // waking it — the relaunch this test cares about must happen
+      // inside the reclaim-retry loop below, not before it.
+      const sidecarRouter = createFakeSidecarRouter({
+        routableAddresses: [staleAddress],
+      });
+
+      const platform = createPlatform({
+        toolGrantsForPins: () => [],
+        db: db as never,
+        sessionService,
+        sidecarRouter,
+        eventCollectors: createFakeEventCollectors(),
+        reclaimRetryDelaysMs: [1],
+        mailDeliveryTimeoutMs: 5_000,
+      });
+
+      const sent = await platform.sendMail({
+        tenantId: "ten_1",
+        workbenchId: "ins_room1",
+        principalId: "prin_sender",
+        content: { content: "hello" },
+      });
+
+      expect(sent.id).toBeTruthy();
+      expect(sessionService.sendUserMessageCalls).toHaveLength(2);
+      const [first, second] = sessionService.sendUserMessageCalls as {
+        agentAddress: string;
+      }[];
+      expect(first?.agentAddress).toBe(staleAddress);
+      // The retry that follows the relaunch targets the run that
+      // replaced `run_gen1`, never the dead address the loop started
+      // with.
+      expect(second?.agentAddress).not.toBe(staleAddress);
+
+      const repointed = db.updated.at(-1)?.values as {
+        currentRunId: string;
+        priorRunIds: string[];
+      };
+      const secondAddress = second?.agentAddress ?? "";
+      expect(repointed.currentRunId).toBe(secondAddress.split("@")[0] ?? "");
+      expect(repointed.priorRunIds).toEqual(["run_gen1"]);
+    });
+  });
+
   // Proves the actual lever an edited system prompt reaches a running
   // instance through: `wakeByAddress` (exercised via `sendMail`'s
   // wake-on-send path above) replays `workbench_launch.foldedBody`

packages/tool-registry-publish/src/publish.ts

@@ -110,7 +110,19 @@ async function ensureRegistryAsset(
   if (created.status === 201) {
     return parseSchema(AssetResponse, created.data, "create asset response").id;
   }
-  if (created.status !== 409) {
+  // CL-7486: two overlapping seed runs for the same tenant (the root
+  // tenant's own boot-time seed racing a second caller seeding that
+  // same tenant) can both pass the `listRegistryAsset` check above
+  // before either has created the row, then both post-create — the
+  // loser hits the `asset_tenant_kind_name` unique constraint. The
+  // substrate answers that as `409` in the common case; asserted here
+  // by name too, in case a raw constraint violation surfaces as a
+  // generic 500 instead, so the loser still resolves the winner's row
+  // rather than throwing on a race this ensure is meant to tolerate.
+  const isNameCollision =
+    created.status === 409 ||
+    JSON.stringify(created.data).includes("asset_tenant_kind_name");
+  if (!isNameCollision) {
     throw new Error(
       `publishCorbitsToolsRegistry: failed to create the ${CORBITS_TOOLS_REGISTRY} asset: ${String(created.status)}`,
     );

packages/webhook-triggers/test/launch.test.ts

@@ -88,35 +88,28 @@ function createFakeDb() {
     query: {
       workflowDefinition: { findFirst: async () => DEFINITION_ROW },
       tenant: { findFirst: async () => TENANT_ROW },
+      // `recordAgentSessionForRun` reads the just-provisioned run back by
+      // id — production never inserts this row itself (Interchange's
+      // `prepareProvisionedDeployment` mints the anchor row
+      // transactionally before returning it), so this models the same
+      // row for `INTERCHANGE_RUN_ID`.
       workflowRun: {
         findFirst: async () => ({
           id: INTERCHANGE_RUN_ID,
           tenantId: TENANT_ROW.id,
           definitionId: DEFINITION_ROW.id,
-          address: INTERCHANGE_ADDRESS,
-          principalId: null,
+          principalId: "prn_interchange",
         }),
       },
     },
-    insert: () => ({
-      values: () => ({
-        onConflictDoNothing: async () => undefined,
+    insert: (_table: unknown) => ({
+      values: (_values: unknown) => ({
+        onConflictDoNothing: async (_opts: unknown) => {},
       }),
     }),
   };
 }
 
-let eventCollectorCreateCalls: unknown[] = [];
-
-function createFakeEventCollectors() {
-  return {
-    create: (...args: unknown[]) => {
-      eventCollectorCreateCalls.push(args);
-    },
-    has: () => false,
-  };
-}
-
 const TRIGGER = {
   id: "wht_1",
   tenantId: "ten_1",
@@ -204,7 +197,10 @@ function baseDeps() {
       },
     },
     isRoutable: () => isRoutableForTest,
-    eventCollectors: createFakeEventCollectors(),
+    eventCollectors: {
+      create: () => undefined,
+      has: () => false,
+    },
   };
 }
 
@@ -216,7 +212,6 @@ function resetLaunchSpies() {
   sendUserMessageCalls = [];
   cryptoGetKeys = [];
   reportErrorCalls = [];
-  eventCollectorCreateCalls = [];
   visibleOfferings = [...DEFAULT_VISIBLE_OFFERINGS];
   frozenProjection = INERT_PROJECTION;
   sendUserMessageImpl = async () => new Uint8Array([1]);

packages/workflows/src/index.ts

@@ -56,8 +56,8 @@ export {
 } from "./schedule/cron";
 export { scheduleCronFromProjection } from "./schedule/from-projection";
 export {
-  ensureRunSession,
   recordAgentSessionAtProvision,
+  ensureRunSession,
   endAgentSessionForPrincipal,
   endAgentSessionForRun,
   type EventCollectorPort,