From 47cb1826a8a37b0f3e39b075a1bc76d79f32710e Mon Sep 17 00:00:00 2001 From: Carlos Escalante Date: Sat, 4 Jul 2026 21:09:24 -0600 Subject: [PATCH] Assistant: migrate to v2 runtime handler with Postgres-backed AgentRunner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the deprecated copilotRuntimeNextJSAppRouterEndpoint wrapper (which was already the v2 single-route handler underneath) with createCopilotRuntimeHandler from @copilotkit/runtime/v2, built once at module scope, and plug in a custom PostgresAgentRunner: - run() proxies the agent's event stream untouched — MAF persists every turn in chatthreadsnapshot, so the BFF records nothing - connect() forwards to MAF's snapshot hydration (empty-messages run → RUN_STARTED → MESSAGES_SNAPSHOT → RUN_FINISHED, no LLM call), making the DB the only conversation store — correct across clears, restarts and future instances - stop() aborts via a process-local live-run map (same pattern as the official sqlite-runner); isRunning() has no caller in the v2 SSE path This deletes the whole workaround layer the in-memory runner forced: the connect intercepts (URL + envelope forms) and Asistente's client-side hydration run. The ask-box flow now awaits connectAgent instead. Auth moves to a single Request rewrite (ws_token cookie → Authorization Bearer) before the handler — the v2 path forwards authorization + x-* headers to both run clones and runner.connect. Stays in single-route mode: same envelope protocol the client already auto-detected, no exposure to CopilotKit#4953, /threads* REST not needed. Telemetry disabled via COPILOTKIT_TELEMETRY_DISABLED. Verified full matrix in dev: fresh ask, reload replay via runner, 3 cleared convos + double navigation (only last survives), card renders once and survives follow-up + reload, Inicio ask-box with history, stale-tab persists only its new turn. 126 backend tests green. Co-Authored-By: Claude Fable 5 --- frontend/server.ts | 279 +++++++++++++++++++------------ frontend/src/pages/Asistente.tsx | 41 ++--- 2 files changed, 188 insertions(+), 132 deletions(-) diff --git a/frontend/server.ts b/frontend/server.ts index a429435..512b1c0 100644 --- a/frontend/server.ts +++ b/frontend/server.ts @@ -1,12 +1,12 @@ import { serve } from "@hono/node-server"; import { serveStatic } from "@hono/node-server/serve-static"; import { + AgentRunner, CopilotRuntime, - ExperimentalEmptyAdapter, - copilotRuntimeNextJSAppRouterEndpoint, -} from "@copilotkit/runtime"; + createCopilotRuntimeHandler, +} from "@copilotkit/runtime/v2"; import { HttpAgent, Middleware, EventType } from "@ag-ui/client"; -import { Observable } from "rxjs"; +import { Observable, ReplaySubject } from "rxjs"; import { Hono } from "hono"; const BACKEND_URL = @@ -383,52 +383,21 @@ app.use("*", async (c, next) => { }); }); -app.all("/api/copilotkit/*", async (c) => { - // CopilotChat fires "connect" on every mount with an explicit threadId, - // and the runtime's in-memory runner answers it by replaying THIS - // PROCESS's event history — every run since the server started, across - // "Nueva conversación" clears (the DB delete can't touch process memory). - // That resurrected cleared conversations on navigation. The - // empty-messages hydration run against MAF's snapshot store is our single - // restore path, so connect gets an immediately-completed stream instead. - // - // Two transports must be caught: multi-route (POST …/agent//connect) - // and single-route (POST /api/copilotkit with {method:"agent/connect"} in - // the body — the one CopilotKit actually uses here; the URL-only check - // silently missed it). - const emptyStream = () => - new Response("", { - status: 200, - headers: { "Content-Type": "text/event-stream" }, - }); - if (new URL(c.req.url).pathname.endsWith("/connect")) { - return emptyStream(); - } - if ( - c.req.method === "POST" && - (c.req.header("content-type") ?? "").includes("application/json") - ) { - try { - const envelope = (await c.req.raw.clone().json()) as { method?: string }; - if (envelope?.method === "agent/connect") { - return emptyStream(); - } - } catch { - // not JSON — fall through to the runtime - } - } +// ── CopilotKit v2 runtime (module scope, built once) ───────────────────────── +// Migrated 2026-07-04 off the deprecated copilotRuntimeNextJSAppRouterEndpoint +// wrapper (which was already the v2 single-route handler underneath) onto +// createCopilotRuntimeHandler directly, with a Postgres-backed AgentRunner: +// run() is a pure proxy to the MAF backend (which persists every turn in +// chatthreadsnapshot), and connect() forwards to MAF's snapshot hydration — +// so the ONLY conversation store is the DB. This deletes the in-memory +// runner (whose process-local replay resurrected cleared conversations) and +// the client-side hydration hack that compensated for it. - const cookieHeader = c.req.header("cookie") ?? ""; - const match = cookieHeader.match(/(?:^|;\s*)ws_token=([^;]+)/); - const token = match?.[1]; - const agentHeaders: Record = token - ? { Authorization: `Bearer ${token}` } - : {}; - // Note: the browser's X-Client-Timezone header (set in App.tsx) reaches - // the backend without help — CopilotKit's runtime forwards authorization - // and all x-* headers to the agent (extractForwardableHeaders). +// The v2 handler nags about anonymous telemetry unless disabled. +process.env.COPILOTKIT_TELEMETRY_DISABLED ??= "true"; - const agent = new HttpAgent({ url: AGENT_URL, headers: agentHeaders }); +function makeWealthysmartAgent(headers: Record = {}): HttpAgent { + const agent = new HttpAgent({ url: AGENT_URL, headers }); // CK_MIDDLEWARES=off runs the stack bare — used to audit which of these // workarounds are still needed after CopilotKit/MAF upgrades. /* eslint-disable @typescript-eslint/no-explicit-any */ @@ -442,74 +411,166 @@ app.all("/api/copilotkit/*", async (c) => { agent.use(new DebugLogMiddleware() as any); } /* eslint-enable @typescript-eslint/no-explicit-any */ + return agent; +} - const runtime = new CopilotRuntime({ - agents: { wealthysmart: agent }, - // a2ui disabled 2026-07-04: CK 1.60+ validates generated ops against a - // catalog and our agent's ops fail validation (surface never paints). - // Markdown tables cover the use case until the pipeline matures. - a2ui: { enabled: false }, - beforeRequestMiddleware: async ({ request: outbound }) => { - if (outbound.method !== "POST") return; - const ct = outbound.headers.get("content-type") ?? ""; - if (!ct.includes("application/json")) return; - try { - const body = (await outbound.clone().json()) as { - messages?: AGUIMessage[]; - context?: { description?: string; value?: string }[]; - }; - if (!Array.isArray(body.messages)) return; - // An empty messages array is MAF's hydration trigger (known threadId - // + no messages → replay the stored snapshot without invoking the - // LLM). Injecting anything would turn it into a normal run. - if (body.messages.length === 0) return; - let messages = pairOrphanToolCalls(body.messages); +// ── PostgresAgentRunner ─────────────────────────────────────────────────────── +// The conversation store is Postgres (chatthreadsnapshot), written by MAF at +// every run end. So the runner records NOTHING: +// - run() proxies the agent's event stream (the per-request clone the v2 +// handler prepared, auth headers already forwarded onto it). +// - connect() asks MAF for its snapshot hydration — an empty-messages run +// returns RUN_STARTED → [STATE_SNAPSHOT] → MESSAGES_SNAPSHOT → RUN_FINISHED +// with no LLM call (unknown thread: RUN_STARTED → RUN_FINISHED). The same +// stream shape the in-memory runner's replay approximated, but sourced +// from the DB, so it is correct across clears, restarts and instances. +// - Live runs are tracked in-process only for stop(); the official +// sqlite-runner does the same (single-instance limitation acknowledged). +// isRunning() has no caller in the v2 SSE path (verified at 1.62.2). - // MAF's AG-UI endpoint declares `context` on its request model but - // never feeds it to the model (agent_framework_ag_ui, checked at - // python-1.10.0). CopilotKit 1.60+ ships the A2UI catalog schema and - // guidelines in that context — without them the model emits ops the - // A2UI validator rejects and the surface never paints. Inject the - // context as a system message ourselves. Remove once MAF consumes - // RunAgentInput.context. - if (Array.isArray(body.context) && body.context.length > 0) { - const contextText = body.context - .map((c) => `## ${c.description ?? "Context"}\n${c.value ?? ""}`) - .join("\n\n"); - messages = [ - { - id: `ctx-${Date.now()}`, - role: "system", - content: `Additional run context:\n\n${contextText}`, - } as AGUIMessage, - ...messages, - ]; - } +// eslint-disable-next-line @typescript-eslint/no-explicit-any +const liveRuns = new Map(); - if (messages === body.messages) return; - return new Request(outbound.url, { - method: outbound.method, - headers: outbound.headers, - body: JSON.stringify({ ...body, messages }), - }); - } catch (err) { - // Skipping the repair is survivable; doing it silently is not — - // orphan tool calls then fail downstream with no trace of why. - console.error("[copilotkit] outbound request repair skipped:", err); - return; +// eslint-disable-next-line @typescript-eslint/no-explicit-any +function proxyAgentRun(agent: any, input: any, threadId: string): Observable { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const subject = new ReplaySubject(Infinity); + liveRuns.set(threadId, agent); + agent + // eslint-disable-next-line @typescript-eslint/no-explicit-any + .runAgent(input, { onEvent: ({ event }: any) => subject.next(event) }) + .then(() => subject.complete()) + // eslint-disable-next-line @typescript-eslint/no-explicit-any + .catch((err: any) => subject.error(err)) + .finally(() => { + if (liveRuns.get(threadId) === agent) liveRuns.delete(threadId); + }); + return subject.asObservable(); +} + +// eslint-disable-next-line @typescript-eslint/no-explicit-any +class PostgresAgentRunner extends (AgentRunner as any) { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + run(request: any): Observable { + return proxyAgentRun(request.agent, request.input, request.threadId); + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + connect(request: any): Observable { + // request.headers = authorization + x-* from the incoming browser + // request (extractForwardableHeaders) — exactly what MAF needs. + const agent = makeWealthysmartAgent(request.headers ?? {}); + agent.threadId = request.threadId; + // No messages + known threadId = MAF's hydration trigger. + return proxyAgentRun(agent, {}, `connect:${request.threadId}`); + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + isRunning(request: any): Promise { + return Promise.resolve(liveRuns.has(request.threadId)); + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + async stop(request: any): Promise { + const agent = liveRuns.get(request.threadId); + if (!agent) return false; + try { + agent.abortRun(); + liveRuns.delete(request.threadId); + return true; + } catch { + return false; + } + } +} + +const runtime = new CopilotRuntime({ + agents: { wealthysmart: makeWealthysmartAgent() }, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + runner: new PostgresAgentRunner() as any, + // a2ui disabled 2026-07-04: CK 1.60+ validates generated ops against a + // catalog and our agent's ops fail validation (surface never paints). + // Markdown tables cover the use case until the pipeline matures. + a2ui: { enabled: false }, + beforeRequestMiddleware: async ({ request: outbound }) => { + if (outbound.method !== "POST") return; + const ct = outbound.headers.get("content-type") ?? ""; + if (!ct.includes("application/json")) return; + try { + const body = (await outbound.clone().json()) as { + messages?: AGUIMessage[]; + context?: { description?: string; value?: string }[]; + }; + if (!Array.isArray(body.messages)) return; + // An empty messages array is MAF's hydration trigger (known threadId + // + no messages → replay the stored snapshot without invoking the + // LLM). Injecting anything would turn it into a normal run. + if (body.messages.length === 0) return; + let messages = pairOrphanToolCalls(body.messages); + + // MAF's AG-UI endpoint declares `context` on its request model but + // never feeds it to the model (agent_framework_ag_ui, checked at + // python-1.10.0). CopilotKit 1.60+ ships the A2UI catalog schema and + // guidelines in that context — without them the model emits ops the + // A2UI validator rejects and the surface never paints. Inject the + // context as a system message ourselves. Remove once MAF consumes + // RunAgentInput.context. + if (Array.isArray(body.context) && body.context.length > 0) { + const contextText = body.context + .map((c) => `## ${c.description ?? "Context"}\n${c.value ?? ""}`) + .join("\n\n"); + messages = [ + { + id: `ctx-${Date.now()}`, + role: "system", + content: `Additional run context:\n\n${contextText}`, + } as AGUIMessage, + ...messages, + ]; } - }, - }); - const { handleRequest } = copilotRuntimeNextJSAppRouterEndpoint({ - runtime, - serviceAdapter: new ExperimentalEmptyAdapter(), - endpoint: "/api/copilotkit", - }); - - return handleRequest(c.req.raw as Parameters[0]); + if (messages === body.messages) return; + return new Request(outbound.url, { + method: outbound.method, + headers: outbound.headers, + body: JSON.stringify({ ...body, messages }), + }); + } catch (err) { + // Skipping the repair is survivable; doing it silently is not — + // orphan tool calls then fail downstream with no trace of why. + console.error("[copilotkit] outbound request repair skipped:", err); + return; + } + }, }); +const ckHandler = createCopilotRuntimeHandler({ + runtime, + basePath: "/api/copilotkit", + // Single-route mode: same envelope protocol the client already resolved + // via auto-detect (its GET /info probe 405s here, so it stays "single"). + // Multi-route would only add /threads* REST endpoints we don't use and is + // exposed to CopilotKit#4953 (frontend tools under multi-route). + mode: "single-route", + cors: false, +}); + +// The v2 handler forwards only `authorization` and `x-*` headers to agents +// and to runner.connect — cookies never pass. Translate the auth cookie into +// a Bearer header before the handler sees the request. +function withBearerFromCookie(req: Request): Request { + if (req.headers.get("authorization")) return req; + const cookie = req.headers.get("cookie") ?? ""; + const match = cookie.match(/(?:^|;\s*)ws_token=([^;]+)/); + if (!match) return req; + const headers = new Headers(req.headers); + headers.set("Authorization", `Bearer ${match[1]}`); + return new Request(req, { headers }); +} + +app.all("/api/copilotkit/*", (c) => ckHandler(withBearerFromCookie(c.req.raw))); +app.all("/api/copilotkit", (c) => ckHandler(withBearerFromCookie(c.req.raw))); + app.get("/api/health", (c) => c.json({ ok: true })); // Proxy backend API calls (FastAPI). In dev these hit Vite's proxy directly, diff --git a/frontend/src/pages/Asistente.tsx b/frontend/src/pages/Asistente.tsx index f857e0a..f94cc42 100644 --- a/frontend/src/pages/Asistente.tsx +++ b/frontend/src/pages/Asistente.tsx @@ -39,39 +39,34 @@ export default function Asistente() { const [confirmClear, setConfirmClear] = useState(false); const [clearing, setClearing] = useState(false); - // Boot sequence, once per page load: hydrate the persisted thread, then - // send any question that arrived from the Inicio ask-box via router state. - // Order matters — hydration first keeps the invariant that every normal - // run's input carries the full client-known history (the BFF's snapshot - // reconciliation relies on it). + // Ask-box flow: a question typed on Inicio arrives via router state. + // History hydration is no longer done here — CopilotChat's mount connect + // replays the persisted thread through the BFF's PostgresAgentRunner. We + // still await our own connect before sending so the run's input carries + // the full client-known history (idempotent with CopilotChat's connect: + // the MESSAGES_SNAPSHOT merge is id-based). const bootRef = useRef(false); useEffect(() => { - if (!agent || bootRef.current) return; - bootRef.current = true; const ask = (location.state as { ask?: string } | null)?.ask; - if (ask) { - // Clear the state so back/refresh doesn't re-send the question. - navigate(location.pathname, { replace: true, state: null }); - } + if (!ask || !agent || bootRef.current) return; + bootRef.current = true; + // Clear the state so back/refresh doesn't re-send the question. + navigate(location.pathname, { replace: true, state: null }); void (async () => { agent.threadId = CHAT_THREAD_ID; if (agent.messages.length === 0) { - // Empty-messages run = MAF's hydration trigger: replays the stored - // snapshot as MESSAGES_SNAPSHOT without invoking the LLM. try { - await copilotkit.runAgent({ agent }); + await copilotkit.connectAgent({ agent }); } catch (err) { - console.error("[asistente] thread hydration failed:", err); + console.error("[asistente] connect before ask failed:", err); } } - if (ask) { - agent.addMessage({ - id: crypto.randomUUID(), - role: "user", - content: ask, - }); - void copilotkit.runAgent({ agent }); - } + agent.addMessage({ + id: crypto.randomUUID(), + role: "user", + content: ask, + }); + void copilotkit.runAgent({ agent }); })(); // eslint-disable-next-line react-hooks/exhaustive-deps }, [agent]);