Record a run's session at provision; re-key it when Interchange anchors the run (CL-7481)

apps/hub/src/index.ts

@@ -168,6 +168,7 @@ import {
 import {
   createHubMailboxAuthorizeSender,
   createHubMailboxResolveRefs,
+  createHubPersistMailWithSessionEnsure,
 } from "./mailbox-persist";
 import {
   createCommandRegistry,
@@ -196,6 +197,8 @@ import {
   createScheduledWorkflowRoutes,
   renderWorkflowSourceTree,
   WORKFLOW_SOURCE_ENTRY,
+  ensureRunSession,
+  recordAgentSessionAtProvision,
 } from "@corbits/workflows";
 import {
   createSidecarProvisioner as createE2BSidecarProvisioner,
@@ -219,6 +222,7 @@ import {
   createWorkflowAllocationService,
   createWorkflowDispatchService,
   DEFAULT_ASSET_REF,
+  resolveRoutableAddress,
   WorkflowProvisioningError,
   type AgentRepoStore,
   type EventCollectorRegistry,
@@ -713,6 +717,15 @@ export async function createHub(config: HubConfig) {
       args: Parameters<typeof baseLookups.registerSignalCorrelation>[0],
     ) => Promise<void>;
   } = {};
+  // Forward reference: `eventCollectors` (the wrapped
+  // `EventCollectorRegistry`) isn't constructed until later in this
+  // composition, but `createHubPersistMailWithSessionEnsure` below needs
+  // it. Set once `eventCollectors` exists; every persistMail call before
+  // that point (there are none — the server isn't serving requests yet)
+  // would just no-op.
+  const eventCollectorsRef: {
+    current?: Pick<EventCollectorRegistry, "create" | "has">;
+  } = {};
   // CL-6499 (native multi-step routines): materializes a mail-triggered
   // run's authorization grants from its deploy-approved snapshot, so
   // ANY plain mail delivered to a workflow deployment's address — not
@@ -746,7 +759,11 @@ export async function createHub(config: HubConfig) {
     // workbench ref is present before the post-commit bus event fires --
     // no out-of-band UPDATE, no polling read.
     persistMail: createMailboxPersist(mailboxDb, {
-      upstream: baseLookups.persistMail,
+      upstream: createHubPersistMailWithSessionEnsure(
+        db,
+        eventCollectorsRef,
+        baseLookups.persistMail,
+      ),
       authorizeSender: createHubMailboxAuthorizeSender(db),
       bus: mailboxBus,
       resolveRefs: createHubMailboxResolveRefs(chatStore, roomMessages),
@@ -885,6 +902,17 @@ export async function createHub(config: HubConfig) {
       baseEventCollectors.create(agentAddress, tenantId, sessionId, runId);
     },
     dispatch(agentAddress, event) {
+      // CL-7480: a run's turn events can start arriving before anything
+      // else ever recorded its session — the first inbound trigger that
+      // just reconciled its principal races this same dispatch. No live
+      // collector for the address is exactly that case: ensure the
+      // session (and, inside it, the collector) before delegating,
+      // rather than silently dropping the event for a collector that
+      // never gets created.
+      if (!baseEventCollectors.has(agentAddress)) {
+        void ensureCollectorThenDispatch(agentAddress, event);
+        return;
+      }
       turnLatency.onEvent(agentAddress, event);
       baseEventCollectors.dispatch(agentAddress, event);
       // Mirrors the registry's own `isTerminal` check (event-collector-registry.ts)
@@ -901,6 +929,36 @@ export async function createHub(config: HubConfig) {
       baseEventCollectors.abandon(agentAddress);
     },
   };
+  eventCollectorsRef.current = eventCollectors;
+  // Named so `dispatch`'s no-collector branch above reads clearly; not
+  // inlined there because it needs to be `async` and `dispatch` itself
+  // must stay synchronous (the `EventCollectorRegistry` contract).
+  async function ensureCollectorThenDispatch(
+    agentAddress: string,
+    event: Parameters<EventCollectorRegistry["dispatch"]>[1],
+  ): Promise<void> {
+    try {
+      const run = await resolveRoutableAddress(db, agentAddress);
+      if (run !== undefined) {
+        await ensureRunSession({
+          db,
+          eventCollectors: baseEventCollectors,
+          runId: run.id,
+        });
+      }
+    } catch (err) {
+      reportError(err, {
+        operation: "hub.eventCollectors.ensureRunSession",
+        extra: { agentAddress },
+      });
+    }
+    turnLatency.onEvent(agentAddress, event);
+    baseEventCollectors.dispatch(agentAddress, event);
+    const isTerminal =
+      event.type === "reactor.done" ||
+      (event.type === "reactor.error" && event.data.fatal);
+    if (isTerminal) turnLatency.onSessionEnd(agentAddress);
+  }
   createHubSessionOrchestrator({
     events: sidecarRouter.events,
     router: sidecarRouter,
@@ -2017,15 +2075,7 @@ export async function createHub(config: HubConfig) {
       }
 
       try {
-        // Not `recordAgentSessionForRun` here: this mints a deployment
-        // with no message sent against it — a `workflow_run` born this
-        // way carries no principal until its first trigger (a call this
-        // deployer never makes) reconciles one onto it, so the call
-        // would only ever no-op. That trigger runs entirely through
-        // Interchange's own native route, with no Workbench-owned hook
-        // to record the session afterward (CL-7477 tracks the chat,
-        // webhook, and one-shot-prompt launch paths, which all send a
-        // message of their own and record there instead).
+        const sessionId = generateId("session");
         const prepared =
           await workflowAllocationService.prepareProvisionedDeployment({
             tenantId,
@@ -2038,12 +2088,26 @@ export async function createHub(config: HubConfig) {
             },
             entry,
             definitionAssetId: assetId,
-            sessionId: generateId("session"),
+            sessionId,
             sourceAuthorityPrincipalId: principalId,
             sourceOfferingIds,
             defaultSourceOfferingId,
             deployContent: { systemPrompt: "" },
           });
+        // The eager record every native launcher makes right after
+        // `prepareProvisionedDeployment` returns (CL-7481): this
+        // deployment mints no opening message of its own, but the
+        // trigger that eventually reconciles a principal onto it runs
+        // entirely through Interchange's own native route, with no
+        // Workbench-owned hook downstream to record the session
+        // afterward — so this is the only chance to record it at all.
+        await recordAgentSessionAtProvision({
+          db,
+          eventCollectors,
+          runId: prepared.anchorRunId,
+          sessionId,
+          sourceAuthorityPrincipalId: principalId,
+        });
         return {
           deploymentId: prepared.anchorRunId,
           definitionAssetId: assetId,
@@ -2269,10 +2333,10 @@ export async function createHub(config: HubConfig) {
             repoStore: agentRepoStore.repoStore,
             workflowAllocationService,
             credentialCipher,
+            eventCollectors,
             isRoutable: (address) =>
               sidecarRouter.getRoutableAddresses().includes(address),
             cryptoProviderCache: cryptoProviders,
-            eventCollectors,
             persistLaunch: async (input) => {
               await workbenchLaunchPersistExtra(input)(db);
             },

apps/hub/src/mailbox-persist.ts

@@ -16,6 +16,7 @@ import { tenant as tenantTable } from "@intx/db/schema";
 import type {
   AuthorizeMailboxSender,
   CreateMailboxPersistOpts,
+  MailboxPersistArgs,
   MailboxRef,
 } from "@corbits/mailbox";
 import { resolveRoutableAddress } from "@intx/hub-sessions";
@@ -24,6 +25,11 @@ import {
   type ChatStore,
   type RoomMessageStore,
 } from "@corbits/chat";
+import { ensureRunSession, type EventCollectorPort } from "@corbits/workflows";
+import { reportError } from "@corbits/error-sink";
+import { getLogger } from "@intx/log";
+
+const logger = getLogger(["hub", "mailbox-persist"]);
 
 /**
  * `@corbits/mailbox` does not export `ResolveMailboxRefs` itself (only the
@@ -57,6 +63,68 @@ type ResolveMailboxRefs = NonNullable<
  * fact from two directions: one extra indexed lookup by address per frame,
  * not per recipient.
  */
+/**
+ * Wraps the hub's own `persistMail` (vendor's `baseLookups.persistMail`)
+ * so a run's `agent_session` and event collector exist before that write
+ * runs — CL-7480: the first inbound mail on a freshly triggered run is
+ * often the earliest point the run is mail-routable at all, since
+ * `workflow_run.principal_id` only reconciles onto the trigger that just
+ * fired. Resolves the frame's own sender address to its run
+ * (`resolveRoutableAddress`, the same resolver `persistMail` itself
+ * uses) rather than assuming the caller already knows it; a sender that
+ * is not a live run resolves to `undefined` and this is a no-op past
+ * that point. Best-effort: `ensureRunSession` failing must not swallow
+ * the mail upstream is about to persist regardless, so this only
+ * degrades to "no session yet" rather than ever throwing past itself.
+ *
+ * `eventCollectorsRef` is a forward reference (the same pattern this
+ * file's other composition-order refs use, e.g. `apps/hub/src/index.ts`'s
+ * `grantAllowanceGateRef`): the wrapped `eventCollectors` registry isn't
+ * constructed until after `lookups` is, so this only reads `.current`
+ * inside the returned closure, never at wiring time.
+ */
+export function createHubPersistMailWithSessionEnsure<
+  R extends readonly unknown[],
+>(
+  db: DB["db"],
+  eventCollectorsRef: { current?: Pick<EventCollectorPort, "create" | "has"> },
+  upstream: (args: MailboxPersistArgs) => Promise<R>,
+): (args: MailboxPersistArgs) => Promise<R> {
+  return async (args) => {
+    const sender = await resolveRoutableAddress(db, args.senderAddress);
+    try {
+      const eventCollectors = eventCollectorsRef.current;
+      if (sender !== undefined && eventCollectors !== undefined) {
+        await ensureRunSession({ db, eventCollectors, runId: sender.id });
+      }
+    } catch (err) {
+      reportError(err, {
+        operation: "hub.mailboxPersist.ensureRunSession",
+        extra: { senderAddress: args.senderAddress },
+      });
+    }
+    // The vendored `session_mail` ledger (`upstream`, `baseLookups.persistMail`)
+    // keys on the sender run's principal, which does not exist until the
+    // run's first trigger anchors one — a run's own first outbound mail can
+    // reach this wrapper before that happens. `@corbits/mailbox`'s own write
+    // (this function's caller, `createMailboxPersist`) is keyed on the run
+    // address instead and runs regardless, so it is already the durable
+    // record for this frame; skip the vendored delegate rather than let it
+    // throw "no session for address" past the caller.
+    if (sender !== undefined && sender.sessionId === null) {
+      logger.debug(
+        "skipping vendored persistMail for run {runId}: no session yet",
+        { runId: sender.id, senderAddress: args.senderAddress },
+      );
+      // No `session_mail` rows were written for this frame: an empty
+      // result, not a thrown error, so a caller iterating this like
+      // `handleMailPersist` does simply emits no `mail.persisted` events.
+      return [] as unknown as R;
+    }
+    return upstream(args);
+  };
+}
+
 export function createHubMailboxAuthorizeSender(
   db: DB["db"],
 ): AuthorizeMailboxSender {

packages/agent-directory/src/one-shot-prompt.ts

@@ -11,7 +11,7 @@ import {
   endAgentSessionForRun,
   readDefinitionProjection,
   readFoldedBody,
-  recordAgentSessionForRun,
+  recordAgentSessionAtProvision,
   WORKFLOW_SOURCE_ENTRY,
   type EventCollectorPort,
 } from "@corbits/workflows";
@@ -166,10 +166,13 @@ async function provisionOnAsset(
         ? { toolPackagePins: input.foldedBody.toolPackagePins }
         : {}),
     });
-  // Not `recordAgentSessionForRun` here: a freshly provisioned run's
-  // `workflow_run.principal_id` is still null until its first trigger
-  // reconciles one onto it — `runOneShotPrompt` records the session
-  // right after the opening prompt actually sends.
+  await recordAgentSessionAtProvision({
+    db: deps.db,
+    eventCollectors: deps.eventCollectors,
+    runId: prepared.anchorRunId,
+    sessionId,
+    sourceAuthorityPrincipalId: input.principalId,
+  });
   return {
     runId: prepared.anchorRunId,
     address: prepared.deploymentAddress,
@@ -322,25 +325,11 @@ export async function runOneShotPrompt(
             cryptoProvider,
           });
         }
-        // The send above just drove the run's trigger path, the only
-        // thing that ever reconciles a principal onto a freshly
-        // provisioned `workflow_run` (CL-7477) — recording only now is
-        // what makes this ever find one.
-        try {
-          await recordAgentSessionForRun(
-            deps.db,
-            {
-              sessionId: launched.sessionId,
-              anchorRunId: launched.runId,
-            },
-            deps.eventCollectors,
-          );
-        } catch (err) {
-          reportError(err, {
-            operation: "agent-directory.one-shot.recordAgentSession",
-            extra: { address: launched.address },
-          });
-        }
+        // A run's session and event collector are ensured lazily now, at
+        // the hub seams that actually see the run become mail-routable
+        // (`apps/hub/src/mailbox-persist.ts`, the wrapped
+        // `eventCollectors` dispatch in `apps/hub/src/index.ts`) —
+        // CL-7480. Nothing here needs to record it.
       } catch (cause) {
         reportError(cause, {
           operation: "agent-directory.one-shot.send",

packages/chat/src/platform-adapter.ts

@@ -17,7 +17,9 @@ import {
   authoredDefinitionCandidates,
   DefinitionProjectionMissingError,
   endAgentSessionForRun,
+  ensureRunSession,
   readFoldedBody,
+  recordAgentSessionAtProvision,
   resolveNewestProjectedDefinition,
   WORKFLOW_SOURCE_ENTRY,
 } from "@corbits/workflows";
@@ -258,7 +260,6 @@ export function createHubChatPlatform(
     db: deps.db,
     sessionService: deps.sessionService,
     sidecarRouter: deps.sidecarRouter,
-    eventCollectors: deps.eventCollectors,
   };
 
   function offeringDigest(sourceOfferingIds: readonly string[]): string {
@@ -344,11 +345,13 @@ export function createHubChatPlatform(
           ? { toolPackagePins: input.foldedBody.toolPackagePins }
           : {}),
       });
-    // Not `recordAgentSessionForRun` here: a freshly provisioned run's
-    // `workflow_run.principal_id` is still null (an invite sits
-    // un-triggered until someone actually writes into it), so this
-    // would only ever no-op. `sendRunMail` records it instead, once the
-    // run's first turn has actually reconciled a principal onto it.
+    await recordAgentSessionAtProvision({
+      db: deps.db,
+      eventCollectors: deps.eventCollectors,
+      runId: prepared.anchorRunId,
+      sessionId,
+      sourceAuthorityPrincipalId: input.sourceAuthorityPrincipalId,
+    });
     return {
       runId: prepared.anchorRunId,
       address: prepared.deploymentAddress,
@@ -1313,6 +1316,25 @@ export function createHubChatPlatform(
       // idle sweep the moment they see traffic.
       lifecycle?.track(deliveryAddress);
 
+      // CL-7480: a run just woken/reconciled above can be routable
+      // before anything ever recorded its session — waking it is not
+      // itself a mail or dispatch seam, so nothing else along this path
+      // would ensure one. Best-effort: an ensure failure here still
+      // falls through to `resolveRunSessionIdOrThrow`'s own error.
+      try {
+        await ensureRunSession({
+          db: deps.db,
+          eventCollectors: deps.eventCollectors,
+          runId: delivery.run.id,
+        });
+      } catch (err) {
+        reportError(err, {
+          operation: "chat.sendMail.ensureRunSession",
+          tenantId: input.tenantId,
+          agentId: deliveryAddress,
+          extra: { runId: delivery.run.id },
+        });
+      }
       const sessionId = await resolveRunSessionIdOrThrow(deps.db, delivery.run);
       const domain = domainOf(deliveryAddress);
       // `fromWorkbenchId` names the room a dispatch speaks for. A room