Chat delivers room mail through Interchange's run-trigger route (CL-7490)

apps/hub/src/index.ts

@@ -23,6 +23,7 @@ import {
   model,
   modelPricing,
   tenant as tenantTable,
+  user as userTable,
   workflowDefinition,
 } from "@intx/db/schema";
 import { and, eq, inArray } from "drizzle-orm";
@@ -92,6 +93,7 @@ import {
   createHubChatPlatform,
   createNoopInferenceRoutes,
   createRelaunchNoticePoster,
+  createRunTriggerClient,
   createWorkflowParticipantRoutes,
   isWorkbenchHostDefinitionName,
   listConnectedProviders,
@@ -105,6 +107,7 @@ import {
   workbenchLaunchPersistExtra,
   createCryptoProviderCache,
   tagCredentialCipher,
+  verifyInternalRunTriggerToken,
 } from "@corbits/chat";
 import {
   createDrizzleMailboxWriter,
@@ -1129,7 +1132,47 @@ export async function createHub(config: HubConfig) {
     workflowAllocationService,
     workflowDispatchService,
     credentialCipher,
+    // `@intx/hub-api`'s `GetSession` is a deliberately pluggable seam
+    // (its own doc comment: "so a third-party identity provider can be
+    // plugged in"), and this composition root owns it. `@corbits/chat`'s
+    // run-trigger client (CL-7490) has no browser cookie to present for
+    // most of its calls — mention fan-out, a relaunch resend, the
+    // mailbox-fanout persist seam all run with no inbound HTTP request
+    // in scope — so it signs a short-lived internal token instead (see
+    // `@corbits/chat`'s `run-trigger-internal-auth.ts`) naming the
+    // better-auth user it authenticates as. This checks that token
+    // FIRST and only falls through to the real cookie-session path when
+    // the header is absent; a PRESENT-but-invalid token fails closed
+    // (401 via a null session) rather than silently retrying as a
+    // session, the same posture `workflow-run-deploy-auth` documents for
+    // its own bearer mirror in `vendor/intx/hub-api`.
     getSession: async (headers) => {
+      const internalRunTriggerToken = headers.get(
+        "x-corbits-internal-run-trigger",
+      );
+      if (internalRunTriggerToken !== null) {
+        const userId = verifyInternalRunTriggerToken(
+          config.sessionSecret,
+          internalRunTriggerToken,
+        );
+        if (userId === null) return null;
+        const userRow = await db.query.user.findFirst({
+          where: eq(userTable.id, userId),
+        });
+        if (userRow === undefined) return null;
+        const now = new Date();
+        return {
+          user: userRow,
+          session: {
+            id: `internal_run_trigger_${userRow.id}`,
+            createdAt: now,
+            updatedAt: now,
+            userId: userRow.id,
+            expiresAt: new Date(now.getTime() + 30_000),
+            token: internalRunTriggerToken,
+          },
+        };
+      }
       const result = await auth.api.getSession({ headers });
       return result ? { user: result.user, session: result.session } : null;
     },
@@ -1430,9 +1473,17 @@ export async function createHub(config: HubConfig) {
   // across consumers. TTL-bounded by `createCryptoProviderCache` itself
   // (CL-7223).
   const cryptoProviders = createCryptoProviderCache();
+  // Delivers chat's outbound run mail through Interchange's own
+  // workflow-run mail-trigger route (CL-7490), in-process against this
+  // same `app` — see `@corbits/chat`'s `run-trigger-client.ts` and the
+  // `getSession` internal-token check above for how it authenticates.
+  const runTrigger = createRunTriggerClient({
+    app,
+    internalAuthSecret: config.sessionSecret,
+  });
   const chatPlatform = createHubChatPlatform({
     db,
-    sessionService,
+    runTrigger,
     repoStore: agentRepoStore.repoStore,
     sidecarRouter,
     eventCollectors,

packages/chat/src/index.ts

@@ -322,6 +322,17 @@ export type {
   CreateHubChatPlatformDeps,
   HubChatPlatform,
 } from "./platform-adapter";
+export { createRunTriggerClient, RunTriggerError } from "./run-trigger-client";
+export type {
+  RunTriggerClient,
+  RunTriggerClientDeps,
+  TriggerWorkflowRunMailInput,
+  TriggeredWorkflowRunMail,
+} from "./run-trigger-client";
+export {
+  signInternalRunTriggerToken,
+  verifyInternalRunTriggerToken,
+} from "./run-trigger-internal-auth";
 export { createCryptoProviderCache } from "./crypto-cache";
 export type { CryptoProviderCache } from "./crypto-cache";
 export { tagCredentialCipher } from "./credential-cipher-tag";

packages/chat/src/platform-adapter.ts

@@ -31,6 +31,7 @@ import type { CryptoProviderCache } from "./crypto-cache";
 import { InferenceResolutionError } from "./model-unavailable";
 import type { ToolGrantsForPins } from "./pin-ports";
 import { resolveRunSessionIdOrThrow } from "./run-session";
+import type { RunTriggerClient } from "./run-trigger-client";
 import { sendRunMail, type SendRunMailParams } from "./send-run-mail";
 import {
   findStandingLaunchByDefinition,
@@ -53,6 +54,7 @@ import type { DB } from "@intx/db";
 import { listVisibleOfferings } from "@intx/db";
 import {
   asset as assetTable,
+  principal as principalTable,
   sessionMail,
   tenant as tenantTable,
   workflowDefinition,
@@ -68,7 +70,6 @@ import {
   DEFAULT_ASSET_REF,
   type EventCollectorRegistry,
   type RepoStore,
-  type SessionService,
   type SidecarRouter,
   type WorkflowAllocationService,
 } from "@intx/hub-sessions";
@@ -94,7 +95,15 @@ function domainOf(address: string): string {
 
 export type CreateHubChatPlatformDeps = {
   db: DB["db"];
-  sessionService: Pick<SessionService, "sendUserMessage">;
+  /**
+   * Delivers outbound run mail through Interchange's own workflow-run
+   * mail-trigger route (CL-7490) — see `./run-trigger-client.ts` for why
+   * a raw `sessionService.sendUserMessage` can no longer stand in for
+   * this: it never reaches the trigger route's durable-dispatch branch,
+   * so a provisioned run invited but never yet triggered natively never
+   * gets its first `agent.deploy`.
+   */
+  runTrigger: RunTriggerClient;
   repoStore: Pick<RepoStore, "resolveRef">;
   /**
    * Interchange's provisioned-deployment front. Invite, wake, and
@@ -263,10 +272,68 @@ export function createHubChatPlatform(
   void tagCredentialCipher(deps.credentialCipher);
   const runMailDeps = {
     db: deps.db,
-    sessionService: deps.sessionService,
+    runTrigger: deps.runTrigger,
     sidecarRouter: deps.sidecarRouter,
   };
 
+  /**
+   * Resolves the principal a run-trigger's DB row names, then the
+   * better-auth `user.id` behind it — the identity
+   * `./run-trigger-client.ts` authenticates the trigger call as. Only a
+   * `kind: "user"` principal has a real login identity to authenticate
+   * as; anything else is a wiring bug this throws loud on rather than
+   * silently authenticating as nobody.
+   */
+  async function refIdForUserPrincipal(principalId: string): Promise<string> {
+    const rows = await deps.db
+      .select({ refId: principalTable.refId, kind: principalTable.kind })
+      .from(principalTable)
+      .where(eq(principalTable.id, principalId))
+      .limit(1);
+    const row = rows[0];
+    if (row === undefined || row.kind !== "user") {
+      throw new Error(
+        `principal "${principalId}" is not a user principal chat can authenticate the run trigger as`,
+      );
+    }
+    return row.refId;
+  }
+
+  /**
+   * The identity a `sendMail` call authenticates its run-trigger
+   * delivery as (see `./run-trigger-client.ts`'s `authAsUserId`):
+   * the sending human when `sendMail` named one directly, and
+   * otherwise the run's own workflow definition asset's
+   * `creatorPrincipalId` — the same authority
+   * `sourceAuthorityForAsset` already resolves for provisioning.
+   */
+  async function resolveTriggerAuthUserId(input: {
+    readonly principalId: string | undefined;
+    readonly definitionId: string | null;
+  }): Promise<string> {
+    if (input.principalId !== undefined) {
+      return refIdForUserPrincipal(input.principalId);
+    }
+    if (input.definitionId === null) {
+      throw new Error(
+        "run has no definitionId to resolve a run-trigger authority from",
+      );
+    }
+    const rows = await deps.db
+      .select({ assetId: workflowDefinition.assetId })
+      .from(workflowDefinition)
+      .where(eq(workflowDefinition.id, input.definitionId))
+      .limit(1);
+    const assetId = rows[0]?.assetId;
+    if (assetId === null || assetId === undefined) {
+      throw new Error(
+        `workflow definition "${input.definitionId}" has no asset`,
+      );
+    }
+    const creatorPrincipalId = await sourceAuthorityForAsset(assetId);
+    return refIdForUserPrincipal(creatorPrincipalId);
+  }
+
   function offeringDigest(sourceOfferingIds: readonly string[]): string {
     // `sources_digest` is a `text` column and this digest rides straight
     // into it (`platform-adapter.ts`'s `insert(workbenchLaunch)`). A
@@ -974,7 +1041,11 @@ export function createHubChatPlatform(
    */
   async function sendRunMailWithReclaimRetry(
     params: SendRunMailParams,
-    reresolve: () => Promise<{ agentAddress: string; sessionId: string }>,
+    reresolve: () => Promise<{
+      agentAddress: string;
+      anchorRunId: string;
+      sessionId: string;
+    }>,
   ): Promise<Awaited<ReturnType<typeof sendRunMail>>> {
     let current = params;
     let wakedOnFailure = false;
@@ -1012,6 +1083,7 @@ export function createHubChatPlatform(
             current = {
               ...current,
               agentAddress: fresh.agentAddress,
+              anchorRunId: fresh.anchorRunId,
               sessionId: fresh.sessionId,
             };
             throw err;
@@ -1370,14 +1442,21 @@ export function createHubChatPlatform(
         }),
       );
 
+      const authAsUserId = await resolveTriggerAuthUserId({
+        principalId: input.principalId,
+        definitionId: delivery.run.definitionId,
+      });
+
       const sendMailBase = {
         tenantId: input.tenantId,
         sessionId,
         agentAddress: deliveryAddress,
+        anchorRunId: delivery.run.id,
         from,
         domain,
         content: input.content.content,
         cryptoProvider,
+        authAsUserId,
       };
       const withAttachments =
         attachments !== undefined
@@ -1424,6 +1503,7 @@ export function createHubChatPlatform(
           }
           return {
             agentAddress: relive.binding.liveAddress,
+            anchorRunId: relive.run.id,
             sessionId: await resolveRunSessionIdOrThrow(deps.db, relive.run),
           };
         },