Skip to content

fix(web): chats are keyed on the agent's definition asset and send to its live run (CL-8441) #876

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TheGreatAxios merged 2 commits into from
Sep 18, 2026
Merged

fix(web): chats are keyed on the agent's definition asset and send to its live run (CL-8441) #876

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
790f245
fix(web): chats are keyed on the agent's definition asset and send to…
TheGreatAxios Sep 18, 2026
c71474e
fix(web): adapt room participants to the agent-asset chat identity
TheGreatAxios Sep 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
181 changes: 127 additions & 54 deletions apps/web/src/chat/threads-api.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -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";
Expand Down Expand Up @@ -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<readonly ChatAgent[]> {
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<string, { addresses: Set<string>; 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<string>(),
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
Expand Down Expand Up @@ -150,15 +198,23 @@ export function frameBody(raw: string): string {
return "";
}

function addressRunId(address: string): string | undefined {
const local = address.split("@")[0]?.trim().replace(/^.*</, "");
return local !== undefined && local.startsWith("run_") ? local : undefined;
function extractAddress(raw: string): string {
return (/<([^>]+)>/.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;
Expand All @@ -171,15 +227,13 @@ type MailTurn = {
async function readFolder(tenantId: string, folder: "INBOX" | "Sent"): Promise<MailTurn[]> {
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),
Expand All @@ -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<void> {
Expand All @@ -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 } : {}),
Expand All @@ -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<string> {
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<void> {
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);
}

// ---------------------------------------------------------------------
Expand All @@ -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<string, ChatAgent> {
const index = new Map<string, ChatAgent>();
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<readonly ChatSummary[]> {
const [agents, turns] = await Promise.all([listChatAgents(tenantId), readTurns(tenantId)]);
const byRun = new Map<string, MailTurn[]>();
const addressToAgent = agentByAddress(agents);
const byAgent = new Map<string, MailTurn[]>();
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),
Expand All @@ -299,15 +373,16 @@ export async function listChats(tenantId: string): Promise<readonly ChatSummary[
.sort((a, b) => 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<ChatThread> {
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),
Expand Down Expand Up @@ -347,10 +422,9 @@ export type RoomParticipant = {
readonly id: string;
readonly kind: "person" | "agent";
readonly name: string;
/** For an agent, its current live run's address — empty when none is
* live, which `sendToRoom` filters out. */
readonly address: string;
/** Present for agents: the deployment's anchor run, what a send is
* addressed to. */
readonly runId?: string;
};

const PrincipalPage = type({
Expand All @@ -370,7 +444,7 @@ const PrincipalPage = type({
* appears after its first run, so the run listing is what makes an agent
* addressable from the moment it is deployed into the room. */
export async function listRoomParticipants(tenantId: string): Promise<readonly RoomParticipant[]> {
const [page, runs] = await Promise.all([
const [page, chatAgents] = await Promise.all([
getJson(`/api/tenants/${encodeURIComponent(tenantId)}/principals?limit=100`, PrincipalPage),
listChatAgents(tenantId),
]);
Expand All @@ -382,12 +456,11 @@ export async function listRoomParticipants(tenantId: string): Promise<readonly R
name: principal.displayName,
address: principal.email ?? principal.refId,
}));
const agents = runs.map((run): RoomParticipant => ({
id: run.runId,
const agents = chatAgents.map((agent): RoomParticipant => ({
id: agent.id,
kind: "agent",
name: run.name,
address: run.address,
runId: run.runId,
name: agent.name,
address: agent.liveAddress ?? "",
}));
return [...people, ...agents];
}
Expand Down
Loading
Loading