Chat: workbench_launch lookups are tenant-scoped (CL-7474)

packages/chat/src/agent-binding.test.ts

@@ -5,6 +5,7 @@ import { describe, expect, test } from "bun:test";
 import {
   isBeyondWake,
   readBindingByAddress,
+  readBindingByAddressAnyTenant,
   resolveRoomAddress,
 } from "./agent-binding";
 
@@ -71,6 +72,7 @@ describe("readBindingByAddress", () => {
     const binding = await readBindingByAddress(
       fakeDb([relaunched]),
       "run_original@acme.example",
+      "ten_1",
     );
     expect(binding?.stableId).toBe("run_original");
     expect(binding?.currentRunId).toBe("run_fresh");
@@ -85,17 +87,78 @@ describe("readBindingByAddress", () => {
     const binding = await readBindingByAddress(
       fakeDb([relaunched]),
       "run_fresh@acme.example",
+      "ten_1",
     );
     expect(binding?.roomAddress).toBe("run_original@acme.example");
   });
 
   test("is undefined for an address this package never launched", async () => {
     expect(
-      await readBindingByAddress(fakeDb([relaunched]), "echo_1@acme.example"),
+      await readBindingByAddress(
+        fakeDb([relaunched]),
+        "echo_1@acme.example",
+        "ten_1",
+      ),
     ).toBeUndefined();
   });
 });
 
