diff --git a/packages/chat/src/agent-dm-mode.ts b/packages/chat/src/agent-dm-mode.ts new file mode 100644 index 0000000000..6e63e19d66 --- /dev/null +++ b/packages/chat/src/agent-dm-mode.ts @@ -0,0 +1,49 @@ +// CL-7108: the agent-DM pin — the ONE place that says what an agent DM +// is. An agent DM is a `kind: "chat"` conversation carrying a +// `chat/definitionId`: the one 1:1 (tenant, agent) conversation, minted by +// `mintAgentDm` and found-or-reopened by `findExistingAgentChat` (see +// `./workbench-service.ts`) rather than invited like a room agent. +// +// Why DMs differ from rooms is chat-layer only — the execution plane is +// identical. A DM's agent runs the same interactive warm-mailbox shape a +// room agent does: one standing per-agent run, provisioned once (at mint, +// ahead of the member's first message) from the conversation's +// single-definition asset, every inbound mail a header-threaded turn on +// that run, the turn rows a durable per-run INBOX verifiable in the hub +// replica, and the agent's `mail_wait` wired in the one place the +// execution plane parks (the run child's watch registry — the chat layer +// never parks awaiting agent mail; dispatch is fire-and-forget and the +// agent-turns projection plus the turn-mail correlation carry the trail). +// What makes a DM a DM lives entirely here at the chat layer: the 1:1 +// identity this pin names, find-or-reopen instead of invite, the sidebar +// bucket, the greeting, mint-not-invite. +// +// Every settings-level reader and writer of DM-ness goes through this +// module's keys and predicate — never a second inline `chat/definitionId` +// literal beside it — so the pin cannot drift between mint, reopen, the +// workbench view, and the launch gate. +export const AGENT_DM_KIND = "chat"; + +export const AGENT_DM_DEFINITION_ID_KEY = "chat/definitionId"; + +/** The agent this DM was minted for, or `undefined` when the settings + * carry no (string) definition id. Non-string values are not a DM's agent + * — validation at the trust boundary, not a fallback path. */ +export function definitionIdOfSettings( + settings: Record, +): string | undefined { + const value = settings[AGENT_DM_DEFINITION_ID_KEY]; + return typeof value === "string" ? value : undefined; +} + +/** Whether these settings are an agent DM: kind `chat` (a missing kind + * reads as `chat`, mirroring `kindOf` in `./workbench-settings.ts`) with + * a definition id naming its agent. */ +export function isAgentDmSettings(settings: Record): boolean { + const kind = settings["chat/kind"]; + const effectiveKind = typeof kind === "string" ? kind : AGENT_DM_KIND; + return ( + effectiveKind === AGENT_DM_KIND && + definitionIdOfSettings(settings) !== undefined + ); +} diff --git a/packages/chat/src/index.ts b/packages/chat/src/index.ts index ac5f1935a5..bd000ce4ee 100644 --- a/packages/chat/src/index.ts +++ b/packages/chat/src/index.ts @@ -129,10 +129,13 @@ export { createInMemoryTurnClaimStore, } from "./turn-claims"; export type { TurnClaim, TurnClaimStore, TurnClaimToken } from "./turn-claims"; +export { workbenchLaunchPersistExtra } from "./standalone-launch"; export { - AGENT_SECTION_MODE, - workbenchLaunchPersistExtra, -} from "./standalone-launch"; + AGENT_DM_DEFINITION_ID_KEY, + AGENT_DM_KIND, + definitionIdOfSettings, + isAgentDmSettings, +} from "./agent-dm-mode"; export { recordSourcesDigest } from "./agent-binding"; export { createWorkbenchTurnQueue, TurnQueuedEvent } from "./turn-queue"; export type { diff --git a/packages/chat/src/routes.ts b/packages/chat/src/routes.ts index 92c1cd5955..58cf3f2bb9 100644 --- a/packages/chat/src/routes.ts +++ b/packages/chat/src/routes.ts @@ -64,6 +64,7 @@ import { validateSettingsPatch, visibilityOf, } from "./workbench-settings"; +import { AGENT_DM_DEFINITION_ID_KEY } from "./agent-dm-mode"; import { listWorkbenchLiveState } from "./workbench-reply-activity"; import { postRoomMessage, type RoomMessageStore } from "./room-messages"; import { WorkbenchOnboardingStep } from "./blocks"; @@ -1191,7 +1192,7 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono { const withDefinitionId: Record = isChatWithDefinition( body, ) - ? { ...baseSettings, "chat/definitionId": body.definitionId } + ? { ...baseSettings, [AGENT_DM_DEFINITION_ID_KEY]: body.definitionId } : baseSettings; const settings: Record = chatTitle !== undefined diff --git a/packages/chat/src/standalone-launch.test.ts b/packages/chat/src/standalone-launch.test.ts index d6a1c6efbd..4f69922a4f 100644 --- a/packages/chat/src/standalone-launch.test.ts +++ b/packages/chat/src/standalone-launch.test.ts @@ -4,11 +4,7 @@ import { describe, expect, test } from "bun:test"; import { workbenchLaunch } from "./schema"; -import { - AGENT_SECTION_MODE, - workbenchLaunchPersistExtra, -} from "./standalone-launch"; -import { CHAT_TURN_TIMEOUT_MS } from "./turn-claims"; +import { workbenchLaunchPersistExtra } from "./standalone-launch"; const FOLDED_BODY = { systemPrompt: "be helpful", @@ -18,15 +14,6 @@ const FOLDED_BODY = { model: "claude-sonnet-5", }; -describe("AGENT_SECTION_MODE", () => { - test("is the onTrigger section shape with the chat turn timeout", () => { - expect(AGENT_SECTION_MODE).toEqual({ - kind: "section", - turnTimeoutMs: CHAT_TURN_TIMEOUT_MS, - }); - }); -}); - describe("workbenchLaunchPersistExtra", () => { test("writes the identity mapping into workbench_launch", async () => { const written: { table: unknown; values: unknown }[] = []; diff --git a/packages/chat/src/standalone-launch.ts b/packages/chat/src/standalone-launch.ts index 5df9518141..74e09aa90b 100644 --- a/packages/chat/src/standalone-launch.ts +++ b/packages/chat/src/standalone-launch.ts @@ -9,32 +9,11 @@ // it, the wake path cannot resolve it, and its next occurrence 409s // (`workflow_run_terminal`) forever. These exports are what a standalone // launcher passes so its run rides the exact same relaunch path a -// room-invited agent does: the section mode the deploy pins, and the -// `persistExtra` that writes the mapping row inside the launch -// transaction itself. +// room-invited agent does: the `persistExtra` that writes the mapping row +// inside the launch transaction itself. import type { DBExecutor } from "@intx/db"; import type { FoldedBody } from "@intx/workflow-deploy"; -import type { AgentRuntimeConfig } from "@corbits/agent-runtime"; import { workbenchLaunch } from "./schema"; -import { CHAT_TURN_TIMEOUT_MS } from "./turn-claims"; - -/** - * The shape every launched agent run deploys as — chat's room invites - * and standalone routine/webhook launches alike: an `onTrigger` section - * (CL-6329), one warm run per agent, each inbound message an occurrence - * running as its own child run (`turn__`) with its own event log. - * That child id is what a reply's `run_id` carries, which is the whole - * reason a reply is traceable. - * - * `onBodyFailure: "tolerate"` — authored in the section shape itself - * (`@corbits/agent-runtime`) — is the failure edge: a turn that throws - * records a failed occurrence and leaves the section subscribed, so one - * bad turn kills neither the agent nor the conversation. - */ -export const AGENT_SECTION_MODE: AgentRuntimeConfig["mode"] = { - kind: "section", - turnTimeoutMs: CHAT_TURN_TIMEOUT_MS, -}; /** * The `persistExtra` a standalone launch hands its provisioned deploy: diff --git a/packages/chat/src/workbench-service.ts b/packages/chat/src/workbench-service.ts index d465829e2c..95a0db4f83 100644 --- a/packages/chat/src/workbench-service.ts +++ b/packages/chat/src/workbench-service.ts @@ -43,6 +43,11 @@ import { participantsOf, resolveContextWindow, } from "./workbench-settings"; +import { + AGENT_DM_DEFINITION_ID_KEY, + AGENT_DM_KIND, + definitionIdOfSettings, +} from "./agent-dm-mode"; import { presetForKind } from "./kinds"; import type { WorkbenchLauncher, @@ -232,9 +237,11 @@ export type FindExistingAgentChatDeps = { * tenant with that agent. Uniqueness is per (bench, definitionId). * Product reopens; it does not clone. * - * Matches forward, by the `chat/definitionId` every agent chat has - * carried in its settings since this landed, and falls back to - * `matchesLegacyAgentChat` for a chat minted before that key existed. + * Matches forward, by the DM pin (`definitionIdOfSettings` in + * `./agent-dm-mode.ts` — the one place that says what an agent DM is) + * every agent chat has carried in its settings since this landed, and + * falls back to `matchesLegacyAgentChat` for a chat minted before that + * key existed. * The comparison is on the definition's ASSET, not the row id: a * code-sourced deploy projects a new `workflow_definition` row per * frozen wire projection, so the id a chat recorded at creation and the @@ -255,11 +262,10 @@ export async function findExistingAgentChat( const assetId = await deps.platform.resolveDefinitionAssetId(definitionId); const matches: { row: WorkbenchSettingsRow; createdAt: Date }[] = []; for (const row of chats) { - const storedDefinitionId = row.settings["chat/definitionId"]; + const storedDefinitionId = definitionIdOfSettings(row.settings); const isMatch = storedDefinitionId !== undefined - ? typeof storedDefinitionId === "string" && - (await sameAgent(deps, storedDefinitionId, definitionId, assetId)) + ? await sameAgent(deps, storedDefinitionId, definitionId, assetId) : await matchesLegacyAgentChat(deps, row, definitionId); if (!isMatch) continue; const link = await deps.tenancy.getWorkbenchTenancy(row.workbenchId); @@ -384,10 +390,10 @@ export async function mintAgentDm( const preset = presetForKind("chat"); const baseSettings: Record = { - "chat/kind": "chat", + "chat/kind": AGENT_DM_KIND, "chat/pinned": preset.pinned, "chat/participants": [], - "chat/definitionId": input.definitionId, + [AGENT_DM_DEFINITION_ID_KEY]: input.definitionId, }; const settings: Record = chatTitle !== undefined @@ -680,8 +686,8 @@ export async function launchAndJoinAgent( }; } - if (kindOf(input.existingSettings) === "chat") { - const boundDefinitionId = input.existingSettings["chat/definitionId"]; + if (kindOf(input.existingSettings) === AGENT_DM_KIND) { + const boundDefinitionId = definitionIdOfSettings(input.existingSettings); const alreadyHasAgent = participants.some((participant) => isAgentAddress(participant.address), ); @@ -1980,9 +1986,9 @@ export type DispatchTurnInput = { * Asks one agent for a turn — the seam between the room (rows on a * timeline) and the execution plane. * - * A room agent deploys as an `onTrigger` section (CL-6329, see - * `./standalone-launch.ts`'s `AGENT_SECTION_MODE`), so this fires that - * section's trigger: one occurrence, running as its own child run + * An agent runs as one warm folded run provisioned from the + * conversation's definition asset, so this fires that run's mail trigger: + * one occurrence, running as its own child run * (`turn__`) with its own event log, against the one warm run the * (agent, workbench) pair already holds. The trigger is a mail trigger — * that is the primitive the section subscribes on — so `sendMail` diff --git a/packages/chat/src/workbench-settings.ts b/packages/chat/src/workbench-settings.ts index 653cd94a47..a2342d5956 100644 --- a/packages/chat/src/workbench-settings.ts +++ b/packages/chat/src/workbench-settings.ts @@ -13,6 +13,7 @@ import { ParticipantsSetting, type ParticipantRecord, } from "./participants"; +import { definitionIdOfSettings } from "./agent-dm-mode"; const PatchSettingsBody = type("Record"); @@ -245,7 +246,7 @@ export function workbenchView(row: { const kind = kindOf(row.settings); const name = row.settings["chat/name"]; const pinned = row.settings["chat/pinned"]; - const definitionId = row.settings["chat/definitionId"]; + const definitionId = definitionIdOfSettings(row.settings); return { id: row.workbenchId, title: typeof name === "string" ? name : row.workbenchId, diff --git a/packages/chat/test/agent-dm-warm-mailbox.test.ts b/packages/chat/test/agent-dm-warm-mailbox.test.ts new file mode 100644 index 0000000000..5983258509 --- /dev/null +++ b/packages/chat/test/agent-dm-warm-mailbox.test.ts @@ -0,0 +1,203 @@ +// CL-7108: agent DMs run on the interactive warm-mailbox shape — one +// standing per-agent run provisioned from the conversation's definition +// asset, every inbound mail a header-threaded turn on that run, the turn +// rows a durable per-run INBOX. This test pins the shape at the chat +// layer: the DM settings pin (kind + definitionId, in exactly one place) +// and a two-turn DM conversation's durable trail (turn rows, RFC 5322 +// threading, turn-mail correlation, mailbox fan-out). +import { describe, expect, test } from "bun:test"; +import type { MailContent } from "../src/codec"; +import { + AGENT_DM_DEFINITION_ID_KEY, + AGENT_DM_KIND, + definitionIdOfSettings, + isAgentDmSettings, +} from "../src/agent-dm-mode"; +import { createInMemoryAgentTurnStore } from "../src/agent-turns"; +import { createInMemoryRoomMessageStore } from "../src/room-messages"; +import { createInMemoryChatStore } from "../src/store"; +import { createInMemoryThreadStore } from "../src/threads"; +import { createInMemoryTurnClaimStore } from "../src/turn-claims"; +import { createTurnCancelRegistry } from "../src/turn-cancellation"; +import { createInMemoryTurnMailCorrelationStore } from "../src/turn-mail-correlation"; +import { createWorkbenchTurnQueue } from "../src/turn-queue"; +import { sendWorkbenchMessage } from "../src/workbench-service"; +import type { MailboxWriter } from "../src/mailbox-fanout"; + +const TENANT_ID = "tnt_1"; +const WORKBENCH_ID = "wb_dm_1"; +const DOMAIN = "acme.example"; +const SENDER = "prn_alice"; +const AGENT_ADDRESS = "ins_echo1@acme.example"; +const DEFINITION_ID = "def_echo"; + +function dmSettings() { + return { + "chat/kind": "chat", + "chat/definitionId": DEFINITION_ID, + "chat/participants": [ + { address: SENDER, handle: SENDER }, + { address: AGENT_ADDRESS, handle: "echo" }, + ], + }; +} + +describe("agent DM pin (CL-7108)", () => { + test("DM settings are a kind:chat conversation with a definition id", () => { + const settings = dmSettings(); + expect(isAgentDmSettings(settings)).toBe(true); + expect(definitionIdOfSettings(settings)).toBe(DEFINITION_ID); + }); + + test("the pin names the exact settings keys on the wire", () => { + expect(AGENT_DM_KIND).toBe("chat"); + expect(AGENT_DM_DEFINITION_ID_KEY).toBe("chat/definitionId"); + }); + + test("a group conversation with a definition id is not a DM", () => { + expect(isAgentDmSettings({ ...dmSettings(), "chat/kind": "group" })).toBe( + false, + ); + }); + + test("a chat without a definition id is not a DM", () => { + const { "chat/definitionId": _dropped, ...rest } = dmSettings(); + expect(isAgentDmSettings(rest)).toBe(false); + expect(definitionIdOfSettings(rest)).toBeUndefined(); + }); + + test("a non-string definition id is not a DM", () => { + expect( + isAgentDmSettings({ ...dmSettings(), "chat/definitionId": 42 }), + ).toBe(false); + expect(definitionIdOfSettings({ "chat/definitionId": 42 })).toBeUndefined(); + }); + + test("a missing kind reads as chat, like the workbench view does", () => { + const { "chat/kind": _dropped, ...rest } = dmSettings(); + expect(isAgentDmSettings(rest)).toBe(true); + }); +}); + +function noopMailbox() { + const writer: MailboxWriter = { + async writeBatch(items) { + return items.map((_item, index) => ({ + messageKey: String(index), + id: `mbx_${String(index)}`, + })); + }, + }; + return { + writer, + resolveKnownPrincipalIds: async () => new Set([SENDER]), + resolveTenantDomain: async () => DOMAIN, + }; +} + +async function makeDmDeps(sentMail: MailContent[]) { + const store = createInMemoryChatStore(); + const roomMessages = createInMemoryRoomMessageStore(); + const threads = createInMemoryThreadStore(); + const claims = createInMemoryTurnClaimStore({ ttlMs: 60_000 }); + const turnQueue = createWorkbenchTurnQueue({ + claims, + publish: () => undefined, + }); + const turnCancellation = createTurnCancelRegistry(); + const agentTurns = createInMemoryAgentTurnStore(); + const turnMailCorrelation = createInMemoryTurnMailCorrelationStore(); + + await store.createWorkbenchSettings({ + tenantId: TENANT_ID, + workbenchId: WORKBENCH_ID, + updatedBy: SENDER, + settings: dmSettings(), + }); + + const deps = { + store, + roomMessages, + threads, + publish: () => undefined, + platform: { + async sendMail(input: { content: MailContent }) { + sentMail.push(input.content); + return { id: "mail_agent_1", createdAt: new Date().toISOString() }; + }, + }, + turnQueue, + turnCancellation, + mailbox: noopMailbox(), + agentTurns, + turnMailCorrelation, + }; + return { deps, agentTurns, turnMailCorrelation }; +} + +async function sendDm( + deps: Awaited>["deps"], + text: string, +) { + const posted = await sendWorkbenchMessage(deps, { + tenantId: TENANT_ID, + principalId: SENDER, + senderAddress: `${SENDER}@${DOMAIN}`, + workbenchId: WORKBENCH_ID, + messageParts: [{ kind: "text", text }], + }); + await posted.fanoutDelivered; + return posted; +} + +describe("agent DM conversation trail (CL-7108)", () => { + test("two DM turns each open a turn row, dispatch a threaded frame, and correlate", async () => { + const sentMail: MailContent[] = []; + const { deps, agentTurns, turnMailCorrelation } = + await makeDmDeps(sentMail); + + const first = await sendDm(deps, "hello"); + const running = await agentTurns.findRunningTurn({ + tenantId: TENANT_ID, + workbenchId: WORKBENCH_ID, + agentAddress: AGENT_ADDRESS, + }); + expect(running?.requestMessageIds).toEqual([first.id]); + // Settle the first turn the way the reply path would, so the second + // inbound serializes behind it into its own turn instead of piling + // onto a still-running one. + await agentTurns.finishTurn({ + tenantId: TENANT_ID, + turnId: running?.id ?? "", + status: "completed", + }); + + const second = await sendDm(deps, "again"); + + const turns = await agentTurns.listTurns({ + tenantId: TENANT_ID, + workbenchId: WORKBENCH_ID, + }); + expect(turns.map((turn) => turn.requestMessageIds)).toEqual([ + [second.id], + [first.id], + ]); + + // Every inbound mail is a threaded turn: each dispatched frame carries + // its own row-derived Message-ID (CL-7450), and each dispatch mail's + // correlation back to the row it answers is recorded for the reply + // path (CL-6314) — keyed by the answered row, whose derived + // Message-ID the sidecar's bracket reports back. + expect(sentMail).toHaveLength(2); + expect(sentMail[0]?.messageId).toBe(`<${first.id}@${DOMAIN}>`); + expect(sentMail[1]?.messageId).toBe(`<${second.id}@${DOMAIN}>`); + for (const posted of [first, second]) { + const source = await turnMailCorrelation.findTurnMailSource({ + tenantId: TENANT_ID, + mailId: posted.id, + }); + expect(source?.workbenchId).toBe(WORKBENCH_ID); + expect(source?.sourceMessageId).toBe(posted.id); + } + }); +});