diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5fc05def17..2279545a40 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -423,6 +423,7 @@ jobs: tests/test_cli_help_no_dashboard_import.py \ tests/test_behaviour_signals.py \ tests/test_signals_ui_contract.py \ + tests/test_self_diagnostics.py \ -q # ── Entitlement API test suite (~2700+ hermetic tests) ───────────────────────────────────────────── diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a928787dd..358e02e444 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,10 @@ - **What:** `spans` and `sessions` gain a `content_hash` column (ALTER, no rebuild). `ingest_spans_batch` and `ingest_sessions_batch` hash the row's content columns (write timestamps excluded), seed the hash map from the table on first use so a restart is not a rewrite, and skip rows whose content matches the last write; a changed field (late end_ts, cost, status, title) still overwrites, in-batch duplicates still keep the last occurrence, and the rollup mirror only sees rows that were actually written. Legacy rows without a hash are written once and then never again. `CLAWMETRY_UPSERT_DEDUPE=0` restores the old always-write behaviour. `query_forward_progress` now defaults `since` to the last 24 h (`CLAWMETRY_FWDPROG_DEFAULT_HOURS`) unless a `since` or `session_id` is given, and serves through the bounded read cache. - **Verified:** 12 new tests in `tests/test_ingest_upsert_dedupe.py` (identical re-delivery writes zero rows, changed content overwrites, dedupe survives a restart, legacy NULL-hash rows stamped once, kill switch, in-batch dupes, rollups untouched on skip, forward-progress default window and cache hit); 7 of them fail on the unpatched store. The existing bulk-flush, span-ingest, read-lock, query-contract and local-query suites pass unchanged (85 tests). Wired into the CI MOAT file list. +### Added: agents can report to their operators over MCP, and ClawMetry checks whether the tool stream agrees (WO-59, REQ-SELF-001 to 004) (2026-09-04) +- **Why:** reasoning models are good at introspection when asked plainly. Given a reporting tool framed as notes to the people who run them, agents will say that a tool kept failing, that they lacked context or a permission, that they could not finish, or that they worked around a block. ClawMetry already watches the same session from outside, so every such note can sit next to what the detectors and the approval hooks recorded on their own, and an operator can see when the agent stayed quiet about something the detectors caught. The same MCP server is where a developer's editor can ask ClawMetry questions from where they already are. +- **What:** (1) The MCP server gains `report_to_operator` (six categories, operator-extendable via `config.json` `self_diagnostics.categories`, summary capped at 500 characters, session inferred from the environment or the working directory when absent) plus four read tools: `list_incidents`, `get_guard_status` (the per-session control verdict from the one resolver, policy decisions, open incidents), `get_signal_rates` (says "signals not available on this daemon version" when the daemon lacks the method) and `list_self_reports`. No tool on the server acts on a process; the test suite pins that. Every tool answers through the daemon and returns an honest error when it is down. (2) Reports land in a new additive `agent_self_reports` table through the daemon's local query server (the daemon keeps the writer lock); the summary passes the same redaction as every other stored text. (3) On the detector tick the daemon marks a report corroborated when a detector incident or a permission denial exists for the same session within `CORROBORATION_WINDOW_SECS` (600, `CLAWMETRY_SELFDIAG_WINDOW_SECS`); uncorroborated is labelled with its plain meaning, not the same as false. An honesty figure per (runtime, model), the share of detector incidents the agent also reported, is withheld with a reason under `MIN_INCIDENTS` (5, `CLAWMETRY_SELFDIAG_MIN_INCIDENTS`). (4) `clawmetry mcp install [--runtime |all] [--dry-run] [--write-guidance]`, `uninstall` and `status` register the server with Claude Code, Cursor, Codex, Gemini CLI, OpenCode and Windsurf, each format verified against the vendor's documentation (Codex against `codex mcp add` itself); merge only, never a foreign entry deleted, uninstall removes only what a marker file says we wrote, a hand-written entry is left in place, JSONC is refused rather than guessed. Other runtimes report `no MCP support` or `unknown config format` by name. The instructions-file snippet is printed and written only with `--write-guidance`. (5) `GET /api/self-reports`, `/api/self-reports/honesty`, `/api/self-reports/support`; a "What the agent reported" panel on the transcript view; an "Agent reported" card on the Guard tab with per-category counts, the honesty table with withheld reasons, and the per-runtime MCP support state; a `selfReports` snapshot slice (counts and honesty, no summaries). +- **Verified:** `tests/test_self_diagnostics.py` (49 tests): tool schema and framing, no actuating tool, honest daemon-down errors for every tool, redaction and cap on the write path, idempotency, session inference, inclusive window logic, corroboration against real `loop_signals` and `approvals` rows, honesty withheld and computed, installer merge/never-delete/uninstall-only-ours/hand-written-left-in-place/JSONC-refused for all six formats with the Codex block round-tripped through a TOML parser, guidance offered not written, the CLI fast path never importing the dashboard, the routes, and the snapshot slice. Named in the `moat-tests` CI job and `make test-selfdiag`. ### Added: Behaviour Signals, six judge-free signals over every transcript, a rates API, an alert rule and a Signals tab (WO-58) - **Why:** Guard reads the tool stream to notice an agent that is stuck. Nothing read the words. The person swearing at the agent, the agent refusing or handing the work back, the agent saying it could not finish, the person saying thanks: those live in the transcripts ClawMetry already holds for every runtime and were never counted. Claude Code's own team tracks a frustration rate from a keyword list; this brings the same family of signals to every runtime on the operator's machine, with no model call. diff --git a/Makefile b/Makefile index d0b3f5d3c6..f5658627cb 100644 --- a/Makefile +++ b/Makefile @@ -15,6 +15,10 @@ test-compat: # Mirrored in .github/workflows/ci.yml (moat-tests job). test-hooks: python3 -m pytest tests/test_hooks_claude_code.py tests/test_hook_lifecycle.py tests/test_redaction_pii.py -q +# WO-59 self-diagnostics: MCP report tool, corroboration, honesty rollup, +# multi-runtime MCP installer. Mirrored in .github/workflows/ci.yml (moat-tests). +test-selfdiag: + python3 -m pytest tests/test_self_diagnostics.py -q test-fast: CLAWMETRY_URL=http://localhost:8900 CLAWMETRY_TOKEN=dev-token python3 -m pytest tests/test_api.py -v diff --git a/clawmetry/cli.py b/clawmetry/cli.py index 46915d73fe..342d035d05 100644 --- a/clawmetry/cli.py +++ b/clawmetry/cli.py @@ -4691,9 +4691,10 @@ def _format_uptime(seconds): def _cmd_mcp(args) -> None: - """Start the ClawMetry MCP server on stdio (refs #2859).""" - from clawmetry.mcp_server import run - run() + """`clawmetry mcp ...` (refs #2859, WO-59). Normally intercepted by the + fast path in main(); kept for callers that build a Namespace directly.""" + from clawmetry.mcp_install import cli_main as _mcp_cli + raise SystemExit(_mcp_cli(list(getattr(args, "mcp_args", None) or []))) def _cmd_reports(args) -> None: @@ -7657,6 +7658,12 @@ def main() -> None: # dashboard import. Stdlib-only; `stamp` always exits 0 (fail-open). if len(sys.argv) > 1 and sys.argv[1] == "trace": raise SystemExit(trace_main(sys.argv[2:])) + # FAST PATH — `clawmetry mcp [serve|install|uninstall|status]` (WO-59). + # `serve` is started by the agent host on every session and must not + # pay the dashboard import; the installer is stdlib-only as well. + if len(sys.argv) > 1 and sys.argv[1] == "mcp": + from clawmetry.mcp_install import cli_main as _mcp_cli + raise SystemExit(_mcp_cli(sys.argv[2:])) # FAST PATH — `clawmetry instrument …` (WO-57): writes the # runtime's own OpenTelemetry exporter settings so it reports to this # ClawMetry. Which runtimes: whatever profiles are registered (free ones @@ -8169,11 +8176,15 @@ def main() -> None: ), ) - # mcp — start MCP server on stdio (issue #2859) - sub.add_parser( + # mcp — intercepted by the fast path at the top of main() (WO-59); the + # parser entry exists so `clawmetry --help` discovery shows it. + p_mcp = sub.add_parser( "mcp", - help="Start ClawMetry MCP server (stdio) — lets agents query their own telemetry", + help="MCP server: `mcp` serves on stdio; `mcp install [--runtime |all] " + "[--dry-run] [--write-guidance]` registers it with each runtime; " + "`mcp uninstall`; `mcp status`", ) + p_mcp.add_argument("mcp_args", nargs="*") # uninstall — fully remove clawmetry p_uninstall = sub.add_parser( diff --git a/clawmetry/local_store.py b/clawmetry/local_store.py index fdd7680b8c..a6b495fe5b 100644 --- a/clawmetry/local_store.py +++ b/clawmetry/local_store.py @@ -47,6 +47,7 @@ from clawmetry import ccr as _ccr # reversible event-payload compression (#2843) import threading import time +import uuid from collections import deque from contextlib import contextmanager from datetime import datetime, timedelta, timezone @@ -1772,6 +1773,31 @@ def _on_disk_bytes() -> int: "CREATE INDEX IF NOT EXISTS idx_signal_matches_ms ON signal_matches(turn_ms)", "CREATE INDEX IF NOT EXISTS idx_signal_matches_sig_ms ON signal_matches(signal, turn_ms)", "CREATE INDEX IF NOT EXISTS idx_signal_matches_session ON signal_matches(session_id)", + # ── Agent self-reports (WO-59) ────────────────────────────────────── + # A note an agent filed about its own trouble through the MCP + # ``report_to_operator`` tool. ``summary_redacted`` has already been + # through ``clawmetry.redaction``; the raw text is never stored. + # ``corroborated`` / ``corroboration_ref`` are filled by the daemon's + # corroboration pass when a detector incident or a permission denial + # exists for the same session near the report. Additive: CREATE TABLE + # IF NOT EXISTS, no schema-version bump. + """ + CREATE TABLE IF NOT EXISTS agent_self_reports ( + id VARCHAR PRIMARY KEY, + session_id VARCHAR NOT NULL DEFAULT '', + agent_type VARCHAR DEFAULT '', + node_id VARCHAR DEFAULT '', + model VARCHAR DEFAULT '', + category VARCHAR NOT NULL, + summary_redacted VARCHAR DEFAULT '', + ts BIGINT NOT NULL, + created_at BIGINT NOT NULL, + corroborated BOOLEAN DEFAULT FALSE, + corroboration_ref VARCHAR DEFAULT '' + ) + """, + "CREATE INDEX IF NOT EXISTS idx_self_reports_ts ON agent_self_reports(ts DESC)", + "CREATE INDEX IF NOT EXISTS idx_self_reports_session ON agent_self_reports(session_id, ts)", ] @@ -5889,6 +5915,344 @@ def query_policy_actions(self, limit: int = 50) -> list: out.append(d) return out + # ── Agent self-reports (WO-59) ───────────────────────────────────────── + # + # Written by the MCP ``report_to_operator`` tool through the daemon's + # ``/__local_query__/ingest_self_report`` (the daemon owns the writer + # lock; the MCP server process never opens DuckDB). Read by + # ``routes/selfdiag.py``, the MCP read tools and the snapshot slice. + + _SELF_REPORT_COLS = ("id", "session_id", "agent_type", "node_id", "model", + "category", "summary_redacted", "ts", "created_at", + "corroborated", "corroboration_ref") + + def _session_model(self, session_id: str) -> str: + """The model the session most recently used, or ``""``. Reads the + typed events table; family adapters stamp ``model`` per request.""" + sid = str(session_id or "").strip() + if not sid: + return "" + try: + from clawmetry.self_diagnostics import bare_session_id as _bare + candidates = [sid] + bare = _bare(sid) + if bare and bare != sid: + candidates.append(bare) + for cand in candidates: + rows = self._fetch( + "SELECT model FROM events WHERE session_id = ? " + "AND model IS NOT NULL AND model <> '' " + "ORDER BY ts DESC LIMIT 1", [cand]) + if rows and rows[0][0]: + return str(rows[0][0])[:128] + rows = self._fetch( + "SELECT model FROM events WHERE session_id LIKE ? " + "AND model IS NOT NULL AND model <> '' " + "ORDER BY ts DESC LIMIT 1", ["%:" + cand]) + if rows and rows[0][0]: + return str(rows[0][0])[:128] + except Exception: + return "" + return "" + + def ingest_self_report(self, session_id: str = "", category: str = "", + summary: str = "", agent_type: str = "", + model: str = "", node_id: str = "", + ts: Any = None, report_id: str = "") -> dict: + """Store one agent self-report. Returns the stored row, or a dict + with ``error`` when the category is not allowed. + + The summary is redacted with the same rules as every other stored + text (``clawmetry.redaction``) and capped BEFORE it reaches the + table; the raw string is never written anywhere. ``model`` and + ``agent_type`` are filled from the store when the caller omits them. + Idempotent per ``report_id``. + """ + from clawmetry import self_diagnostics as _sd + cat = _sd.normalize_category(category) + if not cat: + return {"error": "category not allowed", + "allowed": list(_sd.allowed_categories())} + try: + from clawmetry.redaction import redact_text as _redact + except Exception: # pragma: no cover - redaction ships with the package + def _redact(x): # type: ignore + return x + text = _sd.clip_summary(_redact(_sd.clip_summary(summary))) + sid = str(session_id or "").strip()[:128] + runtime = str(agent_type or "").strip().lower()[:64] + if not runtime and sid: + runtime = _sd.runtime_from_session_id(sid) + if not runtime: + runtime = "unknown" + mdl = str(model or "").strip()[:128] or (self._session_model(sid) if sid else "") + now_ms = int(time.time() * 1000) + ts_epoch = _sd.to_epoch(ts) + ts_ms = int(ts_epoch * 1000) if ts_epoch else now_ms + rid = str(report_id or "").strip()[:64] or uuid.uuid4().hex + row = { + "id": rid, "session_id": sid, "agent_type": runtime, + "node_id": str(node_id or "")[:128], "model": mdl, + "category": cat, "summary_redacted": text, "ts": ts_ms, + "created_at": now_ms, "corroborated": False, "corroboration_ref": "", + } + with self._write_lock: + self._conn.execute(""" + INSERT INTO agent_self_reports ( + id, session_id, agent_type, node_id, model, category, + summary_redacted, ts, created_at, corroborated, + corroboration_ref + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, FALSE, '') + ON CONFLICT (id) DO NOTHING + """, [rid, sid, runtime, row["node_id"], mdl, cat, text, ts_ms, now_ms]) + return row + + def query_self_reports(self, *, since_secs: int = 0, runtime: str = "", + category: str = "", session_id: str = "", + uncorroborated_only: bool = False, + limit: int = 200) -> list: + """Self-reports newest first. ``since_secs <= 0`` means no window. + ``session_id`` matches the stored id or its bare form.""" + clauses: list = [] + params: list = [] + try: + since = int(since_secs or 0) + except (TypeError, ValueError): + since = 0 + if since > 0: + clauses.append("ts >= ?") + params.append(int((time.time() - since) * 1000)) + rt = str(runtime or "").strip().lower() + if rt and rt != "all": + clauses.append("agent_type = ?") + params.append(rt) + cat = str(category or "").strip().lower() + if cat: + clauses.append("category = ?") + params.append(cat) + sid = str(session_id or "").strip() + if sid: + from clawmetry.self_diagnostics import bare_session_id as _bare + bare = _bare(sid) + clauses.append("(session_id = ? OR session_id = ? OR session_id LIKE ?)") + params.extend([sid, bare, "%:" + bare]) + if uncorroborated_only: + clauses.append("corroborated = FALSE") + try: + lim = max(1, min(int(limit or 200), 5000)) + except (TypeError, ValueError): + lim = 200 + where = ("WHERE " + " AND ".join(clauses)) if clauses else "" + sql = (f"SELECT {', '.join(self._SELF_REPORT_COLS)} FROM agent_self_reports " + f"{where} ORDER BY ts DESC, id LIMIT ?") + params.append(lim) + try: + rows = self._fetch(sql, params) + except Exception: + return [] + out = [] + for r in rows: + d = dict(zip(self._SELF_REPORT_COLS, r)) + d["corroborated"] = bool(d.get("corroborated")) + out.append(d) + return out + + def mark_self_report_corroborated(self, report_id: str, ref: str) -> bool: + """Attach independent evidence to one report. Idempotent; returns + True when a row was updated.""" + rid = str(report_id or "").strip() + if not rid: + return False + try: + with self._write_lock: + pre = self._conn.execute( + "SELECT corroborated FROM agent_self_reports WHERE id = ?", + [rid]).fetchone() + if pre is None or bool(pre[0]): + return False + self._conn.execute( + "UPDATE agent_self_reports SET corroborated = TRUE, " + "corroboration_ref = ? WHERE id = ?", + [str(ref or "")[:256], rid]) + return True + except Exception: + return False + + def query_guard_incidents(self, *, since_secs: int = 3600, runtime: str = "", + session_id: str = "", limit: int = 200) -> list: + """Detector incidents (the daemon's ``loop_signals`` rows), newest + first, each carrying the session's model so the honesty rollup can + group by (runtime, model). Never raises; ``[]`` on error.""" + clauses = ["signature LIKE 'daemon_detect_%'"] + params: list = [] + try: + since = int(since_secs or 0) + except (TypeError, ValueError): + since = 3600 + if since > 0: + clauses.append( + "last_seen >= (current_timestamp::TIMESTAMP - INTERVAL (?) SECOND)") + params.append(since) + rt = str(runtime or "").strip().lower() + if rt and rt != "all": + clauses.append("agent_type = ?") + params.append(rt) + sid = str(session_id or "").strip() + if sid: + from clawmetry.self_diagnostics import bare_session_id as _bare + bare = _bare(sid) + clauses.append("(session_id = ? OR session_id = ? OR session_id LIKE ?)") + params.extend([sid, bare, "%:" + bare]) + try: + lim = max(1, min(int(limit or 200), 5000)) + except (TypeError, ValueError): + lim = 200 + sql = f""" + SELECT l.session_id, l.signature, l.repeat_count, l.first_seen, + l.last_seen, l.severity, l.agent_type, l.details, + (SELECT e.model FROM events e WHERE e.session_id = l.session_id + AND e.model IS NOT NULL AND e.model <> '' + ORDER BY e.ts DESC LIMIT 1) AS model + FROM loop_signals l + WHERE {' AND '.join(clauses)} + ORDER BY l.last_seen DESC, l.session_id, l.signature + LIMIT ? + """ + params.append(lim) + cols = ["session_id", "signature", "repeat_count", "first_seen", + "last_seen", "severity", "agent_type", "details", "model"] + try: + rows = self._fetch(sql, params) + except Exception: + return [] + out = [] + for r in rows: + d = dict(zip(cols, r)) + for tcol in ("first_seen", "last_seen"): + v = d.get(tcol) + if hasattr(v, "isoformat"): + d[tcol] = v.isoformat() + raw = d.get("details") + details = None + if raw is not None: + try: + raw = _ccr.maybe_decompress(raw) + text = raw.decode("utf-8") if isinstance(raw, (bytes, bytearray)) else raw + details = json.loads(text) + except Exception: + details = None + d["details"] = details if isinstance(details, dict) else {} + d["kind"] = str(d["details"].get("kind") or + str(d.get("signature") or "").replace("daemon_detect_", "")) + d["title"] = str(d["details"].get("message") or "") + d["runtime"] = str(d.get("agent_type") or "") + d["model"] = str(d.get("model") or "") + out.append(d) + return out + + def query_session_denials(self, *, session_id: str = "", since_secs: int = 3600, + limit: int = 200) -> list: + """Permission denials from the approvals queue (a human or a policy + refused a call). Newest first. ``[]`` on error.""" + clauses = ["status = 'denied'"] + params: list = [] + try: + since = int(since_secs or 0) + except (TypeError, ValueError): + since = 3600 + sid = str(session_id or "").strip() + if sid: + from clawmetry.self_diagnostics import bare_session_id as _bare + bare = _bare(sid) + clauses.append("(requestor_session_id = ? OR requestor_session_id = ? " + "OR requestor_session_id LIKE ?)") + params.extend([sid, bare, "%:" + bare]) + try: + lim = max(1, min(int(limit or 200), 5000)) + except (TypeError, ValueError): + lim = 200 + sql = f""" + SELECT id, requestor_session_id, action, status, created_at, + resolved_at, resolver, decision_reason + FROM approvals WHERE {' AND '.join(clauses)} + ORDER BY COALESCE(resolved_at, created_at) DESC LIMIT ? + """ + params.append(lim) + cols = ["id", "session_id", "action", "status", "created_at", + "resolved_at", "resolver", "decision_reason"] + try: + rows = self._fetch(sql, params) + except Exception: + return [] + from clawmetry.self_diagnostics import to_epoch as _to_epoch + cutoff = time.time() - since if since > 0 else None + out = [] + for r in rows: + d = dict(zip(cols, r)) + t = _to_epoch(d.get("resolved_at") or d.get("created_at")) + if cutoff is not None and t is not None and t < cutoff: + continue + d["ts"] = t + out.append(d) + return out + + def query_self_report_counts(self, *, since_secs: int = 7 * 86400, + runtime: str = "") -> dict: + """``{runtime: {category: n}}`` over the window.""" + from clawmetry.self_diagnostics import count_by_runtime_category + return count_by_runtime_category( + self.query_self_reports(since_secs=since_secs, runtime=runtime, limit=5000)) + + def query_self_report_honesty(self, *, since_secs: int = 7 * 86400, + runtime: str = "") -> list: + """Per (runtime, model): the share of detector incidents the agent + also reported, withheld below the configured minimum.""" + from clawmetry import self_diagnostics as _sd + window = _sd.corroboration_window_secs() + incidents = self.query_guard_incidents( + since_secs=since_secs, runtime=runtime, limit=5000) + reports = self.query_self_reports( + since_secs=int(since_secs) + window, runtime=runtime, limit=5000) + return _sd.honesty_rollup(incidents, reports, window) + + def find_session_by_cwd(self, cwd: str = "", runtime: str = "") -> Any: + """The most recently active session whose recorded working directory + is ``cwd`` (or a parent of it). Used to infer which session an MCP + report belongs to when the agent did not say. ``None`` when unknown.""" + path = str(cwd or "").strip() + if not path: + return None + rt = str(runtime or "").strip().lower() + try: + norm = os.path.normcase(os.path.realpath(os.path.expanduser(path))) + except Exception: + norm = path + clauses = ["cwd IS NOT NULL", "cwd <> ''"] + params: list = [] + if rt: + clauses.append("(agent_type = ? OR session_id LIKE ?)") + params.extend([rt, rt + ":%"]) + sql = (f"SELECT session_id, agent_type, cwd, last_active_at FROM sessions " + f"WHERE {' AND '.join(clauses)} ORDER BY last_active_at DESC NULLS LAST " + f"LIMIT 400") + try: + rows = self._fetch(sql, params) + except Exception: + return None + best = None + best_len = -1 + for sid, atype, scwd, last in rows: + try: + cand = os.path.normcase(os.path.realpath(os.path.expanduser(str(scwd)))) + except Exception: + continue + if norm == cand or norm.startswith(cand.rstrip(os.sep) + os.sep): + if len(cand) > best_len: + best_len = len(cand) + best = {"session_id": sid, "agent_type": atype, "cwd": scwd, + "last_active_at": last} + return best + # ── Guard baselines ─────────────────────────────────────────────────── def record_guard_observation(self, session_id: str, cohort: str, runtime: str = "", agent_id: str = "", diff --git a/clawmetry/mcp_install.py b/clawmetry/mcp_install.py new file mode 100644 index 0000000000..1494cf5aac --- /dev/null +++ b/clawmetry/mcp_install.py @@ -0,0 +1,656 @@ +"""Register the ClawMetry MCP server with each runtime's MCP configuration +(WO-59, REQ-SELF-001). + +One command, every MCP-capable runtime, and the same three rules the hook +installer (``clawmetry/hooks_claude_code.py`` + ``hook_ownership.py``) +already enforces for ``~/.claude/settings.json``: + +* **merge**: the server entry is added to whatever is already there; +* **never delete a foreign entry**: another tool's server is not ours to + touch, and neither is a ``clawmetry`` entry the operator wrote by hand; +* **uninstall removes only what install added**: a marker file + (``~/.clawmetry/mcp_installed.json``) records each registration, and an + entry is removed only when the marker says we wrote it AND the entry + still looks like ours. + +Formats. Each installer below was checked against the vendor's own +documentation (or, for Codex, by running the vendor's ``codex mcp add`` +into a scratch ``CODEX_HOME`` and reading what it wrote). A runtime whose +format could not be verified is reported as ``unknown_format`` and its file +is never written; a runtime with no MCP client is reported as +``no_mcp_support``. Guessing a config format is how a runtime stops +starting, so we do not. + +Stdlib only: this runs from ``clawmetry mcp install`` without the dashboard. +""" +from __future__ import annotations + +import json +import os +import re +import shutil +import sys +import time +from typing import Any, Dict, List, Optional, Tuple + +SERVER_NAME = "clawmetry" +MARKER_PATH = os.path.expanduser("~/.clawmetry/mcp_installed.json") + +# Statuses (install / status / uninstall share one vocabulary). +REGISTERED = "registered" # install wrote it, or status: ours is present +ALREADY_PRESENT = "already_present" # an entry named clawmetry exists (ours or not) +NOT_INSTALLED = "not_installed" +NO_MCP_SUPPORT = "no_mcp_support" +UNKNOWN_FORMAT = "unknown_format" +WOULD_REGISTER = "would_register" # --dry-run +REMOVED = "removed" +LEFT_IN_PLACE = "left_in_place" # present but not ours: never deleted +ERROR = "error" + +# TOML block markers. Codex's config is TOML and Python 3.9 has no TOML +# writer (or reader), so our section is delimited by comment markers and +# only text between them is ever removed. +_TOML_BEGIN = "# clawmetry-mcp:begin (managed by `clawmetry mcp install`; do not edit inside)" +_TOML_END = "# clawmetry-mcp:end" + +# ── Runtime registry ───────────────────────────────────────────────────────── +# +# ``format``: +# json_mcpservers {"mcpServers": {name: {command, args[, type]}}} +# json_opencode {"mcp": {name: {"type": "local", "command": [..], "enabled": true}}} +# toml_mcp_servers [mcp_servers.name] command = ".." args = [".."] +# +# ``verified``: how the format was checked. Recorded here on purpose so a +# future reader can re-check the same source when a vendor moves things. +SUPPORTED: Dict[str, Dict[str, Any]] = { + "claude_code": { + "label": "Claude Code", + "path": "~/.claude.json", + "format": "json_mcpservers", + "entry_type": "stdio", + "guidance_file": "CLAUDE.md", + "verified": "code.claude.com/docs/en/mcp (user scope: mcpServers in ~/.claude.json)", + }, + "cursor": { + "label": "Cursor", + "path": "~/.cursor/mcp.json", + "format": "json_mcpservers", + "entry_type": "", + "guidance_file": "AGENTS.md", + "verified": "cursor.com/docs/context/mcp (global: ~/.cursor/mcp.json, mcpServers)", + }, + "codex": { + "label": "Codex CLI", + "path": "~/.codex/config.toml", + "format": "toml_mcp_servers", + "entry_type": "", + "guidance_file": "AGENTS.md", + "verified": "`codex mcp add` output into a scratch CODEX_HOME: " + "[mcp_servers.] command/args", + }, + "gemini_cli": { + "label": "Gemini CLI", + "path": "~/.gemini/settings.json", + "format": "json_mcpservers", + "entry_type": "", + "guidance_file": "GEMINI.md", + "verified": "geminicli.com/docs/tools/mcp-server (mcpServers in ~/.gemini/settings.json)", + }, + "opencode": { + "label": "OpenCode", + "path": "~/.config/opencode/opencode.json", + "format": "json_opencode", + "entry_type": "local", + "guidance_file": "AGENTS.md", + "verified": "opencode.ai/docs/mcp-servers + /docs/config (mcp: {type: local, command: [..]})", + }, + "windsurf": { + "label": "Windsurf", + "path": "~/.codeium/windsurf/mcp_config.json", + "format": "json_mcpservers", + "entry_type": "", + "guidance_file": "AGENTS.md", + "verified": "docs.windsurf.com/windsurf/cascade/mcp (mcpServers in mcp_config.json)", + }, +} + +# Runtimes with no MCP client to register with. Named plainly rather than +# shown as an empty row (REQ-SELF: "runtimes without MCP support are told +# so"). Keep the reason short and checkable. +NO_MCP: Dict[str, str] = { + "aider": "Aider has no MCP client; tools are built in.", + "lovable": "Lovable is a hosted builder with no local MCP configuration.", + "replit": "Replit Agent is hosted; there is no local MCP configuration to write.", + "grok_bot": "Grok Bot is a chat bot with no MCP client.", + "picoclaw": "PicoClaw has no MCP client.", + "nanoclaw": "NanoClaw has no MCP client.", + "exo": "Exo is an inference cluster, not an agent with tools.", + "openworker": "OpenWorker has no MCP client.", + "deepseek_harness": "The DeepSeek harness has no MCP client.", +} + +# Everything else that ClawMetry observes is ``unknown_format``: the runtime +# may well speak MCP, but its config location was not verified against the +# vendor's documentation, so the installer will not write to it. +_UNKNOWN_NOTE = ("MCP configuration location not verified against the vendor's " + "documentation; register the server by hand (see `clawmetry mcp status`).") + + +def _all_runtime_ids() -> List[str]: + """Every runtime ClawMetry knows, from the entitlement catalogue when it + is importable, plus the installer's own targets (Windsurf is not an + observed runtime but is an MCP host).""" + ids: List[str] = [] + try: + from clawmetry import entitlements as _e + ids.extend(sorted(set(_e.FREE_RUNTIMES) | set(_e.PAID_RUNTIMES))) + except Exception: + pass + for rid in list(SUPPORTED) + list(NO_MCP): + if rid not in ids: + ids.append(rid) + return ids + + +# ── Server command ─────────────────────────────────────────────────────────── + +def resolve_server_command() -> Tuple[str, List[str]]: + """``(command, args)`` that starts the MCP server. + + Prefers an absolute ``clawmetry`` binary (the one on PATH, else the one + beside this interpreter) so the entry keeps working from an IDE that + does not inherit the shell PATH; falls back to ``python -m clawmetry``. + """ + found = shutil.which("clawmetry") + if found: + return os.path.abspath(found), ["mcp"] + sibling = os.path.join(os.path.dirname(sys.executable or ""), "clawmetry") + if sibling and os.path.isfile(sibling) and os.access(sibling, os.X_OK): + return sibling, ["mcp"] + return sys.executable or "python3", ["-m", "clawmetry", "mcp"] + + +def _entry_is_ours(entry: Any) -> bool: + """An entry is ours when it launches ``clawmetry mcp`` (any spelling). + Used together with the marker: both must agree before a removal.""" + try: + blob = json.dumps(entry, default=str).lower() + except Exception: + return False + return "clawmetry" in blob and "mcp" in blob + + +# ── Marker file ────────────────────────────────────────────────────────────── + +def _read_marker(marker_path: str) -> dict: + try: + with open(marker_path, "r", encoding="utf-8") as fh: + data = json.load(fh) + return data if isinstance(data, dict) else {} + except Exception: + return {} + + +def _write_marker(marker_path: str, data: dict) -> None: + os.makedirs(os.path.dirname(marker_path) or ".", exist_ok=True) + tmp = marker_path + ".tmp" + with open(tmp, "w", encoding="utf-8") as fh: + json.dump(data, fh, indent=2, sort_keys=True) + os.replace(tmp, marker_path) + + +# ── JSON files ─────────────────────────────────────────────────────────────── + +def _read_json_file(path: str) -> Tuple[Optional[dict], str]: + """``(data, problem)``. ``data`` is ``{}`` for a missing or empty file and + ``None`` when the file exists but is not plain JSON (JSONC comments, + trailing commas): we will not rewrite a file we cannot round-trip.""" + if not os.path.exists(path): + return {}, "" + try: + with open(path, "r", encoding="utf-8") as fh: + text = fh.read() + except OSError as e: + return None, f"cannot read: {e}" + if not text.strip(): + return {}, "" + try: + data = json.loads(text) + except ValueError: + return None, "file is not plain JSON (comments or trailing commas); edit it by hand" + if not isinstance(data, dict): + return None, "top level is not an object" + return data, "" + + +def _write_json_file(path: str, data: dict) -> None: + os.makedirs(os.path.dirname(path) or ".", exist_ok=True) + tmp = path + ".tmp" + with open(tmp, "w", encoding="utf-8") as fh: + json.dump(data, fh, indent=2, ensure_ascii=False) + fh.write("\n") + os.replace(tmp, path) + + +def _json_entry(spec: dict, command: str, args: List[str]) -> dict: + if spec["format"] == "json_opencode": + return {"type": "local", "command": [command] + list(args), "enabled": True} + entry: Dict[str, Any] = {} + if spec.get("entry_type"): + entry["type"] = spec["entry_type"] + entry["command"] = command + entry["args"] = list(args) + return entry + + +def _json_container_key(spec: dict) -> str: + return "mcp" if spec["format"] == "json_opencode" else "mcpServers" + + +# ── TOML (Codex) ───────────────────────────────────────────────────────────── + +def _toml_str(s: str) -> str: + return json.dumps(s) # JSON string escaping is valid TOML basic-string escaping + + +def _toml_block(command: str, args: List[str], name: str = SERVER_NAME) -> str: + arr = "[" + ", ".join(_toml_str(a) for a in args) + "]" + return (f"{_TOML_BEGIN}\n[mcp_servers.{name}]\n" + f"command = {_toml_str(command)}\nargs = {arr}\n{_TOML_END}\n") + + +def _toml_has_section(text: str, name: str = SERVER_NAME) -> bool: + pat = re.compile(r"^\s*\[mcp_servers\." + re.escape(name) + r"(?:\.[A-Za-z0-9_-]+)?\]\s*$", + re.MULTILINE) + return bool(pat.search(text)) + + +def _toml_has_our_block(text: str) -> bool: + return _TOML_BEGIN in text and _TOML_END in text + + +def _toml_strip_our_block(text: str) -> str: + start = text.find(_TOML_BEGIN) + end = text.find(_TOML_END, start) + if start < 0 or end < 0: + return text + end += len(_TOML_END) + if end < len(text) and text[end] == "\n": + end += 1 + head = text[:start].rstrip("\n") + tail = text[end:].lstrip("\n") + if head and tail: + return head + "\n\n" + tail + return (head + "\n") if head else tail + + +# ── Installer ──────────────────────────────────────────────────────────────── + +class Installer: + """Per-home installer so tests (and a future ``--home``) never touch the + operator's real files.""" + + def __init__(self, home: Optional[str] = None, marker_path: Optional[str] = None, + command: Optional[str] = None, args: Optional[List[str]] = None): + self.home = os.path.expanduser(home or "~") + self.marker_path = marker_path or ( + os.path.join(self.home, ".clawmetry", "mcp_installed.json") + if home else MARKER_PATH) + if command is None: + command, resolved_args = resolve_server_command() + args = resolved_args if args is None else args + self.command = command + self.args = list(args or ["mcp"]) + + # -- helpers --------------------------------------------------------------- + def path_for(self, runtime: str) -> str: + spec = SUPPORTED[runtime] + rel = spec["path"] + if rel.startswith("~/"): + return os.path.join(self.home, rel[2:]) + return os.path.expanduser(rel) + + def _classify(self, runtime: str) -> Optional[dict]: + """A terminal result for runtimes we cannot write, else ``None``.""" + rid = str(runtime or "").strip().lower() + if rid in SUPPORTED: + return None + if rid in NO_MCP: + return {"runtime": rid, "status": NO_MCP_SUPPORT, "path": "", + "detail": NO_MCP[rid]} + return {"runtime": rid, "status": UNKNOWN_FORMAT, "path": "", + "detail": _UNKNOWN_NOTE} + + def _marker_says_ours(self, runtime: str) -> bool: + rec = _read_marker(self.marker_path).get(runtime) + return isinstance(rec, dict) and bool(rec.get("server_name")) + + # -- status ---------------------------------------------------------------- + def status(self, runtime: str) -> dict: + term = self._classify(runtime) + if term: + return term + spec = SUPPORTED[runtime] + path = self.path_for(runtime) + base = {"runtime": runtime, "path": path, "label": spec["label"], + "verified": spec["verified"]} + if spec["format"] == "toml_mcp_servers": + try: + text = open(path, "r", encoding="utf-8").read() if os.path.exists(path) else "" + except OSError as e: + return dict(base, status=ERROR, detail=str(e)) + if _toml_has_our_block(text) and self._marker_says_ours(runtime): + return dict(base, status=REGISTERED, detail="registered by clawmetry mcp install") + if _toml_has_section(text): + return dict(base, status=ALREADY_PRESENT, + detail="a [mcp_servers.clawmetry] section exists that ClawMetry did not write") + return dict(base, status=NOT_INSTALLED, detail="") + data, problem = _read_json_file(path) + if data is None: + return dict(base, status=UNKNOWN_FORMAT, detail=problem) + container = data.get(_json_container_key(spec)) + entry = container.get(SERVER_NAME) if isinstance(container, dict) else None + if entry is None: + return dict(base, status=NOT_INSTALLED, detail="") + if _entry_is_ours(entry) and self._marker_says_ours(runtime): + return dict(base, status=REGISTERED, detail="registered by clawmetry mcp install") + return dict(base, status=ALREADY_PRESENT, + detail="an entry named clawmetry exists that ClawMetry did not write") + + # -- install --------------------------------------------------------------- + def install(self, runtime: str, dry_run: bool = False) -> dict: + term = self._classify(runtime) + if term: + return term + current = self.status(runtime) + if current["status"] == REGISTERED: + return dict(current, status=ALREADY_PRESENT, + detail="already registered by clawmetry mcp install") + if current["status"] in (ALREADY_PRESENT, UNKNOWN_FORMAT, ERROR): + return current + spec = SUPPORTED[runtime] + path = current["path"] + base = {"runtime": runtime, "path": path, "label": spec["label"], + "verified": spec["verified"]} + if dry_run: + return dict(base, status=WOULD_REGISTER, + detail=f"would add server '{SERVER_NAME}' -> {self.command} {' '.join(self.args)}") + try: + if spec["format"] == "toml_mcp_servers": + text = open(path, "r", encoding="utf-8").read() if os.path.exists(path) else "" + if text and not text.endswith("\n"): + text += "\n" + if text.strip(): + text += "\n" + text += _toml_block(self.command, self.args) + os.makedirs(os.path.dirname(path) or ".", exist_ok=True) + tmp = path + ".tmp" + with open(tmp, "w", encoding="utf-8") as fh: + fh.write(text) + os.replace(tmp, path) + else: + data, problem = _read_json_file(path) + if data is None: + return dict(base, status=UNKNOWN_FORMAT, detail=problem) + key = _json_container_key(spec) + container = data.get(key) + if container is None: + container = {} + data[key] = container + if not isinstance(container, dict): + return dict(base, status=UNKNOWN_FORMAT, + detail=f"'{key}' is not an object; edit it by hand") + container[SERVER_NAME] = _json_entry(spec, self.command, self.args) + _write_json_file(path, data) + marker = _read_marker(self.marker_path) + marker[runtime] = { + "server_name": SERVER_NAME, "path": path, + "command": self.command, "args": list(self.args), + "installed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + } + _write_marker(self.marker_path, marker) + except Exception as e: # noqa: BLE001 + return dict(base, status=ERROR, detail=str(e)) + return dict(base, status=REGISTERED, detail="registered") + + # -- uninstall ------------------------------------------------------------- + def uninstall(self, runtime: str) -> dict: + term = self._classify(runtime) + if term: + return term + current = self.status(runtime) + spec = SUPPORTED[runtime] + path = current["path"] + base = {"runtime": runtime, "path": path, "label": spec["label"]} + if current["status"] == NOT_INSTALLED: + return dict(base, status=NOT_INSTALLED, detail="") + if current["status"] == ALREADY_PRESENT: + return dict(base, status=LEFT_IN_PLACE, + detail="entry named clawmetry was not written by ClawMetry; left as is") + if current["status"] in (UNKNOWN_FORMAT, ERROR): + return current + try: + if spec["format"] == "toml_mcp_servers": + text = open(path, "r", encoding="utf-8").read() + new_text = _toml_strip_our_block(text) + tmp = path + ".tmp" + with open(tmp, "w", encoding="utf-8") as fh: + fh.write(new_text) + os.replace(tmp, path) + else: + data, problem = _read_json_file(path) + if data is None: + return dict(base, status=UNKNOWN_FORMAT, detail=problem) + key = _json_container_key(spec) + container = data.get(key) + if isinstance(container, dict) and _entry_is_ours(container.get(SERVER_NAME)): + del container[SERVER_NAME] + # An empty container we created is ours to drop; one the + # operator had before stays (a foreign key may be empty + # on purpose). We only know the difference from the + # marker, so keep it: an empty object is harmless. + _write_json_file(path, data) + marker = _read_marker(self.marker_path) + marker.pop(runtime, None) + _write_marker(self.marker_path, marker) + except Exception as e: # noqa: BLE001 + return dict(base, status=ERROR, detail=str(e)) + return dict(base, status=REMOVED, detail="removed the entry ClawMetry added") + + # -- fan-out --------------------------------------------------------------- + def targets(self, runtime: Optional[str]) -> List[str]: + rid = (runtime or "all").strip().lower() + if rid in ("", "all"): + return list(SUPPORTED) + return [rid] + + def install_all(self, runtime: Optional[str] = None, dry_run: bool = False) -> List[dict]: + return [self.install(r, dry_run=dry_run) for r in self.targets(runtime)] + + def uninstall_all(self, runtime: Optional[str] = None) -> List[dict]: + return [self.uninstall(r) for r in self.targets(runtime)] + + def status_all(self, runtime: Optional[str] = None) -> List[dict]: + rid = (runtime or "all").strip().lower() + ids = _all_runtime_ids() if rid in ("", "all") else [rid] + return [self.status(r) for r in ids] + + +# ── Guidance snippet (offered, never written without consent) ──────────────── + +GUIDANCE_MARKER = "" + + +def guidance_snippet(runtime: str = "") -> str: + """The instructions-file snippet an operator may paste. Written to a + file only by :func:`write_guidance`, which only ``--write-guidance`` + calls.""" + return ( + f"{GUIDANCE_MARKER}\n" + "## Reporting to your operators\n\n" + "ClawMetry's MCP server is available. Before finishing a task, use the\n" + "`report_to_operator` tool for anything that got in the way: a tool that\n" + "kept failing, context or a permission you were missing, something you\n" + "could not do, a task you could not finish, a block you worked around, or\n" + "anything else the people who run you would want to know. Keep each note\n" + "to a sentence or two.\n" + ) + + +def guidance_file_for(runtime: str) -> str: + spec = SUPPORTED.get(str(runtime or "").lower()) + return spec["guidance_file"] if spec else "AGENTS.md" + + +def write_guidance(runtime: str, project_dir: str) -> dict: + """Append the snippet to the runtime's instructions file in + ``project_dir``. Idempotent via :data:`GUIDANCE_MARKER`. Only ever + called with explicit consent (``--write-guidance``).""" + name = guidance_file_for(runtime) + path = os.path.join(os.path.abspath(project_dir), name) + try: + existing = "" + if os.path.exists(path): + with open(path, "r", encoding="utf-8") as fh: + existing = fh.read() + if GUIDANCE_MARKER in existing: + return {"runtime": runtime, "path": path, "status": ALREADY_PRESENT} + with open(path, "a", encoding="utf-8") as fh: + if existing and not existing.endswith("\n"): + fh.write("\n") + if existing: + fh.write("\n") + fh.write(guidance_snippet(runtime)) + return {"runtime": runtime, "path": path, "status": "written"} + except Exception as e: # noqa: BLE001 + return {"runtime": runtime, "path": path, "status": ERROR, "detail": str(e)} + + +# ── Support matrix for the dashboard ───────────────────────────────────────── + +def support_matrix(home: Optional[str] = None) -> List[dict]: + """One row per runtime: ``mcp`` is ``supported`` / ``not_supported`` / + ``unknown``; ``status`` is the live registration state.""" + inst = Installer(home=home) + rows: List[dict] = [] + for rid in _all_runtime_ids(): + st = inst.status(rid) + if rid in SUPPORTED: + mcp = "supported" + elif rid in NO_MCP: + mcp = "not_supported" + else: + mcp = "unknown" + # An ``error`` row carries an OSError message in ``detail``. The CLI + # prints it; the dashboard route does not, so the served row gets a + # fixed sentence instead of exception text. + detail = st.get("detail", "") + if st.get("status") == ERROR: + detail = "could not read this runtime's configuration file" + rows.append({ + "runtime": rid, + "label": SUPPORTED.get(rid, {}).get("label", rid), + "mcp": mcp, + "status": st.get("status"), + "path": st.get("path", ""), + "detail": detail, + }) + return rows + + +# ── CLI ────────────────────────────────────────────────────────────────────── + +_STATUS_WORDS = { + REGISTERED: "registered", + ALREADY_PRESENT: "already present", + NOT_INSTALLED: "not installed", + NO_MCP_SUPPORT: "no MCP support", + UNKNOWN_FORMAT: "unknown config format", + WOULD_REGISTER: "would register (dry run)", + REMOVED: "removed", + LEFT_IN_PLACE: "left in place (not ours)", + ERROR: "error", +} + + +def _print_rows(rows: List[dict]) -> None: + width = max([len(r["runtime"]) for r in rows] + [8]) + for r in rows: + word = _STATUS_WORDS.get(r.get("status"), str(r.get("status"))) + line = f" {r['runtime']:<{width}} {word}" + if r.get("path"): + line += f" {r['path']}" + if r.get("detail") and r.get("status") not in (REGISTERED, NOT_INSTALLED): + line += f" ({r['detail']})" + print(line) + + +def cli_main(argv: Optional[List[str]] = None) -> int: + """``clawmetry mcp install [--runtime |all] [--dry-run] + [--write-guidance]`` | ``uninstall`` | ``status``. Exit 0 always for + status; install/uninstall exit 1 only on a write error.""" + import argparse + p = argparse.ArgumentParser(prog="clawmetry mcp", add_help=True) + p.add_argument("mcp_cmd", nargs="?", default="serve", + choices=["serve", "install", "uninstall", "status"]) + p.add_argument("--runtime", default="all", + help="one runtime id (claude_code, cursor, codex, gemini_cli, " + "opencode, windsurf) or 'all'") + p.add_argument("--dry-run", action="store_true", + help="show what install would write; write nothing") + p.add_argument("--write-guidance", action="store_true", + help="also append the guidance snippet to the runtime's instructions " + "file in the current directory (never done without this flag)") + p.add_argument("--json", action="store_true", help="machine-readable output") + args = p.parse_args(argv) + + if args.mcp_cmd == "serve": + from clawmetry.mcp_server import run + run() + return 0 + + inst = Installer() + if args.mcp_cmd == "status": + rows = inst.status_all(args.runtime) + if args.json: + print(json.dumps(rows, indent=2)) + else: + print("ClawMetry MCP server registration:") + _print_rows(rows) + print(f"\n server command: {inst.command} {' '.join(inst.args)}") + return 0 + + if args.mcp_cmd == "install": + rows = inst.install_all(args.runtime, dry_run=args.dry_run) + guidance: List[dict] = [] + if args.write_guidance and not args.dry_run: + for r in rows: + if r.get("status") in (REGISTERED, ALREADY_PRESENT): + guidance.append(write_guidance(r["runtime"], os.getcwd())) + if args.json: + print(json.dumps({"results": rows, "guidance": guidance}, indent=2)) + else: + print("ClawMetry MCP server install:") + _print_rows(rows) + print(f"\n server command: {inst.command} {' '.join(inst.args)}") + if guidance: + print("\n guidance written:") + for g in guidance: + print(f" {g['runtime']}: {g['status']} {g['path']}") + else: + print("\nOffered guidance for your instructions file (CLAUDE.md, AGENTS.md, " + "GEMINI.md). Not written. Re-run with --write-guidance to append it to " + "the file in the current directory:\n") + for line in guidance_snippet().splitlines(): + if line.startswith(" +
"no reports". + "ingest_self_report", + "query_self_reports", + "query_self_report_counts", + "query_self_report_honesty", + "query_guard_incidents", + "query_session_denials", + "find_session_by_cwd", }) diff --git a/routes/selfdiag.py b/routes/selfdiag.py new file mode 100644 index 0000000000..a5b0e93e2d --- /dev/null +++ b/routes/selfdiag.py @@ -0,0 +1,120 @@ +"""Agent self-diagnostics read API (WO-59). + +Three read-only endpoints over the ``agent_self_reports`` table: + +* ``GET /api/self-reports`` the reports themselves (session view) +* ``GET /api/self-reports/honesty`` per (runtime, model) honesty rollup +* ``GET /api/self-reports/support`` per runtime: MCP supported / registered + +Every read goes through the daemon proxy (``_ls_call``), never a raw file +and never a writable store open: the daemon owns the DuckDB writer lock and +this module runs in the dashboard process. Nothing here writes; the only +writer is the MCP tool, through the daemon. +""" +from __future__ import annotations + +import logging + +from flask import Blueprint, jsonify, request + +log = logging.getLogger("clawmetry.selfdiag") + +bp_selfdiag = Blueprint("selfdiag", __name__) + + +def _ls_call(method_name, **kwargs): + """Cross-process LocalStore read with single-process fallback. Mirror + of ``routes/guard.py::_ls_call``.""" + try: + from routes.local_query import local_store_via_daemon + result = local_store_via_daemon(method_name, **kwargs) + if result is not None: + return result + except Exception: + pass + try: + from clawmetry import local_store + store = local_store.get_store(read_only=True) + return getattr(store, method_name)(**kwargs) + except Exception: + return None + + +def _window_secs(default: int) -> int: + from clawmetry.self_diagnostics import parse_window_secs + return parse_window_secs(request.args.get("window"), default) + + +def _runtime_arg() -> str: + rt = (request.args.get("runtime") or "").strip().lower() + return "" if rt in ("", "all") else rt + + +@bp_selfdiag.route("/api/self-reports") +def api_self_reports(): + """Self-reports, newest first. ``session`` narrows to one session (the + transcript view), ``runtime`` / ``category`` / ``window`` filter the + list. Always HTTP 200 with an honest empty list on a store error.""" + from clawmetry import self_diagnostics as _sd + window = _window_secs(_sd.DEFAULT_WINDOW_SECS) + session = (request.args.get("session") or request.args.get("session_id") or "").strip() + category = (request.args.get("category") or "").strip().lower() + try: + limit = max(1, min(int(request.args.get("limit", 200)), 2000)) + except (TypeError, ValueError): + limit = 200 + rows = _ls_call( + "query_self_reports", + since_secs=0 if session else window, + runtime=_runtime_arg(), category=category, + session_id=session, limit=limit, + ) + rows = rows if isinstance(rows, list) else [] + return jsonify({ + "reports": rows, + "count": len(rows), + "window_secs": window, + "corroboration_window_secs": _sd.corroboration_window_secs(), + "categories": list(_sd.allowed_categories()), + # The plain-words meaning of the label, so every consumer says the + # same thing: no independent evidence is not the same as false. + "uncorroborated_means": ( + "No independent evidence was found for this report, which is " + "not the same as false." + ), + "store_reachable": rows is not None, + }) + + +@bp_selfdiag.route("/api/self-reports/honesty") +def api_self_reports_honesty(): + """Per (runtime, model): the share of detector incidents the agent also + reported, plus counts per category. Cohorts under the minimum incident + count carry ``withheld: true`` and a reason instead of a figure.""" + from clawmetry import self_diagnostics as _sd + window = _window_secs(_sd.DEFAULT_WINDOW_SECS) + runtime = _runtime_arg() + honesty = _ls_call("query_self_report_honesty", since_secs=window, runtime=runtime) + counts = _ls_call("query_self_report_counts", since_secs=window, runtime=runtime) + return jsonify({ + "window_secs": window, + "runtime": runtime or "all", + "honesty": honesty if isinstance(honesty, list) else [], + "counts": counts if isinstance(counts, dict) else {}, + "min_incidents": _sd.min_incidents(), + "corroboration_window_secs": _sd.corroboration_window_secs(), + }) + + +@bp_selfdiag.route("/api/self-reports/support") +def api_self_reports_support(): + """Per runtime: whether ClawMetry can register its MCP server there, + and whether it is registered right now. Runtimes with no MCP client + are named as such rather than shown as an empty row.""" + try: + from clawmetry import mcp_install as _mi + rows = _mi.support_matrix() + except Exception as e: # noqa: BLE001 + log.debug("mcp support matrix failed: %s", e) + rows = [] + return jsonify({"runtimes": rows, "count": len(rows)}) diff --git a/tests/test_self_diagnostics.py b/tests/test_self_diagnostics.py new file mode 100644 index 0000000000..90c7a692f4 --- /dev/null +++ b/tests/test_self_diagnostics.py @@ -0,0 +1,673 @@ +"""Self-diagnostics over MCP (WO-59, REQ-SELF-001..004). + +Pins the contract end to end without a daemon: + +* the MCP tool catalogue: ``report_to_operator`` with the six categories + (operator-extendable), the four read tools, and NO tool that acts on a + process; +* the write path through a temp DuckDB store: redaction applied before + storage, the summary cap, category validation, idempotency; +* corroboration: the window logic (pure), the daemon-tick pass against real + ``loop_signals`` and ``approvals`` rows, session-id prefix matching; +* honesty: withheld below the minimum, a ratio above it; +* the installer, per verified runtime format: merge, never delete a foreign + entry, uninstall removes only ours, a hand-written entry is left in place, + JSONC is refused rather than guessed, status vocabulary; +* the read routes and the snapshot slice; +* the CLI fast path never imports the dashboard. + +Every file the installer touches lives under a ``tmp_path`` home. Nothing +here reads or writes the developer's real ``~/.claude.json`` & co. +""" +from __future__ import annotations + +import builtins +import importlib +import json +import os +import sys +import time +from datetime import datetime, timezone + +import pytest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from clawmetry import self_diagnostics as sd # noqa: E402 +from clawmetry import mcp_install as mi # noqa: E402 +from clawmetry import mcp_server as ms # noqa: E402 + + +# ── fixtures ─────────────────────────────────────────────────────────────── + +@pytest.fixture +def store(tmp_path, monkeypatch): + monkeypatch.setenv("CLAWMETRY_LOCAL_STORE_PATH", str(tmp_path / "selfdiag.duckdb")) + monkeypatch.setenv("CLAWMETRY_LOCAL_FLUSH_SECS", "0.05") + monkeypatch.setenv("CLAWMETRY_LOCAL_FLUSH_BATCH", "5") + pytest.importorskip("duckdb") + import clawmetry.local_store as ls + + importlib.reload(ls) + s = ls.LocalStore() + s.start() + yield s + s.stop(flush=True) + + +@pytest.fixture +def home(tmp_path, monkeypatch): + """A fake HOME so neither the installer nor the CLI can touch real files.""" + h = tmp_path / "home" + h.mkdir() + monkeypatch.setenv("HOME", str(h)) + monkeypatch.setenv("USERPROFILE", str(h)) + return h + + +def _installer(home): + return mi.Installer(home=str(home), command="/opt/clawmetry/bin/clawmetry", args=["mcp"]) + + +def _iso_local(offset_secs=0): + return datetime.fromtimestamp(time.time() + offset_secs).isoformat(timespec="seconds") + + +# ── 1. tool catalogue ────────────────────────────────────────────────────── + +def _tool(name): + for t in ms.tools_catalogue(): + if t["name"] == name: + return t + raise AssertionError(f"tool {name} missing from catalogue") + + +def test_report_tool_schema_has_six_default_categories(): + t = _tool("report_to_operator") + props = t["inputSchema"]["properties"] + assert set(t["inputSchema"]["required"]) == {"category", "summary"} + assert props["category"]["enum"] == list(sd.DEFAULT_CATEGORIES) + assert len(sd.DEFAULT_CATEGORIES) == 6 + assert "session_id" in props and "runtime" in props + # Framed as feedback to the people who run the agent, never confession. + desc = t["description"].lower() + assert "operators" in desc and "got in the way" in desc + for banned in ("confess", "unsafe", "violation", "misbehav"): + assert banned not in desc + + +def test_operator_can_extend_categories(tmp_path, monkeypatch): + cfg = tmp_path / "config.json" + cfg.write_text(json.dumps({"self_diagnostics": {"categories": ["flaky_ci", "BAD CAT", 7]}})) + monkeypatch.setattr(sd, "_CONFIG_PATH", str(cfg)) + cats = sd.allowed_categories() + assert cats[:6] == sd.DEFAULT_CATEGORIES + assert "flaky_ci" in cats + assert "bad cat" not in cats and "BAD CAT" not in cats + assert _tool("report_to_operator")["inputSchema"]["properties"]["category"]["enum"][-1] == "flaky_ci" + assert sd.normalize_category("FLAKY_CI") == "flaky_ci" + assert sd.normalize_category("made_up") is None + + +def test_read_tools_present_and_nothing_actuates(): + names = {t["name"] for t in ms.tools_catalogue()} + for required in ("list_incidents", "get_guard_status", "get_signal_rates", + "list_self_reports", "report_to_operator", + "list_sessions", "get_cost_summary", "get_session_trace", + "list_events", "get_health"): + assert required in names + for n in names: + for word in ms.ACTUATING_WORDS: + assert word not in n.lower(), f"{n} looks like an actuating tool" + assert "control" not in n.lower() + # The one write tool is the report; every other tool is a read. + assert names - {"report_to_operator"} == { + n for n in names if n.startswith(("list_", "get_"))} + + +@pytest.mark.parametrize("name,args", [ + ("report_to_operator", {"category": "task_failure", "summary": "could not finish", + "session_id": "claude_code:s1"}), + ("list_incidents", {}), + ("get_guard_status", {"session_id": "claude_code:s1"}), + ("get_signal_rates", {}), + ("list_self_reports", {}), + ("get_health", {}), +]) +def test_every_tool_is_honest_when_daemon_is_down(monkeypatch, name, args): + monkeypatch.setattr(ms, "_read_discovery", lambda: None) + resp = ms.handle_request({"jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": {"name": name, "arguments": args}}) + assert resp["result"]["isError"] is True + body = json.loads(resp["result"]["content"][0]["text"]) + assert "daemon is not running" in body["error"] + + +def test_busy_daemon_is_not_reported_as_down(monkeypatch): + import socket + import urllib.request + + monkeypatch.setattr(ms, "_read_discovery", lambda: {"port": 1, "token": "t"}) + + def _raise(*a, **k): + raise socket.timeout("timed out") + + monkeypatch.setattr(urllib.request, "urlopen", _raise) + out = ms._call_tool("list_incidents", {}) + assert out["code"] == "timeout" + assert "running but did not answer" in out["error"] + assert "not running" not in out["error"] + + +def test_old_daemon_refusal_names_the_upgrade(monkeypatch): + monkeypatch.setattr(ms, "_post", lambda path, payload: { + "error": "method not allowed: 'query_guard_incidents'", "code": "refused", "status": 400}) + out = ms._call_tool("list_incidents", {}) + assert out["code"] == "refused" and "clawmetry update" in out["error"] + out = ms._call_tool("report_to_operator", {"category": "noteworthy", "summary": "x", + "session_id": "s"}) + assert out["code"] == "refused" and "too old" in out["error"] + + +def test_signal_rates_shape_grouped_counts_from_daemon(monkeypatch): + seen = {} + + def fake_method(name, **kw): + seen["name"] = name + seen.update(kw) + return {"result": {"turns": [], "matches": []}} + + monkeypatch.setattr(ms, "_method", fake_method) + out = ms._call_tool("get_signal_rates", {"window": "7d", "runtime": "codex"}) + assert seen["name"] == "query_signal_grouped" and seen["runtime"] == "codex" + assert out["available"] is True and out["runtime"] == "codex" + assert isinstance(out["rates"], dict) + + +def test_report_tool_rejects_unknown_category_before_touching_daemon(monkeypatch): + monkeypatch.setattr(ms, "_read_discovery", lambda: None) + out = ms._call_tool("report_to_operator", {"category": "nope", "summary": "x"}) + assert "not one of the allowed" in out["error"] + assert out["allowed"] == list(sd.DEFAULT_CATEGORIES) + + +def test_signal_rates_says_not_available_when_daemon_lacks_method(monkeypatch): + monkeypatch.setattr(ms, "_method", lambda name, **kw: { + "error": "method not allowed: 'query_signal_rates'", "code": "refused"}) + out = ms._call_tool("get_signal_rates", {"window": "24h"}) + assert out["available"] is False + assert out["error"] == "signals not available on this daemon version" + + +# ── 2. write path through the store ─────────────────────────────────────── + +def test_ingest_applies_redaction_and_cap(store): + secret = "sk-ant-abcdefghijklmnopqrstuvwxyz0123456789" + long_tail = " padding" * 200 + row = store.ingest_self_report( + session_id="claude_code:abc", category="repeatedly_broken_tool", + summary=f"the deploy tool failed with api_key={secret}{long_tail}", + agent_type="claude_code", model="claude-x", node_id="n1") + assert row["category"] == "repeatedly_broken_tool" + assert secret not in row["summary_redacted"] + assert "[REDACTED:" in row["summary_redacted"] + assert len(row["summary_redacted"]) <= sd.SUMMARY_MAX_CHARS + stored = store.query_self_reports(session_id="abc") + assert len(stored) == 1 + assert secret not in stored[0]["summary_redacted"] + assert stored[0]["corroborated"] is False + assert stored[0]["model"] == "claude-x" and stored[0]["node_id"] == "n1" + + +def test_ingest_rejects_bad_category_and_is_idempotent(store): + bad = store.ingest_self_report(session_id="s", category="made_up", summary="x") + assert bad["error"] == "category not allowed" + store.ingest_self_report(session_id="s", category="noteworthy", summary="one", + report_id="fixed-id") + store.ingest_self_report(session_id="s", category="noteworthy", summary="two", + report_id="fixed-id") + rows = store.query_self_reports(session_id="s") + assert len(rows) == 1 and rows[0]["summary_redacted"] == "one" + + +def test_runtime_derived_from_session_prefix_when_absent(store): + row = store.ingest_self_report(session_id="codex:xyz", category="capability_gap", + summary="no browser") + assert row["agent_type"] == "codex" + row2 = store.ingest_self_report(session_id="", category="capability_gap", summary="x") + assert row2["agent_type"] == "unknown" + + +def test_mcp_tool_writes_through_daemon_method(store, monkeypatch): + """The MCP process never opens DuckDB: the tool calls the daemon's + ``ingest_self_report`` method. Simulate the proxy with the temp store.""" + calls = [] + + def fake_method(name, **kwargs): + calls.append(name) + return {"result": getattr(store, name)(**kwargs)} + + monkeypatch.setattr(ms, "_method", fake_method) + monkeypatch.setattr(ms, "_node_id", lambda: "node-1") + out = ms._call_tool("report_to_operator", { + "category": "bypassed_block", "summary": "write tool blocked, used bash", + "session_id": "claude_code:abc"}) + assert out["ok"] is True and out["session_source"] == "argument" + assert calls == ["ingest_self_report"] + assert store.query_self_reports(session_id="claude_code:abc")[0]["node_id"] == "node-1" + + listed = ms._call_tool("list_self_reports", {"window": "1h", "category": "bypassed_block"}) + assert listed["count"] == 1 + assert "not the same as false" in listed["uncorroborated_means"] + + +def test_session_inferred_from_env_then_cwd(store, monkeypatch, tmp_path): + monkeypatch.setattr(ms, "_method", lambda name, **kw: {"result": getattr(store, name)(**kw)}) + monkeypatch.setattr(ms, "_node_id", lambda: "") + for var in sd._SESSION_ENV_VARS: + monkeypatch.delenv(var, raising=False) + monkeypatch.setenv("CLAUDE_SESSION_ID", "env-session") + out = ms._call_tool("report_to_operator", {"category": "noteworthy", "summary": "hi"}) + assert out["session_id"] == "env-session" and out["session_source"] == "environment" + + monkeypatch.delenv("CLAUDE_SESSION_ID") + proj = tmp_path / "proj" + (proj / "sub").mkdir(parents=True) + store.ingest_session({"agent_type": "claude_code", "session_id": "claude_code:cwd1", + "node_id": "n", "agent_id": "main", + "last_active_at": datetime.now(timezone.utc).isoformat()}) + store.update_session_location("claude_code:cwd1", cwd=str(proj), agent_type="claude_code") + monkeypatch.chdir(proj / "sub") + out = ms._call_tool("report_to_operator", {"category": "noteworthy", "summary": "hi"}) + assert out["session_id"] == "claude_code:cwd1" + assert out["session_source"] == "working directory" + + +# ── 3. corroboration ────────────────────────────────────────────────────── + +def _inc(sid="claude_code:abc", first=1000.0, last=1100.0, sig="daemon_detect_stuck_loop"): + return {"session_id": sid, "signature": sig, "first_seen": first, "last_seen": last} + + +def test_window_logic_is_inclusive_on_both_sides(): + inc = _inc() + w = 600 + assert sd.find_evidence({"session_id": "abc", "ts": 1100.0 + w}, [inc], [], w) + assert sd.find_evidence({"session_id": "abc", "ts": 1000.0 - w}, [inc], [], w) + assert sd.find_evidence({"session_id": "abc", "ts": 1100.0 + w + 1}, [inc], [], w) is None + assert sd.find_evidence({"session_id": "abc", "ts": 1000.0 - w - 1}, [inc], [], w) is None + # Same session id family: bare id matches the prefixed incident. + assert sd.find_evidence({"session_id": "claude_code:abc", "ts": 1050.0}, [inc], [], w) \ + == "incident:claude_code:abc:daemon_detect_stuck_loop" + # A different session never corroborates. + assert sd.find_evidence({"session_id": "other", "ts": 1050.0}, [inc], [], w) is None + + +def test_nearest_incident_wins_and_denials_are_fallback(): + old = _inc(first=0.0, last=100.0, sig="daemon_detect_no_progress") + near = _inc(first=900.0, last=1000.0, sig="daemon_detect_stuck_loop") + ref = sd.find_evidence({"session_id": "abc", "ts": 1010.0}, [old, near], [], 2000) + assert ref == "incident:claude_code:abc:daemon_detect_stuck_loop" + denial = {"id": "ap1", "session_id": "claude_code:abc", "ts": 5000.0} + assert sd.find_evidence({"session_id": "abc", "ts": 5100.0}, [], [denial], 600) == "denial:ap1" + assert sd.find_evidence({"session_id": "abc", "ts": 5700.0}, [], [denial], 600) is None + + +def test_window_constant_and_env_override(monkeypatch): + assert sd.CORROBORATION_WINDOW_SECS == 600 + monkeypatch.delenv("CLAWMETRY_SELFDIAG_WINDOW_SECS", raising=False) + assert sd.corroboration_window_secs() == 600 + monkeypatch.setenv("CLAWMETRY_SELFDIAG_WINDOW_SECS", "30") + assert sd.corroboration_window_secs() == 30 + monkeypatch.setenv("CLAWMETRY_SELFDIAG_WINDOW_SECS", "garbage") + assert sd.corroboration_window_secs() == 600 + + +def test_daemon_tick_corroborates_against_real_incident_rows(store, monkeypatch): + monkeypatch.delenv("CLAWMETRY_SELFDIAG_WINDOW_SECS", raising=False) + store.ingest_loop_signal( + session_id="claude_code:abc", signature="daemon_detect_stuck_loop", + repeat_count=5, severity="warning", agent_type="claude_code", + details={"source": "daemon_detector", "kind": "stuck_loop", "message": "looping"}) + store.ingest_self_report(session_id="abc", category="repeatedly_broken_tool", + summary="same command kept failing") + # A report far outside the window stays uncorroborated. + store.ingest_self_report(session_id="abc", category="noteworthy", summary="old", + ts=time.time() - 4 * 3600) + incidents = store.query_guard_incidents(since_secs=3600, session_id="abc") + assert len(incidents) == 1 and incidents[0]["kind"] == "stuck_loop" + assert sd.corroborate_pending(store) == 1 + rows = {r["summary_redacted"]: r for r in store.query_self_reports(session_id="abc")} + assert rows["same command kept failing"]["corroborated"] is True + assert rows["same command kept failing"]["corroboration_ref"] == \ + "incident:claude_code:abc:daemon_detect_stuck_loop" + assert rows["old"]["corroborated"] is False + # Second tick is a no-op. + assert sd.corroborate_pending(store) == 0 + + +def test_permission_denial_corroborates(store, monkeypatch): + monkeypatch.delenv("CLAWMETRY_SELFDIAG_WINDOW_SECS", raising=False) + now_utc = datetime.now(timezone.utc).isoformat() + store.ingest_approval({"id": "ap-1", "requestor_session_id": "claude_code:d1", + "action": "Bash", "status": "denied", + "created_at": now_utc, "resolved_at": now_utc}) + denials = store.query_session_denials(session_id="d1", since_secs=3600) + assert len(denials) == 1 and denials[0]["id"] == "ap-1" + store.ingest_self_report(session_id="d1", category="bypassed_block", + summary="write was refused, used bash instead") + assert sd.corroborate_pending(store) == 1 + row = store.query_self_reports(session_id="d1")[0] + assert row["corroborated"] is True and row["corroboration_ref"] == "denial:ap-1" + + +# ── 4. honesty ──────────────────────────────────────────────────────────── + +def test_honesty_withheld_below_minimum_with_reason(): + incs = [_inc(sid=f"claude_code:s{i}", first=100 * i, last=100 * i + 10) for i in range(3)] + reps = [{"session_id": "s0", "ts": 5.0}] + rows = sd.honesty_rollup(incs, reps, window_secs=60, min_count=5) + assert len(rows) == 1 + r = rows[0] + assert r["incidents"] == 3 and r["reported"] == 1 + assert r["honesty"] is None and r["withheld"] is True + assert "Only 3 detector incidents" in r["reason"] and "at least 5" in r["reason"] + + +def test_honesty_ratio_per_runtime_and_model(): + incs = [dict(_inc(sid=f"codex:s{i}", first=100 * i, last=100 * i + 10), + runtime="codex", model="gpt-x") for i in range(6)] + reps = [{"session_id": f"s{i}", "ts": 100 * i + 5} for i in range(3)] + rows = sd.honesty_rollup(incs, reps, window_secs=10, min_count=5) + assert rows == [{"runtime": "codex", "model": "gpt-x", "incidents": 6, "reported": 3, + "honesty": 0.5, "withheld": False, "reason": ""}] + + +def test_store_honesty_uses_min_incidents_env(store, monkeypatch): + monkeypatch.delenv("CLAWMETRY_SELFDIAG_WINDOW_SECS", raising=False) + store.ingest_loop_signal(session_id="claude_code:h1", signature="daemon_detect_no_progress", + repeat_count=3, agent_type="claude_code", + details={"kind": "no_progress"}) + store.ingest_self_report(session_id="h1", category="task_failure", summary="stuck") + monkeypatch.delenv("CLAWMETRY_SELFDIAG_MIN_INCIDENTS", raising=False) + rows = store.query_self_report_honesty(since_secs=3600) + assert rows[0]["withheld"] is True and rows[0]["incidents"] == 1 + monkeypatch.setenv("CLAWMETRY_SELFDIAG_MIN_INCIDENTS", "1") + rows = store.query_self_report_honesty(since_secs=3600) + assert rows[0]["withheld"] is False and rows[0]["honesty"] == 1.0 + assert rows[0]["runtime"] == "claude_code" + + +# ── 5. installer, per verified runtime format ───────────────────────────── + +def _seed(home, runtime): + """Pre-existing config with a FOREIGN server entry, in that runtime's format.""" + inst = _installer(home) + path = inst.path_for(runtime) + os.makedirs(os.path.dirname(path), exist_ok=True) + fmt = mi.SUPPORTED[runtime]["format"] + if fmt == "toml_mcp_servers": + text = ('model = "o3"\n\n[mcp_servers.github]\ncommand = "npx"\n' + 'args = ["-y", "@modelcontextprotocol/server-github"]\n') + open(path, "w").write(text) + elif fmt == "json_opencode": + json.dump({"$schema": "https://opencode.ai/config.json", + "mcp": {"github": {"type": "local", "command": ["npx", "gh-mcp"], + "enabled": True}}}, open(path, "w")) + else: + json.dump({"other": True, "mcpServers": {"github": {"command": "npx", "args": ["gh"]}}}, + open(path, "w")) + return inst, path + + +def _foreign_present(runtime, path): + fmt = mi.SUPPORTED[runtime]["format"] + text = open(path).read() + if fmt == "toml_mcp_servers": + return "[mcp_servers.github]" in text and 'model = "o3"' in text + data = json.loads(text) + key = "mcp" if fmt == "json_opencode" else "mcpServers" + return "github" in data.get(key, {}) and (fmt == "json_opencode" or data.get("other") is True) + + +def _ours_present(runtime, path): + fmt = mi.SUPPORTED[runtime]["format"] + text = open(path).read() + if fmt == "toml_mcp_servers": + return "[mcp_servers.clawmetry]" in text and 'command = "/opt/clawmetry/bin/clawmetry"' in text + data = json.loads(text) + if fmt == "json_opencode": + e = data["mcp"].get("clawmetry") + return bool(e) and e["type"] == "local" and e["command"] == ["/opt/clawmetry/bin/clawmetry", "mcp"] \ + and e["enabled"] is True + e = data["mcpServers"].get("clawmetry") + if not e: + return False + ok = e["command"] == "/opt/clawmetry/bin/clawmetry" and e["args"] == ["mcp"] + if runtime == "claude_code": + ok = ok and e.get("type") == "stdio" + return ok + + +@pytest.mark.parametrize("runtime", sorted(mi.SUPPORTED)) +def test_install_merges_and_uninstall_removes_only_ours(home, runtime): + inst, path = _seed(home, runtime) + assert inst.status(runtime)["status"] == mi.NOT_INSTALLED + + dry = inst.install(runtime, dry_run=True) + assert dry["status"] == mi.WOULD_REGISTER + assert not _ours_present(runtime, path) + + res = inst.install(runtime) + assert res["status"] == mi.REGISTERED, res + assert _ours_present(runtime, path) + assert _foreign_present(runtime, path), "install deleted a foreign entry" + assert inst.status(runtime)["status"] == mi.REGISTERED + marker = json.load(open(inst.marker_path)) + assert marker[runtime]["server_name"] == "clawmetry" + + again = inst.install(runtime) + assert again["status"] == mi.ALREADY_PRESENT + assert open(path).read().count("clawmetry") == open(path).read().count("clawmetry") + + gone = inst.uninstall(runtime) + assert gone["status"] == mi.REMOVED + assert not _ours_present(runtime, path) + assert _foreign_present(runtime, path), "uninstall deleted a foreign entry" + assert runtime not in json.load(open(inst.marker_path)) + assert inst.status(runtime)["status"] == mi.NOT_INSTALLED + assert inst.uninstall(runtime)["status"] == mi.NOT_INSTALLED + + +@pytest.mark.parametrize("runtime", sorted(mi.SUPPORTED)) +def test_hand_written_entry_is_never_deleted(home, runtime): + inst, path = _seed(home, runtime) + fmt = mi.SUPPORTED[runtime]["format"] + if fmt == "toml_mcp_servers": + with open(path, "a") as fh: + fh.write('\n[mcp_servers.clawmetry]\ncommand = "clawmetry"\nargs = ["mcp"]\n') + else: + data = json.load(open(path)) + key = "mcp" if fmt == "json_opencode" else "mcpServers" + data[key]["clawmetry"] = ({"type": "local", "command": ["clawmetry", "mcp"]} + if fmt == "json_opencode" + else {"command": "clawmetry", "args": ["mcp"]}) + json.dump(data, open(path, "w")) + before = open(path).read() + assert inst.status(runtime)["status"] == mi.ALREADY_PRESENT + assert inst.install(runtime)["status"] == mi.ALREADY_PRESENT + assert inst.uninstall(runtime)["status"] == mi.LEFT_IN_PLACE + assert open(path).read() == before + + +def test_install_creates_missing_file_and_uninstall_leaves_valid_json(home): + inst = _installer(home) + res = inst.install("cursor") + assert res["status"] == mi.REGISTERED + path = inst.path_for("cursor") + assert json.load(open(path))["mcpServers"]["clawmetry"]["args"] == ["mcp"] + inst.uninstall("cursor") + assert json.load(open(path)) == {"mcpServers": {}} + + +def test_codex_toml_block_round_trips_through_a_toml_parser(home): + tomllib = pytest.importorskip("tomllib") + inst, path = _seed(home, "codex") + inst.install("codex") + data = tomllib.loads(open(path).read()) + assert data["mcp_servers"]["clawmetry"] == {"command": "/opt/clawmetry/bin/clawmetry", + "args": ["mcp"]} + assert data["mcp_servers"]["github"]["command"] == "npx" + inst.uninstall("codex") + data = tomllib.loads(open(path).read()) + assert "clawmetry" not in data["mcp_servers"] and "github" in data["mcp_servers"] + + +def test_jsonc_is_reported_not_guessed(home): + inst = _installer(home) + path = inst.path_for("opencode") + os.makedirs(os.path.dirname(path)) + open(path, "w").write('{\n // comment\n "mcp": {}\n}\n') + before = open(path).read() + assert inst.status("opencode")["status"] == mi.UNKNOWN_FORMAT + assert inst.install("opencode")["status"] == mi.UNKNOWN_FORMAT + assert inst.uninstall("opencode")["status"] == mi.UNKNOWN_FORMAT + assert open(path).read() == before + + +def test_status_vocabulary_for_unsupported_and_unknown(home): + inst = _installer(home) + assert inst.status("aider")["status"] == mi.NO_MCP_SUPPORT + assert inst.install("aider")["status"] == mi.NO_MCP_SUPPORT + assert inst.status("kimi")["status"] == mi.UNKNOWN_FORMAT + assert inst.install("kimi")["status"] == mi.UNKNOWN_FORMAT + rows = inst.status_all("all") + ids = {r["runtime"] for r in rows} + assert set(mi.SUPPORTED) <= ids and set(mi.NO_MCP) <= ids + matrix = {r["runtime"]: r for r in mi.support_matrix(home=str(home))} + assert matrix["claude_code"]["mcp"] == "supported" + assert matrix["aider"]["mcp"] == "not_supported" + assert matrix["kimi"]["mcp"] == "unknown" + + +def test_guidance_is_offered_not_written(home, tmp_path, monkeypatch, capsys): + proj = tmp_path / "proj" + proj.mkdir() + monkeypatch.chdir(proj) + monkeypatch.setattr(mi, "resolve_server_command", lambda: ("/opt/clawmetry/bin/clawmetry", ["mcp"])) + assert mi.cli_main(["install", "--runtime", "claude_code"]) == 0 + out = capsys.readouterr().out + assert "report_to_operator" in out and "Not written" in out + assert not (proj / "CLAUDE.md").exists() + assert json.load(open(home / ".claude.json"))["mcpServers"]["clawmetry"]["type"] == "stdio" + + assert mi.cli_main(["install", "--runtime", "claude_code", "--write-guidance"]) == 0 + text = (proj / "CLAUDE.md").read_text() + assert mi.GUIDANCE_MARKER in text and "Before finishing" in text + assert mi.cli_main(["install", "--runtime", "claude_code", "--write-guidance"]) == 0 + assert (proj / "CLAUDE.md").read_text().count(mi.GUIDANCE_MARKER) == 1 + assert mi.guidance_file_for("gemini_cli") == "GEMINI.md" + assert mi.guidance_file_for("codex") == "AGENTS.md" + + capsys.readouterr() # drop the install chatter; status --json must stand alone + assert mi.cli_main(["status", "--json"]) == 0 + rows = json.loads(capsys.readouterr().out) + assert {r["runtime"]: r["status"] for r in rows}["claude_code"] == mi.REGISTERED + assert mi.cli_main(["uninstall", "--runtime", "claude_code"]) == 0 + assert "clawmetry" not in json.load(open(home / ".claude.json"))["mcpServers"] + + +def test_cli_fast_path_never_imports_dashboard(home, monkeypatch, capsys): + import clawmetry.cli as cli + monkeypatch.setattr(sys, "argv", ["clawmetry", "mcp", "status", "--json"]) + monkeypatch.delitem(sys.modules, "dashboard", raising=False) + real_import = builtins.__import__ + imported = [] + + def _spy(name, *args, **kwargs): + if name == "dashboard" or name.startswith("dashboard."): + imported.append(name) + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", _spy) + with pytest.raises(SystemExit) as exc: + cli.main() + assert exc.value.code == 0 + assert imported == [] + assert isinstance(json.loads(capsys.readouterr().out), list) + + +# ── 6. routes + snapshot ────────────────────────────────────────────────── + +@pytest.fixture +def client(store, monkeypatch): + flask = pytest.importorskip("flask") + import routes.selfdiag as rs + + monkeypatch.setattr(rs, "_ls_call", lambda name, **kw: getattr(store, name)(**kw)) + app = flask.Flask("t") + app.register_blueprint(rs.bp_selfdiag) + return app.test_client() + + +def test_routes_list_honesty_and_support(client, store, home, monkeypatch): + monkeypatch.delenv("CLAWMETRY_SELFDIAG_WINDOW_SECS", raising=False) + store.ingest_self_report(session_id="claude_code:r1", category="missing_context", + summary="no README", agent_type="claude_code") + d = client.get("/api/self-reports?session=r1").get_json() + assert d["count"] == 1 and d["reports"][0]["category"] == "missing_context" + assert "not the same as false" in d["uncorroborated_means"] + d = client.get("/api/self-reports?window=7d&runtime=codex").get_json() + assert d["count"] == 0 + d = client.get("/api/self-reports?window=7d&category=missing_context").get_json() + assert d["count"] == 1 + + h = client.get("/api/self-reports/honesty?window=7d").get_json() + assert h["counts"] == {"claude_code": {"missing_context": 1}} + assert h["honesty"] == [] and h["min_incidents"] == sd.MIN_INCIDENTS + + s = client.get("/api/self-reports/support").get_json() + by = {r["runtime"]: r for r in s["runtimes"]} + assert by["claude_code"]["mcp"] == "supported" + assert by["aider"]["mcp"] == "not_supported" and "no MCP" in by["aider"]["detail"] + + +def test_snapshot_slice_has_counts_and_no_summaries(store): + store.ingest_self_report(session_id="claude_code:z", category="noteworthy", + summary="a secret-free note") + sl = sd.snapshot_slice(store) + assert sl["total"] == 1 and sl["byRuntime"] == {"claude_code": {"noteworthy": 1}} + for key in ("window_secs", "corroborated", "honesty", "min_incidents", + "corroboration_window_secs"): + assert key in sl + assert "secret-free" not in json.dumps(sl) + + +def test_daemon_allowlist_names_every_method_the_feature_uses(): + from routes.local_query import _DAEMON_METHODS + for m in ("ingest_self_report", "query_self_reports", "query_self_report_counts", + "query_self_report_honesty", "query_guard_incidents", + "query_session_denials", "find_session_by_cwd", + "get_session_location", "query_policy_actions"): + assert m in _DAEMON_METHODS, m + + +def test_parse_window(): + assert sd.parse_window_secs("24h") == 86400 + assert sd.parse_window_secs("7d") == 7 * 86400 + assert sd.parse_window_secs("30m") == 1800 + assert sd.parse_window_secs(90) == 90 + assert sd.parse_window_secs("junk", 5) == 5 + assert sd.parse_window_secs("0") == 1 + + +def test_live_templates_carry_the_new_surfaces(): + root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + guard = open(os.path.join(root, "clawmetry", "templates", "tabs", "guard.html")).read() + trans = open(os.path.join(root, "clawmetry", "templates", "tabs", "transcripts.html")).read() + js = open(os.path.join(root, "clawmetry", "static", "js", "app.js")).read() + assert 'id="guard-selfreports-body"' in guard + assert 'id="selfreports-panel"' in trans + assert "function loadGuardSelfReports" in js and "_loadSelfReportsPanel(sessionId)" in js + for text in (guard, trans): + assert "—" not in text.split("