diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index fbf44a094d..b864db59be 100644 --- a/apps/hub/src/index.ts +++ b/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, diff --git a/packages/chat/src/index.ts b/packages/chat/src/index.ts index cc3770f380..ac5f1935a5 100644 --- a/packages/chat/src/index.ts +++ b/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"; diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index 5e8f971812..5809b292c8 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/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; + /** + * 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; /** * 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 { + 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 { + 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>> { 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), }; }, diff --git a/packages/chat/src/run-trigger-client.ts b/packages/chat/src/run-trigger-client.ts new file mode 100644 index 0000000000..7d22d29510 --- /dev/null +++ b/packages/chat/src/run-trigger-client.ts @@ -0,0 +1,177 @@ +// A thin client over Interchange's own workflow-run mail-trigger route +// (`POST /api/tenants/:tenantId/workflows/runs/:runId/mail`, mounted by +// `vendor/intx/hub-api`'s `createRunRoutes`). CL-7490: a provisioned +// run's first `agent.deploy` only fires from inside that route's +// durable-dispatch branch (`createWorkflowRunTrigger` in +// `vendor/intx/hub-api/src/workflow-run-trigger.ts`) — a raw +// `sessionService.sendUserMessage` never reaches it, so an invited run +// that has never been triggered natively stays undeployed and +// unreachable forever. `createWorkflowRunTrigger` itself is not +// exported from `@intx/hub-api` (it is the route's own private +// dependency, not a published seam), so this calls the mounted HTTP +// route instead of reimplementing what it does. +// +// The call is in-process against the hub's own composed Hono `app` — +// no network hop, no separate origin to configure — authenticated the +// way any other request to that route is: through `@intx/hub-api`'s +// session contract. Chat has no browser cookie to present (most sends +// happen with no inbound HTTP request in scope at all — mention +// fan-out, a relaunch resend, the mailbox-fanout persist seam), so it +// presents a short-lived signed token instead (see +// `./run-trigger-internal-auth.ts`) that `apps/hub`'s own `getSession` +// implementation verifies before falling through to the real +// cookie-session path. Neither side touches `vendor/intx`. +import { type } from "arktype"; + +import type { MailContent } from "./codec"; +import { signInternalRunTriggerToken } from "./run-trigger-internal-auth"; + +// Mirrors `vendor/intx/hub-api/src/workflow-run-trigger.ts`'s own +// `WorkflowRunTriggerResponse` — not imported directly since +// `@intx/hub-api` does not re-export it from its package entry point. +// Parsed here rather than trusted as `as` cast: this is a trust +// boundary (an HTTP response), even though the call never leaves the +// process. +const TriggerResponse = type({ + runId: "string", + address: "string", + messageId: "string", +}); + +const TriggerErrorBody = type({ + error: { + code: "string", + message: "string", + }, +}); + +export class RunTriggerError extends Error { + constructor( + readonly status: number, + readonly code: string, + message: string, + ) { + super(`workflow-run trigger failed (${status} ${code}): ${message}`); + this.name = "RunTriggerError"; + } +} + +export type RunTriggerClientDeps = { + /** + * The hub's own composed Hono app (`@intx/hub-api`'s `App`, built by + * `createApp`). Calling `.request()` on it dispatches straight + * through the route's own middleware chain in-process — the same + * path a real inbound request to the hub takes. + */ + app: { + request: ( + input: string, + init?: RequestInit, + ) => Response | Promise; + }; + /** + * The shared secret `./run-trigger-internal-auth.ts` signs the + * per-call token with. Must be the same value `apps/hub` wires into + * its `getSession` verifier — the host's composition root, never a + * package default, owns that pairing. + */ + internalAuthSecret: string; + /** + * Mints the token asserting which better-auth `user.id` this trigger + * call authenticates as. Separated from `internalAuthSecret` only so + * a test can supply a fixed clock; production always uses + * `signInternalRunTriggerToken` from `./run-trigger-internal-auth.ts`. + */ + signToken?: (secret: string, userId: string) => string; +}; + +export type TriggerWorkflowRunMailInput = { + readonly tenantId: string; + /** The deployment's anchor run id (`:runId` in the route path). */ + readonly anchorRunId: string; + readonly content: string; + readonly attachments?: MailContent["attachments"]; + /** + * The better-auth `user.id` this call authenticates as. The route + * derives the delivered mail's `From` header from whatever principal + * this resolves to in the target tenant, so this should be the real + * sending human when one exists (`sendMail`'s own `principalId`), + * and the workflow definition asset's `creatorPrincipalId` (resolved + * to its owning user) otherwise — see the callers in + * `./platform-adapter.ts`. + */ + readonly authAsUserId: string; +}; + +export type TriggeredWorkflowRunMail = { + readonly runId: string; + readonly address: string; + readonly messageId: string; +}; + +export type RunTriggerClient = { + triggerMail( + input: TriggerWorkflowRunMailInput, + ): Promise; +}; + +async function readJson(response: Response): Promise { + try { + return await response.json(); + } catch { + // report-error-ignore: a response body that fails to parse as JSON + // is handled by the caller's own arktype parse — `TriggerErrorBody` + // or `TriggerResponse` rejecting `null` — which throws a real, + // reported error naming the malformed response; this local catch + // only stops a parse failure itself from masking that path. + return null; + } +} + +export function createRunTriggerClient( + deps: RunTriggerClientDeps, +): RunTriggerClient { + return { + async triggerMail(input): Promise { + const signToken = deps.signToken ?? signInternalRunTriggerToken; + const token = signToken(deps.internalAuthSecret, input.authAsUserId); + const body: { + content: string; + attachments?: MailContent["attachments"]; + } = + input.attachments !== undefined + ? { content: input.content, attachments: input.attachments } + : { content: input.content }; + + const response = await deps.app.request( + `/api/tenants/${input.tenantId}/workflows/runs/${input.anchorRunId}/mail`, + { + method: "POST", + headers: { + "content-type": "application/json", + "x-corbits-internal-run-trigger": token, + }, + body: JSON.stringify(body), + }, + ); + + const json = await readJson(response); + if (response.status !== 202) { + const parsedError = TriggerErrorBody(json); + const { code, message } = + parsedError instanceof type.errors + ? { code: "unknown", message: response.statusText } + : parsedError.error; + throw new RunTriggerError(response.status, code, message); + } + + const parsed = TriggerResponse(json); + if (parsed instanceof type.errors) { + throw new Error( + `malformed workflow-run trigger response: ${parsed.summary}`, + ); + } + return parsed; + }, + }; +} diff --git a/packages/chat/src/run-trigger-internal-auth.ts b/packages/chat/src/run-trigger-internal-auth.ts new file mode 100644 index 0000000000..b48242b524 --- /dev/null +++ b/packages/chat/src/run-trigger-internal-auth.ts @@ -0,0 +1,73 @@ +// A server-side identity credential for chat's own calls into the hub's +// `/api/tenants/:tenantId/workflows/runs/:runId/mail` route (see +// `./run-trigger-client.ts`). That route authenticates a caller through +// `@intx/hub-api`'s ordinary session-cookie path — the right answer for a +// human's browser, but chat's own dispatch (mention fan-out, relaunch +// resends, the mailbox-fanout persist seam) runs with no browser request +// in scope at all and cannot present one. +// +// `@intx/hub-api`'s `GetSession` contract is deliberately pluggable (see +// its own doc comment: "so a third-party identity provider can be +// plugged in"), and `apps/hub` is the one composing it (`getSession` is +// supplied there, not fixed by the vendored package). This module is the +// shared signer/verifier for a short-lived, HMAC-signed token naming the +// principal to authenticate as: chat signs one per trigger call +// (`./run-trigger-client.ts`), and `apps/hub`'s `getSession` verifies it +// before falling through to the real cookie-session path. Neither side +// touches `vendor/intx` — this is host-owned wiring through an +// already-open extension seam, not a fork of the session contract. +// +// The signed identity is a better-auth `user.id` (what `resolveTenant` +// looks up a tenant `principal` by by `refId`), never a raw +// `chat.workbench_launch`/`workflow_run` id — those name a run or a +// room, not an authenticated user. +import { createHmac, timingSafeEqual } from "node:crypto"; + +const TOKEN_TTL_MS = 30_000; + +function sign(secret: string, payload: string): string { + return createHmac("sha256", secret).update(payload).digest("base64url"); +} + +/** + * Mint a token asserting `userId` is the caller, valid for + * `TOKEN_TTL_MS` from `now`. Bound to a real wall-clock deadline rather + * than being a durable credential — this authenticates one trigger call, + * not a session a caller could hold onto. + */ +export function signInternalRunTriggerToken( + secret: string, + userId: string, + now: number = Date.now(), +): string { + const expiresAt = now + TOKEN_TTL_MS; + const payload = `${userId}.${expiresAt}`; + return `${payload}.${sign(secret, payload)}`; +} + +/** + * Verify a token minted by `signInternalRunTriggerToken`, returning the + * asserted `userId` when the signature checks out and the token has not + * expired, `null` otherwise. Constant-time signature comparison so a + * caller cannot learn anything about the expected signature from timing. + */ +export function verifyInternalRunTriggerToken( + secret: string, + token: string, + now: number = Date.now(), +): string | null { + const parts = token.split("."); + if (parts.length !== 3) return null; + const [userId, expiresAtRaw, signature] = parts; + if (userId === undefined || userId === "") return null; + if (expiresAtRaw === undefined || signature === undefined) return null; + const expiresAt = Number(expiresAtRaw); + if (!Number.isFinite(expiresAt) || expiresAt < now) return null; + + const expected = sign(secret, `${userId}.${expiresAtRaw}`); + const presented = Buffer.from(signature); + const wanted = Buffer.from(expected); + if (presented.length !== wanted.length) return null; + if (!timingSafeEqual(presented, wanted)) return null; + return userId; +} diff --git a/packages/chat/src/send-run-mail.ts b/packages/chat/src/send-run-mail.ts index 139dfd98a4..86412bd9d1 100644 --- a/packages/chat/src/send-run-mail.ts +++ b/packages/chat/src/send-run-mail.ts @@ -1,13 +1,36 @@ -// Signs one inbound message via Interchange `sendUserMessage` and -// records it on session_mail so the run's own live subscribers see it. +// Delivers one inbound message to a provisioned run through Interchange's +// own workflow-run mail-trigger route (`./run-trigger-client.ts`), then +// records a local, chat-signed copy on `session_mail` so this workbench's +// own subscribers (the `mail.delivered` event and `fetchBlob`'s outbound +// attachment reads) see it exactly as before (CL-7490). +// +// CL-7490: `sessionService.sendUserMessage` used to both sign AND +// deliver. It never reached the trigger route's durable-dispatch branch, +// so a provisioned run invited but never yet triggered natively never +// got its first `agent.deploy` and stayed permanently unreachable. The +// trigger route now owns delivery (and, on its first call, the deploy); +// it returns no raw MIME bytes, so this still assembles and signs its +// own local copy — with `@intx/mime`/`@intx/crypto`, the same primitives +// the route itself uses — purely for that local record. It is never +// delivered a second time: the sidecar send lives entirely on the +// trigger route's side of this call. +import { + assembleMessage, + assembleSignedContent, + createDetachedSignatureFromProvider, + type MessageHeaders, +} from "@intx/mime"; import { sessionMail } from "@intx/db/schema"; import type { DB } from "@intx/db"; -import type { SessionService, SidecarRouter } from "@intx/hub-sessions"; +import type { SidecarRouter } from "@intx/hub-sessions"; import type { CryptoProvider, MessageAttachment } from "@intx/types/runtime"; +import { mailIdFromBracketMessageId } from "./turn-mail-correlation"; +import type { RunTriggerClient } from "./run-trigger-client"; + export type RunMailDeps = { db: DB["db"]; - sessionService: Pick; + runTrigger: RunTriggerClient; sidecarRouter: Pick; }; @@ -15,6 +38,9 @@ export type SendRunMailParams = { tenantId: string; sessionId: string; agentAddress: string; + /** The deployment's anchor run id — `:runId` in the trigger route's + * path, distinct from `agentAddress` (which names the deployment). */ + anchorRunId: string; from: string; domain: string; content: string; @@ -24,6 +50,12 @@ export type SendRunMailParams = { inReplyTo?: string; references?: readonly string[]; cryptoProvider: CryptoProvider; + /** + * The better-auth `user.id` this delivery authenticates the trigger + * call as — see `./run-trigger-client.ts`'s own doc comment on + * `authAsUserId` for how a caller resolves this. + */ + authAsUserId: string; }; export type SentRunMail = { @@ -31,36 +63,65 @@ export type SentRunMail = { readonly createdAt: string; }; -async function deliverRunMailMIME( - deps: Pick, +function toWireAttachments( + attachments: MessageAttachment[] | undefined, +): { mimeType: string; data: string; name?: string }[] | undefined { + if (attachments === undefined || attachments.length === 0) return undefined; + return attachments.map((attachment) => ({ + mimeType: attachment.contentType, + data: Buffer.from(attachment.data).toString("base64"), + ...(attachment.name !== undefined ? { name: attachment.name } : {}), + })); +} + +/** + * Assemble and sign a local MIME copy of the message this delivery sent, + * for `session_mail` — never delivered anywhere itself. Threading + * headers (`inReplyTo`/`references`) are preserved here even though the + * trigger route's own `SendMessage` body carries neither (CL-7490: "a + * trigger occurrence is threading-less at the mail boundary" — + * `vendor/intx/hub-api/src/workflow-run-trigger.ts`), so the run itself + * never sees this message as a reply; only this workbench's own local + * history does. + */ +async function signLocalRunMailCopy( params: SendRunMailParams, - mailId: string, + messageId: string, now: Date, ): Promise { - const userMessageParams = { - agentAddress: params.agentAddress, + const headers: MessageHeaders = { from: params.from, - messageId: params.messageId ?? `<${mailId}@${params.domain}>`, + to: [params.agentAddress], + cc: undefined, date: now, - content: params.content, - sessionId: params.sessionId, - tenantId: params.tenantId, - cryptoProvider: params.cryptoProvider, + messageId, + subject: undefined, + inReplyTo: params.inReplyTo ?? params.replyTo, + references: + params.references !== undefined ? [...params.references] : undefined, + mimeVersion: "1.0", + interchangeType: "conversation.message", + interchangeCorrelationId: undefined, + interchangeTenantId: params.tenantId, + interchangeAgentId: undefined, + interchangeSessionId: params.sessionId, + interchangeOfferingId: undefined, + interchangeSchemaVersion: undefined, + traceparent: undefined, + tracestate: undefined, }; - const withAttachments = - params.attachments !== undefined - ? { ...userMessageParams, attachments: params.attachments } - : userMessageParams; - const inReplyTo = params.inReplyTo ?? params.replyTo; - const withReplyTo = - inReplyTo !== undefined - ? { ...withAttachments, inReplyTo } - : withAttachments; - const withReferences = - params.references !== undefined && params.references.length > 0 - ? { ...withReplyTo, references: [...params.references] } - : withReplyTo; - return deps.sessionService.sendUserMessage(withReferences); + const signedContent = assembleSignedContent({ + kind: "conversation", + text: params.content, + ...(params.attachments !== undefined + ? { attachments: params.attachments } + : {}), + }); + const signature = await createDetachedSignatureFromProvider( + signedContent, + params.cryptoProvider, + ); + return assembleMessage(headers, signedContent, signature); } async function recordRunMail( @@ -97,9 +158,16 @@ export async function sendRunMail( deps: RunMailDeps, params: SendRunMailParams, ): Promise { - const mailId = crypto.randomUUID(); const now = new Date(); - const rawMIME = await deliverRunMailMIME(deps, params, mailId, now); + const triggered = await deps.runTrigger.triggerMail({ + tenantId: params.tenantId, + anchorRunId: params.anchorRunId, + content: params.content, + attachments: toWireAttachments(params.attachments), + authAsUserId: params.authAsUserId, + }); + const mailId = mailIdFromBracketMessageId(triggered.messageId); + const rawMIME = await signLocalRunMailCopy(params, triggered.messageId, now); // A run's session and event collector are ensured lazily now, at the // hub seams that actually see the run become mail-routable // (`apps/hub/src/mailbox-persist.ts`, the wrapped `eventCollectors` diff --git a/packages/chat/test/platform-adapter.test.ts b/packages/chat/test/platform-adapter.test.ts index 47ff59d594..73d4189376 100644 --- a/packages/chat/test/platform-adapter.test.ts +++ b/packages/chat/test/platform-adapter.test.ts @@ -15,7 +15,7 @@ // `@intx/hub-api`'s own contract, not this package's, and is not // re-proven here. // -// `sessionService`/`repoStore`/`sidecarRouter` are fakes recording +// `runTrigger`/`repoStore`/`sidecarRouter` are fakes recording // their calls, and `db` is a minimal chainable stand-in for the // drizzle query builder (no database involved) so the mapping is // exercised without a real Postgres. @@ -30,6 +30,7 @@ import type { DefinitionSourceResolution } from "@intx/hub-api"; import { agentSession, asset, + principal as principalTable, sessionMail, workflowDefinition, workflowDefinitionVersion, @@ -40,6 +41,10 @@ import type { EventCollectorRegistry, SidecarRouter } from "@intx/hub-sessions"; import { workbenchLaunch } from "../src/schema"; import type { CreateHubChatPlatformDeps } from "../src/platform-adapter"; import { MODEL_UNAVAILABLE_CONSUMER_MESSAGE } from "../src/model-unavailable"; +import type { + RunTriggerClient, + TriggerWorkflowRunMailInput, +} from "../src/run-trigger-client"; import { createCryptoProviderCache, type CryptoProviderCache, @@ -535,6 +540,23 @@ function createFakeDb(opts: { }; } if (table === asset) return selectChain([opts.assetRow]); + // The run-trigger client's auth resolution + // (`resolveTriggerAuthUserId` in `../src/platform-adapter.ts`) + // looks up the better-auth user behind a principal id. No + // test here cares which synthetic user a trigger call + // authenticates as, only that it succeeds, so this answers + // every lookup with the same fixed "user" principal. + if (table === principalTable) { + return selectChain([{ refId: "user_test", kind: "user" }]); + } + if (table === workflowDefinition) { + return selectChain([ + { + assetId: + opts.workflowDefinitionRow?.assetId ?? "ast_definition1", + }, + ]); + } if (table === workflowDefinitionVersion) { // `loadFrozenWireProjection` filters on // `and(eq(definitionId, id), eq(version, "1"))`; the first @@ -670,58 +692,35 @@ function createFakeEventCollectors( }; } -type AdoptedDeployCall = { - anchorRunId: string; - agentAddress: string; -}; - -type FakeSessionService = { - adoptedDeployCalls: unknown[]; - sendUserMessageCalls: unknown[]; - sendUserMessage: (params: unknown) => Promise; - stageWorkflowStep: () => Promise; - deployInstanceAtHead: () => Promise; - deployAdoptedWorkflowFromSource: (params: AdoptedDeployCall) => Promise<{ - anchorRunId: string; - deploymentAddress: string; - publicKey: string; - }>; - deployWorkflowDefinition: () => Promise; - endSession: () => Promise; +type FakeRunTrigger = RunTriggerClient & { + triggerCalls: TriggerWorkflowRunMailInput[]; }; -function createFakeSessionService(): FakeSessionService { - const adoptedDeployCalls: unknown[] = []; - const sendUserMessageCalls: unknown[] = []; - return { - adoptedDeployCalls, - sendUserMessageCalls, - async stageWorkflowStep() {}, - async deployInstanceAtHead() { - throw new Error( - "deployInstanceAtHead must not be called: invite, wake, and relaunch provision via Interchange prepareProvisionedDeployment", - ); - }, - async deployAdoptedWorkflowFromSource(params: AdoptedDeployCall) { - adoptedDeployCalls.push(params); +/** + * Fakes `@corbits/chat`'s `RunTriggerClient` (CL-7490): the seam + * `sendRunMail` now delivers every provisioned run's mail through, + * replacing the old raw `runTrigger.triggerMail`. Each call + * mints a fresh `messageId` (``) the same shape + * `vendor/intx/hub-api/src/workflow-run-trigger.ts`'s real route mints, + * so `mailIdFromBracketMessageId` and reply-threading assertions behave + * exactly as they would against the real route. + */ +function createFakeRunTrigger(): FakeRunTrigger { + const triggerCalls: TriggerWorkflowRunMailInput[] = []; + let counter = 0; + const fake: FakeRunTrigger = { + triggerCalls, + async triggerMail(input: TriggerWorkflowRunMailInput) { + triggerCalls.push(input); + counter += 1; return { - anchorRunId: params.anchorRunId, - deploymentAddress: params.agentAddress, - publicKey: "test-public-key", + runId: input.anchorRunId, + address: `${input.anchorRunId}@ten1.workbench.test`, + messageId: ``, }; }, - async deployWorkflowDefinition() { - throw new Error( - "deployWorkflowDefinition must not be called: launchWorkbench " + - "provisions via Interchange prepareProvisionedDeployment", - ); - }, - async sendUserMessage(params: unknown) { - sendUserMessageCalls.push(params); - return new TextEncoder().encode("raw-mime-bytes"); - }, - async endSession() {}, }; + return fake; } function createFakeSidecarRouter( @@ -782,7 +781,7 @@ function createFakeSidecarRouter( }, // Every launch and wake produces the run's `run.grants` frame before its // first mail. Always routable: the frame is sent after the deploy the - // fake `sessionService` just acked, and that deploy is what makes the + // fake `runTrigger` just acked, and that deploy is what makes the // address resident — `routableAddresses` models residency BEFORE the // wake (what `getRoutableAddresses` answers), not after it. sendRunGrants(address: string, runId: string, stepGrants: unknown) { @@ -896,7 +895,7 @@ describe("createHubChatPlatform", () => { table: agentSession, values: { id: "ses_run1", principalId: "prin_run1" }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const deployError = new Error("sidecar unreachable"); const allocationService = createFakeWorkflowAllocationService(); allocationService.prepareProvisionedDeployment = async () => { @@ -908,7 +907,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter, eventCollectors, workflowAllocationService: allocationService, @@ -976,7 +975,7 @@ describe("createHubChatPlatform", () => { table: agentSession, values: { id: "ses_run1", principalId: "prin_run1" }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const allocationService = createFakeWorkflowAllocationService(); allocationService.prepareProvisionedDeployment = async () => { throw new SessionLaunchError("start", new Error("ack timeout"), true); @@ -987,7 +986,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter, eventCollectors, workflowAllocationService: allocationService, @@ -1010,7 +1009,7 @@ describe("createHubChatPlatform", () => { // its replies are real, so it still fails loud without a catalog // source; proven alongside `launchInvite`'s other tests below. - test("sendMail resolves the workbench's run's session via the shared principal and delivers via sessionService", async () => { + test("sendMail resolves the workbench's run's session via the shared principal and delivers via runTrigger", async () => { resolveDefinitionSourcesResult = { ok: true, materials: [], @@ -1047,13 +1046,13 @@ describe("createHubChatPlatform", () => { values: { id: "ses_run1", principalId: "prin_run1" }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const sidecarRouter = createFakeSidecarRouter(); const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter, eventCollectors: createFakeEventCollectors(), }); @@ -1066,15 +1065,10 @@ describe("createHubChatPlatform", () => { }); expect(sent.id).toBeTruthy(); - expect(sessionService.sendUserMessageCalls).toHaveLength(1); - const call = sessionService.sendUserMessageCalls[0] as { - agentAddress: string; - content: string; - sessionId: string; - }; - expect(call.agentAddress).toBe("ins_workbench1@ten1.workbench.test"); - expect(call.content).toBe("hello workbench"); - expect(call.sessionId).toBe("ses_run1"); + expect(runTrigger.triggerCalls).toHaveLength(1); + const call = runTrigger.triggerCalls[0]; + expect(call?.anchorRunId).toBe("ins_workbench1"); + expect(call?.content).toBe("hello workbench"); const mailInsert = db.inserted.find((row) => row.table === sessionMail); expect(mailInsert?.values).toMatchObject({ @@ -1130,13 +1124,13 @@ describe("createHubChatPlatform", () => { }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const sidecarRouter = createFakeSidecarRouter(); const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter, eventCollectors: createFakeEventCollectors(), }); @@ -1149,10 +1143,8 @@ describe("createHubChatPlatform", () => { }); expect(sent.id).toBeTruthy(); - const call = sessionService.sendUserMessageCalls[0] as { - sessionId: string; - }; - expect(call.sessionId).toBe("ses_unanchored"); + const mailInsert = db.inserted.find((row) => row.table === sessionMail); + expect(mailInsert?.values).toMatchObject({ sessionId: "ses_unanchored" }); }); test("sendMail signs with the injected crypto cache keyed by workbenchId", async () => { @@ -1190,10 +1182,14 @@ describe("createHubChatPlatform", () => { values: { id: "ses_run1", principalId: "prin_run1" }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const getKeys: string[] = []; const injectedProvider = { getPublicKey: () => new Uint8Array([7, 2, 8, 4]), + // Exercised for real now (CL-7490): `sendRunMail` signs its own + // local `session_mail` copy directly rather than through a fake + // `sessionService.sendUserMessage`. + sign: async () => new Uint8Array(64), }; const cryptoProviders: CryptoProviderCache = { get: async (key) => { @@ -1205,7 +1201,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter: createFakeSidecarRouter(), eventCollectors: createFakeEventCollectors(), cryptoProviders, @@ -1218,11 +1214,11 @@ describe("createHubChatPlatform", () => { content: { content: "hello workbench" }, }); + // The crypto cache is only consulted for the local `session_mail` + // signing copy now (CL-7490) — delivery itself goes through the + // run-trigger client, which signs its own copy independently. expect(getKeys).toEqual(["ins_workbench1"]); - const call = sessionService.sendUserMessageCalls[0] as { - cryptoProvider: unknown; - }; - expect(call.cryptoProvider).toBe(injectedProvider); + expect(runTrigger.triggerCalls).toHaveLength(1); }); test("sendMail rejects within the mail-delivery deadline instead of hanging forever when delivery never settles (CL-6644)", async () => { @@ -1260,20 +1256,20 @@ describe("createHubChatPlatform", () => { values: { id: "ses_run1", principalId: "prin_run1" }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); // Models the observed CL-6644 symptom: the post-deploy delivery - // step (`sessionService.sendUserMessage`, reached through + // step (`runTrigger.triggerMail`, reached through // `sendFoldedMail`) never resolves and never rejects -- a wedged // ack, not a thrown "agent is unreachable" the reclaim-retry loop // already knows how to handle. Before the fix, `sendMail`'s // returned promise stayed pending forever with nothing logged. - sessionService.sendUserMessage = () => new Promise(() => {}); + runTrigger.triggerMail = () => new Promise(() => {}); const sidecarRouter = createFakeSidecarRouter(); const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter, eventCollectors: createFakeEventCollectors(), mailDeliveryTimeoutMs: 20, @@ -1326,14 +1322,14 @@ describe("createHubChatPlatform", () => { wfd_echo: inertProjection({ id: "wfd_echo" }), }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const sidecarRouter = createFakeSidecarRouter({ routableAddresses: [] }); const eventCollectors = createFakeEventCollectors(); const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter, eventCollectors, }); @@ -1403,7 +1399,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), eventCollectors: createFakeEventCollectors(), }); @@ -1471,7 +1467,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), eventCollectors: createFakeEventCollectors(), }); @@ -1495,7 +1491,7 @@ describe("createHubChatPlatform", () => { createHubChatPlatform({ toolGrantsForPins: () => [], db: {} as never, - sessionService: {} as never, + runTrigger: {} as never, sidecarRouter: {} as never, eventCollectors: createFakeEventCollectors(), credentialCipher: undefined as never, @@ -1508,7 +1504,7 @@ describe("createHubChatPlatform", () => { createHubChatPlatform({ toolGrantsForPins: () => [], db: {} as never, - sessionService: {} as never, + runTrigger: {} as never, sidecarRouter: {} as never, eventCollectors: createFakeEventCollectors(), credentialCipher: {} as never, @@ -1526,7 +1522,7 @@ describe("createHubChatPlatform", () => { const platform = createHubChatPlatform({ toolGrantsForPins: () => [], db: {} as never, - sessionService: {} as never, + runTrigger: {} as never, sidecarRouter: {} as never, eventCollectors: createFakeEventCollectors(), credentialCipher: { @@ -1564,7 +1560,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter(), eventCollectors: createFakeEventCollectors(), }); @@ -1616,7 +1612,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), eventCollectors: createFakeEventCollectors(), }); @@ -1688,7 +1684,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), eventCollectors: createFakeEventCollectors(), }); @@ -1756,7 +1752,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), eventCollectors: createFakeEventCollectors(), }); @@ -1785,7 +1781,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter(), eventCollectors: createFakeEventCollectors(), }); @@ -1829,7 +1825,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), eventCollectors: createFakeEventCollectors(), }); @@ -1878,7 +1874,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), eventCollectors: createFakeEventCollectors(), }); @@ -1920,7 +1916,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), eventCollectors: createFakeEventCollectors(), }); @@ -1971,7 +1967,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter(), eventCollectors: createFakeEventCollectors(), }); @@ -1997,13 +1993,13 @@ describe("createHubChatPlatform", () => { principalId: "prin_run1", }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const sidecarRouter = createFakeSidecarRouter(); const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter, eventCollectors: createFakeEventCollectors(), }); @@ -2097,16 +2093,17 @@ describe("createHubChatPlatform", () => { values: { id: "ses_run1", principalId: "prin_run1" }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); let sendAttempts = 0; - sessionService.sendUserMessage = async (params: unknown) => { - sessionService.sendUserMessageCalls.push(params); + const baseTriggerMail = runTrigger.triggerMail; + runTrigger.triggerMail = async (input) => { sendAttempts += 1; // The first attempt lands while the sidecar is still booting. if (sendAttempts === 1) { + runTrigger.triggerCalls.push(input); throw new Error("agent is unreachable"); } - return new TextEncoder().encode("raw-mime-bytes"); + return baseTriggerMail(input); }; // Not routable yet — the sidecar is still finishing its boot; // it registers a moment later, on its own, with no redeploy. @@ -2119,7 +2116,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter, eventCollectors, lifecycle: { idleSleepMs: 60_000 }, @@ -2136,7 +2133,7 @@ describe("createHubChatPlatform", () => { // No relaunch: the booting run kept its own run id and address — // `wakeByAddress` never called `prepareProvisionedDeployment`. expect(lastAllocationService.prepareCalls).toHaveLength(0); - expect(sessionService.sendUserMessageCalls).toHaveLength(2); + expect(runTrigger.triggerCalls).toHaveLength(2); }); // CL-6267: the sidecar's own park/wake handler now owns respawning @@ -2199,7 +2196,7 @@ describe("createHubChatPlatform", () => { values: { id: "ses_run1", principalId: "prin_run1" }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const sidecarRouter = createFakeSidecarRouter({ routableAddresses: ["ins_workbench1@ten1.workbench.test"], }); @@ -2208,7 +2205,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter, eventCollectors, lifecycle: { idleSleepMs: 60_000 }, @@ -2224,7 +2221,7 @@ describe("createHubChatPlatform", () => { expect(sent.id).toBeTruthy(); expect(lastAllocationService.prepareCalls).toHaveLength(0); expect(sidecarRouter.sendAgentUndeployCalls).toHaveLength(0); - expect(sessionService.sendUserMessageCalls).toHaveLength(1); + expect(runTrigger.triggerCalls).toHaveLength(1); }); test("the idle sweep never undeploys an address the event collector reports as busy", async () => { @@ -2273,7 +2270,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter, eventCollectors, lifecycle: { idleSleepMs: 5, sweepIntervalMs: 5 }, @@ -2337,7 +2334,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter, eventCollectors, lifecycle: { idleSleepMs: 5, sweepIntervalMs: 5 }, @@ -2403,7 +2400,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter, eventCollectors, lifecycle: { idleSleepMs: 5, sweepIntervalMs: 5 }, @@ -2446,7 +2443,7 @@ describe("createHubChatPlatform", () => { createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter(), eventCollectors: createFakeEventCollectors(), }); @@ -2479,7 +2476,7 @@ describe("createHubChatPlatform", () => { createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter(), eventCollectors: createFakeEventCollectors(), lifecycle: { idleSleepMs: 60_000 }, @@ -2521,11 +2518,11 @@ describe("createHubChatPlatform", () => { }, }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter: createFakeSidecarRouter({ routableAddresses: [address], }), @@ -2590,11 +2587,11 @@ describe("createHubChatPlatform", () => { values: { id: "ses_run1", principalId: "prin_run1" }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), eventCollectors: createFakeEventCollectors(), lifecycle: { idleSleepMs: 60_000 }, @@ -2652,11 +2649,11 @@ describe("createHubChatPlatform", () => { values: { id: "ses_run1", principalId: "prin_run1" }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), eventCollectors: createFakeEventCollectors(), }); @@ -2679,7 +2676,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), eventCollectors: createFakeEventCollectors(), }); @@ -2777,166 +2774,17 @@ describe("createHubChatPlatform", () => { values: { id: "ses_run1", principalId: "prin_room1" }, }); - const sessionService = createFakeSessionService(); - sessionService.sendUserMessage = async (params: unknown) => { - sessionService.sendUserMessageCalls.push(params); - const { agentAddress } = params as { agentAddress: string }; - // The dead run never becomes reachable again — only a send to - // whatever address the relaunch actually produced can succeed. - if (agentAddress === staleAddress) { - throw new Error("agent is unreachable"); - } - return new TextEncoder().encode("raw-mime-bytes"); - }; - - // Routable so `sendMail`'s own opening wake-before-send gate skips - // waking it — the relaunch this test cares about must happen - // inside the reclaim-retry loop below, not before it. - const sidecarRouter = createFakeSidecarRouter({ - routableAddresses: [staleAddress], - }); - - // Models the sidecar registering the relaunch's fresh address the - // moment its deploy actually completes — `sendRunMailWithReclaimRetry` - // now waits on this routing-table transition (CL-7488) instead of a - // fixed backoff, so the retried send needs it to flip before it can - // see the run through. - const allocationService = createFakeWorkflowAllocationService(); - const originalPrepare = - allocationService.prepareProvisionedDeployment.bind(allocationService); - allocationService.prepareProvisionedDeployment = async ( - params: Parameters[0], - ) => { - const result = await originalPrepare(params); - sidecarRouter.routableAddresses.push(result.deploymentAddress); - return result; - }; - - const platform = createPlatform({ - toolGrantsForPins: () => [], - db: db as never, - sessionService, - sidecarRouter, - eventCollectors: createFakeEventCollectors(), - workflowAllocationService: allocationService, - routableWaitDeadlineMs: 5_000, - mailDeliveryTimeoutMs: 5_000, - }); - - const sent = await platform.sendMail({ - tenantId: "ten_1", - workbenchId: "ins_room1", - principalId: "prin_sender", - content: { content: "hello" }, - }); - - expect(sent.id).toBeTruthy(); - expect(sessionService.sendUserMessageCalls).toHaveLength(2); - const [first, second] = sessionService.sendUserMessageCalls as { - agentAddress: string; - }[]; - expect(first?.agentAddress).toBe(staleAddress); - // The retry that follows the relaunch targets the run that - // replaced `run_gen1`, never the dead address the loop started - // with. - expect(second?.agentAddress).not.toBe(staleAddress); - - const repointed = db.updated.at(-1)?.values as { - currentRunId: string; - priorRunIds: string[]; - }; - const secondAddress = second?.agentAddress ?? ""; - expect(repointed.currentRunId).toBe(secondAddress.split("@")[0] ?? ""); - expect(repointed.priorRunIds).toEqual(["run_gen1"]); - }); - }); - - // CL-7486: `sendRunMailWithReclaimRetry` used to retry every attempt - // against the address it started with, even after a wake in between - // attempts relaunched the run. A run already one relaunch generation - // in (`instanceId` stable, `currentRunId` already repointed once) that - // dies again while this loop is retrying reproduces the production - // failure exactly: a second relaunch repoints `currentRunId` again, - // and the address this loop still holds is now neither the stable - // `instanceId` nor the current `currentRunId` — invisible to - // `readBindingByAddressAnyTenant`, so the next wake for it throws "No - // workbench_launch binding" instead of finding the run that replaced - // it. - describe("sendRunMailWithReclaimRetry re-resolves the live address", () => { - test("a relaunch between retries is chased to its new address, not retried against the one that died", async () => { - resolveDefinitionSourcesResult = { - ok: true, - materials: [], - sources: [ - { - id: "off_1", - provider: "anthropic", - baseURL: "https://inference.invalid", - credentialId: "cred_placeholder", - model: "claude-sonnet-5", - }, - ], - defaultSource: "off_1", - }; - - const staleAddress = "run_gen1@ten1.workbench.test"; - const db = createFakeDb({ - assetRow: { - tenantId: "ten_1", - creatorPrincipalId: "prin_creator", - name: "ins_room1", - displayName: null, - }, - definitionId: "wfd_room1", - workflowRunRow: { - id: "run_gen1", - address: staleAddress, - principalId: "prin_room1", - definitionId: "wfd_room1", - // Already dead by the time the reclaim retry's own wake reads - // it — the run died again while this loop was mid-retry, not - // at the top of `sendMail` (which would have relaunched it - // before ever reaching this retry loop at all). - status: "failed", - }, - workflowDefinitionRow: { - id: "wfd_room1", - tenantId: "ten_1", - status: "deployed", - origin: "authored", - assetId: "asst_room1", - }, - workbenchLaunchRow: { - tenantId: "ten_1", - // `instanceId` is the room's own stable id, already distinct - // from `currentRunId` — this participant has been relaunched - // once before this test even starts. - instanceId: "ins_room1", - currentRunId: "run_gen1", - foldedBody: { - systemPrompt: "host prompt", - model: "claude-sonnet-5", - toolPackagePins: [], - grantRequirements: [], - credentialBindings: [], - }, - }, - }); - db.inserted.push({ - table: agentSession, - values: { id: "ses_run1", principalId: "prin_room1" }, - }); - - const sessionService = createFakeSessionService(); - sessionService.sendUserMessage = async (params: unknown) => { - sessionService.sendUserMessageCalls.push(params); - const { agentAddress } = params as { agentAddress: string }; + const staleRunId = "run_gen1"; + const runTrigger = createFakeRunTrigger(); + const baseTriggerMail = runTrigger.triggerMail; + runTrigger.triggerMail = async (input) => { // The dead run never becomes reachable again — only a send to - // whatever address the relaunch actually produced can succeed. - if (agentAddress === staleAddress) { + // whatever run the relaunch actually produced can succeed. + if (input.anchorRunId === staleRunId) { + runTrigger.triggerCalls.push(input); throw new Error("agent is unreachable"); } - return new TextEncoder().encode("raw-mime-bytes"); + return baseTriggerMail(input); }; // Routable so `sendMail`'s own opening wake-before-send gate skips @@ -2965,7 +2813,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter, eventCollectors: createFakeEventCollectors(), workflowAllocationService: allocationService, @@ -2981,22 +2829,18 @@ describe("createHubChatPlatform", () => { }); expect(sent.id).toBeTruthy(); - expect(sessionService.sendUserMessageCalls).toHaveLength(2); - const [first, second] = sessionService.sendUserMessageCalls as { - agentAddress: string; - }[]; - expect(first?.agentAddress).toBe(staleAddress); + expect(runTrigger.triggerCalls).toHaveLength(2); + const [first, second] = runTrigger.triggerCalls; + expect(first?.anchorRunId).toBe(staleRunId); // The retry that follows the relaunch targets the run that - // replaced `run_gen1`, never the dead address the loop started - // with. - expect(second?.agentAddress).not.toBe(staleAddress); + // replaced `run_gen1`, never the dead run the loop started with. + expect(second?.anchorRunId).not.toBe(staleRunId); const repointed = db.updated.at(-1)?.values as { currentRunId: string; priorRunIds: string[]; }; - const secondAddress = second?.agentAddress ?? ""; - expect(repointed.currentRunId).toBe(secondAddress.split("@")[0] ?? ""); + expect(repointed.currentRunId).toBe(second?.anchorRunId ?? ""); expect(repointed.priorRunIds).toEqual(["run_gen1"]); }); }); @@ -3060,7 +2904,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter(), eventCollectors: createFakeEventCollectors(), }); @@ -3106,11 +2950,11 @@ describe("createHubChatPlatform", () => { table: agentSession, values: { id: "ses_agent1", principalId: "prin_agent1" }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), eventCollectors: createFakeEventCollectors(), lifecycle: { idleSleepMs: 60_000 }, @@ -3207,7 +3051,7 @@ describe("createHubChatPlatform", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter(), eventCollectors: createFakeEventCollectors(), }); @@ -3332,7 +3176,7 @@ describe("createHubChatPlatform stale-definition reconciliation", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter, eventCollectors: createFakeEventCollectors(), }); @@ -3406,11 +3250,11 @@ describe("createHubChatPlatform stale-definition reconciliation", () => { table: agentSession, values: { id: "ses_stale", principalId: "prin_room1" }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, sidecarRouter: createFakeSidecarRouter({ routableAddresses: ["run_stale@ten1.workbench.test"], }), @@ -3497,7 +3341,7 @@ describe("createHubChatPlatform stale-definition reconciliation", () => { const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter({ routableAddresses: ["run_standalone@ten1.workbench.test"], }), @@ -3557,12 +3401,12 @@ describe("createHubChatPlatform relaunch sweep", () => { noopInference: opts.noopInference ?? true, }, }); - const sessionService = createFakeSessionService(); + const runTrigger = createFakeRunTrigger(); const notices: unknown[] = []; const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService, + runTrigger, // Routable, and dead anyway: that combination is exactly what the // wake path cannot fix — boot restore re-announced the address, so // nothing looks broken until the next message is dropped. @@ -3572,7 +3416,7 @@ describe("createHubChatPlatform relaunch sweep", () => { eventCollectors: createFakeEventCollectors(), relaunchNotice: { current: (notice) => notices.push(notice) }, }); - return { db, platform, sessionService, notices }; + return { db, platform, runTrigger, notices }; } test("relaunches a routable-but-dead participant and tells the room", async () => { @@ -3774,7 +3618,7 @@ describe("createHubChatPlatform inference-source rotation reconciliation", () => const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter({ routableAddresses: ["run_live@ten1.workbench.test"], }), @@ -4003,7 +3847,7 @@ describe("createHubChatPlatform pinned-tool-package connect reconciliation", () const platform = createPlatform({ toolGrantsForPins: () => [], db: db as never, - sessionService: createFakeSessionService(), + runTrigger: createFakeRunTrigger(), sidecarRouter: createFakeSidecarRouter({ routableAddresses: ["run_live@ten1.workbench.test"], }), diff --git a/packages/chat/test/relaunch-close.test.ts b/packages/chat/test/relaunch-close.test.ts index 35f2c0627d..b4a81f59f4 100644 --- a/packages/chat/test/relaunch-close.test.ts +++ b/packages/chat/test/relaunch-close.test.ts @@ -128,7 +128,7 @@ function createPlatform(db: never) { return createHubChatPlatform({ db, toolGrantsForPins: () => [], - sessionService: {} as never, + runTrigger: {} as never, repoStore: { resolveRef: async () => "sha_test" }, sidecarRouter: { getRoutableAddresses: () => [] } as never, eventCollectors: {} as never,