Native launch creates the run's event collector so turns are recorded (CL-7479)

apps/hub/src/index.ts

@@ -2272,6 +2272,7 @@ export async function createHub(config: HubConfig) {
             isRoutable: (address) =>
               sidecarRouter.getRoutableAddresses().includes(address),
             cryptoProviderCache: cryptoProviders,
+            eventCollectors,
             persistLaunch: async (input) => {
               await workbenchLaunchPersistExtra(input)(db);
             },
@@ -3069,6 +3070,7 @@ export async function createHub(config: HubConfig) {
                 repoStore: agentRepoStore.repoStore,
                 workflowAllocationService,
                 sessionService,
+                eventCollectors,
               },
               runnerInput,
             ),

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

@@ -13,6 +13,7 @@ import {
   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 +58,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`.
@@ -267,7 +269,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",
@@ -321,10 +327,14 @@ export async function runOneShotPrompt(
         // 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,
-          });
+          await recordAgentSessionForRun(
+            deps.db,
+            {
+              sessionId: launched.sessionId,
+              anchorRunId: launched.runId,
+            },
+            deps.eventCollectors,
+          );
         } catch (err) {
           reportError(err, {
             operation: "agent-directory.one-shot.recordAgentSession",

packages/chat/src/platform-adapter.ts

@@ -16,7 +16,7 @@ import { and, desc, eq } from "drizzle-orm";
 import {
   authoredDefinitionCandidates,
   DefinitionProjectionMissingError,
-  endAgentSessionForPrincipal,
+  endAgentSessionForRun,
   readFoldedBody,
   resolveNewestProjectedDefinition,
   WORKFLOW_SOURCE_ENTRY,
@@ -258,6 +258,7 @@ export function createHubChatPlatform(
     db: deps.db,
     sessionService: deps.sessionService,
     sidecarRouter: deps.sidecarRouter,
+    eventCollectors: deps.eventCollectors,
   };
 
   function offeringDigest(sourceOfferingIds: readonly string[]): string {
@@ -476,9 +477,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,

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

@@ -1,6 +1,9 @@
 // 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 {
+  recordAgentSessionForRun,
+  type EventCollectorPort,
+} from "@corbits/workflows";
 import { reportError } from "@corbits/error-sink";
 import { sessionMail } from "@intx/db/schema";
 import type { DB } from "@intx/db";
@@ -11,6 +14,7 @@ export type RunMailDeps = {
   db: DB["db"];
   sessionService: Pick<SessionService, "sendUserMessage">;
   sidecarRouter: Pick<SidecarRouter, "dispatchAgentEvent">;
+  eventCollectors: Pick<EventCollectorPort, "create">;
 };
 
 export type SendRunMailParams = {
@@ -109,10 +113,14 @@ export async function sendRunMail(
   // 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,
-    });
+    await recordAgentSessionForRun(
+      deps.db,
+      {
+        sessionId: params.sessionId,
+        address: params.agentAddress,
+      },
+      deps.eventCollectors,
+    );
   } catch (err) {
     reportError(err, {
       operation: "chat.sendRunMail.recordAgentSession",

packages/webhook-triggers/src/launch.ts

@@ -19,6 +19,7 @@ import {
   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";
@@ -58,6 +59,7 @@ export type LaunchWebhookTriggerDeps = {
    */
   isRoutable: (address: string) => boolean;
   cryptoProviderCache: CryptoProviderCache;
+  eventCollectors: Pick<EventCollectorPort, "create">;
   /**
    * Host-supplied cipher. Interchange decrypts bindings inside
    * `prepareProvisionedDeployment`; this port still owns the field so
@@ -245,10 +247,14 @@ export async function launchWebhookTrigger(
   // a run that never became routable stays without a session until its
   // first successful send.
   try {
-    await recordAgentSessionForRun(deps.db, {
-      sessionId,
-      anchorRunId: prepared.anchorRunId,
-    });
+    await recordAgentSessionForRun(
+      deps.db,
+      {
+        sessionId,
+        anchorRunId: prepared.anchorRunId,
+      },
+      deps.eventCollectors,
+    );
   } catch (error) {
     reportError(error, {
       operation: "webhookTriggers.launch.recordAgentSession",

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

@@ -197,6 +197,9 @@ function baseDeps() {
       },
     },
     isRoutable: () => isRoutableForTest,
+    eventCollectors: {
+      create: () => undefined,
+    },
   };
 }
 

packages/workflows/src/index.ts

@@ -60,6 +60,7 @@ export {
   endAgentSessionForPrincipal,
   endAgentSessionForRun,
   type RecordAgentSessionParams,
+  type EventCollectorPort,
 } from "./launch/agent-session";
 export {
   deliverWhenRoutable,

packages/workflows/src/launch/agent-session.ts

@@ -23,12 +23,27 @@
 import { eq } from "drizzle-orm";
 import type { DB } from "@intx/db";
 import { agentSession, workflowRun } from "@intx/db/schema";
+import type { EventCollectorRegistry } from "@intx/hub-sessions";
 
 export type RecordAgentSessionParams = {
   /** The id `prepareProvisionedDeployment` was called with. */
   readonly sessionId: string;
 } & ({ readonly anchorRunId: string } | { readonly address: string });
 
+/**
+ * The one port every launcher threads the same wrapped
+ * `EventCollectorRegistry` through (`apps/hub/src/index.ts`'s
+ * `eventCollectors`) — never a second registry construction. `create`
+ * is what actually makes `inference_turn`/`turn_part` rows exist for a
+ * run (CL-7479: nothing else ever called it once `folded-runs` was
+ * deleted); `abandon` tears the collector down when a run's session
+ * ends.
+ */
+export type EventCollectorPort = Pick<
+  EventCollectorRegistry,
+  "create" | "abandon"
+>;
+
 /**
  * Inserts the `agent_session` row a run needs to be mail-routable: the
  * run's `tenantId`, `definitionId` (the `agent_id` FK — the column keeps
@@ -41,6 +56,7 @@ export type RecordAgentSessionParams = {
 export async function recordAgentSessionForRun(
   db: DB["db"],
   params: RecordAgentSessionParams,
+  eventCollectors: Pick<EventCollectorPort, "create">,
 ): Promise<boolean> {
   const runRow = await db.query.workflowRun.findFirst({
     where:
@@ -65,6 +81,19 @@ export async function recordAgentSessionForRun(
       updatedAt: now,
     })
     .onConflictDoNothing({ target: agentSession.id });
+
+  const agentAddress = "address" in params ? params.address : runRow.address;
+  if (agentAddress !== null) {
+    // Idempotent across wakes: a collector already live for this
+    // address is replaced, matching the registry's own contract
+    // (`event-collector-registry.ts`'s `create`).
+    eventCollectors.create(
+      agentAddress,
+      runRow.tenantId,
+      params.sessionId,
+      runRow.id,
+    );
+  }
   return true;
 }
 
@@ -87,14 +116,19 @@ export async function endAgentSessionForPrincipal(
 
 /**
  * Same as `endAgentSessionForPrincipal`, keyed by the run's own id
- * instead — for a caller (e.g. a one-shot prompt's teardown) that only
- * ever held the run id, never minted or looked up its principal.
- * No-ops for a run row already gone, or one never triggered (no
- * principal, hence no session was ever recorded for it).
+ * instead — for a caller (e.g. a one-shot prompt's teardown, or a chat
+ * relaunch replacing a terminal run) that only ever held the run id,
+ * never minted or looked up its principal. Also abandons the outgoing
+ * run's event collector, through the same shared
+ * `EventCollectorPort` every launcher threads — one mechanism for
+ * ending a run's session. No-ops for a run row already gone, or one
+ * never triggered (no principal, hence no session was ever recorded
+ * for it).
  */
 export async function endAgentSessionForRun(
   db: DB["db"],
   anchorRunId: string,
+  eventCollectors: Pick<EventCollectorPort, "abandon">,
 ): Promise<void> {
   const runRow = await db.query.workflowRun.findFirst({
     where: eq(workflowRun.id, anchorRunId),
@@ -103,4 +137,7 @@ export async function endAgentSessionForRun(
     return;
   }
   await endAgentSessionForPrincipal(db, runRow.principalId);
+  if (runRow.address !== null) {
+    eventCollectors.abandon(runRow.address);
+  }
 }

packages/workflows/test/agent-session.test.ts

@@ -15,7 +15,10 @@ import { generateId } from "@intx/hub-common";
 import { resolveRunSessionId } from "@intx/hub-sessions";
 import { dbGate } from "../../../scripts/e2e/db-gate";
 
-import { recordAgentSessionForRun } from "../src/launch/agent-session";
+import {
+  endAgentSessionForRun,
+  recordAgentSessionForRun,
+} from "../src/launch/agent-session";
 
 function dbConfigFromUrl(databaseUrl: string) {
   const url = new URL(databaseUrl);
@@ -98,9 +101,55 @@ describeIfDb("recordAgentSessionForRun", () => {
   });
 
   test("writes an agent_session resolveRunSessionId reads back", async () => {
-    await recordAgentSessionForRun(db.db, { sessionId, anchorRunId });
+    await recordAgentSessionForRun(
+      db.db,
+      { sessionId, anchorRunId },
+      { create: () => undefined },
+    );
 
     const resolved = await resolveRunSessionId(db.db, principalId);
     expect(resolved).toBe(sessionId);
   });
+
+  test("creates the run's event collector with its address/tenant/session/run id", async () => {
+    const createCalls: unknown[] = [];
+    const otherSessionId = generateId("session");
+
+    await recordAgentSessionForRun(
+      db.db,
+      { sessionId: otherSessionId, anchorRunId },
+      {
+        create: (...args: unknown[]) => {
+          createCalls.push(args);
+        },
+      },
+    );
+
+    expect(createCalls).toEqual([
+      [
+        `${anchorRunId}@agent-session-${tenantId}.localhost`,
+        tenantId,
+        otherSessionId,
+        anchorRunId,
+      ],
+    ]);
+
+    await db.db
+      .delete(schema.agentSession)
+      .where(eq(schema.agentSession.id, otherSessionId));
+  });
+
+  test("abandons the run's event collector when the session ends", async () => {
+    const abandonCalls: string[] = [];
+
+    await endAgentSessionForRun(db.db, anchorRunId, {
+      abandon: (address: string) => {
+        abandonCalls.push(address);
+      },
+    });
+
+    expect(abandonCalls).toEqual([
+      `${anchorRunId}@agent-session-${tenantId}.localhost`,
+    ]);
+  });
 });