diff --git a/packages/chat/src/agent-binding.test.ts b/packages/chat/src/agent-binding.test.ts index ed89cec9db..f11d61a60f 100644 --- a/packages/chat/src/agent-binding.test.ts +++ b/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( diff --git a/packages/chat/src/agent-binding.ts b/packages/chat/src/agent-binding.ts index d53399366c..47b4d0d714 100644 --- a/packages/chat/src/agent-binding.ts +++ b/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 { + 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 { + 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 { - 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 { 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 { + const row = await readLaunchRow(db, "instanceId", stableId); + return row === undefined ? undefined : resolveLiveRunForRow(db, row); +} + +async function resolveLiveRunForRow( + db: DB["db"], + row: LaunchRow, +): Promise { const run = await readRun(db, row.currentRunId); if (run === undefined || run.address === null) return undefined; return { binding: bindingFrom(row, requireDomain(run.address)), run }; diff --git a/packages/chat/src/chat-orchestrator.ts b/packages/chat/src/chat-orchestrator.ts index 808c755ba2..73ce91b0cb 100644 --- a/packages/chat/src/chat-orchestrator.ts +++ b/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 diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index 6490af3c7e..39e52ff2c8 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/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 { - 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 { - 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 { - const live = await resolveLiveByStableId(deps.db, stableId); + async function requireLive( + stableId: string, + expectedTenantId?: string, + ): Promise { + 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 { - 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 { - 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}"`); }