mirror of
https://github.com/escalante29/WealthySmart.git
synced 2026-07-17 10:08:46 +02:00
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 <noreply@anthropic.com>
617 lines
27 KiB
TypeScript
617 lines
27 KiB
TypeScript
import { serve } from "@hono/node-server";
|
|
import { serveStatic } from "@hono/node-server/serve-static";
|
|
import {
|
|
AgentRunner,
|
|
CopilotRuntime,
|
|
createCopilotRuntimeHandler,
|
|
} from "@copilotkit/runtime/v2";
|
|
import { HttpAgent, Middleware, EventType } from "@ag-ui/client";
|
|
import { Observable, ReplaySubject } from "rxjs";
|
|
import { Hono } from "hono";
|
|
|
|
const BACKEND_URL =
|
|
process.env.BACKEND_URL ?? "http://localhost:8001";
|
|
// Prod sets AGENT_URL explicitly (compose network); the default derives from
|
|
// BACKEND_URL so local dev reaches the docker backend published on :8001.
|
|
const AGENT_URL =
|
|
process.env.AGENT_URL ?? `${BACKEND_URL}/api/v1/agent/agui`;
|
|
|
|
const isProd = process.env.NODE_ENV === "production";
|
|
const PORT = parseInt(process.env.PORT ?? (isProd ? "3000" : "3001"));
|
|
|
|
// ── pairOrphanToolCalls ──────────────────────────────────────────────────────
|
|
// CopilotKit's browser-side store sometimes drops the tool-role message
|
|
// between turns, producing assistant(toolCalls=[X]) → assistant(text) which
|
|
// OpenAI rejects. Inject a synthetic empty tool response for each orphan id.
|
|
|
|
interface AGUIMessage {
|
|
id?: string;
|
|
role?: string;
|
|
content?: unknown;
|
|
toolCalls?: Array<{ id?: string }>;
|
|
tool_calls?: Array<{ id?: string }>;
|
|
toolCallId?: string;
|
|
tool_call_id?: string;
|
|
}
|
|
|
|
function pairOrphanToolCalls(messages: AGUIMessage[]): AGUIMessage[] {
|
|
const out: AGUIMessage[] = [];
|
|
let pending: string[] = [];
|
|
|
|
const flush = () => {
|
|
for (const callId of pending) {
|
|
out.push({ id: `synth-${Math.random().toString(36).slice(2)}`, role: "tool", toolCallId: callId, content: "" });
|
|
}
|
|
pending = [];
|
|
};
|
|
|
|
for (const msg of messages) {
|
|
const role = String(msg?.role ?? "");
|
|
if (role === "tool") {
|
|
const callId = msg.toolCallId ?? msg.tool_call_id;
|
|
if (callId) pending = pending.filter((p) => p !== callId);
|
|
out.push(msg);
|
|
continue;
|
|
}
|
|
if (role === "assistant") {
|
|
flush();
|
|
const toolCalls = msg.toolCalls ?? msg.tool_calls ?? [];
|
|
out.push(msg);
|
|
for (const tc of toolCalls) {
|
|
if (tc?.id) pending.push(String(tc.id));
|
|
}
|
|
continue;
|
|
}
|
|
flush();
|
|
out.push(msg);
|
|
}
|
|
flush();
|
|
return out;
|
|
}
|
|
|
|
// ── DeduplicateToolCallMiddleware ─────────────────────────────────────────────
|
|
// MAF re-emits TOOL_CALL_START for declaration-only calls when they are invoked
|
|
// in parallel with other tools, causing verifyEvents to throw an AGUIError.
|
|
// This middleware tracks completed tool calls and silently drops any duplicate
|
|
// START/ARGS/END events for a call ID that has already been closed.
|
|
//
|
|
// NOTE: runNextWithState emits { event, messages, state } wrappers; always
|
|
// extract `.event` and forward the raw BaseEvent — never the wrapper.
|
|
|
|
// ── DebugLogMiddleware ──────────────────────────────────────────────────────
|
|
// Temporary: logs every event type flowing out to diagnose double-response bug.
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
class DebugLogMiddleware extends (Middleware as any) {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
run(input: any, next: any): Observable<any> {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
return new Observable<any>((observer) => {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const sub = (this as any).runNextWithState(input, next).subscribe({
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
next: (eventWithState: any) => {
|
|
const event = eventWithState.event;
|
|
const type: string = event?.type ?? "";
|
|
const extra = event?.toolCallName ? ` [${event.toolCallName}]`
|
|
: event?.activityType ? ` [${event.activityType}]`
|
|
: event?.messageId ? ` [msg:${event.messageId.slice(0, 8)}]`
|
|
: "";
|
|
console.log(`[DBG] ${type}${extra}`);
|
|
if (type === EventType.MESSAGES_SNAPSHOT) {
|
|
const msgs = (event as any)?.messages ?? [];
|
|
console.log(`[DBG] SNAPSHOT msgs: ${msgs.map((m: any) => `${m.role}:${(m.id ?? "").slice(0,8)}:${typeof m.content === "string" ? m.content.length : "?"}c`).join(" | ")}`);
|
|
}
|
|
observer.next(event);
|
|
},
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
error: (err: any) => observer.error(err),
|
|
complete: () => observer.complete(),
|
|
});
|
|
return () => sub.unsubscribe();
|
|
});
|
|
}
|
|
}
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
class DeduplicateToolCallMiddleware extends (Middleware as any) {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
run(input: any, next: any): Observable<any> {
|
|
const open = new Set<string>(); // tool call IDs currently in progress
|
|
const closed = new Set<string>(); // tool call IDs that already received END
|
|
|
|
// Render tools are display-once: after the client returns their result,
|
|
// the model sometimes re-issues the same call (fresh id) on the
|
|
// follow-up run, painting a second identical card. Track which render
|
|
// tools already ran in the current user turn (from the request's own
|
|
// message history) and suppress repeats — both their streamed events
|
|
// and their entries in this run's MESSAGES_SNAPSHOT.
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const inputMessages: any[] = Array.isArray(input?.messages) ? input.messages : [];
|
|
let lastUserIdx = -1;
|
|
inputMessages.forEach((m, i) => { if (m?.role === "user") lastUserIdx = i; });
|
|
const renderedThisTurn = new Set<string>();
|
|
for (const m of inputMessages.slice(lastUserIdx + 1)) {
|
|
const calls = m?.toolCalls ?? m?.tool_calls;
|
|
if (m?.role !== "assistant" || !Array.isArray(calls)) continue;
|
|
for (const tc of calls) {
|
|
const name = tc?.function?.name;
|
|
if (name && RENDER_TOOLS.has(name)) renderedThisTurn.add(name);
|
|
}
|
|
}
|
|
const suppressed = new Set<string>(); // toolCallIds of dropped repeats
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
return new Observable<any>((observer) => {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const sub = (this as any).runNextWithState(input, next).subscribe({
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
next: (eventWithState: any) => {
|
|
const event = eventWithState.event; // raw BaseEvent
|
|
const type: string = event?.type ?? "";
|
|
const id: string | undefined = event?.toolCallId;
|
|
|
|
if (type === EventType.TOOL_CALL_START) {
|
|
if (!id || closed.has(id) || open.has(id)) return; // duplicate
|
|
const name: string | undefined = event?.toolCallName;
|
|
if (name && RENDER_TOOLS.has(name)) {
|
|
if (renderedThisTurn.has(name)) {
|
|
suppressed.add(id);
|
|
return; // repeat render call — drop the whole call
|
|
}
|
|
renderedThisTurn.add(name);
|
|
}
|
|
open.add(id);
|
|
} else if (type === EventType.TOOL_CALL_ARGS || type === EventType.TOOL_CALL_END) {
|
|
if (id && suppressed.has(id)) return;
|
|
if (id && closed.has(id)) return; // already completed, drop
|
|
if (type === EventType.TOOL_CALL_END && id) {
|
|
open.delete(id);
|
|
closed.add(id);
|
|
}
|
|
} else if (type === EventType.MESSAGES_SNAPSHOT && suppressed.size > 0) {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const msgs: any[] = Array.isArray(event?.messages) ? event.messages : [];
|
|
const cleaned = msgs
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
.map((m: any) => {
|
|
const calls = m?.toolCalls ?? m?.tool_calls;
|
|
if (m?.role !== "assistant" || !Array.isArray(calls)) return m;
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const kept = calls.filter((tc: any) => !suppressed.has(String(tc?.id ?? "")));
|
|
if (kept.length === calls.length) return m;
|
|
const { toolCalls: _a, tool_calls: _b, ...rest } = m;
|
|
return kept.length > 0 ? { ...rest, toolCalls: kept } : rest;
|
|
})
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
.filter((m: any) => {
|
|
if (m?.role === "tool" && suppressed.has(String(m?.toolCallId ?? m?.tool_call_id ?? ""))) {
|
|
return false;
|
|
}
|
|
if (m?.role === "assistant") {
|
|
const calls = m?.toolCalls ?? m?.tool_calls;
|
|
const hasCalls = Array.isArray(calls) && calls.length > 0;
|
|
const hasText = typeof m?.content === "string" && m.content.length > 0;
|
|
if (!hasCalls && !hasText) return false;
|
|
}
|
|
return true;
|
|
});
|
|
observer.next({ ...event, messages: cleaned });
|
|
return;
|
|
}
|
|
|
|
observer.next(event); // emit raw BaseEvent
|
|
},
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
error: (err: any) => observer.error(err),
|
|
complete: () => observer.complete(),
|
|
});
|
|
return () => sub.unsubscribe();
|
|
});
|
|
}
|
|
}
|
|
|
|
// ── StripModelArtifactsMiddleware ────────────────────────────────────────────
|
|
// Some OpenAI models occasionally leak training-data special tokens such as
|
|
// `<|ipynb_marker|>` into completion text. Filter them out of streamed deltas
|
|
// before they reach the chat UI.
|
|
|
|
const ARTIFACT_RE = /<\|[^|>]+\|>/g;
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
class StripModelArtifactsMiddleware extends (Middleware as any) {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
run(input: any, next: any): Observable<any> {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
return new Observable<any>((observer) => {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const sub = (this as any).runNextWithState(input, next).subscribe({
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
next: (eventWithState: any) => {
|
|
const event = eventWithState.event;
|
|
if (
|
|
event?.type === EventType.TEXT_MESSAGE_CONTENT &&
|
|
typeof event?.delta === "string" &&
|
|
ARTIFACT_RE.test(event.delta)
|
|
) {
|
|
const cleaned = event.delta.replace(ARTIFACT_RE, "");
|
|
if (cleaned.length === 0) return;
|
|
observer.next({ ...event, delta: cleaned });
|
|
return;
|
|
}
|
|
observer.next(event);
|
|
},
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
error: (err: any) => observer.error(err),
|
|
complete: () => observer.complete(),
|
|
});
|
|
return () => sub.unsubscribe();
|
|
});
|
|
}
|
|
}
|
|
|
|
// ── DropRunMessagesSnapshotMiddleware ────────────────────────────────────────
|
|
// MAF's run-end MESSAGES_SNAPSHOT re-mints ids for post-tool-call text (their
|
|
// #3619) and, with thread persistence on, rebuilds history under STORE ids.
|
|
// ag-ui's apply treats the snapshot as authoritative: client messages absent
|
|
// from it are dropped, matching ids are replaced. Because streamed ids and
|
|
// snapshot ids never agree for tool+text turns, every mid-session snapshot
|
|
// either duplicates the current answer (fresh-id append) or wipes earlier
|
|
// ones (store-id replace with empty content) — both observed live. During a
|
|
// normal run the client already built the exact turn from streamed events,
|
|
// so the snapshot adds nothing: swallow it. Hydration replays (empty
|
|
// input.messages) pass through untouched — there the snapshot IS the
|
|
// conversation, and the client converges to store ids on every reload.
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
class DropRunMessagesSnapshotMiddleware extends (Middleware as any) {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
run(input: any, next: any): Observable<any> {
|
|
const isHydration =
|
|
!Array.isArray(input?.messages) || input.messages.length === 0;
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
return new Observable<any>((observer) => {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const sub = (this as any).runNextWithState(input, next).subscribe({
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
next: (eventWithState: any) => {
|
|
const event = eventWithState.event;
|
|
if (event?.type === EventType.MESSAGES_SNAPSHOT && !isHydration) {
|
|
return; // client state from streaming is already correct
|
|
}
|
|
observer.next(event);
|
|
},
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
error: (err: any) => observer.error(err),
|
|
complete: () => observer.complete(),
|
|
});
|
|
return () => sub.unsubscribe();
|
|
});
|
|
}
|
|
}
|
|
|
|
// ── SuppressRenderToolTextMiddleware ──────────────────────────────────────────
|
|
// When the LLM emits text content in the same response as a render tool call
|
|
// (render_a2ui or render_spending_summary), the text appears as a duplicate
|
|
// below the card. This middleware buffers TEXT_MESSAGE_* events and discards
|
|
// them if any render tool call is detected in the same turn.
|
|
|
|
const RENDER_TOOLS = new Set(["render_spending_summary"]);
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
class SuppressRenderToolTextMiddleware extends (Middleware as any) {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
run(input: any, next: any): Observable<any> {
|
|
let renderToolSeen = false;
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const textBuffer: any[] = [];
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
return new Observable<any>((observer) => {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const sub = (this as any).runNextWithState(input, next).subscribe({
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
next: (eventWithState: any) => {
|
|
const event = eventWithState.event; // raw BaseEvent
|
|
const type: string = event?.type ?? "";
|
|
|
|
if (type === EventType.TOOL_CALL_START) {
|
|
const toolName: string = event?.toolCallName ?? "";
|
|
if (RENDER_TOOLS.has(toolName)) {
|
|
renderToolSeen = true;
|
|
textBuffer.length = 0; // discard text buffered before we saw the tool call
|
|
}
|
|
}
|
|
|
|
if (
|
|
type === EventType.TEXT_MESSAGE_START ||
|
|
type === EventType.TEXT_MESSAGE_CONTENT ||
|
|
type === EventType.TEXT_MESSAGE_END
|
|
) {
|
|
if (!renderToolSeen) textBuffer.push(event); // buffer raw event
|
|
return; // always hold — flush at turn end
|
|
}
|
|
|
|
if (type === EventType.RUN_FINISHED || type === EventType.RUN_ERROR) {
|
|
if (!renderToolSeen) {
|
|
for (const e of textBuffer) observer.next(e); // flush raw events
|
|
}
|
|
textBuffer.length = 0;
|
|
observer.next(event); // emit raw event
|
|
return;
|
|
}
|
|
|
|
observer.next(event); // emit raw event
|
|
},
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
error: (err: any) => observer.error(err),
|
|
complete: () => {
|
|
if (!renderToolSeen) {
|
|
for (const e of textBuffer) observer.next(e);
|
|
}
|
|
observer.complete();
|
|
},
|
|
});
|
|
return () => sub.unsubscribe();
|
|
});
|
|
}
|
|
}
|
|
|
|
// ── Hono app ─────────────────────────────────────────────────────────────────
|
|
|
|
const app = new Hono();
|
|
|
|
// Security headers on every response. Responses returned by fetch() (the
|
|
// backend proxy) carry immutable headers, so rebuild the Response instead of
|
|
// mutating in place.
|
|
const SECURITY_HEADERS: Record<string, string> = {
|
|
"X-Frame-Options": "DENY",
|
|
"X-Content-Type-Options": "nosniff",
|
|
"Referrer-Policy": "strict-origin-when-cross-origin",
|
|
};
|
|
|
|
app.use("*", async (c, next) => {
|
|
await next();
|
|
const res = c.res;
|
|
const headers = new Headers(res.headers);
|
|
for (const [k, v] of Object.entries(SECURITY_HEADERS)) headers.set(k, v);
|
|
c.res = new Response(res.body, {
|
|
status: res.status,
|
|
statusText: res.statusText,
|
|
headers,
|
|
});
|
|
});
|
|
|
|
// ── 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.
|
|
|
|
// The v2 handler nags about anonymous telemetry unless disabled.
|
|
process.env.COPILOTKIT_TELEMETRY_DISABLED ??= "true";
|
|
|
|
function makeWealthysmartAgent(headers: Record<string, string> = {}): 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 */
|
|
if (process.env.CK_MIDDLEWARES !== "off") {
|
|
agent.use(new SuppressRenderToolTextMiddleware() as any);
|
|
agent.use(new StripModelArtifactsMiddleware() as any);
|
|
agent.use(new DropRunMessagesSnapshotMiddleware() as any);
|
|
agent.use(new DeduplicateToolCallMiddleware() as any);
|
|
}
|
|
if (process.env.CK_DEBUG_EVENTS === "1" || process.env.CK_MIDDLEWARES === "off") {
|
|
agent.use(new DebugLogMiddleware() as any);
|
|
}
|
|
/* eslint-enable @typescript-eslint/no-explicit-any */
|
|
return agent;
|
|
}
|
|
|
|
// ── 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).
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const liveRuns = new Map<string, any>();
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
function proxyAgentRun(agent: any, input: any, threadId: string): Observable<any> {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const subject = new ReplaySubject<any>(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<any> {
|
|
return proxyAgentRun(request.agent, request.input, request.threadId);
|
|
}
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
connect(request: any): Observable<any> {
|
|
// 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<boolean> {
|
|
return Promise.resolve(liveRuns.has(request.threadId));
|
|
}
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
async stop(request: any): Promise<boolean | undefined> {
|
|
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,
|
|
];
|
|
}
|
|
|
|
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,
|
|
// but in prod the browser talks to this Hono server, which must forward
|
|
// `/api/v1/*` and `/api/auth/*` to the FastAPI container — otherwise the SPA
|
|
// fallback below swallows them and returns index.html.
|
|
const proxyToBackend = async (c: import("hono").Context) => {
|
|
const url = new URL(c.req.url);
|
|
const target = `${BACKEND_URL}${url.pathname}${url.search}`;
|
|
const method = c.req.method;
|
|
const headers = new Headers(c.req.raw.headers);
|
|
headers.delete("host");
|
|
const init: RequestInit = {
|
|
method,
|
|
headers,
|
|
redirect: "manual",
|
|
};
|
|
if (method !== "GET" && method !== "HEAD") {
|
|
init.body = c.req.raw.body;
|
|
// @ts-expect-error undici requires duplex for streamed bodies
|
|
init.duplex = "half";
|
|
}
|
|
try {
|
|
return await fetch(target, init);
|
|
} catch (err) {
|
|
console.error(`[proxy] ${method} ${url.pathname} → backend failed:`, err);
|
|
return c.json({ error: "Backend unavailable" }, 502);
|
|
}
|
|
};
|
|
|
|
app.all("/api/v1/*", proxyToBackend);
|
|
app.all("/api/auth/*", proxyToBackend);
|
|
|
|
// In production, serve the Vite build output.
|
|
if (isProd) {
|
|
app.use("/*", serveStatic({ root: "./dist" }));
|
|
// SPA fallback: any non-asset path returns index.html
|
|
app.get("/*", serveStatic({ path: "./dist/index.html" }));
|
|
}
|
|
|
|
serve({ fetch: app.fetch, port: PORT }, (info) => {
|
|
console.log(`CopilotKit server on port ${info.port} [${isProd ? "prod" : "dev"}]`);
|
|
});
|