Hub ensures a run's session and collector on first use (CL-7480)

apps/hub/src/index.ts

@@ -168,6 +168,7 @@ import {
 import {
   createHubMailboxAuthorizeSender,
   createHubMailboxResolveRefs,
+  createHubPersistMailWithSessionEnsure,
 } from "./mailbox-persist";
 import {
   createCommandRegistry,
@@ -196,6 +197,7 @@ import {
   createScheduledWorkflowRoutes,
   renderWorkflowSourceTree,
   WORKFLOW_SOURCE_ENTRY,
+  ensureRunSession,
 } from "@corbits/workflows";
 import {
   createSidecarProvisioner as createE2BSidecarProvisioner,
@@ -219,6 +221,7 @@ import {
   createWorkflowAllocationService,
   createWorkflowDispatchService,
   DEFAULT_ASSET_REF,
+  resolveRoutableAddress,
   WorkflowProvisioningError,
   type AgentRepoStore,
   type EventCollectorRegistry,
@@ -713,6 +716,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 +758,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 +901,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 +928,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,
@@ -3069,6 +3126,7 @@ export async function createHub(config: HubConfig) {
                 repoStore: agentRepoStore.repoStore,
                 workflowAllocationService,
                 sessionService,
+                eventCollectors,
               },
               runnerInput,
             ),

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,8 @@ import {
   type ChatStore,
   type RoomMessageStore,
 } from "@corbits/chat";
+import { ensureRunSession, type EventCollectorPort } from "@corbits/workflows";
+import { reportError } from "@corbits/error-sink";
 
 /**
  * `@corbits/mailbox` does not export `ResolveMailboxRefs` itself (only the
@@ -57,6 +60,48 @@ 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>(
+  db: DB["db"],
+  eventCollectorsRef: { current?: Pick<EventCollectorPort, "create" | "has"> },
+  upstream: (args: MailboxPersistArgs) => Promise<R>,
+): (args: MailboxPersistArgs) => Promise<R> {
+  return async (args) => {
+    try {
+      const eventCollectors = eventCollectorsRef.current;
+      const sender = await resolveRoutableAddress(db, args.senderAddress);
+      if (sender !== undefined && eventCollectors !== undefined) {
+        await ensureRunSession({ db, eventCollectors, runId: sender.id });
+      }
+    } catch (err) {
+      reportError(err, {
+        operation: "hub.mailboxPersist.ensureRunSession",
+        extra: { senderAddress: args.senderAddress },
+      });
+    }
+    return upstream(args);
+  };
+}
+
 export function createHubMailboxAuthorizeSender(
   db: DB["db"],
 ): AuthorizeMailboxSender {

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

@@ -11,8 +11,8 @@ import {
   endAgentSessionForRun,
   readDefinitionProjection,
   readFoldedBody,
-  recordAgentSessionForRun,
   WORKFLOW_SOURCE_ENTRY,
+  type EventCollectorPort,
 } from "@corbits/workflows";
 import { listVisibleOfferings, type DB } from "@intx/db";
 import { tenant as tenantTable, workflowDefinition } from "@intx/db/schema";
@@ -57,6 +57,7 @@ export type OneShotRunnerDeps = {
     "prepareProvisionedDeployment"
   >;
   readonly sessionService: Pick<SessionService, "sendUserMessage">;
+  readonly eventCollectors: EventCollectorPort;
   /**
    * Test seam only. Production never sets these; they default to
    * Interchange `prepareProvisionedDeployment` and `sendUserMessage`.
@@ -164,10 +165,10 @@ async function provisionOnAsset(
         ? { toolPackagePins: input.foldedBody.toolPackagePins }
         : {}),
     });
-  // Not `recordAgentSessionForRun` here: a freshly provisioned run's
+  // Not `ensureRunSession` 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.
+  // reconciles one onto it, and the hub's own mail/dispatch seams
+  // ensure the session lazily once that happens (CL-7480).
   return {
     runId: prepared.anchorRunId,
     address: prepared.deploymentAddress,
@@ -267,7 +268,11 @@ export async function runOneShotPrompt(
       clearTimeout(timer);
       unsubscribe();
       try {
-        await endAgentSessionForRun(deps.db, launched.runId);
+        await endAgentSessionForRun(
+          deps.db,
+          launched.runId,
+          deps.eventCollectors,
+        );
       } catch (err) {
         reportError(err, {
           operation: "agent-directory.one-shot.end-session",
@@ -316,21 +321,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,
-          });
-        } 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

@@ -16,7 +16,8 @@ import { and, desc, eq } from "drizzle-orm";
 import {
   authoredDefinitionCandidates,
   DefinitionProjectionMissingError,
-  endAgentSessionForPrincipal,
+  endAgentSessionForRun,
+  ensureRunSession,
   readFoldedBody,
   resolveNewestProjectedDefinition,
   WORKFLOW_SOURCE_ENTRY,
@@ -343,11 +344,12 @@ export function createHubChatPlatform(
           ? { toolPackagePins: input.foldedBody.toolPackagePins }
           : {}),
       });
-    // Not `recordAgentSessionForRun` here: a freshly provisioned run's
+    // Not `ensureRunSession` 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.
+    // would only ever no-op. The hub's own mail/dispatch seams ensure
+    // the session lazily instead, once the run's first turn has
+    // actually reconciled a principal onto it (CL-7480).
     return {
       runId: prepared.anchorRunId,
       address: prepared.deploymentAddress,
@@ -476,9 +478,7 @@ export function createHubChatPlatform(
       foldedBody: binding.foldedBody,
     });
 
-    if (run.principalId !== null) {
-      await endAgentSessionForPrincipal(deps.db, run.principalId);
-    }
+    await endAgentSessionForRun(deps.db, run.id, deps.eventCollectors);
 
     await repointBinding(
       deps.db,
@@ -1314,6 +1314,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

packages/chat/src/send-run-mail.ts

@@ -1,7 +1,5 @@
 // Signs one inbound message via Interchange `sendUserMessage` and
 // records it on session_mail so the run's own live subscribers see it.
-import { recordAgentSessionForRun } from "@corbits/workflows";
-import { reportError } from "@corbits/error-sink";
 import { sessionMail } from "@intx/db/schema";
 import type { DB } from "@intx/db";
 import type { SessionService, SidecarRouter } from "@intx/hub-sessions";
@@ -102,24 +100,10 @@ export async function sendRunMail(
   const mailId = crypto.randomUUID();
   const now = new Date();
   const rawMIME = await deliverRunMailMIME(deps, params, mailId, now);
-  // `sendUserMessage` just drove this run's trigger path, which is the
-  // only thing that ever reconciles a principal onto a freshly
-  // provisioned `workflow_run` (CL-7477) — so the run is mail-routable
-  // now if it never was before. Best-effort and idempotent past the
-  // first successful call; a failure here must not swallow the mail
-  // that already sent.
-  try {
-    await recordAgentSessionForRun(deps.db, {
-      sessionId: params.sessionId,
-      address: params.agentAddress,
-    });
-  } catch (err) {
-    reportError(err, {
-      operation: "chat.sendRunMail.recordAgentSession",
-      tenantId: params.tenantId,
-      agentId: params.agentAddress,
-      extra: { sessionId: params.sessionId },
-    });
-  }
+  // 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.
   return recordRunMail(deps, params, mailId, now, rawMIME);
 }

packages/webhook-triggers/src/launch.ts

@@ -17,7 +17,6 @@ import {
   deliverWhenRoutable,
   readDefinitionProjection,
   readFoldedBody,
-  recordAgentSessionForRun,
   WORKFLOW_SOURCE_ENTRY,
 } from "@corbits/workflows";
 import { listVisibleOfferings, type DB } from "@intx/db";
@@ -238,25 +237,11 @@ export async function launchWebhookTrigger(
     });
   }
 
-  // The delivery attempt above is what 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, after
-  // that attempt, is what makes this call ever find one. Best-effort:
-  // a run that never became routable stays without a session until its
-  // first successful send.
-  try {
-    await recordAgentSessionForRun(deps.db, {
-      sessionId,
-      anchorRunId: prepared.anchorRunId,
-    });
-  } catch (error) {
-    reportError(error, {
-      operation: "webhookTriggers.launch.recordAgentSession",
-      tenantId: trigger.tenantId,
-      agentId: prepared.deploymentAddress,
-      extra: { instanceId: prepared.anchorRunId },
-    });
-  }
+  // 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.
 
   return {
     instanceId: prepared.anchorRunId,