diff --git a/headroom/proxy/interceptors/astgrep.py b/headroom/proxy/interceptors/astgrep.py
index 1516723d9..35ff5836e 100644
--- a/headroom/proxy/interceptors/astgrep.py
+++ b/headroom/proxy/interceptors/astgrep.py
@@ -15,6 +15,7 @@ import logging
 import os
 import re
 import shutil
+import stat
 import subprocess
 import tempfile
 from enum import Enum
@@ -24,7 +25,7 @@ from typing import Any
 from headroom import binaries
 from headroom._subprocess import run
 from headroom.proxy import runtime_env
-from headroom.proxy.project_context import get_current_cwd
+from headroom.proxy.project_context import get_current_cwd, is_current_request_trusted
 
 from . import base
 
@@ -151,32 +152,108 @@ def _verify_truncation_on_disk_enabled() -> bool:
     )
 
 
+def _max_disk_verify_bytes() -> int:
+    # Live read, same hot-reload pattern as _min_chars_to_rewrite(). 5 MB
+    # comfortably covers real source files while bounding worst-case read
+    # time/memory for the disk-verify fallback.
+    try:
+        return int(runtime_env.getenv("HEADROOM_VERIFY_TRUNCATION_MAX_BYTES", "5000000"))
+    except (TypeError, ValueError):
+        return 5_000_000
+
+
+def _resolve_read_path_in_workspace(file_path: str, resolved_root: Path) -> Path | None:
+    """Confine `file_path` (relative or absolute) strictly under `resolved_root`.
+
+    Mirrors memory_handler._resolve_native_path's join/resolve/relative_to
+    pattern. `resolved_root` must already be canonicalized by the caller (a
+    single source of truth for "inside the workspace" across a call) --
+    this does not re-resolve it. `.resolve()` collapses symlinks, including
+    in intermediate path components, before the containment check runs, so
+    a symlink that points outside the workspace is rejected the same way a
+    `..` traversal is -- this is a path-segment containment check via
+    `relative_to()`, not a string-prefix check, so a sibling directory that
+    merely shares a string prefix with the root is correctly rejected too.
+    """
+    candidate = Path(file_path) if Path(file_path).is_absolute() else resolved_root / file_path
+    try:
+        resolved = candidate.resolve(strict=True)
+    except OSError:
+        return None
+    try:
+        resolved.relative_to(resolved_root)
+    except ValueError:
+        return None
+    return resolved
+
+
+def _read_disk_content_bounded(resolved: Path, max_bytes: int) -> str | None:
+    """Read `resolved` iff it's a regular file no larger than `max_bytes`.
+
+    Opens with O_NONBLOCK so a FIFO/special file with no writer returns
+    immediately instead of blocking the calling thread indefinitely --
+    O_NONBLOCK has no effect on reads once fstat confirms a regular file.
+    fstat runs on the already-open fd (not the path) so the type/size
+    check and the read happen on the same underlying file object, closing
+    the TOCTOU gap a separate path-based stat-then-open would leave.
+    """
+    try:
+        fd = os.open(resolved, os.O_RDONLY | os.O_NONBLOCK)
+    except OSError:
+        return None
+    try:
+        st = os.fstat(fd)
+        if not stat.S_ISREG(st.st_mode) or st.st_size > max_bytes:
+            return None
+        with os.fdopen(fd, "r", encoding="utf-8") as f:
+            fd = -1  # ownership transferred to the file object
+            return f.read()
+    except (OSError, UnicodeDecodeError):
+        return None
+    finally:
+        if fd >= 0:
+            os.close(fd)
+
+
 def _verify_read_against_disk(
     file_path: str | None,
     received_content: str,
     cwd: str | None,
+    *,
+    trusted: bool,
 ) -> tuple[ReadVerificationResult, tuple[int, int] | None]:
     """Compare `received_content` against the real file at `file_path`.
 
-    `cwd` (the `x-headroom-cwd` header) resolves a relative `file_path`.
+    `cwd` (the `x-headroom-cwd` header) is never authority on its own --
+    `trusted` (whether the request's peer is loopback, a server-observed
+    fact, not a header) must hold before `cwd` or `file_path` are even
+    inspected, let alone touch disk. Once trusted, every read must be a
+    regular file whose fully resolved path is beneath the canonically
+    resolved `cwd`; anything else is UNKNOWN, never a guess.
+
     TRUNCATED requires an exact-prefix match with strictly more on disk —
     a weaker match means the file diverged since the client read it, not
     a provable truncation, so it's UNKNOWN. Returns `(visible_lines,
     total_lines)` alongside TRUNCATED so the header can cite real numbers
     without a second, potentially racy, read.
     """
