From c57dd9d75327e738894292a0bb006f5c7d607c17 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Sun, 6 Sep 2026 14:15:26 -0700 Subject: [PATCH 1/5] Fix merge fallout: restore CL-7481 agent-session exports and drop a duplicate key Merging cl-7107's repin/re-vendor audit into this branch left packages/workflows/src/index.ts re-exporting the pre-CL-7481 agent-session API (recordAgentSessionForRun) that no longer exists, its matching test file testing that same stale API, and a duplicated eventCollectors key in apps/hub/src/index.ts's webhook-trigger launch wiring. Restores the current exports (recordAgentSessionAtProvision, ensureRunSession) and their test, and drops the duplicate object key. --- apps/hub/src/index.ts | 1 - packages/workflows/src/index.ts | 4 +- packages/workflows/test/agent-session.test.ts | 204 ++++++++++++------ 3 files changed, 140 insertions(+), 69 deletions(-) diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index b62dbcb1ff..fbf44a094d 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -2337,7 +2337,6 @@ export async function createHub(config: HubConfig) { isRoutable: (address) => sidecarRouter.getRoutableAddresses().includes(address), cryptoProviderCache: cryptoProviders, - eventCollectors, persistLaunch: async (input) => { await workbenchLaunchPersistExtra(input)(db); }, diff --git a/packages/workflows/src/index.ts b/packages/workflows/src/index.ts index 244d0478ab..64bfbdff98 100644 --- a/packages/workflows/src/index.ts +++ b/packages/workflows/src/index.ts @@ -56,10 +56,10 @@ export { } from "./schedule/cron"; export { scheduleCronFromProjection } from "./schedule/from-projection"; export { - recordAgentSessionForRun, + recordAgentSessionAtProvision, + ensureRunSession, endAgentSessionForPrincipal, endAgentSessionForRun, - type RecordAgentSessionParams, type EventCollectorPort, } from "./launch/agent-session"; export { diff --git a/packages/workflows/test/agent-session.test.ts b/packages/workflows/test/agent-session.test.ts index 3ae50e221e..6fe8dcfb64 100644 --- a/packages/workflows/test/agent-session.test.ts +++ b/packages/workflows/test/agent-session.test.ts @@ -2,11 +2,16 @@ // repo's convention for tests that talk to a real Postgres (see // `packages/access-tools/test/routes.test.ts`). // -// CL-7477: before this fix, nothing wrote `agent_session` for a native -// launch, so `resolveRunSessionId` (vendor's mail-routing lookup) could -// never find a session for a run's principal and every launch's first -// turn failed to persist mail. This proves `recordAgentSessionForRun` -// writes a row `resolveRunSessionId` actually reads back. +// CL-7481: `recordAgentSessionAtProvision` is the eager write every +// native launcher makes right after `prepareProvisionedDeployment` +// returns — this proves it lands the `agent_session` row under the +// *deploying* principal (the run's own principal does not exist yet) +// and creates the run's event collector exactly once, idempotently. +// `ensureRunSession` is now purely a re-key: once the run's first +// trigger anchors a real principal onto it, this proves the session +// moves onto that principal without ever touching the session id, and +// that a run with no launch spec row is a bug this throws on rather +// than papering over. import { afterAll, beforeAll, expect, test } from "bun:test"; import { eq } from "drizzle-orm"; @@ -16,8 +21,8 @@ import { resolveRunSessionId } from "@intx/hub-sessions"; import { dbGate } from "../../../scripts/e2e/db-gate"; import { - endAgentSessionForRun, - recordAgentSessionForRun, + ensureRunSession, + recordAgentSessionAtProvision, } from "../src/launch/agent-session"; function dbConfigFromUrl(databaseUrl: string) { @@ -34,14 +39,17 @@ function dbConfigFromUrl(databaseUrl: string) { const databaseUrl = process.env["DATABASE_URL"]; const describeIfDb = dbGate(databaseUrl, import.meta.path); -describeIfDb("recordAgentSessionForRun", () => { +describeIfDb("recordAgentSessionAtProvision / ensureRunSession", () => { let db: DB; const tenantId = generateId("tenant"); - const principalId = generateId("principal"); + const deployingPrincipalId = generateId("principal"); + const runPrincipalId = generateId("principal"); const definitionId = generateId("workflowDefinition"); - const anchorRunId = generateId("workflowRun"); + const provisionedRunId = generateId("workflowRun"); + const noLaunchSpecRunId = generateId("workflowRun"); const sessionId = generateId("session"); + const domain = `agent-session-${tenantId}.localhost`; beforeAll(async () => { if (databaseUrl === undefined) return; @@ -51,35 +59,70 @@ describeIfDb("recordAgentSessionForRun", () => { id: tenantId, name: "Agent Session Test Tenant", slug: `agent-session-${tenantId}`, - domain: `agent-session-${tenantId}.localhost`, + domain, parentId: null, config: null, }); - await db.db.insert(schema.principal).values({ - id: principalId, - tenantId, - kind: "workflow", - refId: anchorRunId, - status: "active", - }); + await db.db.insert(schema.principal).values([ + { + id: deployingPrincipalId, + tenantId, + kind: "user", + refId: "usr_test", + status: "active", + }, + { + id: runPrincipalId, + tenantId, + kind: "workflow", + refId: provisionedRunId, + status: "active", + }, + ]); await db.db.insert(schema.workflowDefinition).values({ id: definitionId, tenantId, - creatorPrincipalId: principalId, + creatorPrincipalId: deployingPrincipalId, assetId: null, name: "agent-session-test-definition", }); + // Born the instant `prepareProvisionedDeployment` returns: a run row + // exists, its own principal is still null, and its launch spec names + // the session `recordAgentSessionAtProvision` will record. + await db.db.insert(schema.workflowRun).values({ + id: provisionedRunId, + definitionId, + anchorRunId: provisionedRunId, + tenantId, + principalId: null, + address: `${provisionedRunId}@${domain}`, + status: "deployed", + }); + await db.db.insert(schema.workflowRunLaunchSpec).values({ + anchorRunId: provisionedRunId, + sessionId, + deploymentDomain: domain, + sourceAuthorityPrincipalId: deployingPrincipalId, + frozenApprovalBundle: {}, + sourceOfferingIds: [], + defaultSourceOfferingId: "off_test", + deployContent: { systemPrompt: "" }, + }); + + // A run with no launch spec row at all: a launcher that skipped the + // shared provisioning path — a bug `ensureRunSession` must surface, + // not silently no-op past. await db.db.insert(schema.workflowRun).values({ - id: anchorRunId, + id: noLaunchSpecRunId, definitionId, - anchorRunId, + anchorRunId: noLaunchSpecRunId, tenantId, - principalId, - address: `${anchorRunId}@agent-session-${tenantId}.localhost`, - status: "running", + principalId: null, + address: `${noLaunchSpecRunId}@${domain}`, + status: "deployed", }); }); @@ -88,68 +131,97 @@ describeIfDb("recordAgentSessionForRun", () => { await db.db .delete(schema.agentSession) .where(eq(schema.agentSession.id, sessionId)); + await db.db + .delete(schema.workflowRunLaunchSpec) + .where(eq(schema.workflowRunLaunchSpec.anchorRunId, provisionedRunId)); + await db.db + .delete(schema.workflowRun) + .where(eq(schema.workflowRun.id, provisionedRunId)); await db.db .delete(schema.workflowRun) - .where(eq(schema.workflowRun.id, anchorRunId)); + .where(eq(schema.workflowRun.id, noLaunchSpecRunId)); await db.db .delete(schema.workflowDefinition) .where(eq(schema.workflowDefinition.id, definitionId)); await db.db .delete(schema.principal) - .where(eq(schema.principal.id, principalId)); + .where(eq(schema.principal.id, runPrincipalId)); + await db.db + .delete(schema.principal) + .where(eq(schema.principal.id, deployingPrincipalId)); await db.db.delete(schema.tenant).where(eq(schema.tenant.id, tenantId)); }); - test("writes an agent_session resolveRunSessionId reads back", async () => { - await recordAgentSessionForRun( - db.db, - { sessionId, anchorRunId }, - { create: () => undefined }, - ); - - const resolved = await resolveRunSessionId(db.db, principalId); - expect(resolved).toBe(sessionId); - }); - - test("creates the run's event collector with its address/tenant/session/run id", async () => { + test("records the session under the deploying principal and creates the collector once", async () => { const createCalls: unknown[] = []; - const otherSessionId = generateId("session"); - - await recordAgentSessionForRun( - db.db, - { sessionId: otherSessionId, anchorRunId }, - { - create: (...args: unknown[]) => { - createCalls.push(args); - }, + let hasCollector = false; + const eventCollectors = { + create: (...args: unknown[]) => { + createCalls.push(args); + hasCollector = true; }, - ); + has: () => hasCollector, + }; + + await recordAgentSessionAtProvision({ + db: db.db, + eventCollectors, + runId: provisionedRunId, + sessionId, + sourceAuthorityPrincipalId: deployingPrincipalId, + }); + // A second call for the same run is a no-op, not a conflict, and + // must not create a second collector. + await recordAgentSessionAtProvision({ + db: db.db, + eventCollectors, + runId: provisionedRunId, + sessionId, + sourceAuthorityPrincipalId: deployingPrincipalId, + }); expect(createCalls).toEqual([ - [ - `${anchorRunId}@agent-session-${tenantId}.localhost`, - tenantId, - otherSessionId, - anchorRunId, - ], + [`${provisionedRunId}@${domain}`, tenantId, sessionId, provisionedRunId], ]); - await db.db - .delete(schema.agentSession) - .where(eq(schema.agentSession.id, otherSessionId)); + const sessionRow = await db.db.query.agentSession.findFirst({ + where: eq(schema.agentSession.id, sessionId), + }); + expect(sessionRow?.principalId).toBe(deployingPrincipalId); + expect(sessionRow?.agentId).toBe(definitionId); }); - test("abandons the run's event collector when the session ends", async () => { - const abandonCalls: string[] = []; + test("re-keys the session onto the run principal once Interchange anchors it", async () => { + await db.db + .update(schema.workflowRun) + .set({ principalId: runPrincipalId, status: "running" }) + .where(eq(schema.workflowRun.id, provisionedRunId)); + + const resolved = await ensureRunSession({ + db: db.db, + eventCollectors: { create: () => {}, has: () => true }, + runId: provisionedRunId, + }); - await endAgentSessionForRun(db.db, anchorRunId, { - abandon: (address: string) => { - abandonCalls.push(address); - }, + expect(resolved).toBe(sessionId); + const sessionRow = await db.db.query.agentSession.findFirst({ + where: eq(schema.agentSession.id, sessionId), }); + expect(sessionRow?.principalId).toBe(runPrincipalId); - expect(abandonCalls).toEqual([ - `${anchorRunId}@agent-session-${tenantId}.localhost`, - ]); + // Interchange's own mail-routing lookup now finds the session by the + // run's own principal, exactly as it would for a folded run. + const viaVendorLookup = await resolveRunSessionId(db.db, runPrincipalId); + expect(viaVendorLookup).toBe(sessionId); + }); + + test("throws for a run with no launch spec row", async () => { + await expect( + ensureRunSession({ + db: db.db, + eventCollectors: { create: () => {}, has: () => false }, + runId: noLaunchSpecRunId, + }), + ).rejects.toThrow(/no workflow_run_launch_spec/); }); }); From 42f051164eff60ddaed7e32b58090d48740be9f0 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Sun, 6 Sep 2026 14:15:31 -0700 Subject: [PATCH 2/5] Add tests for wake retry re-resolving the live address (CL-7486) sendRunMailWithReclaimRetry retried every attempt against the address it started with. Proves the reproduction of the production failure: a participant already one relaunch generation in that dies again mid-retry must have its next retry chase the run that replaced it, not keep hammering the address that just died. --- packages/chat/test/platform-adapter.test.ts | 133 ++++++++++++++++++++ 1 file changed, 133 insertions(+) diff --git a/packages/chat/test/platform-adapter.test.ts b/packages/chat/test/platform-adapter.test.ts index 120538c926..89ad655d58 100644 --- a/packages/chat/test/platform-adapter.test.ts +++ b/packages/chat/test/platform-adapter.test.ts @@ -2781,6 +2781,139 @@ describe("createHubChatPlatform", () => { }); }); + // 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 }; + // 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], + }); + + const platform = createPlatform({ + toolGrantsForPins: () => [], + db: db as never, + sessionService, + sidecarRouter, + eventCollectors: createFakeEventCollectors(), + reclaimRetryDelaysMs: [1], + 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"]); + }); + }); + // Proves the actual lever an edited system prompt reaches a running // instance through: `wakeByAddress` (exercised via `sendMail`'s // wake-on-send path above) replays `workbench_launch.foldedBody` From c5cbcebe417a51018a868b831334a6171a617c33 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Sun, 6 Sep 2026 14:15:41 -0700 Subject: [PATCH 3/5] Chat: wake retry chases the run a relaunch replaced it with (CL-7486) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sendRunMailWithReclaimRetry retried every delivery attempt, and woke, against the address it captured at the top of the loop. A wake in between retries can relaunch the run (its dead-run branch repoints the room's stable participant at a fresh address without telling this loop), and a participant already one relaunch generation in that dies again mid-retry leaves this loop holding an address that is no longer either the stable participant id or the current run id — invisible to readBindingByAddressAnyTenant. The next wake for it then threw "No workbench_launch binding" instead of finding the run that replaced it, which is what chat e2e's mention-fanout and invite-drives-a-turn cases were hitting. Re-resolves the live address (and its session) off the stable participant id after each wake, so the retry loop always chases whichever run is actually current. Also hardens publishCorbitsToolsRegistry's create-then-list-on-conflict ensure: two overlapping seed runs racing the same tenant's package registry asset can still throw a raw asset_tenant_kind_name violation if the substrate's own 409 mapping is bypassed; the ensure now also recognizes that violation by name and resolves the winner's row instead of throwing. --- packages/chat/src/platform-adapter.ts | 63 ++++++++++++++++--- packages/tool-registry-publish/src/publish.ts | 14 ++++- 2 files changed, 69 insertions(+), 8 deletions(-) diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index b7c5c51477..1113171348 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -989,10 +989,24 @@ export function createHubChatPlatform( * more time regardless. Exhausting every delay means the condition * is not transient and the caller gets a clean `AgentUnreachableError` * rather than an unhandled 500. + * + * CL-7486: a wake this loop runs between retries can itself relaunch + * the run — `wakeByAddress`'s dead-run branch repoints the room's + * stable participant at a fresh address without telling this loop. + * Retrying `sendRunMail` against the address it started with would + * then keep hammering a run that will never become routable again, + * and the wake ahead of the NEXT retry would fail outright (no + * `workbench_launch` row still names that address at all) instead of + * the "unreachable" this loop knows how to ride out. `reresolve` + * re-reads the live address (and its session) off the stable + * participant id before every wake/retry, so this loop always chases + * whichever run is actually current. */ async function sendRunMailWithReclaimRetry( params: SendRunMailParams, + reresolve: () => Promise<{ agentAddress: string; sessionId: string }>, ): Promise>> { + let current = params; let loggedRetryStart = false; for (let attempt = 0; ; attempt++) { try { @@ -1001,29 +1015,44 @@ export function createHubChatPlatform( // to accept regardless — the signal parameter is unused here, // same abandon-on-timeout behavior as before. return await withTimeout( - () => sendRunMail(runMailDeps, params), + () => sendRunMail(runMailDeps, current), MAIL_DELIVERY_TIMEOUT_MS, - `mail to ${params.agentAddress} did not settle within ${String(MAIL_DELIVERY_TIMEOUT_MS)}ms`, + `mail to ${current.agentAddress} did not settle within ${String(MAIL_DELIVERY_TIMEOUT_MS)}ms`, ); } catch (err) { const delay = RECLAIM_RETRY_DELAYS_MS[attempt]; if (!isAgentUnreachable(err) || delay === undefined) { if (loggedRetryStart) { - wakeLogger.info`mail to ${params.agentAddress} exhausted every reclaim retry; giving up`; + wakeLogger.info`mail to ${current.agentAddress} exhausted every reclaim retry; giving up`; } if (isAgentUnreachable(err)) { - throw new AgentUnreachableError(params.agentAddress, { + throw new AgentUnreachableError(current.agentAddress, { cause: err, }); } throw err; } if (!loggedRetryStart) { - wakeLogger.info`mail to ${params.agentAddress} hit "agent is unreachable"; retrying with backoff while the post-restart reclaim settles`; + wakeLogger.info`mail to ${current.agentAddress} hit "agent is unreachable"; retrying with backoff while the post-restart reclaim settles`; loggedRetryStart = true; } await sleep(delay); - await wakeByAddressBounded(params.agentAddress); + // Wake whatever this loop still believes is live — `wakeByAddress` + // itself tolerates an address one relaunch behind the room's + // current one (its lookup matches either the stable participant + // id or the live run id) — and only THEN re-resolve, since the + // wake just run is exactly the kind of relaunch that can leave + // `current.agentAddress` pointing at a run that no longer exists + // at all. Re-resolving before the wake instead would still hand + // the next retry whatever address was current a moment ago, + // stale again the instant this wake relaunches it. + await wakeByAddressBounded(current.agentAddress); + const fresh = await reresolve(); + current = { + ...current, + agentAddress: fresh.agentAddress, + sessionId: fresh.sessionId, + }; } } } @@ -1399,7 +1428,27 @@ export function createHubChatPlatform( ? { references: input.content.references } : {}), }; - const sent = await sendRunMailWithReclaimRetry(withThreading); + const sent = await sendRunMailWithReclaimRetry(withThreading, async () => { + const relive = await requireLive(input.workbenchId, input.tenantId); + try { + await ensureRunSession({ + db: deps.db, + eventCollectors: deps.eventCollectors, + runId: relive.run.id, + }); + } catch (err) { + reportError(err, { + operation: "chat.sendMail.ensureRunSession", + tenantId: input.tenantId, + agentId: relive.binding.liveAddress, + extra: { runId: relive.run.id }, + }); + } + return { + agentAddress: relive.binding.liveAddress, + sessionId: await resolveRunSessionIdOrThrow(deps.db, relive.run), + }; + }); lifecycle?.recordActivity(deliveryAddress); diff --git a/packages/tool-registry-publish/src/publish.ts b/packages/tool-registry-publish/src/publish.ts index 301bdd6c64..d6eebbd36d 100644 --- a/packages/tool-registry-publish/src/publish.ts +++ b/packages/tool-registry-publish/src/publish.ts @@ -110,7 +110,19 @@ async function ensureRegistryAsset( if (created.status === 201) { return parseSchema(AssetResponse, created.data, "create asset response").id; } - if (created.status !== 409) { + // CL-7486: two overlapping seed runs for the same tenant (the root + // tenant's own boot-time seed racing a second caller seeding that + // same tenant) can both pass the `listRegistryAsset` check above + // before either has created the row, then both post-create — the + // loser hits the `asset_tenant_kind_name` unique constraint. The + // substrate answers that as `409` in the common case; asserted here + // by name too, in case a raw constraint violation surfaces as a + // generic 500 instead, so the loser still resolves the winner's row + // rather than throwing on a race this ensure is meant to tolerate. + const isNameCollision = + created.status === 409 || + JSON.stringify(created.data).includes("asset_tenant_kind_name"); + if (!isNameCollision) { throw new Error( `publishCorbitsToolsRegistry: failed to create the ${CORBITS_TOOLS_REGISTRY} asset: ${String(created.status)}`, ); From ebc173bdd5aaa12cad9068df676586788a827314 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Sun, 6 Sep 2026 14:16:53 -0700 Subject: [PATCH 4/5] Fix webhook-triggers launch test fixture: eventCollectors needs has() --- packages/webhook-triggers/test/launch.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/webhook-triggers/test/launch.test.ts b/packages/webhook-triggers/test/launch.test.ts index 01b3b6785c..e671da0093 100644 --- a/packages/webhook-triggers/test/launch.test.ts +++ b/packages/webhook-triggers/test/launch.test.ts @@ -199,6 +199,7 @@ function baseDeps() { isRoutable: () => isRoutableForTest, eventCollectors: { create: () => undefined, + has: () => false, }, }; } From 47d3409e29f6a6b93ea8332d26d18437a64d5c23 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Sun, 6 Sep 2026 14:34:15 -0700 Subject: [PATCH 5/5] chat: format platform-adapter.ts --- packages/chat/src/platform-adapter.ts | 45 ++++++++++++++------------- 1 file changed, 24 insertions(+), 21 deletions(-) diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index 1113171348..d32fb65d15 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -1428,27 +1428,30 @@ export function createHubChatPlatform( ? { references: input.content.references } : {}), }; - const sent = await sendRunMailWithReclaimRetry(withThreading, async () => { - const relive = await requireLive(input.workbenchId, input.tenantId); - try { - await ensureRunSession({ - db: deps.db, - eventCollectors: deps.eventCollectors, - runId: relive.run.id, - }); - } catch (err) { - reportError(err, { - operation: "chat.sendMail.ensureRunSession", - tenantId: input.tenantId, - agentId: relive.binding.liveAddress, - extra: { runId: relive.run.id }, - }); - } - return { - agentAddress: relive.binding.liveAddress, - sessionId: await resolveRunSessionIdOrThrow(deps.db, relive.run), - }; - }); + const sent = await sendRunMailWithReclaimRetry( + withThreading, + async () => { + const relive = await requireLive(input.workbenchId, input.tenantId); + try { + await ensureRunSession({ + db: deps.db, + eventCollectors: deps.eventCollectors, + runId: relive.run.id, + }); + } catch (err) { + reportError(err, { + operation: "chat.sendMail.ensureRunSession", + tenantId: input.tenantId, + agentId: relive.binding.liveAddress, + extra: { runId: relive.run.id }, + }); + } + return { + agentAddress: relive.binding.liveAddress, + sessionId: await resolveRunSessionIdOrThrow(deps.db, relive.run), + }; + }, + ); lifecycle?.recordActivity(deliveryAddress);