diff --git a/.env.example b/.env.example index 4ffae83bb..c4d98c207 100644 --- a/.env.example +++ b/.env.example @@ -54,11 +54,16 @@ WORKER_HOST=127.0.0.1 # AGENT_TOKEN= # OpenBot's Intelligence runtime is not this raw AG-UI URL. See its adapter docs. -# CopilotKit Intelligence is required in every mode. Create or select a project -# with the commands below, then keep the generated key server-only. +# CopilotKit Intelligence is the default threads backend and requires a project +# key in every mode. Create or select a project with the commands below, then +# keep the generated key server-only. # npx copilotkit@latest login # npx copilotkit@latest project select CPK_INTELLIGENCE_API_KEY= +# THREADS_BACKEND=local keeps chat threads on the server's own database with +# no Intelligence key and no Intelligence requests. Rename, archive and +# realtime thread metadata remain Intelligence-only. See docs/PLUGGABLE-THREADS.md. +# THREADS_BACKEND=intelligence # CopilotKit SDK telemetry goes through the shared sink to PostHog, tagged OpenMuse. # project select also writes CPK_TELEMETRY_ID: preserve that generated value and diff --git a/README.md b/README.md index c1c6982ca..e93b052d5 100644 --- a/README.md +++ b/README.md @@ -57,13 +57,13 @@ The computer combines **persistent Chromium and an optional Linux workspace**. T | **Finance** | Import transaction CSV to create a spending summary with categories, transactions, and a savings-goal action. | | **Gmail & Calendar** | Google OAuth adapters, complete mail threads, drafts/attachments, calendar discovery, and reviewed event creation/update/deletion. Live credentials required. | | **Personal context** | Editable name, tone, avatar, and memories. Background-update preferences and durable in-app notifications. | -| **Rich Threads** | CopilotKit Intelligence persistence in every mode, with a stable main conversation, side chats, renaming, archiving, restoring, and replay. A server-only project key is required. | +| **Rich Threads** | CopilotKit Intelligence persistence in every mode, with a stable main conversation, side chats, renaming, archiving, restoring, and replay. A server-only project key is required — or set `THREADS_BACKEND=local` to persist threads on your own database with no key. | The [feature inventory](docs/FEATURES.md) describes implemented capabilities and planned extensions. Health/bank/social connectors, device push, voice, generated executable tools, and automatic reservations/payments are on the [roadmap](ROADMAP.md). ## Quick start -**Requirements:** Node 24 LTS, pnpm 11.19.0, and a CopilotKit Intelligence project key. The local sample app needs no model, Google account, or Docker. +**Requirements:** Node 24 LTS, pnpm 11.19.0, and either a CopilotKit Intelligence project key (default) or `THREADS_BACKEND=local` for fully self-hosted thread persistence. The local sample app needs no model, Google account, or Docker. ```sh git clone https://github.com/CopilotKit/OpenMuse.git openmuse @@ -73,6 +73,7 @@ cp .env.example .env npx copilotkit@latest login npx copilotkit@latest project select # Set CPK_INTELLIGENCE_API_KEY in .env to the generated server-only project key. +# Or set THREADS_BACKEND=local in .env to skip Intelligence entirely. # Keep the CLI-generated CPK_TELEMETRY_ID too, including in deployed server environments. pnpm dev ``` @@ -111,7 +112,7 @@ The deploy form asks for three values you provide. Render generates the other tw | Variable | Set by | If it is missing | |---|---|---| -| `CPK_INTELLIGENCE_API_KEY` | You. Run `npx copilotkit@latest login`, then `npx copilotkit@latest project select`. Keep it on the server. | Chat cannot open a thread. | +| `CPK_INTELLIGENCE_API_KEY` | You. Run `npx copilotkit@latest login`, then `npx copilotkit@latest project select`. Keep it on the server. Required unless `THREADS_BACKEND=local`. | Chat cannot open a thread (the API refuses to start when the backend is `intelligence`). | | `CPK_TELEMETRY_ID` | You. Keep the identity generated by the same project selection. | Runtime usage cannot be attributed to the Intelligence signup account. | | `OPENAI_API_KEY` | You. Used by the default `openai/gpt-5`. Change `MODEL` and supply the matching provider key for Anthropic or Google. | The model call fails. | | `OPENMUSE_ACCESS_KEY` | Render | You cannot sign in. | @@ -129,7 +130,7 @@ Health check: `https:///api/health`. **Standard** is the smallest plan that stays up. At 512 MB the process runs out of memory before it binds a port, because PGlite loads an embedded Postgres build. -**The disk** holds the database, PDFs, and the signing key. A redeploy without it wipes that data. Chat threads are stored by CopilotKit Intelligence, so a thread can still load after you sign back in even when the disk was never attached. +**The disk** holds the database, PDFs, and the signing key. A redeploy without it wipes that data. With the default `THREADS_BACKEND=intelligence`, chat threads are stored by CopilotKit Intelligence, so a thread can still load after you sign back in even when the disk was never attached. With `THREADS_BACKEND=local`, chat threads live on that same disk and are lost with it. **Live mode** is required. Render binds `0.0.0.0`, and sample mode rejects any host that is not loopback. The Blueprint sets `WORKSPACE_MODE=live`. @@ -142,8 +143,8 @@ The Docker computer and Google mail or calendar need the setup in the sections b Copy the commented settings in [.env.example](.env.example) into your private `.env`: 1. Set `AGENT_BACKEND=model`, `MODEL=provider/model-id`, and the matching provider key. CopilotKit supports the configured OpenAI, Anthropic or Google provider. Fictional data can still be used with a real model. Provider keys stay on the server. -2. Create or select a CopilotKit Intelligence project with `npx copilotkit@latest login` and `npx copilotkit@latest project select`. Keep the generated `CPK_INTELLIGENCE_API_KEY` on the server. -3. For personal mail/calendar, set `WORKSPACE_MODE=live`, the generated `CPK_INTELLIGENCE_API_KEY`, a random `OPENMUSE_ACCESS_KEY` of at least 24 characters, and `TOKEN_ENCRYPTION_KEY` containing 32 random bytes encoded as base64. Restart the API. +2. Create or select a CopilotKit Intelligence project with `npx copilotkit@latest login` and `npx copilotkit@latest project select`. Keep the generated `CPK_INTELLIGENCE_API_KEY` on the server. (Skip this and set `THREADS_BACKEND=local` to keep threads on your own database instead.) +3. For personal mail/calendar, set `WORKSPACE_MODE=live`, the generated `CPK_INTELLIGENCE_API_KEY` (unless threads are local), a random `OPENMUSE_ACCESS_KEY` of at least 24 characters, and `TOKEN_ENCRYPTION_KEY` containing 32 random bytes encoded as base64. Restart the API. 4. Configure a Google OAuth web client with Gmail and Calendar APIs enabled. Set `GOOGLE_CLIENT_ID` and `GOOGLE_CLIENT_SECRET`; register `${PUBLIC_API_URL}/api/google/callback` as its redirect URI. Configure consent/test-user access in your Google project. 5. Open **Apps → Gmail** (or **Google Calendar**), connect read access, and grant write access when needed. Every send or calendar change still requires its own stored review. Changing/disconnecting the account invalidates pending connection-bound work. @@ -189,7 +190,9 @@ No hidden retry occurs after an uncertain external write. Review its provider ou ## CopilotKit Rich Threads -Every deployment requires `CPK_INTELLIGENCE_API_KEY` on the API server for CopilotKit Intelligence conversation persistence and replay. Create or select a project with `npx copilotkit@latest login` and `npx copilotkit@latest project select`, set the generated server-only key, and restart the API. The native menu uses `useThreads`; rich tool results link back to saved tasks, documents, and browser sessions. +By default every deployment requires `CPK_INTELLIGENCE_API_KEY` on the API server for CopilotKit Intelligence conversation persistence and replay. Create or select a project with `npx copilotkit@latest login` and `npx copilotkit@latest project select`, set the generated server-only key, and restart the API. The native menu uses `useThreads`; rich tool results link back to saved tasks, documents, and browser sessions. + +Set `THREADS_BACKEND=local` to run fully self-hosted instead: threads persist in the server's own database with no Intelligence key and no Intelligence requests. Local mode supports the main conversation, side chats, listing, replay, and clear; rename, archive, and realtime metadata remain Intelligence-only and are hidden in the native UI. [Backend selection, durability boundary and capability matrix](docs/PLUGGABLE-THREADS.md). Intelligence is a separate service and is not included in this repository's MIT license. No project key is shipped. [Configuration and validation boundaries](docs/RICH-THREADS.md). @@ -203,7 +206,7 @@ The CopilotKit SDK sends runtime telemetry through the shared CopilotKit sink to flowchart TD Client[Expo / React Native / Web] -->|AG-UI and authenticated API| API[Hono + CopilotKit runtime] API --> Tasks[Durable task worker] - API --> Threads[CopilotKit Intelligence required in every mode] + API --> Threads[CopilotKit Intelligence (default) or THREADS_BACKEND=local] API --> Store[(PGlite or PostgreSQL)] Tasks --> Store Tasks --> Review[Stored action review] diff --git a/apps/mobile/src/threads.tsx b/apps/mobile/src/threads.tsx index 4f3fb51b0..4254d2756 100644 --- a/apps/mobile/src/threads.tsx +++ b/apps/mobile/src/threads.tsx @@ -1,6 +1,7 @@ import { useThreads } from "@copilotkit/react-native/headless"; import { Archive, + ArchiveRestore, CalendarDays, FileText, MessageCircle, @@ -8,6 +9,7 @@ import { Plus, RefreshCw, Settings2, + Trash2, } from "lucide-react-native"; import { createContext, type ReactNode, useContext, useEffect, useRef, useState } from "react"; import { ActivityIndicator, Pressable, Text, View } from "react-native"; @@ -32,6 +34,7 @@ const ThreadContext = createContext<{ retry: () => void; select: (selection: Selection) => void; start: () => void; + removeVisited: (id: string) => void; claimPrompt: (id: number) => boolean; } | null>(null); export function ThreadsProvider({ children }: { children: ReactNode }) { @@ -88,6 +91,10 @@ export function ThreadsProvider({ children }: { children: ReactNode }) { selection, select, start: () => select({ id: newThreadId(), existing: false }), + removeVisited: (id) => { + setVisited((items) => items.filter((item) => item.id !== id)); + if (selection.id === id) select({ id: mainId, existing: true }); + }, }} > {children} @@ -99,6 +106,81 @@ export function useMuseThread() { if (!context) throw new Error("Threads provider is unavailable"); return context; } + +type ThreadSummary = { id: string; name: string | null; archived: boolean }; + +/** + * One saved conversation row. The layout matches the Intelligence backend: + * the title sits on its own line, and a row of small labeled buttons + * (Rename / Archive|Restore / Delete) sits beneath it. + */ +function ThreadRow({ + thread, + selected, + editing, + busy, + draft, + onDraft, + onOpen, + onSave, + onArchive, + onDelete, +}: { + thread: ThreadSummary; + selected: boolean; + editing: boolean; + busy: boolean; + draft: string; + onDraft: (value: string) => void; + onOpen: () => void; + onSave: () => void; + onArchive: () => void; + onDelete: () => void; +}) { + const title = thread.name || "Untitled conversation"; + return ( + + + + + {title} + + + {editing && ( + + )} + + + + + + + ); +} export function ThreadsSheet({ onClose }: { onClose: () => void }) { const { enabled, @@ -110,6 +192,7 @@ export function ThreadsSheet({ onClose }: { onClose: () => void }) { retry, select, start, + removeVisited, } = useMuseThread(); const { workspace, open, navigate, refresh } = useWorkspace(); const threads = useThreads({ agentId: "default", enabled, includeArchived: true, limit: 20 }); @@ -201,64 +284,40 @@ export function ThreadsSheet({ onClose }: { onClose: () => void }) { {threads.threads .filter((thread) => thread.id !== mainId && thread.archived === archived) .map((thread) => ( - { + select({ id: thread.id, existing: true }); + onClose(); }} - > - { - select({ id: thread.id, existing: true }); - onClose(); - }} - style={[s.row, { gap: 10 }]} - > - - - {thread.name || "Untitled conversation"} - - - {editing === thread.id && ( - - )} - - - - - + onSave={() => { + if (editing === thread.id) + void mutate(() => threads.renameThread(thread.id, name.trim())); + else { + setEditing(thread.id); + setName(thread.name || ""); + } + }} + onArchive={() => + void mutate(() => + thread.archived + ? threads.unarchiveThread(thread.id) + : threads.archiveThread(thread.id), + ) + } + onDelete={() => + void mutate(async () => { + await threads.deleteThread(thread.id); + removeVisited(thread.id); + }) + } + /> ))} {!threads.isLoading && !threads.error && diff --git a/apps/mobile/src/ui.tsx b/apps/mobile/src/ui.tsx index 0346f6478..5babbf02e 100644 --- a/apps/mobile/src/ui.tsx +++ b/apps/mobile/src/ui.tsx @@ -162,28 +162,39 @@ export function IconButton({ icon: Icon, label, onPress, + disabled, + small, + danger, }: { icon: LucideIcon; label: string; onPress: () => void; + disabled?: boolean; + small?: boolean; + danger?: boolean; }) { + const color = danger ? colors.danger : colors.text; + const size = small ? 32 : 44; return ( [ { - width: 44, - height: 44, + width: size, + height: size, alignItems: "center", justifyContent: "center", - borderRadius: 22, + borderRadius: small ? 16 : 22, backgroundColor: pressed ? colors.line : "#FFFFFF", }, + disabled && { opacity: 0.4 }, ]} > - + ); } diff --git a/apps/server/src/agent.ts b/apps/server/src/agent.ts index 4a5a75865..8eca127a3 100644 --- a/apps/server/src/agent.ts +++ b/apps/server/src/agent.ts @@ -2,7 +2,6 @@ import "./config.ts"; import { HttpAgent } from "@ag-ui/client"; import { type AgentsFactory, - type CopilotKitIntelligence, CopilotRuntime, createCopilotHonoHandler, } from "@copilotkit/runtime/v2"; @@ -11,6 +10,7 @@ import type { Config } from "./config.ts"; import { ConversationAgent } from "./engine/conversation.ts"; import type { AgentService } from "./engine/service.ts"; import { createJevAdapter, type JevAdapter } from "./jev/adapter.ts"; +import type { ThreadProvider } from "./threads/provider.ts"; export function agentConfigured(config: Config) { return ( @@ -29,42 +29,49 @@ export function makeRuntime( config: Config, service: AgentService, auth: Auth, - intelligence: CopilotKitIntelligence, + threads: ThreadProvider, + agentOwners: WeakMap, ) { // Built on first use, then shared so live mode reuses one TypeSafe client across requests. let jevAdapter: JevAdapter | undefined; const sharedJevAdapter = () => (jevAdapter ??= createJevAdapter(config)); - const agents: AgentsFactory = async ({ request }) => ({ - default: + // The runner's run/persist methods receive no request context, so the factory + // records the owning user on each agent it builds and the lookup reaches the + // thread provider (PersistentAgentRunner scopes saved threads by owner). + const agents: AgentsFactory = async ({ request }) => { + const owner = await auth.owner(request.headers.get("authorization") ?? undefined); + const agent = config.agentBackend === "sample" - ? new ConversationAgent( - config, - service, - await auth.owner(request.headers.get("authorization") ?? undefined), - sharedJevAdapter(), - ) + ? new ConversationAgent(config, service, owner, sharedJevAdapter()) : config.agentBackend === "agui" ? new HttpAgent({ url: config.agentUrl ?? "http://127.0.0.1:1/unconfigured", headers: config.agentToken ? { Authorization: `Bearer ${config.agentToken}` } : {}, }) - : new ConversationAgent( - config, - service, - await auth.owner(request.headers.get("authorization") ?? undefined), - sharedJevAdapter(), - ), - }); - const runtime = new CopilotRuntime({ - agents, - intelligence, - // The shared CopilotKit sink carries this tag onto existing PostHog events. - telemetryProperties: { accessibility_title: "OpenMuse" }, - identifyUser: async (request) => ({ - id: await auth.owner(request.headers.get("authorization") ?? undefined), - name: "OpenMuse user", - }), - generateThreadNames: false, - }); + : new ConversationAgent(config, service, owner, sharedJevAdapter()); + if (agent instanceof ConversationAgent) agentOwners.set(agent, owner); + return { default: agent }; + }; + // CopilotRuntime dispatches on this option set: with `intelligence` it builds + // the Intelligence runtime; with `runner` (and no intelligence) the SSE + // runtime that serves threads from the local runner itself. + const runtime = + threads.backend === "intelligence" + ? new CopilotRuntime({ + agents, + intelligence: threads.intelligence, + // The shared CopilotKit sink carries this tag onto existing PostHog events. + telemetryProperties: { accessibility_title: "OpenMuse" }, + identifyUser: async (request) => ({ + id: await auth.owner(request.headers.get("authorization") ?? undefined), + name: "OpenMuse user", + }), + generateThreadNames: false, + }) + : new CopilotRuntime({ + agents, + runner: threads.runner, + telemetryProperties: { accessibility_title: "OpenMuse" }, + }); return createCopilotHonoHandler({ runtime, basePath: "/api/copilotkit" }); } diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index 258d7a22b..19a059ae2 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -1,8 +1,7 @@ import "./config.ts"; import { randomUUID } from "node:crypto"; import { MessageSchema } from "@ag-ui/core"; -import { CopilotKitIntelligence } from "@copilotkit/runtime/v2"; -import { Hono } from "hono"; +import { type Context, Hono } from "hono"; import { bodyLimit } from "hono/body-limit"; import { cors } from "hono/cors"; import { z } from "zod"; @@ -17,14 +16,17 @@ import { createAuth } from "./auth.ts"; import { BrowserService } from "./browser.ts"; import { ComputerService, type DockerRunner } from "./computer.ts"; import { computerRoutes } from "./computer-routes.ts"; -import { assertApiDeploymentConfig, type Config } from "./config.ts"; +import { assertThreadsBackendConfig, type Config } from "./config.ts"; import type { Store } from "./db.ts"; +import { ConversationAgent } from "./engine/conversation.ts"; import { agentRoutes } from "./engine/routes.ts"; import { AgentService } from "./engine/service.ts"; import { AppError } from "./errors.ts"; import { Files } from "./files.ts"; import { GoogleAuth } from "./google-auth.ts"; import { rateLimit } from "./rate-limit.ts"; +import { CHAT_THREADS_KIND, type LocalThreadRecord } from "./threads/local-runner.ts"; +import { createThreadProvider, type ThreadProvider } from "./threads/provider.ts"; import { WorkspaceService } from "./workspace.ts"; export async function createApp( @@ -32,7 +34,7 @@ export async function createApp( config: Config, options: { docker?: DockerRunner } = {}, ) { - assertApiDeploymentConfig(config); + assertThreadsBackendConfig(config); const auth = await createAuth(db, config), files = new Files(db, config, auth), google = new GoogleAuth(db, config), @@ -47,8 +49,20 @@ export async function createApp( const browser = new BrowserService(db, config, auth, files); const computer = new ComputerService(db, config, options.docker); const agent = new AgentService(db, config, workspace, files, actions, browser, computer); - const intelligence = new CopilotKitIntelligence({ apiKey: config.intelligenceApiKey }); - const runtime = makeRuntime(config, agent, auth, intelligence); + // The runner's persist path receives no request context, so the agents factory + // (in makeRuntime) records each built agent's owner in this shared registry. + // The runtime clones the agent per request before handing it to the runner, + // so the WeakMap alone cannot resolve an owner; ConversationAgent carries + // its owner (clone() preserves it) and the map remains as a fallback. + const agentOwners = new WeakMap(); + const ownerOf = (agent: unknown) => + agent instanceof ConversationAgent + ? agent.owner + : typeof agent === "object" && agent !== null + ? agentOwners.get(agent) + : undefined; + const threads: ThreadProvider = createThreadProvider(config, { db, ownerOf }); + const runtime = makeRuntime(config, agent, auth, threads, agentOwners); const app = new Hono<{ Variables: { owner: string } }>(); const origins = new Set([...config.allowedOrigins, new URL(config.publicUrl).origin]); app.use("*", async (c, next) => { @@ -101,6 +115,7 @@ export async function createApp( mode: config.mode, agentConfigured: agentConfigured(config), browserConfigured: Boolean(config.workerUrl && config.workerToken), + threadsBackend: config.threadsBackend ?? "intelligence", }), ); let loginWindow = 0, @@ -215,14 +230,12 @@ export async function createApp( const main = await db.get<{ threadId: string }>(owner, "conversation-settings", "main"); if (!main) throw new AppError("Main conversation could not be loaded", 503); try { - await intelligence.getOrCreateThread({ - threadId: main.threadId, - userId: owner, - agentId: "default", - }); + await threads.getOrCreateMainThread({ owner, threadId: main.threadId }); } catch { throw new AppError( - "Main conversation is unavailable. Check the Rich Threads connection and try again.", + threads.backend === "intelligence" + ? "Main conversation is unavailable. Check the Rich Threads connection and try again." + : "Main conversation could not be saved. Check the server database and try again.", 502, ); } @@ -329,12 +342,142 @@ export async function createApp( await browser.input(c.get("owner"), c.req.param("id"), await c.req.json()); return c.json({ ok: true }); }); + // The runtime reports thread mutations as Intelligence-only, which makes the + // client SDK refuse rename/archive/delete before ever issuing a request. + // Local mode implements those endpoints owner-scoped below, so advertise + // them here; realtime metadata stays off. Intelligence passes through. + app.get("/api/copilotkit/info", async (c) => { + if (threads.backend !== "local") return proxyRuntime(c); + const response = await runtime.fetch(c.req.raw); + const body = (await response.json()) as { + threadEndpoints?: { mutations?: boolean; realtimeMetadata?: boolean }; + }; + if (body.threadEndpoints) body.threadEndpoints.mutations = true; + return c.json(body); + }); + // The runtime's own local thread-list fallback is process-global and cannot + // scope by owner, so local mode lists from the database instead. The static + // route wins over the /api/copilotkit/* wildcard below; intelligence mode + // forwards to the runtime untouched. + app.get("/api/copilotkit/threads", async (c) => { + if (threads.backend !== "local") return proxyRuntime(c); + const url = new URL(c.req.url); + const agentId = url.searchParams.get("agentId") ?? "default"; + const limit = Math.min(Math.max(Number(url.searchParams.get("limit")) || 20, 1), 100); + const offset = Math.max(Number(url.searchParams.get("cursor")) || 0, 0); + const records = (await db.list(c.get("owner"), CHAT_THREADS_KIND)).filter( + (record) => record.agentId === agentId, + ); + const page = records.slice(offset, offset + limit); + const nextOffset = offset + page.length; + return c.json({ + threads: page.map(threadListItem), + nextCursor: nextOffset < records.length ? String(nextOffset) : null, + }); + }); + // The runtime's local fallbacks for these endpoints are process-global with + // no owner concept: any signed-in user could read another owner's thread, and + // clear would wipe every owner. Local mode serves them owner-scoped from the + // database instead; other owners' threads are indistinguishable from missing + // ones (404, never 403) so the endpoints leak no existence information. + app.get("/api/copilotkit/threads/:id/messages", async (c) => + threadDetail(c, "messages", (record) => record.messages), + ); + app.get("/api/copilotkit/threads/:id/events", async (c) => + threadDetail(c, "events", (record) => record.runs.flatMap((run) => run.events)), + ); + app.get("/api/copilotkit/threads/:id/state", async (c) => threadDetail(c, "state", () => null)); + // Thread management is Intelligence-only in the runtime (422 without it). + // Local mode implements the same contract owner-scoped against the database + // so the app UI works identically on both backends. + app.patch("/api/copilotkit/threads/:id", async (c) => { + if (threads.backend !== "local") return proxyRuntime(c); + const body = z + .object({ + agentId: z.string().optional(), + name: z.string().min(1).optional(), + archived: z.boolean().optional(), + }) + .parse(await c.req.json().catch(() => ({}))); + const record = await ownedThread(c); + const updated: LocalThreadRecord = { + ...record, + ...(body.name !== undefined ? { name: body.name } : {}), + ...(body.archived !== undefined ? { archived: body.archived } : {}), + }; + await db.put(c.get("owner"), CHAT_THREADS_KIND, updated); + threads.runner.syncRecord(updated); + return c.json(threadListItem(updated)); + }); + app.post("/api/copilotkit/threads/:id/archive", async (c) => { + if (threads.backend !== "local") return proxyRuntime(c); + await c.req.json().catch(() => ({})); + const record = await ownedThread(c); + const updated: LocalThreadRecord = { ...record, archived: true }; + await db.put(c.get("owner"), CHAT_THREADS_KIND, updated); + threads.runner.syncRecord(updated); + return c.json({ threadId: record.id, archived: true }); + }); + app.delete("/api/copilotkit/threads/:id", async (c) => { + if (threads.backend !== "local") return proxyRuntime(c); + await c.req.json().catch(() => ({})); + const record = await ownedThread(c); + await db.remove(c.get("owner"), CHAT_THREADS_KIND, record.id); + threads.runner.deleteThread(record.id); + return c.json({ threadId: record.id, deleted: true }); + }); + app.post("/api/copilotkit/threads/clear", async (c) => { + if (threads.backend !== "local") return proxyRuntime(c); + const owner = c.get("owner"); + // Delete only this owner's durable rows; memory is process-global, so it + // is cleared whole — other owners rehydrate from the database on their + // next access. + const records = await db.list(owner, CHAT_THREADS_KIND); + for (const record of records) await db.remove(owner, CHAT_THREADS_KIND, record.id); + threads.runner.clearMemory(); + return c.json({ ok: true }); + }); + async function threadDetail( + c: Context, + key: "messages" | "events" | "state", + pick: (record: LocalThreadRecord) => unknown, + ) { + if (threads.backend !== "local") return proxyRuntime(c); + const record = await ownedThread(c); + return c.json({ [key]: pick(record) }); + } + /** Local-mode record lookup scoped to the authenticated owner: 404, never 403. */ + async function ownedThread(c: Context): Promise { + const id = c.req.param("id"); + if (!id) throw new AppError("Thread not found", 404); + const record = await db.get(c.get("owner"), CHAT_THREADS_KIND, id); + // Another owner's thread is indistinguishable from a missing one, so the + // endpoint leaks no existence information. + if (!record) throw new AppError("Thread not found", 404); + return record; + } + /** The list shape the CopilotKit SDK parses; shared by list and rename. */ + function threadListItem(record: LocalThreadRecord) { + return { + id: record.id, + name: record.name, + agentId: record.agentId, + organizationId: "", + createdById: "", + archived: record.archived, + createdAt: record.createdAt, + updatedAt: record.updatedAt, + }; + } app.all("/api/copilotkit/*", async (c) => { if (!agentConfigured(config)) throw new AppError( "Configure a model and provider API key, or a valid AG-UI endpoint, to start chat", 503, ); + return proxyRuntime(c); + }); + async function proxyRuntime(c: { req: { raw: Request } }) { const response = await runtime.fetch(c.req.raw); // Runtime 1.70 emits SSE strings; a WHATWG Response body requires byte chunks. const encoder = new TextEncoder(); @@ -346,7 +489,7 @@ export async function createApp( }), ); return new Response(body, { status: response.status, headers: response.headers }); - }); + } app.get("/", (c) => c.json({ name: "OpenMuse", app: "http://localhost:8081", health: "/api/health" }), ); diff --git a/apps/server/src/config.ts b/apps/server/src/config.ts index d34a90929..cb84709c9 100644 --- a/apps/server/src/config.ts +++ b/apps/server/src/config.ts @@ -44,6 +44,8 @@ export interface Config { agentUrl?: string; agentToken?: string; intelligenceApiKey?: string; + /** Set by readConfig; undefined only in hand-built fixtures, meaning the default. */ + threadsBackend?: ThreadsBackend; googleClientId?: string; googleClientSecret?: string; googleRedirectUri: string; @@ -63,6 +65,10 @@ export interface Config { export type ComputerProvider = "docker" | "e2b-desktop"; +/** Chat thread persistence backend. Intelligence is the CopilotKit-hosted default; + * "local" keeps threads on the server's own database with no external service. */ +export type ThreadsBackend = "intelligence" | "local"; + /** Pinned so live rankings do not shift when TypeSafe moves the `jev-latest` alias. */ export const defaultJevModel = "jev-1.13.0"; @@ -77,14 +83,17 @@ export function required(name: string, message: string, value = process.env[name return value.trim(); } -export function assertApiDeploymentConfig( +export function assertThreadsBackendConfig( config: Config, ): asserts config is Config & { intelligenceApiKey: string } { - required( - "CPK_INTELLIGENCE_API_KEY", - intelligenceKeyRequiredMessage, - config.intelligenceApiKey ?? "", - ); + // An explicit Intelligence selection without a key fails loudly at startup; + // it never silently downgrades to local persistence. + if ((config.threadsBackend ?? "intelligence") === "intelligence") + required( + "CPK_INTELLIGENCE_API_KEY", + intelligenceKeyRequiredMessage, + config.intelligenceApiKey ?? "", + ); } /** Accept a full worker URL, or host:port from a platform that omits the scheme. */ @@ -131,6 +140,9 @@ export function readConfig(): Config { "COMPUTER_PROVIDER=e2b-desktop requires a unique COMPUTER_DEPLOYMENT_ID, e.g. from `openssl rand -hex 12`", ); } + const threadsBackend = process.env.THREADS_BACKEND ?? "intelligence"; + if (threadsBackend !== "intelligence" && threadsBackend !== "local") + throw new Error("THREADS_BACKEND must be intelligence or local"); const port = Number(process.env.PORT ?? 8787); const publicUrl = process.env.PUBLIC_API_URL ?? `http://localhost:${port}`; const config: Config = { @@ -149,7 +161,8 @@ export function readConfig(): Config { agentBackend: backend, agentUrl: process.env.AGENT_URL, agentToken: process.env.AGENT_TOKEN, - intelligenceApiKey: required("CPK_INTELLIGENCE_API_KEY", intelligenceKeyRequiredMessage), + intelligenceApiKey: process.env.CPK_INTELLIGENCE_API_KEY?.trim(), + threadsBackend, googleClientId: process.env.GOOGLE_CLIENT_ID, googleClientSecret: process.env.GOOGLE_CLIENT_SECRET, googleRedirectUri: `${publicUrl}/api/google/callback`, diff --git a/apps/server/src/demo/entry.ts b/apps/server/src/demo/entry.ts index d399c6182..e9229b232 100644 --- a/apps/server/src/demo/entry.ts +++ b/apps/server/src/demo/entry.ts @@ -8,7 +8,15 @@ import { createStore } from "../db.ts"; import { createDemoModel, demoModel } from "./model.ts"; const root = fileURLToPath(new URL("../../../../", import.meta.url)); -const intelligenceApiKey = required("CPK_INTELLIGENCE_API_KEY", intelligenceKeyRequiredMessage); +const intelligenceApiKey = process.env.CPK_INTELLIGENCE_API_KEY?.trim(); +// An explicit choice wins; otherwise the demo follows the key. Either way the +// child process gets THREADS_BACKEND verbatim, never a silent downgrade. +const threadsBackend = + process.env.THREADS_BACKEND ?? (intelligenceApiKey ? "intelligence" : "local"); +if (threadsBackend !== "intelligence" && threadsBackend !== "local") + throw new Error("THREADS_BACKEND must be intelligence or local"); +if (threadsBackend === "intelligence" && !intelligenceApiKey) + throw new Error(intelligenceKeyRequiredMessage); const jevMode = process.env.DEMO_JEV_MODE ?? "sample"; if (jevMode !== "sample" && jevMode !== "live") throw new Error("DEMO_JEV_MODE must be sample or live"); @@ -96,7 +104,8 @@ const api = spawn( process.env.DEMO_ALLOWED_ORIGINS ?? "http://localhost:8081,http://127.0.0.1:8081", DO_NOT_TRACK: "1", COPILOTKIT_TELEMETRY_DISABLED: "true", - CPK_INTELLIGENCE_API_KEY: intelligenceApiKey, + ...(intelligenceApiKey ? { CPK_INTELLIGENCE_API_KEY: intelligenceApiKey } : {}), + THREADS_BACKEND: threadsBackend, }, }, ); diff --git a/apps/server/src/engine/conversation.ts b/apps/server/src/engine/conversation.ts index e2c0a57b7..f1df63c7b 100644 --- a/apps/server/src/engine/conversation.ts +++ b/apps/server/src/engine/conversation.ts @@ -30,7 +30,8 @@ export class ConversationAgent extends AbstractAgent { constructor( private readonly config: Config, private readonly service: AgentService, - private readonly owner: string, + /** Read by the local threads backend after the runtime clones the agent per request. */ + readonly owner: string, private readonly jevAdapter: JevAdapter | undefined = createJevAdapter(config), ) { super({ agentId: "default" }); diff --git a/apps/server/src/threads/local-runner.ts b/apps/server/src/threads/local-runner.ts new file mode 100644 index 000000000..c6bba5351 --- /dev/null +++ b/apps/server/src/threads/local-runner.ts @@ -0,0 +1,458 @@ +import { type BaseEvent, EventType } from "@ag-ui/client"; +import type { AgentRunnerConnectRequest, AgentRunnerRunRequest } from "@copilotkit/runtime/v2"; +import { InMemoryAgentRunner, ɵGLOBAL_STORE } from "@copilotkit/runtime/v2"; +import { type AnySummarizeAdapter, summarize } from "@tanstack/ai"; +import { anthropicSummarize } from "@tanstack/ai-anthropic"; +import { geminiSummarize } from "@tanstack/ai-gemini"; +import { Observable, ReplaySubject } from "rxjs"; +import type { Store } from "../db.ts"; +import { backgroundFailure } from "../log.ts"; + +export const THREADS_CONTRACT_REF = "@copilotkit/runtime 1.70.1"; + +export const CHAT_THREADS_KIND = "chat-threads"; + +const TITLE_MAX_LENGTH = 60; + +/** Turns a user's opening message into a conversation title, or null to skip. */ +export type TitleGenerator = (text: string) => Promise; + +/** + * Single-line title text capped at TITLE_MAX_LENGTH chars. Newlines collapse + * to spaces so a message never breaks the one-line thread list layout. + */ +export function fallbackTitle(text: string): string { + const singleLine = text.replace(/\s+/g, " ").trim(); + return singleLine.slice(0, TITLE_MAX_LENGTH); +} + +function cleanTitle(summary: string): string | null { + const title = summary + .replace(/^["'`“”‘’]+|["'`“”‘’]+$/g, "") + .replace(/\s+/g, " ") + .trim(); + return title ? fallbackTitle(title) : null; +} + +/** + * Default title chain: ask the configured model for a short title (mirroring + * the Intelligence backend's generateThreadNames), and fall back to a plain + * truncation of the first user message when the model is unconfigured or the + * call fails. + * + * OpenAI-compatible providers (including Moonshot/Kimi gateways) go through a + * direct chat-completions fetch instead of the @tanstack summarize activity: + * the activity maps maxLength to provider-native keys across several adapter + * layers, and gateways like Moonshot reject or truncate under those mappings + * (kimi-k3 only allows temperature=1 and spends ~200 reasoning tokens before + * the summary text). A direct request keeps every knob explicit and the + * failure surface is a single catch. + */ +export function defaultTitleGenerator(model: string | undefined): TitleGenerator { + return async (text) => { + const clip = fallbackTitle(text); + if (!clip) return null; + const [, provider = "", modelId = ""] = model?.trim().match(/^([^/:]*)[/:](.*)$/) ?? []; + if (!provider || !modelId.trim()) return clip; + try { + if (provider.toLowerCase() === "openai") { + const baseURL = (process.env.OPENAI_BASE_URL || "https://api.openai.com/v1").replace( + /\/+$/, + "", + ); + const apiKey = process.env.OPENAI_API_KEY; + if (!apiKey) return clip; + // Moonshot/Kimi rejects any temperature other than 1; plain OpenAI + // (0-2) accepts 1 as well. 256 completion tokens leaves room for + // reasoning tokens before the short summary text. + const response = await fetch(`${baseURL}/chat/completions`, { + method: "POST", + headers: { "Content-Type": "application/json", Authorization: `Bearer ${apiKey}` }, + body: JSON.stringify({ + model: modelId.trim(), + messages: [ + { + role: "user", + content: `用一句话概括以下内容的主题，直接输出标题本身，不要加引号、标点或解释：${clip}`, + }, + ], + temperature: 1, + max_tokens: 256, + }), + }); + if (!response.ok) return clip; + const body = (await response.json()) as { + choices?: { message?: { content?: string } }[]; + }; + return cleanTitle(body.choices?.[0]?.message?.content ?? "") ?? clip; + } + const id = modelId.trim(); + const adapter = ( + provider.toLowerCase() === "anthropic" + ? anthropicSummarize(id as never, { baseURL: process.env.ANTHROPIC_BASE_URL } as never) + : provider.toLowerCase() === "google" + ? geminiSummarize( + id as never, + { + baseURL: process.env.GOOGLE_GENERATIVE_AI_BASE_URL, + } as never, + ) + : null + ) as AnySummarizeAdapter | null; + if (!adapter) return clip; + const result = await summarize({ + adapter, + text: clip, + maxLength: 10, + style: "concise", + modelOptions: { temperature: 1, max_tokens: 256 }, + }); + return cleanTitle(result.summary) ?? clip; + } catch { + // A title must never surface as a chat error; the truncated message is + // a perfectly good stand-in. + return clip; + } + }; +} + +/** One persisted run: its compacted events, keyed for merge-on-write. */ +export interface LocalThreadRun { + runId: string; + createdAt: string; + events: unknown[]; +} + +/** + * Durable per-thread record stored in the server's own database. `runs` grows + * incrementally: memory eviction in the base runner may drop old runs from + * the in-memory store, so a persist must merge new runs into the stored set + * instead of overwriting it. `events` is the v1 shape, read for migration. + */ +export interface LocalThreadRecord { + id: string; + owner: string; + agentId: string; + name: string | null; + archived: boolean; + createdAt: string; + updatedAt: string; + /** Thread-level message snapshot (AG-UI Message[]) replayed on connect. */ + messages: unknown[]; + runs: LocalThreadRun[]; + /** v1 records stored all events at thread level; migrated on read. */ + events?: unknown[]; +} + +type OwnerLookup = (agent: unknown) => string | undefined; + +type MessageLike = { role?: unknown; content?: unknown }; + +/** Plain text of the first user message in a snapshot, for titling. */ +function firstUserText(messages: unknown): string | null { + if (!Array.isArray(messages)) return null; + for (const raw of messages) { + const message = raw as MessageLike; + if (message?.role !== "user") continue; + if (typeof message.content === "string" && message.content.trim()) + return message.content.trim(); + if (Array.isArray(message.content)) { + for (const part of message.content as { type?: unknown; text?: unknown }[]) { + if (part && part.type === "text" && typeof part.text === "string" && part.text.trim()) + return part.text.trim(); + } + } + } + return null; +} + +/** The historic-run fields this class reads, decoupled from runtime internals. */ +type HistoricRun = { + runId?: unknown; + createdAt?: unknown; + events?: unknown; +}; + +/** + * InMemoryAgentRunner durably backed by the OpenMuse store. The base runner + * keeps everything in a process-global store that dies with the process; this + * subclass rehydrates threads from the database before run/connect and merges + * each finalized run into the durable record. + * + * Durability boundary: the terminal RUN_FINISHED event is held until the + * snapshot write completes, so a client never sees a completed run that a + * restart would lose. A failed write on a successful run is reported as a + * visible RUN_ERROR; a failed run always terminates the stream cleanly. + */ +export class PersistentAgentRunner extends InMemoryAgentRunner { + /** threadId -> record, so owner/name survive restarts without a scan. */ + private readonly index = new Map(); + private readonly scanIndex = new Map(); + + constructor( + private readonly db: Store, + private readonly ownerOf: OwnerLookup, + private readonly titleGenerator: TitleGenerator = () => Promise.resolve(null), + ) { + super(); + } + + override run(request: AgentRunnerRunRequest): Observable { + const subject = new ReplaySubject(Infinity); + this.hydrate(request.threadId).then( + () => { + // A concurrent run on the thread throws synchronously; deliver it as + // an observable error instead of an unhandled rejection or a hang. + let source: Observable; + try { + source = super.run(request); + } catch (error) { + subject.error(error); + return; + } + this.gatePersistence(source, request.threadId, request.agent).subscribe(subject); + }, + (error: unknown) => subject.error(error), + ); + return subject.asObservable(); + } + + override connect(request: AgentRunnerConnectRequest): Observable { + const subject = new ReplaySubject(Infinity); + this.hydrate(request.threadId).then( + () => void super.connect(request).subscribe(subject), + // A failed hydration must not break the connection: replay continues + // from memory (empty for a fresh thread), the failure is logged. + (error: unknown) => { + backgroundFailure("chat thread hydration", error); + super.connect(request).subscribe(subject); + }, + ); + return subject.asObservable(); + } + + /** Clears in-memory history and every durable record. The runtime handler + * does not await the returned promise; failures are logged, not thrown. */ + override clearThreads(): Promise { + super.clearThreads(); + return this.db + .scan(CHAT_THREADS_KIND) + .then((records) => + Promise.all( + records.map(({ owner, value }) => + this.db.remove(value.owner ?? owner, CHAT_THREADS_KIND, value.id), + ), + ), + ) + .then(() => { + this.index.clear(); + this.scanIndex.clear(); + }) + .catch((error: unknown) => backgroundFailure("chat thread clear", error)); + } + + /** Memory-only clear for the app route: durable rows are deleted per owner there. */ + clearMemory(): void { + super.clearThreads(); + this.index.clear(); + this.scanIndex.clear(); + } + + /** Forget one thread in memory; the app route deletes the durable row per owner. */ + deleteThread(threadId: string): void { + const store = ɵGLOBAL_STORE.peek(threadId); + if (store) { + // removeThread exists at runtime (in-memory.mjs:222) but is typed private; + // the runner has no single-thread delete, so reach the one store primitive. + (ɵGLOBAL_STORE as unknown as { removeThread(id: string, store: unknown): void }).removeThread( + threadId, + store, + ); + } + this.index.delete(threadId); + this.scanIndex.delete(threadId); + } + + /** Keep the runner's caches in step after the app layer renames/archives a thread. */ + syncRecord(record: LocalThreadRecord): void { + this.index.set(record.id, record); + this.scanIndex.set(record.id, record); + } + + /** + * Forward every event except the terminal RUN_FINISHED, which is held until + * the durable write settles. The base runner finalizes (appendRun) + * synchronously after emitting RUN_FINISHED, so the persist defers one + * macrotask and reads the finalized store. A run that ends without + * RUN_FINISHED (internal failure) still persists what ran and completes. + */ + private gatePersistence( + source: Observable, + threadId: string, + agent: AgentRunnerRunRequest["agent"], + ): Observable { + return new Observable((subscriber) => { + let finishEvent: BaseEvent | undefined; + let persisted: Promise | undefined; + let sourceComplete = false; + const close = (errorEvent: BaseEvent | undefined) => { + if (subscriber.closed) return; + if (errorEvent) subscriber.next(errorEvent); + subscriber.complete(); + }; + const flush = () => { + if (!sourceComplete) return; + if (persisted) { + void persisted.then( + () => close(finishEvent), + (error: unknown) => { + backgroundFailure("chat thread persistence", error); + close({ + type: EventType.RUN_ERROR, + message: + "The run completed but the conversation could not be saved. " + + "Check the server database and try again.", + } as BaseEvent); + }, + ); + } else { + // No RUN_FINISHED means the run already failed; the runner appended + // a RUN_ERROR. Save what exists (best effort) and close the stream. + void new Promise((resolve) => setImmediate(resolve)) + .then(() => this.persist(threadId, agent)) + .catch((error: unknown) => backgroundFailure("chat thread persistence", error)) + .finally(() => close(undefined)); + } + }; + const subscription = source.subscribe({ + next: (event) => { + if (event.type === EventType.RUN_FINISHED) { + finishEvent = event; + persisted = new Promise((resolve) => setImmediate(resolve)).then(() => + this.persist(threadId, agent), + ); + } else if (!subscriber.closed) { + subscriber.next(event); + } + }, + error: (error: unknown) => { + if (!subscriber.closed) subscriber.error(error); + }, + complete: () => { + sourceComplete = true; + flush(); + }, + }); + return () => subscription.unsubscribe(); + }); + } + + /** Load a thread's record into the process-global store unless it is already there. */ + private async hydrate(threadId: string): Promise { + if (ɵGLOBAL_STORE.peek(threadId)) return; + const record = await this.findRecord(threadId); + if (!record || ɵGLOBAL_STORE.peek(threadId)) return; // a concurrent run hydrated first + const store = ɵGLOBAL_STORE.getOrCreate(threadId); + store.createdAt = Date.parse(record.createdAt); + store.messagesSnapshot = record.messages as never; + // Rebuild historic runs exactly as appendRun would have produced them, so + // connect/replay and further appends work unchanged. + store.historicRuns = this.recordRuns(record).map( + (run) => + ({ + threadId, + runId: run.runId, + agentId: record.agentId, + parentRunId: null, + events: run.events, + messages: [], + createdAt: Date.parse(run.createdAt), + }) as never, + ); + } + + /** Write the finalized thread snapshot back, merging runs already stored. */ + private async persist(threadId: string, agent: unknown): Promise { + const store = ɵGLOBAL_STORE.peek(threadId); + if (!store) return; + const prior = await this.findRecord(threadId); + const owner = this.ownerOf(agent) ?? prior?.owner; + if (!owner) throw new Error("No owner is associated with the running agent"); + const priorRuns = prior ? this.recordRuns(prior) : []; + const known = new Set(priorRuns.map((run) => run.runId)); + const freshRuns = (store.historicRuns as HistoricRun[]) + .filter( + (run) => + typeof run.runId === "string" && + run.runId !== "persisted" && + !known.has(run.runId) && + Array.isArray(run.events), + ) + .map((run) => ({ + runId: run.runId as string, + createdAt: new Date(Number(run.createdAt) || Date.now()).toISOString(), + events: run.events as unknown[], + })); + // A run that produced no events (immediate failure) must not create an + // empty or clobbering record; the prior durable state is authoritative. + if (freshRuns.length === 0) return; + const record: LocalThreadRecord = { + id: threadId, + owner, + agentId: prior?.agentId ?? "default", + name: prior?.name ?? null, + archived: prior?.archived ?? false, + createdAt: prior?.createdAt ?? new Date(store.createdAt ?? Date.now()).toISOString(), + updatedAt: new Date().toISOString(), + messages: store.messagesSnapshot as unknown[], + runs: [...priorRuns, ...freshRuns], + }; + await this.db.put(owner, CHAT_THREADS_KIND, record); + this.index.set(threadId, record); + this.scanIndex.set(threadId, record); + if (record.name === null) this.titleThread(record); + } + + /** + * Name an untitled thread after its first run. Fire-and-forget: a title is + * cosmetic and must never delay RUN_FINISHED or surface as a chat error. + * An untitled record is retried on the next persist, so a transient model + * failure self-heals. + */ + private titleThread(record: LocalThreadRecord): void { + const text = firstUserText(record.messages); + if (!text) return; + void this.titleGenerator(text) + .then((title) => { + if (!title) return; + const current = this.index.get(record.id) ?? this.scanIndex.get(record.id); + // The thread may have been renamed or deleted while the title call was in flight. + if (!current || current.name !== null) return; + const named = { ...current, name: title }; + return this.db.put(current.owner, CHAT_THREADS_KIND, named).then(() => { + this.index.set(record.id, named); + this.scanIndex.set(record.id, named); + }); + }) + .catch((error: unknown) => backgroundFailure("chat thread titling", error)); + } + + /** Runs of a record, migrating the v1 thread-level `events` shape on read. */ + private recordRuns(record: LocalThreadRecord): LocalThreadRun[] { + if (Array.isArray(record.runs) && record.runs.length > 0) return record.runs; + if (Array.isArray(record.events) && record.events.length > 0) + return [{ runId: "migrated", createdAt: record.createdAt, events: record.events }]; + return []; + } + + /** Resolve a record by thread id; owners are not knowable from a thread id alone. */ + private async findRecord(threadId: string): Promise { + const cached = this.index.get(threadId) ?? this.scanIndex.get(threadId); + if (cached) return cached; + for (const { owner, value } of await this.db.scan(CHAT_THREADS_KIND)) { + this.scanIndex.set(value.id, { ...value, owner: value.owner ?? owner }); + if (value.id === threadId) return this.scanIndex.get(threadId) ?? null; + } + return null; + } +} diff --git a/apps/server/src/threads/provider.ts b/apps/server/src/threads/provider.ts new file mode 100644 index 000000000..e804b0606 --- /dev/null +++ b/apps/server/src/threads/provider.ts @@ -0,0 +1,71 @@ +import { CopilotKitIntelligence } from "@copilotkit/runtime/v2"; +import type { Config, ThreadsBackend } from "../config.ts"; +import { intelligenceKeyRequiredMessage } from "../config.ts"; +import type { Store } from "../db.ts"; +import { CHAT_THREADS_KIND, defaultTitleGenerator, PersistentAgentRunner } from "./local-runner.ts"; + +/** + * Persistence boundary for chat threads. "intelligence" is the CopilotKit-hosted + * default; "local" keeps threads on the server's own database and never makes + * Intelligence requests. See docs/PLUGGABLE-THREADS.md for the capability matrix. + */ +export type ThreadProvider = + | { + backend: "intelligence"; + intelligence: CopilotKitIntelligence; + getOrCreateMainThread(input: { owner: string; threadId: string }): Promise; + } + | { + backend: "local"; + runner: PersistentAgentRunner; + getOrCreateMainThread(input: { owner: string; threadId: string }): Promise; + }; + +export function createThreadProvider( + config: Config, + deps: { db: Store; ownerOf: (agent: unknown) => string | undefined }, +): ThreadProvider { + const backend: ThreadsBackend = config.threadsBackend ?? "intelligence"; + if (backend === "intelligence") { + // assertThreadsBackendConfig guards the real startup path; this keeps the + // factory honest when it is built directly (tests, demos). + if (!config.intelligenceApiKey?.trim()) throw new Error(intelligenceKeyRequiredMessage); + const intelligence = new CopilotKitIntelligence({ apiKey: config.intelligenceApiKey }); + return { + backend: "intelligence", + intelligence, + getOrCreateMainThread: ({ owner, threadId }) => + intelligence + .getOrCreateThread({ threadId, userId: owner, agentId: "default" }) + .then(() => undefined), + }; + } + if (config.intelligenceApiKey) + console.warn("[OpenMuse] THREADS_BACKEND=local ignores CPK_INTELLIGENCE_API_KEY."); + const runner = new PersistentAgentRunner( + deps.db, + deps.ownerOf, + defaultTitleGenerator(config.model), + ); + const { db } = deps; + return { + backend: "local", + runner, + getOrCreateMainThread: async ({ owner, threadId }) => { + const now = new Date().toISOString(); + await db.insertIfAbsent(owner, CHAT_THREADS_KIND, { + id: threadId, + owner, + agentId: "default", + name: null, + archived: false, + createdAt: now, + updatedAt: now, + messages: [], + runs: [], + }); + }, + }; +} + +export type { ThreadsBackend }; diff --git a/apps/server/src/workspace.ts b/apps/server/src/workspace.ts index 1bbe4ae51..3192d35ef 100644 --- a/apps/server/src/workspace.ts +++ b/apps/server/src/workspace.ts @@ -365,6 +365,7 @@ export class WorkspaceService { configured: agentConfigured(this.config), openbotConfigured: false, richThreads: true, + threadsBackend: this.config.threadsBackend ?? "intelligence", }, }; } diff --git a/docs/PLUGGABLE-THREADS.md b/docs/PLUGGABLE-THREADS.md new file mode 100644 index 000000000..1cdd2527a --- /dev/null +++ b/docs/PLUGGABLE-THREADS.md @@ -0,0 +1,75 @@ +# Pluggable thread persistence + +OpenMuse keeps chat threads behind a provider boundary with two backends, +selected by `THREADS_BACKEND`: + +| `THREADS_BACKEND` | `CPK_INTELLIGENCE_API_KEY` | Behavior | +|---|---|---| +| `intelligence` (default) | set | Current behavior, unchanged: CopilotKit Intelligence persists and replays threads. | +| `intelligence` | missing/blank | The API **refuses to start** with the Intelligence key message. There is never a silent downgrade to local storage. | +| `local` | set or missing | Threads persist in the server's own database (PGlite, or PostgreSQL with `DATABASE_URL`). No Intelligence requests are made; a set key is ignored with a one-line warning. | +| anything else | — | The API refuses to start: `THREADS_BACKEND must be intelligence or local`. | + +## How the local backend works + +`CopilotRuntime` dispatches on its options: with `intelligence` it builds the +Intelligence runtime; with `runner` (and no `intelligence`) it builds the SSE +runtime and serves thread endpoints from that runner. The local backend is a +`PersistentAgentRunner` (`apps/server/src/threads/local-runner.ts`) that +subclasses the runtime's own `InMemoryAgentRunner`: + +- **Hydrate**: before `run`/`connect`, the thread's snapshot is loaded from the + database (`chat-threads` records: message snapshot plus compacted events) + into the runner's store, so history survives API restarts. +- **Durability boundary**: the terminal `RUN_FINISHED` event is held until the + snapshot write completes. A client never sees a completed run that a restart + would lose. A failed write surfaces as a visible `RUN_ERROR`, not a silent + loss. +- **Owner scoping**: the agents factory records the owning user on each agent + it builds; saved records carry that owner, and the thread list is served per + owner from the database (the runtime's built-in local list fallback is + process-global and cannot scope). + +The only internal CopilotKit API this touches is the process-global in-memory +store used for hydration/snapshot (pinned in `THREADS_CONTRACT_REF`); a runtime +upgrade that moves it breaks compilation in that one module. + +## Capability matrix + +Local mode implements the same thread-management contract as Intelligence, +owner-scoped against the database. The app UI is identical on both backends. + +| Capability | intelligence | local | +|---|---|---| +| Main conversation, side chats | ✓ | ✓ | +| Thread list, pagination | ✓ | ✓ (owner-scoped, database-backed) | +| Replay on open (`connect`) | ✓ | ✓ | +| Inspect messages/events/state | ✓ | ✓ | +| Clear threads | ✓ | ✓ | +| Rename | ✓ | ✓ (`PATCH`, owner-scoped) | +| Archive / restore | ✓ | ✓ (owner-scoped) | +| Delete thread | ✓ | ✓ (owner-scoped) | +| Thread naming | cloud-generated | generated by the configured `MODEL`, falling back to a truncated first message | +| Realtime thread metadata | ✓ | unavailable (the client SDK skips the subscription) | + +Thread naming runs after a run persists: the runner asks the configured model +for a short title and, on any failure (no model, provider error), falls back to +the first user message truncated to one line. Titration never delays +`RUN_FINISHED` and never overwrites a title that is already set. + +## Operator notes + +- Set `THREADS_BACKEND=local` to run fully self-hosted with no CopilotKit + account and no Intelligence key. Chat history then lives (and dies) with the + `DATA_DIR`/`DATABASE_URL` database — back it up like the rest of the data. +- Switching backends is not a migration: local threads start empty, and + Intelligence threads stay in Intelligence. Import/export across backends is + out of scope. +- The demo harness (`pnpm dev:demo`) follows the same setting: with a key it + uses Intelligence, without one it runs keyless on the local backend. +- `POST /api/copilotkit/threads/clear` wipes local thread records as well as + in-memory history. + +See [RICH-THREADS.md](RICH-THREADS.md) for the Intelligence path and +[OPENBOT-INTEGRATION.md](OPENBOT-INTEGRATION.md) for the adapter seam pattern +this provider follows. diff --git a/packages/domain/src/index.ts b/packages/domain/src/index.ts index 764ab9c8a..1fb9adb45 100644 --- a/packages/domain/src/index.ts +++ b/packages/domain/src/index.ts @@ -195,6 +195,7 @@ export interface Workspace { configured: boolean; openbotConfigured: boolean; richThreads?: boolean; + threadsBackend?: "intelligence" | "local"; }; } diff --git a/tests/config.test.ts b/tests/config.test.ts index b323dd3b9..04b3c9cc7 100644 --- a/tests/config.test.ts +++ b/tests/config.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import { test } from "node:test"; import { - assertApiDeploymentConfig, + assertThreadsBackendConfig, browserWorkerUrl, type Config, readConfig, @@ -17,6 +17,7 @@ const sampleConfig: Config = { agentBackend: "sample", googleRedirectUri: "http://localhost:8787/api/google/callback", allowedOrigins: ["http://localhost:8081"], + threadsBackend: "intelligence", }; function liveConfig(intelligenceApiKey?: string): Config { @@ -37,7 +38,7 @@ const missingKeyMessage = test("every API mode rejects a missing or blank Intelligence key", () => { for (const mode of [sampleConfig, liveConfig()]) { for (const key of [undefined, "", " \t\n"]) { - assert.throws(() => assertApiDeploymentConfig({ ...mode, intelligenceApiKey: key }), { + assert.throws(() => assertThreadsBackendConfig({ ...mode, intelligenceApiKey: key }), { name: "Error", message: missingKeyMessage, }); @@ -48,11 +49,48 @@ test("every API mode rejects a missing or blank Intelligence key", () => { test("every API mode accepts a non-empty Intelligence key", () => { for (const mode of [sampleConfig, liveConfig()]) { assert.doesNotThrow(() => - assertApiDeploymentConfig({ ...mode, intelligenceApiKey: "test-project-key-never-sent" }), + assertThreadsBackendConfig({ ...mode, intelligenceApiKey: "test-project-key-never-sent" }), ); } }); +test("the local threads backend starts without an Intelligence key", () => { + for (const mode of [sampleConfig, liveConfig()]) { + assert.doesNotThrow(() => + assertThreadsBackendConfig({ + ...mode, + threadsBackend: "local", + intelligenceApiKey: undefined, + }), + ); + } +}); + +test("readConfig rejects an unknown THREADS_BACKEND", (t) => { + const previous = { ...process.env }; + t.after(() => { + process.env = previous; + }); + process.env.WORKSPACE_MODE = "sample"; + process.env.CPK_INTELLIGENCE_API_KEY = "test-project-key-never-sent"; + process.env.THREADS_BACKEND = "sqlite"; + assert.throws(() => readConfig(), /THREADS_BACKEND must be intelligence or local/); +}); + +test("readConfig defaults to the intelligence backend and reports local explicitly", (t) => { + const previous = { ...process.env }; + t.after(() => { + process.env = previous; + }); + process.env.WORKSPACE_MODE = "sample"; + process.env.CPK_INTELLIGENCE_API_KEY = "test-project-key-never-sent"; + delete process.env.THREADS_BACKEND; + assert.equal(readConfig().threadsBackend, "intelligence"); + process.env.THREADS_BACKEND = "local"; + assert.equal(readConfig().threadsBackend, "local"); + assert.equal(readConfig().intelligenceApiKey, "test-project-key-never-sent"); +}); + test("web search is enabled by default with an explicit opt-out", (t) => { const previous = { ...process.env }; t.after(() => { diff --git a/tests/local-threads.test.ts b/tests/local-threads.test.ts new file mode 100644 index 000000000..ce2ed522f --- /dev/null +++ b/tests/local-threads.test.ts @@ -0,0 +1,293 @@ +import assert from "node:assert/strict"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { after, before, test } from "node:test"; +import { CopilotKitIntelligence } from "@copilotkit/runtime/v2"; +import { createApp } from "../apps/server/src/app.ts"; +import { createStore, type Store } from "../apps/server/src/db.ts"; +import { + CHAT_THREADS_KIND, + type LocalThreadRecord, +} from "../apps/server/src/threads/local-runner.ts"; + +let db: Store, directory: string, token: string; +let app: Awaited>["app"]; +const headers = () => ({ Authorization: `Bearer ${token}`, "Content-Type": "application/json" }); + +before(async () => { + directory = await mkdtemp(join(tmpdir(), "openmuse-local-threads-")); + db = await createStore(); + // No CPK_INTELLIGENCE_API_KEY at all: local mode must boot and stay keyless. + ({ app } = await createApp(db, { + mode: "sample", + port: 8787, + host: "127.0.0.1", + publicUrl: "http://localhost:8787", + dataDir: directory, + agentBackend: "sample", + googleRedirectUri: "http://localhost:8787/api/google/callback", + allowedOrigins: ["http://localhost:8081"], + threadsBackend: "local", + })); + const session = await app.request("/api/session", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: "{}", + }); + token = (await session.json()).token; +}); +after(async () => { + await db.close(); + await rm(directory, { recursive: true, force: true }); +}); + +test("health reports the local threads backend", async () => { + const health = await app.request("/api/health"); + assert.equal((await health.json()).threadsBackend, "local"); +}); + +test("the main thread is provisioned locally with no Intelligence contact", async (t) => { + let intelligenceCalls = 0; + t.mock.method(CopilotKitIntelligence.prototype, "getOrCreateThread", async () => { + intelligenceCalls += 1; + return { id: "unused" }; + }); + const first = await (await app.request("/api/main-thread", { headers: headers() })).json(); + const reopened = await (await app.request("/api/main-thread", { headers: headers() })).json(); + assert.equal(first.existing, true); + assert.equal(reopened.threadId, first.threadId); + assert.equal(intelligenceCalls, 0, "local mode never reaches for Intelligence"); + const record = await db.get("local-user", CHAT_THREADS_KIND, first.threadId); + assert.ok(record, "the main thread has a durable local record"); + assert.equal(record.owner, "local-user"); +}); + +test("thread listing is owner-scoped from the database and paginates", async () => { + const main = await (await app.request("/api/main-thread", { headers: headers() })).json(); + for (let index = 0; index < 2; index++) { + const now = new Date().toISOString(); + await db.put("local-user", CHAT_THREADS_KIND, { + id: `side-${index}`, + owner: "local-user", + agentId: "default", + name: `Side ${index}`, + archived: false, + createdAt: now, + updatedAt: now, + messages: [], + runs: [], + }); + } + const seen: string[] = []; + let cursor: string | null = null; + for (let page = 0; page < 3; page++) { + const query = `agentId=default&limit=1${cursor ? `&cursor=${cursor}` : ""}`; + const response = await app.request(`/api/copilotkit/threads?${query}`, { headers: headers() }); + assert.equal(response.status, 200, await response.clone().text()); + const body = await response.json(); + assert.equal(body.threads.length, 1); + seen.push(body.threads[0].id); + cursor = body.nextCursor; + } + assert.equal(cursor, null, "the third page exhausts the list"); + assert.equal(new Set(seen).size, 3, "pages do not overlap or repeat"); + assert.ok(seen.includes(main.threadId)); + assert.equal( + (await app.request("/api/copilotkit/threads?agentId=default")).status, + 401, + "listing still requires authentication", + ); +}); + +test("threads can be renamed, archived, restored and deleted in local mode", async () => { + const now = new Date().toISOString(); + await db.put("local-user", CHAT_THREADS_KIND, { + id: "manage-me", + owner: "local-user", + agentId: "default", + name: null, + archived: false, + createdAt: now, + updatedAt: now, + messages: [], + runs: [], + }); + // Rename returns the updated thread in the list shape the SDK parses. + const renamed = await app.request("/api/copilotkit/threads/manage-me", { + method: "PATCH", + headers: headers(), + body: JSON.stringify({ agentId: "default", name: "Weekly report" }), + }); + assert.equal(renamed.status, 200, await renamed.clone().text()); + assert.equal((await renamed.json()).name, "Weekly report"); + // Archive moves it out of the active list into the archived filter. + const archived = await app.request("/api/copilotkit/threads/manage-me/archive", { + method: "POST", + headers: headers(), + body: JSON.stringify({ agentId: "default" }), + }); + assert.equal(archived.status, 200); + assert.deepEqual(await archived.json(), { threadId: "manage-me", archived: true }); + const list = async () => + ( + await ( + await app.request("/api/copilotkit/threads?agentId=default&limit=100", { + headers: headers(), + }) + ).json() + ).threads as { id: string; archived: boolean }[]; + assert.ok( + (await list()).some((thread) => thread.id === "manage-me" && thread.archived), + "the archived thread shows in the archived list", + ); + // Restore puts it back in the active list. + const restored = await app.request("/api/copilotkit/threads/manage-me", { + method: "PATCH", + headers: headers(), + body: JSON.stringify({ agentId: "default", archived: false }), + }); + assert.equal(restored.status, 200); + assert.ok( + (await list()).some((thread) => thread.id === "manage-me" && !thread.archived), + "the restored thread shows in the active list", + ); + // Delete removes the durable record and the list entry. + const deleted = await app.request("/api/copilotkit/threads/manage-me", { + method: "DELETE", + headers: headers(), + body: JSON.stringify({ agentId: "default" }), + }); + assert.equal(deleted.status, 200); + assert.deepEqual(await deleted.json(), { threadId: "manage-me", deleted: true }); + assert.equal(await db.get("local-user", CHAT_THREADS_KIND, "manage-me"), null); + assert.ok( + !(await list()).some((thread) => thread.id === "manage-me"), + "the deleted thread is gone from the list", + ); +}); + +test("thread management stays owner-scoped", async () => { + const now = new Date().toISOString(); + await db.put("someone-else", CHAT_THREADS_KIND, { + id: "manage-theirs", + owner: "someone-else", + agentId: "default", + name: "Secret", + archived: false, + createdAt: now, + updatedAt: now, + messages: [{ id: "m1", role: "user", content: "hi" }], + runs: [], + }); + for (const [method, path, body] of [ + ["PATCH", "/api/copilotkit/threads/manage-theirs", { agentId: "default", name: "Hijack" }], + ["POST", "/api/copilotkit/threads/manage-theirs/archive", { agentId: "default" }], + ["DELETE", "/api/copilotkit/threads/manage-theirs", { agentId: "default" }], + ] as const) { + const response = await app.request(path, { + method, + headers: headers(), + body: JSON.stringify(body), + }); + assert.equal(response.status, 404, `${method} ${path} must not touch another owner's thread`); + } + const kept = await db.get("someone-else", CHAT_THREADS_KIND, "manage-theirs"); + assert.ok(kept, "another owner's record is untouched"); + assert.equal(kept.name, "Secret"); + assert.equal(kept.archived, false); +}); + +test("the runtime info advertises thread mutations in local mode", async () => { + const response = await app.request("/api/copilotkit/info", { headers: headers() }); + assert.equal(response.status, 200, await response.clone().text()); + const body = await response.json(); + assert.equal(body.threadEndpoints.mutations, true, "the SDK must allow rename/archive/delete"); + assert.equal( + body.threadEndpoints.realtimeMetadata, + false, + "streaming metadata stays Intelligence-only", + ); +}); + +test("thread detail endpoints are owner-scoped and clear stays owner-local", async () => { + const main = await (await app.request("/api/main-thread", { headers: headers() })).json(); + const now = new Date().toISOString(); + await db.put("local-user", CHAT_THREADS_KIND, { + id: "mine-detail", + owner: "local-user", + agentId: "default", + name: null, + archived: false, + createdAt: now, + updatedAt: now, + messages: [{ id: "m1", role: "user", content: "hi" }], + runs: [{ runId: "r1", createdAt: now, events: [{ type: "RUN_STARTED" }] }], + }); + await db.put("someone-else", CHAT_THREADS_KIND, { + id: "theirs-detail", + owner: "someone-else", + agentId: "default", + name: null, + archived: false, + createdAt: now, + updatedAt: now, + messages: [{ id: "secret", role: "user", content: "not yours" }], + runs: [], + }); + const mine = await app.request("/api/copilotkit/threads/mine-detail/messages", { + headers: headers(), + }); + assert.equal(mine.status, 200); + assert.deepEqual((await mine.json()).messages, [{ id: "m1", role: "user", content: "hi" }]); + const events = await app.request("/api/copilotkit/threads/mine-detail/events", { + headers: headers(), + }); + assert.equal(events.status, 200); + assert.equal((await events.json()).events.length, 1); + // Another owner's thread is indistinguishable from a missing one. + const theirs = await app.request("/api/copilotkit/threads/theirs-detail/messages", { + headers: headers(), + }); + assert.equal(theirs.status, 404); + const missing = await app.request("/api/copilotkit/threads/nope/messages", { + headers: headers(), + }); + assert.equal(missing.status, 404); + // Clear deletes only the authenticated owner's records. + const clear = await app.request("/api/copilotkit/threads/clear", { + method: "POST", + headers: headers(), + }); + assert.equal(clear.status, 200); + const kept = await db.get("someone-else", CHAT_THREADS_KIND, "theirs-detail"); + assert.ok(kept, "another owner's record survives this owner's clear"); + assert.equal(kept.owner, "someone-else"); + assert.equal(await db.get("local-user", CHAT_THREADS_KIND, "mine-detail"), null); + assert.equal(await db.get("local-user", CHAT_THREADS_KIND, main.threadId), null); +}); + +test("approvals, browser tools and background tasks behave as in intelligence mode", async () => { + // Local mode changes only thread persistence; the action review flow is untouched. + const proposal = await app.request("/api/actions", { + method: "POST", + headers: headers(), + body: JSON.stringify({ + kind: "calendar.create", + data: { + title: "Local mode smoke", + start: "2026-10-08T09:00:00Z", + end: "2026-10-08T10:00:00Z", + }, + }), + }); + assert.equal(proposal.status, 201, await proposal.clone().text()); + const created = await proposal.json(); + assert.equal(created.status, "awaiting_review"); + const decision = await app.request(`/api/actions/${created.id}/decide`, { + method: "POST", + headers: headers(), + body: JSON.stringify({ hash: created.hash, decision: "approve" }), + }); + assert.equal(decision.status, 200); +}); diff --git a/tests/threads-provider.test.ts b/tests/threads-provider.test.ts new file mode 100644 index 000000000..eae1bb320 --- /dev/null +++ b/tests/threads-provider.test.ts @@ -0,0 +1,435 @@ +import assert from "node:assert/strict"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { after, before, test } from "node:test"; +import { EventType } from "@ag-ui/client"; +import { VERSION, ɵGLOBAL_STORE } from "@copilotkit/runtime/v2"; +import { firstValueFrom, toArray } from "rxjs"; +import type { Config } from "../apps/server/src/config.ts"; +import { createStore, type Store } from "../apps/server/src/db.ts"; +import { + CHAT_THREADS_KIND, + defaultTitleGenerator, + type LocalThreadRecord, + PersistentAgentRunner, + THREADS_CONTRACT_REF, +} from "../apps/server/src/threads/local-runner.ts"; +import { createThreadProvider } from "../apps/server/src/threads/provider.ts"; + +let db: Store, directory: string; +before(async () => { + directory = await mkdtemp(join(tmpdir(), "openmuse-threads-provider-")); + db = await createStore(); +}); +after(async () => { + await db.close(); + await rm(directory, { recursive: true, force: true }); +}); + +/** + * A scripted agent that mimics the runtime's per-request cloning: `clone()` + * returns a NEW object, so owner lookups keyed on the original instance miss. + * Mirrors ConversationAgent, which carries its owner and preserves it in clone(). + */ +function fakeAgent( + threadId: string, + owner: string, + options: { fail?: boolean; runId?: string } = {}, +) { + const agent = { + agentId: "default", + owner, + messages: [ + { id: "user-1", role: "user", content: "Hello" }, + { id: "assistant-1", role: "assistant", content: "Hi there" }, + ], + clone() { + return { ...this, messages: [...this.messages] }; + }, + async runAgent( + _input: unknown, + callbacks: { + onEvent: (payload: { event: unknown }) => void; + onNewMessage: (payload: { message: unknown }) => void; + }, + ) { + if (options.fail) throw new Error("model provider exploded"); + callbacks.onEvent({ event: { type: EventType.RUN_STARTED, threadId, runId: "run-1" } }); + callbacks.onEvent({ + event: { + type: EventType.TEXT_MESSAGE_CONTENT, + messageId: "assistant-1", + delta: "Hi there", + }, + }); + callbacks.onNewMessage({ message: { id: "assistant-1" } }); + callbacks.onEvent({ event: { type: EventType.RUN_FINISHED } }); + }, + }; + return agent as never; +} + +/** Owner resolution that survives cloning: read the agent's own owner first. */ +const ownerOf = (agent: unknown): string | undefined => { + if (!agent || typeof agent !== "object" || !("owner" in agent)) return undefined; + const owner = (agent as { owner?: unknown }).owner; + return typeof owner === "string" ? owner : undefined; +}; + +/** The runtime hands the runner a per-request CLONE; mimic that at call sites. */ +function cloneOf(agent: never): never { + return (agent as unknown as { clone(): unknown }).clone() as never; +} + +function runnerFor(titleGenerator?: (text: string) => Promise) { + return new PersistentAgentRunner(db, ownerOf, titleGenerator); +} + +async function runOnce( + runner: PersistentAgentRunner, + threadId: string, + agent: never, + runId = "run-1", +) { + return firstValueFrom( + runner + .run({ threadId, agent, input: { threadId, runId, messages: [] } } as never) + .pipe(toArray()), + ); +} + +test("the pinned runtime version still matches the store contract this runner relies on", () => { + assert.ok(THREADS_CONTRACT_REF.includes(VERSION)); +}); + +test("RUN_FINISHED is held until the snapshot write completes", async () => { + const order: string[] = []; + const original = db.put.bind(db); + db.put = ((owner: string, kind: string, value: { id: string }) => { + order.push("put"); + return new Promise((resolve) => setTimeout(resolve, 40)).then(() => + original(owner, kind, value), + ); + }) as Store["put"]; + try { + const events = await runOnce(runnerFor(), "thread-gated", fakeAgent("thread-gated", "owner-1")); + order.push(...events.map((event) => (event as { type: string }).type)); + const putAt = order.indexOf("put"); + const finishedAt = order.indexOf(EventType.RUN_FINISHED); + assert.notEqual(putAt, -1, "the snapshot was written"); + assert.notEqual(finishedAt, -1, "the run finished"); + assert.ok(putAt < finishedAt, "persistence lands before the client sees completion"); + assert.equal(finishedAt, order.length - 1, "RUN_FINISHED is still the terminal event"); + } finally { + db.put = original; + } +}); + +test("a failed snapshot write is a visible RUN_ERROR, not a silent loss", async () => { + const original = db.put.bind(db); + db.put = (() => Promise.reject(new Error("disk full"))) as Store["put"]; + try { + const events = await runOnce( + runnerFor(), + "thread-failing", + fakeAgent("thread-failing", "owner-1"), + ); + const types = events.map((event) => (event as { type: string }).type); + assert.ok(types.includes(EventType.RUN_ERROR), "the failure reaches the client"); + assert.ok( + types.every((type) => type !== EventType.RUN_FINISHED), + "no completion is reported for a run that was not saved", + ); + const saved = await db.get("owner-1", CHAT_THREADS_KIND, "thread-failing"); + assert.equal(saved, null); + } finally { + db.put = original; + } +}); + +test("history survives a restart: a fresh runner replays the persisted thread", async () => { + // Pass the CLONE, exactly like the runtime does after cloneAgentForRequest. + const original = fakeAgent("thread-restart", "owner-1"); + await runOnce(runnerFor(), "thread-restart", cloneOf(original)); + const record = await db.get("owner-1", CHAT_THREADS_KIND, "thread-restart"); + assert.ok(record, "the thread was persisted"); + assert.equal(record.owner, "owner-1"); + assert.deepEqual( + (record.messages as { id: string }[]).map((message) => message.id), + ["user-1", "assistant-1"], + ); + assert.equal(record.runs.length, 1); + assert.ok((record.runs[0].events as unknown[]).length > 0); + // Simulate a process restart: same database, empty in-memory store, new runner. + ɵGLOBAL_STORE.clear(); + const revived = runnerFor(); + const replayed = await firstValueFrom( + revived.connect({ threadId: "thread-restart" }).pipe(toArray()), + ); + const types = replayed.map((event) => (event as { type: string }).type); + assert.ok(types.includes(EventType.RUN_STARTED)); + assert.ok(types.includes(EventType.TEXT_MESSAGE_CONTENT)); + assert.ok(types.includes(EventType.RUN_FINISHED)); + assert.deepEqual( + revived.listThreads().map((thread) => thread.id), + ["thread-restart"], + ); +}); + +test("threads are scoped and replayed per owner", async () => { + await runOnce(runnerFor(), "thread-a", fakeAgent("thread-a", "owner-a")); + await runOnce(runnerFor(), "thread-b", fakeAgent("thread-b", "owner-b")); + assert.equal((await db.list("owner-a", CHAT_THREADS_KIND)).length, 1); + assert.equal((await db.list("owner-b", CHAT_THREADS_KIND)).length, 1); + ɵGLOBAL_STORE.clear(); + const runner = runnerFor(); + const replayed = await firstValueFrom(runner.connect({ threadId: "thread-b" }).pipe(toArray())); + assert.ok( + replayed.some((event) => (event as { type: string }).type === EventType.TEXT_MESSAGE_CONTENT), + ); +}); + +test("a failed run still terminates the stream cleanly", async () => { + const events = await runOnce( + runnerFor(), + "thread-failed-run", + fakeAgent("thread-failed-run", "owner-1", { fail: true }), + ); + const types = events.map((event) => (event as { type: string }).type); + assert.ok(types.includes(EventType.RUN_ERROR), "the run error is delivered"); + assert.ok( + types.every((type) => type !== EventType.RUN_FINISHED), + "a failed run never reports completion", + ); + // firstValueFrom resolved, so the stream completed — no client-side hang. + const record = await db.get("owner-1", CHAT_THREADS_KIND, "thread-failed-run"); + assert.equal(record, null, "nothing is stored for a run that produced no events"); +}); + +test("a concurrent run on one thread errors instead of hanging", async () => { + // Gate the first run inside runAgent so it stays in flight while the second + // run arrives — deterministic ordering for the "already running" rejection. + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const stalled = { + agentId: "default", + owner: "owner-1", + messages: [], + async runAgent( + _input: unknown, + callbacks: { + onEvent: (payload: { event: unknown }) => void; + onNewMessage: (payload: { message: unknown }) => void; + }, + ) { + callbacks.onEvent({ + event: { type: EventType.RUN_STARTED, threadId: "thread-busy", runId: "run-1" }, + }); + await gate; + callbacks.onEvent({ event: { type: EventType.RUN_FINISHED } }); + }, + } as never; + const runner = runnerFor(); + const first = firstValueFrom( + runner + .run({ + threadId: "thread-busy", + agent: stalled, + input: { threadId: "thread-busy", runId: "run-1", messages: [] }, + } as never) + .pipe(toArray()), + ); + // Give the first run a macrotask to reach its in-flight state. + await new Promise((resolve) => setTimeout(resolve, 20)); + await assert.rejects( + firstValueFrom( + runner + .run({ + threadId: "thread-busy", + agent: fakeAgent("thread-busy", "owner-1"), + input: { threadId: "thread-busy", runId: "run-2", messages: [] }, + } as never) + .pipe(toArray()), + ), + /Thread already running/, + ); + release(); + await first; +}); + +test("memory eviction does not truncate the persisted conversation", async () => { + const threadId = "thread-evict"; + for (let run = 1; run <= 3; run++) { + const original = fakeAgent(threadId, "owner-1"); + await runOnce(runnerFor(), threadId, cloneOf(original), `run-${run}`); + } + const before = await db.get("owner-1", CHAT_THREADS_KIND, threadId); + assert.ok(before); + assert.equal(before.runs.length, 3, "three runs are stored"); + // Simulate the in-memory run cap dropping old runs after a fourth run starts. + const store = ɵGLOBAL_STORE.peek(threadId); + assert.ok(store); + store.historicRuns.splice(0, 2); + const fourth = fakeAgent(threadId, "owner-1"); + await runOnce(runnerFor(), threadId, cloneOf(fourth), "run-4"); + const after = await db.get("owner-1", CHAT_THREADS_KIND, threadId); + assert.ok(after); + assert.deepEqual( + after.runs.map((run) => run.runId), + ["run-1", "run-2", "run-3", "run-4"], + "evicted runs survive in the durable record", + ); +}); + +test("v1 records with thread-level events migrate on read", async () => { + const now = new Date().toISOString(); + await db.put("owner-1", CHAT_THREADS_KIND, { + id: "thread-legacy", + owner: "owner-1", + agentId: "default", + name: null, + archived: false, + createdAt: now, + updatedAt: now, + messages: [], + events: [{ type: EventType.RUN_STARTED }], + }); + ɵGLOBAL_STORE.clear(); + const replayed = await firstValueFrom( + runnerFor().connect({ threadId: "thread-legacy" }).pipe(toArray()), + ); + assert.ok( + replayed.some((event) => (event as { type: string }).type === EventType.RUN_STARTED), + "legacy thread-level events still replay", + ); +}); + +test("clearThreads wipes the durable records as well as memory", async () => { + await runOnce(runnerFor(), "thread-clear", fakeAgent("thread-clear", "owner-1")); + const runner = runnerFor(); + await runner.clearThreads(); + assert.equal((await db.list("owner-1", CHAT_THREADS_KIND)).length, 0); +}); + +test("deleteThread drops one thread from the in-memory store", async () => { + // A store whose owner lookup fails never persists, so connect cannot + // rehydrate from the database — the assertion stays memory-only. + await runOnce( + new PersistentAgentRunner(db, () => undefined), + "thread-delete", + fakeAgent("thread-delete", "owner-1"), + ); + const runner = runnerFor(); + runner.deleteThread("thread-delete"); + assert.equal(ɵGLOBAL_STORE.peek("thread-delete"), undefined); + const replayed = await firstValueFrom( + runner.connect({ threadId: "thread-delete" }).pipe(toArray()), + ); + assert.equal(replayed.length, 0, "a deleted thread replays nothing from memory"); +}); + +test("an untitled thread gets a generated title after the run", async () => { + const calls: string[] = []; + const runner = runnerFor(async (text) => { + calls.push(text); + return "周报助手"; + }); + await runOnce(runner, "thread-titled", fakeAgent("thread-titled", "owner-1")); + await new Promise((resolve) => setTimeout(resolve, 20)); + const record = await db.get("owner-1", CHAT_THREADS_KIND, "thread-titled"); + assert.ok(record); + assert.equal(record.name, "周报助手"); + assert.equal(calls[0], "Hello", "the title input is the first user message"); +}); + +test("an existing title is never overwritten by auto-titling", async () => { + const now = new Date().toISOString(); + await db.put("owner-1", CHAT_THREADS_KIND, { + id: "thread-named", + owner: "owner-1", + agentId: "default", + name: "Hand picked", + archived: false, + createdAt: now, + updatedAt: now, + messages: [{ id: "user-1", role: "user", content: "Hello" }], + runs: [], + }); + const runner = runnerFor(async () => { + throw new Error("must not be called"); + }); + await runOnce(runner, "thread-named", fakeAgent("thread-named", "owner-1")); + await new Promise((resolve) => setTimeout(resolve, 20)); + const record = await db.get("owner-1", CHAT_THREADS_KIND, "thread-named"); + assert.ok(record); + assert.equal(record.name, "Hand picked"); +}); + +test("a title generator failure leaves the thread untitled without breaking the run", async () => { + const runner = runnerFor(async () => { + throw new Error("model down"); + }); + const events = await runOnce(runner, "thread-untitled", fakeAgent("thread-untitled", "owner-1")); + await new Promise((resolve) => setTimeout(resolve, 20)); + assert.ok(events.some((event) => (event as { type: string }).type === EventType.RUN_FINISHED)); + const record = await db.get("owner-1", CHAT_THREADS_KIND, "thread-untitled"); + assert.ok(record); + assert.equal(record.name, null); +}); + +test("defaultTitleGenerator falls back to a truncated message without a model", async () => { + const long = "帮我写一份".repeat(30); + const title = await defaultTitleGenerator(undefined)(long); + assert.ok(title, "a message without a configured model still gets a title"); + assert.ok(title.length <= 60, "the fallback title is clipped"); + assert.equal(await defaultTitleGenerator(undefined)(" "), null, "blank text titles nothing"); + assert.equal(await defaultTitleGenerator(undefined)("帮我写一份"), "帮我写一份"); + assert.equal( + await defaultTitleGenerator(undefined)("line one\nline two"), + "line one line two", + "newlines collapse to spaces", + ); +}); + +test("createThreadProvider selects the backend from config", async () => { + const owners = new Map(); + const weakOwnerOf = (agent: unknown) => owners.get(agent); + const base: Config = { + mode: "sample", + port: 8787, + host: "127.0.0.1", + publicUrl: "http://localhost:8787", + dataDir: directory, + agentBackend: "sample", + googleRedirectUri: "http://localhost:8787/api/google/callback", + allowedOrigins: ["http://localhost:8081"], + }; + const intelligence = createThreadProvider( + { ...base, threadsBackend: "intelligence", intelligenceApiKey: "test-project-key-never-sent" }, + { db, ownerOf: weakOwnerOf }, + ); + assert.equal(intelligence.backend, "intelligence"); + const local = createThreadProvider( + { ...base, threadsBackend: "local" }, + { + db, + ownerOf: weakOwnerOf, + }, + ); + assert.equal(local.backend, "local"); + if (local.backend !== "local") throw new Error("unreachable"); + assert.ok(local.runner instanceof PersistentAgentRunner); + // A missing backend means the documented default, which still requires a key. + const defaulted = createThreadProvider( + { ...base, intelligenceApiKey: "test-project-key-never-sent" }, + { db, ownerOf: weakOwnerOf }, + ); + assert.equal(defaulted.backend, "intelligence"); + assert.throws( + () => createThreadProvider({ ...base }, { db, ownerOf: weakOwnerOf }), + /requires CPK_INTELLIGENCE_API_KEY/, + ); +});