From 99cdacaada8942bce9a18728c2045a8f06dc2814 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E5=BB=BA=E5=86=9B?= Date: Wed, 7 Oct 2026 23:10:10 +0800 Subject: [PATCH 1/7] feat: pluggable thread persistence with THREADS_BACKEND=local Make CopilotKit Intelligence replaceable via the runtime's own AgentRunner seam so self-hosted deployments run keyless: - THREADS_BACKEND=intelligence|local (default intelligence, unchanged behavior; an explicit intelligence selection without a key still fails loudly at startup, never a silent downgrade) - PersistentAgentRunner extends InMemoryAgentRunner: hydrates threads from the database before run/connect, holds RUN_FINISHED until the snapshot write completes, and surfaces a failed write as a visible RUN_ERROR - owner-scoped, database-backed thread listing (the runtime's built-in local fallback is process-global); rename/archive stay Intelligence-only and are hidden in the native UI rather than faked - the demo harness follows the same setting and runs keyless Addresses the review feedback on #94 (awaited persistence, runner regressions, operator docs) and fixes #73. --- .env.example | 9 +- README.md | 19 +- apps/mobile/src/threads.tsx | 71 ++++---- apps/server/src/agent.ts | 63 ++++--- apps/server/src/app.ts | 60 +++++-- apps/server/src/config.ts | 27 ++- apps/server/src/demo/entry.ts | 13 +- apps/server/src/threads/local-runner.ts | 223 ++++++++++++++++++++++++ apps/server/src/threads/provider.ts | 67 +++++++ apps/server/src/workspace.ts | 1 + docs/PLUGGABLE-THREADS.md | 69 ++++++++ packages/domain/src/index.ts | 1 + tests/config.test.ts | 44 ++++- tests/local-threads.test.ts | 142 +++++++++++++++ tests/threads-provider.test.ts | 212 ++++++++++++++++++++++ 15 files changed, 928 insertions(+), 93 deletions(-) create mode 100644 apps/server/src/threads/local-runner.ts create mode 100644 apps/server/src/threads/provider.ts create mode 100644 docs/PLUGGABLE-THREADS.md create mode 100644 tests/local-threads.test.ts create mode 100644 tests/threads-provider.test.ts diff --git a/.env.example b/.env.example index 36e6580c9..d0839d3a6 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..8b8846ed9 100644 --- a/apps/mobile/src/threads.tsx +++ b/apps/mobile/src/threads.tsx @@ -113,6 +113,9 @@ export function ThreadsSheet({ onClose }: { onClose: () => void }) { } = useMuseThread(); const { workspace, open, navigate, refresh } = useWorkspace(); const threads = useThreads({ agentId: "default", enabled, includeArchived: true, limit: 20 }); + // Local persistence serves list/replay but not rename/archive (the runtime + // returns 422 for those without Intelligence), so hide rather than fake them. + const canManageThreads = (workspace.runtime.threadsBackend ?? "intelligence") === "intelligence"; const [editing, setEditing] = useState(); const [name, setName] = useState(""); const [error, setError] = useState(""); @@ -169,9 +172,11 @@ export function ThreadsSheet({ onClose }: { onClose: () => void }) { Side chats - + {canManageThreads && ( + + )} {threads.isLoading && } @@ -228,36 +233,38 @@ export function ThreadsSheet({ onClose }: { onClose: () => void }) { {editing === thread.id && ( )} - - + - - + > + {thread.archived ? "Restore" : "Archive"} + + + )} ))} {!threads.isLoading && 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 bff91dd2f..5153c76ee 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -1,7 +1,6 @@ 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 { bodyLimit } from "hono/body-limit"; import { cors } from "hono/cors"; @@ -17,7 +16,7 @@ import { createAuth } from "./auth.ts"; import { BrowserService } from "./browser.ts"; import { ComputerService, type DockerRunner, resolveComputerDeployment } 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 { agentRoutes } from "./engine/routes.ts"; import { AgentService } from "./engine/service.ts"; @@ -25,6 +24,8 @@ 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,6 +33,7 @@ export async function createApp( config: Config, options: { docker?: DockerRunner } = {}, ) { + assertThreadsBackendConfig(config); assertApiDeploymentConfig(config); await resolveComputerDeployment(config); const auth = await createAuth(db, config), @@ -48,8 +50,13 @@ 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. + const agentOwners = new WeakMap(); + const ownerOf = (agent: unknown) => + 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) => { @@ -102,6 +109,7 @@ export async function createApp( mode: config.mode, agentConfigured: agentConfigured(config), browserConfigured: Boolean(config.workerUrl && config.workerToken), + threadsBackend: config.threadsBackend ?? "intelligence", }), ); let loginWindow = 0, @@ -224,14 +232,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, ); } @@ -338,12 +344,44 @@ 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'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((record) => ({ + id: record.id, + name: record.name, + agentId: record.agentId, + organizationId: "", + createdById: "", + archived: record.archived, + createdAt: record.createdAt, + updatedAt: record.updatedAt, + })), + nextCursor: nextOffset < records.length ? String(nextOffset) : null, + }); + }); 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(); @@ -355,7 +393,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 25797a979..724d3cd9c 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/threads/local-runner.ts b/apps/server/src/threads/local-runner.ts new file mode 100644 index 000000000..7fd0fea01 --- /dev/null +++ b/apps/server/src/threads/local-runner.ts @@ -0,0 +1,223 @@ +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 { Observable, ReplaySubject } from "rxjs"; +import type { Store } from "../db.ts"; +import { backgroundFailure } from "../log.ts"; + +/** + * Pinned against the @copilotkit/runtime store contract (1.70.1). The only + * internal-API (ɵ) surface this module touches is ɵGLOBAL_STORE, used to + * rehydrate a thread after a server restart and to snapshot it for saving; a + * runtime upgrade that moves it breaks compilation here, in one place, + * rather than silently dropping persistence. + */ +export const THREADS_CONTRACT_REF = "@copilotkit/runtime 1.70.1"; + +export const CHAT_THREADS_KIND = "chat-threads"; + +/** Durable per-thread record stored in the server's own database. */ +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[]; + /** Compacted AG-UI events across all runs of the thread, replayed on connect. */ + events: unknown[]; +} + +type OwnerLookup = (agent: unknown) => string | undefined; + +/** + * 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 saves + * the snapshot once a run finalizes. + * + * 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 is reported as a visible RUN_ERROR + * instead of being swallowed. + */ +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, + ) { + super(); + } + + override run(request: AgentRunnerRunRequest): Observable { + const subject = new ReplaySubject(Infinity); + this.hydrate(request.threadId).then( + () => + this.gatePersistence(super.run(request), 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)); + } + + /** + * Forward every event except the terminal RUN_FINISHED, which is held until + * the durable snapshot write settles. The base runner finalizes (appendRun) + * synchronously after emitting RUN_FINISHED, so the persist defers one + * macrotask and reads the finalized store. + */ + 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 flush = () => { + if (!sourceComplete || !persisted) return; + void persisted.then( + () => { + if (!subscriber.closed) { + if (finishEvent) subscriber.next(finishEvent); + subscriber.complete(); + } + }, + (error: unknown) => { + backgroundFailure("chat thread persistence", error); + if (!subscriber.closed) { + subscriber.next({ + type: EventType.RUN_ERROR, + message: + "The run completed but the conversation could not be saved. " + + "Check the server database and try again.", + } as BaseEvent); + subscriber.complete(); + } + }, + ); + }; + 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; + // One synthetic run holding the persisted events is the canonical + // post-appendRun shape the base runner itself produces. + store.historicRuns = [ + { + threadId, + runId: "persisted", + agentId: record.agentId, + parentRunId: null, + events: record.events, + messages: [], + createdAt: Date.parse(record.createdAt), + } as never, + ]; + } + + /** Write the finalized thread snapshot back to the database. */ + 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 createdAt = store.createdAt ?? Date.now(); + const record: LocalThreadRecord = { + id: threadId, + owner, + agentId: prior?.agentId ?? "default", + name: prior?.name ?? null, + archived: prior?.archived ?? false, + createdAt: prior?.createdAt ?? new Date(createdAt).toISOString(), + updatedAt: new Date().toISOString(), + messages: store.messagesSnapshot as unknown[], + events: store.historicRuns.flatMap((run) => run.events) as unknown[], + }; + await this.db.put(owner, CHAT_THREADS_KIND, record); + this.index.set(threadId, record); + this.scanIndex.set(threadId, record); + } + + /** 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..237e4d3f1 --- /dev/null +++ b/apps/server/src/threads/provider.ts @@ -0,0 +1,67 @@ +import { type AgentRunner, 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, 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: AgentRunner; + 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); + 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: [], + events: [], + }); + }, + }; +} + +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..318ad3266 --- /dev/null +++ b/docs/PLUGGABLE-THREADS.md @@ -0,0 +1,69 @@ +# 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 intentionally serves a subset. Unsupported operations are hidden in +the native UI rather than faked, and return the runtime's own `422` if called +directly. + +| 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 | ✓ | hidden / `422` | +| Archive / restore | ✓ | hidden / `422` | +| Realtime thread metadata | ✓ | unavailable (the client SDK skips the subscription) | + +## 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 20008d2be..2d3ac9991 100644 --- a/packages/domain/src/index.ts +++ b/packages/domain/src/index.ts @@ -199,6 +199,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 a2845fb7e..29e3f966d 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..795dddf10 --- /dev/null +++ b/tests/local-threads.test.ts @@ -0,0 +1,142 @@ +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: [], + events: [], + }); + } + 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("rename and archive stay honest 422s in local mode", async () => { + const main = await (await app.request("/api/main-thread", { headers: headers() })).json(); + const rename = await app.request(`/api/copilotkit/threads/${main.threadId}`, { + method: "PATCH", + headers: headers(), + body: JSON.stringify({ agentId: "default", name: "Renamed" }), + }); + assert.equal(rename.status, 422); + const archive = await app.request(`/api/copilotkit/threads/${main.threadId}/archive`, { + method: "POST", + headers: headers(), + body: JSON.stringify({ agentId: "default" }), + }); + assert.equal(archive.status, 422); +}); + +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..dd28ab66f --- /dev/null +++ b/tests/threads-provider.test.ts @@ -0,0 +1,212 @@ +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, + 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: emits a couple of events, then resolves like a finished run. */ +function fakeAgent(id: string, owner: string, owners: Map) { + const agent = { + agentId: "default", + messages: [ + { id: "user-1", role: "user", content: "Hello" }, + { id: "assistant-1", role: "assistant", content: "Hi there" }, + ], + async runAgent( + _input: unknown, + callbacks: { + onEvent: (payload: { event: unknown }) => void; + onNewMessage: (payload: { message: unknown }) => void; + }, + ) { + callbacks.onEvent({ event: { type: EventType.RUN_STARTED, threadId: id, 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 } }); + }, + }; + owners.set(agent, owner); + return agent as never; +} + +function runnerFor(owners: Map) { + return new PersistentAgentRunner(db, (agent) => owners.get(agent)); +} + +async function runOnce(runner: PersistentAgentRunner, threadId: string, agent: never) { + return firstValueFrom( + runner + .run({ threadId, agent, input: { threadId, runId: "run-1", 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 owners = new Map(); + 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(owners), + "thread-gated", + fakeAgent("thread-gated", "owner-1", owners), + ); + 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 owners = new Map(); + const original = db.put.bind(db); + db.put = (() => Promise.reject(new Error("disk full"))) as Store["put"]; + try { + const events = await runOnce( + runnerFor(owners), + "thread-failing", + fakeAgent("thread-failing", "owner-1", owners), + ); + 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 () => { + const owners = new Map(); + await runOnce( + runnerFor(owners), + "thread-restart", + fakeAgent("thread-restart", "owner-1", owners), + ); + 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.ok((record.events as unknown[]).length > 0); + // Simulate a process restart: same database, empty in-memory store, new runner. + ɵGLOBAL_STORE.clear(); + const revived = runnerFor(new Map()); + 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 () => { + const owners = new Map(); + await runOnce(runnerFor(owners), "thread-a", fakeAgent("thread-a", "owner-a", owners)); + await runOnce(runnerFor(owners), "thread-b", fakeAgent("thread-b", "owner-b", owners)); + 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(new Map()); + 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("clearThreads wipes the durable records as well as memory", async () => { + const owners = new Map(); + await runOnce(runnerFor(owners), "thread-clear", fakeAgent("thread-clear", "owner-1", owners)); + const runner = runnerFor(owners); + await runner.clearThreads(); + assert.equal((await db.list("owner-1", CHAT_THREADS_KIND)).length, 0); +}); + +test("createThreadProvider selects the backend from config", async () => { + const owners = new Map(); + const ownerOf = (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 }, + ); + assert.equal(intelligence.backend, "intelligence"); + const local = createThreadProvider({ ...base, threadsBackend: "local" }, { db, ownerOf }); + 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 }, + ); + assert.equal(defaulted.backend, "intelligence"); + assert.throws( + () => createThreadProvider({ ...base }, { db, ownerOf }), + /requires CPK_INTELLIGENCE_API_KEY/, + ); +}); From 4b81ae3a0048de97ac5d98a92a7a5995cb225bb5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E5=BB=BA=E5=86=9B?= Date: Thu, 8 Oct 2026 20:59:09 +0800 Subject: [PATCH 2/7] fix: harden local threads against cloning, eviction, and cross-owner access Address the four blockers from review on #166: - side chats now persist through the runtime's per-request agent cloning: ConversationAgent exposes its owner (clone() preserves it) and the owner lookup reads it, instead of relying on a WeakMap keyed on the original factory instance - thread detail endpoints (messages/events/state) are served owner-scoped from the database; other owners' threads return 404 and clear deletes only the authenticated owner's durable rows - failed runs terminate the stream cleanly (best-effort persist, no hang when no RUN_FINISHED arrives) and concurrent sends surface the runner's Thread already running error instead of hanging - persistence merges runs by runId into the stored record, so in-memory eviction can no longer truncate the durable conversation; v1 records with thread-level events migrate on read Regression coverage: clone-path persistence, failed-run termination, concurrent-run rejection, eviction-merge survival, v1 migration, and owner-scoped detail endpoints with owner-local clear. --- apps/server/src/app.ts | 49 +++++- apps/server/src/engine/conversation.ts | 3 +- apps/server/src/threads/local-runner.ts | 160 +++++++++++++----- apps/server/src/threads/provider.ts | 6 +- tests/local-threads.test.ts | 59 ++++++- tests/threads-provider.test.ts | 216 ++++++++++++++++++++---- 6 files changed, 403 insertions(+), 90 deletions(-) diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index 5153c76ee..f73fe0aed 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -1,7 +1,7 @@ import "./config.ts"; import { randomUUID } from "node:crypto"; import { MessageSchema } from "@ag-ui/core"; -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"; @@ -18,6 +18,7 @@ import { ComputerService, type DockerRunner, resolveComputerDeployment } from ". import { computerRoutes } from "./computer-routes.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"; @@ -52,9 +53,16 @@ export async function createApp( const agent = new AgentService(db, config, workspace, files, actions, browser, computer); // 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) => - typeof agent === "object" && agent !== null ? agentOwners.get(agent) : undefined; + 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 } }>(); @@ -373,6 +381,43 @@ export async function createApp( 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)); + 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 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 c.json({ [key]: pick(record) }); + } app.all("/api/copilotkit/*", async (c) => { if (!agentConfigured(config)) throw new AppError( 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 index 7fd0fea01..3aa663072 100644 --- a/apps/server/src/threads/local-runner.ts +++ b/apps/server/src/threads/local-runner.ts @@ -16,7 +16,19 @@ export const THREADS_CONTRACT_REF = "@copilotkit/runtime 1.70.1"; export const CHAT_THREADS_KIND = "chat-threads"; -/** Durable per-thread record stored in the server's own database. */ +/** 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; @@ -27,22 +39,30 @@ export interface LocalThreadRecord { updatedAt: string; /** Thread-level message snapshot (AG-UI Message[]) replayed on connect. */ messages: unknown[]; - /** Compacted AG-UI events across all runs of the thread, replayed on connect. */ - events: unknown[]; + runs: LocalThreadRun[]; + /** v1 records stored all events at thread level; migrated on read. */ + events?: unknown[]; } type OwnerLookup = (agent: unknown) => string | undefined; +/** 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 saves - * the snapshot once a run finalizes. + * 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 is reported as a visible RUN_ERROR - * instead of being swallowed. + * 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. */ @@ -59,10 +79,18 @@ export class PersistentAgentRunner extends InMemoryAgentRunner { override run(request: AgentRunnerRunRequest): Observable { const subject = new ReplaySubject(Infinity); this.hydrate(request.threadId).then( - () => - this.gatePersistence(super.run(request), request.threadId, request.agent).subscribe( - subject, - ), + () => { + // 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(); @@ -102,11 +130,19 @@ export class PersistentAgentRunner extends InMemoryAgentRunner { .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(); + } + /** * Forward every event except the terminal RUN_FINISHED, which is held until - * the durable snapshot write settles. The base runner finalizes (appendRun) + * 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. + * 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, @@ -117,28 +153,34 @@ export class PersistentAgentRunner extends InMemoryAgentRunner { 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 || !persisted) return; - void persisted.then( - () => { - if (!subscriber.closed) { - if (finishEvent) subscriber.next(finishEvent); - subscriber.complete(); - } - }, - (error: unknown) => { - backgroundFailure("chat thread persistence", error); - if (!subscriber.closed) { - subscriber.next({ + 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); - subscriber.complete(); - } - }, - ); + }, + ); + } 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) => { @@ -171,45 +213,71 @@ export class PersistentAgentRunner extends InMemoryAgentRunner { const store = ɵGLOBAL_STORE.getOrCreate(threadId); store.createdAt = Date.parse(record.createdAt); store.messagesSnapshot = record.messages as never; - // One synthetic run holding the persisted events is the canonical - // post-appendRun shape the base runner itself produces. - store.historicRuns = [ - { - threadId, - runId: "persisted", - agentId: record.agentId, - parentRunId: null, - events: record.events, - messages: [], - createdAt: Date.parse(record.createdAt), - } 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 to the database. */ + /** 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 createdAt = store.createdAt ?? Date.now(); + 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(createdAt).toISOString(), + createdAt: prior?.createdAt ?? new Date(store.createdAt ?? Date.now()).toISOString(), updatedAt: new Date().toISOString(), messages: store.messagesSnapshot as unknown[], - events: store.historicRuns.flatMap((run) => run.events) as unknown[], + runs: [...priorRuns, ...freshRuns], }; await this.db.put(owner, CHAT_THREADS_KIND, record); this.index.set(threadId, record); this.scanIndex.set(threadId, record); } + /** 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); diff --git a/apps/server/src/threads/provider.ts b/apps/server/src/threads/provider.ts index 237e4d3f1..3318f0684 100644 --- a/apps/server/src/threads/provider.ts +++ b/apps/server/src/threads/provider.ts @@ -1,4 +1,4 @@ -import { type AgentRunner, CopilotKitIntelligence } from "@copilotkit/runtime/v2"; +import { CopilotKitIntelligence } from "@copilotkit/runtime/v2"; import type { Config, ThreadsBackend } from "../config.ts"; import { intelligenceKeyRequiredMessage } from "../config.ts"; import type { Store } from "../db.ts"; @@ -17,7 +17,7 @@ export type ThreadProvider = } | { backend: "local"; - runner: AgentRunner; + runner: PersistentAgentRunner; getOrCreateMainThread(input: { owner: string; threadId: string }): Promise; }; @@ -58,7 +58,7 @@ export function createThreadProvider( createdAt: now, updatedAt: now, messages: [], - events: [], + runs: [], }); }, }; diff --git a/tests/local-threads.test.ts b/tests/local-threads.test.ts index 795dddf10..2b122520d 100644 --- a/tests/local-threads.test.ts +++ b/tests/local-threads.test.ts @@ -76,7 +76,7 @@ test("thread listing is owner-scoped from the database and paginates", async () createdAt: now, updatedAt: now, messages: [], - events: [], + runs: [], }); } const seen: string[] = []; @@ -116,6 +116,63 @@ test("rename and archive stay honest 422s in local mode", async () => { assert.equal(archive.status, 422); }); +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", { diff --git a/tests/threads-provider.test.ts b/tests/threads-provider.test.ts index dd28ab66f..fa15681cc 100644 --- a/tests/threads-provider.test.ts +++ b/tests/threads-provider.test.ts @@ -26,14 +26,26 @@ after(async () => { await rm(directory, { recursive: true, force: true }); }); -/** A scripted agent: emits a couple of events, then resolves like a finished run. */ -function fakeAgent(id: string, owner: string, owners: Map) { +/** + * 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: { @@ -41,7 +53,8 @@ function fakeAgent(id: string, owner: string, owners: Map) { onNewMessage: (payload: { message: unknown }) => void; }, ) { - callbacks.onEvent({ event: { type: EventType.RUN_STARTED, threadId: id, runId: "run-1" } }); + 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, @@ -53,18 +66,34 @@ function fakeAgent(id: string, owner: string, owners: Map) { callbacks.onEvent({ event: { type: EventType.RUN_FINISHED } }); }, }; - owners.set(agent, owner); return agent as never; } -function runnerFor(owners: Map) { - return new PersistentAgentRunner(db, (agent) => owners.get(agent)); +/** 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() { + return new PersistentAgentRunner(db, ownerOf); } -async function runOnce(runner: PersistentAgentRunner, threadId: string, agent: never) { +async function runOnce( + runner: PersistentAgentRunner, + threadId: string, + agent: never, + runId = "run-1", +) { return firstValueFrom( runner - .run({ threadId, agent, input: { threadId, runId: "run-1", messages: [] } } as never) + .run({ threadId, agent, input: { threadId, runId, messages: [] } } as never) .pipe(toArray()), ); } @@ -74,7 +103,6 @@ test("the pinned runtime version still matches the store contract this runner re }); test("RUN_FINISHED is held until the snapshot write completes", async () => { - const owners = new Map(); const order: string[] = []; const original = db.put.bind(db); db.put = ((owner: string, kind: string, value: { id: string }) => { @@ -84,11 +112,7 @@ test("RUN_FINISHED is held until the snapshot write completes", async () => { ); }) as Store["put"]; try { - const events = await runOnce( - runnerFor(owners), - "thread-gated", - fakeAgent("thread-gated", "owner-1", owners), - ); + 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); @@ -102,14 +126,13 @@ test("RUN_FINISHED is held until the snapshot write completes", async () => { }); test("a failed snapshot write is a visible RUN_ERROR, not a silent loss", async () => { - const owners = new Map(); const original = db.put.bind(db); db.put = (() => Promise.reject(new Error("disk full"))) as Store["put"]; try { const events = await runOnce( - runnerFor(owners), + runnerFor(), "thread-failing", - fakeAgent("thread-failing", "owner-1", owners), + 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"); @@ -125,12 +148,9 @@ test("a failed snapshot write is a visible RUN_ERROR, not a silent loss", async }); test("history survives a restart: a fresh runner replays the persisted thread", async () => { - const owners = new Map(); - await runOnce( - runnerFor(owners), - "thread-restart", - fakeAgent("thread-restart", "owner-1", owners), - ); + // 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"); @@ -138,10 +158,11 @@ test("history survives a restart: a fresh runner replays the persisted thread", (record.messages as { id: string }[]).map((message) => message.id), ["user-1", "assistant-1"], ); - assert.ok((record.events as unknown[]).length > 0); + 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(new Map()); + const revived = runnerFor(); const replayed = await firstValueFrom( revived.connect({ threadId: "thread-restart" }).pipe(toArray()), ); @@ -156,30 +177,145 @@ test("history survives a restart: a fresh runner replays the persisted thread", }); test("threads are scoped and replayed per owner", async () => { - const owners = new Map(); - await runOnce(runnerFor(owners), "thread-a", fakeAgent("thread-a", "owner-a", owners)); - await runOnce(runnerFor(owners), "thread-b", fakeAgent("thread-b", "owner-b", owners)); + 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(new Map()); + 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 () => { - const owners = new Map(); - await runOnce(runnerFor(owners), "thread-clear", fakeAgent("thread-clear", "owner-1", owners)); - const runner = runnerFor(owners); + 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("createThreadProvider selects the backend from config", async () => { const owners = new Map(); - const ownerOf = (agent: unknown) => owners.get(agent); + const weakOwnerOf = (agent: unknown) => owners.get(agent); const base: Config = { mode: "sample", port: 8787, @@ -192,21 +328,27 @@ test("createThreadProvider selects the backend from config", async () => { }; const intelligence = createThreadProvider( { ...base, threadsBackend: "intelligence", intelligenceApiKey: "test-project-key-never-sent" }, - { db, ownerOf }, + { db, ownerOf: weakOwnerOf }, ); assert.equal(intelligence.backend, "intelligence"); - const local = createThreadProvider({ ...base, threadsBackend: "local" }, { db, ownerOf }); + 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 }, + { db, ownerOf: weakOwnerOf }, ); assert.equal(defaulted.backend, "intelligence"); assert.throws( - () => createThreadProvider({ ...base }, { db, ownerOf }), + () => createThreadProvider({ ...base }, { db, ownerOf: weakOwnerOf }), /requires CPK_INTELLIGENCE_API_KEY/, ); }); From 54272277855ff5bcd0150d5a89a19f019e2acb44 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E5=BB=BA=E5=86=9B?= Date: Thu, 8 Oct 2026 23:04:48 +0800 Subject: [PATCH 3/7] feat(server): align local thread management with the Intelligence backend Local mode now serves rename, archive/restore and delete owner-scoped from the database, and names untitled threads with the configured model, falling back to a truncated first user message. The app UI exposes the same controls on both backends; intelligence behavior is unchanged. --- apps/mobile/src/threads.tsx | 93 +++++++++------- apps/server/src/app.ts | 70 ++++++++++-- apps/server/src/threads/local-runner.ts | 139 ++++++++++++++++++++++-- apps/server/src/threads/provider.ts | 8 +- docs/PLUGGABLE-THREADS.md | 16 ++- tests/local-threads.test.ts | 95 +++++++++++++++- tests/threads-provider.test.ts | 85 ++++++++++++++- 7 files changed, 433 insertions(+), 73 deletions(-) diff --git a/apps/mobile/src/threads.tsx b/apps/mobile/src/threads.tsx index 8b8846ed9..09a055f13 100644 --- a/apps/mobile/src/threads.tsx +++ b/apps/mobile/src/threads.tsx @@ -8,6 +8,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 +33,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 +90,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} @@ -110,12 +116,10 @@ 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 }); - // Local persistence serves list/replay but not rename/archive (the runtime - // returns 422 for those without Intelligence), so hide rather than fake them. - const canManageThreads = (workspace.runtime.threadsBackend ?? "intelligence") === "intelligence"; const [editing, setEditing] = useState(); const [name, setName] = useState(""); const [error, setError] = useState(""); @@ -172,11 +176,9 @@ export function ThreadsSheet({ onClose }: { onClose: () => void }) { Side chats - {canManageThreads && ( - - )} + {threads.isLoading && } @@ -226,45 +228,56 @@ export function ThreadsSheet({ onClose }: { onClose: () => void }) { style={[s.row, { gap: 10 }]} > - + {thread.name || "Untitled conversation"} {editing === thread.id && ( )} - {canManageThreads && ( - - - - - )} + }} + > + {editing === thread.id ? "Save name" : "Rename"} + + + + ))} {!threads.isLoading && diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index f73fe0aed..ef8daa6a3 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -368,16 +368,7 @@ export async function createApp( const page = records.slice(offset, offset + limit); const nextOffset = offset + page.length; return c.json({ - threads: page.map((record) => ({ - id: record.id, - name: record.name, - agentId: record.agentId, - organizationId: "", - createdById: "", - archived: record.archived, - createdAt: record.createdAt, - updatedAt: record.updatedAt, - })), + threads: page.map(threadListItem), nextCursor: nextOffset < records.length ? String(nextOffset) : null, }); }); @@ -393,6 +384,45 @@ export async function createApp( 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"); @@ -410,13 +440,31 @@ export async function createApp( 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 c.json({ [key]: pick(record) }); + 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)) diff --git a/apps/server/src/threads/local-runner.ts b/apps/server/src/threads/local-runner.ts index 3aa663072..7dba70e90 100644 --- a/apps/server/src/threads/local-runner.ts +++ b/apps/server/src/threads/local-runner.ts @@ -1,21 +1,79 @@ 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 { openaiSummarize } from "@tanstack/ai-openai"; import { Observable, ReplaySubject } from "rxjs"; import type { Store } from "../db.ts"; import { backgroundFailure } from "../log.ts"; -/** - * Pinned against the @copilotkit/runtime store contract (1.70.1). The only - * internal-API (ɵ) surface this module touches is ɵGLOBAL_STORE, used to - * rehydrate a thread after a server restart and to snapshot it for saving; a - * runtime upgrade that moves it breaks compilation here, in one place, - * rather than silently dropping persistence. - */ 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. + */ +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 { + const id = modelId.trim(); + const adapter = ( + provider.toLowerCase() === "openai" + ? openaiSummarize(id as never, { baseURL: process.env.OPENAI_BASE_URL } as never) + : 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" }); + 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; @@ -46,6 +104,26 @@ export interface LocalThreadRecord { 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; @@ -72,6 +150,7 @@ export class PersistentAgentRunner extends InMemoryAgentRunner { constructor( private readonly db: Store, private readonly ownerOf: OwnerLookup, + private readonly titleGenerator: TitleGenerator = () => Promise.resolve(null), ) { super(); } @@ -137,6 +216,27 @@ export class PersistentAgentRunner extends InMemoryAgentRunner { 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) @@ -268,6 +368,31 @@ export class PersistentAgentRunner extends InMemoryAgentRunner { 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. */ diff --git a/apps/server/src/threads/provider.ts b/apps/server/src/threads/provider.ts index 3318f0684..e804b0606 100644 --- a/apps/server/src/threads/provider.ts +++ b/apps/server/src/threads/provider.ts @@ -2,7 +2,7 @@ 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, PersistentAgentRunner } from "./local-runner.ts"; +import { CHAT_THREADS_KIND, defaultTitleGenerator, PersistentAgentRunner } from "./local-runner.ts"; /** * Persistence boundary for chat threads. "intelligence" is the CopilotKit-hosted @@ -42,7 +42,11 @@ export function createThreadProvider( } if (config.intelligenceApiKey) console.warn("[OpenMuse] THREADS_BACKEND=local ignores CPK_INTELLIGENCE_API_KEY."); - const runner = new PersistentAgentRunner(deps.db, deps.ownerOf); + const runner = new PersistentAgentRunner( + deps.db, + deps.ownerOf, + defaultTitleGenerator(config.model), + ); const { db } = deps; return { backend: "local", diff --git a/docs/PLUGGABLE-THREADS.md b/docs/PLUGGABLE-THREADS.md index 318ad3266..1cdd2527a 100644 --- a/docs/PLUGGABLE-THREADS.md +++ b/docs/PLUGGABLE-THREADS.md @@ -36,9 +36,8 @@ upgrade that moves it breaks compilation in that one module. ## Capability matrix -Local mode intentionally serves a subset. Unsupported operations are hidden in -the native UI rather than faked, and return the runtime's own `422` if called -directly. +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 | |---|---|---| @@ -47,10 +46,17 @@ directly. | Replay on open (`connect`) | ✓ | ✓ | | Inspect messages/events/state | ✓ | ✓ | | Clear threads | ✓ | ✓ | -| Rename | ✓ | hidden / `422` | -| Archive / restore | ✓ | hidden / `422` | +| 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 diff --git a/tests/local-threads.test.ts b/tests/local-threads.test.ts index 2b122520d..392ede26f 100644 --- a/tests/local-threads.test.ts +++ b/tests/local-threads.test.ts @@ -100,20 +100,103 @@ test("thread listing is owner-scoped from the database and paginates", async () ); }); -test("rename and archive stay honest 422s in local mode", async () => { +test("threads can be renamed, archived, restored and deleted in local mode", async () => { const main = await (await app.request("/api/main-thread", { headers: headers() })).json(); - const rename = await app.request(`/api/copilotkit/threads/${main.threadId}`, { + 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: "Renamed" }), + body: JSON.stringify({ agentId: "default", name: "Weekly report" }), }); - assert.equal(rename.status, 422); - const archive = await app.request(`/api/copilotkit/threads/${main.threadId}/archive`, { + 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(archive.status, 422); + 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("thread detail endpoints are owner-scoped and clear stays owner-local", async () => { diff --git a/tests/threads-provider.test.ts b/tests/threads-provider.test.ts index fa15681cc..eae1bb320 100644 --- a/tests/threads-provider.test.ts +++ b/tests/threads-provider.test.ts @@ -10,6 +10,7 @@ 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, @@ -81,8 +82,8 @@ function cloneOf(agent: never): never { return (agent as unknown as { clone(): unknown }).clone() as never; } -function runnerFor() { - return new PersistentAgentRunner(db, ownerOf); +function runnerFor(titleGenerator?: (text: string) => Promise) { + return new PersistentAgentRunner(db, ownerOf, titleGenerator); } async function runOnce( @@ -313,6 +314,86 @@ test("clearThreads wipes the durable records as well as memory", async () => { 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); From 432b8e7939fbf468a3b8345187a64352531971ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E5=BB=BA=E5=86=9B?= Date: Fri, 9 Oct 2026 00:18:13 +0800 Subject: [PATCH 4/7] fix(server): advertise local thread mutations and pin summarize model options - intercept GET /api/copilotkit/info in local mode and set threadEndpoints.mutations=true so the SDK issues rename/archive/delete calls against the app-layer endpoints instead of refusing client-side; realtimeMetadata stays false (Intelligence-only feature) - pass modelOptions {temperature: 1, max_output_tokens: 64} to summarize: Moonshot/Kimi rejects any temperature other than 1, and the default maxLength-derived token cap truncates CJK titles; caller always wins - thread list rows match the Intelligence backend layout: title line plus Rename / Archive|Restore / Delete buttons (Delete in danger style); numberOfLines=1 keeps long titles on one line --- apps/mobile/src/threads.tsx | 177 +++++++++++++++--------- apps/mobile/src/ui.tsx | 19 ++- apps/server/src/app.ts | 13 ++ apps/server/src/threads/local-runner.ts | 14 +- tests/local-threads.test.ts | 13 +- 5 files changed, 161 insertions(+), 75 deletions(-) diff --git a/apps/mobile/src/threads.tsx b/apps/mobile/src/threads.tsx index 09a055f13..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, @@ -105,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, @@ -208,77 +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/app.ts b/apps/server/src/app.ts index ef8daa6a3..3d56d33a6 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -352,6 +352,19 @@ 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 diff --git a/apps/server/src/threads/local-runner.ts b/apps/server/src/threads/local-runner.ts index 7dba70e90..3cce5692f 100644 --- a/apps/server/src/threads/local-runner.ts +++ b/apps/server/src/threads/local-runner.ts @@ -64,7 +64,19 @@ export function defaultTitleGenerator(model: string | undefined): TitleGenerator : null ) as AnySummarizeAdapter | null; if (!adapter) return clip; - const result = await summarize({ adapter, text: clip, maxLength: 10, style: "concise" }); + // Moonshot/Kimi rejects any temperature other than 1; OpenAI (0-2), + // Anthropic (0-1) and Gemini all accept 1, so pin it explicitly. The + // summarize default would otherwise inject a low temperature. + // maxLength 10 becomes max_output_tokens 10 for openai adapters, which + // truncates CJK titles mid-phrase (each hanzi costs several tokens), so + // an explicit 64-token cap overrides it (caller always wins). + const result = await summarize({ + adapter, + text: clip, + maxLength: 10, + style: "concise", + modelOptions: { temperature: 1, max_output_tokens: 64 }, + }); return cleanTitle(result.summary) ?? clip; } catch { // A title must never surface as a chat error; the truncated message is diff --git a/tests/local-threads.test.ts b/tests/local-threads.test.ts index 392ede26f..ce2ed522f 100644 --- a/tests/local-threads.test.ts +++ b/tests/local-threads.test.ts @@ -101,7 +101,6 @@ test("thread listing is owner-scoped from the database and paginates", async () }); test("threads can be renamed, archived, restored and deleted in local mode", 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: "manage-me", @@ -199,6 +198,18 @@ test("thread management stays owner-scoped", async () => { 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(); From 60bcbe1ec1d163dc6df4d5c1d071a3314922acdd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E5=BB=BA=E5=86=9B?= Date: Fri, 9 Oct 2026 00:43:49 +0800 Subject: [PATCH 5/7] fix(server): generate openai-compatible thread titles via direct chat call The @tanstack summarize activity maps maxLength across several adapter layers and Moonshot/Kimi gateways reject or truncate under those mappings (kimi-k3 allows only temperature=1 and spends reasoning tokens before the summary). Call the chat-completions endpoint directly for openai-compatible providers so every knob (temperature, max_tokens) is explicit; anthropic and google keep the summarize activity. Any failure still degrades to the truncated first-message title. --- apps/server/src/threads/local-runner.ts | 70 ++++++++++++++++++------- 1 file changed, 50 insertions(+), 20 deletions(-) diff --git a/apps/server/src/threads/local-runner.ts b/apps/server/src/threads/local-runner.ts index 3cce5692f..c6bba5351 100644 --- a/apps/server/src/threads/local-runner.ts +++ b/apps/server/src/threads/local-runner.ts @@ -4,7 +4,6 @@ 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 { openaiSummarize } from "@tanstack/ai-openai"; import { Observable, ReplaySubject } from "rxjs"; import type { Store } from "../db.ts"; import { backgroundFailure } from "../log.ts"; @@ -40,6 +39,14 @@ function cleanTitle(summary: string): string | null { * 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) => { @@ -48,34 +55,57 @@ export function defaultTitleGenerator(model: string | undefined): TitleGenerator 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() === "openai" - ? openaiSummarize(id as never, { baseURL: process.env.OPENAI_BASE_URL } as never) - : 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 + 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; - // Moonshot/Kimi rejects any temperature other than 1; OpenAI (0-2), - // Anthropic (0-1) and Gemini all accept 1, so pin it explicitly. The - // summarize default would otherwise inject a low temperature. - // maxLength 10 becomes max_output_tokens 10 for openai adapters, which - // truncates CJK titles mid-phrase (each hanzi costs several tokens), so - // an explicit 64-token cap overrides it (caller always wins). const result = await summarize({ adapter, text: clip, maxLength: 10, style: "concise", - modelOptions: { temperature: 1, max_output_tokens: 64 }, + modelOptions: { temperature: 1, max_tokens: 256 }, }); return cleanTitle(result.summary) ?? clip; } catch { From 2fcdea93dc21172946610f5afdd6edeff4ccc3aa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E5=BB=BA=E5=86=9B?= Date: Thu, 29 Oct 2026 21:18:49 +0800 Subject: [PATCH 6/7] fix(server): enforce thread ownership at the persistence boundary Addresses the review note that the runner's isolation unit was the bare threadId: a warm process-global store entry skipped any owner check, and the findRecord fallback scan resolved an id across every owner's rows, so the invariant only held while every caller remembered to authorize first. - findRecord takes an expectedOwner and resolves a foreign record to null (cache hit and database scan alike). - run() passes the agent's owner into hydrate; hydrate and persist refuse with "Thread not found" when the durable record belongs to someone else, instead of appending to the other owner's in-memory history. - connect carries no agent, so the route layer remains the authority for replay, as before. Also rebases onto upstream/main (signed-URL method scoping, DELETE /api/session, computer deployment resolution) and restores the resolveComputerDeployment startup call alongside the threads-backend assertion. Regression tests cover a warm in-memory foreign threadId and a cold-start cross-owner run. Full suite: 549 tests, 0 failures. --- apps/server/src/app.ts | 1 - apps/server/src/threads/local-runner.ts | 54 +++++++++++++++++++------ tests/threads-provider.test.ts | 29 +++++++++++++ 3 files changed, 70 insertions(+), 14 deletions(-) diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index 3d56d33a6..4cd4f9611 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -35,7 +35,6 @@ export async function createApp( options: { docker?: DockerRunner } = {}, ) { assertThreadsBackendConfig(config); - assertApiDeploymentConfig(config); await resolveComputerDeployment(config); const auth = await createAuth(db, config), files = new Files(db, config, auth), diff --git a/apps/server/src/threads/local-runner.ts b/apps/server/src/threads/local-runner.ts index c6bba5351..4d2f2ad46 100644 --- a/apps/server/src/threads/local-runner.ts +++ b/apps/server/src/threads/local-runner.ts @@ -199,7 +199,7 @@ export class PersistentAgentRunner extends InMemoryAgentRunner { override run(request: AgentRunnerRunRequest): Observable { const subject = new ReplaySubject(Infinity); - this.hydrate(request.threadId).then( + this.hydrate(request.threadId, this.ownerOf(request.agent)).then( () => { // A concurrent run on the thread throws synchronously; deliver it as // an observable error instead of an unhandled rejection or a hang. @@ -348,10 +348,21 @@ export class PersistentAgentRunner extends InMemoryAgentRunner { } /** Load a thread's record into the process-global store unless it is already there. */ - private async hydrate(threadId: string): Promise { + private async hydrate(threadId: string, owner?: string): Promise { + // The process-global store is keyed by bare threadId, so a cache hit alone + // cannot prove ownership: an internal caller that reaches the runner with + // just a threadId must not append onto another owner's thread. Verify the + // durable record whenever the owner is knowable, even on a memory hit. + const record = await this.findRecord(threadId, owner); + if (owner && !record) { + // A durable record exists but belongs to someone else: refuse rather + // than run against the other owner's in-memory history. A thread with + // no record at all is simply new. + const exists = this.index.has(threadId) || this.scanIndex.has(threadId); + if (exists) throw new Error("Thread not found"); + } if (ɵGLOBAL_STORE.peek(threadId)) return; - const record = await this.findRecord(threadId); - if (!record || ɵGLOBAL_STORE.peek(threadId)) return; // a concurrent run hydrated first + if (!record) return; const store = ɵGLOBAL_STORE.getOrCreate(threadId); store.createdAt = Date.parse(record.createdAt); store.messagesSnapshot = record.messages as never; @@ -375,9 +386,14 @@ export class PersistentAgentRunner extends InMemoryAgentRunner { 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 owner = this.ownerOf(agent); + // An owner-less agent (tests, internal tools) keeps the record's owner; + // a known owner never appends to another owner's durable record. + const prior = await this.findRecord(threadId, owner); + if (owner && !prior && (this.index.has(threadId) || this.scanIndex.has(threadId))) + throw new Error("Thread not found"); + const effectiveOwner = owner ?? prior?.owner; + if (!effectiveOwner) 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[]) @@ -398,7 +414,7 @@ export class PersistentAgentRunner extends InMemoryAgentRunner { if (freshRuns.length === 0) return; const record: LocalThreadRecord = { id: threadId, - owner, + owner: effectiveOwner, agentId: prior?.agentId ?? "default", name: prior?.name ?? null, archived: prior?.archived ?? false, @@ -407,7 +423,7 @@ export class PersistentAgentRunner extends InMemoryAgentRunner { messages: store.messagesSnapshot as unknown[], runs: [...priorRuns, ...freshRuns], }; - await this.db.put(owner, CHAT_THREADS_KIND, record); + await this.db.put(effectiveOwner, CHAT_THREADS_KIND, record); this.index.set(threadId, record); this.scanIndex.set(threadId, record); if (record.name === null) this.titleThread(record); @@ -445,13 +461,25 @@ export class PersistentAgentRunner extends InMemoryAgentRunner { return []; } - /** Resolve a record by thread id; owners are not knowable from a thread id alone. */ - private async findRecord(threadId: string): Promise { + /** + * Resolve a record by thread id. The scan covers every owner's rows because + * a thread id alone does not name an owner; when `expectedOwner` is given a + * record owned by someone else resolves to null, so a caller holding only a + * threadId can never hydrate or append onto another owner's thread — the + * invariant lives here, not in each route. + */ + private async findRecord( + threadId: string, + expectedOwner?: string, + ): Promise { const cached = this.index.get(threadId) ?? this.scanIndex.get(threadId); - if (cached) return cached; + if (cached) return !expectedOwner || cached.owner === expectedOwner ? cached : null; 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; + if (value.id === threadId) { + const record = this.scanIndex.get(threadId) ?? null; + return !expectedOwner || record?.owner === expectedOwner ? record : null; + } } return null; } diff --git a/tests/threads-provider.test.ts b/tests/threads-provider.test.ts index eae1bb320..702d717b7 100644 --- a/tests/threads-provider.test.ts +++ b/tests/threads-provider.test.ts @@ -190,6 +190,35 @@ test("threads are scoped and replayed per owner", async () => { ); }); +test("the persistence layer refuses a foreign threadId even when it is warm in memory", async () => { + // owner-a creates a thread; its record is now cached in the runner's index + // and its history lives in the process-global store. A later internal caller + // (background job, webhook replay) that reaches the runner with just a + // threadId and owner-b's agent must not append onto owner-a's thread. + await runOnce(runnerFor(), "shared-thread", fakeAgent("shared-thread", "owner-a")); + const foreign = runnerFor(); + await assert.rejects( + runOnce(foreign, "shared-thread", fakeAgent("shared-thread", "owner-b")), + /Thread not found/, + "a cross-owner run is refused at the persistence boundary", + ); + const record = await db.get("owner-a", CHAT_THREADS_KIND, "shared-thread"); + assert.ok(record, "the original owner's record is intact"); + assert.equal(record.owner, "owner-a"); + assert.equal(await db.get("owner-b", CHAT_THREADS_KIND, "shared-thread"), null); +}); + +test("a cross-owner run after a cold start is refused from the database scan", async () => { + await runOnce(runnerFor(), "cold-thread", fakeAgent("cold-thread", "owner-a")); + ɵGLOBAL_STORE.clear(); + const foreign = runnerFor(); + await assert.rejects( + runOnce(foreign, "cold-thread", fakeAgent("cold-thread", "owner-b")), + /Thread not found/, + ); + assert.equal(await db.get("owner-b", CHAT_THREADS_KIND, "cold-thread"), null); +}); + test("a failed run still terminates the stream cleanly", async () => { const events = await runOnce( runnerFor(), From 832ddf557d5aac20317eb6fb349e4712e26e028a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E5=BB=BA=E5=86=9B?= Date: Thu, 29 Oct 2026 21:46:51 +0800 Subject: [PATCH 7/7] fix(server): bound title generation with a 30s timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The direct chat-completions fetch had no abort signal, and the summarize activity path for anthropic/google had no deadline either — a hung model gateway would leak the fire-and-forget titling promise (and its connection) forever. Both paths now time out after 30s and degrade to the truncated first-message title; the summarize timer is cleared when the call settles first so it never holds the process open. --- apps/server/src/threads/local-runner.ts | 30 +++++++++++++++++-------- 1 file changed, 21 insertions(+), 9 deletions(-) diff --git a/apps/server/src/threads/local-runner.ts b/apps/server/src/threads/local-runner.ts index 4d2f2ad46..80155d0b7 100644 --- a/apps/server/src/threads/local-runner.ts +++ b/apps/server/src/threads/local-runner.ts @@ -64,9 +64,11 @@ export function defaultTitleGenerator(model: string | undefined): TitleGenerator 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. + // reasoning tokens before the short summary text. A hung gateway must + // not leak the fire-and-forget promise forever: time out and degrade. const response = await fetch(`${baseURL}/chat/completions`, { method: "POST", + signal: AbortSignal.timeout(30_000), headers: { "Content-Type": "application/json", Authorization: `Bearer ${apiKey}` }, body: JSON.stringify({ model: modelId.trim(), @@ -100,14 +102,24 @@ export function defaultTitleGenerator(model: string | undefined): TitleGenerator : 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; + let timer: ReturnType | undefined; + try { + const result = await Promise.race([ + summarize({ + adapter, + text: clip, + maxLength: 10, + style: "concise", + modelOptions: { temperature: 1, max_tokens: 256 }, + }), + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error("Title generation timed out")), 30_000); + }), + ]); + return cleanTitle(result.summary) ?? clip; + } finally { + clearTimeout(timer); + } } catch { // A title must never surface as a chat error; the truncated message is // a perfectly good stand-in.