Native launch records the run's agent_session so mail and spans persist (CL-7477)

apps/hub/src/index.ts

@@ -2017,6 +2017,15 @@ 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 prepared =
           await workflowAllocationService.prepareProvisionedDeployment({
             tenantId,

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

@@ -8,8 +8,10 @@ import type { AgentLifecycle } from "@corbits/agent-lifecycle";
 import { connectorReplyContent, messageRunEnded } from "@corbits/agent-events";
 import { reportError } from "@corbits/error-sink";
 import {
+  endAgentSessionForRun,
   readDefinitionProjection,
   readFoldedBody,
+  recordAgentSessionForRun,
   WORKFLOW_SOURCE_ENTRY,
 } from "@corbits/workflows";
 import { listVisibleOfferings, type DB } from "@intx/db";
@@ -162,6 +164,10 @@ 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.
   return {
     runId: prepared.anchorRunId,
     address: prepared.deploymentAddress,
@@ -260,6 +266,14 @@ export async function runOneShotPrompt(
       settled = true;
       clearTimeout(timer);
       unsubscribe();
+      try {
+        await endAgentSessionForRun(deps.db, launched.runId);
+      } catch (err) {
+        reportError(err, {
+          operation: "agent-directory.one-shot.end-session",
+          extra: { runId: launched.runId, reason },
+        });
+      }
       try {
         await deps.undeploy(launched.address, reason);
       } catch (err) {
@@ -290,18 +304,33 @@ export async function runOneShotPrompt(
             content: input.prompt,
             cryptoProvider,
           });
-          return;
+        } else {
+          await deps.sessionService.sendUserMessage({
+            agentAddress: launched.address,
+            from: `${input.principalId}@${tenantRow.domain}`,
+            messageId: `<${crypto.randomUUID()}@${tenantRow.domain}>`,
+            date: new Date(),
+            content: input.prompt,
+            sessionId: launched.sessionId,
+            tenantId: input.tenantId,
+            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 },
+          });
         }
-        await deps.sessionService.sendUserMessage({
-          agentAddress: launched.address,
-          from: `${input.principalId}@${tenantRow.domain}`,
-          messageId: `<${crypto.randomUUID()}@${tenantRow.domain}>`,
-          date: new Date(),
-          content: input.prompt,
-          sessionId: launched.sessionId,
-          tenantId: input.tenantId,
-          cryptoProvider,
-        });
       } catch (cause) {
         reportError(cause, {
           operation: "agent-directory.one-shot.send",

packages/chat/src/platform-adapter.ts

@@ -16,6 +16,7 @@ import { and, desc, eq } from "drizzle-orm";
 import {
   authoredDefinitionCandidates,
   DefinitionProjectionMissingError,
+  endAgentSessionForPrincipal,
   readFoldedBody,
   resolveNewestProjectedDefinition,
   WORKFLOW_SOURCE_ENTRY,
@@ -342,6 +343,11 @@ 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.
     return {
       runId: prepared.anchorRunId,
       address: prepared.deploymentAddress,
@@ -470,6 +476,10 @@ export function createHubChatPlatform(
       foldedBody: binding.foldedBody,
     });
 
+    if (run.principalId !== null) {
+      await endAgentSessionForPrincipal(deps.db, run.principalId);
+    }
+
     await repointBinding(
       deps.db,
       binding,

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

@@ -1,5 +1,7 @@
 // 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";
@@ -100,5 +102,24 @@ 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 },
+    });
+  }
   return recordRunMail(deps, params, mailId, now, rawMIME);
 }

packages/webhook-triggers/src/launch.ts

@@ -17,6 +17,7 @@ import {
   deliverWhenRoutable,
   readDefinitionProjection,
   readFoldedBody,
+  recordAgentSessionForRun,
   WORKFLOW_SOURCE_ENTRY,
 } from "@corbits/workflows";
 import { listVisibleOfferings, type DB } from "@intx/db";
@@ -237,6 +238,26 @@ 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 },
+    });
+  }
+
   return {
     instanceId: prepared.anchorRunId,
     triggerAddress: prepared.deploymentAddress,

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

@@ -88,7 +88,25 @@ 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,
+          principalId: "prn_interchange",
+        }),
+      },
     },
+    insert: (_table: unknown) => ({
+      values: (_values: unknown) => ({
+        onConflictDoNothing: async (_opts: unknown) => {},
+      }),
+    }),
   };
 }
 

packages/workflows/src/index.ts

@@ -55,6 +55,12 @@ export {
   type ZonedParts,
 } from "./schedule/cron";
 export { scheduleCronFromProjection } from "./schedule/from-projection";
