Router + hooks hardening: production-grade fixes from user-simulation stress test

- handle_dhee_read: coerce offset/limit, return error dict on non-numeric
  input instead of raising ValueError up the MCP stack.
- Rename digest tags <dhee:read>/<dhee:bash>/<dhee:agent> to underscore
  variants. Colon-prefixed names require a namespace declaration; any
  downstream XML parse chokes on them (same class as the <dhee:ctx> fix).
- Scrub internal degradation warnings at the assembler boundary before
  they reach the renderer. Prevents 'Context assembly degraded: ...'
  error strings (including API-key-laden 401 bodies) from burning
  tokens and leaking into every injected context.
- PreToolUse heavy-pattern gate: strip quoted regions before regex scan
  so echo 'git log stuff' is no longer denied. Real git log and piped
  variants still deny.

Adds 15 regression tests covering argument validation, digest XML
safety, warning scrub, and quote-safety in the enforcement gate.

1098 pass, 10 skipped, 0 failed.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: Ashish-dwi99

dhee/hooks/claude_code/assembler.py

@@ -5,14 +5,19 @@
 only the chunks relevant to THIS specific prompt and assembles them into
 a token-budgeted injection.
 
-The economics:
-    CLAUDE.md is ~2000 tokens. Over a 20-turn conversation, that's 40K
-    input tokens of mostly-irrelevant context. If Dhee injects ~200 tokens
-    of relevant chunks per turn, that's 4K tokens — 10x savings on the
-    costliest model (Opus).
-
-    Even with a Dhee-side embedding call for retrieval (~$0.0001), the
-    savings on Opus input ($0.015/1K tokens) yield >100x ROI.
+The economics (revised 2026-04-17):
+    Phase 0 audit (`dhee/benchmarks/phase0_context_audit.py`) parsed real
+    Anthropic `usage` fields across 14 sessions / 4,680 assistant turns.
+    CLAUDE.md first-turn cost bundled with system prompt + skill catalog
+    + tool schemas is ~5–8K tokens — a rounding error, not the fat. The
+    actual fat is **tool-result accumulation** (~57% of growing content)
+    and tool-use inputs (~31%). Top 10% of turns account for 75% of new
+    input.
+
+    This assembler therefore does relevance-filtering on *docs* (a small
+    marginal win), not the primary win. The primary router lever is
+    `dhee.router` — digest-at-source MCP wrappers (`dhee_read`, etc.)
+    that keep raw tool output out of the context entirely.
 
 The assembler is a pure selection pipeline:
     query → vector search → filter(kind, score) → budget → render
@@ -110,7 +115,7 @@ def assemble(
                 user_id=os.environ.get("DHEE_USER_ID", "default"),
             )
             if isinstance(ctx, dict):
-                typed = ctx
+                typed = _strip_internal_warnings(ctx)
         except Exception:
             pass
 
@@ -121,6 +126,32 @@ def assemble(
     )
 
 
