From 36c4e8099bd17df6b1f19a179850eb58a77ba5e2 Mon Sep 17 00:00:00 2001 From: Emi <74990983+canoo@users.noreply.github.com> Date: Sun, 28 Jun 2026 02:38:51 -0600 Subject: [PATCH 1/2] feat(observability): log route metadata --- tools/mcp/server.mjs | 22 ++++++++++++++++++++++ tools/tui/main.go | 21 ++++++++++++++++++++- tools/tui/main_test.go | 27 +++++++++++++++++++++------ 3 files changed, 63 insertions(+), 7 deletions(-) diff --git a/tools/mcp/server.mjs b/tools/mcp/server.mjs index 0457967..dc4b4bc 100644 --- a/tools/mcp/server.mjs +++ b/tools/mcp/server.mjs @@ -47,16 +47,38 @@ function estimateCloudCost(tokensIn, tokensOut) { (tokensOut / 1_000_000) * CLOUD_OUTPUT_USD_PER_1M; } +function taskTypeForTool(tool) { + return { + ollama_commit_msg: "commit-msg", + ollama_boilerplate: "boilerplate", + ollama_test_scaffold: "test-scaffold", + ollama_lint_fix: "lint-fix", + ollama_logic_refactor: "logic-refactor", + }[tool] || "unknown"; +} + +function routeBandForTask(taskType, model) { + if (model === "fast-path") return "fast-path"; + if (["commit-msg", "boilerplate", "test-scaffold"].includes(taskType)) return "supervisor"; + if (["lint-fix", "logic-refactor"].includes(taskType)) return "logic"; + return "unknown"; +} + function taskLogEntry({ tool, model, ms, ok, prompt = "", response = "", error }) { + const taskType = taskTypeForTool(tool); const tokensIn = estimateTokens(prompt); const tokensOut = estimateTokens(response); const routing = model === "fast-path" ? "deterministic" : "local"; const entry = { tool, + task_type: taskType, model, + model_provider: model === "fast-path" ? "fast-path" : "ollama", + route_band: routeBandForTask(taskType, model), routing, tokens_in: tokensIn, tokens_out: tokensOut, + cost_usd: 0, cloud_cost_equivalent: estimateCloudCost(tokensIn, tokensOut), ms, ok, diff --git a/tools/tui/main.go b/tools/tui/main.go index f13ac0a..e25c1ac 100644 --- a/tools/tui/main.go +++ b/tools/tui/main.go @@ -84,10 +84,14 @@ type healthMsg struct { type taskLogEntry struct { Tool string `json:"tool"` + TaskType string `json:"task_type,omitempty"` Model string `json:"model"` + ModelProvider string `json:"model_provider,omitempty"` + RouteBand string `json:"route_band,omitempty"` Routing string `json:"routing,omitempty"` TokensIn int `json:"tokens_in,omitempty"` TokensOut int `json:"tokens_out,omitempty"` + CostUSD float64 `json:"cost_usd,omitempty"` CloudCostEquivalent float64 `json:"cloud_cost_equivalent,omitempty"` Ms int `json:"ms"` Ok bool `json:"ok"` @@ -107,6 +111,9 @@ type taskLogStats struct { p95Ms int modelTasks map[string]int routes map[string]int + routeBands map[string]int + costUSD float64 + cloudUSD float64 savingsUSD float64 } @@ -881,6 +888,10 @@ func taskLogView(m model) string { s += fmt.Sprintf(" Est. local savings: $%.4f\n", stats.savingsUSD) } s += fmt.Sprintf(" Routes: %s\n", summarizeIntCounts(stats.routes, 3)) + s += fmt.Sprintf(" Route bands: %s\n", summarizeIntCounts(stats.routeBands, 3)) + if stats.cloudUSD > 0 { + s += fmt.Sprintf(" Cloud equivalent: $%.4f Local cost: $%.4f\n", stats.cloudUSD, stats.costUSD) + } s += fmt.Sprintf(" Models: %s\n\n", summarizeModelUsage(stats.modelTasks, 3)) // Header @@ -909,6 +920,7 @@ func summarizeTaskLog(entries []taskLogEntry) taskLogStats { stats := taskLogStats{ modelTasks: map[string]int{}, routes: map[string]int{}, + routeBands: map[string]int{}, } if len(entries) == 0 { return stats @@ -926,8 +938,10 @@ func summarizeTaskLog(entries []taskLogEntry) taskLogStats { stats.failures++ } if e.Routing == "local" || e.Routing == "deterministic" { - stats.savingsUSD += e.CloudCostEquivalent + stats.savingsUSD += e.CloudCostEquivalent - e.CostUSD } + stats.costUSD += e.CostUSD + stats.cloudUSD += e.CloudCostEquivalent model := strings.TrimSpace(e.Model) if model == "" { model = "unknown" @@ -938,6 +952,11 @@ func summarizeTaskLog(entries []taskLogEntry) taskLogStats { route = "unknown" } stats.routes[route]++ + routeBand := strings.TrimSpace(e.RouteBand) + if routeBand == "" { + routeBand = "unknown" + } + stats.routeBands[routeBand]++ } stats.avgMs = totalMs / stats.total stats.p95Ms = percentile95(latencies) diff --git a/tools/tui/main_test.go b/tools/tui/main_test.go index bb0afd5..b9e5fd7 100644 --- a/tools/tui/main_test.go +++ b/tools/tui/main_test.go @@ -298,10 +298,10 @@ func TestViewsDoNotPanic(t *testing.T) { func TestSummarizeTaskLog(t *testing.T) { entries := []taskLogEntry{ - {Tool: "ollama_commit_msg", Model: "qwen2.5-coder:1.5b", Routing: "local", CloudCostEquivalent: 0.001, Ms: 10, Ok: true}, - {Tool: "ollama_boilerplate", Model: "qwen2.5-coder:1.5b", Routing: "local", CloudCostEquivalent: 0.002, Ms: 20, Ok: true}, - {Tool: "ollama_commit_msg", Model: "fast-path", Routing: "deterministic", CloudCostEquivalent: 0.001, Ms: 0, Ok: true}, - {Tool: "ollama_lint_fix", Model: "llama3.2:3b", Routing: "local", CloudCostEquivalent: 0.003, Ms: 30, Ok: false}, + {Tool: "ollama_commit_msg", Model: "qwen2.5-coder:1.5b", RouteBand: "supervisor", Routing: "local", CostUSD: 0, CloudCostEquivalent: 0.001, Ms: 10, Ok: true}, + {Tool: "ollama_boilerplate", Model: "qwen2.5-coder:1.5b", RouteBand: "supervisor", Routing: "local", CostUSD: 0, CloudCostEquivalent: 0.002, Ms: 20, Ok: true}, + {Tool: "ollama_commit_msg", Model: "fast-path", RouteBand: "fast-path", Routing: "deterministic", CostUSD: 0, CloudCostEquivalent: 0.001, Ms: 0, Ok: true}, + {Tool: "ollama_lint_fix", Model: "llama3.2:3b", RouteBand: "logic", Routing: "local", CostUSD: 0.001, CloudCostEquivalent: 0.003, Ms: 30, Ok: false}, } stats := summarizeTaskLog(entries) @@ -320,8 +320,14 @@ func TestSummarizeTaskLog(t *testing.T) { if stats.p95Ms != 30 { t.Errorf("p95Ms: got %d, want 30", stats.p95Ms) } - if math.Abs(stats.savingsUSD-0.007) > 0.000001 { - t.Errorf("savingsUSD: got %f, want 0.007", stats.savingsUSD) + if math.Abs(stats.savingsUSD-0.006) > 0.000001 { + t.Errorf("savingsUSD: got %f, want 0.006", stats.savingsUSD) + } + if math.Abs(stats.cloudUSD-0.007) > 0.000001 { + t.Errorf("cloudUSD: got %f, want 0.007", stats.cloudUSD) + } + if math.Abs(stats.costUSD-0.001) > 0.000001 { + t.Errorf("costUSD: got %f, want 0.001", stats.costUSD) } if stats.modelTasks["qwen2.5-coder:1.5b"] != 2 { t.Errorf("qwen count: got %d, want 2", stats.modelTasks["qwen2.5-coder:1.5b"]) @@ -335,6 +341,15 @@ func TestSummarizeTaskLog(t *testing.T) { if stats.routes["deterministic"] != 1 { t.Errorf("deterministic route count: got %d, want 1", stats.routes["deterministic"]) } + if stats.routeBands["supervisor"] != 2 { + t.Errorf("supervisor route band count: got %d, want 2", stats.routeBands["supervisor"]) + } + if stats.routeBands["fast-path"] != 1 { + t.Errorf("fast-path route band count: got %d, want 1", stats.routeBands["fast-path"]) + } + if stats.routeBands["logic"] != 1 { + t.Errorf("logic route band count: got %d, want 1", stats.routeBands["logic"]) + } } func TestSummarizeIntCounts(t *testing.T) { From b3df22c921320e6bbc1289444396f455f0475897 Mon Sep 17 00:00:00 2001 From: Emi <74990983+canoo@users.noreply.github.com> Date: Sun, 28 Jun 2026 16:00:42 -0600 Subject: [PATCH 2/2] feat(observability): write MCP tasks to sqlite --- docs/observability-schema.md | 10 ++- tools/mcp/server.mjs | 166 ++++++++++++++++++++++++++++++++++- 2 files changed, 170 insertions(+), 6 deletions(-) diff --git a/docs/observability-schema.md b/docs/observability-schema.md index bffc3fc..919d7c8 100644 --- a/docs/observability-schema.md +++ b/docs/observability-schema.md @@ -30,8 +30,10 @@ The existing MCP JSONL log remains a compatibility source during migration: ~/.config/nexus/logs/mcp-tasks.jsonl ``` -Writers should prefer SQLite once implemented. Readers may fall back to JSONL -until the migration is complete. +The MCP writer prefers SQLite when the Node runtime exposes `node:sqlite`. +Readers may fall back to JSONL until the migration is complete. JSONL writes +continue during the transition so older TUI builds can still display task +history. ## Entity Model @@ -220,9 +222,9 @@ LIMIT 20; ## Migration Plan -1. Add a small SQLite writer used by the MCP server for new task rows. +1. Add a small SQLite writer used by the MCP server for new task rows. [DONE] 2. Keep JSONL writes temporarily so older TUI builds can still display task - history. + history. [DONE] 3. Add a one-time importer that reads `mcp-tasks.jsonl` and writes missing rows into SQLite using deterministic task IDs. 4. Update the TUI Task Log and dashboard screens to prefer SQLite and fall back diff --git a/tools/mcp/server.mjs b/tools/mcp/server.mjs index dc4b4bc..7f4cfb3 100644 --- a/tools/mcp/server.mjs +++ b/tools/mcp/server.mjs @@ -12,6 +12,7 @@ import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; import { z } from "zod"; import { appendFileSync, mkdirSync } from "node:fs"; +import { createHash, randomUUID } from "node:crypto"; import { homedir } from "node:os"; import { join } from "node:path"; @@ -20,15 +21,170 @@ const CONNECT_TIMEOUT_MS = 5000; const REQUEST_TIMEOUT_MS = 120000; // ── Task log ──────────────────────────────────────────────────────────────── -// Writes JSONL entries to ~/.config/nexus/logs/mcp-tasks.jsonl for TUI display. +// Writes SQLite entries to ~/.config/nexus/logs/observability.sqlite when the +// runtime supports node:sqlite. JSONL remains a compatibility source for older +// runtimes and older TUI builds. const LOG_DIR = join(homedir(), ".config", "nexus", "logs"); const LOG_FILE = join(LOG_DIR, "mcp-tasks.jsonl"); +const DB_FILE = join(LOG_DIR, "observability.sqlite"); try { mkdirSync(LOG_DIR, { recursive: true }); } catch {} +let sqlite = null; +try { + const { DatabaseSync } = await import("node:sqlite"); + sqlite = new DatabaseSync(DB_FILE); + sqlite.exec(` +CREATE TABLE IF NOT EXISTS schema_migrations ( + version INTEGER PRIMARY KEY, + applied_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS sessions ( + id TEXT PRIMARY KEY, + start_time TEXT NOT NULL, + end_time TEXT, + status TEXT, + cli_tool TEXT, + persona TEXT, + nexus_mode TEXT, + project_path_hash TEXT, + nexus_version TEXT, + metadata_json TEXT +); + +CREATE TABLE IF NOT EXISTS tasks ( + id TEXT PRIMARY KEY, + session_id TEXT NOT NULL REFERENCES sessions(id), + parent_task_id TEXT REFERENCES tasks(id), + timestamp TEXT NOT NULL, + source TEXT NOT NULL, + tool TEXT, + task_type TEXT, + model TEXT NOT NULL, + model_provider TEXT, + route_band TEXT, + routing TEXT NOT NULL, + routing_reason TEXT, + trace_id TEXT, + span_id TEXT, + parent_span_id TEXT, + client_request_id TEXT, + upstream_session_id TEXT, + upstream_tool TEXT, + idempotency_key TEXT, + input_bytes INTEGER, + output_bytes INTEGER, + input_hash TEXT, + tokens_in INTEGER, + tokens_out INTEGER, + total_tokens INTEGER, + latency_ms INTEGER, + cost_usd REAL, + cloud_cost_equivalent REAL, + quality_rating INTEGER, + ok INTEGER NOT NULL DEFAULT 1, + error TEXT +); + +CREATE TABLE IF NOT EXISTS routing_decisions ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES tasks(id), + decided_at TEXT NOT NULL, + reason TEXT NOT NULL, + alternatives_considered TEXT, + classifier_version TEXT, + fallback_from TEXT, + fallback_to TEXT, + circuit_breaker_triggered INTEGER NOT NULL DEFAULT 0, + latency_budget_ms INTEGER +); + +CREATE INDEX IF NOT EXISTS idx_sessions_start_time + ON sessions(start_time); +CREATE INDEX IF NOT EXISTS idx_tasks_session_id + ON tasks(session_id); +CREATE INDEX IF NOT EXISTS idx_tasks_timestamp + ON tasks(timestamp); +CREATE INDEX IF NOT EXISTS idx_tasks_model + ON tasks(model); +CREATE INDEX IF NOT EXISTS idx_tasks_status + ON tasks(ok); +CREATE INDEX IF NOT EXISTS idx_tasks_routing + ON tasks(routing); +CREATE INDEX IF NOT EXISTS idx_routing_decisions_task_id + ON routing_decisions(task_id); +INSERT OR IGNORE INTO schema_migrations (version, applied_at) +VALUES (1, datetime('now')); +`); +} catch {} + function logTask(entry) { try { appendFileSync(LOG_FILE, JSON.stringify(entry) + "\n"); } catch {} + + if (!sqlite) return; + try { + writeSqliteTask(entry); + } catch {} +} + +function writeSqliteTask(entry) { + const sessionId = `mcp-${entry.timestamp.slice(0, 10)}`; + sqlite.prepare(` +INSERT OR IGNORE INTO sessions + (id, start_time, status, cli_tool, nexus_mode, metadata_json) +VALUES + (?, ?, 'active', 'mcp', 'hybrid', ?) +`).run(sessionId, `${entry.timestamp.slice(0, 10)}T00:00:00.000Z`, JSON.stringify({ source: "nexus-ollama" })); + + sqlite.prepare(` +INSERT INTO tasks + (id, session_id, timestamp, source, tool, task_type, model, model_provider, + route_band, routing, routing_reason, input_bytes, output_bytes, input_hash, + tokens_in, tokens_out, total_tokens, latency_ms, cost_usd, + cloud_cost_equivalent, ok, error) +VALUES + (?, ?, ?, 'mcp-tool', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +`).run( + entry.id, + sessionId, + entry.timestamp, + entry.tool, + entry.task_type, + entry.model, + entry.model_provider, + entry.route_band, + entry.routing, + `mcp:${entry.tool}`, + entry.input_bytes, + entry.output_bytes, + entry.input_hash, + entry.tokens_in, + entry.tokens_out, + entry.tokens_in + entry.tokens_out, + entry.ms, + entry.cost_usd, + entry.cloud_cost_equivalent, + entry.ok ? 1 : 0, + entry.error || null, + ); + + sqlite.prepare(` +INSERT INTO routing_decisions + (id, task_id, decided_at, reason, alternatives_considered, + classifier_version, circuit_breaker_triggered, latency_budget_ms) +VALUES + (?, ?, ?, ?, ?, 'rules-v1', ?, ?) +`).run( + randomUUID(), + entry.id, + entry.timestamp, + `Selected ${entry.route_band} route for ${entry.task_type}`, + JSON.stringify([]), + entry.ok ? 0 : 1, + REQUEST_TIMEOUT_MS, + ); } // Early cost tracking uses a conservative cloud-equivalent estimate. Local @@ -69,7 +225,9 @@ function taskLogEntry({ tool, model, ms, ok, prompt = "", response = "", error } const tokensIn = estimateTokens(prompt); const tokensOut = estimateTokens(response); const routing = model === "fast-path" ? "deterministic" : "local"; + const timestamp = new Date(); const entry = { + id: randomUUID(), tool, task_type: taskType, model, @@ -78,11 +236,15 @@ function taskLogEntry({ tool, model, ms, ok, prompt = "", response = "", error } routing, tokens_in: tokensIn, tokens_out: tokensOut, + input_bytes: Buffer.byteLength(prompt, "utf8"), + output_bytes: Buffer.byteLength(response, "utf8"), + input_hash: prompt ? createHash("sha256").update(prompt).digest("hex") : "", cost_usd: 0, cloud_cost_equivalent: estimateCloudCost(tokensIn, tokensOut), ms, ok, - ts: Date.now(), + ts: timestamp.getTime(), + timestamp: timestamp.toISOString(), }; if (error) entry.error = error; return entry;