fix(proxy): guard feedback endpoints and add CSRF checks to loopback writes (#3060)

## Description

`#2927` brought eight telemetry/TOIN routes under `require_loopback`.
Two structurally identical siblings 60 lines above them were missed:

```
GET /v1/feedback
GET /v1/feedback/{tool_name}
```

Neither is an aggregate-counter endpoint. Their `common_queries` /
`queried_fields` keys are built verbatim from agent search text —
`event.query.lower()` at `headroom/cache/compression_feedback.py:311` —
and up to 100 queries are retained per tool, keyed by real tool name.
Under the shipped Docker default (`--host 0.0.0.0`) a LAN peer gets a
404 from `/v1/toin/patterns` and the query corpus from `/v1/feedback`.

Separately, five mutating loopback-only routes had no CSRF guard.
`require_loopback` cannot stop that attack: a remote page POSTing to a
known `127.0.0.1` URL with `Content-Type: text/plain` is a CORS *simple*
request, so there is no preflight, and the browser still sends the real
loopback `Host` header — both of the guard's gates pass. Only `Origin`
betrays the caller, and only `require_same_origin` inspects it. That
guard already existed at `headroom/proxy/loopback_guard.py:219` and was
applied solely to `/settings`.

Closes #2927 (completes it — the original eight routes were already
done).

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update

## Changes Made

- Added `Depends(_require_loopback)` to `/v1/feedback` and
`/v1/feedback/{tool_name}`.
- Stripped `common_queries` / `queried_fields` from both response bodies
even on the guarded path, matching the whitelist discipline #2930
applied at `server.py:4909-4916`.
- Added `_feedback_stats_without_query_text()` so the scrub happens at
the HTTP boundary; `get_stats()` is unchanged and in-process compression
decisions are untouched.
- Added `Depends(_require_same_origin)` to `POST /stats/reset`,
`/cache/clear`, `/v1/retrieve`, `/v1/telemetry/import`,
`/admin/runtime-env`.

## Testing

- [x] Unit tests pass
- [x] Linting passes (ruff check + format)
- [ ] Type checking passes (`uv run mypy headroom`) — not run
- [x] New tests added for new functionality

### Test Output

```text
$ .venv/bin/python -m pytest tests/test_proxy_loopback_gating.py -q
99 passed, 1 warning in 4.18s

$ .venv/bin/python -m pytest tests/test_proxy_settings_endpoints.py tests/test_telemetry.py \
    tests/test_proxy_cache_telemetry.py tests/test_proxy_telemetry_env.py tests/test_telemetry_context.py -q
101 passed, 1 warning in 3.67s

$ .venv/bin/python -m pytest tests/test_critical_fixes.py tests/test_compression_store.py \
    tests/test_toin_full_integration.py tests/test_ccr_feedback.py tests/test_critical_gaps.py \
    tests/test_proxy_ccr.py tests/test_proxy_dashboard_stats_cache.py -q
168 passed, 4 skipped, 3 warnings in 13.18s

$ .venv/bin/python -m ruff check headroom/proxy/server.py tests/test_proxy_loopback_gating.py
All checks passed!
```

Against the parent commit (`git stash` of `server.py` only), all 14 new
tests fail:

```text
FAILED test_non_loopback_caller_gets_404[get-/v1/feedback]
FAILED test_non_loopback_caller_gets_404[get-/v1/feedback/example]
FAILED test_cross_origin_post_rejected[/stats/reset]
FAILED test_cross_origin_post_rejected[/cache/clear]
FAILED test_cross_origin_post_rejected[/v1/retrieve]
FAILED test_cross_origin_post_rejected[/v1/telemetry/import]
FAILED test_cross_origin_post_rejected[/admin/runtime-env]
FAILED test_sandboxed_null_origin_post_rejected[...]  (5 cases)
FAILED test_feedback_stats_exclude_agent_query_text
FAILED test_feedback_tool_detail_excludes_agent_query_text
14 failed, 85 passed
```

## Real Behavior Proof

- Environment: macOS 15 (darwin 25.4.0), Python 3.12.13, this branch,
FastAPI `TestClient` against the real `create_app` proxy.
- Exact command / steps: drive `/v1/feedback` with a feedback singleton
whose `common_queries` contains `"find the customer api key rotation
runbook"`, once from a non-loopback peer and once from a loopback peer;
POST each of the five mutating routes with `Origin:
https://attacker.example` and `Content-Type: text/plain`.
- Observed result: non-loopback callers now receive 404 where they
previously received 200 with the query corpus; on the loopback path the
response no longer contains `common_queries`, `queried_fields`, or the
substring `customer api key rotation`, while `retrieval_rate` still
resolves to `0.25`. All five cross-origin POSTs return 403; the same
requests with no `Origin`, or with `Origin: http://127.0.0.1`, are
unaffected.
- Not tested: a real browser issuing the cross-origin POST (the CORS
simple-request shape is reproduced at the header level, not in a
browser), and a live non-loopback deployment.

