feat(transcript): noise tagging + llm review-relevance grading

src/vouch/capture.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import json
+import os
 import subprocess
 import time
 from dataclasses import dataclass
@@ -38,7 +39,15 @@ class CaptureConfig:
 
 
 def load_config(store: KBStore) -> CaptureConfig:
-    """Read ``capture:`` from config.yaml; fall back to defaults."""
+    """Read ``capture:`` from config.yaml; fall back to defaults.
+
+    ``VOUCH_CAPTURE_DISABLE=1`` overrides to disabled: vouch's own LLM
+    subprocesses (compile, session-split, transcript grading) set it so the
+    agent session they spawn does not capture itself back into the KB as a
+    fresh pending summary on every run.
+    """
+    if os.environ.get("VOUCH_CAPTURE_DISABLE") == "1":
+        return CaptureConfig(enabled=False)
     try:
         loaded = yaml.safe_load(store.config_path.read_text(encoding="utf-8"))
     except (OSError, UnicodeDecodeError, yaml.YAMLError):

src/vouch/cli.py

@@ -2228,13 +2228,26 @@ def session_list_cmd(as_json: bool) -> None:
     default=None,
     help="Pin the transcript source instead of auto-detecting.",
 )
-def session_transcript_cmd(session_id: str, agent: str | None) -> None:
+@click.option(
+    "--grade", is_flag=True,
+    help="Annotate dialog with cached LLM review-relevance grades.",
+)
+@click.option(
+    "--regrade", is_flag=True,
+    help="Re-run the LLM grading even when a cached grade exists.",
+)
+def session_transcript_cmd(
+    session_id: str, agent: str | None, grade: bool, regrade: bool
+) -> None:
     """Emit the normalized transcript of a captured session as JSON."""
     from . import transcript
 
     store = _load_store()
     with _cli_errors():
-        _emit_json(transcript.load_transcript(store, session_id, agent=agent))
+        _emit_json(transcript.load_transcript(
+            store, session_id, agent=agent,
+            grade=grade or regrade, regrade=regrade,
+        ))
 
 
 @session.command("summarize")

src/vouch/index_db.py

@@ -14,7 +14,7 @@
 import datetime as _dt
 import json
 import sqlite3
-from collections.abc import Iterable
+from collections.abc import Iterable, Mapping
 from contextlib import contextmanager, suppress
 from pathlib import Path
 
@@ -75,6 +75,13 @@
     detected_at     TEXT NOT NULL
 );
 
+CREATE TABLE IF NOT EXISTS transcript_grades (
+    session_id      TEXT PRIMARY KEY,
+    content_hash    TEXT NOT NULL,
+    grades          TEXT NOT NULL,
+    graded_at       TEXT NOT NULL
+);
+
 CREATE TABLE IF NOT EXISTS prov_edges (
     src_id          TEXT NOT NULL,
     dst_id          TEXT NOT NULL,
@@ -329,6 +336,51 @@ def stats(kb_dir: Path) -> dict[str, int]:
         }
 
 
+# --- transcript grade cache -------------------------------------------------
+#
+# LLM review-relevance grades for a session transcript. Derived cache like
+# everything else in state.db: keyed by a hash of the raw session file so a
+# session that grew since grading re-grades, and losing the DB just means
+# paying for the grading again.
+
+
+def get_transcript_grades(
+    kb_dir: Path, session_id: str, content_hash: str
+) -> dict[str, object] | None:
+    """The cached grades payload for (session, exact file content), or None."""
+    with open_db(kb_dir) as conn:
+        row = conn.execute(
+            "SELECT content_hash, grades, graded_at FROM transcript_grades "
+            "WHERE session_id = ?",
+            (session_id,),
+        ).fetchone()
+    if row is None or row[0] != content_hash:
+        return None
+    try:
+        grades = json.loads(row[1])
+    except (TypeError, ValueError):
+        return None
+    if not isinstance(grades, dict):
+        return None
+    return {"grades": grades, "graded_at": row[2]}
+
+
+def put_transcript_grades(
+    kb_dir: Path,
+    session_id: str,
+    *,
+    content_hash: str,
+    grades: Mapping[str, object],
+    graded_at: str,
+) -> None:
+    with open_db(kb_dir) as conn:
+        conn.execute(
+            "INSERT OR REPLACE INTO transcript_grades "
+            "(session_id, content_hash, grades, graded_at) VALUES (?, ?, ?, ?)",
+            (session_id, content_hash, json.dumps(grades), graded_at),
+        )
+
+
 # --- embeddings storage --------------------------------------------------
 
 

src/vouch/jsonl_server.py