+# Degradation messages come from internal error paths (e.g. embedder 401s,
+# store read failures). They're meant for logs and telemetry, not for the
+# LLM — surfacing them wastes tokens and can leak error-string content
+# (headers, stack traces, model names). Stripped here so the renderer
+# never sees them.
+_INTERNAL_WARNING_PREFIXES = (
+    "Context assembly degraded:",
+    "Cognitive state degraded:",
+)
+
+
+def _strip_internal_warnings(ctx: dict[str, Any]) -> dict[str, Any]:
+    warnings = ctx.get("warnings")
+    if not isinstance(warnings, list):
+        return ctx
+    visible = [
+        w for w in warnings
+        if isinstance(w, str) and not w.startswith(_INTERNAL_WARNING_PREFIXES)
+    ]
+    if len(visible) == len(warnings):
+        return ctx
+    scrubbed = dict(ctx)
+    scrubbed["warnings"] = visible
+    return scrubbed
+
+
 def assemble_docs_only(
     dhee: Any,
     query: str,

dhee/hooks/claude_code/renderer.py

@@ -8,10 +8,17 @@
 Token philosophy (Caveman-inspired): drop structural fluff, keep
 technical substance exact. No indentation, short tags, no wrapper
 nesting, no redundant metadata. Every byte earns its place.
+
+Phase 5 (2026-04-17): root is ``<dhee v="1">`` with typed
+sections. Doc chunks carry ``src``/``head``/``s`` so the model knows
+the origin, not just a weighted fragment. ``<edits>`` is a first-class
+section, fed in from the edit ledger. Schema stays terse — JSON-heavy
+envelopes were rejected as more expensive than the caveman tags.
 """
 
 from __future__ import annotations
 
+import os
 from typing import Any
 from xml.sax.saxutils import escape as _xml_escape
 
@@ -28,12 +35,15 @@ def render_context(
     max_insights: int = 5,
     max_intentions: int = 3,
     doc_matches: list | None = None,
+    edits_block: str | None = None,
 ) -> str:
     """Render Dhee context dict as flat XML for Claude Code injection.
 
     Returns empty string when nothing to inject.
     """
     sections: list[tuple[int, list[str]]] = [
+        (120, _router_block()),
+        (115, _edits_section(edits_block)),
         (110, _docs_block(doc_matches)),
         (100, _session_block(ctx.get("last_session"))),
         (90, _performance_block(ctx.get("performance", []))),
@@ -55,6 +65,7 @@ def render_context(
     attrs = ""
     if task_description:
         attrs = f' task="{_esc_attr(task_description[:120])}"'
+    attrs += ' v="1"'
 
     open_tag = f"<dhee{attrs}>"
     close_tag = "</dhee>"
@@ -88,22 +99,50 @@ def estimate_tokens(text: str) -> int:
 # ---------------------------------------------------------------------------
 
 
+_ROUTER_NUDGE = (
+    "router=on. Prefer mcp__dhee__dhee_read over Read for files >200 lines "
+    "(pass offset/limit for ranges). Prefer mcp__dhee__dhee_bash over Bash "
+    "for commands likely >2KB output (git log/diff, pytest, find, grep, "
+    "ls -R). After a subagent/large tool return, pass the text through "
+    "mcp__dhee__dhee_agent to keep raw out of context. Call "
+    "mcp__dhee__dhee_expand_result(ptr) only when a digest is genuinely "
+    "insufficient — raw re-enters context."
+)
+
+
+def _router_block() -> list[str]:
+    """One-time router nudge. Only rendered when DHEE_ROUTER=1."""
+    if os.environ.get("DHEE_ROUTER") != "1":
+        return []
+    return [f"<router>{_xml_escape(_ROUTER_NUDGE)}</router>"]
+
+
 def _docs_block(doc_matches: list | None) -> list[str]:
     if not doc_matches:
         return []
     items: list[str] = []
     for m in doc_matches:
-        path = getattr(m, "heading_breadcrumb", "") or ""
+        head = getattr(m, "heading_breadcrumb", "") or ""
+        src = getattr(m, "source_name", "") or ""
         score = getattr(m, "score", 0.0)
         text = getattr(m, "text", "")
         if not text:
             continue
-        if path and text.startswith(path):
-            text = text[len(path):].lstrip("\n")
-        items.append(_tag("r", _score_attr(score), text))
+        if head and text.startswith(head):
+            text = text[len(head):].lstrip("\n")
+        attrs = _attrs(src=src, head=head)
+        score_attr = _score_attr(score)
+        a = f"{attrs} {score_attr}" if attrs else score_attr
+        items.append(_tag("doc", a, text))
     return items
 
 
+def _edits_section(edits_block: str | None) -> list[str]:
+    if not edits_block:
+        return []
+    return [edits_block]
+
+
 def _session_block(session: dict[str, Any] | None) -> list[str]:
     if not session or not isinstance(session, dict):
         return []

dhee/router/agent_digest.py

@@ -0,0 +1,158 @@
+"""Digest for subagent / long-text tool returns.
+
+`dhee_agent` accepts a text blob (typically a subagent's final message)
+and returns a factual, compact digest: which files were referenced,
+bulleted findings, any error indicators, and head/tail excerpts. Raw is
+stored behind a ptr. Honest about what was extracted — never invents
+references.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+
+
+CHARS_PER_TOKEN = 3.5
+
+# file:line patterns — matches path/to/file.py:123 (optionally :col)
+_FILE_LINE_RE = re.compile(
+    r"(?<![\w/])"                             # not preceded by word/slash
+    r"([A-Za-z0-9_./\-]+\.[A-Za-z0-9]{1,8})"  # path with extension
+    r":(\d+)"                                  # :lineno
+    r"(?::\d+)?"                               # optional :col
+)
+
+_BULLET_RE = re.compile(r"^\s*(?:[-*+]|\d+\.)\s+(.+)$", re.MULTILINE)
+
+_ERROR_RE = re.compile(
+    r"\b(error|exception|failed|failure|traceback|fatal)\b",
+    re.IGNORECASE,
+)
+
+_HEADING_RE = re.compile(r"^(#{1,6})\s+(.+?)\s*#*\s*$", re.MULTILINE)
+
+
+@dataclass
+class AgentDigest:
+    char_count: int
+    line_count: int
+    est_tokens: int
+    kind: str
+    file_refs: list[str] = field(default_factory=list)
+    headings: list[str] = field(default_factory=list)
+    bullets: list[str] = field(default_factory=list)
+    error_hits: int = 0
+    head: str = ""
+    tail: str = ""
+    notes: list[str] = field(default_factory=list)
+
+    def render(self, ptr: str) -> str:
+        lines: list[str] = [f'<dhee_agent ptr="{ptr}">']
+        lines.append(
+            f"size={self.line_count} lines, {self.char_count} chars, "
+            f"~{self.est_tokens} tokens"
+        )
+        lines.append(f"kind={self.kind}")
+        if self.headings:
+            lines.append("headings:")
+            for h in self.headings[:8]:
+                lines.append(f"  {h}")
+            if len(self.headings) > 8:
+                lines.append(f"  (+{len(self.headings)-8} more)")
+        if self.file_refs:
+            shown = self.file_refs[:15]
+            lines.append("file_refs:")
+            for r in shown:
+                lines.append(f"  {r}")
+            if len(self.file_refs) > 15:
+                lines.append(f"  (+{len(self.file_refs)-15} more)")
+        if self.bullets:
+            lines.append("bullets:")
+            for b in self.bullets[:10]:
+                lines.append(f"  - {b}")
+            if len(self.bullets) > 10:
+                lines.append(f"  (+{len(self.bullets)-10} more)")
+        if self.error_hits:
+            lines.append(f"error_signals={self.error_hits}")
+        if self.head:
+            lines.append("head:")
+            for hl in self.head.splitlines()[:6]:
+                lines.append(f"  {hl}")
+        if self.tail:
+            lines.append("tail:")
+            for tl in self.tail.splitlines()[-4:]:
+                lines.append(f"  {tl}")
+        for n in self.notes:
+            lines.append(f"note: {n}")
+        lines.append(f'(expand: dhee_expand_result(ptr="{ptr}"))')
+        lines.append("</dhee_agent>")
+        return "\n".join(lines)
+
+
+def _head_tail(text: str, head_lines: int = 6, tail_lines: int = 4) -> tuple[str, str]:
+    lines = text.splitlines()
+    if len(lines) <= head_lines + tail_lines:
+        return text, ""
+    return "\n".join(lines[:head_lines]), "\n".join(lines[-tail_lines:])
+
+
+def digest_agent(text: str, *, kind: str | None = None) -> AgentDigest:
+    """Build an AgentDigest from an arbitrary text blob."""
+    char_count = len(text)
+    line_count = text.count("\n") + (1 if text and not text.endswith("\n") else 0)
+    if text == "":
+        line_count = 0
+
+    # file:line refs — de-dupe, preserve order
+    seen: set[str] = set()
+    file_refs: list[str] = []
+    for m in _FILE_LINE_RE.finditer(text):
+        ref = f"{m.group(1)}:{m.group(2)}"
+        if ref in seen:
+            continue
+        seen.add(ref)
+        file_refs.append(ref)
+
+    headings = [
+        f"{'#' * len(m.group(1))} {m.group(2)}"
+        for m in _HEADING_RE.finditer(text)
+    ]
+
+    bullets: list[str] = []
+    for m in _BULLET_RE.finditer(text):
+        b = m.group(1).strip()
+        if b and len(b) <= 200:
+            bullets.append(b)
+
+    error_hits = len(_ERROR_RE.findall(text))
+
+    head, tail = _head_tail(text)
+
+    # Auto-classify kind if not given.
+    if kind is None:
+        if "```" in text and file_refs:
+            kind = "code-review"
+        elif error_hits and file_refs:
+            kind = "error-report"
+        elif headings:
+            kind = "structured-summary"
+        elif file_refs:
+            kind = "code-survey"
+        elif bullets:
+            kind = "bulleted-findings"
+        else:
+            kind = "prose"
+
+    return AgentDigest(
+        char_count=char_count,
+        line_count=line_count,
+        est_tokens=int(char_count / CHARS_PER_TOKEN),
+        kind=kind,
+        file_refs=file_refs,
+        headings=headings,
+        bullets=bullets,
+        error_hits=error_hits,
+        head=head,
+        tail=tail,
+    )