diff --git a/apps/web/src/app.css b/apps/web/src/app.css index d74a520f2b..d921efe24a 100644 --- a/apps/web/src/app.css +++ b/apps/web/src/app.css @@ -3793,3 +3793,70 @@ button.mission-control-jump-row:hover { font-size: 0.85rem; font-weight: 600; } + +/* Sidebar sections: Workbenches above Chats, each with its own label. */ +.shell-panel-section-label { + padding: 0.25rem 0.5rem; + font-size: 0.7rem; + font-weight: 600; + letter-spacing: 0.04em; + text-transform: uppercase; + color: var(--muted-foreground); +} +.shell-panel-see-all { + padding: 0.25rem 0.5rem; + text-align: left; + font-size: 0.78rem; + color: var(--muted-foreground); + background: none; + border: 0; + cursor: pointer; +} +.shell-panel-see-all:hover { + color: var(--foreground); +} + +/* A chat: transcript above, composer below. */ +.chat-thread-title { + margin: 0 0 1rem; + font-size: 1.1rem; + font-weight: 600; +} +.chat-thread-messages { + display: flex; + flex-direction: column; + gap: 1rem; + padding-bottom: 1rem; +} +.chat-thread-message { + display: flex; + gap: 0.75rem; + align-items: flex-start; +} +.chat-thread-body { + min-width: 0; + font-size: 0.9rem; + line-height: 1.55; +} +.chat-thread-error { + margin: 0.5rem 0; + font-size: 0.82rem; + color: var(--destructive); +} +.chat-agent-picker { + display: flex; + align-items: center; + gap: 0.5rem; + margin-bottom: 1rem; + font-size: 0.82rem; + color: var(--muted-foreground); +} +.chat-composer { + display: flex; + flex-direction: column; + gap: 0.5rem; + align-items: flex-end; +} +.chat-composer > textarea { + width: 100%; +} diff --git a/apps/web/src/chat-path.ts b/apps/web/src/chat-path.ts new file mode 100644 index 0000000000..8a3c46d640 --- /dev/null +++ b/apps/web/src/chat-path.ts @@ -0,0 +1,31 @@ +// Chat deep links. `/chats/new` composes a first message to an agent; +// `/chats/:id` is one chat, addressed by its mailbox thread root uid (or +// a local id while the agent has not answered yet). + +/** One key factory for every chat read, so a send can invalidate the + * listing, the transcript, and the agent roster in one call. */ +export const chatKeys = { + scope: (tenantId: string) => ["tenant", tenantId, "chats"] as const, + list: (tenantId: string) => ["tenant", tenantId, "chats", "list"] as const, + agents: (tenantId: string) => ["tenant", tenantId, "chats", "agents"] as const, + one: (tenantId: string, chatId: string) => ["tenant", tenantId, "chats", "one", chatId] as const, + childTenants: (tenantId: string) => ["tenant", tenantId, "child-tenants"] as const, +}; + +export const CHATS_PATH_PREFIX = "/chats"; +export const NEW_CHAT_PATH = `${CHATS_PATH_PREFIX}/new`; + +export function chatPath(chatId: string): string { + return `${CHATS_PATH_PREFIX}/${encodeURIComponent(chatId)}`; +} + +export function chatIdFromPath(path: string): string | null { + if (!path.startsWith(`${CHATS_PATH_PREFIX}/`)) return null; + const segment = path.slice(CHATS_PATH_PREFIX.length + 1); + if (segment === "" || segment === "new" || segment.includes("/")) return null; + return decodeURIComponent(segment); +} + +export function isChatPath(path: string): boolean { + return path === CHATS_PATH_PREFIX || path.startsWith(`${CHATS_PATH_PREFIX}/`); +} diff --git a/apps/web/src/chat/threads-api.ts b/apps/web/src/chat/threads-api.ts new file mode 100644 index 0000000000..9d9071b013 --- /dev/null +++ b/apps/web/src/chat/threads-api.ts @@ -0,0 +1,439 @@ +// Chats are mail threads. One chat has exactly one agent: the person +// triggers the agent's deployment run over the stock +// `POST /workflows/:runId/mail`, and the agent's reply lands in the +// person's own `@corbits/mailbox` inbox, threaded by In-Reply-To. This +// file is the only seam between the chat UI and those two surfaces — +// nothing here touches a chat-specific hub route, because none exist. +// +// The person's own outgoing turns are held client-side: mailbox rows are +// written for addressed principals only, and a run address is not a +// mailbox, so a trigger leaves no durable copy anywhere the person can +// read it back. + +import { type } from "arktype"; + +import { listTopLevelRuns } from "../agents-api"; +import { MYRA_SOURCE_CONFIG } from "../myra-source"; + +export class ChatApiError extends Error { + constructor( + message: string, + readonly status?: number, + ) { + super(message); + this.name = "ChatApiError"; + } +} + +export type ChatAgent = { + /** The deployment's anchor run id — what a trigger is addressed to. */ + readonly runId: string; + readonly address: string; + readonly name: string; +}; + +export type ChatMessage = { + readonly id: string; + readonly author: "me" | "agent"; + readonly authorName: string; + readonly body: string; + readonly at: string; +}; + +export type ChatSummary = { + /** Route id: a mailbox thread's root uid, or a local id while the + * agent has not answered a brand-new chat yet. */ + readonly id: string; + readonly title: string; + readonly agentName: string; + readonly preview: string; + readonly lastActivityAt: string; +}; + +/** Myra is the one default agent; her deploy asset name is not a display + * name anyone should have to read. */ +function displayAgentName(definitionName: string): string { + return definitionName === MYRA_SOURCE_CONFIG.assetName + ? MYRA_SOURCE_CONFIG.displayName + : definitionName; +} + +export function agentInitials(name: string): string { + const words = name.trim().split(/\s+/).filter(Boolean); + const letters = words.slice(0, 2).map((word) => word[0] ?? ""); + 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. */ +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), + })); +} + +/** The agent an `@name` first message picks, matched case-insensitively + * against the agent's display name with spaces removed. */ +export function agentFromMention( + text: string, + agents: readonly ChatAgent[], +): ChatAgent | undefined { + const mention = /(?:^|\s)@([\w-]+)/.exec(text)?.[1]?.toLowerCase(); + if (mention === undefined) return undefined; + return agents.find((agent) => agent.name.toLowerCase().replace(/\s+/g, "") === mention); +} + +// --------------------------------------------------------------------- +// Mailbox reads +// --------------------------------------------------------------------- + +function mailboxPath(tenantId: string): string { + return `/api/tenants/${encodeURIComponent(tenantId)}/mailbox/me/inbox`; +} + +const Envelope = type({ + messageId: "string", + from: "string", + to: "string[]", + subject: "string", + date: "string", + "inReplyTo?": "string", + references: "string[]", +}); + +const InboxPage = type({ + messages: type({ uid: "number", envelope: Envelope, raw: "string" }).array(), +}); + +/** Thread nodes nest arbitrarily; arktype expresses recursion only through a + * named scope, so the tree is validated one level at a time as it is walked. */ +const ThreadNode = type({ uid: "number", envelope: Envelope, children: "unknown[]" }); +const ThreadList = type({ threads: "unknown[]" }); +const ThreadOne = type({ thread: "unknown" }); + +type ThreadNodeShape = typeof ThreadNode.infer; + +async function getJson(path: string, schema: (value: unknown) => T | type.errors): Promise { + let response: Response; + try { + response = await fetch(path, { headers: { accept: "application/json" } }); + } catch (cause) { + throw new ChatApiError(cause instanceof Error ? cause.message : String(cause)); + } + if (!response.ok) { + throw new ChatApiError(`The server answered ${String(response.status)}.`, response.status); + } + const parsed = schema(await response.json().catch(() => undefined)); + if (parsed instanceof type.errors) { + throw new ChatApiError(`Unexpected response shape from ${path}: ${parsed.summary}`); + } + return parsed; +} + +function parseNode(value: unknown): ThreadNodeShape { + const parsed = ThreadNode(value); + if (parsed instanceof type.errors) { + throw new ChatApiError(`Unexpected mailbox thread node: ${parsed.summary}`); + } + return parsed; +} + +function flatten(node: ThreadNodeShape): ThreadNodeShape[] { + return [node, ...node.children.map(parseNode).flatMap(flatten)]; +} + +/** The readable text of an RFC 5322 frame: the bytes after the header + * section, or the first `text/plain` part of a multipart one. */ +export function frameBody(raw: string): string { + let decoded: string; + try { + decoded = atob(raw); + } catch { + return ""; + } + const text = decoded.replace(/\r\n/g, "\n"); + const split = text.indexOf("\n\n"); + if (split < 0) return ""; + const headers = text.slice(0, split).toLowerCase(); + const body = text.slice(split + 2); + const boundary = /boundary="?([^";\n]+)"?/.exec(headers)?.[1]; + if (boundary === undefined) return body.trim(); + for (const part of body.split(`--${boundary}`)) { + const partSplit = part.indexOf("\n\n"); + if (partSplit < 0) continue; + if (!part.slice(0, partSplit).toLowerCase().includes("text/plain")) continue; + return part.slice(partSplit + 2).trim(); + } + return ""; +} + +async function inboxBodies(tenantId: string): Promise> { + const page = await getJson(`${mailboxPath(tenantId)}?limit=100`, InboxPage); + return new Map(page.messages.map((message) => [message.uid, frameBody(message.raw)])); +} + +// --------------------------------------------------------------------- +// The person's own turns +// --------------------------------------------------------------------- + +const SentTurn = type({ + localId: "string", + runId: "string", + agentName: "string", + messageId: "string", + body: "string", + at: "string", +}); +const SentTurns = SentTurn.array(); +type SentTurn = typeof SentTurn.infer; + +function storageKey(tenantId: string): string { + return `workbench.chat.sent.${tenantId}`; +} + +function readSentTurns(tenantId: string): SentTurn[] { + const raw = window.localStorage.getItem(storageKey(tenantId)); + if (raw === null) return []; + let json: unknown; + try { + json = JSON.parse(raw); + } catch { + return []; + } + const parsed = SentTurns(json); + return parsed instanceof type.errors ? [] : parsed; +} + +function writeSentTurn(tenantId: string, turn: SentTurn): void { + window.localStorage.setItem( + storageKey(tenantId), + JSON.stringify([...readSentTurns(tenantId), turn]), + ); +} + +// --------------------------------------------------------------------- +// Sends +// --------------------------------------------------------------------- + +const TriggerAccepted = type({ runId: "string", address: "string", messageId: "string" }); + +async function triggerAgent(tenantId: string, runId: string, content: string): Promise { + const path = `/api/tenants/${encodeURIComponent(tenantId)}/workflows/${encodeURIComponent(runId)}/mail`; + let response: Response; + try { + response = await fetch(path, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ content }), + }); + } catch (cause) { + throw new ChatApiError(cause instanceof Error ? cause.message : String(cause)); + } + if (!response.ok) { + throw new ChatApiError( + `The agent could not be reached (${String(response.status)}).`, + response.status, + ); + } + const parsed = TriggerAccepted(await response.json().catch(() => undefined)); + if (parsed instanceof type.errors) { + throw new ChatApiError(`Unexpected trigger response: ${parsed.summary}`); + } + return parsed.messageId; +} + +/** Starts a chat with one agent. Returns the local chat id to route to — + * the mailbox has no thread for it until the agent answers. */ +export async function startChat( + tenantId: string, + agent: ChatAgent, + content: string, +): Promise { + const messageId = await triggerAgent(tenantId, agent.runId, content); + const localId = `local-${messageId}`; + writeSentTurn(tenantId, { + localId, + runId: agent.runId, + agentName: agent.name, + messageId, + body: content, + at: new Date().toISOString(), + }); + return localId; +} + +/** A reply rides the same trigger route: the run is the conversation, and + * the agent threads its answer onto the chat it is already holding. */ +export async function replyInChat( + tenantId: string, + chat: { readonly id: string; readonly runId: string; readonly agentName: string }, + content: string, +): Promise { + const messageId = await triggerAgent(tenantId, chat.runId, content); + writeSentTurn(tenantId, { + localId: chat.id, + runId: chat.runId, + agentName: chat.agentName, + messageId, + body: content, + at: new Date().toISOString(), + }); +} + +// --------------------------------------------------------------------- +// Chat listing and reads +// --------------------------------------------------------------------- + +function addressRunId(address: string): string | undefined { + const local = address.split("@")[0]?.trim().replace(/^.* { + const page = await getJson(`${mailboxPath(tenantId)}/threads?folder=INBOX`, ThreadList); + return page.threads.map((raw) => { + const root = parseNode(raw); + return { root, nodes: flatten(root) }; + }); +} + +/** Every chat the person has: one per mailbox thread, plus any chat they + * started that the agent has not answered yet. */ +export async function listChats(tenantId: string): Promise { + const sent = readSentTurns(tenantId); + const trees = await listThreadTrees(tenantId); + const answeredRuns = new Set(); + + const fromMail: ChatSummary[] = trees.map(({ root, nodes }) => { + const runId = threadRunId(nodes); + if (runId !== undefined) answeredRuns.add(runId); + const newest = nodes[nodes.length - 1] ?? root; + const agentName = sent.find((turn) => turn.runId === runId)?.agentName ?? runId ?? "Agent"; + return { + id: String(root.uid), + title: root.envelope.subject.length > 0 ? root.envelope.subject : agentName, + agentName, + preview: newest.envelope.subject, + lastActivityAt: newest.envelope.date, + }; + }); + + const pending: ChatSummary[] = sent + .filter((turn) => !answeredRuns.has(turn.runId)) + .filter((turn, index, rows) => rows.findIndex((row) => row.localId === turn.localId) === index) + .map((turn) => ({ + id: turn.localId, + title: turn.body.slice(0, 60), + agentName: turn.agentName, + preview: turn.body.slice(0, 80), + lastActivityAt: turn.at, + })); + + return [...fromMail, ...pending].sort( + (a, b) => Date.parse(b.lastActivityAt) - Date.parse(a.lastActivityAt), + ); +} + +/** One chat's full transcript: the agent's mail turns merged with the + * person's own trigger turns, oldest first. */ +export async function readChat(tenantId: string, chatId: string): Promise { + const sent = readSentTurns(tenantId); + if (chatId.startsWith("local-")) { + const turns = sent.filter((turn) => turn.localId === chatId); + const first = turns[0]; + if (first === undefined) throw new ChatApiError("That chat is not on this device.", 404); + return { + id: chatId, + title: first.body.slice(0, 60), + agentName: first.agentName, + runId: first.runId, + messages: turns.map((turn) => ({ + id: turn.messageId, + author: "me", + authorName: "You", + body: turn.body, + at: turn.at, + })), + }; + } + + const [page, bodies] = await Promise.all([ + getJson( + `${mailboxPath(tenantId)}/threads/${encodeURIComponent(chatId)}?folder=INBOX`, + ThreadOne, + ), + inboxBodies(tenantId), + ]); + const root = parseNode(page.thread); + const nodes = flatten(root); + const runId = threadRunId(nodes) ?? ""; + const agentName = sent.find((turn) => turn.runId === runId)?.agentName ?? runId; + + const mine = sent.filter((turn) => turn.runId === runId); + const messages: ChatMessage[] = [ + ...mine.map((turn): ChatMessage => ({ + id: turn.messageId, + author: "me", + authorName: "You", + body: turn.body, + at: turn.at, + })), + ...nodes.map((node): ChatMessage => ({ + id: String(node.uid), + author: "agent", + authorName: agentName, + body: bodies.get(node.uid) ?? node.envelope.subject, + at: node.envelope.date, + })), + ].sort((a, b) => Date.parse(a.at) - Date.parse(b.at)); + + return { + id: chatId, + title: root.envelope.subject.length > 0 ? root.envelope.subject : agentName, + agentName, + runId, + messages, + }; +} + +/** Live mailbox updates. The stream carries no chat identity, so a caller + * refetches on any event rather than patching a thread in place. */ +export function subscribeToInbox(tenantId: string, onChange: () => void): () => void { + const source = new EventSource(`${mailboxPath(tenantId)}/events`); + const handler = () => onChange(); + source.addEventListener("message", handler); + return () => { + source.removeEventListener("message", handler); + source.close(); + }; +} diff --git a/apps/web/src/fetch-dedupe.test.tsx b/apps/web/src/fetch-dedupe.test.tsx deleted file mode 100644 index 3af04812e0..0000000000 --- a/apps/web/src/fetch-dedupe.test.tsx +++ /dev/null @@ -1,181 +0,0 @@ -// the shell used to fire the same listing 4-5x per navigation -// because independent components each fetched independently instead of -// sharing a cache. Every mount of `useBenchActivity` (the sidebar's -// `WorkbenchList`, and any second subscriber) shares the same TanStack -// Query keys (`tenantKeys.workbenches`, `.routineActivity` — see -// `../src/query-client.ts`) under one `QueryClient`, so two mounts fetch -// each listing exactly once. -import { afterEach, describe, expect, test } from "bun:test"; -import { act } from "react"; -import { createRoot } from "react-dom/client"; -import type { Root } from "react-dom/client"; - -import { BenchProvider } from "./bench-context"; -import { requestWorkbenchRename } from "./workbench-rename-events"; -import { WorkbenchList } from "./shell/workbench-list"; -import { createTestQueryClient, TestQueryProvider } from "./test-query-provider"; - -const realFetch = globalThis.fetch; - -let container: HTMLDivElement | null = null; -let root: Root | null = null; - -afterEach(() => { - globalThis.fetch = realFetch; - if (root !== null) { - act(() => root?.unmount()); - root = null; - } - if (container !== null) { - container.remove(); - container = null; - } -}); - -const membership = { - data: [ - { - principalId: "prn_1", - tenantId: "tnt_1", - tenantName: "Corbits Bench", - tenantSlug: "corbits-bench", - kind: "user", - status: "active", - roles: [], - }, - ], - nextCursor: null, -}; - -function json(body: unknown): Response { - return new Response(JSON.stringify(body), { - status: 200, - headers: { "content-type": "application/json" }, - }); -} - -function stubFetch(calls: string[]): void { - globalThis.fetch = ((input: RequestInfo | URL) => { - const path = typeof input === "string" ? input : String(input); - calls.push(path); - if (path.includes("/api/me/principals")) return Promise.resolve(json(membership)); - if (path.includes("/agent-definitions/visible")) - return Promise.resolve(json({ definitions: [] })); - return Promise.resolve(json({ items: [] })); - }) as typeof fetch; -} - -async function settle() { - for (let i = 0; i < 20; i++) { - await act(async () => { - await new Promise((resolve) => setTimeout(resolve, 0)); - }); - } -} - -function countsByMatch(calls: readonly string[], predicate: (path: string) => boolean): number { - return calls.filter(predicate).length; -} - -describe("shell listing dedupe", () => { - test("two WorkbenchList mounts together fetch each listing exactly once", async () => { - const calls: string[] = []; - stubFetch(calls); - const queryClient = createTestQueryClient(); - - container = document.createElement("div"); - document.body.appendChild(container); - root = createRoot(container); - await act(async () => { - root?.render( - - - undefined} /> - undefined} /> - - , - ); - }); - await settle(); - - // Before the fix, each of these fired once per mounted band (2x here, - // and up to 5x across the shell's real surfaces) — now exactly once - // per (tenant, kind) no matter how many bands subscribe. - expect(countsByMatch(calls, (p) => p.includes("kind=workbench"))).toBe(1); - expect(countsByMatch(calls, (p) => p.includes("kind=chat"))).toBe(1); - // the routines query resolves with no items and no fetch (the - // `feed=fires` route is deleted and the native listing has no fires - // equivalent), so both mounts share that without any request at all. - expect(countsByMatch(calls, (p) => p.includes("/top-level-runs"))).toBe(0); - expect(countsByMatch(calls, (p) => p.includes("/workflows/runs"))).toBe(0); - }); - - test("a rename invalidates the shared listing query, triggering exactly one refetch", async () => { - const calls: string[] = []; - const workbench = { - id: "ch_1", - title: "General", - kind: "workbench", - pinned: false, - participants: [], - }; - globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => { - const path = typeof input === "string" ? input : String(input); - calls.push(path); - if (path.includes("/api/me/principals")) return Promise.resolve(json(membership)); - if (path.includes("/agent-definitions/visible")) - return Promise.resolve(json({ definitions: [] })); - if (init?.method === "PATCH") { - return Promise.resolve( - json({ - ...workbench, - title: "Renamed", - settings: {}, - contextWindow: { value: 20, source: "inherit" }, - }), - ); - } - if (path.includes("kind=workbench")) return Promise.resolve(json({ items: [workbench] })); - return Promise.resolve(json({ items: [] })); - }) as typeof fetch; - const queryClient = createTestQueryClient(); - - container = document.createElement("div"); - document.body.appendChild(container); - root = createRoot(container); - await act(async () => { - root?.render( - - - undefined} /> - - , - ); - }); - await settle(); - - expect(countsByMatch(calls, (p) => p.includes("kind=workbench"))).toBe(1); - - act(() => requestWorkbenchRename("ch_1")); - await settle(); - const input = container.querySelector('input[aria-label="Rename"]') as HTMLInputElement | null; - expect(input).not.toBeNull(); - const setter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, "value")?.set; - if (setter === undefined) { - throw new Error("HTMLInputElement.prototype.value has no native setter"); - } - act(() => { - setter.call(input, "Renamed"); - input?.dispatchEvent(new Event("input", { bubbles: true })); - }); - await act(async () => { - input?.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true })); - }); - await settle(); - - // One refetch, triggered by the rename's invalidateQueries — not zero - // (freshness must actually fire) and not more than one (no duplicate - // invalidation sites). - expect(countsByMatch(calls, (p) => p.includes("kind=workbench"))).toBe(2); - }); -}); diff --git a/apps/web/src/pages/chat-thread-page.tsx b/apps/web/src/pages/chat-thread-page.tsx new file mode 100644 index 0000000000..06a02f5397 --- /dev/null +++ b/apps/web/src/pages/chat-thread-page.tsx @@ -0,0 +1,235 @@ +// A chat: one person, one agent, one mail thread. `/chats/new` composes a +// first message (agent picked in the select, or by @tagging one in the +// message itself); `/chats/:id` is the transcript plus a reply box. + +import { Button, EmptyState, PageShell, Select, Textarea } from "@corbits/react-ui"; +import { WarningCircle } from "@/lib/icons"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { useEffect, useMemo, useState } from "react"; + +import { Markdown } from "@/chat/markdown"; +import { + agentFromMention, + agentInitials, + listChatAgents, + readChat, + replyInChat, + startChat, + subscribeToInbox, + type ChatAgent, +} from "@/chat/threads-api"; +import { MYRA_SOURCE_CONFIG } from "../myra-source"; +import { useBench } from "../bench-context"; +import { chatIdFromPath, chatKeys, chatPath, NEW_CHAT_PATH } from "../chat-path"; + +function errorText(cause: unknown): string { + return cause instanceof Error ? cause.message : String(cause); +} + +function Composer({ + placeholder, + busy, + onSend, +}: { + readonly placeholder: string; + readonly busy: boolean; + readonly onSend: (text: string) => void; +}) { + const [text, setText] = useState(""); + const send = () => { + const trimmed = text.trim(); + if (trimmed === "" || busy) return; + setText(""); + onSend(trimmed); + }; + return ( +
+