@@ -401,7 +401,11 @@ def _h_session_transcript(p: dict) -> dict:
     agent = p.get("agent")
     if agent is not None and agent not in ("claude", "codex"):
         raise ValueError(f"unknown agent: {agent!r} (expected 'claude' or 'codex')")
-    return transcript.load_transcript(_store(), session_id, agent=agent)
+    return transcript.load_transcript(
+        _store(), session_id, agent=agent,
+        grade=bool(p.get("grade", False)),
+        regrade=bool(p.get("regrade", False)),
+    )
 
 
 def _h_propose_entity(p: dict) -> dict:

src/vouch/llm_draft.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import json
+import os
 import re
 import subprocess
 import tempfile
@@ -47,6 +48,12 @@ def run_llm(
                 input=prompt, capture_output=True, text=True,
                 encoding="utf-8", errors="replace",
                 timeout=timeout_seconds,
+                # An agent CLI used as llm_cmd fires its own vouch hooks; the
+                # temp cwd shields cwd-discovery, but hooks that resolve the
+                # project dir from their own env still capture the drafting
+                # session back into the KB as a new pending summary each run.
+                # capture.load_config honors this kill-switch.
+                env={**os.environ, "VOUCH_CAPTURE_DISABLE": "1"},
             )
         except subprocess.TimeoutExpired as e:
             raise LLMDraftError(f"{label} timed out after {timeout_seconds:.0f}s") from e

src/vouch/server.py

@@ -595,18 +595,29 @@ def kb_list_sessions() -> dict[str, Any]:
 
 
 @mcp.tool()
-def kb_session_transcript(session_id: str, agent: str | None = None) -> dict[str, Any]:
+def kb_session_transcript(
+    session_id: str,
+    agent: str | None = None,
+    grade: bool = False,
+    regrade: bool = False,
+) -> dict[str, Any]:
     """Render a captured session's full transcript from its raw agent JSONL.
 
     Read-only. Locates the raw Claude Code / Codex file on disk and normalizes
     it into message blocks (text, thinking, tool_use with paired results).
+    Injected scaffolding (system reminders, hook payloads, command output)
+    arrives tagged with a ``noise`` kind so clients can collapse it.
     ``agent`` restricts the search ("claude" | "codex"); omit to try both.
+    ``grade=True`` adds cached LLM review-relevance annotations via
+    ``compile.llm_cmd`` (``regrade=True`` bypasses the cache).
     Degrades to compact capture observations when the raw file is unavailable.
     """
     from . import transcript
     if agent is not None and agent not in ("claude", "codex"):
         raise ValueError(f"unknown agent: {agent!r} (expected 'claude' or 'codex')")
-    return transcript.load_transcript(_store(), session_id, agent=agent)
+    return transcript.load_transcript(
+        _store(), session_id, agent=agent, grade=grade, regrade=regrade,
+    )
 
 
 @mcp.tool()

src/vouch/transcript.py

@@ -72,6 +72,58 @@ def i(key: str) -> int:
     }
 
 
+# Injected/scaffolding user-role content the reviewer never typed. Tagged, not
+# dropped: the raw file stays the evidence of record and the console renders
+# noise behind a stub. Kinds:
+#   meta            — the JSONL entry itself carries isMeta: true
+#   system-reminder — <system-reminder> blocks (memory recalls, hook nags)
+#   hook-context    — SessionStart / UserPromptSubmit / notification payloads,
+#                     including vouch's own per-prompt kb injection
+#   command         — slash-command and IDE scaffolding
+_NOISE_PREFIXES: tuple[tuple[str, str], ...] = (
+    ("<system-reminder>", "system-reminder"),
+    ("<user-prompt-submit-hook>", "hook-context"),
+    ("<session-start-hook>", "hook-context"),
+    ("<task-notification>", "hook-context"),
+    ("[SYSTEM NOTIFICATION", "hook-context"),
+    ("SessionStart hook", "hook-context"),
+    ("UserPromptSubmit hook", "hook-context"),
+    ("<command-name>", "command"),
+    ("<command-message>", "command"),
+    ("<local-command-stdout>", "command"),
+    ("<local-command-stderr>", "command"),
+    ("<bash-input>", "command"),
+    ("<bash-stdout>", "command"),
+    ("<bash-stderr>", "command"),
+    ("<ide_opened_file>", "command"),
+    ("<ide_selection>", "command"),
+    ("Caveat: The messages below were generated by the user while running", "command"),
+)
+
+
+def classify_noise(text: str) -> str | None:
+    """The noise kind for an injected user-role text block, or None.
+
+    Prefix-matched on the stripped text: injected payloads arrive as whole
+    content parts, so a genuine prompt that merely *mentions* one of these
+    tags mid-sentence is never tagged.
+    """
+    stripped = text.lstrip()
+    for prefix, kind in _NOISE_PREFIXES:
+        if stripped.startswith(prefix):
+            return kind
+    return None
+
+
+def _rollup_noise(message: dict[str, Any]) -> None:
+    """Stamp message["noise"] when every block is noise (one-line UI stub)."""
+    blocks = message["blocks"]
+    kinds = {b.get("noise") for b in blocks}
+    if blocks and None not in kinds:
+        only = kinds.pop() if len(kinds) == 1 else "mixed"
+        message["noise"] = only
+
+
 def _result_text(content: Any) -> str:
     """tool_result.content is a string, or a list of {type:text,text} parts."""
     if isinstance(content, str):
