From 7a88f98bd7673f87ac84f2e199ee780476c85fc1 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 11 Aug 2026 11:57:47 +0000 Subject: [PATCH] Stream Cursor Cloud Agent chat updates with richer empty-result errors Prefer the Cloud Agents SSE run stream so AI Chat can show live status and assistant text, recover replies when terminal result is empty, and surface agent/run links when a run truly finishes without text. Co-authored-by: Damon --- CHANGELOG.md | 2 + apps/api/src/ai.cursor.test.ts | 199 +++++++++++++++++++- apps/api/src/ai.ts | 39 +++- apps/api/src/cursorCloud.ts | 332 +++++++++++++++++++++++++++++++-- apps/api/src/index.ts | 79 ++++++++ apps/ui/src/lib/api.ts | 127 +++++++++++++ apps/ui/src/pages/ChatPage.tsx | 130 ++++++++++--- apps/ui/src/styles.css | 7 + docs/ai/AI_INTEGRATION.md | 2 +- 9 files changed, 865 insertions(+), 52 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3fd3739..c2157fd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ Versions follow [SemVer](https://semver.org/) (`0.1.0-alpha.x` while the public - MCP / AGENTS docs no longer imply a silent `PRM_PASSWORD=workbench` default - Settings → AI: OpenAI-compatible gateways are a separate provider from Cursor Cloud Agents - Empty workspaces land in setup until finished or skipped; demo seed marks onboarding complete +- **Cursor Cloud Agents** chat: stream status and assistant text live via SSE; empty-result errors include agent/run ids and a dashboard link ### Fixed @@ -29,6 +30,7 @@ Versions follow [SemVer](https://semver.org/) (`0.1.0-alpha.x` while the public - Desktop AI: clearer network errors; Test saves then probes with visible status - Empty-workspace onboarding: optional EM name at create, single setup checklist, less Home clutter - Settings / long-page scroll jank: section nav no longer forces layout on scroll; drop sticky chrome backdrop blur +- Cursor Cloud Agents: recover reply text from streamed assistant deltas when the terminal `result` field is empty ## [0.1.0-alpha.1] - TBD diff --git a/apps/api/src/ai.cursor.test.ts b/apps/api/src/ai.cursor.test.ts index 5f1f5ef..8fa49cb 100644 --- a/apps/api/src/ai.cursor.test.ts +++ b/apps/api/src/ai.cursor.test.ts @@ -50,6 +50,25 @@ describe("normalizeAiProvider", () => { }); }); +function sseResponse(chunks: string[]): Response { + const encoder = new TextEncoder(); + let i = 0; + const stream = new ReadableStream({ + pull(controller) { + if (i >= chunks.length) { + controller.close(); + return; + } + controller.enqueue(encoder.encode(chunks[i])); + i += 1; + }, + }); + return new Response(stream, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); +} + describe("Cursor Cloud Agents client", () => { it("formats a grounded prompt", () => { const text = formatCloudAgentPrompt( @@ -65,7 +84,182 @@ describe("Cursor Cloud Agents client", () => { assert.match(text, /Summarize \[ach_1\]/); }); - it("creates a no-repo agent, polls to FINISHED, archives", async () => { + it("streams assistant text via SSE and archives", async () => { + const progress: Array<{ type: string; detail?: string }> = []; + const calls: Array<{ url: string; method: string }> = []; + const fetchFn: typeof fetch = async (input, init) => { + const url = String(input); + const method = (init?.method ?? "GET").toUpperCase(); + calls.push({ url, method }); + + if (method === "POST" && url.endsWith("/agents")) { + return new Response( + JSON.stringify({ + agent: { + id: "bc-1", + url: "https://cursor.com/agents/bc-1", + latestRunId: "run-1", + }, + run: { id: "run-1", agentId: "bc-1", status: "CREATING" }, + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + } + if (method === "GET" && url.endsWith("/stream")) { + return sseResponse([ + 'event: status\ndata: {"runId":"run-1","status":"RUNNING"}\n\n', + 'id: 1\nevent: assistant\ndata: {"text":"Hello "}\n\n', + 'id: 2\nevent: assistant\ndata: {"text":"from stream"}\n\n', + 'id: 3\nevent: result\ndata: {"runId":"run-1","status":"FINISHED","text":"Hello from stream","durationMs":1200}\n\n', + "id: 4\nevent: done\ndata: {}\n\n", + ]); + } + if (method === "POST" && url.endsWith("/archive")) { + return new Response(JSON.stringify({ id: "bc-1", status: "ARCHIVED" }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + } + return new Response("unexpected", { status: 500 }); + }; + + const result = await runCursorCloudAgent( + { + apiKey: "crsr_test", + messages: [{ role: "user", content: "ping" }], + feature: "chat_dossier", + model: "composer-2", + pollMs: 1, + timeoutMs: 5_000, + onProgress: (p) => { + if (p.type === "status") progress.push({ type: "status", detail: p.status }); + if (p.type === "text") progress.push({ type: "text", detail: p.cumulative }); + }, + }, + { fetchFn, sleep: async () => undefined, now: () => 1 }, + ); + + assert.equal(result.text, "Hello from stream"); + assert.equal(result.model, "cursor-cloud:composer-2"); + assert.equal(result.agentId, "bc-1"); + assert.ok(progress.some((p) => p.type === "text" && p.detail === "Hello from stream")); + assert.ok(calls.some((c) => c.method === "POST" && c.url.endsWith("/archive"))); + }); + + it("uses streamed assistant deltas when result text is empty", async () => { + const fetchFn: typeof fetch = async (input, init) => { + const url = String(input); + const method = (init?.method ?? "GET").toUpperCase(); + + if (method === "POST" && url.endsWith("/agents")) { + return new Response( + JSON.stringify({ + agent: { id: "bc-2", url: "https://cursor.com/agents/bc-2" }, + run: { id: "run-2", agentId: "bc-2", status: "RUNNING" }, + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + } + if (method === "GET" && url.endsWith("/stream")) { + return sseResponse([ + 'event: assistant\ndata: {"text":"Recovered answer"}\n\n', + 'event: result\ndata: {"runId":"run-2","status":"FINISHED","durationMs":800}\n\n', + "event: done\ndata: {}\n\n", + ]); + } + if (method === "GET" && url.includes("/runs/run-2") && !url.endsWith("/stream")) { + return new Response( + JSON.stringify({ + id: "run-2", + agentId: "bc-2", + status: "FINISHED", + result: "", + durationMs: 800, + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + } + if (method === "POST" && url.endsWith("/archive")) { + return new Response("{}", { status: 200 }); + } + return new Response("unexpected", { status: 500 }); + }; + + const result = await runCursorCloudAgent( + { + apiKey: "crsr_test", + messages: [{ role: "user", content: "ping" }], + feature: "chat_dossier", + pollMs: 1, + timeoutMs: 5_000, + }, + { fetchFn, sleep: async () => undefined, now: () => 1 }, + ); + assert.equal(result.text, "Recovered answer"); + }); + + it("reports agent url and ids on truly empty FINISHED runs", async () => { + const fetchFn: typeof fetch = async (input, init) => { + const url = String(input); + const method = (init?.method ?? "GET").toUpperCase(); + + if (method === "POST" && url.endsWith("/agents")) { + return new Response( + JSON.stringify({ + agent: { id: "bc-3", url: "https://cursor.com/agents/bc-3" }, + run: { id: "run-3", agentId: "bc-3", status: "RUNNING" }, + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + } + if (method === "GET" && url.endsWith("/stream")) { + return sseResponse([ + 'event: result\ndata: {"runId":"run-3","status":"FINISHED","durationMs":500}\n\n', + "event: done\ndata: {}\n\n", + ]); + } + if (method === "GET" && url.includes("/runs/run-3") && !url.endsWith("/stream")) { + return new Response( + JSON.stringify({ + id: "run-3", + agentId: "bc-3", + status: "FINISHED", + result: null, + durationMs: 500, + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + } + if (method === "POST" && url.endsWith("/archive")) { + return new Response("{}", { status: 200 }); + } + return new Response("unexpected", { status: 500 }); + }; + + await assert.rejects( + () => + runCursorCloudAgent( + { + apiKey: "crsr_test", + messages: [{ role: "user", content: "ping" }], + feature: "chat_dossier", + pollMs: 1, + timeoutMs: 5_000, + }, + { fetchFn, sleep: async () => undefined, now: () => 1 }, + ), + (err: unknown) => { + assert.ok(err instanceof Error); + assert.match(err.message, /empty result/); + assert.match(err.message, /agent bc-3/); + assert.match(err.message, /run run-3/); + assert.match(err.message, /https:\/\/cursor\.com\/agents\/bc-3/); + return true; + }, + ); + }); + + it("falls back to polling when stream endpoint is unavailable", async () => { const calls: Array<{ url: string; method: string; body?: unknown }> = []; let poll = 0; const fetchFn: typeof fetch = async (input, init) => { @@ -89,6 +283,9 @@ describe("Cursor Cloud Agents client", () => { { status: 200, headers: { "content-type": "application/json" } }, ); } + if (method === "GET" && url.endsWith("/stream")) { + return new Response("gone", { status: 410 }); + } if (method === "GET" && url.includes("/runs/run-1")) { poll += 1; const status = poll === 1 ? "RUNNING" : "FINISHED"; diff --git a/apps/api/src/ai.ts b/apps/api/src/ai.ts index 9f755fe..b52e654 100644 --- a/apps/api/src/ai.ts +++ b/apps/api/src/ai.ts @@ -2,7 +2,12 @@ import { eq } from "drizzle-orm"; import { aiSettings } from "@prm/db"; import { decryptSecret, getDb, getWorkspaceSecret, id, logActivity, nowIso } from "./store.js"; import { aiGenerations } from "@prm/db"; -import { CURSOR_CLOUD_API_BASE, cursorCloudMe, runCursorCloudAgent } from "./cursorCloud.js"; +import { + CURSOR_CLOUD_API_BASE, + cursorCloudMe, + runCursorCloudAgent, + type CursorCloudProgress, +} from "./cursorCloud.js"; export type AiFeature = | "evidence_digest" @@ -223,13 +228,24 @@ export async function probeCursorCloudKey(apiKey: string) { }; } -export async function runChat(feature: AiFeature, messages: ChatMessage[]) { +export type RunChatOpts = { + /** Live updates (Cursor Cloud Agents SSE status / text / tools). */ + onProgress?: (event: CursorCloudProgress) => void; +}; + +export async function runChat(feature: AiFeature, messages: ChatMessage[], opts: RunChatOpts = {}) { const cfg = getAiConfig(); if (!cfg.enabled) { throw new Error("AI disabled — enable in Settings → AI"); } const maxTokens = feature === "framework_extract" ? 4096 : 1600; if (cfg.localOnly || cfg.provider === "ollama") { + opts.onProgress?.({ + type: "status", + status: "RUNNING", + agentId: "ollama", + runId: "local", + }); const text = await callOllama(cfg.ollamaBaseUrl, cfg.modelDraft || "llama3.1", messages); return { text, model: `ollama:${cfg.modelDraft || "llama3.1"}`, provider: "ollama" as const }; } @@ -244,6 +260,12 @@ export async function runChat(feature: AiFeature, messages: ChatMessage[]) { } const model = feature === "evidence_digest" ? cfg.modelDigest : cfg.modelDraft; if (cfg.provider === "openai") { + opts.onProgress?.({ + type: "status", + status: "RUNNING", + agentId: "openai", + runId: "chat", + }); const text = await callOpenAI(cfg.apiKey, model || "gpt-4o-mini", messages); return { text, model: `openai:${model}`, provider: "openai" as const }; } @@ -254,6 +276,12 @@ export async function runChat(feature: AiFeature, messages: ChatMessage[]) { "or switch provider to Cursor Cloud Agents to use a crsr_… dashboard key.", ); } + opts.onProgress?.({ + type: "status", + status: "RUNNING", + agentId: "openai_compatible", + runId: "chat", + }); const text = await callOpenAI( cfg.apiKey, model || "gpt-4o-mini", @@ -273,9 +301,16 @@ export async function runChat(feature: AiFeature, messages: ChatMessage[]) { messages, feature, model: model || null, + onProgress: opts.onProgress, }); return { text: result.text, model: result.model, provider: "cursor" as const }; } + opts.onProgress?.({ + type: "status", + status: "RUNNING", + agentId: "anthropic", + runId: "chat", + }); const text = await callAnthropic(cfg.apiKey, model || "claude-sonnet-4-5", messages, maxTokens); return { text, model: `anthropic:${model}`, provider: "anthropic" as const }; } diff --git a/apps/api/src/cursorCloud.ts b/apps/api/src/cursorCloud.ts index f677e5c..d9f8820 100644 --- a/apps/api/src/cursorCloud.ts +++ b/apps/api/src/cursorCloud.ts @@ -3,6 +3,7 @@ * Dashboard keys (`crsr_…`) authenticate here — not OpenAI chat completions. * * PRM uses no-repo agents so digests/drafts can run without a GitHub repo. + * Prefers the run SSE stream for live text; falls back to polling when needed. * @see https://cursor.com/docs/cloud-agent/api/endpoints */ @@ -35,6 +36,19 @@ export type CursorCloudMe = { createdAt?: string; }; +/** Progress events for UI / callers while a Cloud Agent run is in flight. */ +export type CursorCloudProgress = + | { + type: "status"; + status: CursorCloudRunStatus; + agentId: string; + runId: string; + agentUrl?: string; + } + | { type: "text"; delta: string; cumulative: string } + | { type: "thinking"; delta: string } + | { type: "tool_call"; name: string; status: string; callId?: string }; + export type CursorCloudDeps = { fetchFn?: typeof fetch; sleep?: (ms: number) => Promise; @@ -48,10 +62,11 @@ const TERMINAL: ReadonlySet = new Set([ "EXPIRED", ]); -function authHeaders(apiKey: string): HeadersInit { +function authHeaders(apiKey: string, extra?: HeadersInit): HeadersInit { return { authorization: `Bearer ${apiKey}`, "content-type": "application/json", + ...extra, }; } @@ -60,6 +75,29 @@ async function readError(res: Response): Promise { return text.slice(0, 400) || res.statusText; } +function formatEmptyResultError(opts: { + agentId: string; + runId: string; + agentUrl?: string; + durationMs?: number | null; + assistantChars: number; + status: string; +}): string { + const parts = [ + `Cursor Cloud Agent finished with empty result (status ${opts.status}`, + `agent ${opts.agentId}`, + `run ${opts.runId}`, + ]; + if (opts.durationMs != null) parts.push(`${Math.round(opts.durationMs / 1000)}s`); + if (opts.assistantChars > 0) { + parts.push(`streamed ${opts.assistantChars} assistant chars that did not resolve to a final reply`); + } + const tip = opts.agentUrl + ? ` Open ${opts.agentUrl} to inspect the transcript.` + : " Open https://cursor.com/agents to inspect the run."; + return `${parts.join(", ")}).${tip}`; +} + export function formatCloudAgentPrompt( messages: Array<{ role: string; content: string }>, feature: string, @@ -146,20 +184,217 @@ async function archiveAgent(apiKey: string, agentId: string, deps: CursorCloudDe } } +type StreamOutcome = { + status: CursorCloudRunStatus; + /** Final reply from the `result` SSE event when present. */ + resultText: string; + /** Concatenated `assistant` SSE deltas (useful when result.text is empty). */ + assistantText: string; + durationMs?: number | null; +}; + +function parseSseBlock(block: string): { event?: string; data?: string; id?: string } | null { + const lines = block.split(/\r?\n/); + let event: string | undefined; + let id: string | undefined; + const dataLines: string[] = []; + for (const line of lines) { + if (!line || line.startsWith(":")) continue; + if (line.startsWith("event:")) event = line.slice(6).trim(); + else if (line.startsWith("id:")) id = line.slice(3).trim(); + else if (line.startsWith("data:")) dataLines.push(line.slice(5).trimStart()); + } + if (!event && dataLines.length === 0) return null; + return { event, id, data: dataLines.join("\n") }; +} + +/** + * Consume GET /v1/agents/{id}/runs/{runId}/stream until result/done/error. + * Returns null when the stream endpoint is unavailable (caller should poll). + */ +async function streamRun( + apiKey: string, + agentId: string, + runId: string, + agentUrl: string | undefined, + onProgress: ((p: CursorCloudProgress) => void) | undefined, + deps: CursorCloudDeps, + timeoutMs: number, + now: () => number, +): Promise { + const fetchFn = deps.fetchFn ?? fetch; + const res = await fetchFn(`${CURSOR_CLOUD_API_BASE}/agents/${agentId}/runs/${runId}/stream`, { + method: "GET", + headers: authHeaders(apiKey, { Accept: "text/event-stream" }), + }); + + // Stream expired or unsupported → fall back to polling. + if (res.status === 404 || res.status === 410 || res.status === 405) { + return null; + } + if (!res.ok) { + throw new Error(`Cursor Cloud Agents stream failed: ${res.status} ${await readError(res)}`); + } + if (!res.body) { + return null; + } + + const deadline = now() + timeoutMs; + let assistantText = ""; + let resultText = ""; + let status: CursorCloudRunStatus = "CREATING"; + let durationMs: number | null | undefined; + let finished = false; + + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + + const handleEvent = (raw: { event?: string; data?: string }) => { + const eventName = (raw.event || "message").trim(); + if (eventName === "heartbeat" || eventName === "done") { + if (eventName === "done") finished = true; + return; + } + + let payload: Record = {}; + if (raw.data) { + try { + payload = JSON.parse(raw.data) as Record; + } catch { + payload = { text: raw.data }; + } + } + + if (eventName === "status") { + const next = String(payload.status || status) as CursorCloudRunStatus; + status = next; + onProgress?.({ type: "status", status: next, agentId, runId, agentUrl }); + return; + } + + if (eventName === "assistant") { + const delta = typeof payload.text === "string" ? payload.text : ""; + if (delta) { + assistantText += delta; + onProgress?.({ type: "text", delta, cumulative: assistantText }); + } + return; + } + + if (eventName === "thinking") { + const delta = typeof payload.text === "string" ? payload.text : ""; + if (delta) onProgress?.({ type: "thinking", delta }); + return; + } + + if (eventName === "tool_call") { + onProgress?.({ + type: "tool_call", + name: String(payload.name || "tool"), + status: String(payload.status || "running"), + callId: typeof payload.callId === "string" ? payload.callId : undefined, + }); + return; + } + + if (eventName === "result") { + const next = String(payload.status || "FINISHED") as CursorCloudRunStatus; + status = next; + if (typeof payload.text === "string") resultText = payload.text; + if (typeof payload.durationMs === "number") durationMs = payload.durationMs; + onProgress?.({ type: "status", status: next, agentId, runId, agentUrl }); + finished = true; + return; + } + + if (eventName === "error") { + const message = String(payload.message || payload.code || "stream error"); + throw new Error(`Cursor Cloud Agent stream error: ${message}`); + } + + // Ignore interaction_update and unknown events (assistant/text already covered). + }; + + try { + while (!finished) { + if (now() > deadline) { + throw new Error( + `Cursor Cloud Agent timed out after ${Math.round(timeoutMs / 1000)}s (status ${status}). ` + + `Open ${agentUrl ?? "https://cursor.com/agents"} to inspect the run.`, + ); + } + const { done, value } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + // SSE events are separated by a blank line. + while (true) { + const match = buffer.match(/\r?\n\r?\n/); + if (!match || match.index === undefined) break; + const block = buffer.slice(0, match.index); + buffer = buffer.slice(match.index + match[0].length); + const parsed = parseSseBlock(block); + if (parsed) handleEvent(parsed); + if (finished) break; + } + } + } finally { + try { + reader.releaseLock(); + } catch { + /* ignore */ + } + } + + return { status, resultText, assistantText, durationMs }; +} + +async function pollRun( + apiKey: string, + agentId: string, + runId: string, + agentUrl: string | undefined, + onProgress: ((p: CursorCloudProgress) => void) | undefined, + deps: CursorCloudDeps, + timeoutMs: number, + pollMs: number, + now: () => number, + sleep: (ms: number) => Promise, + initial: CursorCloudRun, +): Promise { + const deadline = now() + timeoutMs; + let run = initial; + onProgress?.({ type: "status", status: run.status, agentId, runId, agentUrl }); + while (!TERMINAL.has(run.status)) { + if (now() > deadline) { + throw new Error( + `Cursor Cloud Agent timed out after ${Math.round(timeoutMs / 1000)}s (status ${run.status}). ` + + `Open ${agentUrl ?? "https://cursor.com/agents"} to inspect the run.`, + ); + } + await sleep(pollMs); + run = await getRun(apiKey, agentId, runId, deps); + onProgress?.({ type: "status", status: run.status, agentId, runId, agentUrl }); + } + return run; +} + export type RunCursorCloudAgentOpts = { apiKey: string; messages: Array<{ role: string; content: string }>; feature: string; /** Cursor model id from GET /v1/models; omit to use account default. */ model?: string | null; - /** Polling timeout (ms). Default 180s. */ + /** Polling/stream timeout (ms). Default 180s. */ timeoutMs?: number; - /** Poll interval (ms). Default 2s. */ + /** Poll interval (ms) when SSE is unavailable. Default 2s. */ pollMs?: number; + /** Live status / text / tool updates (Cloud Agents SSE). */ + onProgress?: (event: CursorCloudProgress) => void; }; /** - * Launch a no-repo Cloud Agent, poll until terminal, return assistant text, then archive. + * Launch a no-repo Cloud Agent, stream (or poll) until terminal, return assistant text, then archive. */ export async function runCursorCloudAgent( opts: RunCursorCloudAgentOpts, @@ -181,37 +416,94 @@ export async function runCursorCloudAgent( ); const agentId = created.agent.id; const runId = created.run.id; + const agentUrl = created.agent.url; const modelLabel = opts.model?.trim() || "default"; + opts.onProgress?.({ + type: "status", + status: created.run.status, + agentId, + runId, + agentUrl, + }); + try { - const deadline = now() + timeoutMs; - let run = created.run as CursorCloudRun; - while (!TERMINAL.has(run.status)) { - if (now() > deadline) { - throw new Error( - `Cursor Cloud Agent timed out after ${Math.round(timeoutMs / 1000)}s (status ${run.status}). ` + - `Open ${created.agent.url ?? "https://cursor.com/agents"} to inspect the run.`, - ); + let text = ""; + let status: CursorCloudRunStatus = created.run.status; + let durationMs: number | null | undefined; + let assistantChars = 0; + + const streamed = await streamRun( + opts.apiKey, + agentId, + runId, + agentUrl, + opts.onProgress, + deps, + timeoutMs, + now, + ).catch(async (err) => { + // Network / parse failures → try polling instead of failing hard mid-run. + if (err instanceof Error && /timed out|stream error/i.test(err.message)) throw err; + return null; + }); + + if (streamed) { + status = streamed.status; + durationMs = streamed.durationMs; + assistantChars = streamed.assistantText.trim().length; + text = (streamed.resultText || streamed.assistantText || "").trim(); + // If SSE ended without a clear final reply, refresh via GET. + if (!text || !TERMINAL.has(status)) { + const run = await getRun(opts.apiKey, agentId, runId, deps); + status = run.status; + durationMs = run.durationMs ?? durationMs; + if (!text) text = (run.result ?? "").trim(); } - await sleep(pollMs); - run = await getRun(opts.apiKey, agentId, runId, deps); + } else { + const run = await pollRun( + opts.apiKey, + agentId, + runId, + agentUrl, + opts.onProgress, + deps, + timeoutMs, + pollMs, + now, + sleep, + created.run as CursorCloudRun, + ); + status = run.status; + durationMs = run.durationMs; + text = (run.result ?? "").trim(); } - if (run.status !== "FINISHED") { + + if (status !== "FINISHED") { throw new Error( - `Cursor Cloud Agent ended with ${run.status}` + - (run.result ? `: ${String(run.result).slice(0, 200)}` : ""), + `Cursor Cloud Agent ended with ${status}` + + (text ? `: ${text.slice(0, 200)}` : "") + + (agentUrl ? ` — ${agentUrl}` : ""), ); } - const text = (run.result ?? "").trim(); if (!text) { - throw new Error("Cursor Cloud Agent finished with empty result"); + throw new Error( + formatEmptyResultError({ + agentId, + runId, + agentUrl, + durationMs, + assistantChars, + status, + }), + ); } return { text, model: `cursor-cloud:${modelLabel}`, agentId, runId, - agentUrl: created.agent.url, + agentUrl, }; } finally { await archiveAgent(opts.apiKey, agentId, deps); diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index 0b61a98..3254bea 100644 --- a/apps/api/src/index.ts +++ b/apps/api/src/index.ts @@ -1,6 +1,7 @@ import { serve } from "@hono/node-server"; import { Hono } from "hono"; import { cors } from "hono/cors"; +import { streamSSE } from "hono/streaming"; import { access, readFile } from "node:fs/promises"; import { basename, extname, join, normalize, resolve, sep } from "node:path"; import { and, eq, isNull } from "drizzle-orm"; @@ -3484,6 +3485,84 @@ app.post("/api/ai/chat", async (c) => { ...history, ]; + const wantsStream = (c.req.header("accept") || "").includes("text/event-stream"); + + if (wantsStream) { + return streamSSE(c, async (stream) => { + let eventId = 0; + let writeChain: Promise = Promise.resolve(); + const send = (event: string, data: unknown) => { + writeChain = writeChain.then(async () => { + eventId += 1; + await stream.writeSSE({ + id: String(eventId), + event, + data: JSON.stringify(data), + }); + }); + return writeChain; + }; + try { + await send("status", { + status: "CREATING", + detail: "Starting model…", + }); + const result = await runChat("chat_dossier", packed, { + onProgress: (p) => { + if (p.type === "status") { + void send("status", { + status: p.status, + agentId: p.agentId, + runId: p.runId, + agentUrl: p.agentUrl, + detail: + p.status === "CREATING" + ? "Creating Cursor Cloud Agent…" + : p.status === "RUNNING" + ? "Cloud Agent is working…" + : `Cloud Agent ${p.status.toLowerCase()}`, + }); + } else if (p.type === "text") { + void send("delta", { text: p.delta, cumulative: p.cumulative }); + } else if (p.type === "thinking") { + void send("thinking", { text: p.delta }); + } else if (p.type === "tool_call") { + void send("tool", { + name: p.name, + status: p.status, + callId: p.callId, + }); + } + }, + }); + await writeChain; + const filtered = filterCitationsAgainstAllowlist(result.text, ctx.citations); + const generationId = storeGeneration({ + feature: "chat_dossier", + subjectPersonId: personId, + cycleId, + model: result.model, + outputText: filtered.text, + citations: filtered.citationsUsed, + dataClassesSent: ctx.dataClassesSent, + }); + await send("done", { + generationId, + reply: filtered.text, + model: result.model, + provider: result.provider, + citations: filtered.citationsUsed, + citationsDropped: filtered.citationsDropped, + dataClassesSent: ctx.dataClassesSent, + scopeLabel: ctx.scopeLabel, + }); + } catch (e) { + await writeChain.catch(() => undefined); + await send("error", { error: e instanceof Error ? e.message : "AI failed" }); + } + }); + } + try { const result = await runChat("chat_dossier", packed); const filtered = filterCitationsAgainstAllowlist(result.text, ctx.citations); diff --git a/apps/ui/src/lib/api.ts b/apps/ui/src/lib/api.ts index 7bacc87..0cd2147 100644 --- a/apps/ui/src/lib/api.ts +++ b/apps/ui/src/lib/api.ts @@ -83,3 +83,130 @@ export async function api(path: string, init: RequestInit = {}): Promise { } return res.json() as Promise; } + +export type ChatStreamHandlers = { + onStatus?: (payload: { + status?: string; + detail?: string; + agentId?: string; + runId?: string; + agentUrl?: string; + }) => void; + onDelta?: (payload: { text: string; cumulative: string }) => void; + onTool?: (payload: { name: string; status: string; callId?: string }) => void; + onThinking?: (payload: { text: string }) => void; +}; + +export type ChatStreamResult = { + reply: string; + model: string; + scopeLabel?: string; + generationId?: string; + provider?: string; +}; + +/** + * POST /api/ai/chat with SSE. Falls back to JSON if the server does not stream. + */ +export async function streamChat( + body: unknown, + handlers: ChatStreamHandlers = {}, +): Promise { + const headers = authHeaders({ + "Content-Type": "application/json", + Accept: "text/event-stream", + }); + const res = await fetch(apiUrl("/api/ai/chat"), { + method: "POST", + headers, + body: JSON.stringify(body), + }); + + const contentType = res.headers.get("content-type") || ""; + if (!res.ok) { + const err = await res.json().catch(() => ({ error: res.statusText })); + throw new Error((err as { error?: string }).error || "Request failed"); + } + + if (!contentType.includes("text/event-stream") || !res.body) { + const json = (await res.json()) as ChatStreamResult & { error?: string }; + if (json.error) throw new Error(json.error); + return json; + } + + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + const state: { done: ChatStreamResult | null; error: string | null } = { + done: null, + error: null, + }; + + const handleBlock = (block: string) => { + const lines = block.split(/\r?\n/); + let event = "message"; + const dataLines: string[] = []; + for (const line of lines) { + if (!line || line.startsWith(":")) continue; + if (line.startsWith("event:")) event = line.slice(6).trim(); + else if (line.startsWith("data:")) dataLines.push(line.slice(5).trimStart()); + } + if (!dataLines.length) return; + let data: Record = {}; + try { + data = JSON.parse(dataLines.join("\n")) as Record; + } catch { + return; + } + if (event === "status") { + handlers.onStatus?.({ + status: typeof data.status === "string" ? data.status : undefined, + detail: typeof data.detail === "string" ? data.detail : undefined, + agentId: typeof data.agentId === "string" ? data.agentId : undefined, + runId: typeof data.runId === "string" ? data.runId : undefined, + agentUrl: typeof data.agentUrl === "string" ? data.agentUrl : undefined, + }); + } else if (event === "delta") { + handlers.onDelta?.({ + text: String(data.text ?? ""), + cumulative: String(data.cumulative ?? data.text ?? ""), + }); + } else if (event === "thinking") { + handlers.onThinking?.({ text: String(data.text ?? "") }); + } else if (event === "tool") { + handlers.onTool?.({ + name: String(data.name ?? "tool"), + status: String(data.status ?? "running"), + callId: typeof data.callId === "string" ? data.callId : undefined, + }); + } else if (event === "done") { + state.done = { + reply: String(data.reply ?? ""), + model: String(data.model ?? ""), + scopeLabel: typeof data.scopeLabel === "string" ? data.scopeLabel : undefined, + generationId: typeof data.generationId === "string" ? data.generationId : undefined, + provider: typeof data.provider === "string" ? data.provider : undefined, + }; + } else if (event === "error") { + state.error = String(data.error ?? "Chat failed"); + } + }; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + while (true) { + const match = buffer.match(/\r?\n\r?\n/); + if (!match || match.index === undefined) break; + const block = buffer.slice(0, match.index); + buffer = buffer.slice(match.index + match[0].length); + handleBlock(block); + } + } + if (buffer.trim()) handleBlock(buffer); + + if (state.error) throw new Error(state.error); + if (!state.done?.reply) throw new Error("Chat stream ended without a reply"); + return state.done; +} diff --git a/apps/ui/src/pages/ChatPage.tsx b/apps/ui/src/pages/ChatPage.tsx index 6ba93f2..b0b222f 100644 --- a/apps/ui/src/pages/ChatPage.tsx +++ b/apps/ui/src/pages/ChatPage.tsx @@ -1,7 +1,7 @@ import { useEffect, useRef, useState } from "react"; import { Link } from "react-router-dom"; import type { PersonDTO } from "@prm/shared"; -import { api } from "../lib/api"; +import { api, streamChat } from "../lib/api"; import { estimateThenConfirm } from "../lib/aiEstimate"; import { NextStep, PageHeader } from "../components/PageChrome"; @@ -26,6 +26,9 @@ type ChatTurn = { content: string; model?: string; error?: boolean; + pending?: boolean; + statusDetail?: string; + agentUrl?: string; }; const WORKSPACE_SUGGESTIONS = [ @@ -100,8 +103,16 @@ export function ChatPage() { } const userTurn: ChatTurn = { id: newId(), role: "user", content }; + const assistantId = newId(); + const pendingTurn: ChatTurn = { + id: assistantId, + role: "assistant", + content: "", + pending: true, + statusDetail: "Reading evidence…", + }; const nextHistory = [...messages, userTurn]; - setMessages(nextHistory); + setMessages([...nextHistory, pendingTurn]); setDraft(""); setSending(true); setError(null); @@ -130,30 +141,87 @@ export function ChatPage() { const payload = nextHistory .filter((m) => !m.error) .map((m) => ({ role: m.role, content: m.content })); - const res = await api<{ - reply: string; - model: string; - scopeLabel: string; - }>("/api/ai/chat", { - method: "POST", - body: JSON.stringify({ + const res = await streamChat( + { messages: payload, personId: personId || null, cycleId: cycleId || null, confirmWorkspaceScope: !personId, - }), - }); - setMessages((prev) => [ - ...prev, - { id: newId(), role: "assistant", content: res.reply, model: res.model }, - ]); + }, + { + onStatus: (s) => { + setMessages((prev) => + prev.map((m) => + m.id === assistantId + ? { + ...m, + statusDetail: s.detail || s.status || m.statusDetail, + agentUrl: s.agentUrl || m.agentUrl, + } + : m, + ), + ); + }, + onDelta: (d) => { + setMessages((prev) => + prev.map((m) => + m.id === assistantId + ? { + ...m, + content: d.cumulative, + statusDetail: undefined, + pending: true, + } + : m, + ), + ); + }, + onTool: (t) => { + setMessages((prev) => + prev.map((m) => + m.id === assistantId + ? { + ...m, + statusDetail: `${t.name} ${t.status}`, + } + : m, + ), + ); + }, + }, + ); + setMessages((prev) => + prev.map((m) => + m.id === assistantId + ? { + id: assistantId, + role: "assistant", + content: res.reply, + model: res.model, + pending: false, + statusDetail: undefined, + } + : m, + ), + ); } catch (e) { const msg = e instanceof Error ? e.message : "Chat failed"; setError(msg); - setMessages((prev) => [ - ...prev, - { id: newId(), role: "assistant", content: msg, error: true }, - ]); + setMessages((prev) => + prev.map((m) => + m.id === assistantId + ? { + id: assistantId, + role: "assistant", + content: msg, + error: true, + pending: false, + statusDetail: undefined, + agentUrl: m.agentUrl, + } + : m, + ), + ); } finally { setSending(false); inputRef.current?.focus(); @@ -263,7 +331,7 @@ export function ChatPage() {
- {messages.length === 0 && !sending ? ( + {messages.length === 0 ? (

Ask anything about loaded workbench data

@@ -288,22 +356,28 @@ export function ChatPage() { messages.map((m) => (

{m.role === "user" ? "You" : "Assistant"} {m.model ? · {m.model} : null} + {m.pending && m.statusDetail && m.content ? ( + · {m.statusDetail} + ) : null}
-
{m.content}
+
+ {m.content || (m.pending ? m.statusDetail || "Working…" : "")} +
+ {m.error && m.agentUrl ? ( +

+ + Open Cloud Agent run + +

+ ) : null}
)) )} - {sending ? ( -
-
Assistant
-
Reading evidence…
-
- ) : null}
diff --git a/apps/ui/src/styles.css b/apps/ui/src/styles.css index 6f7fb2c..331b323 100644 --- a/apps/ui/src/styles.css +++ b/apps/ui/src/styles.css @@ -1027,6 +1027,13 @@ li { margin-bottom: 0.4rem; line-height: 1.55; color: var(--ink-soft); } background: var(--danger-tint); color: var(--danger); } +.chat-bubble-link { + margin: 0.4rem 0 0; + font-size: 0.85rem; +} +.chat-bubble-link a { + color: inherit; +} .chat-bubble-pending { opacity: 0.75; } .chat-bubble-meta { font-size: 0.75rem; diff --git a/docs/ai/AI_INTEGRATION.md b/docs/ai/AI_INTEGRATION.md index c83aa48..791c321 100644 --- a/docs/ai/AI_INTEGRATION.md +++ b/docs/ai/AI_INTEGRATION.md @@ -24,7 +24,7 @@ Workbench trust: [TRUST_MODEL.md](../architecture/TRUST_MODEL.md). - Provider: Anthropic / OpenAI / **Cursor Cloud Agents** / OpenAI-compatible (custom URL) / Ollama - API key → encrypted with workspace secret (masked in UI); Test connection saves then probes - Anthropic / OpenAI: standard cloud keys - - **Cursor Cloud Agents**: dashboard key (`crsr_…` from [cursor.com/dashboard/api](https://cursor.com/dashboard/api)); digests/drafts launch a short-lived **no-repo** agent via `https://api.cursor.com/v1` (Test probes `/v1/me`) + - **Cursor Cloud Agents**: dashboard key (`crsr_…` from [cursor.com/dashboard/api](https://cursor.com/dashboard/api)); digests/drafts/chat launch a short-lived **no-repo** agent via `https://api.cursor.com/v1` (Test probes `/v1/me`). Runs prefer the Cloud Agents **SSE stream** (`/runs/{id}/stream`) so chat can show status and assistant text live; empty terminal replies include agent/run ids and a link to inspect the transcript. - OpenAI-compatible: any gateway that implements `/v1/chat/completions` (set Base URL explicitly) - Optional env: `CURSOR_API_KEY` (Cloud Agents or gateway); `CURSOR_API_BASE_URL` (gateway only) - Models per capability (blank Cursor models → account default)