From 790f2453406af4dda820b7131b077de65ba8954e Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 17 Sep 2026 22:07:05 -0700 Subject: [PATCH 1/2] fix(web): chats are keyed on the agent's definition asset and send to its live run (CL-8441) --- apps/web/src/chat/threads-api.ts | 165 +++++++++++++++++------- apps/web/src/pages/chat-thread-page.tsx | 41 +++++- 2 files changed, 154 insertions(+), 52 deletions(-) diff --git a/apps/web/src/chat/threads-api.ts b/apps/web/src/chat/threads-api.ts index c60e8d26f..b4fa03039 100644 --- a/apps/web/src/chat/threads-api.ts +++ b/apps/web/src/chat/threads-api.ts @@ -5,10 +5,15 @@ // INBOX. This file is the only seam between the chat UI and that surface — // nothing here touches a chat-specific hub route, because none exist. // -// A chat is identified by its agent's run id: the run is the conversation, -// and every message in the chat carries that run address on one side. +// A chat is identified by its agent's definition asset id, not a run id — +// every hub restart releases the old run and redeploys under a new one, so +// keying on a run id would 409 the moment it turns terminal. An agent's +// address set spans every run it has ever had (releases included), which is +// what keeps a chat's history intact across a redeploy; sends resolve the +// current live run's address at send time. import { type } from "arktype"; +import { WorkflowDeploymentResponse } from "@intx/types"; import { listTopLevelRuns } from "../agents-api"; import { MYRA_SOURCE_CONFIG } from "../myra-source"; @@ -24,12 +29,22 @@ export class ChatApiError extends Error { } export type ChatAgent = { - /** The deployment's anchor run id — what a message is addressed to. */ - readonly runId: string; - readonly address: string; + /** Route/chat id: the agent's definition asset id — stable across + * redeploys, unlike a run id. */ + readonly id: string; readonly name: string; + /** Every address this agent has ever run under, releases included. */ + readonly addresses: readonly string[]; + /** The address of the agent's currently live run, or null when none is + * live (mid-redeploy). */ + readonly liveAddress: string | null; }; +const LIVE_DEPLOYMENT_STATUSES = new Set(["deployed", "pending", "recovering"]); + +const DeploymentsSchema = WorkflowDeploymentResponse.array(); +const WorkflowAssetSchema = type({ id: "string", name: "string" }).array(); + export type ChatMessage = { readonly id: string; readonly author: "me" | "agent"; @@ -61,18 +76,51 @@ export function agentInitials(name: string): string { return (letters.join("") || "?").toUpperCase(); } -/** Every deployed agent the person can start a chat with. The stock run - * listing already excludes non-top-level runs, so each row here is a real - * deployment with a routable address. */ +function deploymentsPath(tenantId: string): string { + return `/api/tenants/${encodeURIComponent(tenantId)}/workflows/deployments`; +} + +function workflowAssetsPath(tenantId: string): string { + return `/api/tenants/${encodeURIComponent(tenantId)}/assets?kind=workflow&inherited=false`; +} + +/** Every agent the person has ever had a deployment of, keyed by its + * definition asset id. Deployments list every anchor run ever created for + * an asset (releases included, most recent first), joined here against the + * run listing for each run's address and against the workflow assets for a + * display name. */ export async function listChatAgents(tenantId: string): Promise { - const runs = await listTopLevelRuns(tenantId); - return runs - .filter((run) => run.address.length > 0 && run.status !== "stopped") - .map((run) => ({ - runId: run.id, - address: run.address, - name: displayAgentName(run.definitionName), - })); + const [deployments, assets, runs] = await Promise.all([ + getJson(deploymentsPath(tenantId), DeploymentsSchema), + getJson(workflowAssetsPath(tenantId), WorkflowAssetSchema), + listTopLevelRuns(tenantId), + ]); + const addressByRunId = new Map(runs.map((run) => [run.id, run.address])); + const nameByAssetId = new Map(assets.map((asset) => [asset.id, asset.name])); + + const byAsset = new Map; liveAddress: string | null }>(); + for (const deployment of deployments) { + const address = addressByRunId.get(deployment.id); + if (address === undefined || address.length === 0) continue; + const entry = byAsset.get(deployment.definitionAssetId) ?? { + addresses: new Set(), + liveAddress: null, + }; + entry.addresses.add(address); + // Deployments come back newest-first, so the first live one seen per + // asset is the current one. + if (entry.liveAddress === null && LIVE_DEPLOYMENT_STATUSES.has(deployment.status)) { + entry.liveAddress = address; + } + byAsset.set(deployment.definitionAssetId, entry); + } + + return [...byAsset.entries()].map(([assetId, entry]) => ({ + id: assetId, + name: displayAgentName(nameByAssetId.get(assetId) ?? assetId), + addresses: [...entry.addresses], + liveAddress: entry.liveAddress, + })); } /** The agent an `@name` first message picks, matched case-insensitively @@ -150,15 +198,23 @@ export function frameBody(raw: string): string { return ""; } -function addressRunId(address: string): string | undefined { - const local = address.split("@")[0]?.trim().replace(/^.*]+)>/.exec(raw)?.[1] ?? raw).trim(); +} + +/** The run address on a message, from whichever side (from/to) carries + * one — a plain address, not just its local part, so it can be matched + * against an agent's full address set. */ +function participantAddress(envelope: { from: string; to: readonly string[] }): string | undefined { + return [envelope.from, ...envelope.to] + .map(extractAddress) + .find((address) => address.split("@")[0]?.startsWith("run_")); } type MailTurn = { readonly id: string; readonly messageId: string; - readonly runId: string; + readonly address: string; readonly author: "me" | "agent"; readonly subject: string; readonly body: string; @@ -171,15 +227,13 @@ type MailTurn = { async function readFolder(tenantId: string, folder: "INBOX" | "Sent"): Promise { const page = await getJson(`${mailboxPath(tenantId)}?folder=${folder}&limit=100`, InboxPage); return page.messages.flatMap((message) => { - const runId = [message.envelope.from, ...message.envelope.to] - .map(addressRunId) - .find((candidate) => candidate !== undefined); - if (runId === undefined) return []; + const address = participantAddress(message.envelope); + if (address === undefined) return []; return [ { id: `${folder}:${String(message.uid)}`, messageId: message.envelope.messageId, - runId, + address, author: folder === "Sent" ? ("me" as const) : ("agent" as const), subject: message.envelope.subject, body: frameBody(message.raw), @@ -205,7 +259,7 @@ const SendAccepted = type({ messageId: "string", uid: "number" }); async function sendToAgent( tenantId: string, - agent: ChatAgent, + address: string, body: string, inReplyTo: string | undefined, ): Promise { @@ -215,7 +269,7 @@ async function sendToAgent( method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ - to: [agent.address], + to: [address], subject: body.slice(0, 60), body, ...(inReplyTo !== undefined ? { inReplyTo } : {}), @@ -237,23 +291,31 @@ async function sendToAgent( } /** Starts a chat with one agent. Returns the chat id to route to — the - * agent's run id, which the Sent copy already carries. */ + * agent's definition asset id. Callers should keep the composer disabled + * until `liveAddress` is set; this still guards against a stale click. */ export async function startChat( tenantId: string, agent: ChatAgent, content: string, ): Promise { - await sendToAgent(tenantId, agent, content, undefined); - return agent.runId; + if (agent.liveAddress === null) { + throw new ChatApiError(`${agent.name} is starting…`); + } + await sendToAgent(tenantId, agent.liveAddress, content, undefined); + return agent.id; } -/** A reply is the same send, threaded onto the chat's newest message. */ +/** A reply is the same send, threaded onto the chat's newest message, sent + * to the agent's current live run — not whichever run last answered. */ export async function replyInChat( tenantId: string, chat: ChatThread, content: string, ): Promise { - await sendToAgent(tenantId, chat.agent, content, chat.lastMessageId); + if (chat.agent.liveAddress === null) { + throw new ChatApiError(`${chat.agent.name} is starting…`); + } + await sendToAgent(tenantId, chat.agent.liveAddress, content, chat.lastMessageId); } // --------------------------------------------------------------------- @@ -276,20 +338,32 @@ function chatTitle(turns: readonly MailTurn[], agentName: string): string { return first.subject.length > 0 ? first.subject : first.body.slice(0, 60); } -/** Every chat the person has: one per agent they have exchanged mail - * with. */ +function agentByAddress(agents: readonly ChatAgent[]): Map { + const index = new Map(); + for (const agent of agents) { + for (const address of agent.addresses) index.set(address, agent); + } + return index; +} + +/** Every chat the person has: one per agent they have exchanged mail with, + * grouped by the agent's asset id so a redeploy's new run still lands in + * the same chat. */ export async function listChats(tenantId: string): Promise { const [agents, turns] = await Promise.all([listChatAgents(tenantId), readTurns(tenantId)]); - const byRun = new Map(); + const addressToAgent = agentByAddress(agents); + const byAgent = new Map(); for (const turn of turns) { - byRun.set(turn.runId, [...(byRun.get(turn.runId) ?? []), turn]); + const agent = addressToAgent.get(turn.address); + if (agent === undefined) continue; + byAgent.set(agent.id, [...(byAgent.get(agent.id) ?? []), turn]); } - return [...byRun.entries()] - .map(([runId, rows]) => { - const agentName = agents.find((agent) => agent.runId === runId)?.name ?? runId; + return [...byAgent.entries()] + .map(([agentId, rows]) => { + const agentName = agents.find((agent) => agent.id === agentId)?.name ?? agentId; const newest = rows[rows.length - 1]!; return { - id: runId, + id: agentId, title: chatTitle(rows, agentName), agentName, preview: newest.body.slice(0, 80), @@ -299,15 +373,16 @@ export async function listChats(tenantId: string): Promise Date.parse(b.lastActivityAt) - Date.parse(a.lastActivityAt)); } -/** One chat's full transcript: every mail turn on that agent's run, - * oldest first. */ +/** One chat's full transcript: every mail turn addressed to any run this + * agent has ever had, oldest first. */ export async function readChat(tenantId: string, chatId: string): Promise { const [agents, turns] = await Promise.all([listChatAgents(tenantId), readTurns(tenantId)]); - const agent = agents.find((candidate) => candidate.runId === chatId); + const agent = agents.find((candidate) => candidate.id === chatId); if (agent === undefined) { - throw new ChatApiError("That agent is no longer deployed.", 404); + throw new ChatApiError("That agent could not be found.", 404); } - const rows = turns.filter((turn) => turn.runId === chatId); + const addresses = new Set(agent.addresses); + const rows = turns.filter((turn) => addresses.has(turn.address)); return { id: chatId, title: chatTitle(rows, agent.name), diff --git a/apps/web/src/pages/chat-thread-page.tsx b/apps/web/src/pages/chat-thread-page.tsx index 06a02f539..d0320b7b2 100644 --- a/apps/web/src/pages/chat-thread-page.tsx +++ b/apps/web/src/pages/chat-thread-page.tsx @@ -29,16 +29,18 @@ function errorText(cause: unknown): string { function Composer({ placeholder, busy, + disabled, onSend, }: { readonly placeholder: string; readonly busy: boolean; + readonly disabled?: boolean; readonly onSend: (text: string) => void; }) { const [text, setText] = useState(""); const send = () => { const trimmed = text.trim(); - if (trimmed === "" || busy) return; + if (trimmed === "" || busy || disabled) return; setText(""); onSend(trimmed); }; @@ -49,6 +51,7 @@ function Composer({ rows={3} placeholder={placeholder} aria-label={placeholder} + disabled={disabled} onChange={(event) => setText(event.target.value)} onKeyDown={(event) => { if (event.key === "Enter" && !event.shiftKey) { @@ -57,7 +60,7 @@ function Composer({ } }} /> - @@ -75,6 +78,16 @@ function NewChat({ const agentsQuery = useQuery({ queryKey: chatKeys.agents(tenantId), queryFn: () => listChatAgents(tenantId), + // Keep polling while the chosen agent has no live run yet, so the + // composer unlocks itself once a redeploy finishes. + refetchInterval: (query) => { + const agentsData = query.state.data; + if (agentsData === undefined) return false; + const defaultAgent = agentsData.find( + (agent) => agent.name === MYRA_SOURCE_CONFIG.displayName, + ); + return (defaultAgent ?? agentsData[0])?.liveAddress === null ? 3000 : false; + }, }); const agents = useMemo(() => agentsQuery.data ?? [], [agentsQuery.data]); const [selected, setSelected] = useState(""); @@ -89,8 +102,9 @@ function NewChat({ }); const defaultAgent = agents.find((agent) => agent.name === MYRA_SOURCE_CONFIG.displayName); - const chosen = agents.find((agent) => agent.runId === selected) ?? defaultAgent ?? agents[0]; + const chosen = agents.find((agent) => agent.id === selected) ?? defaultAgent ?? agents[0]; const error: unknown = start.error ?? agentsQuery.error; + const isLive = chosen?.liveAddress !== null && chosen?.liveAddress !== undefined; return ( @@ -98,12 +112,12 @@ function NewChat({