From c181204a3abc6612e2c2b242bec8712aff1b26ba Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 15:42:14 -0700 Subject: [PATCH 1/6] Add tests for a run's first reply reaching the chat --- e2e/workflows/first-reply-session.test.ts | 142 ++++++++++++++++++++++ 1 file changed, 142 insertions(+) create mode 100644 e2e/workflows/first-reply-session.test.ts diff --git a/e2e/workflows/first-reply-session.test.ts b/e2e/workflows/first-reply-session.test.ts new file mode 100644 index 000000000..52a5c3f8e --- /dev/null +++ b/e2e/workflows/first-reply-session.test.ts @@ -0,0 +1,142 @@ +// DB-gated. A run's first reply must reach the chat: its event collector has +// to exist before the first inference event, and its first outbound mail has +// to land in session_mail even though the agent_session is created mid-call. +import { afterAll, beforeAll, expect, test } from "bun:test"; +import { eq, inArray } from "drizzle-orm"; + +import { createDB, schema, type DB } from "@intx/db"; +import { generateId } from "@intx/hub-common"; +import type { EventCollectorRegistry } from "@intx/hub-sessions"; +import { dbGate } from "../lib/db-gate"; + +import { createHubPersistMailWithSessionEnsure } from "../../apps/hub/src/mailbox-persist"; +import { withLazyRunCollector } from "../../packages/workflows/src/launch/agent-session"; + +function dbConfigFromUrl(databaseUrl: string) { + const url = new URL(databaseUrl); + return { + host: url.hostname, + port: url.port === "" ? 5432 : Number(url.port), + user: decodeURIComponent(url.username), + password: decodeURIComponent(url.password), + database: url.pathname.replace(/^\//, ""), + }; +} + +const databaseUrl = process.env["DATABASE_URL"]; +const describeIfDb = dbGate(databaseUrl, import.meta.path); + +type Registry = Pick; + +describeIfDb("first reply of a run", () => { + let db: DB; + + const tenantId = generateId("tenant"); + const userPrincipalId = generateId("principal"); + const runPrincipalId = generateId("principal"); + const definitionId = generateId("workflowDefinition"); + const runId = generateId("workflowRun"); + const sessionId = generateId("session"); + const domain = `first-reply-${tenantId}.localhost`; + const address = `${runId}@${domain}`; + + beforeAll(async () => { + if (databaseUrl === undefined) return; + db = createDB(dbConfigFromUrl(databaseUrl)); + + await db.db.insert(schema.tenant).values({ + id: tenantId, + name: "First Reply Test Tenant", + slug: `first-reply-${tenantId.replaceAll("_", "-")}`, + domain, + parentId: null, + config: null, + }); + await db.db.insert(schema.principal).values([ + { id: userPrincipalId, tenantId, kind: "user", refId: "usr_test", status: "active" }, + { id: runPrincipalId, tenantId, kind: "workflow", refId: runId, status: "active" }, + ]); + await db.db.insert(schema.workflowDefinition).values({ + id: definitionId, + tenantId, + creatorPrincipalId: userPrincipalId, + assetId: null, + name: "first-reply-test-definition", + }); + await db.db.insert(schema.workflowRun).values({ + id: runId, + definitionId, + anchorRunId: runId, + tenantId, + principalId: runPrincipalId, + address, + status: "running", + }); + await db.db.insert(schema.workflowRunLaunchSpec).values({ + anchorRunId: runId, + sessionId, + deploymentDomain: domain, + sourceAuthorityPrincipalId: userPrincipalId, + frozenApprovalBundle: {}, + sourceOfferingIds: [], + defaultSourceOfferingId: "off_test", + deployContent: { systemPrompt: "" }, + }); + }); + + afterAll(async () => { + if (databaseUrl === undefined) return; + await db.db.delete(schema.agentSession).where(eq(schema.agentSession.id, sessionId)); + await db.db + .delete(schema.workflowRunLaunchSpec) + .where(eq(schema.workflowRunLaunchSpec.anchorRunId, runId)); + await db.db.delete(schema.workflowRun).where(eq(schema.workflowRun.id, runId)); + await db.db + .delete(schema.workflowDefinition) + .where(eq(schema.workflowDefinition.id, definitionId)); + await db.db + .delete(schema.principal) + .where(inArray(schema.principal.id, [runPrincipalId, userPrincipalId])); + await db.db.delete(schema.tenant).where(eq(schema.tenant.id, tenantId)); + }); + + test("creates the collector before dispatching the run's first event, in order", async () => { + const collectors = new Set(); + const dispatched: string[] = []; + const registry: Registry = { + has: (a) => collectors.has(a), + create: (a) => { + collectors.add(a); + dispatched.push("create"); + }, + dispatch: (a, event) => { + if (!collectors.has(a)) return; + dispatched.push(event.type); + }, + }; + const lazy = withLazyRunCollector(registry, db.db); + + lazy.dispatch(address, { type: "custom.first", seq: 1, data: {} }); + lazy.dispatch(address, { type: "custom.second", seq: 2, data: {} }); + await Bun.sleep(200); + + expect(dispatched).toEqual(["create", "custom.first", "custom.second"]); + }); + + test("persists the run's first outbound mail although its session is created mid-call", async () => { + await db.db.delete(schema.agentSession).where(eq(schema.agentSession.id, sessionId)); + const upstreamCalls: unknown[] = []; + const persist = createHubPersistMailWithSessionEnsure( + db.db, + { current: { create: () => {}, has: () => true } }, + async (args) => { + upstreamCalls.push(args); + return []; + }, + ); + + await persist({ senderAddress: address } as Parameters[0]); + + expect(upstreamCalls).toHaveLength(1); + }); +}); From eb2d00d81c8a38961e172260d73803d2d55100ee Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 15:42:14 -0700 Subject: [PATCH 2/6] hub: create run collector on first event and persist first mail after session ensure (CL-9907) --- apps/hub/src/mailbox-persist.ts | 10 ++--- apps/hub/src/server.ts | 38 ++++++++-------- packages/workflows/src/index.ts | 6 ++- .../workflows/src/launch/agent-session.ts | 43 +++++++++++++++++++ 4 files changed, 74 insertions(+), 23 deletions(-) diff --git a/apps/hub/src/mailbox-persist.ts b/apps/hub/src/mailbox-persist.ts index e33bc1e46..89f0e835d 100644 --- a/apps/hub/src/mailbox-persist.ts +++ b/apps/hub/src/mailbox-persist.ts @@ -25,11 +25,12 @@ export function createHubPersistMailWithSessionEnsure Promise, ): (args: MailboxPersistArgs) => Promise { return async (args) => { - const sender = await resolveRoutableAddress(db, args.senderAddress); + let sender = await resolveRoutableAddress(db, args.senderAddress); try { const eventCollectors = eventCollectorsRef.current; if (sender !== undefined && eventCollectors !== undefined) { await ensureRunSession({ db, eventCollectors, runId: sender.id }); + sender = await resolveRoutableAddress(db, args.senderAddress); } } catch (err) { reportError(err, { @@ -37,10 +38,9 @@ export function createHubPersistMailWithSessionEnsure, +>(registry: R, db: DB["db"]): R { + const pending = new Map>(); + + async function ensureForAddress(address: string): Promise { + try { + const runRow = await db.query.workflowRun.findFirst({ + where: eq(workflowRun.address, address), + }); + if (runRow !== undefined) { + await ensureRunSession({ db, eventCollectors: registry, runId: runRow.id }); + } + } catch (err) { + reportError(err, { + operation: "workflows.lazyRunCollector.ensure", + extra: { address }, + }); + } + } + + const dispatch: R["dispatch"] = (address, event) => { + const inFlight = pending.get(address); + if (inFlight === undefined && (registry.has(address) || event.type === "reactor.done")) { + registry.dispatch(address, event); + return; + } + const next = (inFlight ?? ensureForAddress(address)).then(() => { + registry.dispatch(address, event); + }); + pending.set(address, next); + void next.finally(() => { + if (pending.get(address) === next) pending.delete(address); + }); + }; + + return { ...registry, dispatch }; +} From 4848968f9cf9ead96a986d2acfb7487693c1974e Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 15:42:14 -0700 Subject: [PATCH 3/6] Update docs: lazy run collector --- docs/agent-session-provisioning.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/docs/agent-session-provisioning.md b/docs/agent-session-provisioning.md index 975511688..ba98bfe3e 100644 --- a/docs/agent-session-provisioning.md +++ b/docs/agent-session-provisioning.md @@ -24,3 +24,7 @@ that first trigger may write the principal first. `ensureRunSession` is a true upsert, called from every seam that might be a run's first mail-routable moment: it creates the row once a principal is anchored, and re-keys an existing row onto it. + +The hub also wraps its event collector registry with `withLazyRunCollector`, +so a run's collector exists before its first inference event rather than +after its first outbound mail; otherwise the first reply's parts are dropped. From f5fb67957840d65ea0a7d6175bc3980f021192b5 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 15:58:31 -0700 Subject: [PATCH 4/6] workflows: lazy run collector needs a session, never resurrects, reports errors (CL-9907) --- e2e/workflows/first-reply-session.test.ts | 153 ++++++++++++------ .../workflows/src/launch/agent-session.ts | 85 ++++++++-- 2 files changed, 175 insertions(+), 63 deletions(-) diff --git a/e2e/workflows/first-reply-session.test.ts b/e2e/workflows/first-reply-session.test.ts index 52a5c3f8e..59544fafd 100644 --- a/e2e/workflows/first-reply-session.test.ts +++ b/e2e/workflows/first-reply-session.test.ts @@ -6,7 +6,7 @@ import { eq, inArray } from "drizzle-orm"; import { createDB, schema, type DB } from "@intx/db"; import { generateId } from "@intx/hub-common"; -import type { EventCollectorRegistry } from "@intx/hub-sessions"; +import { createEventCollectorRegistry } from "@intx/hub-sessions"; import { dbGate } from "../lib/db-gate"; import { createHubPersistMailWithSessionEnsure } from "../../apps/hub/src/mailbox-persist"; @@ -26,8 +26,6 @@ function dbConfigFromUrl(databaseUrl: string) { const databaseUrl = process.env["DATABASE_URL"]; const describeIfDb = dbGate(databaseUrl, import.meta.path); -type Registry = Pick; - describeIfDb("first reply of a run", () => { let db: DB; @@ -36,8 +34,17 @@ describeIfDb("first reply of a run", () => { const runPrincipalId = generateId("principal"); const definitionId = generateId("workflowDefinition"); const runId = generateId("workflowRun"); - const sessionId = generateId("session"); const domain = `first-reply-${tenantId}.localhost`; + const sessionId = generateId("session"); + const unanchoredRunId = generateId("workflowRun"); + const unanchoredSessionId = generateId("session"); + const unanchoredAddress = `${unanchoredRunId}@${domain}`; + const closedRunId = generateId("workflowRun"); + const closedSessionId = generateId("session"); + const closedAddress = `${closedRunId}@${domain}`; + const closedPrincipalId = generateId("principal"); + const runIds = [runId, unanchoredRunId, closedRunId]; + const sessionIds = [sessionId, unanchoredSessionId, closedSessionId]; const address = `${runId}@${domain}`; beforeAll(async () => { @@ -55,6 +62,7 @@ describeIfDb("first reply of a run", () => { await db.db.insert(schema.principal).values([ { id: userPrincipalId, tenantId, kind: "user", refId: "usr_test", status: "active" }, { id: runPrincipalId, tenantId, kind: "workflow", refId: runId, status: "active" }, + { id: closedPrincipalId, tenantId, kind: "workflow", refId: closedRunId, status: "active" }, ]); await db.db.insert(schema.workflowDefinition).values({ id: definitionId, @@ -63,64 +71,117 @@ describeIfDb("first reply of a run", () => { assetId: null, name: "first-reply-test-definition", }); - await db.db.insert(schema.workflowRun).values({ - id: runId, - definitionId, - anchorRunId: runId, - tenantId, - principalId: runPrincipalId, - address, - status: "running", - }); - await db.db.insert(schema.workflowRunLaunchSpec).values({ - anchorRunId: runId, - sessionId, - deploymentDomain: domain, - sourceAuthorityPrincipalId: userPrincipalId, - frozenApprovalBundle: {}, - sourceOfferingIds: [], - defaultSourceOfferingId: "off_test", - deployContent: { systemPrompt: "" }, - }); + const seedRun = async ( + id: string, + session: string, + principalId: string | null, + runAddress: string, + ) => { + await db.db.insert(schema.workflowRun).values({ + id, + definitionId, + anchorRunId: id, + tenantId, + principalId, + address: runAddress, + status: "running", + }); + await db.db.insert(schema.workflowRunLaunchSpec).values({ + anchorRunId: id, + sessionId: session, + deploymentDomain: domain, + sourceAuthorityPrincipalId: userPrincipalId, + frozenApprovalBundle: {}, + sourceOfferingIds: [], + defaultSourceOfferingId: "off_test", + deployContent: { systemPrompt: "" }, + }); + }; + await seedRun(runId, sessionId, runPrincipalId, address); + await seedRun(unanchoredRunId, unanchoredSessionId, null, unanchoredAddress); + await seedRun(closedRunId, closedSessionId, closedPrincipalId, closedAddress); }); afterAll(async () => { if (databaseUrl === undefined) return; - await db.db.delete(schema.agentSession).where(eq(schema.agentSession.id, sessionId)); + await db.db.delete(schema.inferenceTurn).where(eq(schema.inferenceTurn.tenantId, tenantId)); + await db.db.delete(schema.agentSession).where(inArray(schema.agentSession.id, sessionIds)); await db.db .delete(schema.workflowRunLaunchSpec) - .where(eq(schema.workflowRunLaunchSpec.anchorRunId, runId)); - await db.db.delete(schema.workflowRun).where(eq(schema.workflowRun.id, runId)); + .where(inArray(schema.workflowRunLaunchSpec.anchorRunId, runIds)); + await db.db.delete(schema.workflowRun).where(inArray(schema.workflowRun.id, runIds)); await db.db .delete(schema.workflowDefinition) .where(eq(schema.workflowDefinition.id, definitionId)); await db.db .delete(schema.principal) - .where(inArray(schema.principal.id, [runPrincipalId, userPrincipalId])); + .where(inArray(schema.principal.id, [runPrincipalId, closedPrincipalId, userPrincipalId])); await db.db.delete(schema.tenant).where(eq(schema.tenant.id, tenantId)); }); - test("creates the collector before dispatching the run's first event, in order", async () => { - const collectors = new Set(); - const dispatched: string[] = []; - const registry: Registry = { - has: (a) => collectors.has(a), - create: (a) => { - collectors.add(a); - dispatched.push("create"); - }, - dispatch: (a, event) => { - if (!collectors.has(a)) return; - dispatched.push(event.type); - }, - }; - const lazy = withLazyRunCollector(registry, db.db); + const start = (seq: number) => ({ type: "inference.start", seq, data: { model: "m" } }) as const; + const turnsFor = (session: string) => + db.db.query.inferenceTurn.findMany({ where: eq(schema.inferenceTurn.sessionId, session) }); + + test("persists the run's first turn although no collector existed yet", async () => { + const lazy = withLazyRunCollector(createEventCollectorRegistry({ db: db.db }), db.db); + + lazy.dispatch(address, start(1)); + await Bun.sleep(300); + + expect(lazy.has(address)).toBe(true); + expect(await turnsFor(sessionId)).toHaveLength(1); + }); + + test("creates no collector until the run has a session, then retries on the next event", async () => { + const lazy = withLazyRunCollector(createEventCollectorRegistry({ db: db.db }), db.db); + + lazy.dispatch(unanchoredAddress, start(1)); + await Bun.sleep(300); + expect(lazy.has(unanchoredAddress)).toBe(false); + + const [principal] = await db.db + .insert(schema.principal) + .values({ + id: generateId("principal"), + tenantId, + kind: "workflow", + refId: unanchoredRunId, + status: "active", + }) + .returning(); + await db.db + .update(schema.workflowRun) + .set({ principalId: principal!.id }) + .where(eq(schema.workflowRun.id, unanchoredRunId)); + + lazy.dispatch(unanchoredAddress, start(2)); + await Bun.sleep(300); + expect(lazy.has(unanchoredAddress)).toBe(true); + expect(await turnsFor(unanchoredSessionId)).toHaveLength(1); + + await db.db.delete(schema.inferenceTurn).where(eq(schema.inferenceTurn.tenantId, tenantId)); + await db.db.delete(schema.agentSession).where(eq(schema.agentSession.id, unanchoredSessionId)); + await db.db + .update(schema.workflowRun) + .set({ principalId: null }) + .where(eq(schema.workflowRun.id, unanchoredRunId)); + await db.db.delete(schema.principal).where(eq(schema.principal.id, principal!.id)); + }); + + test("does not resurrect a collector after abandon", async () => { + const lazy = withLazyRunCollector(createEventCollectorRegistry({ db: db.db }), db.db); + + lazy.dispatch(closedAddress, start(1)); + await Bun.sleep(300); + expect(lazy.has(closedAddress)).toBe(true); - lazy.dispatch(address, { type: "custom.first", seq: 1, data: {} }); - lazy.dispatch(address, { type: "custom.second", seq: 2, data: {} }); - await Bun.sleep(200); + lazy.abandon(closedAddress); + await Bun.sleep(300); + lazy.dispatch(closedAddress, start(2)); + await Bun.sleep(300); - expect(dispatched).toEqual(["create", "custom.first", "custom.second"]); + expect(lazy.has(closedAddress)).toBe(false); }); test("persists the run's first outbound mail although its session is created mid-call", async () => { diff --git a/packages/workflows/src/launch/agent-session.ts b/packages/workflows/src/launch/agent-session.ts index a17bfba3b..2a3e08769 100644 --- a/packages/workflows/src/launch/agent-session.ts +++ b/packages/workflows/src/launch/agent-session.ts @@ -2,7 +2,12 @@ // turn. See docs/agent-session-provisioning.md. import { eq } from "drizzle-orm"; import type { DB } from "@intx/db"; -import { agentSession, workflowRun, workflowRunLaunchSpec } from "@intx/db/schema"; +import { + agentSession, + isLiveWorkflowRunStatus, + workflowRun, + workflowRunLaunchSpec, +} from "@intx/db/schema"; import type { EventCollectorRegistry } from "@intx/hub-sessions"; import { reportError } from "@corbits/error-sink"; @@ -24,6 +29,15 @@ export async function ensureRunSession(params: { if (runRow === undefined) { return null; } + return ensureSessionForRun(db, eventCollectors, runRow); +} + +async function ensureSessionForRun( + db: DB["db"], + eventCollectors: Pick, + runRow: typeof workflowRun.$inferSelect, +): Promise { + const runId = runRow.id; const launchSpecRow = await db.query.workflowRunLaunchSpec.findFirst({ where: eq(workflowRunLaunchSpec.anchorRunId, runId), @@ -59,52 +73,89 @@ export async function ensureRunSession(params: { .set({ principalId: runRow.principalId, updatedAt: new Date() }) .where(eq(agentSession.id, sessionId)); } - } - if (runRow.address !== null && !eventCollectors.has(runRow.address)) { - eventCollectors.create(runRow.address, runRow.tenantId, sessionId, runRow.id); + if (runRow.address !== null && !eventCollectors.has(runRow.address)) { + eventCollectors.create(runRow.address, runRow.tenantId, sessionId, runRow.id); + } } return sessionId; } +const MAX_TRACKED_ADDRESSES = 1000; + +function track(set: Set, address: string): void { + if (set.size >= MAX_TRACKED_ADDRESSES) { + const oldest = set.values().next().value; + if (oldest !== undefined) set.delete(oldest); + } + set.add(address); +} + /** Creates a run's event collector on its first inference event instead of * its first outbound mail, so the first reply's parts are not dropped. Events - * for an address queue behind its in-flight ensure to keep their order. */ + * for an address queue behind its in-flight ensure to keep their order. A + * collector is never created for an address whose collector already ended + * (terminal event or abandon), nor before the run has a session row. */ export function withLazyRunCollector< - R extends Pick, + R extends Pick, >(registry: R, db: DB["db"]): R { const pending = new Map>(); + const closed = new Set(); + const notRuns = new Set(); async function ensureForAddress(address: string): Promise { + if (closed.has(address) || notRuns.has(address)) return; try { const runRow = await db.query.workflowRun.findFirst({ where: eq(workflowRun.address, address), }); - if (runRow !== undefined) { - await ensureRunSession({ db, eventCollectors: registry, runId: runRow.id }); + if (runRow === undefined) { + track(notRuns, address); + return; } + if (!isLiveWorkflowRunStatus(runRow.status)) return; + await ensureSessionForRun( + db, + { + has: registry.has, + create: (...args) => { + if (!closed.has(address)) registry.create(...args); + }, + }, + runRow, + ); } catch (err) { - reportError(err, { - operation: "workflows.lazyRunCollector.ensure", - extra: { address }, - }); + reportError(err, { operation: "workflows.lazyRunCollector.ensure", extra: { address } }); } } const dispatch: R["dispatch"] = (address, event) => { + const terminal = + event.type === "reactor.done" || (event.type === "reactor.error" && event.data.fatal); const inFlight = pending.get(address); - if (inFlight === undefined && (registry.has(address) || event.type === "reactor.done")) { + if (inFlight === undefined && (registry.has(address) || terminal)) { + if (terminal) track(closed, address); registry.dispatch(address, event); return; } - const next = (inFlight ?? ensureForAddress(address)).then(() => { - registry.dispatch(address, event); - }); + if (terminal) track(closed, address); + const next = (inFlight ?? ensureForAddress(address)) + .then(() => { + registry.dispatch(address, event); + }) + .catch((err: unknown) => { + reportError(err, { operation: "workflows.lazyRunCollector.dispatch", extra: { address } }); + }); pending.set(address, next); void next.finally(() => { if (pending.get(address) === next) pending.delete(address); }); }; - return { ...registry, dispatch }; + const abandon: R["abandon"] = (address) => { + track(closed, address); + registry.abandon(address); + }; + + return { ...registry, dispatch, abandon }; } From f34b19219cf6ae30eef3732b1fe7fe606051c910 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 16:01:06 -0700 Subject: [PATCH 5/6] workflows: close a run's collector only after its terminal event is delivered (CL-9907) --- e2e/workflows/first-reply-session.test.ts | 56 ++++++++++++++++--- .../workflows/src/launch/agent-session.ts | 52 +++++++---------- 2 files changed, 67 insertions(+), 41 deletions(-) diff --git a/e2e/workflows/first-reply-session.test.ts b/e2e/workflows/first-reply-session.test.ts index 59544fafd..368092963 100644 --- a/e2e/workflows/first-reply-session.test.ts +++ b/e2e/workflows/first-reply-session.test.ts @@ -10,7 +10,10 @@ import { createEventCollectorRegistry } from "@intx/hub-sessions"; import { dbGate } from "../lib/db-gate"; import { createHubPersistMailWithSessionEnsure } from "../../apps/hub/src/mailbox-persist"; -import { withLazyRunCollector } from "../../packages/workflows/src/launch/agent-session"; +import { + ensureRunSession, + withLazyRunCollector, +} from "../../packages/workflows/src/launch/agent-session"; function dbConfigFromUrl(databaseUrl: string) { const url = new URL(databaseUrl); @@ -42,9 +45,13 @@ describeIfDb("first reply of a run", () => { const closedRunId = generateId("workflowRun"); const closedSessionId = generateId("session"); const closedAddress = `${closedRunId}@${domain}`; + const doneRunId = generateId("workflowRun"); + const doneSessionId = generateId("session"); + const doneAddress = `${doneRunId}@${domain}`; + const donePrincipalId = generateId("principal"); const closedPrincipalId = generateId("principal"); - const runIds = [runId, unanchoredRunId, closedRunId]; - const sessionIds = [sessionId, unanchoredSessionId, closedSessionId]; + const runIds = [runId, unanchoredRunId, closedRunId, doneRunId]; + const sessionIds = [sessionId, unanchoredSessionId, closedSessionId, doneSessionId]; const address = `${runId}@${domain}`; beforeAll(async () => { @@ -62,6 +69,7 @@ describeIfDb("first reply of a run", () => { await db.db.insert(schema.principal).values([ { id: userPrincipalId, tenantId, kind: "user", refId: "usr_test", status: "active" }, { id: runPrincipalId, tenantId, kind: "workflow", refId: runId, status: "active" }, + { id: donePrincipalId, tenantId, kind: "workflow", refId: doneRunId, status: "active" }, { id: closedPrincipalId, tenantId, kind: "workflow", refId: closedRunId, status: "active" }, ]); await db.db.insert(schema.workflowDefinition).values({ @@ -99,6 +107,7 @@ describeIfDb("first reply of a run", () => { }; await seedRun(runId, sessionId, runPrincipalId, address); await seedRun(unanchoredRunId, unanchoredSessionId, null, unanchoredAddress); + await seedRun(doneRunId, doneSessionId, donePrincipalId, doneAddress); await seedRun(closedRunId, closedSessionId, closedPrincipalId, closedAddress); }); @@ -115,10 +124,20 @@ describeIfDb("first reply of a run", () => { .where(eq(schema.workflowDefinition.id, definitionId)); await db.db .delete(schema.principal) - .where(inArray(schema.principal.id, [runPrincipalId, closedPrincipalId, userPrincipalId])); + .where( + inArray(schema.principal.id, [ + runPrincipalId, + closedPrincipalId, + donePrincipalId, + userPrincipalId, + ]), + ); await db.db.delete(schema.tenant).where(eq(schema.tenant.id, tenantId)); }); + async function waitFor(done: () => boolean | Promise): Promise { + for (let i = 0; i < 50 && !(await done()); i++) await Bun.sleep(20); + } const start = (seq: number) => ({ type: "inference.start", seq, data: { model: "m" } }) as const; const turnsFor = (session: string) => db.db.query.inferenceTurn.findMany({ where: eq(schema.inferenceTurn.sessionId, session) }); @@ -173,17 +192,36 @@ describeIfDb("first reply of a run", () => { const lazy = withLazyRunCollector(createEventCollectorRegistry({ db: db.db }), db.db); lazy.dispatch(closedAddress, start(1)); - await Bun.sleep(300); - expect(lazy.has(closedAddress)).toBe(true); - + await waitFor(() => lazy.has(closedAddress)); lazy.abandon(closedAddress); - await Bun.sleep(300); lazy.dispatch(closedAddress, start(2)); - await Bun.sleep(300); + await Bun.sleep(100); expect(lazy.has(closedAddress)).toBe(false); }); + test("keeps the reply of a run whose done follows its start immediately", async () => { + const lazy = withLazyRunCollector(createEventCollectorRegistry({ db: db.db }), db.db); + + lazy.dispatch(doneAddress, start(1)); + lazy.dispatch(doneAddress, { type: "reactor.done", seq: 2, data: {} }); + await waitFor(async () => (await turnsFor(doneSessionId)).length === 1); + + expect(await turnsFor(doneSessionId)).toHaveLength(1); + }); + + test("does not recreate the collector for mail after the run is done", async () => { + const lazy = withLazyRunCollector(createEventCollectorRegistry({ db: db.db }), db.db); + + lazy.dispatch(doneAddress, start(3)); + lazy.dispatch(doneAddress, { type: "reactor.done", seq: 4, data: {} }); + await waitFor(async () => (await turnsFor(doneSessionId)).length === 2); + await Bun.sleep(100); + await ensureRunSession({ db: db.db, eventCollectors: lazy, runId: doneRunId }); + + expect(lazy.has(doneAddress)).toBe(false); + }); + test("persists the run's first outbound mail although its session is created mid-call", async () => { await db.db.delete(schema.agentSession).where(eq(schema.agentSession.id, sessionId)); const upstreamCalls: unknown[] = []; diff --git a/packages/workflows/src/launch/agent-session.ts b/packages/workflows/src/launch/agent-session.ts index 2a3e08769..fe4e2d2ae 100644 --- a/packages/workflows/src/launch/agent-session.ts +++ b/packages/workflows/src/launch/agent-session.ts @@ -93,37 +93,27 @@ function track(set: Set, address: string): void { /** Creates a run's event collector on its first inference event instead of * its first outbound mail, so the first reply's parts are not dropped. Events - * for an address queue behind its in-flight ensure to keep their order. A - * collector is never created for an address whose collector already ended - * (terminal event or abandon), nor before the run has a session row. */ + * for an address queue behind its in-flight ensure to keep their order. Once + * an address's collector ended (terminal event or abandon), `create` ignores + * it, so no path (lazy dispatch or mail) can resurrect one. */ export function withLazyRunCollector< R extends Pick, >(registry: R, db: DB["db"]): R { const pending = new Map>(); const closed = new Set(); - const notRuns = new Set(); + + const create: R["create"] = (address, ...rest) => { + if (!closed.has(address)) registry.create(address, ...rest); + }; async function ensureForAddress(address: string): Promise { - if (closed.has(address) || notRuns.has(address)) return; + if (closed.has(address)) return; try { const runRow = await db.query.workflowRun.findFirst({ where: eq(workflowRun.address, address), }); - if (runRow === undefined) { - track(notRuns, address); - return; - } - if (!isLiveWorkflowRunStatus(runRow.status)) return; - await ensureSessionForRun( - db, - { - has: registry.has, - create: (...args) => { - if (!closed.has(address)) registry.create(...args); - }, - }, - runRow, - ); + if (runRow === undefined || !isLiveWorkflowRunStatus(runRow.status)) return; + await ensureSessionForRun(db, { has: registry.has, create }, runRow); } catch (err) { reportError(err, { operation: "workflows.lazyRunCollector.ensure", extra: { address } }); } @@ -132,20 +122,18 @@ export function withLazyRunCollector< const dispatch: R["dispatch"] = (address, event) => { const terminal = event.type === "reactor.done" || (event.type === "reactor.error" && event.data.fatal); - const inFlight = pending.get(address); - if (inFlight === undefined && (registry.has(address) || terminal)) { - if (terminal) track(closed, address); + const deliver = () => { registry.dispatch(address, event); + if (terminal) track(closed, address); + }; + const inFlight = pending.get(address); + if (inFlight === undefined && (registry.has(address) || closed.has(address))) { + deliver(); return; } - if (terminal) track(closed, address); - const next = (inFlight ?? ensureForAddress(address)) - .then(() => { - registry.dispatch(address, event); - }) - .catch((err: unknown) => { - reportError(err, { operation: "workflows.lazyRunCollector.dispatch", extra: { address } }); - }); + const next = (inFlight ?? ensureForAddress(address)).then(deliver).catch((err: unknown) => { + reportError(err, { operation: "workflows.lazyRunCollector.dispatch", extra: { address } }); + }); pending.set(address, next); void next.finally(() => { if (pending.get(address) === next) pending.delete(address); @@ -157,5 +145,5 @@ export function withLazyRunCollector< registry.abandon(address); }; - return { ...registry, dispatch, abandon }; + return { ...registry, create, dispatch, abandon }; } From 9ce1c67c311285dfa9b518e88c35909554c3bb33 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 16:03:12 -0700 Subject: [PATCH 6/6] workflows: only a terminal event closes a run's collector, not a disconnect abandon (CL-9907) --- e2e/workflows/first-reply-session.test.ts | 8 +++++--- packages/workflows/src/launch/agent-session.ts | 14 +++++--------- 2 files changed, 10 insertions(+), 12 deletions(-) diff --git a/e2e/workflows/first-reply-session.test.ts b/e2e/workflows/first-reply-session.test.ts index 368092963..864a6dec2 100644 --- a/e2e/workflows/first-reply-session.test.ts +++ b/e2e/workflows/first-reply-session.test.ts @@ -188,16 +188,18 @@ describeIfDb("first reply of a run", () => { await db.db.delete(schema.principal).where(eq(schema.principal.id, principal!.id)); }); - test("does not resurrect a collector after abandon", async () => { + test("recreates the collector after an abandon so a reconnected run keeps persisting", async () => { const lazy = withLazyRunCollector(createEventCollectorRegistry({ db: db.db }), db.db); lazy.dispatch(closedAddress, start(1)); await waitFor(() => lazy.has(closedAddress)); lazy.abandon(closedAddress); lazy.dispatch(closedAddress, start(2)); - await Bun.sleep(100); + await waitFor(() => lazy.has(closedAddress)); - expect(lazy.has(closedAddress)).toBe(false); + expect(lazy.has(closedAddress)).toBe(true); + await waitFor(async () => (await turnsFor(closedSessionId)).length === 2); + expect(await turnsFor(closedSessionId)).toHaveLength(2); }); test("keeps the reply of a run whose done follows its start immediately", async () => { diff --git a/packages/workflows/src/launch/agent-session.ts b/packages/workflows/src/launch/agent-session.ts index fe4e2d2ae..539af23fe 100644 --- a/packages/workflows/src/launch/agent-session.ts +++ b/packages/workflows/src/launch/agent-session.ts @@ -94,10 +94,11 @@ function track(set: Set, address: string): void { /** Creates a run's event collector on its first inference event instead of * its first outbound mail, so the first reply's parts are not dropped. Events * for an address queue behind its in-flight ensure to keep their order. Once - * an address's collector ended (terminal event or abandon), `create` ignores - * it, so no path (lazy dispatch or mail) can resurrect one. */ + * a run delivered its terminal event, `create` ignores its address, so no path + * (lazy dispatch or mail) can resurrect one. A disconnect's abandon is not + * terminal: the next event recreates the collector. */ export function withLazyRunCollector< - R extends Pick, + R extends Pick, >(registry: R, db: DB["db"]): R { const pending = new Map>(); const closed = new Set(); @@ -140,10 +141,5 @@ export function withLazyRunCollector< }); }; - const abandon: R["abandon"] = (address) => { - track(closed, address); - registry.abandon(address); - }; - - return { ...registry, create, dispatch, abandon }; + return { ...registry, create, dispatch }; }