diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index c3352c5a3..1faa70f3a 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -2272,6 +2272,7 @@ export async function createHub(config: HubConfig) { isRoutable: (address) => sidecarRouter.getRoutableAddresses().includes(address), cryptoProviderCache: cryptoProviders, + eventCollectors, persistLaunch: async (input) => { await workbenchLaunchPersistExtra(input)(db); }, @@ -3069,6 +3070,7 @@ export async function createHub(config: HubConfig) { repoStore: agentRepoStore.repoStore, workflowAllocationService, sessionService, + eventCollectors, }, runnerInput, ), diff --git a/packages/agent-directory/src/one-shot-prompt.ts b/packages/agent-directory/src/one-shot-prompt.ts index f3084ea66..36efb6022 100644 --- a/packages/agent-directory/src/one-shot-prompt.ts +++ b/packages/agent-directory/src/one-shot-prompt.ts @@ -13,6 +13,7 @@ import { readFoldedBody, recordAgentSessionForRun, WORKFLOW_SOURCE_ENTRY, + type EventCollectorPort, } from "@corbits/workflows"; import { listVisibleOfferings, type DB } from "@intx/db"; import { tenant as tenantTable, workflowDefinition } from "@intx/db/schema"; @@ -57,6 +58,7 @@ export type OneShotRunnerDeps = { "prepareProvisionedDeployment" >; readonly sessionService: Pick; + readonly eventCollectors: EventCollectorPort; /** * Test seam only. Production never sets these; they default to * Interchange `prepareProvisionedDeployment` and `sendUserMessage`. @@ -267,7 +269,11 @@ export async function runOneShotPrompt( clearTimeout(timer); unsubscribe(); try { - await endAgentSessionForRun(deps.db, launched.runId); + await endAgentSessionForRun( + deps.db, + launched.runId, + deps.eventCollectors, + ); } catch (err) { reportError(err, { operation: "agent-directory.one-shot.end-session", @@ -321,10 +327,14 @@ export async function runOneShotPrompt( // provisioned `workflow_run` (CL-7477) — recording only now is // what makes this ever find one. try { - await recordAgentSessionForRun(deps.db, { - sessionId: launched.sessionId, - anchorRunId: launched.runId, - }); + await recordAgentSessionForRun( + deps.db, + { + sessionId: launched.sessionId, + anchorRunId: launched.runId, + }, + deps.eventCollectors, + ); } catch (err) { reportError(err, { operation: "agent-directory.one-shot.recordAgentSession", diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index 5a432c5d7..8b820493d 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -16,7 +16,7 @@ import { and, desc, eq } from "drizzle-orm"; import { authoredDefinitionCandidates, DefinitionProjectionMissingError, - endAgentSessionForPrincipal, + endAgentSessionForRun, readFoldedBody, resolveNewestProjectedDefinition, WORKFLOW_SOURCE_ENTRY, @@ -258,6 +258,7 @@ export function createHubChatPlatform( db: deps.db, sessionService: deps.sessionService, sidecarRouter: deps.sidecarRouter, + eventCollectors: deps.eventCollectors, }; function offeringDigest(sourceOfferingIds: readonly string[]): string { @@ -476,9 +477,7 @@ export function createHubChatPlatform( foldedBody: binding.foldedBody, }); - if (run.principalId !== null) { - await endAgentSessionForPrincipal(deps.db, run.principalId); - } + await endAgentSessionForRun(deps.db, run.id, deps.eventCollectors); await repointBinding( deps.db, diff --git a/packages/chat/src/send-run-mail.ts b/packages/chat/src/send-run-mail.ts index e79a51e8f..b570a22d2 100644 --- a/packages/chat/src/send-run-mail.ts +++ b/packages/chat/src/send-run-mail.ts @@ -1,6 +1,9 @@ // Signs one inbound message via Interchange `sendUserMessage` and // records it on session_mail so the run's own live subscribers see it. -import { recordAgentSessionForRun } from "@corbits/workflows"; +import { + recordAgentSessionForRun, + type EventCollectorPort, +} from "@corbits/workflows"; import { reportError } from "@corbits/error-sink"; import { sessionMail } from "@intx/db/schema"; import type { DB } from "@intx/db"; @@ -11,6 +14,7 @@ export type RunMailDeps = { db: DB["db"]; sessionService: Pick; sidecarRouter: Pick; + eventCollectors: Pick; }; export type SendRunMailParams = { @@ -109,10 +113,14 @@ export async function sendRunMail( // first successful call; a failure here must not swallow the mail // that already sent. try { - await recordAgentSessionForRun(deps.db, { - sessionId: params.sessionId, - address: params.agentAddress, - }); + await recordAgentSessionForRun( + deps.db, + { + sessionId: params.sessionId, + address: params.agentAddress, + }, + deps.eventCollectors, + ); } catch (err) { reportError(err, { operation: "chat.sendRunMail.recordAgentSession", diff --git a/packages/webhook-triggers/src/launch.ts b/packages/webhook-triggers/src/launch.ts index 6fda59774..22e686d40 100644 --- a/packages/webhook-triggers/src/launch.ts +++ b/packages/webhook-triggers/src/launch.ts @@ -19,6 +19,7 @@ import { readFoldedBody, recordAgentSessionForRun, WORKFLOW_SOURCE_ENTRY, + type EventCollectorPort, } from "@corbits/workflows"; import { listVisibleOfferings, type DB } from "@intx/db"; import { tenant as tenantTable, workflowDefinition } from "@intx/db/schema"; @@ -58,6 +59,7 @@ export type LaunchWebhookTriggerDeps = { */ isRoutable: (address: string) => boolean; cryptoProviderCache: CryptoProviderCache; + eventCollectors: Pick; /** * Host-supplied cipher. Interchange decrypts bindings inside * `prepareProvisionedDeployment`; this port still owns the field so @@ -245,10 +247,14 @@ export async function launchWebhookTrigger( // a run that never became routable stays without a session until its // first successful send. try { - await recordAgentSessionForRun(deps.db, { - sessionId, - anchorRunId: prepared.anchorRunId, - }); + await recordAgentSessionForRun( + deps.db, + { + sessionId, + anchorRunId: prepared.anchorRunId, + }, + deps.eventCollectors, + ); } catch (error) { reportError(error, { operation: "webhookTriggers.launch.recordAgentSession", diff --git a/packages/webhook-triggers/test/launch.test.ts b/packages/webhook-triggers/test/launch.test.ts index 177a5090b..01b3b6785 100644 --- a/packages/webhook-triggers/test/launch.test.ts +++ b/packages/webhook-triggers/test/launch.test.ts @@ -197,6 +197,9 @@ function baseDeps() { }, }, isRoutable: () => isRoutableForTest, + eventCollectors: { + create: () => undefined, + }, }; } diff --git a/packages/workflows/src/index.ts b/packages/workflows/src/index.ts index 7eac44dbb..244d0478a 100644 --- a/packages/workflows/src/index.ts +++ b/packages/workflows/src/index.ts @@ -60,6 +60,7 @@ export { endAgentSessionForPrincipal, endAgentSessionForRun, type RecordAgentSessionParams, + type EventCollectorPort, } from "./launch/agent-session"; export { deliverWhenRoutable, diff --git a/packages/workflows/src/launch/agent-session.ts b/packages/workflows/src/launch/agent-session.ts index f95c53960..58ec8fd40 100644 --- a/packages/workflows/src/launch/agent-session.ts +++ b/packages/workflows/src/launch/agent-session.ts @@ -23,12 +23,27 @@ import { eq } from "drizzle-orm"; import type { DB } from "@intx/db"; import { agentSession, workflowRun } from "@intx/db/schema"; +import type { EventCollectorRegistry } from "@intx/hub-sessions"; export type RecordAgentSessionParams = { /** The id `prepareProvisionedDeployment` was called with. */ readonly sessionId: string; } & ({ readonly anchorRunId: string } | { readonly address: string }); +/** + * The one port every launcher threads the same wrapped + * `EventCollectorRegistry` through (`apps/hub/src/index.ts`'s + * `eventCollectors`) — never a second registry construction. `create` + * is what actually makes `inference_turn`/`turn_part` rows exist for a + * run (CL-7479: nothing else ever called it once `folded-runs` was + * deleted); `abandon` tears the collector down when a run's session + * ends. + */ +export type EventCollectorPort = Pick< + EventCollectorRegistry, + "create" | "abandon" +>; + /** * Inserts the `agent_session` row a run needs to be mail-routable: the * run's `tenantId`, `definitionId` (the `agent_id` FK — the column keeps @@ -41,6 +56,7 @@ export type RecordAgentSessionParams = { export async function recordAgentSessionForRun( db: DB["db"], params: RecordAgentSessionParams, + eventCollectors: Pick, ): Promise { const runRow = await db.query.workflowRun.findFirst({ where: @@ -65,6 +81,19 @@ export async function recordAgentSessionForRun( updatedAt: now, }) .onConflictDoNothing({ target: agentSession.id }); + + const agentAddress = "address" in params ? params.address : runRow.address; + if (agentAddress !== null) { + // Idempotent across wakes: a collector already live for this + // address is replaced, matching the registry's own contract + // (`event-collector-registry.ts`'s `create`). + eventCollectors.create( + agentAddress, + runRow.tenantId, + params.sessionId, + runRow.id, + ); + } return true; } @@ -87,14 +116,19 @@ export async function endAgentSessionForPrincipal( /** * Same as `endAgentSessionForPrincipal`, keyed by the run's own id - * instead — for a caller (e.g. a one-shot prompt's teardown) that only - * ever held the run id, never minted or looked up its principal. - * No-ops for a run row already gone, or one never triggered (no - * principal, hence no session was ever recorded for it). + * instead — for a caller (e.g. a one-shot prompt's teardown, or a chat + * relaunch replacing a terminal run) that only ever held the run id, + * never minted or looked up its principal. Also abandons the outgoing + * run's event collector, through the same shared + * `EventCollectorPort` every launcher threads — one mechanism for + * ending a run's session. No-ops for a run row already gone, or one + * never triggered (no principal, hence no session was ever recorded + * for it). */ export async function endAgentSessionForRun( db: DB["db"], anchorRunId: string, + eventCollectors: Pick, ): Promise { const runRow = await db.query.workflowRun.findFirst({ where: eq(workflowRun.id, anchorRunId), @@ -103,4 +137,7 @@ export async function endAgentSessionForRun( return; } await endAgentSessionForPrincipal(db, runRow.principalId); + if (runRow.address !== null) { + eventCollectors.abandon(runRow.address); + } } diff --git a/packages/workflows/test/agent-session.test.ts b/packages/workflows/test/agent-session.test.ts index a7446adda..3ae50e221 100644 --- a/packages/workflows/test/agent-session.test.ts +++ b/packages/workflows/test/agent-session.test.ts @@ -15,7 +15,10 @@ import { generateId } from "@intx/hub-common"; import { resolveRunSessionId } from "@intx/hub-sessions"; import { dbGate } from "../../../scripts/e2e/db-gate"; -import { recordAgentSessionForRun } from "../src/launch/agent-session"; +import { + endAgentSessionForRun, + recordAgentSessionForRun, +} from "../src/launch/agent-session"; function dbConfigFromUrl(databaseUrl: string) { const url = new URL(databaseUrl); @@ -98,9 +101,55 @@ describeIfDb("recordAgentSessionForRun", () => { }); test("writes an agent_session resolveRunSessionId reads back", async () => { - await recordAgentSessionForRun(db.db, { sessionId, anchorRunId }); + 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 () => { + const createCalls: unknown[] = []; + const otherSessionId = generateId("session"); + + await recordAgentSessionForRun( + db.db, + { sessionId: otherSessionId, anchorRunId }, + { + create: (...args: unknown[]) => { + createCalls.push(args); + }, + }, + ); + + expect(createCalls).toEqual([ + [ + `${anchorRunId}@agent-session-${tenantId}.localhost`, + tenantId, + otherSessionId, + anchorRunId, + ], + ]); + + await db.db + .delete(schema.agentSession) + .where(eq(schema.agentSession.id, otherSessionId)); + }); + + test("abandons the run's event collector when the session ends", async () => { + const abandonCalls: string[] = []; + + await endAgentSessionForRun(db.db, anchorRunId, { + abandon: (address: string) => { + abandonCalls.push(address); + }, + }); + + expect(abandonCalls).toEqual([ + `${anchorRunId}@agent-session-${tenantId}.localhost`, + ]); + }); });