## Runtime Rollout Safety

- Rollout-managed feature(s): none.
- Minimum rollout channel: N/A.
- Stable/default behavior changed: yes — `/v1/feedback*` now 404 for
non-loopback callers and no longer return query text; five POST routes
reject cross-origin browser callers.
- Kill switch / disable path: none; these are security guards and are
deliberately not configurable.
- Unsafe override required: none.
- Qualification impact: none.
- Rollback path: revert this commit.

## Review Readiness

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

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective
- [x] New and existing unit tests pass locally with my changes

## Additional Notes

`/stats` also calls `feedback.get_stats()` (`server.py:3896`) but only
reads aggregate counters at `:4303-4311` and never emits query text —
verified, and the reason the scrub is applied at the HTTP boundary
rather than inside `get_stats()`.

The five POST routes are strictly loopback-gated, so the
trusted-dashboard wrapper `/settings` uses is unnecessary here; for a
loopback caller that wrapper falls through to the same raw guard. No
dashboard asset calls them, and the TypeScript SDK
(`sdk/typescript/src/client.ts:322,443`) sends no `Origin` header, which
the guard passes through unchanged.

Co-authored-by: Tejas Chopra <tejas@Tejass-MacBook-Pro.local>

headroom/proxy/server.py

@@ -2438,6 +2438,35 @@ def _normalized_http_origin(value: str) -> tuple[str, str, int] | None:
     return scheme, parsed.hostname.lower(), port
 
 
+#: Feedback-pattern keys built verbatim from agent query text. They are useful
+#: in-process for compression decisions but must never reach an HTTP response —
+#: same privacy contract the TOIN endpoints were brought under in #2926/#2927.
+_FEEDBACK_QUERY_TEXT_KEYS = ("common_queries", "queried_fields")
+
+
+def _feedback_stats_without_query_text(stats: dict[str, Any]) -> dict[str, Any]:
+    """Return ``stats`` with per-tool query text stripped from ``tool_patterns``.
+
+    Copies only the levels it edits; the aggregate counters are shared with the
+    caller's dict, which is fine because they are scalars.
+    """
+
+    patterns = stats.get("tool_patterns")
+    if not isinstance(patterns, dict):
+        return stats
+
+    scrubbed: dict[str, Any] = {}
+    for name, pattern in patterns.items():
+        if isinstance(pattern, dict):
+            scrubbed[name] = {
+                key: value for key, value in pattern.items() if key not in _FEEDBACK_QUERY_TEXT_KEYS
+            }
+        else:
+            scrubbed[name] = pattern
+
+    return {**stats, "tool_patterns": scrubbed}
+
+
 _is_known_websocket_callback_failure = is_known_websocket_callback_failure
 
 
@@ -3506,7 +3535,10 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
         payload["runtime"] = _runtime_payload()
         return JSONResponse(status_code=200, content=payload)
 
-    @app.post("/admin/runtime-env", dependencies=[Depends(_require_loopback)])
+    @app.post(
+        "/admin/runtime-env",
+        dependencies=[Depends(_require_loopback), Depends(_require_same_origin)],
+    )
     async def admin_runtime_env(request: Request):
         """Hot-reload live env knobs (the output-shaper family, the ast-grep
         read threshold) without restarting the proxy.
@@ -4447,7 +4479,10 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
                 payload["persistence"] = {**persistence, "error": None}
         return payload
 
-    @app.post("/stats/reset", dependencies=[Depends(_require_loopback)])
+    @app.post(
+        "/stats/reset",
+        dependencies=[Depends(_require_loopback), Depends(_require_same_origin)],
+    )
     async def stats_reset():
         """Reset in-memory proxy stats for local test/debug isolation."""
         await proxy.metrics.reset_runtime()
@@ -4584,7 +4619,10 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
         report = tracker.get_report()
         return report.to_dict()
 
-    @app.post("/cache/clear", dependencies=[Depends(_require_loopback)])
+    @app.post(
+        "/cache/clear",
+        dependencies=[Depends(_require_loopback), Depends(_require_same_origin)],
+    )
     async def clear_cache():
         """Clear the response cache.
 
