Hands-On: Build and Test a Passive Proxy Lab
The main Part 6 article explains why I use a passive model proxy as a debugging instrument. This companion gives you a small version you can run, break, and inspect without pointing it at a real model or putting private prompt content into the retained metrics.
The lab has three pieces:
passive_proxy_lab.pyforwards HTTP request and response bodies and writes content-free JSONL metrics.fake_model_upstream.pybehaves like a tiny deterministic chat-completion endpoint.test_passive_proxy_lab.pyproves the important claims with six canary tests.
All three use the Python standard library. There is no package installation step.
What This Lab Proves
The tests do not try to prove that this is a production model gateway. They prove a smaller set of properties that are easy to understand and hard to fake accidentally.
| Property | Canary |
|---|---|
| The request body is passive | The fake upstream must receive exactly the bytes submitted to the proxy |
| The response body is passive | The caller must receive exactly the body bytes produced by the fake upstream |
| Routine metrics are content-free | Prompt, authorization, private query, and private path-segment markers must be absent from JSONL |
| Existing evidence is restricted | A pre-created 0644 metric file must become 0600 before the append |
| Failure is bounded | An unavailable upstream returns 502 and records an exception class, not its potentially sensitive message |
| Upstream selection is deliberate | A non-loopback upstream requires explicit opt-in, and an upstream URL containing user information is rejected |
The proxy still forwards the Authorization header because a real upstream may require it. It never writes headers to the metrics file. That distinction is worth testing: a value can be required for transport and still be forbidden from retained operational evidence.
Get the Companion Files
The complete files are available through the site’s source viewer. You can inspect them in place before downloading anything.
passive_proxy_lab.py python View source
#!/usr/bin/env python3
"""A small content-free passive HTTP proxy lab for local experimentation.
The lab forwards request and response bodies without decoding or rewriting them.
Its JSONL metrics contain a bounded route class, sizes, timing, status, and an
error class only. It does not record request targets, headers, bodies, client
addresses, prompts, or replies.
This is an educational artifact, not a production model gateway. It intentionally
omits streaming passthrough, TLS termination, authentication, log rotation, and
service supervision.
"""
from __future__ import annotations
import argparse
import http.client
import json
import os
import re
import stat
import threading
import time
import uuid
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Type
from urllib.parse import SplitResult, urlsplit
HOP_BY_HOP_HEADERS = {
"connection",
"keep-alive",
"proxy-authenticate",
"proxy-authorization",
"te",
"trailer",
"transfer-encoding",
"upgrade",
}
ROUTE_CLASSES = {
"/api/chat": "chat_completions",
"/chat/completions": "chat_completions",
"/v1/chat/completions": "chat_completions",
"/v1/responses": "responses",
"/v1/models": "models",
"/health": "health",
"/healthz": "health",
}
HEADER_NAME_RE = re.compile(r"^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$")
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat()
def parse_upstream(value: str, *, allow_remote: bool = False) -> SplitResult:
parsed = urlsplit(value)
if parsed.scheme not in {"http", "https"}:
raise ValueError("upstream scheme must be http or https")
if not parsed.hostname or parsed.port is None:
raise ValueError("upstream must include an explicit host and port")
if parsed.username is not None or parsed.password is not None:
raise ValueError("upstream must not include user information")
if parsed.query or parsed.fragment:
raise ValueError("upstream must not include a query or fragment")
if not allow_remote and parsed.hostname not in {"127.0.0.1", "localhost", "::1"}:
raise ValueError("remote upstreams require --allow-remote-upstream")
return parsed
def classify_route(request_target: str) -> str:
"""Map a request target to a bounded class without retaining path segments."""
path = urlsplit(request_target).path.rstrip("/") or "/"
return ROUTE_CLASSES.get(path, "other")
def safe_header_name(name: str) -> str | None:
"""Return a header name only if it is a valid RFC 9110 field name."""
sanitized = name.replace("\r", "").replace("\n", "").replace(":", "")
if sanitized != name:
return None
return sanitized if HEADER_NAME_RE.fullmatch(sanitized) else None
def safe_header_value(value: str) -> str:
"""Remove line breaks before writing a value to an HTTP response header."""
return value.replace("\r", "").replace("\n", "")
class JsonlMetrics:
"""Append content-free records with process-local write serialization."""
def __init__(self, path: Path) -> None:
self.path = path
self._lock = threading.Lock()
def write(self, record: dict[str, object]) -> None:
line = json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n"
with self._lock:
self.path.parent.mkdir(parents=True, exist_ok=True)
if not hasattr(os, "O_NOFOLLOW") and self.path.is_symlink():
raise ValueError("metrics path must not be a symbolic link")
flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0)
descriptor = os.open(self.path, flags, 0o600)
try:
if not stat.S_ISREG(os.fstat(descriptor).st_mode):
raise ValueError("metrics path must be a regular file")
os.fchmod(descriptor, 0o600)
handle = os.fdopen(descriptor, "a", encoding="utf-8")
descriptor = -1
with handle:
handle.write(line)
handle.flush()
os.fsync(handle.fileno())
finally:
if descriptor >= 0:
os.close(descriptor)
def joined_upstream_path(upstream: SplitResult, request_target: str) -> str:
base = upstream.path.rstrip("/")
target = request_target if request_target.startswith("/") else f"/{request_target}"
return f"{base}{target}" if base else target
def make_proxy_handler(
upstream: SplitResult,
metrics: JsonlMetrics,
*,
timeout: float = 30.0,
) -> Type[BaseHTTPRequestHandler]:
"""Create an isolated handler class bound to one upstream and metrics sink."""
class PassiveProxyHandler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def log_message(self, _format: str, *_args: object) -> None:
return
def read_request_body(self) -> bytes:
length = int(self.headers.get("Content-Length", "0") or "0")
return self.rfile.read(length) if length > 0 else b""
def filtered_request_headers(self, body: bytes) -> dict[str, str]:
headers: dict[str, str] = {}
for name, value in self.headers.items():
lowered = name.lower()
if lowered in HOP_BY_HOP_HEADERS or lowered in {"host", "content-length"}:
continue
headers[name] = value
headers["Host"] = upstream.netloc
headers["Content-Length"] = str(len(body))
return headers
def send_upstream_response(
self,
status: int,
reason: str,
headers: list[tuple[str, str]],
body: bytes,
) -> None:
self.send_response(status, reason)
for name, value in headers:
lowered = name.lower()
if lowered in HOP_BY_HOP_HEADERS or lowered == "content-length":
continue
safe_name = safe_header_name(name)
if safe_name is None:
continue
self.send_header(safe_name, safe_header_value(value))
self.send_header("Content-Length", str(len(body)))
self.send_header("Connection", "close")
self.end_headers()
if body:
self.wfile.write(body)
def send_proxy_error(self, request_id: str) -> bytes:
body = json.dumps(
{"error": "upstream unavailable", "request_id": request_id},
separators=(",", ":"),
).encode("utf-8")
self.send_response(502, "Bad Gateway")
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.send_header("Connection", "close")
self.end_headers()
self.wfile.write(body)
return body
def proxy(self) -> None:
request_id = uuid.uuid4().hex
started = time.monotonic()
request_body = self.read_request_body()
response_body = b""
status = 502
error_class: str | None = None
connection: http.client.HTTPConnection | http.client.HTTPSConnection | None = None
try:
connection_type = (
http.client.HTTPSConnection
if upstream.scheme == "https"
else http.client.HTTPConnection
)
connection = connection_type(upstream.hostname, upstream.port, timeout=timeout)
connection.request(
self.command,
joined_upstream_path(upstream, self.path),
body=request_body,
headers=self.filtered_request_headers(request_body),
)
response = connection.getresponse()
status = response.status
response_body = response.read()
self.send_upstream_response(
response.status,
response.reason,
response.getheaders(),
response_body,
)
except Exception as exc: # The lab records the class, never the message.
error_class = type(exc).__name__
response_body = self.send_proxy_error(request_id)
finally:
if connection is not None:
connection.close()
metrics.write(
{
"duration_ms": round((time.monotonic() - started) * 1000, 3),
"error_class": error_class,
"method": self.command,
"request_bytes": len(request_body),
"request_id": request_id,
"response_bytes": len(response_body),
"route_class": classify_route(self.path),
"status": status,
"ts": utc_now(),
}
)
do_GET = proxy
do_POST = proxy
do_PUT = proxy
do_PATCH = proxy
do_DELETE = proxy
do_OPTIONS = proxy
return PassiveProxyHandler
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Content-free passive proxy lab")
parser.add_argument("--listen-host", default="127.0.0.1")
parser.add_argument("--listen-port", type=int, default=18080)
parser.add_argument("--upstream", required=True, help="Example: http://127.0.0.1:18081")
parser.add_argument("--metrics", type=Path, default=Path("passive-proxy-metrics.jsonl"))
parser.add_argument("--timeout", type=float, default=30.0)
parser.add_argument("--allow-remote-upstream", action="store_true")
return parser.parse_args()
def main() -> int:
args = parse_args()
upstream = parse_upstream(args.upstream, allow_remote=args.allow_remote_upstream)
handler = make_proxy_handler(upstream, JsonlMetrics(args.metrics), timeout=args.timeout)
server = ThreadingHTTPServer((args.listen_host, args.listen_port), handler)
print(
f"passive proxy lab listening on http://{args.listen_host}:{server.server_port} "
f"and forwarding to {upstream.scheme}://{upstream.netloc}{upstream.path}",
flush=True,
)
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
server.server_close()
return 0
if __name__ == "__main__":
raise SystemExit(main())
fake_model_upstream.py python View source
#!/usr/bin/env python3
"""Deterministic local upstream for the passive proxy lab."""
from __future__ import annotations
import argparse
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
class FakeModelHandler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def log_message(self, _format: str, *_args: object) -> None:
return
def do_POST(self) -> None:
length = int(self.headers.get("Content-Length", "0") or "0")
self.rfile.read(length)
body = json.dumps(
{
"id": "lab-response",
"object": "chat.completion",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "The fake upstream received the request.",
},
"finish_reason": "stop",
}
],
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0,
},
},
separators=(",", ":"),
).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def main() -> int:
parser = argparse.ArgumentParser(description="Fake local model endpoint")
parser.add_argument("--listen-host", default="127.0.0.1")
parser.add_argument("--listen-port", type=int, default=18081)
args = parser.parse_args()
server = ThreadingHTTPServer((args.listen_host, args.listen_port), FakeModelHandler)
print(
f"fake model upstream listening on http://{args.listen_host}:{server.server_port}",
flush=True,
)
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
server.server_close()
return 0
if __name__ == "__main__":
raise SystemExit(main())
test_passive_proxy_lab.py python View source
#!/usr/bin/env python3
from __future__ import annotations
import json
import socket
import tempfile
import threading
import time
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.error import HTTPError
from urllib.request import Request, urlopen
from passive_proxy_lab import (
JsonlMetrics,
classify_route,
make_proxy_handler,
parse_upstream,
safe_header_name,
safe_header_value,
)
class CaptureUpstream(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
received: list[dict[str, object]] = []
response_body = b'{"ok":true,"spacing":" preserved "}\n'
def log_message(self, _format: str, *_args: object) -> None:
return
def do_POST(self) -> None:
length = int(self.headers.get("Content-Length", "0") or "0")
body = self.rfile.read(length)
type(self).received.append(
{
"authorization": self.headers.get("Authorization"),
"body": body,
"path": self.path,
}
)
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(self.response_body)))
self.end_headers()
self.wfile.write(self.response_body)
def start_server(handler: type[BaseHTTPRequestHandler]) -> tuple[ThreadingHTTPServer, threading.Thread]:
server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
return server, thread
def read_metrics(path: Path, timeout: float = 2.0) -> str:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if path.exists() and path.stat().st_size:
return path.read_text(encoding="utf-8")
time.sleep(0.01)
raise AssertionError(f"metrics were not written within {timeout} seconds")
class PassiveProxyLabTests(unittest.TestCase):
def setUp(self) -> None:
CaptureUpstream.received = []
def test_request_and_response_bodies_are_preserved(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
metrics_path = Path(tmpdir) / "metrics.jsonl"
upstream, upstream_thread = start_server(CaptureUpstream)
self.addCleanup(upstream.shutdown)
self.addCleanup(upstream.server_close)
self.addCleanup(upstream_thread.join, 1)
parsed = parse_upstream(f"http://127.0.0.1:{upstream.server_port}")
proxy_handler = make_proxy_handler(parsed, JsonlMetrics(metrics_path))
proxy, proxy_thread = start_server(proxy_handler)
self.addCleanup(proxy.shutdown)
self.addCleanup(proxy.server_close)
self.addCleanup(proxy_thread.join, 1)
secret_marker = "private prompt text must not enter metrics"
request_body = json.dumps(
{"messages": [{"role": "user", "content": secret_marker}]},
separators=(",", ":"),
).encode("utf-8")
path_marker = "private-path-marker"
request = Request(
f"http://127.0.0.1:{proxy.server_port}/sessions/{path_marker}/messages?trace=private",
data=request_body,
headers={
"Authorization": "Bearer lab-secret",
"Content-Type": "application/json",
},
method="POST",
)
with urlopen(request, timeout=3) as response:
self.assertEqual(response.status, 200)
self.assertEqual(response.read(), CaptureUpstream.response_body)
self.assertEqual(CaptureUpstream.received[0]["body"], request_body)
self.assertEqual(
CaptureUpstream.received[0]["path"],
f"/sessions/{path_marker}/messages?trace=private",
)
self.assertEqual(
CaptureUpstream.received[0]["authorization"],
"Bearer lab-secret",
)
metrics_text = read_metrics(metrics_path)
self.assertNotIn(secret_marker, metrics_text)
self.assertNotIn("lab-secret", metrics_text)
self.assertNotIn("trace=private", metrics_text)
self.assertNotIn(path_marker, metrics_text)
record = json.loads(metrics_text)
self.assertEqual(
set(record),
{
"duration_ms",
"error_class",
"method",
"request_bytes",
"request_id",
"response_bytes",
"route_class",
"status",
"ts",
},
)
self.assertEqual(record["route_class"], "other")
self.assertEqual(record["request_bytes"], len(request_body))
self.assertEqual(record["response_bytes"], len(CaptureUpstream.response_body))
self.assertEqual(record["status"], 200)
self.assertIsNone(record["error_class"])
self.assertEqual(metrics_path.stat().st_mode & 0o777, 0o600)
def test_route_classes_are_bounded(self) -> None:
self.assertEqual(classify_route("/v1/chat/completions?trace=private"), "chat_completions")
self.assertEqual(classify_route("/sessions/private-path-marker/messages"), "other")
def test_existing_metrics_file_is_restricted_before_append(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
metrics_path = Path(tmpdir) / "metrics.jsonl"
metrics_path.write_text("", encoding="utf-8")
metrics_path.chmod(0o644)
JsonlMetrics(metrics_path).write({"route_class": "health", "status": 200})
self.assertEqual(metrics_path.stat().st_mode & 0o777, 0o600)
self.assertEqual(json.loads(metrics_path.read_text(encoding="utf-8"))["status"], 200)
def test_upstream_failure_returns_502_without_logging_exception_text(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
metrics_path = Path(tmpdir) / "metrics.jsonl"
unused = socket.socket()
unused.bind(("127.0.0.1", 0))
unused_port = unused.getsockname()[1]
unused.close()
parsed = parse_upstream(f"http://127.0.0.1:{unused_port}")
proxy_handler = make_proxy_handler(parsed, JsonlMetrics(metrics_path), timeout=0.5)
proxy, proxy_thread = start_server(proxy_handler)
self.addCleanup(proxy.shutdown)
self.addCleanup(proxy.server_close)
self.addCleanup(proxy_thread.join, 1)
request = Request(
f"http://127.0.0.1:{proxy.server_port}/v1/models",
data=b"sensitive failure payload",
method="POST",
)
with self.assertRaises(HTTPError) as raised:
urlopen(request, timeout=3)
self.assertEqual(raised.exception.code, 502)
metrics_text = read_metrics(metrics_path)
self.assertNotIn("sensitive failure payload", metrics_text)
self.assertNotIn(str(unused_port), metrics_text)
record = json.loads(metrics_text)
self.assertEqual(record["status"], 502)
self.assertIsNotNone(record["error_class"])
def test_remote_upstream_requires_explicit_opt_in(self) -> None:
with self.assertRaisesRegex(ValueError, "--allow-remote-upstream"):
parse_upstream("https://example.invalid:443")
def test_upstream_user_information_is_rejected(self) -> None:
with self.assertRaisesRegex(ValueError, "must not include user information"):
parse_upstream("http://invented-user:invented-secret@127.0.0.1:18081")
def test_forwarded_response_headers_reject_response_splitting_material(self) -> None:
self.assertEqual(safe_header_name("X-Lab-Trace"), "X-Lab-Trace")
self.assertIsNone(safe_header_name("X-Bad\r\nInjected"))
self.assertIsNone(safe_header_name("X-Bad:Injected"))
self.assertEqual(safe_header_value("ok\r\nX-Injected: yes"), "okX-Injected: yes")
if __name__ == "__main__":
unittest.main()
Create a disposable working directory and download the three files:
mkdir -p /tmp/passive-proxy-lab
cd /tmp/passive-proxy-lab
curl -fsSLO https://unixwzrd.ai/assets/code/agent-optimization/post-06/passive_proxy_lab.py
curl -fsSLO https://unixwzrd.ai/assets/code/agent-optimization/post-06/fake_model_upstream.py
curl -fsSLO https://unixwzrd.ai/assets/code/agent-optimization/post-06/test_passive_proxy_lab.py
Run the Regression Tests First
I prefer to begin with the executable contract before starting any servers manually:
python3 -m unittest -v test_passive_proxy_lab.py
The expected result is six passing tests. The suite creates temporary loopback listeners with operating-system-assigned ports, sends invented content through the proxy, checks the captured bytes, checks the returned body bytes, reads the metric record, and then removes the temporary evidence.
The privacy test deliberately sends four markers through different surfaces:
private prompt text must not enter metrics
Bearer lab-secret
trace=private
private-path-marker
The body and header must reach the fake upstream, and the query and private path segment must remain part of the upstream request target. None may appear in the metrics file. The lab maps only a small fixed set of known endpoints to bounded values such as chat_completions, responses, models, and health; every other target becomes other. It never retains the arbitrary request path.
This is a useful pattern beyond this lab. If a log format is supposed to exclude content, do not verify that by reading a pleasant example. Inject markers into every sensitive surface and fail the test if any marker survives.
Start the Fake Model Endpoint
Open one terminal in the companion directory and run:
python3 fake_model_upstream.py
It listens on loopback and accepts POST requests. It reads the body so the HTTP exchange completes correctly, but it does not print or retain that body. Every request receives the same invented OpenAI-compatible JSON response.
The zero token counts in that response are deliberate. This fake endpoint does not tokenize anything, so claiming a synthetic token count would make the example look more realistic at the expense of being truthful.
Start the Passive Proxy
In a second terminal, still in the companion directory, run:
python3 passive_proxy_lab.py \
--upstream http://127.0.0.1:18081 \
--metrics ./passive-proxy-metrics.jsonl
The proxy listens on a separate loopback port and forwards to the fake endpoint. Both listener and upstream default to local-only behavior. Before every append, the metrics writer verifies that the destination is a regular file and enforces mode 0600, including when a permissive file already exists. A symbolic-link target is rejected where the platform provides the standard no-follow open flag.
The proxy accepts GET, POST, PUT, PATCH, DELETE, and OPTIONS for experimentation. It filters a fixed set of common hop-by-hop headers, sets the upstream Host and Content-Length values, and otherwise leaves end-to-end headers available to the upstream. It does not parse additional header names nominated by an incoming Connection header, which is one reason this remains a teaching artifact. Request and response bodies are handled as bytes. They are not decoded, normalized, reformatted, or rendered.
The upstream URL must include an explicit host and port and may not contain user information, a query, or a fragment. Remote hosts also require --allow-remote-upstream. Rejecting user information prevents a startup status line from reproducing embedded credentials.
Send an Invented Chat Request
Use a third terminal to send a request through the proxy:
curl --silent --show-error \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer invented-lab-token' \
--data '{"messages":[{"role":"user","content":"Explain why passive observation matters."}]}' \
'http://127.0.0.1:18080/v1/chat/completions?session=invented'
You should receive the deterministic response from the fake endpoint. The proxy did not generate that response and did not parse it before returning it.
Now inspect the operational record:
tail -n 1 passive-proxy-metrics.jsonl | python3 -m json.tool
The exact timestamp, request ID, byte counts, and duration will differ. The shape should resemble this sanitized example:
{
"duration_ms": 1.234,
"error_class": null,
"method": "POST",
"request_bytes": 88,
"request_id": "generated-lab-identifier",
"response_bytes": 251,
"route_class": "chat_completions",
"status": 200,
"ts": "generated UTC timestamp"
}
Notice what is missing. There is no arbitrary request path, and there are no headers, token values, query parameters, request bodies, prompt fragments, replies, or client addresses. The route class and byte counts are useful for trend monitoring, but they cannot reconstruct the content.
You can check the canaries directly:
if grep -Eq 'invented-lab-token|session=invented|passive observation matters' passive-proxy-metrics.jsonl; then
echo 'privacy canary failed'
else
echo 'privacy canaries absent from metrics'
fi
Force an Upstream Failure
Stop the fake endpoint with Control-C, leave the proxy running, and send the request again. The proxy should return a short 502 response containing a generated request ID. Its metric record will have status 502 and an error_class, but it will not record the exception message or upstream address.
This is not enough error reporting for a production operator. It is the right amount for this lab because it demonstrates the boundary: routine metrics can identify the class and correlation point while detailed incident evidence remains a separate, more sensitive channel.
Restart the fake endpoint and the next request should succeed. That gives you a quick way to watch the status transition without changing the proxy configuration.
Try Three Useful Experiments
First, change whitespace inside the JSON body and rerun the byte-preservation test. A proxy that parses and reserializes JSON may preserve meaning while changing bytes. This lab’s contract is stricter, so the upstream capture must match the original body exactly.
Second, add more canaries. Put an invented secret in a header, a query parameter, a variable path segment, and a nested message. Then extend the test assertion before changing the proxy. This turns the privacy rule into a regression boundary rather than a comment future code can ignore.
Third, add a Content-Encoding or unusual content type and observe that the proxy does not need to understand the payload to forward its bytes. Be careful not to mistake this for complete HTTP compliance. The lab buffers complete bodies and does not implement chunked request streaming.
What I Would Add Before Real Use
I would not place this teaching proxy on a shared network. A production component needs streaming request and response handling, bounded body sizes, TLS policy, authentication, concurrency limits, timeouts suited to model workloads, structured diagnostic retention, rotation, health checks, service supervision, and careful handling of disconnects and partial responses.
It also needs separate evidence modes. Content-free operational metrics can have one access and retention policy. Raw requests, rendered prompts, model reasoning, tool calls, and responses need a much stricter policy. Combining both into one convenient log makes it too easy for a routine monitoring tool to become a private-conversation archive.
The production LLM-Ops-Kit proxy covers substantially more of that operational surface, including streaming diagnostics, rendered prompt inspection, correlation, cancellation classification, wrapper-managed operations, and timed numbered diagnostic-log rotation. That rotation renames the active file and changes its inode, so monitoring readers must reopen the active filename. This lab is here so readers can understand and test the core invariant without copying private configuration or deploying a full stack.
Current State
The proxy lab, fake upstream, and six regression tests pass with the system Python 3 interpreter in the private publication workspace. The tests verify exact request and response body preservation, bounded route classification, prompt, header, query, and path-segment marker exclusion, 0600 enforcement for new and pre-existing metric files, sanitized upstream failure, credential-bearing upstream rejection, and explicit remote-upstream opt-in.
The package is presented as a teaching artifact and is not represented as production-ready. It passed the documented technical checks before appearing here.
Next Work
A useful follow-up exercise would add streaming SSE fixtures while preserving the same content canaries and exact transport assertions. That belongs in a later revision only after the buffered teaching boundary remains clear.
For now, the important result is already executable: the proxy can observe sizes, timing, status, and failure class while leaving the request and response bodies alone and keeping their contents out of its routine metric record.
That is the habit I want readers to take away from the exercise. Write the boundary down, put canaries on both sides of it, and make the test fail before a future convenience quietly turns observation into mutation or routine metrics into a content archive.
Join the Discussion
Comments for this post live in GitHub Discussions. That keeps moderation in one place and gives the conversation a stable home.