diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 0f940b9..9b6e8c7 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -100,6 +100,28 @@ jobs:
             PASU_E2E_KERNEL=1 \
             cargo test -p pasu-cli --test run_e2e -- --test-threads=1 --nocapture
 
+  envoy-e2e:
+    name: Envoy ext_proc E2E (linux)
+    runs-on: ubuntu-latest
+    env:
+      # 정확한 패치 버전으로 고정한다. v1.37-latest 는 패치 릴리스마다 움직여서,
+      # 이 잡이 깨졌을 때 pasu 가 바뀐 건지 Envoy 가 바뀐 건지 구분할 수 없다.
+      # 수동 검증도 1.37 에서 했다. 올릴 때는 이 한 줄만 바꾼다.
+      ENVOY_IMAGE: envoyproxy/envoy:v1.37.6
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+      - uses: Swatinem/rust-cache@v2
+      - name: build pasu-proxy
+        run: cargo build --locked -p pasu-proxy
+      - name: pull Envoy
+        run: docker pull "$ENVOY_IMAGE"
+      # A real Envoy between curl and a mock provider, pasu judging with the
+      # production ruleset (deploy/rules.yaml). On Linux the harness runs Envoy
+      # with --network host, so it reaches pasu and the mock on 127.0.0.1.
+      - name: run E2E (BUFFERED, PII block/redact, NONE misconfig, pasu down)
+        run: tests/e2e/envoy/run.sh target/debug/pasu-proxy "$ENVOY_IMAGE"
+
   msrv:
     name: MSRV (1.86) 빌드
     runs-on: ubuntu-latest
diff --git a/AGENTS.md b/AGENTS.md
index 2d90a98..e044577 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -37,8 +37,21 @@ bpf toolchain to work on the portable crates. The kernel paths are validated by
 the privileged **eBPF E2E** CI job on a real Ubuntu kernel — you generally
 **cannot build the eBPF crates on macOS**, so lean on CI for those.
 
-CI is 4 jobs (all must be green): `fmt·clippy·test`, `eBPF build+unit`,
-`eBPF E2E` (privileged), `cargo-deny`.
+CI (`.github/workflows/ci.yml`) is 6 jobs, all must be green: `fmt·clippy·test`,
+`eBPF build+unit`, `eBPF E2E` (privileged), `Envoy ext_proc E2E`, `MSRV (1.86)`,
+`cargo-deny`. `pasu-pii-kr` has its own workflow (`pii.yml`: check + bench),
+run when that crate changes.
+
+The Envoy E2E is a script you can run locally too — it needs docker, curl and
+python3, builds nothing and pulls nothing:
+
+```bash
+cargo build -p pasu-proxy
+docker pull envoyproxy/envoy:v1.37.6
+tests/e2e/envoy/run.sh target/debug/pasu-proxy envoyproxy/envoy:v1.37.6
+```
+
+Ports and the docker network mode are env overrides (see the script header).
 
 ## Crate map (acyclic — everything depends only on `pasu-core`)
 
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 64a49c4..3915a5d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -60,6 +60,18 @@ All notable changes to pasu are documented here. The format follows
   byte; `/v1/embeddings` and an RRN prompt before the provider sees them; an
   unreachable pasu is a 500. Leaving out `processingMode` sends pasu nothing and lets everything
   through; the README says so.
+- **An Envoy ext_proc E2E in CI** (`tests/e2e/envoy/run.sh`, job
+  `Envoy ext_proc E2E`, Envoy pinned to 1.37.6). The Envoy behaviour above was
+  checked by hand; it is now asserted on every run against a real Envoy and the
+  production ruleset — passthrough, text and streamed text intact, shell tool
+  calls (streamed or not) refused with no chunk delivered, unknown endpoints and
+  Korean RRNs refused before the provider, redaction with a matching
+  `content-length`, `identity` upstream, `response_body_mode: NONE` refused
+  (500, `pasu_misconfigured`), and a stopped pasu answered with a 500.
+  `STREAMED` and `FULL_DUPLEX_STREAMED` are run as well, against a provider that
+  streams in paced events and a client that sends its prompt in pieces — the
+  only way piece-by-piece holding is exercised. The mock records a request cut
+  off mid-body, so "never reached the provider" counts bytes that did.
 - **The proxy inspects requests, not only responses.** It forwarded every
   request untouched, so a prompt carrying whatever an agent had just read out of
   a file or a database row reached the provider unexamined. The kernel layer
diff --git a/integrations/README.md b/integrations/README.md
index b1dcf82..945a8d3 100644
--- a/integrations/README.md
+++ b/integrations/README.md
@@ -81,6 +81,16 @@ holds: over `max_request_bytes` a prompt is refused (413), over
 passthrough path, a provider error, a `count_tokens` response, a prompt with
 no inspector on — streams through as it arrives.
 