@@ -194,14 +246,25 @@ def flush() -> None:
 
             # user entry
             flush()
+            is_meta = bool(obj.get("isMeta"))
+
+            def _text_block(text: str, *, _meta: bool = is_meta) -> dict[str, Any]:
+                block: dict[str, Any] = {"type": "text", "text": text}
+                kind = "meta" if _meta else classify_noise(text)
+                if kind is not None:
+                    block["noise"] = kind
+                return block
+
             if isinstance(content, str):
                 text = content.strip()
                 if text:
-                    messages.append({
+                    message = {
                         "role": "user", "id": None, "model": None,
                         "timestamp": ts if isinstance(ts, str) else None,
-                        "tokens": None, "blocks": [{"type": "text", "text": text}],
-                    })
+                        "tokens": None, "blocks": [_text_block(text)],
+                    }
+                    _rollup_noise(message)
+                    messages.append(message)
                 continue
             parts = content if isinstance(content, list) else []
             user_blocks: list[dict[str, Any]] = []
@@ -224,13 +287,15 @@ def flush() -> None:
                 elif part.get("type") == "text":
                     text = str(part.get("text", "")).strip()
                     if text:
-                        user_blocks.append({"type": "text", "text": text})
+                        user_blocks.append(_text_block(text))
             if user_blocks:
-                messages.append({
+                message = {
                     "role": "user", "id": None, "model": None,
                     "timestamp": ts if isinstance(ts, str) else None,
                     "tokens": None, "blocks": user_blocks,
-                })
+                }
+                _rollup_noise(message)
+                messages.append(message)
     flush()
     return {"session": session, "messages": messages, "truncated": truncated}
 
@@ -325,11 +390,17 @@ def flush() -> None:
                 if role == "user":
                     flush()
                     if text:
-                        messages.append({
+                        text_block: dict[str, Any] = {"type": "text", "text": text}
+                        kind = classify_noise(text)
+                        if kind is not None:
+                            text_block["noise"] = kind
+                        message = {
                             "role": "user", "id": None, "model": None,
                             "timestamp": None, "tokens": None,
-                            "blocks": [{"type": "text", "text": text}],
-                        })
+                            "blocks": [text_block],
+                        }
+                        _rollup_noise(message)
+                        messages.append(message)
                 elif role == "assistant":
                     if current is None:
                         current = new_assistant()
@@ -370,13 +441,22 @@ def _degraded(store: KBStore, session_id: str, reason: str) -> dict[str, Any]:
 
 
 def load_transcript(
-    store: KBStore, session_id: str, *, agent: str | None = None
+    store: KBStore,
+    session_id: str,
+    *,
+    agent: str | None = None,
+    grade: bool = False,
+    regrade: bool = False,
 ) -> dict[str, Any]:
     """Locate + parse the raw transcript for ``session_id``.
 
     ``agent`` restricts the search ("claude" | "codex"); when None both are
     tried. Returns the normalized schema on success, or a degraded result
     (compact capture observations) when the raw file is missing/too large.
+
+    ``grade=True`` additionally annotates dialog messages with LLM
+    review-relevance grades (cached; ``regrade=True`` bypasses the cache).
+    Grades are presentation metadata — they never reorder or drop messages.
     """
     path: Path | None = None
     source_agent = ""
@@ -403,4 +483,17 @@ def load_transcript(
         parsed = parse_claude_transcript(path, max_messages=MAX_MESSAGES)
     else:
         parsed = parse_codex_transcript(path, max_messages=MAX_MESSAGES)
-    return {"available": True, "source": {"agent": source_agent, "path": str(path)}, **parsed}
+    result = {
+        "available": True,
+        "source": {"agent": source_agent, "path": str(path)},
+        **parsed,
+    }
+
+    from . import transcript_grading
+
+    result["grading_available"] = transcript_grading.grading_available(store)
+    if grade or regrade:
+        transcript_grading.apply_grades(
+            store, session_id, result, raw_path=path, regrade=regrade,
+        )
+    return result