audit fixes: checked types + curated lint; install-mcp drift guard; strict policy keys; transport credential guard

.github/workflows/tests.yml

@@ -17,6 +17,7 @@ jobs:
           enable-cache: true
       - run: uv sync --dev          # no anthropic extra: keeps CI keyless, matches design
       - run: uv run ruff check .
+      - run: uv run mypy            # checked annotations (audit fix #1); config in pyproject
   pytest:
     runs-on: ubuntu-latest
     strategy:

.gitignore

@@ -29,3 +29,4 @@ reports/
 # Ad-hoc experiment harnesses (e.g. the fidelity probe) — not part of the lib
 scratch_fidelity/
 .codegraph
+.coverage

.runechoguardignore

@@ -81,3 +81,6 @@ DEFAULT_PROBE_REPORT
 # probes.py cross_call_overlap docstring: "When `idf` is supplied (from `token_idf`)" —
 # "supplied (from" reads as a bare call to the guard's pattern match. Not code.
 supplied
+# proxy.py _emit_audit: `audit = self.audit` narrowed local (mypy baseline, audit
+# fix #1), called bare — same local-variable false-positive class as real/orig_send.
+audit

USAGE.md

@@ -137,9 +137,11 @@ install invocation. It honors `$CLAUDE_CONFIG` if your config isn't at
 `~/.claude.json`. Start with one high-win, read-only server (e.g. `runecho`) and
 confirm it works before wrapping more.
 