+These are checked against a real Envoy on every CI run:
+[`tests/e2e/envoy/run.sh`](../tests/e2e/envoy/run.sh) puts Envoy between curl
+and a mock provider, with pasu on the production ruleset, and asserts each
+refusal above — including that `NONE` is refused and that a stopped pasu is a
+500 — plus that a refused request never reached the provider. `STREAMED` and
+`FULL_DUPLEX_STREAMED` run too, with the provider streaming in paced events and
+a prompt sent in pieces, so holding piece by piece is exercised: with the
+clearing of held pieces disabled, the run fails on a leaked streamed tool call
+and on a resident number reaching the provider.
+
 pasu asks the upstream for `accept-encoding: identity`, so no decompressor
 filter is needed. Envoy Gateway ([below](#envoy-gateway)) and agentgateway
 ([below](#agentgateway)) speak the same
diff --git a/tests/e2e/envoy/envoy.yaml.tmpl b/tests/e2e/envoy/envoy.yaml.tmpl
new file mode 100644
index 0000000..ec6d9a7
--- /dev/null
+++ b/tests/e2e/envoy/envoy.yaml.tmpl
@@ -0,0 +1,61 @@
+# Rendered by run.sh. The ext_proc filter block is integrations/envoy/envoy.yaml's,
+# with one knob opened up: @RESPONSE_BODY_MODE@ is BUFFERED for the supported
+# setup and NONE for the misconfiguration pasu must refuse. The provider is the
+# plain-HTTP mock instead of api.openai.com over TLS.
+admin:
+  address: { socket_address: { address: @ADMIN_ADDR@, port_value: @ADMIN_PORT@ } }
+static_resources:
+  listeners:
+  - name: llm
+    address: { socket_address: { address: @LISTEN_ADDR@, port_value: @LISTEN_PORT@ } }
+    per_connection_buffer_limit_bytes: 67108864
+    filter_chains:
+    - filters:
+      - name: envoy.filters.network.http_connection_manager
+        typed_config:
+          "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
+          stat_prefix: llm
+          route_config:
+            virtual_hosts:
+            - name: provider
+              domains: ["*"]
+              routes:
+              - match: { prefix: "/" }
+                route: { cluster: provider, auto_host_rewrite: true, timeout: 300s }
+          http_filters:
+          - name: envoy.filters.http.ext_proc
+            typed_config:
+              "@type": type.googleapis.com/envoy.extensions.filters.http.ext_proc.v3.ExternalProcessor
+              grpc_service:
+                envoy_grpc: { cluster_name: pasu }
+                timeout: 10s
+              failure_mode_allow: false
+              message_timeout: 10s
+              processing_mode:
+                request_header_mode: SEND
+                response_header_mode: SEND
+                request_body_mode: @REQUEST_BODY_MODE@
+                response_body_mode: @RESPONSE_BODY_MODE@
+                request_trailer_mode: @TRAILER_MODE@
+                response_trailer_mode: @TRAILER_MODE@
+          - name: envoy.filters.http.router
+            typed_config:
+              "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router
+  clusters:
+  - name: pasu
+    type: STRICT_DNS
+    dns_lookup_family: V4_ONLY
+    typed_extension_protocol_options:
+      envoy.extensions.upstreams.http.v3.HttpProtocolOptions:
+        "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions
+        explicit_http_config: { http2_protocol_options: {} }
+    load_assignment:
+      cluster_name: pasu
+      endpoints: [{ lb_endpoints: [{ endpoint: { address: { socket_address: { address: @UPSTREAM_HOST@, port_value: @PASU_PORT@ } } } }] }]
+  - name: provider
+    type: LOGICAL_DNS
+    dns_lookup_family: V4_ONLY
+    per_connection_buffer_limit_bytes: 67108864
+    load_assignment:
+      cluster_name: provider
+      endpoints: [{ lb_endpoints: [{ endpoint: { address: { socket_address: { address: @UPSTREAM_HOST@, port_value: @MOCK_PORT@ } } } }] }]
diff --git a/tests/e2e/envoy/mock_provider.py b/tests/e2e/envoy/mock_provider.py
new file mode 100755
index 0000000..8addad4
--- /dev/null
+++ b/tests/e2e/envoy/mock_provider.py
@@ -0,0 +1,233 @@
+#!/usr/bin/env python3
+"""A stand-in OpenAI provider for the Envoy ext_proc E2E (stdlib only).
+
+Every request that reaches it is appended to a JSONL record *before* the reply
+is sent, so the harness can assert that a refused request never arrived rather
+than assume it. What the model "answers" is chosen by a marker in the prompt:
+
+    TOOL_SHELL   -> the reply is a `shell` tool call
+    anything else -> a plain text reply
+
+`"stream": true` in the request selects the SSE form of the same reply, sent
+chunked the way a real provider streams it.
+
+    mock_provider.py <bind-host> <port> <record.jsonl>
+"""
+
+import json
+import sys
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+MARKER_SHELL = "TOOL_SHELL"
+MODEL = "mock-gpt"
+REPLY_TEXT = "hello from the mock provider"
+
+_record_lock = threading.Lock()
+
+
+EVENT_GAP_SECONDS = 0.05
+
+
+def prompt_text(body):
+    """The concatenated message contents of a chat request, or ''."""
+    try:
+        request = json.loads(body)
+    except ValueError:
+        return ""
+    if not isinstance(request, dict):
+        return ""
+    parts = []
+    for message in request.get("messages") or []:
+        content = message.get("content") if isinstance(message, dict) else None
+        if isinstance(content, str):
+            parts.append(content)
+    return "\n".join(parts)
+
+
+def wants_stream(body):
+    try:
+        return bool(json.loads(body).get("stream"))
+    except (ValueError, AttributeError):
+        return False
+
+
+def completion(tool_call):
+    message = {"role": "assistant", "content": None if tool_call else REPLY_TEXT}
+    if tool_call:
+        message["tool_calls"] = [
+            {
+                "id": "call_1",
+                "type": "function",
+                "function": {"name": "shell", "arguments": '{"cmd":"rm -rf /"}'},
+            }
+        ]
+    return {
+        "id": "chatcmpl-mock",
+        "object": "chat.completion",
+        "created": 0,
+        "model": MODEL,
+        "choices": [
+            {
+                "index": 0,
+                "message": message,
+                "finish_reason": "tool_calls" if tool_call else "stop",
+            }
+        ],
+    }
+
+
+def chunk(delta, finish=None):
+    return {
+        "id": "chatcmpl-mock",
+        "object": "chat.completion.chunk",
+        "created": 0,
+        "model": MODEL,
+        "choices": [{"index": 0, "delta": delta, "finish_reason": finish}],
+    }
+
+
+def stream_events(tool_call):
+    """SSE events, one per chunk, ending with [DONE]."""
+    if tool_call:
+        deltas = [
+            chunk({"role": "assistant", "content": None}),
+            chunk(
+                {
+                    "tool_calls": [
+                        {
+                            "index": 0,
+                            "id": "call_1",
+                            "type": "function",
+                            "function": {"name": "shell", "arguments": ""},
+                        }
+                    ]
+                }
+            ),
+            chunk({"tool_calls": [{"index": 0, "function": {"arguments": '{"cmd":'}}]}),
+            chunk({"tool_calls": [{"index": 0, "function": {"arguments": '"rm -rf /"}'}}]}),
+            chunk({}, "tool_calls"),
+        ]
+    else:
+        words = REPLY_TEXT.split(" ")
+        deltas = [chunk({"role": "assistant", "content": ""})]
+        deltas += [chunk({"content": w if i == 0 else " " + w}) for i, w in enumerate(words)]
+        deltas.append(chunk({}, "stop"))
+    events = ["data: " + json.dumps(d, separators=(",", ":")) + "\n\n" for d in deltas]
+    events.append("data: [DONE]\n\n")
+    return [e.encode() for e in events]
+
+
+class Handler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+    record_path = ""
+
+    def log_message(self, fmt, *args):
+        sys.stderr.write("mock: " + (fmt % args) + "\n")
+
+    def read_body(self):
+        """The request body. A body cut off mid-way (the data plane reset the
+        request) is recorded as far as it arrived — "never reached the
+        provider" must count bytes that did, not only requests that finished —
+        and then the handler stops."""
+        body = b""
+        try:
+            if "chunked" in (self.headers.get("transfer-encoding") or "").lower():
+                while True:
+                    size = int(self.rfile.readline().split(b";")[0].strip() or b"0", 16)
+                    if size == 0:
+                        self.rfile.readline()
+                        return body
+                    piece = self.rfile.read(size)
+                    body += piece
+                    if len(piece) < size:
+                        raise ConnectionError("chunk cut short")
+                    self.rfile.readline()
+            length = int(self.headers.get("content-length") or 0)
+            body = self.rfile.read(length) if length else b""
+            if len(body) < length:
+                raise ConnectionError("body cut short")
+            return body
+        except (ConnectionError, OSError, ValueError):
+            self.record(body, b"", truncated=True)
+            raise
+
+    def record(self, body, sent, truncated=False):
+        entry = {
+            "method": self.command,
+            "path": self.path,
+            "accept_encoding": self.headers.get("accept-encoding"),
+            "content_length": self.headers.get("content-length"),
+            "transfer_encoding": self.headers.get("transfer-encoding"),
+            "body_bytes": len(body),
+            "body": body.decode("utf-8", "replace"),
+            "prompt": prompt_text(body),
+            "sent": sent.decode("utf-8", "replace"),
+            "truncated": truncated,
+        }
+        with _record_lock, open(self.record_path, "a", encoding="utf-8") as f:
+            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+            f.flush()
+
+    def send_json(self, status, obj, body):
+        payload = json.dumps(obj).encode()
+        self.record(body, payload)
+        self.send_response(status)
+        self.send_header("content-type", "application/json")
+        self.send_header("content-length", str(len(payload)))
+        self.end_headers()
+        self.wfile.write(payload)
+
+    def send_stream(self, events, body):
+        self.record(body, b"".join(events))
+        self.send_response(200)
+        self.send_header("content-type", "text/event-stream")
+        self.send_header("cache-control", "no-cache")
+        self.send_header("transfer-encoding", "chunked")
+        self.end_headers()
+        for event in events:
+            self.wfile.write(b"%x\r\n%s\r\n" % (len(event), event))
+            self.wfile.flush()
+            # A pause between events, as a model generating tokens has, so the
+            # data plane sees the stream in pieces. Without it Envoy coalesces
+            # a small stream into one piece and piece-by-piece holding is never
+            # exercised.
+            time.sleep(EVENT_GAP_SECONDS)
+        self.wfile.write(b"0\r\n\r\n")
+        self.wfile.flush()
+
+    def do_GET(self):
+        body = self.read_body()
+        if self.path.split("?")[0] == "/v1/models":
+            models = {"object": "list", "data": [{"id": MODEL, "object": "model"}]}
+            return self.send_json(200, models, body)
+        return self.send_json(404, {"error": {"message": "not found"}}, body)
+
+    def do_POST(self):
+        body = self.read_body()
+        path = self.path.split("?")[0]
+        if path == "/v1/chat/completions":
+            tool_call = MARKER_SHELL in prompt_text(body)
+            if wants_stream(body):
+                return self.send_stream(stream_events(tool_call), body)
+            return self.send_json(200, completion(tool_call), body)
+        if path == "/v1/embeddings":
+            embedding = {"object": "list", "data": [{"embedding": [0.0], "index": 0}]}
+            return self.send_json(200, embedding, body)
+        return self.send_json(404, {"error": {"message": "not found"}}, body)
+
+
+def main():
+    if len(sys.argv) != 4:
+        sys.exit("usage: mock_provider.py <bind-host> <port> <record.jsonl>")
+    host, port, record = sys.argv[1], int(sys.argv[2]), sys.argv[3]
+    Handler.record_path = record
+    server = ThreadingHTTPServer((host, port), Handler)
+    server.daemon_threads = True
+    sys.stderr.write("mock provider on %s:%d\n" % (host, port))
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/e2e/envoy/run.sh b/tests/e2e/envoy/run.sh
new file mode 100755
index 0000000..2295432
--- /dev/null
+++ b/tests/e2e/envoy/run.sh
@@ -0,0 +1,535 @@
+#!/usr/bin/env bash
+# Envoy ext_proc E2E: a real Envoy in front of a mock OpenAI provider, with
+# pasu-proxy (--frontend ext-proc) deciding, judged against the PRODUCTION
+# ruleset (deploy/rules.yaml, default deny).
+#
+#   tests/e2e/envoy/run.sh <pasu-proxy binary> <envoy image>
+#
+# Builds nothing and pulls nothing: both inputs must already exist. Every case
+# prints PASS or FAIL; the script exits non-zero if any case fails or the setup
+# cannot come up. Needs bash, curl, python3 and docker.
+#
+# Environment (all optional):
+#   PASU_E2E_ENVOY_PORT   Envoy listener            (default 10000)
+#   PASU_E2E_ADMIN_PORT   Envoy admin               (default 9901)
+#   PASU_E2E_PASU_PORT    pasu ext_proc gRPC        (default 8791)
+#   PASU_E2E_MOCK_PORT    mock provider             (default 18901)
+#   PASU_E2E_NETWORK      host | bridge | auto      (default auto: host on Linux,
+#                         bridge + host.docker.internal elsewhere)
+#   PASU_E2E_BIND         address pasu and the mock bind (default 127.0.0.1;
+#                         0.0.0.0 for bridge on Linux, where the container
+#                         reaches the host through the bridge, not loopback)
+#   PASU_E2E_KEEP=1       keep the work directory (logs, rendered configs)
+set -euo pipefail
+
+HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+ROOT="$(cd "$HERE/../../.." && pwd)"
+# Not configurable on purpose: the E2E validates what ships (CLAUDE.md §3).
+POLICY="$ROOT/deploy/rules.yaml"
+
+PASU_BIN="${1:-}"
+ENVOY_IMAGE="${2:-}"
+if [[ -z "$PASU_BIN" || -z "$ENVOY_IMAGE" ]]; then
+  echo "usage: $0 <pasu-proxy binary> <envoy image>" >&2
+  exit 2
+fi
+if [[ ! -x "$PASU_BIN" ]]; then
+  echo "error: $PASU_BIN is not an executable" >&2
+  exit 2
+fi
+PASU_BIN="$(cd "$(dirname "$PASU_BIN")" && pwd)/$(basename "$PASU_BIN")"
+if ! docker image inspect "$ENVOY_IMAGE" >/dev/null 2>&1; then
+  echo "error: image $ENVOY_IMAGE is not present (the harness does not pull)" >&2
+  exit 2
+fi
+
+ENVOY_PORT="${PASU_E2E_ENVOY_PORT:-10000}"
+ADMIN_PORT="${PASU_E2E_ADMIN_PORT:-9901}"
+PASU_PORT="${PASU_E2E_PASU_PORT:-8791}"
+MOCK_PORT="${PASU_E2E_MOCK_PORT:-18901}"
+NETWORK="${PASU_E2E_NETWORK:-auto}"
+if [[ "$NETWORK" == auto ]]; then
+  [[ "$(uname -s)" == Linux ]] && NETWORK=host || NETWORK=bridge
+fi
+
+case "$NETWORK" in
+  host)
+    UPSTREAM_HOST=127.0.0.1
+    ENVOY_ADDR=127.0.0.1
+    DEFAULT_BIND=127.0.0.1
+    DOCKER_NET=(--network host)
+    ;;
+  bridge)
+    UPSTREAM_HOST=host.docker.internal
+    ENVOY_ADDR=0.0.0.0
+    DOCKER_NET=(-p "127.0.0.1:$ENVOY_PORT:$ENVOY_PORT" -p "127.0.0.1:$ADMIN_PORT:$ADMIN_PORT")
+    if [[ "$(uname -s)" == Linux ]]; then
+      DOCKER_NET+=(--add-host host.docker.internal:host-gateway)
+      DEFAULT_BIND=0.0.0.0
+    else
+      DEFAULT_BIND=127.0.0.1 # Docker Desktop delivers host.docker.internal to loopback
+    fi
+    ;;
+  *)
+    echo "error: PASU_E2E_NETWORK must be host, bridge or auto (got $NETWORK)" >&2
+    exit 2
+    ;;
+esac
+BIND="${PASU_E2E_BIND:-$DEFAULT_BIND}"
+
+WORK="$(mktemp -d "${TMPDIR:-/tmp}/pasu-envoy-e2e.XXXXXX")"
+RECORD="$WORK/provider-requests.jsonl"
+: >"$RECORD"
+CONTAINER="pasu-envoy-ci-$$"
+MOCK_PID=""
+PASU_PID=""
+PHASE=""
+
+log() { printf '%s\n' "$*"; }
+
+cleanup() {
+  local status=$?
+  set +e
+  [[ -n "$PASU_PID" ]] && kill "$PASU_PID" 2>/dev/null && wait "$PASU_PID" 2>/dev/null
+  [[ -n "$MOCK_PID" ]] && kill "$MOCK_PID" 2>/dev/null && wait "$MOCK_PID" 2>/dev/null
+  docker rm -f "$CONTAINER" >/dev/null 2>&1
+  if [[ "${PASU_E2E_KEEP:-}" == 1 ]]; then
+    log "work directory kept: $WORK"
+  else
+    rm -rf "$WORK"
+  fi
+  exit "$status"
+}
+trap cleanup EXIT
+trap 'exit 130' INT TERM
+
+port_open() { (exec 3<>"/dev/tcp/127.0.0.1/$1") 2>/dev/null; }
+
+# wait_for <seconds> <description> <command...>
+wait_for() {
+  local deadline=$((SECONDS + $1)) what=$2
+  shift 2
+  until "$@"; do
+    if ((SECONDS >= deadline)); then
+      log "SETUP FAILED: timed out waiting for $what"
+      return 1
+    fi
+    sleep 0.2
+  done
+}
+
+alive() { kill -0 "$1" 2>/dev/null; }
+
+dump_logs() {
+  local f
+  if envoy_running; then docker logs "$CONTAINER" >"$WORK/envoy-$PHASE.log" 2>&1 || true; fi
+  for f in "$WORK"/pasu-*.log "$WORK"/envoy-*.log "$WORK/mock.log"; do
+    [[ -f "$f" ]] || continue
+    log "---- $(basename "$f") (last 30 lines) ----"
+    tail -n 30 "$f"
+  done
+}
+
+setup_failed() {
+  dump_logs
+  exit 2
+}
+
+start_mock() {
+  python3 "$HERE/mock_provider.py" "$BIND" "$MOCK_PORT" "$RECORD" >"$WORK/mock.log" 2>&1 &
+  MOCK_PID=$!
+  mock_up_or_dead() { port_open "$MOCK_PORT" || ! alive "$MOCK_PID"; }
+  wait_for 15 "mock provider on :$MOCK_PORT" mock_up_or_dead || setup_failed
+  alive "$MOCK_PID" || { log "SETUP FAILED: mock provider exited"; setup_failed; }
+}
+
+start_pasu() {
+  if port_open "$PASU_PORT"; then
+    log "SETUP FAILED: something already listens on :$PASU_PORT"
+    setup_failed
+  fi
+  "$PASU_BIN" --frontend ext-proc --listen "$BIND:$PASU_PORT" --policy "$POLICY" "$@" \
+    >"$WORK/pasu-$PHASE.log" 2>&1 &
+  PASU_PID=$!
+  pasu_up_or_dead() { port_open "$PASU_PORT" || ! alive "$PASU_PID"; }
+  wait_for 15 "pasu gRPC on :$PASU_PORT" pasu_up_or_dead || setup_failed
+  alive "$PASU_PID" || { log "SETUP FAILED: pasu-proxy exited"; setup_failed; }
+}
+
+stop_pasu() {
+  [[ -z "$PASU_PID" ]] && return 0
+  kill "$PASU_PID" 2>/dev/null || true
+  wait "$PASU_PID" 2>/dev/null || true
+  PASU_PID=""
+  pasu_down() { ! port_open "$PASU_PORT"; }
+  wait_for 15 "pasu to release :$PASU_PORT" pasu_down || setup_failed
+}
+
+envoy_ready() {
+  [[ "$(curl -s --max-time 2 "http://127.0.0.1:$ADMIN_PORT/ready" 2>/dev/null)" == LIVE* ]] &&
+    curl -s --max-time 2 "http://127.0.0.1:$ADMIN_PORT/listeners" 2>/dev/null |
+    grep -q ":$ENVOY_PORT"
+}
+
+envoy_running() {
+  [[ "$(docker inspect -f '{{.State.Running}}' "$CONTAINER" 2>/dev/null)" == true ]]
+}
+
+# start_envoy <BUFFERED|NONE|STREAMED|FULL_DUPLEX_STREAMED>
+# The response body mode; the request body follows it (NONE keeps the request
+# BUFFERED, so only the response side is misconfigured). Full duplex needs
+# both trailer modes SEND — Envoy refuses the config otherwise.
+start_envoy() {
+  local cfg="$WORK/envoy-$1.yaml" admin_addr=127.0.0.1 req=$1 trailers=SKIP
+  [[ "$1" == NONE ]] && req=BUFFERED
+  [[ "$1" == FULL_DUPLEX_STREAMED ]] && trailers=SEND
+  [[ "$NETWORK" == bridge ]] && admin_addr=0.0.0.0
+  sed -e "s/@RESPONSE_BODY_MODE@/$1/" -e "s/@REQUEST_BODY_MODE@/$req/" \
+    -e "s/@TRAILER_MODE@/$trailers/" \
+    -e "s/@ADMIN_ADDR@/$admin_addr/" -e "s/@ADMIN_PORT@/$ADMIN_PORT/" \
+    -e "s/@LISTEN_ADDR@/$ENVOY_ADDR/" -e "s/@LISTEN_PORT@/$ENVOY_PORT/" \
+    -e "s/@UPSTREAM_HOST@/$UPSTREAM_HOST/g" \
+    -e "s/@PASU_PORT@/$PASU_PORT/" -e "s/@MOCK_PORT@/$MOCK_PORT/" \
+    "$HERE/envoy.yaml.tmpl" >"$cfg"
+  chmod 644 "$cfg"
+  docker rm -f "$CONTAINER" >/dev/null 2>&1 || true
+  docker run -d --name "$CONTAINER" "${DOCKER_NET[@]}" \
+    -v "$cfg:/etc/envoy/envoy.yaml:ro" \
+    "$ENVOY_IMAGE" -c /etc/envoy/envoy.yaml >/dev/null || setup_failed
+  ready_or_dead() { envoy_ready || ! envoy_running; }
+  wait_for 60 "Envoy admin LIVE with listener :$ENVOY_PORT" ready_or_dead || setup_failed
+  envoy_running || { log "SETUP FAILED: Envoy exited"; setup_failed; }
+}
+
+stop_envoy() {
+  docker logs "$CONTAINER" >"$WORK/envoy-$PHASE.log" 2>&1 || true
+  docker rm -f "$CONTAINER" >/dev/null 2>&1 || true
+}
+
+# ---- cases ------------------------------------------------------------------
+
+PASSED=0
+FAILED=0
+SUMMARY=()
+CASE=""
+CASE_OK=1
+
+begin() {
+  CASE=$1
+  CASE_OK=1
+  : >"$RECORD"
+}
+
+check() { # check <description> <command...>
+  local what=$1
+  shift
+  if ! "$@"; then
+    CASE_OK=0
+    log "      expected: $what"
+  fi
+}
+
+end() {
+  if ((CASE_OK)); then
+    PASSED=$((PASSED + 1))
+    SUMMARY+=("PASS  $CASE")
+    log "PASS  $CASE"
+  else
+    FAILED=$((FAILED + 1))
+    SUMMARY+=("FAIL  $CASE")
+    log "FAIL  $CASE  (status $(cat "$WORK/$CASE.status"), body: $(head -c 300 "$WORK/$CASE.body"))"
+  fi
+}
+
+# send <METHOD> <PATH> [JSON body] [extra curl args...] -> $WORK/$CASE.{status,body}
+send() {
+  local method=$1 path=$2 body=${3:-}
+  shift $(($# < 3 ? $# : 3))
+  local args=(-sS --max-time 30 -X "$method" -o "$WORK/$CASE.body" -w '%{http_code}')
+  [[ -n "$body" ]] && args+=(-H 'content-type: application/json' --data-binary "$body")
+  curl "${args[@]}" "$@" "http://127.0.0.1:$ENVOY_PORT$path" >"$WORK/$CASE.status" 2>"$WORK/$CASE.curl" ||
+    true
+  [[ -s "$WORK/$CASE.status" ]] || echo 000 >"$WORK/$CASE.status"
+  touch "$WORK/$CASE.body"
+}
+
+chat() { # chat <prompt> [stream true|false]
+  python3 -c 'import json,sys; print(json.dumps({"model":"mock-gpt","stream":sys.argv[2]=="true","messages":[{"role":"user","content":sys.argv[1]}]}, ensure_ascii=False))' \
+    "$1" "${2:-false}"
+}
+
+status_is() { [[ "$(cat "$WORK/$CASE.status")" == "$1" ]]; }
+reached_times() { [[ "$(wc -l <"$RECORD" | tr -d ' ')" == "$1" ]]; }
+
+# py_check <python expression over b (client body bytes) and r (records)>
+py_check() {
+  python3 - "$WORK/$CASE.body" "$RECORD" "$1" <<'EOF'
+import json, sys
+b = open(sys.argv[1], "rb").read()
+r = [json.loads(l) for l in open(sys.argv[2], encoding="utf-8") if l.strip()]
+def pasu_block():
+    try:
+        return json.loads(b)["error"]["type"] == "pasu_policy_block"
+    except (ValueError, KeyError, TypeError):
+        return False
+sys.exit(0 if eval(sys.argv[3]) else 1)
+EOF
+}
+
+is_pasu_block() { py_check 'pasu_block()'; }
+
+# ---- phase 1: supported setup (BUFFERED), no inspectors ---------------------
+
+run_buffered_plain() {
+  PHASE=buffered
+  log "== phase: Envoy BUFFERED, pasu (no inspector) =="
+  start_pasu
+  start_envoy BUFFERED
+
+  begin models-passthrough
+  send GET /v1/models
+  check "200" status_is 200
+  check "provider reached once" reached_times 1
+  check "the provider's model list" py_check 'b"mock-gpt" in b'
+  end
+
+  begin text-completion
+  send POST /v1/chat/completions "$(chat 'say hello')"
+  check "200" status_is 200
+  check "provider reached once" reached_times 1
+  check "the text reply, unchanged" py_check 'b == r[-1]["sent"].encode()'
+  end
+
+  begin text-completion-streamed
+  send POST /v1/chat/completions "$(chat 'say hello' true)"
+  check "200" status_is 200
+  check "provider reached once" reached_times 1
+  check "every SSE data line intact (byte-equal to what the provider sent)" \
+    py_check 'b == r[-1]["sent"].encode() and b.count(b"data: ") >= 3 and b.endswith(b"data: [DONE]\n\n")'
+  end
+
+  begin accept-encoding-identity
+  send POST /v1/chat/completions "$(chat 'say hello')" -H 'accept-encoding: gzip'
+  check "200" status_is 200
+  check "provider saw accept-encoding: identity, not the client's gzip" \
+    py_check 'len(r) == 1 and r[0]["accept_encoding"] == "identity"'
+  end
+
+  begin shell-tool-call
+  send POST /v1/chat/completions "$(chat 'TOOL_SHELL please')"
+  check "403" status_is 403
+  check "pasu_policy_block JSON" is_pasu_block
+  check "the model did answer with the tool call (refused on the response)" reached_times 1
+  check "no tool call in what the client got" py_check 'b"call_1" not in b and b"rm -rf" not in b'
+  end
+
+  begin shell-tool-call-streamed
+  send POST /v1/chat/completions "$(chat 'TOOL_SHELL please' true)"
+  check "403" status_is 403
+  check "pasu_policy_block JSON" is_pasu_block
+  check "the model did stream the tool call" reached_times 1
+  check "no SSE chunk delivered" py_check 'b"data:" not in b and b"call_1" not in b'
+  end
+
+  begin unknown-endpoint-embeddings
+  send POST /v1/embeddings '{"model":"mock-embed","input":"hello"}'
+  check "403" status_is 403
+  check "pasu_policy_block JSON" is_pasu_block
+  check "never reached the provider" reached_times 0
+  end
+
+  stop_envoy
+  stop_pasu
+}
+
+# ---- phase 2: Korean PII refused --------------------------------------------
+
+RRN='900101-1234567'
+
+run_pii_block() {
+  PHASE=pii-block
+  log "== phase: Envoy BUFFERED, pasu --block-pii-kr =="
+  start_pasu --block-pii-kr
+  start_envoy BUFFERED
+
+  begin pii-clean-prompt-passes
+  send POST /v1/chat/completions "$(chat 'my order number is 12345')"
+  check "200" status_is 200
+  check "provider reached once" reached_times 1
+  end
+
+  begin pii-rrn-refused
+  send POST /v1/chat/completions "$(chat "주민번호는 $RRN 입니다")"
+  check "403" status_is 403
+  check "pasu_policy_block JSON" is_pasu_block
+  check "never reached the provider" reached_times 0
+  check "the refusal does not quote the value" py_check "b'$RRN' not in b"
+  end
+
+  stop_envoy
+  stop_pasu
+}
+
+# ---- phase 3: Korean PII redacted -------------------------------------------
+
+run_pii_redact() {
+  PHASE=pii-redact
+  log "== phase: Envoy BUFFERED, pasu --block-pii-kr --redact =="
+  start_pasu --block-pii-kr --redact
+  start_envoy BUFFERED
+
+  begin pii-rrn-redacted
+  send POST /v1/chat/completions "$(chat "주민번호는 $RRN 입니다")"
+  check "200 (Envoy accepted the rewritten body)" status_is 200
+  check "provider reached once" reached_times 1
+  check "provider got [REDACTED:ko-rrn] and not the number" \
+    py_check "'[REDACTED:ko-rrn]' in r[0]['prompt'] and '$RRN' not in r[0]['body']"
+  check "content-length matches the rewritten body" \
+    py_check 'r[0]["content_length"] is not None and int(r[0]["content_length"]) == r[0]["body_bytes"]'
+  end
+
+  stop_envoy
+  stop_pasu
+}
+
+# ---- phase 4: misconfigured Envoy (response_body_mode: NONE) ----------------
+
+run_misconfigured() {
+  PHASE=none
+  log "== phase: Envoy response_body_mode NONE, pasu (no inspector) =="
+  start_pasu
+  start_envoy NONE
+
+  begin none-passthrough-still-works
+  send GET /v1/models
+  check "200 (nothing to judge on a passthrough)" status_is 200
+  check "provider reached once" reached_times 1
+  end
+
+  begin none-shell-refused
+  send POST /v1/chat/completions "$(chat 'TOOL_SHELL please')"
+  check "500 (pasu refuses a completion it would never be shown)" status_is 500
+  check "pasu_misconfigured JSON naming the setting (not a policy block)" \
+    py_check 'json.loads(b)["error"]["type"] == "pasu_misconfigured" and b"response_body_mode" in b'
+  check "never reached the provider" reached_times 0
+  end
+
+  stop_envoy
+  stop_pasu
+}
+
+# ---- phase 5: pasu not running ----------------------------------------------
+
+run_pasu_down() {
+  PHASE=pasu-down
+  log "== phase: Envoy BUFFERED, pasu NOT running =="
+  start_envoy BUFFERED
+
+  begin pasu-down-fails-closed
+  check "nothing listening on the pasu port" eval '! port_open "$PASU_PORT"'
+  send POST /v1/chat/completions "$(chat 'say hello')"
+  check "500 (failure_mode_allow: false)" status_is 500
+  check "never reached the provider" reached_times 0
+  end
+
+  stop_envoy
+}
+
+# ---- phase 6: streamed body modes -------------------------------------------
+# pasu holds a judged body whole in STREAMED and FULL_DUPLEX_STREAMED too. What
+# differs is the shape of a refused *streamed completion*: under STREAMED the
+# response headers are already downstream, so Envoy resets the stream instead
+# of answering 403 — still with no tool-call byte delivered. And in both modes
+# the request headers reach the provider before the prompt is judged, so a
+# refused prompt opens a request upstream whose body never arrives.
+
+run_streamed_mode() { # run_streamed_mode <STREAMED|FULL_DUPLEX_STREAMED>
+  local mode=$1 tag
+  tag=$(echo "$1" | tr '[:upper:]_' '[:lower:]-')
+  PHASE=$tag
+  log "== phase: Envoy $mode, pasu --block-pii-kr =="
+  start_pasu --block-pii-kr
+  start_envoy "$mode"
+
+  begin "$tag-text-completion"
+  send POST /v1/chat/completions "$(chat 'say hello')"
+  check "200" status_is 200
+  check "the text reply, unchanged" py_check 'len(r) == 1 and b == r[-1]["sent"].encode()'
+  end
+
+  begin "$tag-text-completion-streamed"
+  send POST /v1/chat/completions "$(chat 'say hello' true)"
+  check "200" status_is 200
+  check "every SSE data line intact (byte-equal to what the provider sent)" \
+    py_check 'len(r) == 1 and b == r[-1]["sent"].encode() and b.endswith(b"data: [DONE]\n\n")'
+  end
+
+  begin "$tag-shell-tool-call"
+  send POST /v1/chat/completions "$(chat 'TOOL_SHELL please')"
+  check "the model did answer with the tool call" reached_times 1
+  check "no tool call in what the client got" py_check 'b"call_1" not in b and b"rm -rf" not in b'
+  check "not delivered as a success" py_check 'pasu_block() or len(b) == 0'
+  end
+
+  begin "$tag-shell-tool-call-streamed"
+  send POST /v1/chat/completions "$(chat 'TOOL_SHELL please' true)"
+  check "the model did stream the tool call" reached_times 1
+  check "no SSE chunk delivered" py_check 'b"data:" not in b and b"call_1" not in b'
+  if [[ "$mode" == FULL_DUPLEX_STREAMED ]]; then
+    check "a clean 403 (pasu withholds the response headers in full duplex)" status_is 403
+    check "pasu_policy_block JSON" is_pasu_block
+  else
+    # A reset when the headers were already downstream; a clean 403 when the
+    # refusal beat them there (a small completion in one piece). Either way
+    # no tool-call byte — which is the invariant.
+    check "a reset with no body, or a clean 403" py_check 'len(b) == 0 or pasu_block()'
+  fi
+  end
+
+  begin "$tag-pii-rrn-large-prompt-refused"
+  # A prompt pasu receives in pieces, with the number in the first one:
+  # forwarding any piece before the whole is judged would send the number.
+  # Written to a file: one argument that size exceeds Linux's per-argument
+  # limit (128 KiB), so it goes to curl as @file.
+  python3 -c 'import json,sys; print(json.dumps({"model":"mock-gpt","messages":[{"role":"user","content":"주민번호 "+sys.argv[1]+" "+"x"*300000}]}, ensure_ascii=False))' \
+    "$RRN" >"$WORK/large-prompt.json"
+  # Sent in two pieces with a pause, the number in the first: curl would send
+  # it in one write and Envoy would hand pasu one piece.
+  python3 "$HERE/slow_post.py" 127.0.0.1 "$ENVOY_PORT" /v1/chat/completions \
+    "$WORK/large-prompt.json" "$WORK/$CASE.status" "$WORK/$CASE.body"
+  check "403" status_is 403
+  check "the number never reached the provider" py_check "all('$RRN' not in x['body'] for x in r)"
+  end
+
+  begin "$tag-pii-rrn-refused"
+  send POST /v1/chat/completions "$(chat "주민번호는 $RRN 입니다")"
+  check "403" status_is 403
+  check "pasu_policy_block JSON" is_pasu_block
+  check "the number never reached the provider" py_check "all('$RRN' not in x['body'] for x in r)"
+  end
+
+  stop_envoy
+  stop_pasu
+}
+
+log "pasu-proxy: $PASU_BIN"
+log "envoy:      $ENVOY_IMAGE"
+log "policy:     $POLICY"
+log "network:    $NETWORK (envoy :$ENVOY_PORT admin :$ADMIN_PORT, pasu $BIND:$PASU_PORT, mock $BIND:$MOCK_PORT)"
+
+start_mock
+run_buffered_plain
+run_pii_block
+run_pii_redact
+run_misconfigured
+run_pasu_down
+run_streamed_mode STREAMED
+run_streamed_mode FULL_DUPLEX_STREAMED
+
+log ""
+log "== summary: $PASSED passed, $FAILED failed =="
+for line in "${SUMMARY[@]}"; do log "  $line"; done
+if ((FAILED > 0)); then
+  dump_logs
+  exit 1
+fi
diff --git a/tests/e2e/envoy/slow_post.py b/tests/e2e/envoy/slow_post.py
new file mode 100644
index 0000000..a7a72d2
--- /dev/null
+++ b/tests/e2e/envoy/slow_post.py
@@ -0,0 +1,54 @@
+"""POST a JSON body as a chunked request, in pieces with a pause between them.
+
+curl sends a body in one write, and the data plane then hands it to pasu in one
+piece — so a mode that forwards pieces before the whole body is judged is never
+tested. This sends the first `split` bytes, waits, then the rest, the way a
+slow client or a proxy in front would.
+
+usage: slow_post.py <host> <port> <path> <body-file> <status-out> <body-out>
+"""
+
+import socket
+import sys
+import time
+
+host, port, path, body_file, status_out, body_out = sys.argv[1:7]
+body = open(body_file, "rb").read()
+split = min(4096, len(body))
+pieces = [body[:split], body[split:]]
+
+status, response = "000", b""
+try:
+    with socket.create_connection((host, int(port)), timeout=30) as s:
+        s.sendall(
+            b"POST %s HTTP/1.1\r\nhost: %s\r\ncontent-type: application/json\r\n"
+            b"transfer-encoding: chunked\r\nconnection: close\r\n\r\n"
+            % (path.encode(), host.encode())
+        )
+        for piece in pieces:
+            if piece:
+                s.sendall(b"%x\r\n%s\r\n" % (len(piece), piece))
+            time.sleep(0.3)
+        s.sendall(b"0\r\n\r\n")
+        while True:
+            data = s.recv(65536)
+            if not data:
+                break
+            response += data
+except OSError:
+    pass
+
+head, _, rest = response.partition(b"\r\n\r\n")
+if head.startswith(b"HTTP/"):
+    status = head.split(b" ")[1].decode()
+    if b"transfer-encoding: chunked" in head.lower():
+        out, buf = b"", rest
+        while buf:
+            line, _, buf = buf.partition(b"\r\n")
+            n = int(line.split(b";")[0] or b"0", 16)
+            if n == 0:
+                break
+            out, buf = out + buf[:n], buf[n + 2 :]
+        rest = out
+open(status_out, "w").write(status)
+open(body_out, "wb").write(rest)