feat(B): superseded-context detector + ts_stale_context (agent-agnostic)

src/token_savior/discover/superseded.py

@@ -0,0 +1,71 @@
+"""Detect superseded / redundant tool results in a transcript (feature B).
+
+Agent-agnostic: consumes the same Events ``transcript_scanner`` yields for any
+harness. This module is PURE — it decides WHAT is stale, not how to act on it.
+The findings feed two delivery paths: a callable MCP tool (works on every
+agent) and, on Claude Code only, the PreCompact hook. Nothing here is
+harness-specific.
+"""
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+from token_savior.discover.transcript_scanner import Event
+
+_READ_TOOLS = {"Read", "NotebookRead"}
+_EDIT_TOOLS = {
+    "Edit", "Write", "MultiEdit", "NotebookEdit",
+    "replace_symbol_source", "edit_lines_in_symbol",
+    "insert_near_symbol", "move_symbol", "add_field_to_model",
+}
+
+
+def _base(tool: str) -> str:
+    """Strip an ``mcp__<server>__`` prefix so tool names match by their leaf."""
+    return tool.rsplit("__", 1)[-1] if "__" in tool else tool
+
+
+def find_superseded(events: Iterable[Event]) -> list[dict]:
+    """References to tool results made obsolete LATER in the same session:
+
+    - ``reread``          a file read, then read again (first read redundant)
+    - ``read_then_edit``  a file read, then edited (the read is now stale)
+    - ``rerun``           a Bash command run, then re-run (first output stale)
+
+    Each finding names the stale event (``stale_index``) and the fresh one
+    (``fresh_index``) so a caller can drop the stale content or replace it with
+    a pointer. Order-preserving, deterministic, no model.
+    """
+    findings: list[dict] = []
+    last_read: dict[tuple[str, str], int] = {}
+    last_bash: dict[tuple[str, str], int] = {}
+    for i, ev in enumerate(events):
+        base = _base(ev.tool_name)
+        sess = ev.session_id
+        if base in _READ_TOOLS:
+            fp = ev.args.get("file_path")
+            if fp:
+                key = (sess, fp)
+                if key in last_read:
+                    findings.append({"reason": "reread", "tool": base, "target": fp,
+                                     "session": sess, "stale_index": last_read[key],
+                                     "fresh_index": i})
+                last_read[key] = i
+        elif base in _EDIT_TOOLS:
+            fp = ev.args.get("file_path")
+            if fp:
+                key = (sess, fp)
+                if key in last_read:
+                    findings.append({"reason": "read_then_edit", "tool": base, "target": fp,
+                                     "session": sess, "stale_index": last_read.pop(key),
+                                     "fresh_index": i})
+        elif base == "Bash":
+            cmd = ev.args.get("command")
+            if cmd:
+                key = (sess, cmd)
+                if key in last_bash:
+                    findings.append({"reason": "rerun", "tool": "Bash", "target": cmd[:80],
+                                     "session": sess, "stale_index": last_bash[key],
+                                     "fresh_index": i})
+                last_bash[key] = i
+    return findings

src/token_savior/server_handlers/discover.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import json
+from datetime import UTC
 from typing import Any
 
 from token_savior._compat import TextContent, types
@@ -27,6 +28,8 @@
     discover_adoption as _discover_adoption,
 )
 from token_savior.discover.patterns import AdoptionReport
+from token_savior.discover.superseded import find_superseded
+from token_savior.discover.transcript_scanner import iter_events, transcript_root
 
 
 def _fmt_table(findings: list) -> str:
@@ -124,6 +127,58 @@ def _hm_ts_discover(arguments: dict[str, Any]) -> list[types.TextContent]:
     return [TextContent(type="text", text=_fmt_table(findings))]
 
 
+def _fmt_stale(findings: list[dict]) -> str:
+    if not findings:
+        return "No superseded context found in window (nothing to drop)."
+    label = {
+        "reread": "read again — drop the earlier read",
+        "read_then_edit": "edited after read — the read is stale",
+        "rerun": "command re-run — earlier output stale",
+    }
+    rows = [
+        "# ts_stale_context — tool results made obsolete later in the session",
+        "",
+        "| reason | target | session | stale→fresh |",
+        "| ------ | ------ | ------- | ----------- |",
+    ]
+    for f in findings:
+        sid = (f["session"][:8] + "…") if len(f["session"]) > 8 else f["session"]
+        tgt = f["target"]
+        tgt = (tgt[:48] + "…") if len(tgt) > 48 else tgt
+        rows.append(f"| {label.get(f['reason'], f['reason'])} | `{tgt}` | {sid} "
+                    f"| #{f['stale_index']}→#{f['fresh_index']} |")
+    rows += [
+        "",
+        (f"_{len(findings)} stale reference(s). Drop the stale content or "
+         "refetch the fresh copy. On Claude Code the PreCompact hook prunes "
+         "these automatically; elsewhere this is advisory._"),
+    ]
+    return "\n".join(rows)
+
+
+def _hm_ts_stale_context(arguments: dict[str, Any]) -> list[types.TextContent]:
+    from datetime import datetime, timedelta
+
+    since_days = int(arguments.get("since_days", 1) or 1)
+    project = arguments.get("project") or None
+    fmt = (arguments.get("format") or "table").lower()
+    limit = arguments.get("limit")
+    try:
+        since = datetime.now(UTC) - timedelta(days=since_days)
+        events = iter_events(transcript_root(), since=since, project=project)
+        findings = find_superseded(events)
+    except Exception as exc:  # defensive — never break dispatch
+        return [TextContent(type="text", text=f"ts_stale_context error: {exc}")]
+    if isinstance(limit, int) and limit > 0:
+        findings = findings[:limit]
+    if fmt == "json":
+        return [TextContent(type="text", text=json.dumps(
+            {"since_days": since_days, "project": project,
+             "count": len(findings), "stale": findings}, indent=2))]
+    return [TextContent(type="text", text=_fmt_stale(findings))]
+
+
 HANDLERS: dict[str, Any] = {
     "ts_discover": _hm_ts_discover,
+    "ts_stale_context": _hm_ts_stale_context,
 }