-One sharp edge worth knowing: a re-wrap rebuilds the entry from the **stashed
-original**, so any hand-edit you made to the *wrapped* entry (say, an `env.PATH`
-pin) is dropped — apply such edits to the stash original too.
+Hand-edits to a wrapped entry (say, an `env.PATH` pin) survive re-installs: the
+re-wrap rebuilds `command`/`args` from the stashed original but keeps every other
+live key, and reports what it carried (`kept hand-edited key(s) …`). One edge
+remains: `uninstall-mcp` restores the **pre-terse original**, which does not carry
+such edits — put them on the original entry too if they must survive an uninstall.
 
 Claude Code has three MCP scopes, and `--scope` targets any of them (default
 `user`, i.e. today's behavior):

pyproject.toml

@@ -14,11 +14,35 @@ dependencies = [
 terse = "terse.cli:main"
 
 [dependency-groups]
-dev = ["pytest", "ruff"]
+dev = ["pytest", "ruff", "mypy"]
 
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/terse"]
+
+[tool.ruff.lint]
+# Curated beyond ruff's minimal default (E4/E7/E9/F): import order, real bug
+# patterns, modern-syntax upgrades, comprehension and simplification wins. Style-only
+# packs are deliberately left out — the gate is for defects, not formatting taste.
+select = ["E4", "E7", "E9", "F", "I", "B", "UP", "C4", "SIM", "RET", "PLE"]
+ignore = [
+    "SIM108",  # ternary-vs-if is a readability call, not a defect
+    "SIM105",  # contextlib.suppress: try/except-pass is clearer at fail-open sites
+]
+
+[tool.mypy]
+# Lenient baseline (audit fix #1): the goal is to make the existing annotations a
+# CHECKED contract, not to win strictness awards on day one. Ratchet later.
+python_version = "3.11"
+files = ["src/terse"]
+warn_unused_ignores = true
+warn_redundant_casts = true
+no_implicit_optional = true
+check_untyped_defs = true
+
+[[tool.mypy.overrides]]
+module = "tiktoken"
+ignore_missing_imports = true

src/terse/capture.py

@@ -21,8 +21,9 @@
 from typing import Any
 
 from ._secure_io import append_restricted, write_restricted
-
-from .transforms import _uniform_dict_list  # the one canonical "what tabularize folds" rule
+from .transforms import (
+    _uniform_dict_list,  # the one canonical "what tabularize folds" rule
+)
 
 # Shape buckets. classify_shape returns one of these.
 PRETTY_JSON = "pretty-json"

src/terse/cli.py

@@ -15,15 +15,22 @@
 
 import argparse
 import json
+import json as _json
 import re
 import sys
+from datetime import UTC
 from pathlib import Path
 
-import json as _json
-
 from . import transforms
 from ._secure_io import write_restricted
-from .capture import capture_payload, classify_shape, coverage, extract_records, load_corpus
+from .capture import (
+    capture_payload,
+    classify_shape,
+    coverage,
+    extract_records,
+    load_corpus,
+)
+from .html_report import build_html_report
 from .measure import cross_tokenizer_savings, measure_corpus
 from .probes import (
     cross_call_overlap,
@@ -32,7 +39,6 @@
     server_of_tool,
     value_redundancy,
 )
-from .html_report import build_html_report
 from .report import (
     build_cross_server_probe_report,
     build_probe_report,
@@ -81,14 +87,14 @@ def _record_and_print_trend(history_path: str, rows: list, label: str) -> None:
     in this command that reads the real clock (principle #31: inject nondeterminism
     at the edge, not inside the pure summarize_run/build_trend_report/
     trend_sparkline_lines core)."""
-    from datetime import datetime, timezone
+    from datetime import datetime
 
     from .history import append_run, load_history, summarize_run
     from .report import build_trend_report
     from .terminal_report import trend_sparkline_lines
 
     path = Path(history_path)
-    ts = datetime.now(timezone.utc).isoformat(timespec="seconds")
+    ts = datetime.now(UTC).isoformat(timespec="seconds")
     run = summarize_run(rows, ts, label=label)
     all_runs = load_history(path) + [run]
     append_run(path, run)
@@ -302,7 +308,7 @@ def _cmd_probe(args: argparse.Namespace) -> int:
         by_tool.setdefault(env["tool"], []).append(env)
     for tool, envs in by_tool.items():
         envs = sorted(envs, key=lambda e: e.get("sha", ""))
-        for prev, curr in zip(envs, envs[1:]):
+        for prev, curr in zip(envs, envs[1:], strict=False):  # sliding pairs
             res = cross_call_overlap(prev["raw"], curr["raw"])
             if res.get("available"):
                 overlap_rows.append({"tool": tool, "prev_sha": prev.get("sha", "?"),
@@ -604,6 +610,10 @@ def _cmd_install_mcp(args: argparse.Namespace) -> int:
         print(f"{tag} {c['server']}:")
         print(f"    before: {_short_cmd(c['before'])}")
         print(f"    after:  {_short_cmd(c['after'])}")
+        if c.get("preserved"):
+            print(f"    kept hand-edited key(s) from the live entry: "
+                  f"{', '.join(c['preserved'])} (note: uninstall restores the "
+                  f"pre-terse original, which does NOT carry them)")
     print(f"config: {res['config']}  scope: {res['scope']}  policy: {res['policy']}")
     if res.get("capture_dir"):
         print(f"capture: raw tool results → {res['capture_dir']}")

src/terse/dropeval.py

@@ -26,15 +26,16 @@
 
 import json
 import urllib.request
+from collections.abc import Callable
 from dataclasses import dataclass, field
-from typing import Any, Callable, Optional
+from typing import Any
 
-from . import capture
-from . import fluency
+from . import capture, fluency
 from . import lossy as lossy_mod
 from . import policy as policy_mod
 from .proxy import TERSE_PRIMER
 
+
 # --------------------------------------------------------------------------- #
 # Tool-loop answerer protocol — the existing fluency.Answerer (system, user) -> str
 # can't express a tool call; this is a provider-neutral running conversation instead.
@@ -94,7 +95,7 @@ def _staged_apply(obj: Any, rule: Any, tool: str) -> tuple[policy_mod.Applied, d
 
 def _questions_and_staging(
     obj: Any, rule: Any, tool: str
-) -> tuple[list[DropQuestion], Optional[policy_mod.Applied], Optional[dict[str, Any]]]:
+) -> tuple[list[DropQuestion], policy_mod.Applied | None, dict[str, Any] | None]:
     """Shared core of `gen_drop_questions`: generates the (recall, precision) question
     pair AND returns the `(applied, staging)` that `_staged_apply` computed along the
     way, so `run_drop_payload` can reuse them instead of a second `policy.apply()` pass
@@ -190,7 +191,7 @@ def gen_drop_questions(obj: Any, rule: Any, tool: str) -> list[DropQuestion]:
 # --------------------------------------------------------------------------- #
 # The 2-turn tool-loop driver — mirrors the real proxy's retrieve protocol exactly
 # --------------------------------------------------------------------------- #
-def _miss_text(handle: str) -> str:
+def _miss_text(handle: Any) -> str:
     """The exact miss string `proxy.Interceptor.answer_retrieve` emits for an unresolved
     handle, copied verbatim so this eval's miss-handling matches production behavior — a
     model that has learned to recover from a real miss must see the same words here."""

src/terse/fluency.py

@@ -34,15 +34,23 @@
 import json
 import re
 import urllib.request
-from urllib.parse import urlsplit
+from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Any, Callable
+from typing import Any
+from urllib.parse import urlsplit
 
 from . import text_diff
 from .capture import LONG_TEXT, OTHER, classify_shape, extract_records
 from .tokenize import count_cl100k
-from .transforms import (compress, compress_structure, dict_encode, diff_wire,
-                         has_terse_marker, minify, _uniform_dict_list)
+from .transforms import (
+    _uniform_dict_list,
+    compress,
+    compress_structure,
+    dict_encode,
+    diff_wire,
+    has_terse_marker,
+    minify,
+)
 
 # Loopback hosts where cleartext http is safe (never leaves the machine), so a Bearer
 # key over http to one of these is fine — a local LiteLLM/CCR gateway is a common setup.
@@ -187,7 +195,7 @@ def _records_of(lst: Any) -> list[str] | None:
         # dict-map of parent records -> the first parent (map order) with a child record list
         if isinstance(v, dict) and v and all(isinstance(pv, dict) for pv in v.values()):
             for pkey, parent in v.items():
-                for ck, cv in parent.items():
+                for _ck, cv in parent.items():
                     if _records_of(cv):
                         return f"{k}[{json.dumps(pkey, ensure_ascii=False)}]", cv, _intersection_cols(cv)
         # a NON-uniform top-level list of dicts (uniform ones are extract_records' domain)
@@ -561,7 +569,7 @@ def _iter_consecutive_pairs(envelopes: list[dict]):
         by_tool.setdefault(env["tool"], []).append(env)
     for tool, envs in by_tool.items():
         envs = sorted(envs, key=lambda e: e.get("sha", ""))
-        for prev_env, curr_env in zip(envs, envs[1:]):
+        for prev_env, curr_env in zip(envs, envs[1:], strict=False):  # sliding pairs
             yield tool, curr_env.get("sha", "?"), prev_env["raw"], curr_env["raw"]
 
 
@@ -627,7 +635,7 @@ def build_chain_windows(envelopes: list[dict], max_depth: int = 5,
             except (json.JSONDecodeError, TypeError):
                 seq.append((e.get("sha", "?"), None))
         cur: list[tuple] = []
-        for (psha, p), (csha, c) in zip(seq, seq[1:]):
+        for (psha, p), (csha, c) in zip(seq, seq[1:], strict=False):  # sliding pairs
             ok = (p is not None and c is not None
                   and diff_wire(p, c, tool) is not None)
             if ok:
@@ -643,9 +651,9 @@ def build_chain_windows(envelopes: list[dict], max_depth: int = 5,
     windows: list[tuple] = []
     for depth in range(1, max_depth + 1):
         # every eligible start per tool, spread over its runs; then round-robin
-        candidates: dict[str, list[tuple]] = {}
+        candidates: dict[str, list[list[tuple]]] = {}
         for tool, tool_runs in sorted(runs.items()):
-            starts: list[tuple] = []
+            starts: list[list[tuple]] = []
             for run in tool_runs:
                 for i in range(0, len(run) - depth, depth):   # non-overlapping
                     window = run[i:i + depth + 1]
@@ -656,7 +664,7 @@ def build_chain_windows(envelopes: list[dict], max_depth: int = 5,
         picked = 0
         idx = 0
         while picked < per_depth_cap and candidates:
-            for tool in sorted(list(candidates)):
+            for tool in sorted(candidates):
                 starts = candidates[tool]
                 if idx >= len(starts):
                     del candidates[tool]
@@ -682,7 +690,7 @@ def run_chain_payload(objs: list, answerer: Answerer, tool: str = "",
     if not questions:
         return []
     wires: list[str] = []
-    for prev, curr in zip(objs, objs[1:]):
+    for prev, curr in zip(objs, objs[1:], strict=False):  # sliding pairs
         wire = diff_wire(prev, curr, tool)
         if wire is None:
             return []

src/terse/html_report.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import html as _html
+from collections.abc import Sequence
 from typing import Any
 
 from .report import _ci, _form_stats, _pct, _sum, _worst_case_gap
@@ -147,7 +148,7 @@ def diverging_bar_chart(items: list[tuple[str, float]], unit: str = "%") -> str:
     return "".join(out)
 
 
-def stacked_bar_chart(items: list[tuple[str, list[float]]],
+def stacked_bar_chart(items: Sequence[tuple[str, Sequence[float]]],
                        series_labels: tuple[str, ...],
                        series_colors: tuple[str, ...] = ("var(--series-1)", "var(--series-2)", "var(--series-3)")
                        ) -> str:
@@ -187,7 +188,7 @@ def sx(v: float) -> float:
     zero_x = sx(0.0)
     out = [_svg_open(width, height)]
     lx, ly = label_w, 14
-    for name, color in zip(series_labels, series_colors):
+    for name, color in zip(series_labels, series_colors, strict=True):
         out.append(f'<rect x="{lx}" y="{ly - 9}" width="10" height="10" rx="2" fill="{color}"/>')
         out.append(f'<text x="{lx + 15}" y="{ly}" class="axis-label">{_esc(name)}</text>')
         lx += 18 + 9 * len(name)

src/terse/install_mcp.py

@@ -149,7 +149,9 @@ def wrap(config: dict, stash: dict, server: str, policy: str,
     """Wrap `server`'s entry with the terse proxy. Idempotent: if already managed
     (present in stash), re-wrap from the stashed original so policy/cmd updates
     apply cleanly without nesting proxies. Preserves all non-command/args (and, for a
-    URL entry, non-url/headers) keys (env, cwd, type, …) of the original entry. With
+    URL entry, non-url/headers) keys (env, cwd, type, …) of the original entry — and,
+    on a re-wrap, hand-edits made to those keys on the LIVE wrapped entry win over the
+    stashed original's values (the drift guard below). With
     `capture_dir`, the wrapped proxy tees raw tool results into that corpus for later
     measurement (#32). `diff` is tri-state: None writes no flag (the entry inherits
     the proxy default — ON since #75), True/False bake `--diff`/`--no-diff` into the
@@ -161,10 +163,11 @@ def wrap(config: dict, stash: dict, server: str, policy: str,
     forwarded as repeated `--header k=v` (see `transport.HttpTransport`). Anything with
     neither key is not a valid MCP server entry and can't be wrapped."""
     servers = config.setdefault("mcpServers", {})
+    live = servers.get(server)
     if server in stash:
         original = stash[server]
-    elif server in servers:
-        original = servers[server]
+    elif live is not None:
+        original = live
         stash[server] = original
     else:
         raise KeyError(server)
@@ -199,6 +202,18 @@ def wrap(config: dict, stash: dict, server: str, policy: str,
                     if k not in ("command", "args", "url", "headers")}
         new_entry["command"] = terse_cmd[0]
         new_entry["args"] = [*terse_cmd[1:], "proxy", *proxy_opts, *header_opts, "--", orig_url]
+
+    if live is not None and live is not original:
+        # Drift guard: a re-wrap used to rebuild the entry purely from the stashed
+        # original, silently reverting any hand-edit made to the WRAPPED entry since
+        # the last install (a scoped env.PATH pin, a cwd) — that reverted codegraph's
+        # node pin in production on 2026-07-13. command/args are terse-owned and always
+        # rebuilt (flags must reflect this invocation); url/headers never appear on a
+        # wrapped entry (they're folded into args), so a drifted live copy of them must
+        # not be resurrected either. Everything else on the live entry wins.
+        for k, v in live.items():
+            if k not in ("command", "args", "url", "headers"):
+                new_entry[k] = v
     servers[server] = new_entry
     return config, stash
 
@@ -306,10 +321,19 @@ def do_install(servers: list[str], policy: str, *, dry_run: bool = False,
     changes = []
     for s in servers:
         before = (node.get("mcpServers") or {}).get(s)
+        # Hand-edits = non-terse-owned keys on the live WRAPPED entry that differ from
+        # the stashed original — the drift guard in wrap() carries them forward; name
+        # them in the result so the operator sees what survived (and what to move into
+        # the original entry if it should also survive an uninstall).
+        preserved = sorted(
+            k for k in (before or {})
+            if k not in ("command", "args", "url", "headers")
+            and s in stash and (before or {}).get(k) != stash[s].get(k)
+        )
         wrap(node, stash, s, policy_abs, terse_cmd, capture_dir=capture_abs,
              diff=diff, diff_keyframe_interval=diff_keyframe_interval)
         changes.append({"server": s, "before": before,
-                        "after": node["mcpServers"][s]})
+                        "after": node["mcpServers"][s], "preserved": preserved})
 
     result = {"config": str(target.cfg), "scope": scope, "policy": policy_abs,
               "available": available, "changes": changes, "dry_run": dry_run,