Skip to content

feat(observability): write MCP task metrics #97

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
canoo wants to merge 2 commits into
base: main
Choose a base branch
from
Open

feat(observability): write MCP task metrics #97

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
36c4e80
feat(observability): log route metadata
canoo Jun 28, 2026
b3df22c
feat(observability): write MCP tasks to sqlite
canoo Jun 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 6 additions & 4 deletions docs/observability-schema.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand Down
188 changes: 186 additions & 2 deletions tools/mcp/server.mjs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";

Expand All @@ -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
Expand All @@ -47,20 +203,48 @@ 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 timestamp = new Date();
const entry = {
id: randomUUID(),
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,
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;
Expand Down
21 changes: 20 additions & 1 deletion tools/tui/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"`
Expand All @@ -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
}

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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"
Expand All @@ -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)
Expand Down
27 changes: 21 additions & 6 deletions tools/tui/main_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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"])
Expand All @@ -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) {
Expand Down
Loading