@@ -4600,7 +4638,10 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
         return {"status": "cache disabled"}
 
     # CCR (Compress-Cache-Retrieve) endpoints
-    @app.post("/v1/retrieve", dependencies=[Depends(_require_loopback)])
+    @app.post(
+        "/v1/retrieve",
+        dependencies=[Depends(_require_loopback), Depends(_require_same_origin)],
+    )
     async def ccr_retrieve(request: Request):
         """Retrieve original content from CCR compression cache.
 
@@ -4669,21 +4710,25 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
             ],
         }
 
-    @app.get("/v1/feedback")
+    @app.get("/v1/feedback", dependencies=[Depends(_require_loopback)])
     async def ccr_feedback():
         """Get CCR feedback loop statistics and learned patterns.
 
         This endpoint exposes the feedback loop's learned patterns for monitoring
         and debugging. It shows:
         - Per-tool retrieval rates (high = compress less aggressively)
-        - Common search queries per tool
+        - Aggregate compression/retrieval counters per tool
-        - Queried fields (suggest what to preserve)
 
         Use this to understand how well compression is working and whether
         the feedback loop is adjusting appropriately.
+
+        Loopback-guarded and query-text free for the same reason as the
+        telemetry and TOIN endpoints (#2926/#2927): ``common_queries`` and
+        ``queried_fields`` are built verbatim from agent search queries, so
+        they stay out of the response even on the guarded path.
         """
         feedback = get_compression_feedback()
-        stats = feedback.get_stats()
+        stats = _feedback_stats_without_query_text(feedback.get_stats())
         return {
             "feedback": stats,
             "hints_example": {
@@ -4702,12 +4747,15 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
             },
         }
 
-    @app.get("/v1/feedback/{tool_name}")
+    @app.get("/v1/feedback/{tool_name}", dependencies=[Depends(_require_loopback)])
     async def ccr_feedback_for_tool(tool_name: str):
         """Get compression hints for a specific tool.
 
         Returns feedback-based hints that would be used for compressing
         this tool's output.
+
+        Loopback-guarded, and the pattern block excludes ``common_queries``
+        and ``queried_fields`` — both are raw agent query text (#2926/#2927).
         """
         feedback = get_compression_feedback()
         hints = feedback.get_compression_hints(tool_name)
@@ -4730,8 +4778,6 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
                 "retrieval_rate": patterns.retrieval_rate if patterns else 0.0,
                 "full_retrieval_rate": patterns.full_retrieval_rate if patterns else 0.0,
                 "search_rate": patterns.search_rate if patterns else 0.0,
-                "common_queries": list(patterns.common_queries.keys())[:10] if patterns else [],
-                "queried_fields": list(patterns.queried_fields.keys())[:10] if patterns else [],
             }
             if patterns
             else None,
@@ -4777,7 +4823,10 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
         telemetry = get_telemetry_collector()
         return telemetry.export_stats()
 
-    @app.post("/v1/telemetry/import", dependencies=[Depends(_require_loopback)])
+    @app.post(
+        "/v1/telemetry/import",
+        dependencies=[Depends(_require_loopback), Depends(_require_same_origin)],
+    )
     async def telemetry_import(request: Request):
         """Import telemetry data from another source.
 

tests/test_proxy_loopback_gating.py

@@ -15,6 +15,7 @@ from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
 from headroom.cache.backends import InMemoryBackend
+from headroom.cache.compression_feedback import CompressionHints
 from headroom.cache.compression_store import get_compression_store, reset_compression_store
 from headroom.proxy.loopback_guard import is_ip_literal_host_header
 from headroom.proxy.server import ProxyConfig, create_app
@@ -30,6 +31,11 @@ GATED = [
     ("get", "/v1/toin/stats"),
     ("get", "/v1/toin/patterns"),
     ("get", "/v1/toin/pattern/example"),
+    # #2927 guarded the eight telemetry/TOIN routes the issue enumerated but
+    # left these two siblings open, and their payload carries the same raw
+    # agent query text (``common_queries``, built from ``event.query``).
+    ("get", "/v1/feedback"),
+    ("get", "/v1/feedback/example"),
 ]
 
 
@@ -119,6 +125,135 @@ def test_toin_pattern_detail_whitelists_learned_payload(monkeypatch: pytest.Monk
     }
 
 
+# Mutating routes reachable from loopback. `require_loopback` cannot stop a
+# remote page from POSTing to a known 127.0.0.1 URL: a "simple" cross-origin
+# request (Content-Type: text/plain carrying JSON) skips preflight, and the
+# browser still sends the real loopback Host header. Only `Origin` betrays the
+# attacker, and only `require_same_origin` inspects it.
+CSRF_GUARDED = [
+    "/stats/reset",
+    "/cache/clear",
+    "/v1/retrieve",
+    "/v1/telemetry/import",
+    "/admin/runtime-env",
+]
+
+
+@pytest.mark.parametrize("path", CSRF_GUARDED)
+def test_cross_origin_post_rejected(path: str) -> None:
+    resp = _loopback_client().post(
+        path,
+        headers={"Origin": "https://attacker.example", "Content-Type": "text/plain"},
+        content="{}",
+    )
+    assert resp.status_code == 403, resp.text
+
+
+@pytest.mark.parametrize("path", CSRF_GUARDED)
+def test_sandboxed_null_origin_post_rejected(path: str) -> None:
+    # A sandboxed iframe or file:// page sends the opaque literal "null".
+    resp = _loopback_client().post(
+        path,
+        headers={"Origin": "null", "Content-Type": "text/plain"},
+        content="{}",
+    )
+    assert resp.status_code == 403, resp.text
+
+
+@pytest.mark.parametrize("path", CSRF_GUARDED)
+def test_loopback_origin_post_allowed(path: str) -> None:
+    # The local dashboard is same-origin on loopback and must keep working.
+    resp = _loopback_client().post(
+        path,
+        headers={"Origin": "http://127.0.0.1"},
+        json={},
+    )
+    assert resp.status_code != 403, resp.text
+
+
+@pytest.mark.parametrize("path", CSRF_GUARDED)
+def test_originless_post_allowed(path: str) -> None:
+    # CLI tools and the TypeScript SDK send no Origin header at all; the guard
+    # must pass them through or it breaks every non-browser client.
+    resp = _loopback_client().post(path, json={})
+    assert resp.status_code != 403, resp.text
+
+
+def _feedback_with_query_text():
+    """A feedback singleton whose patterns carry raw agent query text."""
+
+    class FakePattern:
+        total_compressions = 8
+        total_retrievals = 2
+        retrieval_rate = 0.25
+        full_retrieval_rate = 0.1
+        search_rate = 0.5
+        common_queries = {"find the customer api key rotation runbook": 3}
+        queried_fields = {"internal_field_name": 2}
+
+    class FakeFeedback:
+        def get_stats(self):
+            return {
+                "total_compressions": 8,
+                "total_retrievals": 2,
+                "global_retrieval_rate": 0.25,
+                "tools_tracked": 1,
+                "tool_patterns": {
+                    "Grep": {
+                        "compressions": 8,
+                        "retrievals": 2,
+                        "retrieval_rate": 0.25,
+                        "full_rate": 0.1,
+                        "search_rate": 0.5,
+                        "common_queries": ["find the customer api key rotation runbook"],
+                        "queried_fields": ["internal_field_name"],
+                    }
+                },
+            }
+
+        def get_compression_hints(self, tool_name):
+            # The real implementation is annotated ``-> CompressionHints`` and
+            # always returns one, so the double must too.
+            return CompressionHints()
+
+        def get_all_patterns(self):
+            return {"Grep": FakePattern()}
+
+    return FakeFeedback()
+
+
+def test_feedback_stats_exclude_agent_query_text(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "headroom.proxy.server.get_compression_feedback",
+        _feedback_with_query_text,
+    )
+    response = _loopback_client().get("/v1/feedback")
+
+    assert response.status_code == 200
+    pattern = response.json()["feedback"]["tool_patterns"]["Grep"]
+    assert "common_queries" not in pattern
+    assert "queried_fields" not in pattern
+    # The aggregate counters the endpoint exists to expose still survive.
+    assert pattern["retrieval_rate"] == 0.25
+    assert "customer api key rotation" not in response.text
+
+
+def test_feedback_tool_detail_excludes_agent_query_text(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "headroom.proxy.server.get_compression_feedback",
+        _feedback_with_query_text,
+    )
+    response = _loopback_client().get("/v1/feedback/Grep")
+
+    assert response.status_code == 200
+    pattern = response.json()["pattern"]
+    assert "common_queries" not in pattern
+    assert "queried_fields" not in pattern
+    assert pattern["retrieval_rate"] == 0.25
+    assert "customer api key rotation" not in response.text
+    assert "internal_field_name" not in response.text
+
+
 # CCR data endpoints — cached session content, gated to 404 off-loopback (#1227).
 def test_stats_lifetime_route_uses_dashboard_metadata_access_policy(
     monkeypatch: pytest.MonkeyPatch,