+    if not trusted:
+        return ReadVerificationResult.UNKNOWN, None
     if not file_path:
         return ReadVerificationResult.UNKNOWN, None
-    if os.path.isabs(file_path):
-        resolved = file_path
-    elif cwd:
-        resolved = os.path.join(cwd, file_path)
-    else:
+    if not cwd or not os.path.isabs(cwd):
         return ReadVerificationResult.UNKNOWN, None
     try:
-        disk_content = Path(resolved).read_text(encoding="utf-8")
-    except (OSError, UnicodeDecodeError) as e:
-        logger.debug("disk verification: cannot read %s: %s", resolved, e)
+        resolved_root = Path(cwd).resolve(strict=True)
+    except OSError:
+        return ReadVerificationResult.UNKNOWN, None
+    if not resolved_root.is_dir():
+        return ReadVerificationResult.UNKNOWN, None
+    resolved = _resolve_read_path_in_workspace(file_path, resolved_root)
+    if resolved is None:
+        return ReadVerificationResult.UNKNOWN, None
+    disk_content = _read_disk_content_bounded(resolved, _max_disk_verify_bytes())
+    if disk_content is None:
         return ReadVerificationResult.UNKNOWN, None
     if disk_content == received_content:
         return ReadVerificationResult.COMPLETE, None
@@ -233,7 +310,10 @@ class AstGrepReadOutline:
         truncation = _detect_truncation(tool_output)
         if truncation is None and _verify_truncation_on_disk_enabled():
             verdict, disk_truncation = _verify_read_against_disk(
-                _path_from_input(tool_input), tool_output, get_current_cwd()
+                _path_from_input(tool_input),
+                tool_output,
+                get_current_cwd(),
+                trusted=is_current_request_trusted(),
             )
             if verdict is ReadVerificationResult.TRUNCATED:
                 truncation = disk_truncation
diff --git a/headroom/proxy/project_context.py b/headroom/proxy/project_context.py
index a047325bc..a1d003834 100644
--- a/headroom/proxy/project_context.py
+++ b/headroom/proxy/project_context.py
@@ -17,6 +17,16 @@ second, unsanitized contextvar for consumers that need the literal
 filesystem path (e.g. verifying a Read tool_result against disk). Not
 (yet) bound at the WebSocket accept paths — an absent cwd there is already
 treated as "can't resolve, don't guess."
+
+A third contextvar, ``_current_request_trusted``, records whether the
+active request's peer is loopback — a fact the HTTP middleware observes
+from the TCP connection itself, not something a caller can assert via a
+header. Consumers that turn ``_current_cwd`` into a filesystem read (again,
+disk verification) must gate on this first: the cwd header alone is never
+sufficient authority for touching disk, only a signal to interpret once the
+peer is already known to be trusted. Defaults to ``False`` so any request
+path that never calls the setter (e.g. the WebSocket accept path) is safe
+by construction.
 """
 
 from __future__ import annotations
@@ -41,6 +51,11 @@ _current_project: ContextVar[str | None] = ContextVar("headroom_current_project"
 # tool's file_path and read from disk, so it must stay the literal path.
 _current_cwd: ContextVar[str | None] = ContextVar("headroom_current_cwd", default=None)
 
+# Server-observed (not header-derived) loopback signal — see module docstring.
+_current_request_trusted: ContextVar[bool] = ContextVar(
+    "headroom_current_request_trusted", default=False
+)
+
 
 def set_current_project(project: str | None) -> None:
     """Bind the active request's project for downstream outcome recording."""
@@ -62,6 +77,16 @@ def get_current_cwd() -> str | None:
     return _current_cwd.get()
 
 
