diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index bef0a6d6c..c3352c5a3 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -2017,6 +2017,15 @@ export async function createHub(config: HubConfig) { } try { + // Not `recordAgentSessionForRun` here: this mints a deployment + // with no message sent against it — a `workflow_run` born this + // way carries no principal until its first trigger (a call this + // deployer never makes) reconciles one onto it, so the call + // would only ever no-op. That trigger runs entirely through + // Interchange's own native route, with no Workbench-owned hook + // to record the session afterward (CL-7477 tracks the chat, + // webhook, and one-shot-prompt launch paths, which all send a + // message of their own and record there instead). const prepared = await workflowAllocationService.prepareProvisionedDeployment({ tenantId, diff --git a/packages/agent-directory/src/one-shot-prompt.ts b/packages/agent-directory/src/one-shot-prompt.ts index bf742b2d3..f3084ea66 100644 --- a/packages/agent-directory/src/one-shot-prompt.ts +++ b/packages/agent-directory/src/one-shot-prompt.ts @@ -8,8 +8,10 @@ import type { AgentLifecycle } from "@corbits/agent-lifecycle"; import { connectorReplyContent, messageRunEnded } from "@corbits/agent-events"; import { reportError } from "@corbits/error-sink"; import { + endAgentSessionForRun, readDefinitionProjection, readFoldedBody, + recordAgentSessionForRun, WORKFLOW_SOURCE_ENTRY, } from "@corbits/workflows"; import { listVisibleOfferings, type DB } from "@intx/db"; @@ -162,6 +164,10 @@ async function provisionOnAsset( ? { toolPackagePins: input.foldedBody.toolPackagePins } : {}), }); + // Not `recordAgentSessionForRun` here: a freshly provisioned run's + // `workflow_run.principal_id` is still null until its first trigger + // reconciles one onto it — `runOneShotPrompt` records the session + // right after the opening prompt actually sends. return { runId: prepared.anchorRunId, address: prepared.deploymentAddress, @@ -260,6 +266,14 @@ export async function runOneShotPrompt( settled = true; clearTimeout(timer); unsubscribe(); + try { + await endAgentSessionForRun(deps.db, launched.runId); + } catch (err) { + reportError(err, { + operation: "agent-directory.one-shot.end-session", + extra: { runId: launched.runId, reason }, + }); + } try { await deps.undeploy(launched.address, reason); } catch (err) { @@ -290,18 +304,33 @@ export async function runOneShotPrompt( content: input.prompt, cryptoProvider, }); - return; + } else { + await deps.sessionService.sendUserMessage({ + agentAddress: launched.address, + from: `${input.principalId}@${tenantRow.domain}`, + messageId: `<${crypto.randomUUID()}@${tenantRow.domain}>`, + date: new Date(), + content: input.prompt, + sessionId: launched.sessionId, + tenantId: input.tenantId, + cryptoProvider, + }); + } + // The send above just drove the run's trigger path, the only + // thing that ever reconciles a principal onto a freshly + // 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, + }); + } catch (err) { + reportError(err, { + operation: "agent-directory.one-shot.recordAgentSession", + extra: { address: launched.address }, + }); } - await deps.sessionService.sendUserMessage({ - agentAddress: launched.address, - from: `${input.principalId}@${tenantRow.domain}`, - messageId: `<${crypto.randomUUID()}@${tenantRow.domain}>`, - date: new Date(), - content: input.prompt, - sessionId: launched.sessionId, - tenantId: input.tenantId, - cryptoProvider, - }); } catch (cause) { reportError(cause, { operation: "agent-directory.one-shot.send", diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index 39e52ff2c..5a432c5d7 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -16,6 +16,7 @@ import { and, desc, eq } from "drizzle-orm"; import { authoredDefinitionCandidates, DefinitionProjectionMissingError, + endAgentSessionForPrincipal, readFoldedBody, resolveNewestProjectedDefinition, WORKFLOW_SOURCE_ENTRY, @@ -342,6 +343,11 @@ export function createHubChatPlatform( ? { toolPackagePins: input.foldedBody.toolPackagePins } : {}), }); + // Not `recordAgentSessionForRun` here: a freshly provisioned run's + // `workflow_run.principal_id` is still null (an invite sits + // un-triggered until someone actually writes into it), so this + // would only ever no-op. `sendRunMail` records it instead, once the + // run's first turn has actually reconciled a principal onto it. return { runId: prepared.anchorRunId, address: prepared.deploymentAddress, @@ -470,6 +476,10 @@ export function createHubChatPlatform( foldedBody: binding.foldedBody, }); + if (run.principalId !== null) { + await endAgentSessionForPrincipal(deps.db, run.principalId); + } + await repointBinding( deps.db, binding, diff --git a/packages/chat/src/send-run-mail.ts b/packages/chat/src/send-run-mail.ts index 909fefa09..e79a51e8f 100644 --- a/packages/chat/src/send-run-mail.ts +++ b/packages/chat/src/send-run-mail.ts @@ -1,5 +1,7 @@ // 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 { reportError } from "@corbits/error-sink"; import { sessionMail } from "@intx/db/schema"; import type { DB } from "@intx/db"; import type { SessionService, SidecarRouter } from "@intx/hub-sessions"; @@ -100,5 +102,24 @@ export async function sendRunMail( const mailId = crypto.randomUUID(); const now = new Date(); const rawMIME = await deliverRunMailMIME(deps, params, mailId, now); + // `sendUserMessage` just drove this run's trigger path, which is the + // only thing that ever reconciles a principal onto a freshly + // provisioned `workflow_run` (CL-7477) — so the run is mail-routable + // now if it never was before. Best-effort and idempotent past the + // 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, + }); + } catch (err) { + reportError(err, { + operation: "chat.sendRunMail.recordAgentSession", + tenantId: params.tenantId, + agentId: params.agentAddress, + extra: { sessionId: params.sessionId }, + }); + } return recordRunMail(deps, params, mailId, now, rawMIME); } diff --git a/packages/webhook-triggers/src/launch.ts b/packages/webhook-triggers/src/launch.ts index 6f3b3adfa..6fda59774 100644 --- a/packages/webhook-triggers/src/launch.ts +++ b/packages/webhook-triggers/src/launch.ts @@ -17,6 +17,7 @@ import { deliverWhenRoutable, readDefinitionProjection, readFoldedBody, + recordAgentSessionForRun, WORKFLOW_SOURCE_ENTRY, } from "@corbits/workflows"; import { listVisibleOfferings, type DB } from "@intx/db"; @@ -237,6 +238,26 @@ export async function launchWebhookTrigger( }); } + // The delivery attempt above is what drove the run's trigger path, + // the only thing that ever reconciles a principal onto a freshly + // provisioned `workflow_run` (CL-7477) — recording only now, after + // that attempt, is what makes this call ever find one. Best-effort: + // a run that never became routable stays without a session until its + // first successful send. + try { + await recordAgentSessionForRun(deps.db, { + sessionId, + anchorRunId: prepared.anchorRunId, + }); + } catch (error) { + reportError(error, { + operation: "webhookTriggers.launch.recordAgentSession", + tenantId: trigger.tenantId, + agentId: prepared.deploymentAddress, + extra: { instanceId: prepared.anchorRunId }, + }); + } + return { instanceId: prepared.anchorRunId, triggerAddress: prepared.deploymentAddress, diff --git a/packages/webhook-triggers/test/launch.test.ts b/packages/webhook-triggers/test/launch.test.ts index fa7559086..177a5090b 100644 --- a/packages/webhook-triggers/test/launch.test.ts +++ b/packages/webhook-triggers/test/launch.test.ts @@ -88,7 +88,25 @@ function createFakeDb() { query: { workflowDefinition: { findFirst: async () => DEFINITION_ROW }, tenant: { findFirst: async () => TENANT_ROW }, + // `recordAgentSessionForRun` reads the just-provisioned run back by + // id — production never inserts this row itself (Interchange's + // `prepareProvisionedDeployment` mints the anchor row + // transactionally before returning it), so this models the same + // row for `INTERCHANGE_RUN_ID`. + workflowRun: { + findFirst: async () => ({ + id: INTERCHANGE_RUN_ID, + tenantId: TENANT_ROW.id, + definitionId: DEFINITION_ROW.id, + principalId: "prn_interchange", + }), + }, }, + insert: (_table: unknown) => ({ + values: (_values: unknown) => ({ + onConflictDoNothing: async (_opts: unknown) => {}, + }), + }), }; } diff --git a/packages/workflows/src/index.ts b/packages/workflows/src/index.ts index 92c132180..7eac44dbb 100644 --- a/packages/workflows/src/index.ts +++ b/packages/workflows/src/index.ts @@ -55,6 +55,12 @@ export { type ZonedParts, } from "./schedule/cron"; export { scheduleCronFromProjection } from "./schedule/from-projection"; +export { + recordAgentSessionForRun, + endAgentSessionForPrincipal, + endAgentSessionForRun, + type RecordAgentSessionParams, +} from "./launch/agent-session"; export { deliverWhenRoutable, isAgentUnreachableError, diff --git a/packages/workflows/src/launch/agent-session.ts b/packages/workflows/src/launch/agent-session.ts new file mode 100644 index 000000000..f95c53960 --- /dev/null +++ b/packages/workflows/src/launch/agent-session.ts @@ -0,0 +1,106 @@ +// Every Workbench launch path provisions through Interchange's +// `prepareProvisionedDeployment` with a `sessionId` it mints itself, but +// nothing ever wrote that id into `agent_session` (vendor-owned, +// `vendor/intx/db/src/schema/sessions.ts`) — the table +// `resolveRunSessionId` (`vendor/intx/hub-sessions/src/hub-session-lookups.ts`) +// reads to route a run's outbound mail. Before the folded-runs package +// was deleted it was the only writer of that table; every current +// launcher must now do this itself, through this one shared helper, so +// mail and spans persist against a real session from the run's first +// turn (CL-7477). +// +// Timing matters: `workflow_run.principal_id` (the FK `agent_session` +// keys on) is still null the instant `prepareProvisionedDeployment` +// returns — a provisioned anchor is born "deployed" with no principal, +// and only the run's first trigger reconciles one onto it (Interchange's +// `anchorWithPrincipal`, `vendor/intx/db/src/workflow-run-store.ts`), +// through the exact same trigger path every `sendUserMessage` drives. So +// this cannot run right after provisioning (an invite sits un-triggered +// until someone actually writes into it) — call it after a send that +// just delivered the run's first turn instead. A run with no principal +// yet is a normal, expected state, not a bug: this returns `false` +// rather than throwing, and idempotent past its first successful call. +import { eq } from "drizzle-orm"; +import type { DB } from "@intx/db"; +import { agentSession, workflowRun } from "@intx/db/schema"; + +export type RecordAgentSessionParams = { + /** The id `prepareProvisionedDeployment` was called with. */ + readonly sessionId: string; +} & ({ readonly anchorRunId: string } | { readonly address: string }); + +/** + * Inserts the `agent_session` row a run needs to be mail-routable: the + * run's `tenantId`, `definitionId` (the `agent_id` FK — the column keeps + * its old name but targets `workflow_definition` now), and `principalId` + * are all read fresh off the `workflow_run` row, never guessed by the + * caller. Returns `false` (and writes nothing) when the row has no + * principal yet, `true` once the session is recorded (or already was — + * a second call for the same `sessionId` is a no-op, not a conflict). + */ +export async function recordAgentSessionForRun( + db: DB["db"], + params: RecordAgentSessionParams, +): Promise { + const runRow = await db.query.workflowRun.findFirst({ + where: + "anchorRunId" in params + ? eq(workflowRun.id, params.anchorRunId) + : eq(workflowRun.address, params.address), + }); + if (runRow === undefined || runRow.principalId === null) { + return false; + } + + const now = new Date(); + await db + .insert(agentSession) + .values({ + id: params.sessionId, + tenantId: runRow.tenantId, + agentId: runRow.definitionId, + principalId: runRow.principalId, + status: "active", + createdAt: now, + updatedAt: now, + }) + .onConflictDoNothing({ target: agentSession.id }); + return true; +} + +/** + * Marks a run's `agent_session` ended — called only from the places that + * already know the run died: a chat relaunch (the old run's session) or + * a one-shot prompt's teardown. No sweeper; a run nobody reacts to stays + * `active` until something does. + */ +export async function endAgentSessionForPrincipal( + db: DB["db"], + principalId: string, +): Promise { + const endedAt = new Date(); + await db + .update(agentSession) + .set({ status: "ended", endedAt, updatedAt: endedAt }) + .where(eq(agentSession.principalId, principalId)); +} + +/** + * 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). + */ +export async function endAgentSessionForRun( + db: DB["db"], + anchorRunId: string, +): Promise { + const runRow = await db.query.workflowRun.findFirst({ + where: eq(workflowRun.id, anchorRunId), + }); + if (runRow === undefined || runRow.principalId === null) { + return; + } + await endAgentSessionForPrincipal(db, runRow.principalId); +} diff --git a/packages/workflows/test/agent-session.test.ts b/packages/workflows/test/agent-session.test.ts new file mode 100644 index 000000000..a7446adda --- /dev/null +++ b/packages/workflows/test/agent-session.test.ts @@ -0,0 +1,106 @@ +// DB-gated: skipped when DATABASE_URL is unreachable, matching this +// 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. +import { afterAll, beforeAll, expect, test } from "bun:test"; +import { eq } from "drizzle-orm"; + +import { createDB, schema, type DB } from "@intx/db"; +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"; + +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); + +describeIfDb("recordAgentSessionForRun", () => { + let db: DB; + + const tenantId = generateId("tenant"); + const principalId = generateId("principal"); + const definitionId = generateId("workflowDefinition"); + const anchorRunId = generateId("workflowRun"); + const sessionId = generateId("session"); + + beforeAll(async () => { + if (databaseUrl === undefined) return; + db = createDB(dbConfigFromUrl(databaseUrl)); + + await db.db.insert(schema.tenant).values({ + id: tenantId, + name: "Agent Session Test Tenant", + slug: `agent-session-${tenantId}`, + domain: `agent-session-${tenantId}.localhost`, + 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.workflowDefinition).values({ + id: definitionId, + tenantId, + creatorPrincipalId: principalId, + assetId: null, + name: "agent-session-test-definition", + }); + + await db.db.insert(schema.workflowRun).values({ + id: anchorRunId, + definitionId, + anchorRunId, + tenantId, + principalId, + address: `${anchorRunId}@agent-session-${tenantId}.localhost`, + status: "running", + }); + }); + + afterAll(async () => { + if (databaseUrl === undefined) return; + await db.db + .delete(schema.agentSession) + .where(eq(schema.agentSession.id, sessionId)); + await db.db + .delete(schema.workflowRun) + .where(eq(schema.workflowRun.id, anchorRunId)); + await db.db + .delete(schema.workflowDefinition) + .where(eq(schema.workflowDefinition.id, definitionId)); + await db.db + .delete(schema.principal) + .where(eq(schema.principal.id, principalId)); + 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 }); + + const resolved = await resolveRunSessionId(db.db, principalId); + expect(resolved).toBe(sessionId); + }); +});