src/token_savior/tool_schemas.py

@@ -692,6 +692,38 @@
             },
         },
     },
+    "ts_stale_context": {
+        "description": (
+            "Scan the recent session transcript(s) and list tool results made "
+            "obsolete later in the SAME session — a file read then re-read, a "
+            "file read then edited, a Bash command re-run. Returns the stale "
+            "refs so you can drop that content or refetch the fresh copy, "
+            "keeping the context window lean. Agent-agnostic, read-only, "
+            "PII-safe. format='table' (default) or 'json'."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "since_days": {
+                    "type": "integer",
+                    "description": "Only consider events newer than now - since_days (default 1 = the current session).",
+                },
+                "project": {
+                    "type": "string",
+                    "description": "Filter to project dirs whose name contains this substring (omit = all recent).",
+                },
+                "format": {
+                    "type": "string",
+                    "enum": ["table", "json"],
+                    "description": "Output format. 'table' (default) or 'json'.",
+                },
+                "limit": {
+                    "type": "integer",
+                    "description": "Cap the number of stale refs returned (default unlimited).",
+                },
+            },
+        },
+    },
     # ── Stats (unified) ───────────────────────────────────────────────────
     "get_stats": {
         "description": (

tests/maison/test_01_tous_les_outils.py

@@ -152,6 +152,7 @@
     # --- meta ----------------------------------------------------------
     "ts_search": {"query": "trouver un symbole"},
     "ts_discover": {},
+    "ts_stale_context": {},
     "ts_execute": {"script": "return 1 + 1;"},
 }
 

tests/test_superseded.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from token_savior.discover.superseded import find_superseded
+from token_savior.discover.transcript_scanner import Event
+
+
+def _ev(tool, session="s1", **args):
+    return Event(ts=None, tool_name=tool, args=args, session_id=session, project="p")
+
+
+def test_reread_is_flagged():
+    evs = [_ev("Read", file_path="a.py"), _ev("Bash", command="ls"), _ev("Read", file_path="a.py")]
+    f = find_superseded(evs)
+    assert any(x["reason"] == "reread" and x["target"] == "a.py" for x in f)
+
+
+def test_read_then_edit_is_flagged():
+    evs = [_ev("Read", file_path="a.py"), _ev("Edit", file_path="a.py")]
+    assert any(x["reason"] == "read_then_edit" for x in find_superseded(evs))
+
+
+def test_duplicate_bash_is_flagged():
+    evs = [_ev("Bash", command="pytest -q"), _ev("Bash", command="pytest -q")]
+    assert any(x["reason"] == "rerun" for x in find_superseded(evs))
+
+
+def test_mcp_prefixed_edit_matches():
+    evs = [_ev("Read", file_path="a.py"),
+           _ev("mcp__token-savior__replace_symbol_source", file_path="a.py")]
+    assert any(x["reason"] == "read_then_edit" for x in find_superseded(evs))
+
+
+def test_distinct_files_not_flagged():
+    evs = [_ev("Read", file_path="a.py"), _ev("Read", file_path="b.py")]
+    assert find_superseded(evs) == []
+
+
+def test_cross_session_not_a_reread():
+    evs = [_ev("Read", file_path="a.py", session="s1"),
+           _ev("Read", file_path="a.py", session="s2")]
+    assert find_superseded(evs) == []
+
+
+def test_deterministic():
+    evs = [_ev("Read", file_path="a.py"), _ev("Read", file_path="a.py"),
+           _ev("Bash", command="ls"), _ev("Bash", command="ls")]
+    assert find_superseded(evs) == find_superseded(list(evs))
+
+
+def test_fmt_stale_and_handler_smoke():
+    """ts_stale_context: formatting + the handler returns valid TextContent
+    over real transcripts without crashing (agent-agnostic delivery)."""
+    import json as _j
+
+    from token_savior.server_handlers.discover import _fmt_stale, _hm_ts_stale_context
+    md = _fmt_stale([{"reason": "reread", "tool": "Read", "target": "a.py",
+                      "session": "sess1234abcd", "stale_index": 0, "fresh_index": 3}])
+    assert "a.py" in md and "#0→#3" in md
+    assert _fmt_stale([]).startswith("No superseded")
+    out = _hm_ts_stale_context({"since_days": 1, "format": "json"})
+    assert out[0].type == "text"
+    payload = _j.loads(out[0].text)
+    assert "stale" in payload and "count" in payload

tests/test_tool_schemas.py

@@ -84,7 +84,9 @@ def test_tool_count(self):
         #   collapsed into one round-trip; whitelisted via ALLOWED_TOOLS) = 68.
         # +1 ts_discover (v3.3 F4 — transcript scanner for missed TS
         #   opportunities; META handler, read-only on ~/.claude/projects) = 69.
-        assert len(TOOL_SCHEMAS) == 69, f"Expected 69 tools, got {len(TOOL_SCHEMAS)}"
+        # +1 ts_stale_context (feature B — superseded-context detector exposed
+        #   as an agent-agnostic MCP tool; META handler, read-only) = 70.
+        assert len(TOOL_SCHEMAS) == 70, f"Expected 70 tools, got {len(TOOL_SCHEMAS)}"
 
     def test_server_tools_match_schemas(self):
         from token_savior.server import TOOLS