+export {
+  recordAgentSessionForRun,
+  endAgentSessionForPrincipal,
+  endAgentSessionForRun,
+  type RecordAgentSessionParams,
+} from "./launch/agent-session";
 export {
   deliverWhenRoutable,
   isAgentUnreachableError,

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

@@ -0,0 +1,106 @@
+// Every Workbench launch path provisions through Interchange's
+// `prepareProvisionedDeployment` with a `sessionId` it mints itself, but
+// nothing ever wrote that id into `agent_session` (vendor-owned,
+// `vendor/intx/db/src/schema/sessions.ts`) — the table
+// `resolveRunSessionId` (`vendor/intx/hub-sessions/src/hub-session-lookups.ts`)
+// reads to route a run's outbound mail. Before the folded-runs package
+// was deleted it was the only writer of that table; every current
+// launcher must now do this itself, through this one shared helper, so
+// mail and spans persist against a real session from the run's first
+// turn (CL-7477).
+//
+// Timing matters: `workflow_run.principal_id` (the FK `agent_session`
+// keys on) is still null the instant `prepareProvisionedDeployment`
+// returns — a provisioned anchor is born "deployed" with no principal,
+// and only the run's first trigger reconciles one onto it (Interchange's
+// `anchorWithPrincipal`, `vendor/intx/db/src/workflow-run-store.ts`),
+// through the exact same trigger path every `sendUserMessage` drives. So
+// this cannot run right after provisioning (an invite sits un-triggered
+// until someone actually writes into it) — call it after a send that
+// just delivered the run's first turn instead. A run with no principal
+// yet is a normal, expected state, not a bug: this returns `false`
+// rather than throwing, and idempotent past its first successful call.
+import { eq } from "drizzle-orm";
+import type { DB } from "@intx/db";
+import { agentSession, workflowRun } from "@intx/db/schema";
+
+export type RecordAgentSessionParams = {
+  /** The id `prepareProvisionedDeployment` was called with. */
+  readonly sessionId: string;
+} & ({ readonly anchorRunId: string } | { readonly address: string });
+
+/**
+ * Inserts the `agent_session` row a run needs to be mail-routable: the
+ * run's `tenantId`, `definitionId` (the `agent_id` FK — the column keeps
+ * its old name but targets `workflow_definition` now), and `principalId`
+ * are all read fresh off the `workflow_run` row, never guessed by the
+ * caller. Returns `false` (and writes nothing) when the row has no
+ * principal yet, `true` once the session is recorded (or already was —
+ * a second call for the same `sessionId` is a no-op, not a conflict).
+ */
+export async function recordAgentSessionForRun(
+  db: DB["db"],
+  params: RecordAgentSessionParams,
+): Promise<boolean> {
+  const runRow = await db.query.workflowRun.findFirst({
+    where:
+      "anchorRunId" in params
+        ? eq(workflowRun.id, params.anchorRunId)
+        : eq(workflowRun.address, params.address),
+  });
+  if (runRow === undefined || runRow.principalId === null) {
+    return false;
+  }
+
+  const now = new Date();
+  await db
+    .insert(agentSession)
+    .values({
+      id: params.sessionId,
+      tenantId: runRow.tenantId,
+      agentId: runRow.definitionId,
+      principalId: runRow.principalId,
+      status: "active",
+      createdAt: now,
+      updatedAt: now,
+    })
+    .onConflictDoNothing({ target: agentSession.id });
+  return true;
+}
+
+/**
+ * Marks a run's `agent_session` ended — called only from the places that
+ * already know the run died: a chat relaunch (the old run's session) or
+ * a one-shot prompt's teardown. No sweeper; a run nobody reacts to stays
+ * `active` until something does.
+ */
+export async function endAgentSessionForPrincipal(
+  db: DB["db"],
+  principalId: string,
+): Promise<void> {
+  const endedAt = new Date();
+  await db
+    .update(agentSession)
+    .set({ status: "ended", endedAt, updatedAt: endedAt })
+    .where(eq(agentSession.principalId, principalId));
+}
+
+/**
+ * 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).
+ */
+export async function endAgentSessionForRun(
+  db: DB["db"],
+  anchorRunId: string,
+): Promise<void> {
+  const runRow = await db.query.workflowRun.findFirst({
+    where: eq(workflowRun.id, anchorRunId),
+  });
+  if (runRow === undefined || runRow.principalId === null) {
+    return;
+  }
+  await endAgentSessionForPrincipal(db, runRow.principalId);
+}