+def set_current_request_trusted(trusted: bool) -> None:
+    """Bind whether the active request's peer is loopback."""
+    _current_request_trusted.set(trusted)
+
+
+def is_current_request_trusted() -> bool:
+    """Whether the active request's peer is loopback, or ``False`` if unset."""
+    return _current_request_trusted.get()
+
+
 def strip_project_path_prefix(scope: MutableMapping[str, Any]) -> str | None:
     """Strip a ``/p/<name>`` prefix from an ASGI scope, returning the name.
 
@@ -80,8 +105,10 @@ __all__ = [
     "classify_project",
     "get_current_cwd",
     "get_current_project",
+    "is_current_request_trusted",
     "set_current_cwd",
     "set_current_project",
+    "set_current_request_trusted",
     "split_project_path",
     "strip_project_path_prefix",
     "with_project_prefix",
diff --git a/headroom/proxy/runtime_env.py b/headroom/proxy/runtime_env.py
index b20d1607d..b9d255f34 100644
--- a/headroom/proxy/runtime_env.py
+++ b/headroom/proxy/runtime_env.py
@@ -80,6 +80,11 @@ RUNTIME_ENV_KNOBS: tuple[Knob, ...] = (
         "bool",
         "Verify ast-grep Read truncation against disk when no client banner is found.",
     ),
+    Knob(
+        "HEADROOM_VERIFY_TRUNCATION_MAX_BYTES",
+        "int",
+        "Max on-disk file size (bytes) the disk-verify fallback will read.",
+    ),
 )
 
 _KNOBS_BY_ENV: dict[str, Knob] = {k.env: k for k in RUNTIME_ENV_KNOBS}
diff --git a/headroom/proxy/server.py b/headroom/proxy/server.py
index 4c807127b..f6138bf3e 100644
--- a/headroom/proxy/server.py
+++ b/headroom/proxy/server.py
@@ -162,6 +162,7 @@ from headroom.proxy.project_context import (
     classify_project,
     set_current_cwd,
     set_current_project,
+    set_current_request_trusted,
     strip_project_path_prefix,
 )
 from headroom.proxy.prometheus_metrics import PrometheusMetrics  # noqa: F401
@@ -3263,8 +3264,19 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
         method = request.method
         query = request.url.query
         headers = dict(request.headers.items())
+        client = getattr(request, "client", None)
+        client_addr = ""
+        client_host = None
+        if client is not None:
+            client_host = getattr(client, "host", None)
+            client_port = getattr(client, "port", None)
+            client_addr = f"{client_host}:{client_port}" if client_port else str(client_host)
         set_current_project(classify_project(headers) or prefix_project)
         set_current_cwd(headers.get("x-headroom-cwd"))
+        # Server-observed (not header-derived) trust signal for consumers that
+        # turn x-headroom-cwd into a filesystem read (e.g. astgrep disk
+        # verification) -- the header alone is never sufficient authority.
+        set_current_request_trusted(is_loopback_host(client_host))
         # Path-based Codex identification: stamp X-Client: codex on the
         # Responses endpoint for callers that don't otherwise classify (e.g.
         # Codex Desktop, whose User-Agent isn't a known codex UA). Without it
@@ -3274,12 +3286,6 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
         # makes every downstream classify_client(headers) read "codex".
         if should_stamp_codex_client(path, headers):
             request.scope["headers"].append((b"x-client", b"codex"))
-        client = getattr(request, "client", None)
-        client_addr = ""
-        if client is not None:
-            client_host = getattr(client, "host", None)
-            client_port = getattr(client, "port", None)
-            client_addr = f"{client_host}:{client_port}" if client_port else str(client_host)
         try:
             proxy.metrics.record_inbound_request(method=method, path=path)
         except Exception:
diff --git a/tests/test_proxy_loopback_gating.py b/tests/test_proxy_loopback_gating.py
index 49381d1ef..33dd8c7a5 100644
--- a/tests/test_proxy_loopback_gating.py
+++ b/tests/test_proxy_loopback_gating.py
@@ -745,3 +745,44 @@ def test_dashboard_client_cidr_does_not_expand_other_management_endpoints(
     assert client.get("/admin/upstream").status_code == 404
     assert client.get("/debug/tasks").status_code == 404
     assert client.post("/stats/reset").status_code == 404
+
+
+def _app_with_trust_probe() -> FastAPI:
+    """`_record_headroom_stack` binds `is_current_request_trusted()` from the
+    peer's loopback-ness before any route runs; this test-only route reads
+    it back so the binding itself (not just a downstream consumer like
+    astgrep's disk-verify fallback) can be asserted directly."""
+    from headroom.proxy.project_context import is_current_request_trusted
+
+    app = _make_app()
+
+    @app.get("/__test/trusted")
+    def _trusted_probe() -> dict[str, bool]:
+        return {"trusted": is_current_request_trusted()}
+
+    # A catch-all passthrough route registered by create_app() would
+    # otherwise shadow this path -- Starlette matches routes in
+    # registration order, not by specificity.
+    app.router.routes.insert(0, app.router.routes.pop())
+    return app
+
+
+def test_record_headroom_stack_binds_trusted_true_for_loopback_client() -> None:
+    client = TestClient(
+        _app_with_trust_probe(), base_url="http://127.0.0.1", client=("127.0.0.1", 12345)
+    )
+    resp = client.get("/__test/trusted", headers={"x-headroom-cwd": "/some/spoofed/path"})
+    assert resp.json() == {"trusted": True}
+
+
+def test_record_headroom_stack_binds_trusted_false_for_non_loopback_client() -> None:
+    """A spoofed x-headroom-cwd header from a non-loopback peer must not
+    make the request trusted -- the header carries no authority on its
+    own. Direct regression test for the disk-verify oracle finding."""
+    client = TestClient(
+        _app_with_trust_probe(),
+        base_url="http://100.82.0.2:8787",
+        client=("100.90.0.5", 12345),
+    )
+    resp = client.get("/__test/trusted", headers={"x-headroom-cwd": "/some/spoofed/path"})
+    assert resp.json() == {"trusted": False}
diff --git a/tests/test_tool_result_interceptors.py b/tests/test_tool_result_interceptors.py
index efe39f8c6..bc61e216a 100644
--- a/tests/test_tool_result_interceptors.py
+++ b/tests/test_tool_result_interceptors.py
@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import os
+import sys
 import textwrap
 
 import pytest
@@ -20,7 +22,11 @@ from headroom.proxy.interceptors.astgrep import (
     _verify_read_against_disk,
 )
 from headroom.proxy.interceptors.base import reset_interceptor_failure_counts
-from headroom.proxy.project_context import get_current_cwd, set_current_cwd
+from headroom.proxy.project_context import (
+    get_current_cwd,
+    set_current_cwd,
+    set_current_request_trusted,
+)
 from headroom.tokenizer import Tokenizer
 
 
@@ -370,24 +376,47 @@ def test_set_current_cwd_none_and_blank_both_clear():
 
 class TestVerifyReadAgainstDisk:
     def test_missing_file_path_is_unknown(self):
-        verdict, info = _verify_read_against_disk(None, "abc", "/repo")
+        verdict, info = _verify_read_against_disk(None, "abc", "/repo", trusted=True)
         assert verdict is ReadVerificationResult.UNKNOWN
         assert info is None
 
     def test_relative_path_without_cwd_is_unknown(self):
-        verdict, info = _verify_read_against_disk("payments.py", "abc", None)
+        verdict, info = _verify_read_against_disk("payments.py", "abc", None, trusted=True)
         assert verdict is ReadVerificationResult.UNKNOWN
         assert info is None
 
-    def test_unreadable_file_is_unknown(self, tmp_path):
-        verdict, info = _verify_read_against_disk(str(tmp_path / "missing.py"), "abc", None)
+    def test_untrusted_request_never_touches_disk(self, tmp_path, monkeypatch):
+        """A fully valid cwd/file_path/content combo that would resolve
+        COMPLETE if trusted must short-circuit to UNKNOWN -- and never call
+        os.open -- when the request isn't trusted. The regression guarantee
+        is "untrusted -> UNKNOWN without any filesystem operation," not just
+        "untrusted -> UNKNOWN.\""""
+        f = tmp_path / "payments.py"
+        f.write_text(_PY_FIXTURE, encoding="utf-8")
+        opened: list[object] = []
+        real_open = os.open
+
+        def _tracking_open(*args, **kwargs):
+            opened.append(args)
+            return real_open(*args, **kwargs)
+
+        monkeypatch.setattr(os, "open", _tracking_open)
+        verdict, info = _verify_read_against_disk(str(f), _PY_FIXTURE, str(tmp_path), trusted=False)
+        assert verdict is ReadVerificationResult.UNKNOWN
+        assert info is None
+        assert opened == []
+
+    def test_missing_file_under_workspace_root_is_unknown(self, tmp_path):
+        verdict, info = _verify_read_against_disk(
+            str(tmp_path / "missing.py"), "abc", str(tmp_path), trusted=True
+        )
         assert verdict is ReadVerificationResult.UNKNOWN
         assert info is None
 
     def test_exact_match_is_complete(self, tmp_path):
         f = tmp_path / "payments.py"
         f.write_text(_PY_FIXTURE, encoding="utf-8")
-        verdict, info = _verify_read_against_disk(str(f), _PY_FIXTURE, None)
+        verdict, info = _verify_read_against_disk(str(f), _PY_FIXTURE, str(tmp_path), trusted=True)
         assert verdict is ReadVerificationResult.COMPLETE
         assert info is None
 
@@ -395,7 +424,7 @@ class TestVerifyReadAgainstDisk:
         f = tmp_path / "payments.py"
         f.write_text(_PY_FIXTURE, encoding="utf-8")
         partial = _PY_FIXTURE[:200]
-        verdict, info = _verify_read_against_disk(str(f), partial, None)
+        verdict, info = _verify_read_against_disk(str(f), partial, str(tmp_path), trusted=True)
         assert verdict is ReadVerificationResult.TRUNCATED
         assert info == (len(partial.splitlines()), len(_PY_FIXTURE.splitlines()))
 
@@ -403,7 +432,9 @@ class TestVerifyReadAgainstDisk:
         f = tmp_path / "payments.py"
         f.write_text(_PY_FIXTURE, encoding="utf-8")
         partial = _PY_FIXTURE[:200]
-        verdict, info = _verify_read_against_disk("payments.py", partial, str(tmp_path))
+        verdict, info = _verify_read_against_disk(
+            "payments.py", partial, str(tmp_path), trusted=True
+        )
         assert verdict is ReadVerificationResult.TRUNCATED
         assert info is not None
 
@@ -411,7 +442,78 @@ class TestVerifyReadAgainstDisk:
         # File diverged since the client read it -- not a clean prefix.
         f = tmp_path / "payments.py"
         f.write_text(_PY_FIXTURE.replace("compute_subtotal", "compute_total"), encoding="utf-8")
-        verdict, info = _verify_read_against_disk(str(f), _PY_FIXTURE, None)
+        verdict, info = _verify_read_against_disk(str(f), _PY_FIXTURE, str(tmp_path), trusted=True)
+        assert verdict is ReadVerificationResult.UNKNOWN
+        assert info is None
+
+    def test_absolute_path_outside_workspace_root_is_unknown(self, tmp_path):
+        root = tmp_path / "project"
+        root.mkdir()
+        sibling = tmp_path / "other"
+        sibling.mkdir()
+        f = sibling / "secret.py"
+        f.write_text(_PY_FIXTURE, encoding="utf-8")
+        verdict, info = _verify_read_against_disk(str(f), _PY_FIXTURE, str(root), trusted=True)
+        assert verdict is ReadVerificationResult.UNKNOWN
+        assert info is None
+
+    def test_relative_traversal_escapes_workspace_is_unknown(self, tmp_path):
+        root = tmp_path / "project"
+        root.mkdir()
+        f = tmp_path / "secret.py"
+        f.write_text(_PY_FIXTURE, encoding="utf-8")
+        verdict, info = _verify_read_against_disk(
+            "../secret.py", _PY_FIXTURE, str(root), trusted=True
+        )
+        assert verdict is ReadVerificationResult.UNKNOWN
+        assert info is None
+
+    def test_sibling_directory_sharing_string_prefix_is_not_inside_workspace(self, tmp_path):
+        """`relative_to()` is a path-segment containment check, not a string
+        prefix check -- a target that merely starts with the root's string
+        must not be treated as inside it."""
+        root = tmp_path / "project"
+        root.mkdir()
+        other = tmp_path / "project-other"
+        other.mkdir()
+        f = other / "secret.py"
+        f.write_text(_PY_FIXTURE, encoding="utf-8")
+        verdict, info = _verify_read_against_disk(str(f), _PY_FIXTURE, str(root), trusted=True)
+        assert verdict is ReadVerificationResult.UNKNOWN
+        assert info is None
+
+    @pytest.mark.skipif(
+        sys.platform == "win32", reason="symlinks need elevated privilege on Windows"
+    )
+    def test_symlinked_file_escapes_workspace_is_unknown(self, tmp_path):
+        root = tmp_path / "project"
+        root.mkdir()
+        outside = tmp_path / "outside.py"
+        outside.write_text(_PY_FIXTURE, encoding="utf-8")
+        link = root / "link.py"
+        link.symlink_to(outside)
+        verdict, info = _verify_read_against_disk("link.py", _PY_FIXTURE, str(root), trusted=True)
+        assert verdict is ReadVerificationResult.UNKNOWN
+        assert info is None
+
+    def test_directory_passed_as_file_path_is_unknown(self, tmp_path):
+        # Also proves O_NONBLOCK doesn't need a FIFO fixture to matter: this
+        # exercises the same os.open(..., O_NONBLOCK) path, just rejected by
+        # the S_ISREG check rather than by not blocking on a missing writer.
+        root = tmp_path / "project"
+        subdir = root / "subdir"
+        subdir.mkdir(parents=True)
+        verdict, info = _verify_read_against_disk("subdir", "abc", str(root), trusted=True)
+        assert verdict is ReadVerificationResult.UNKNOWN
+        assert info is None
+
+    def test_oversized_file_exceeding_byte_cap_is_unknown(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HEADROOM_VERIFY_TRUNCATION_MAX_BYTES", "10")
+        f = tmp_path / "payments.py"
+        f.write_text(_PY_FIXTURE, encoding="utf-8")
+        assert len(_PY_FIXTURE.encode("utf-8")) > 10
+        # Content matches exactly -- would be COMPLETE without the cap.
+        verdict, info = _verify_read_against_disk(str(f), _PY_FIXTURE, str(tmp_path), trusted=True)
         assert verdict is ReadVerificationResult.UNKNOWN
         assert info is None
 
@@ -432,6 +534,7 @@ def test_astgrep_disk_verification_flags_truncation_when_no_banner(
     assert "truncated" not in partial.lower()
 
     set_current_cwd(str(tmp_path))
+    set_current_request_trusted(True)
     try:
         messages = [
             {
@@ -453,6 +556,7 @@ def test_astgrep_disk_verification_flags_truncation_when_no_banner(
         result = apply_to_messages(messages, tokenizer)
     finally:
         set_current_cwd(None)
+        set_current_request_trusted(False)
 
     assert len(result.spans) == 1
     new_content = result.messages[1]["content"][0]["content"]
@@ -477,6 +581,47 @@ def test_astgrep_disk_verification_disabled_by_default(tokenizer, tmp_path, monk
     partial = _PY_FIXTURE[: _PY_FIXTURE.index(marker)]
 
     set_current_cwd(str(tmp_path))
+    set_current_request_trusted(True)
+    try:
+        messages = [
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "tool_use",
+                        "id": "abc",
+                        "name": "Read",
+                        "input": {"file_path": "payments.py"},
+                    }
+                ],
+            },
+            {
+                "role": "user",
+                "content": [{"type": "tool_result", "tool_use_id": "abc", "content": partial}],
+            },
+        ]
+        result = apply_to_messages(messages, tokenizer)
+    finally:
+        set_current_cwd(None)
+        set_current_request_trusted(False)
+
+    new_content = result.messages[1]["content"][0]["content"]
+    assert "truncated upstream" not in new_content
+
+
+def test_astgrep_disk_verification_skips_when_request_untrusted(tokenizer, tmp_path, monkeypatch):
+    """Same truncated-on-disk scenario, opted in and cwd bound, but the
+    request's peer was never marked loopback-trusted (the contextvar's
+    False default) -- must behave exactly like the no-signal case."""
+    monkeypatch.setenv("HEADROOM_VERIFY_TRUNCATION_ON_DISK", "1")
+    monkeypatch.setenv("HEADROOM_INTERCEPT_READ_MIN_CHARS", "50")
+    f = tmp_path / "payments.py"
+    f.write_text(_PY_FIXTURE, encoding="utf-8")
+    marker = "\n\ndef format_receipt"
+    partial = _PY_FIXTURE[: _PY_FIXTURE.index(marker)]
+
+    set_current_cwd(str(tmp_path))
+    # Deliberately not calling set_current_request_trusted(True).
     try:
         messages = [
             {