diff --git a/mcp-server/dashboard.ts b/mcp-server/dashboard.ts index f9f6eca..2cc106c 100644 --- a/mcp-server/dashboard.ts +++ b/mcp-server/dashboard.ts @@ -1,4 +1,5 @@ -const GW = "http://localhost:3700"; +// Use the same host the browser loaded the page from — works on LAN + localhost +const GW = window.location.origin; let simData: any = null; let currentDay = 0; diff --git a/mcp-server/index.ts b/mcp-server/index.ts index c596414..3e6b696 100644 --- a/mcp-server/index.ts +++ b/mcp-server/index.ts @@ -17,19 +17,48 @@ import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js"; import { z } from "zod"; +import { startTrace, logSpan, logGeneration, scoreTrace, flush as flushTraces } from "./tracing.js"; const BASE = process.env.LAKEHOUSE_URL || "http://localhost:3100"; const PORT = parseInt(process.env.MCP_PORT || "3700"); const MODE = process.env.MCP_TRANSPORT || "http"; // "stdio" or "http" +// Active trace for the current request — set per-request in the HTTP handler +let activeTrace: ReturnType | null = null; + async function api(method: string, path: string, body?: any) { + const t0 = Date.now(); const resp = await fetch(`${BASE}${path}`, { method, headers: body ? { "Content-Type": "application/json" } : {}, body: body ? JSON.stringify(body) : undefined, }); const text = await resp.text(); - try { return JSON.parse(text); } catch { return { raw: text, status: resp.status }; } + const ms = Date.now() - t0; + let parsed: any; + try { parsed = JSON.parse(text); } catch { parsed = { raw: text, status: resp.status }; } + + // Trace the call if we have an active trace + if (activeTrace) { + const isGen = path.includes("/generate"); + if (isGen) { + logGeneration(activeTrace, `lakehouse${path}`, { + model: body?.model || "unknown", + prompt: typeof body?.prompt === "string" ? body.prompt.slice(0, 500) : JSON.stringify(body).slice(0, 300), + completion: typeof parsed?.text === "string" ? parsed.text.slice(0, 500) : JSON.stringify(parsed).slice(0, 300), + duration_ms: ms, + tokens_in: parsed?.prompt_eval_count, + tokens_out: parsed?.eval_count, + }); + } else { + logSpan(activeTrace, `lakehouse${path}`, body, { + rows: parsed?.row_count, sources: parsed?.sources?.length, + sql_matches: parsed?.sql_matches, method: parsed?.method, + }, ms); + } + } + + return parsed; } const server = new McpServer({ name: "lakehouse", version: "1.0.0" }); @@ -267,13 +296,27 @@ async function main() { async fetch(req) { const url = new URL(req.url); const json = async () => req.method === "POST" ? await req.json() : {}; - const ok = (data: any) => Response.json(data); - const err = (msg: string, status = 400) => Response.json({ error: msg }, { status }); + + // CORS — dashboard runs in the browser, gateway is a different origin + const cors = { + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "GET, POST, OPTIONS", + "Access-Control-Allow-Headers": "Content-Type", + }; + if (req.method === "OPTIONS") return new Response(null, { status: 204, headers: cors }); + + const ok = (data: any) => Response.json(data, { headers: cors }); + const err = (msg: string, status = 400) => Response.json({ error: msg }, { status, headers: cors }); try { - // Health + // Health — no trace needed if (url.pathname === "/health") return ok({ status: "ok", lakehouse: BASE, tools: 11 }); + // Start a Langfuse trace for every non-static request + if (req.method === "POST" || !["/","/dashboard","/dashboard.css","/dashboard.ts","/dashboard.js"].includes(url.pathname)) { + activeTrace = startTrace(`gw:${url.pathname}`, { method: req.method, path: url.pathname }); + } + // Self-orientation: any agent calls this first to understand the system if (url.pathname === "/context") { const instructions = await Bun.file("/home/profit/lakehouse/mcp-server/AGENT_INSTRUCTIONS.md").text().catch(() => ""); @@ -431,9 +474,16 @@ async function main() { return ok(await runWeekSimulation()); } + activeTrace = null; return err("Unknown path. Available: / /health /search /sql /match /worker/:id /ask /log /playbooks /profile/:id /vram /context /verify /simulation/run", 404); } catch (e: any) { + if (activeTrace) { scoreTrace(activeTrace, "error", 0, e.message); } + activeTrace = null; return err(e.message || String(e), 500); + } finally { + // Flush traces async — don't block the response + flushTraces().catch(() => {}); + activeTrace = null; } }, });