+describe("readBindingByAddress: tenant scoping (CL-7474)", () => {
+  // Two tenants that each independently invited "the same" agent
+  // (definitionId happens to differ per tenant in practice, but the
+  // launch rows below are shaped exactly like two independent
+  // launches — different stable ids, different run ids, different
+  // tenants). `instanceId`/`currentRunId` are collision-resistant
+  // generated ids, not scoped to a tenant, so nothing but an explicit
+  // `expectedTenantId` check stops tenant B's caller from resolving
+  // tenant A's row if it ever guessed or replayed tenant A's address.
+  const tenantALaunch: LaunchRow = {
+    tenantId: "tnt_a",
+    instanceId: "run_a1",
+    currentRunId: "run_a1",
+    priorRunIds: [],
+    foldedBody: FOLDED_BODY,
+  };
+  const tenantBLaunch: LaunchRow = {
+    tenantId: "tnt_b",
+    instanceId: "run_b1",
+    currentRunId: "run_b1",
+    priorRunIds: [],
+    foldedBody: FOLDED_BODY,
+  };
+  const db = fakeDb([tenantALaunch, tenantBLaunch]);
+
+  test("each tenant's DM resolves to its own distinct run id and address", async () => {
+    const a = await readBindingByAddress(db, "run_a1@acme.example", "tnt_a");
+    const b = await readBindingByAddress(db, "run_b1@acme.example", "tnt_b");
+    expect(a?.currentRunId).toBe("run_a1");
+    expect(b?.currentRunId).toBe("run_b1");
+    expect(a?.currentRunId).not.toBe(b?.currentRunId);
+    expect(a?.roomAddress).not.toBe(b?.roomAddress);
+  });
+
+  test("a bench invite in tenant B never resolves tenant A's launch", async () => {
+    expect(
+      await readBindingByAddress(db, "run_a1@acme.example", "tnt_b"),
+    ).toBeUndefined();
+    expect(
+      await readBindingByAddress(db, "run_b1@acme.example", "tnt_a"),
+    ).toBeUndefined();
+  });
+});
+
+describe("readBindingByAddressAnyTenant", () => {
+  test("the address-only resolution event-stream discovery needs, with no tenant to check against", async () => {
+    const anyTenantDb = fakeDb([
+      { ...relaunched, instanceId: "run_a1", currentRunId: "run_a1" },
+    ]);
+    expect(
+      (await readBindingByAddressAnyTenant(anyTenantDb, "run_a1@acme.example"))
+        ?.tenantId,
+    ).toBe("ten_1");
+  });
+});
+
 describe("resolveRoomAddress", () => {
   test("leaves a non-participant address alone", async () => {
     expect(

packages/chat/src/agent-binding.ts

@@ -135,13 +135,48 @@ async function readLaunchRow(
 export async function readBindingByAddress(
   db: DB["db"],
   address: string,
+  tenantId: string,
 ): Promise<AgentBinding | undefined> {
+  const row = await readBindingRowByAddress(db, address);
+  if (row === undefined) return undefined;
+  // A caller that already knows which tenant it is acting for must
+  // never accept a binding minted by another one — `instanceId` and
+  // `currentRunId` are collision-resistant generated ids, not scoped to
+  // a tenant, so nothing else stops an address that happens to name a
+  // row from a different tenant from resolving here. `tenantId` is
+  // required precisely so a caller cannot silently fall back to the
+  // unscoped read by forgetting to pass it — see
+  // `readBindingByAddressAnyTenant` for the one place that read is
+  // actually correct.
+  if (row.row.tenantId !== tenantId) return undefined;
+  return bindingFrom(row.row, row.domain);
+}
+
+/**
+ * The unscoped counterpart of `readBindingByAddress`, for the one caller
+ * that genuinely cannot know the tenant yet: the inbound event-stream
+ * handler discovering which tenant an address even belongs to before it
+ * can act as that tenant. Every other caller already knows its own
+ * tenant and must use `readBindingByAddress` instead.
+ */
+export async function readBindingByAddressAnyTenant(
+  db: DB["db"],
+  address: string,
+): Promise<AgentBinding | undefined> {
+  const row = await readBindingRowByAddress(db, address);
+  return row === undefined ? undefined : bindingFrom(row.row, row.domain);
+}
+
+async function readBindingRowByAddress(
+  db: DB["db"],
+  address: string,
+): Promise<{ row: LaunchRow; domain: string } | undefined> {
   const domain = requireDomain(address);
   const localPart = localPartOf(address);
   const byStableId = await readLaunchRow(db, "instanceId", localPart);
-  if (byStableId !== undefined) return bindingFrom(byStableId, domain);
-  const byRunId = await readLaunchRow(db, "currentRunId", localPart);
-  return byRunId === undefined ? undefined : bindingFrom(byRunId, domain);
+  const row =
+    byStableId ?? (await readLaunchRow(db, "currentRunId", localPart));
+  return row === undefined ? undefined : { row, domain };
 }
 
 /**
@@ -154,7 +189,7 @@ export async function resolveRoomAddress(
   db: DB["db"],
   liveAddress: string,
 ): Promise<string> {
-  const binding = await readBindingByAddress(db, liveAddress);
+  const binding = await readBindingByAddressAnyTenant(db, liveAddress);
   return binding?.roomAddress ?? liveAddress;
 }
 
@@ -183,9 +218,36 @@ export async function resolveLiveAgent(
 export async function resolveLiveByStableId(
   db: DB["db"],
   stableId: string,
+  tenantId: string,
 ): Promise<LiveAgent | undefined> {
   const row = await readLaunchRow(db, "instanceId", stableId);
   if (row === undefined) return undefined;
+  // See the matching note on `readBindingByAddress`: a caller that
+  // knows its own tenant must never be handed another tenant's launch
+  // for a stable id it happens to guess or receive by mistake. Required,
+  // not optional — see `resolveLiveByStableIdAnyTenant` for the unscoped
+  // read.
+  if (row.tenantId !== tenantId) return undefined;
+  return resolveLiveRunForRow(db, row);
+}
+
+/**
+ * The unscoped counterpart of `resolveLiveByStableId`, for a caller that
+ * has no tenant of its own to check against (e.g. it is resolving a
+ * stable id it does not yet know the tenant for).
+ */
+export async function resolveLiveByStableIdAnyTenant(
+  db: DB["db"],
+  stableId: string,
+): Promise<LiveAgent | undefined> {
+  const row = await readLaunchRow(db, "instanceId", stableId);
+  return row === undefined ? undefined : resolveLiveRunForRow(db, row);
+}
+
+async function resolveLiveRunForRow(
+  db: DB["db"],
+  row: LaunchRow,
+): Promise<LiveAgent | undefined> {
   const run = await readRun(db, row.currentRunId);
   if (run === undefined || run.address === null) return undefined;
   return { binding: bindingFrom(row, requireDomain(run.address)), run };

packages/chat/src/chat-orchestrator.ts

@@ -57,7 +57,10 @@ import { consumerFacingInferenceText } from "./consumer-inference-text";
 import type { ConnectedProviderLister } from "./inference-preferences";
 import { mentionedParticipants } from "./mentions";
 import { domainOf, localPartOf } from "./agent-address";
-import { readBindingByAddress, resolveLiveAgent } from "./agent-binding";
+import {
+  readBindingByAddressAnyTenant,
+  resolveLiveAgent,
+} from "./agent-binding";
 import { parseParticipants, type ParticipantRecord } from "./participants";
 import type { Part, TextPart } from "./parts";
 import type { ChatPlatform } from "./platform-port";
@@ -448,7 +451,10 @@ async function resolveMemberWorkbenches(
   // event's address against `workflow_run.address` directly: after a
   // relaunch the two differ, and only the mapping knows that the run
   // announcing itself under a fresh address is the same room teammate.
-  const binding = await readBindingByAddress(deps.db, agentAddress);
+  // The tenant an inbound event's address belongs to is exactly what
+  // this discovery is trying to find out, so it cannot pass one yet —
+  // see `readBindingByAddressAnyTenant`.
+  const binding = await readBindingByAddressAnyTenant(deps.db, agentAddress);
   if (binding === undefined) {
     // Not every agent address on the event stream belongs to a chat
     // workbench (an echo instance, say) — an address this package's own

packages/chat/src/platform-adapter.ts

@@ -33,11 +33,13 @@ import {
   listLaunchesBeyondWake,
   listLaunchesForTenant,
   readBindingByAddress,
+  readBindingByAddressAnyTenant,
   readPriorRuns,
   recordSourcesDigest,
   repointBinding,
   resolveLiveAgent,
   resolveLiveByStableId,
+  resolveLiveByStableIdAnyTenant,
   type AgentBinding,
   type LiveAgent,
 } from "./agent-binding";
@@ -258,7 +260,15 @@ export function createHubChatPlatform(
   };
 
   function offeringDigest(sourceOfferingIds: readonly string[]): string {
-    return sourceOfferingIds.join("\0");
+    // `sources_digest` is a `text` column and this digest rides straight
+    // into it (`platform-adapter.ts`'s `insert(workbenchLaunch)`). A
+    // `\0`-joined id list is not valid UTF-8 text to Postgres
+    // (`invalid byte sequence for encoding "UTF8": 0x00`) the moment a
+    // tenant's catalog carries more than one visible offering, which
+    // permanently failed every agent launch for such a tenant. `\n` never
+    // appears inside a generated offering id, so this stays a lossless,
+    // order-sensitive join — just one Postgres can actually store.
+    return sourceOfferingIds.join("\n");
   }
 
   async function catalogOfferings(tenantId: string): Promise<{
@@ -695,7 +705,13 @@ export function createHubChatPlatform(
    * that treatment, just triggered by content drift instead of death.
    */
   async function wakeByAddress(address: string): Promise<void> {
-    const binding = await readBindingByAddress(deps.db, address);
+    // Every caller of `wakeByAddress` already reached `address` through
+    // a tenant-scoped resolution of its own (`sendMail`'s
+    // `requireLive(workbenchId, tenantId)`, `ensureAwake`'s caller-held
+    // address) — re-deriving the binding here is bookkeeping for a wake
+    // already authorized upstream, not a new trust boundary crossing, so
+    // the unscoped read is correct.
+    const binding = await readBindingByAddressAnyTenant(deps.db, address);
     if (binding === undefined) {
       throw new Error(
         `No workbench_launch binding for address "${address}"; instances ` +
@@ -801,7 +817,9 @@ export function createHubChatPlatform(
    * check that used to never run for it.
    */
   async function reconcileDriftedRun(address: string): Promise<boolean> {
-    const binding = await readBindingByAddress(deps.db, address);
+    // See the matching note in `wakeByAddress`: `address` is always
+    // reached through a tenant-scoped resolution upstream.
+    const binding = await readBindingByAddressAnyTenant(deps.db, address);
     if (binding === undefined) return false;
     const live = await resolveLiveAgent(deps.db, binding);
     if (live === undefined || live.run.address === null) return false;
@@ -913,8 +931,14 @@ export function createHubChatPlatform(
    * participant id — not the room's own address, once anything has been
    * relaunched.
    */
-  async function requireLive(stableId: string): Promise<LiveAgent> {
-    const live = await resolveLiveByStableId(deps.db, stableId);
+  async function requireLive(
+    stableId: string,
+    expectedTenantId?: string,
+  ): Promise<LiveAgent> {
+    const live =
+      expectedTenantId === undefined
+        ? await resolveLiveByStableIdAnyTenant(deps.db, stableId)
+        : await resolveLiveByStableId(deps.db, stableId, expectedTenantId);
     if (live === undefined) {
       throw new Error(`No live workbench run for "${stableId}"`);
     }
@@ -1151,7 +1175,9 @@ export function createHubChatPlatform(
     },
 
     async resolveDefinitionIdByAddress(address): Promise<string | undefined> {
-      const binding = await readBindingByAddress(deps.db, address);
+      // No tenant is threaded through this platform method's own
+      // signature; unscoped by necessity, not by omission.
+      const binding = await readBindingByAddressAnyTenant(deps.db, address);
       if (binding === undefined) return undefined;
       const live = await resolveLiveAgent(deps.db, binding);
       return live?.run.definitionId ?? undefined;
@@ -1176,7 +1202,7 @@ export function createHubChatPlatform(
       _workbenchId,
       address,
     ): Promise<void> {
-      const binding = await readBindingByAddress(deps.db, address);
+      const binding = await readBindingByAddress(deps.db, address, tenantId);
       if (binding === undefined) return;
       const live = await resolveLiveAgent(deps.db, binding);
       const definitionId = live?.run.definitionId;
@@ -1216,7 +1242,7 @@ export function createHubChatPlatform(
       // The stable id names the room's participant; the run it resolves
       // to is whichever one is alive right now, which is a different
       // run (and a different address) after every relaunch.
-      const { binding } = await requireLive(input.workbenchId);
+      const { binding } = await requireLive(input.workbenchId, input.tenantId);
       const liveAddress = binding.liveAddress;
 
       // Wake before send: a sleeping instance (the lifecycle package's
@@ -1270,7 +1296,7 @@ export function createHubChatPlatform(
       } catch (error) {
         throw wrapWakeInferenceError(error);
       }
-      const delivery = await requireLive(input.workbenchId);
+      const delivery = await requireLive(input.workbenchId, input.tenantId);
       const deliveryAddress = delivery.binding.liveAddress;
       // Tracking here (not only at launch) brings instances that were
       // already resident before this hub process started — restored by
@@ -1394,7 +1420,7 @@ export function createHubChatPlatform(
       let cancelled = false;
       let unsubscribeAgent: (() => void) | undefined;
 
-      void resolveLiveByStableId(deps.db, workbenchId)
+      void resolveLiveByStableIdAnyTenant(deps.db, workbenchId)
         .then((live) => {
           if (cancelled || live === undefined) return;
           unsubscribeAgent = deps.sidecarRouter.subscribeAgent(
@@ -1422,7 +1448,7 @@ export function createHubChatPlatform(
       // undelivered-mail handler holds whatever the envelope named), so
       // the lifecycle is driven on the LIVE address it resolves to —
       // that is the only address the sidecar ever announces.
-      const binding = await readBindingByAddress(deps.db, address);
+      const binding = await readBindingByAddressAnyTenant(deps.db, address);
       if (binding === undefined) {
         throw new Error(`No workbench_launch binding for address "${address}"`);
       }