fix(transforms/content-router): detect on inner tool-output payload (#1717)

## Description

Coding-agent harnesses wrap each tool result in an envelope such as
`<returncode>0</returncode>\n<output>…</output>` (also `<stdout>`,
`<stderr>`,
`<tool_result>`, `<result>`). The native content detector read those
wrapper
tags as markup and classified the whole payload as HTML/XML — so source
code,
grep results, and logs were misrouted to the HTML article-extractor,
which
blanks or corrupts them (dropping identifiers and route converters).

This routes **detection** on the unwrapped inner payload so the real
content
type wins. **Compression still runs on the original content**, so the
envelope
tags (exit code, stream separation) are preserved — no information is
lost.

Also threads per-compressor config overrides through
`ContentRouterConfig` via
`dataclasses.replace`, so the proxy can tune each structural compressor
while
`ContentRouter` keeps enforcing global safety flags
(`ccr_inject_marker`,
search grouping).

Closes #

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [x] Code refactoring (config-override plumbing; no change to default
behavior)

## Changes Made

- `_strip_detection_envelope()` + `_DETECTION_ENVELOPE_RE`: unwrap a
whole-string
tool-output envelope for detection only. Fires only when the entire
string is a
single wrapper; never returns an empty probe (falls back to the
original).
- `_detect_content()` now detects on the unwrapped payload.
- `ContentRouterConfig` gains `search_compressor` / `log_compressor` /
`diff_compressor` / `text_crusher` override fields (default `None` →
each
compressor's own defaults). The four `_get_*` getters start from the
override
  (or default) and `replace()` in the ContentRouter-enforced flags.
- Regression tests for both behaviors.

## Testing

- [x] Unit tests pass (targeted suites below)
- [x] Linting passes (`ruff check`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality

### Test Output

```text
$ pytest tests/test_transforms_content_router.py tests/test_transforms_content_detection.py -q
45 passed in 0.50s

$ ruff check headroom/transforms/content_router.py tests/test_transforms_content_router.py
All checks passed!

$ mypy headroom/transforms/content_router.py
Success: no issues found in 1 source file
```

## Real Behavior Proof

- Environment: local, Python 3.12.6, native `headroom._core` detect
backend.
- Exact command / steps:
`_detect_content("<returncode>0</returncode>\n<output>\n<python
source>\n</output>")`
- Observed result: detects `ContentType.SOURCE_CODE` (identical to the
same code
unwrapped). Before this change the wrapper tags made it detect as HTML.
- Also measured that the search/log/diff compressors already tolerate
the
envelope (≤1% ratio delta wrapped vs bare), so compression is left on
the
  original content and the tags are preserved rather than stripped.
- Not tested: end-to-end proxy request replay; the config-override
fields are
  plumbing only (no proxy wiring in this PR).

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Additional Notes

- The four config-override fields are wiring only; the proxy is not yet
passing
  overrides through them (follow-up).
- Pre-existing mypy findings in
`tests/test_transforms_content_router.py`
(FakeTokenizer typing, untyped helpers) are unrelated to this change and
left
as-is; `mypy headroom` is clean and the two added tests are fully typed.

headroom/transforms/content_router.py

@@ -45,7 +45,7 @@ import sys
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from enum import Enum
 from typing import Any
 
@@ -185,6 +185,40 @@ def _rust_detect_watchdogged(rust_detect: Any, content: str, timeout: float) ->
     return box["result"]
 
 
+# Coding agents commonly wrap each tool result in an envelope such as
+# ``<returncode>0</returncode>\n<output>...</output>`` (or <stdout>/<stderr>/
+# <tool_result>). Those wrapper tags make the native detector read the whole
+# payload as markup (HTML/XML) even though the inner content is source code, a
+# grep result, or a log. That misroutes to the HTML article-extractor, which
+# blanks or corrupts code (dropping identifiers and route converters). Detect on
+# the inner payload so the real content type wins; compression still runs on the
+# original content.
+_DETECTION_ENVELOPE_RE = re.compile(
+    r"\A\s*(?:<returncode>\s*-?\d+\s*</returncode>\s*)?"
+    r"<(?P<tag>output|stdout|stderr|tool_result|result)>\n?"
+    r"(?P<body>.*?)"
+    r"\n?</(?P=tag)>\s*\Z",
+    re.DOTALL,
+)
+
+
+def _strip_detection_envelope(content: str) -> str:
+    """Return the inner payload of a tool-output envelope, for detection only.
+
+    Only strips when the ENTIRE string is a single wrapper envelope, so content
+    that merely mentions these tags is left untouched. Never returns an empty
+    probe (falls back to the original when the body is blank).
+    """
+    if "<" not in content:
+        return content
+    match = _DETECTION_ENVELOPE_RE.match(content)
+    if match:
+        body = match.group("body")
+        if body.strip():
+            return body
+    return content
+
+
 def _detect_content(content: str) -> DetectionResult:
     """Detect content type via the native chain, with a safe Windows default.
 
@@ -202,6 +236,10 @@ def _detect_content(content: str) -> DetectionResult:
     """
     global _detect_backend_warned, _detect_panic_warned, _detect_native_unhealthy
 
+    # Detect on the unwrapped payload so a tool-output envelope's tags don't get
+    # the whole result misclassified as HTML/XML (#route-converter corruption).
+    content = _strip_detection_envelope(content)
+
     backend = _resolve_detect_backend()
     if backend == "python":
         if not _detect_backend_warned:
@@ -800,6 +838,15 @@ class ContentRouterConfig:
     # the crusher themselves.
     smart_crusher: Any | None = None
 
+    # Structural compressor configuration overrides. None preserves each
+    # compressor's dataclass defaults. The proxy wires environment-backed
+    # overrides into these objects, while ccr_inject_marker/search grouping are
+    # still enforced by ContentRouter so global safety flags win consistently.
+    search_compressor: Any | None = None
+    log_compressor: Any | None = None
+    diff_compressor: Any | None = None
+    text_crusher: Any | None = None
+
     # Group search-compressor output by file (`rg --heading` style).
     # Default False; the proxy enables it in token mode.
     search_group_by_file: bool = False
@@ -1995,12 +2042,13 @@ class ContentRouter(Transform):
             try:
                 from .search_compressor import SearchCompressor, SearchCompressorConfig
 
-                self._search_compressor = SearchCompressor(
-                    SearchCompressorConfig(
-                        group_by_file=self.config.search_group_by_file,
-                        enable_ccr=self.config.ccr_inject_marker,
-                    )
+                cfg = self.config.search_compressor or SearchCompressorConfig()
+                cfg = replace(
+                    cfg,
+                    group_by_file=self.config.search_group_by_file,
+                    enable_ccr=self.config.ccr_inject_marker,
                 )
+                self._search_compressor = SearchCompressor(cfg)
             except ImportError:
                 logger.debug("SearchCompressor not available")
         return self._search_compressor
@@ -2011,9 +2059,9 @@ class ContentRouter(Transform):
             try:
                 from .log_compressor import LogCompressor, LogCompressorConfig
 
-                self._log_compressor = LogCompressor(
-                    LogCompressorConfig(enable_ccr=self.config.ccr_inject_marker)
-                )
+                cfg = self.config.log_compressor or LogCompressorConfig()
+                cfg = replace(cfg, enable_ccr=self.config.ccr_inject_marker)
+                self._log_compressor = LogCompressor(cfg)
             except ImportError:
                 logger.debug("LogCompressor not available")
         return self._log_compressor
@@ -2026,9 +2074,10 @@ class ContentRouter(Transform):
             return None
         if self._text_crusher is None:
             try:
-                from .text_crusher import TextCrusher
+                from .text_crusher import TextCrusher, TextCrusherConfig
 
-                self._text_crusher = TextCrusher()
+                cfg = self.config.text_crusher or TextCrusherConfig()
+                self._text_crusher = TextCrusher(cfg)
             except ImportError:
                 logger.debug("TextCrusher (headroom._core) unavailable; disabling gate route")
                 self._text_crusher_enabled = False
@@ -2052,9 +2101,9 @@ class ContentRouter(Transform):
         if self._diff_compressor is None:
             from .diff_compressor import DiffCompressor, DiffCompressorConfig
 
-            self._diff_compressor = DiffCompressor(
-                DiffCompressorConfig(enable_ccr=self.config.ccr_inject_marker)
-            )
+            cfg = self.config.diff_compressor or DiffCompressorConfig()
+            cfg = replace(cfg, enable_ccr=self.config.ccr_inject_marker)
+            self._diff_compressor = DiffCompressor(cfg)
         return self._diff_compressor
 
     def _get_html_extractor(self) -> Any:

tests/test_transforms_content_router.py

@@ -16,6 +16,7 @@ from headroom.transforms.content_router import (
     _create_content_signature,
     _detect_content,
     _extract_json_block,
+    _strip_detection_envelope,
     is_mixed_content,
     split_into_sections,
 )
@@ -989,3 +990,37 @@ def test_detect_content_circuit_breaker_skips_native_after_hang(
         assert calls == 1  # breaker tripped: native entered once, 2nd call skipped it
     finally:
         release.set()  # let the lone daemon worker finish
+
+
+def test_strip_detection_envelope_isolates_tool_output_payload() -> None:
+    """Only a whole-string tool-output envelope is unwrapped; content that
+    merely mentions the tags, or has an empty body, is left untouched."""
+    body = "def main():\n    return 1"
+    wrapped = f"<returncode>0</returncode>\n<output>\n{body}\n</output>"
+    assert _strip_detection_envelope(wrapped) == body
+    # <output> alias tags and a bare envelope (no returncode) also unwrap.
+    assert _strip_detection_envelope(f"<stdout>\n{body}\n</stdout>") == body
+    # Non-envelope content is returned verbatim (no "<" fast-path + no match).
+    prose = "see the <output> tag docs for details"
+    assert _strip_detection_envelope(prose) == prose
+    # Empty body never yields an empty probe — falls back to the original.
+    empty = "<output>\n\n</output>"
+    assert _strip_detection_envelope(empty) == empty
+
+
+def test_detect_content_sees_through_tool_output_envelope() -> None:
+    """Regression: a tool-result envelope's tags used to make the detector
+    read the whole payload as markup and misroute code to the HTML extractor.
+    Detection now runs on the inner payload, so the real type wins."""
+    code = "\n".join(
+        [
+            "import os",
+            "from pathlib import Path",
+            "",
+            "def main() -> int:",
+            "    return len(os.listdir(Path.cwd()))",
+        ]
+    )
+    wrapped = f"<returncode>0</returncode>\n<output>\n{code}\n</output>"
+    assert _detect_content(wrapped).content_type is ContentType.SOURCE_CODE
+    assert _detect_content(wrapped).content_type is _detect_content(code).content_type