附录：随书实验完整代码

"""Loopback-only HTTP experiments for AI gateway engineering."""

"""Minimal HTTP/SSE client used by both the demo and integration tests."""
import http.client
import json
from urllib.parse import urlsplit

from .protocol import read_events


def open_stream(url, request, credential="tenant-a-secret", timeout=5):
    parsed = urlsplit(url)
    connection = http.client.HTTPConnection(parsed.hostname, parsed.port, timeout=timeout)
    connection.request("POST", "/v1/generate", body=json.dumps(request).encode(), headers={
        "Authorization": "Bearer " + credential, "Content-Type": "application/json",
    })
    return connection, connection.getresponse()


def generate(url, request, credential="tenant-a-secret"):
    connection, response = open_stream(url, request, credential)
    try:
        if response.status != 200:
            return response.status, json.loads(response.read())
        return response.status, list(read_events(response))
    finally:
        response.close()
        connection.close()


def cancel(url, request_id, credential="tenant-a-secret"):
    parsed = urlsplit(url)
    connection = http.client.HTTPConnection(parsed.hostname, parsed.port, timeout=5)
    connection.request("DELETE", "/v1/runs/" + request_id,
                       headers={"Authorization": "Bearer " + credential})
    response = connection.getresponse()
    try:
        return response.status, json.loads(response.read())
    finally:
        response.close()
        connection.close()

"""Run complete HTTP fault experiments, or keep a loopback gateway running."""
import argparse
from contextlib import ExitStack
from datetime import datetime, timezone
import http.client
import json
from pathlib import Path
import socket
import time
from urllib.parse import urlsplit

from .client import cancel, generate, open_stream
from .gateway import Gateway, Route, Settings, TenantPolicy
from .protocol import UNIT_CONTRACT, read_events
from .upstream import ModelStub, Plan


def payload(request_id, **changes):
    value = {"request_id": request_id, "prompt": "hi", "capabilities": ["text"],
             "max_output_units": 100}
    value.update(changes)
    return value


def await_condition(check, timeout=3):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        value = check()
        if value:
            return value
        time.sleep(0.01)
    raise RuntimeError("experiment_condition_timed_out")


def await_finished(gateway, request_id):
    return await_condition(lambda: (record if not record["active"] else None)
                           if (record := gateway.ledger.record("tenant-a", request_id)) else None)


def experiments():
    cases = []
    with ExitStack() as stack:
        first = stack.enter_context(ModelStub([Plan("safe_503")]))
        second = stack.enter_context(ModelStub([Plan(chunks=("route ", "works"))]))
        gateway = stack.enter_context(Gateway([
            Route("primary", first.url, trust_not_started=True), Route("backup", second.url)]))
        status, events = generate(gateway.url, payload("safe-failover", allow_failover=True))
        record = await_finished(gateway, "safe-failover")
        assert status == 200 and record["actual_units"] == 13
        assert first.operations_started == 0 and second.operations_started == 1
        cases.append({"name": "authorized_safe_failover", "http_status": status,
                      "events": [name for name, _ in events],
                      "upstream_operations": [first.operations_started, second.operations_started],
                      "accounting": record, "balance": gateway.ledger.snapshot("tenant-a")})
    with ExitStack() as stack:
        first = stack.enter_context(ModelStub([Plan("midstream_disconnect", chunks=("part", "never"))]))
        second = stack.enter_context(ModelStub())
        gateway = stack.enter_context(Gateway([
            Route("primary", first.url, trust_not_started=True), Route("backup", second.url)],
            policies={"tenant-a": TenantPolicy(budget_units=110)}))
        status, events = generate(gateway.url, payload("interrupted", max_output_units=20, allow_failover=True))
        record = await_finished(gateway, "interrupted")
        assert status == 200 and record["usage_state"] == "unknown" and second.requests == 0
        held = gateway.ledger.snapshot("tenant-a")
        # The fixture executed four output bytes and two input bytes. This is
        # authoritative in-process fixture evidence, not an HTTP client claim.
        gateway.ledger.reconcile("tenant-a", "interrupted", 6)
        cases.append({"name": "interrupted_not_replayed", "http_status": status,
                      "events": [name for name, _ in events], "backup_requests": second.requests,
                      "accounting_before_reconciliation": record, "balance_held": held,
                      "authoritative_fixture_units": 6,
                      "balance_after_reconciliation": gateway.ledger.snapshot("tenant-a")})
    with ExitStack() as stack:
        upstream = stack.enter_context(ModelStub([Plan(chunks=("x",) * 1000, delay=0.005)]))
        gateway = stack.enter_context(Gateway([Route("primary", upstream.url)],
            policies={"tenant-a": TenantPolicy(max_concurrent=1)}))
        connection, response = open_stream(gateway.url, payload("cancel", max_output_units=1000))
        try:
            assert next(read_events(response))[0] == "token"
            concurrency_status, _ = generate(gateway.url, payload("rejected-concurrent"))
            cancellation_status, _ = cancel(gateway.url, "cancel")
            record = await_finished(gateway, "cancel")
            await_condition(lambda: upstream.disconnects > 0)
            assert concurrency_status == 429 and cancellation_status == 202
            assert record["usage_state"] == "unknown"
            cases.append({"name": "cancel_and_concurrency", "concurrent_http_status": concurrency_status,
                          "cancel_http_status": cancellation_status, "upstream_observed_disconnect": True,
                          "accounting": record, "balance": gateway.ledger.snapshot("tenant-a")})
        finally:
            response.close()
            connection.close()
    with ExitStack() as stack:
        upstream = stack.enter_context(ModelStub([Plan(chunks=("x" * 2048,) * 2000, delay=0)]))
        gateway = stack.enter_context(Gateway([Route("primary", upstream.url)],
            policies={"tenant-a": TenantPolicy(budget_units=5_000_000)},
            settings=Settings(queue_events=2, client_write_timeout=0.5)))
        parsed = urlsplit(gateway.url)
        connection = http.client.HTTPConnection(parsed.hostname, parsed.port, timeout=5)
        connection.connect()
        connection.sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024)
        connection.request("POST", "/v1/generate", body=json.dumps(payload("slow", max_output_units=4_096_000)),
            headers={"Authorization": "Bearer tenant-a-secret", "Content-Type": "application/json"})
        response = connection.getresponse()
        try:
            assert response.status == 200
            await_condition(lambda: gateway.run_stats("tenant-a", "slow")["backpressure_waits"] > 0)
            stats = gateway.run_stats("tenant-a", "slow")
            assert stats["queue_peak"] <= 2
            cancel(gateway.url, "slow")
            record = await_finished(gateway, "slow")
            await_condition(lambda: upstream.disconnects > 0)
            cases.append({"name": "slow_consumer", "queue_capacity_events": 2,
                          "queue_peak": stats["queue_peak"], "backpressure_observed": True,
                          "upstream_observed_disconnect": True, "usage_state": record["usage_state"]})
        finally:
            response.close()
            connection.close()
    with ExitStack() as stack:
        upstream = stack.enter_context(ModelStub([Plan(chunks=("private-generated-value",))]))
        gateway = stack.enter_context(Gateway([Route("primary", upstream.url)]))
        generate(gateway.url, payload("privacy", prompt="private-input-value"))
        await_finished(gateway, "privacy")
        audit = gateway.audit.jsonl()
        forbidden = ["private-input-value", "private-generated-value", "tenant-a-secret", "fixture-upstream-secret"]
        assert not any(value in audit for value in forbidden)
        cases.append({"name": "audit_allowlist", "sensitive_values_absent": True,
                      "audit": [json.loads(line) for line in audit.splitlines()]})
    return {"experiment": "loopback-http-gateway-v1", "checked_at": datetime.now(timezone.utc).isoformat(),
            "unit_contract": UNIT_CONTRACT, "cases": cases,
            "limits": ["one process; volatile ledger and audit", "loopback HTTP; fixture credentials",
                       "no vendor API or real model capability evaluation",
                       "queue bound excludes socket and HTTP library buffers",
                       "socket cancellation does not prove provider compute or billing stopped"]}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--evidence", type=Path, help="Also save the executed experiment record as JSON")
    parser.add_argument("--serve", action="store_true", help="Serve a normal text fixture until Ctrl-C")
    parser.add_argument("--port", type=int, default=8088, help="Loopback gateway port in --serve mode")
    args = parser.parse_args()
    if args.serve:
        with ModelStub() as upstream, Gateway([Route("local-text", upstream.url)], port=args.port) as gateway:
            print(json.dumps({"gateway": gateway.url, "credential": "tenant-a-secret",
                              "unit_contract": UNIT_CONTRACT, "capabilities": ["text"]}), flush=True)
            try:
                while True:
                    time.sleep(0.2)
            except KeyboardInterrupt:
                print(gateway.audit.jsonl())
        return
    result = experiments()
    text = json.dumps(result, ensure_ascii=False, indent=2) + "\n"
    if args.evidence:
        args.evidence.parent.mkdir(parents=True, exist_ok=True)
        args.evidence.write_text(text)
    print(text, end="")


if __name__ == "__main__":
    main()

{
  "experiment": "loopback-http-gateway-v1",
  "checked_at": "2026-10-21T08:14:07.810190+00:00",
  "unit_contract": "utf8-bytes-v1",
  "cases": [
    {
      "name": "authorized_safe_failover",
      "http_status": 200,
      "events": [
        "token",
        "token",
        "usage",
        "done"
      ],
      "upstream_operations": [
        0,
        1
      ],
      "accounting": {
        "reservation_units": 102,
        "actual_units": 13,
        "usage_state": "settled",
        "outcome": "completed",
        "active": false
      },
      "balance": {
        "budget_units": 10000,
        "spent_units": 13,
        "held_units": 0,
        "available_units": 9987,
        "active": 0
      }
    },
    {
      "name": "interrupted_not_replayed",
      "http_status": 200,
      "events": [
        "token",
        "error"
      ],
      "backup_requests": 0,
      "accounting_before_reconciliation": {
        "reservation_units": 22,
        "actual_units": null,
        "usage_state": "unknown",
        "outcome": "failed",
        "active": false
      },
      "balance_held": {
        "budget_units": 110,
        "spent_units": 0,
        "held_units": 22,
        "available_units": 88,
        "active": 0
      },
      "authoritative_fixture_units": 6,
      "balance_after_reconciliation": {
        "budget_units": 110,
        "spent_units": 6,
        "held_units": 0,
        "available_units": 104,
        "active": 0
      }
    },
    {
      "name": "cancel_and_concurrency",
      "concurrent_http_status": 429,
      "cancel_http_status": 202,
      "upstream_observed_disconnect": true,
      "accounting": {
        "reservation_units": 1002,
        "actual_units": null,
        "usage_state": "unknown",
        "outcome": "cancelled",
        "active": false
      },
      "balance": {
        "budget_units": 10000,
        "spent_units": 0,
        "held_units": 1002,
        "available_units": 8998,
        "active": 0
      }
    },
    {
      "name": "slow_consumer",
      "queue_capacity_events": 2,
      "queue_peak": 2,
      "backpressure_observed": true,
      "upstream_observed_disconnect": true,
      "usage_state": "unknown"
    },
    {
      "name": "audit_allowlist",
      "sensitive_values_absent": true,
      "audit": [
        {
          "event": "admitted",
          "request_id": "privacy",
          "reservation_units": 119,
          "tenant": "tenant-a",
          "time": 1792570447.765
        },
        {
          "attempt": 1,
          "event": "route_attempt",
          "request_id": "privacy",
          "route": "primary",
          "tenant": "tenant-a",
          "time": 1792570447.765
        },
        {
          "actual_units": 42,
          "event": "finished",
          "outcome": "completed",
          "reason": "final_usage_received",
          "request_id": "privacy",
          "tenant": "tenant-a",
          "time": 1792570447.783,
          "usage_state": "settled"
        }
      ]
    }
  ],
  "limits": [
    "one process; volatile ledger and audit",
    "loopback HTTP; fixture credentials",
    "no vendor API or real model capability evaluation",
    "queue bound excludes socket and HTTP library buffers",
    "socket cancellation does not prove provider compute or billing stopped"
  ]
}

"""Small inspectable HTTP gateway with conservative replay and accounting rules.

Teaching constraints: loopback HTTP only, one process/in-memory state, one credit
per UTF-8 byte, no real model SDKs. No claim of durable accounting, authentication
for an untrusted network, precise model cancellation or production isolation.
"""
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler
import http.client
import json
import queue
import re
import socket
import threading
import time
from urllib.parse import urlsplit

from .protocol import UNIT_CONTRACT, ProtocolError, encode_event, end_chunks, read_events, send_chunk
from .upstream import QuietServer


@dataclass(frozen=True)
class Route:
    name: str
    url: str
    capabilities: frozenset = frozenset({"text"})
    credential: str = "fixture-upstream-secret"
    trust_not_started: bool = False

    def __post_init__(self):
        parsed = urlsplit(self.url)
        if parsed.scheme != "http" or parsed.hostname != "127.0.0.1" or not parsed.port:
            raise ValueError("The lab accepts loopback HTTP upstreams only")


@dataclass(frozen=True)
class TenantPolicy:
    budget_units: int = 10000
    max_concurrent: int = 2


@dataclass(frozen=True)
class Settings:
    queue_events: int = 2
    max_attempts: int = 2
    upstream_timeout: float = 1.0
    client_write_timeout: float = 0.3
    max_output_units: int = 8 * 1024 * 1024
    max_input_bytes: int = 32768


class AdmissionError(Exception):
    def __init__(self, code, status):
        self.code, self.status = code, status


class StreamFailure(Exception):
    pass


class Cancelled(StreamFailure):
    pass


class OpenFailure(Exception):
    def __init__(self, code, safe=False):
        self.code, self.safe = code, safe


class Audit:
    ALLOWED = {"event", "request_id", "tenant", "route", "attempt", "reason",
               "outcome", "usage_state", "reservation_units", "actual_units"}

    def __init__(self):
        self.records, self.lock = [], threading.Lock()

    def emit(self, event, **fields):
        # Field allowlist excludes prompt, generated text, request headers and
        # arbitrary exception strings; do not log arbitrary upstream objects.
        record = {"event": event, **fields}
        if not set(record) <= self.ALLOWED:
            raise ValueError("Unapproved audit field")
        with self.lock:
            self.records.append({"time": round(time.time(), 3), **record})

    def jsonl(self):
        with self.lock:
            return "\n".join(json.dumps(r, ensure_ascii=False, sort_keys=True) for r in self.records)


class Ledger:
    def __init__(self, policies):
        self.policies = policies
        self.runs, self.lock = {}, threading.Lock()

    def _totals(self, tenant):
        records = [r for (owner, _), r in self.runs.items() if owner == tenant]
        held = sum(r["reservation_units"] for r in records if r["usage_state"] in {"held", "unknown"})
        spent = sum(r["actual_units"] or 0 for r in records if r["usage_state"] == "settled")
        return records, held, spent, sum(r["active"] for r in records)

    def reserve(self, tenant, request_id, units):
        with self.lock:
            if (tenant, request_id) in self.runs:
                raise AdmissionError("duplicate_request_id", 409)
            _, held, spent, active = self._totals(tenant)
            policy = self.policies[tenant]
            if active >= policy.max_concurrent:
                raise AdmissionError("concurrency_limit", 429)
            if units > policy.budget_units - held - spent:
                raise AdmissionError("budget_exhausted", 402)
            self.runs[tenant, request_id] = {
                "reservation_units": units, "actual_units": None,
                "usage_state": "held", "outcome": "admitted", "active": True,
            }

    def finish(self, tenant, request_id, outcome, actual_units=None, known_unstarted=False):
        with self.lock:
            record = self.runs[tenant, request_id]
            if not record["active"]:
                return dict(record)
            if actual_units is not None and not 0 <= actual_units <= record["reservation_units"]:
                raise ValueError("usage_outside_reservation")
            record.update(active=False, outcome=outcome, actual_units=actual_units)
            record["usage_state"] = ("settled" if actual_units is not None else
                                     "released" if known_unstarted else "unknown")
            return dict(record)

    def reconcile(self, tenant, request_id, actual_units):
        """Operator-only in-process action after authoritative billing evidence.

        There is deliberately no client HTTP endpoint that can forgive debt.
        Zero is valid only if an operator has evidence that nothing was billed.
        """
        with self.lock:
            record = self.runs[tenant, request_id]
            if record["usage_state"] != "unknown":
                raise ValueError("only_unknown_usage_can_be_reconciled")
            if type(actual_units) is not int or not 0 <= actual_units <= record["reservation_units"]:
                raise ValueError("invalid_authoritative_usage")
            record.update(usage_state="settled", actual_units=actual_units)

    def record(self, tenant, request_id):
        with self.lock:
            record = self.runs.get((tenant, request_id))
            return dict(record) if record else None

    def snapshot(self, tenant):
        with self.lock:
            _, held, spent, active = self._totals(tenant)
            return {"budget_units": self.policies[tenant].budget_units,
                    "spent_units": spent, "held_units": held,
                    "available_units": self.policies[tenant].budget_units - held - spent,
                    "active": active}


class Running:
    def __init__(self, tenant, request_id):
        self.tenant, self.request_id = tenant, request_id
        self.cancelled = threading.Event()
        self.socket = None
        self.lock = threading.Lock()
        self.processing_possible = False
        self.metrics = {"queue_peak": 0, "backpressure_waits": 0, "producer_stopped": True}

    def bind_socket(self, sock):
        with self.lock:
            self.socket = sock
            if self.cancelled.is_set():
                self._shutdown(sock)

    @staticmethod
    def _shutdown(sock):
        if sock is not None:
            try:
                sock.shutdown(socket.SHUT_RDWR)
            except OSError:
                pass

    def close_upstream(self):
        with self.lock:
            self._shutdown(self.socket)

    def cancel(self):
        self.cancelled.set()
        self.close_upstream()

    def metric(self, key, value=None):
        with self.lock:
            if value is None:
                self.metrics[key] += 1
            else:
                self.metrics[key] = value

    def stats(self):
        with self.lock:
            return dict(self.metrics)


class Gateway:
    def __init__(self, routes, policies=None, credentials=None, settings=None, port=0):
        self.routes = routes
        self.settings = settings or Settings()
        if self.settings.queue_events < 1 or self.settings.max_attempts < 1:
            raise ValueError("queue_events and max_attempts must be positive")
        self.ledger = Ledger(policies or {"tenant-a": TenantPolicy()})
        self.credentials = credentials or {"tenant-a-secret": "tenant-a"}
        self.audit, self.running, self.lock = Audit(), {}, threading.Lock()
        owner = self

        class Handler(BaseHTTPRequestHandler):
            protocol_version = "HTTP/1.1"

            def log_message(self, *args):
                pass

            def authenticate(self):
                value = self.headers.get("Authorization", "")
                return owner.credentials.get(value[7:]) if value.startswith("Bearer ") else None

            def reply(self, status, payload):
                self.close_connection = True
                data = json.dumps(payload).encode()
                try:
                    self.send_response(status)
                    self.send_header("Content-Type", "application/json")
                    self.send_header("Content-Length", str(len(data)))
                    self.send_header("Connection", "close")
                    self.end_headers()
                    self.wfile.write(data)
                except OSError:
                    pass

            def do_DELETE(self):
                tenant = self.authenticate()
                if tenant is None:
                    self.reply(401, {"error": "unauthorized"})
                    return
                prefix = "/v1/runs/"
                request_id = self.path[len(prefix):] if self.path.startswith(prefix) else ""
                with owner.lock:
                    run = owner.running.get((tenant, request_id))
                if run is None:
                    self.reply(404, {"error": "not_found"})
                    return
                if not owner.ledger.record(tenant, request_id)["active"]:
                    self.reply(409, {"error": "already_finished"})
                    return
                run.cancel()
                owner.audit.emit("cancel_requested", tenant=tenant, request_id=request_id)
                self.reply(202, {"request_id": request_id, "state": "cancellation_requested"})

            def do_POST(self):
                self.close_connection = True
                self.connection.settimeout(owner.settings.client_write_timeout)
                if self.path != "/v1/generate":
                    self.reply(404, {"error": "not_found"})
                    return
                tenant = self.authenticate()
                if tenant is None:
                    self.reply(401, {"error": "unauthorized"})
                    return
                try:
                    size = int(self.headers.get("Content-Length", "0"))
                    if not 0 < size <= 65536:
                        raise ValueError
                    request = json.loads(self.rfile.read(size))
                    owner.validate(request)
                except (ValueError, TypeError, KeyError, OSError):
                    self.reply(400, {"error": "invalid_request"})
                    return
                candidates = [route for route in owner.routes
                              if set(request["capabilities"]) <= route.capabilities]
                if not candidates:
                    self.reply(422, {"error": "unsupported_capability"})
                    return
                request_id = request["request_id"]
                reservation = len(request["prompt"].encode("utf-8")) + request["max_output_units"]
                try:
                    owner.ledger.reserve(tenant, request_id, reservation)
                except AdmissionError as exc:
                    self.reply(exc.status, {"error": exc.code})
                    return
                run = Running(tenant, request_id)
                with owner.lock:
                    owner.running[tenant, request_id] = run
                owner.audit.emit("admitted", tenant=tenant, request_id=request_id, reservation_units=reservation)
                owner.serve(self, run, request, candidates)

        self.server = QuietServer(("127.0.0.1", port), Handler)
        self.thread = threading.Thread(target=lambda: self.server.serve_forever(poll_interval=0.02), daemon=True)

    def validate(self, request):
        allowed = {"request_id", "prompt", "capabilities", "max_output_units", "allow_failover"}
        if not isinstance(request, dict) or not set(request) <= allowed:
            raise ValueError
        if not isinstance(request["request_id"], str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", request["request_id"]):
            raise ValueError
        if not isinstance(request["prompt"], str) or len(request["prompt"].encode("utf-8")) > self.settings.max_input_bytes:
            raise ValueError
        if type(request["max_output_units"]) is not int or not 1 <= request["max_output_units"] <= self.settings.max_output_units:
            raise ValueError
        capabilities = request["capabilities"]
        if not isinstance(capabilities, list) or not capabilities or len(capabilities) > 16:
            raise ValueError
        if any(not isinstance(c, str) or not re.fullmatch(r"[a-z][a-z0-9_]{0,31}", c) for c in capabilities):
            raise ValueError
        if "allow_failover" in request and type(request["allow_failover"]) is not bool:
            raise ValueError

    def _open(self, run, route, request):
        parsed = urlsplit(route.url)
        connection = http.client.HTTPConnection(parsed.hostname, parsed.port,
                                                timeout=self.settings.upstream_timeout)
        try:
            connection.connect()
        except OSError as exc:
            connection.close()
            raise OpenFailure("connect_failed") from exc
        run.bind_socket(connection.sock)
        if run.cancelled.is_set():
            connection.close()
            raise Cancelled("cancelled")
        # Once sending begins, a transport failure is ambiguous. The code does
        # not infer safe replay from zero model tokens or an empty response.
        run.processing_possible = True
        try:
            connection.request("POST", "/generate", body=json.dumps({
                "prompt": request["prompt"], "max_output_units": request["max_output_units"],
                "capabilities": request["capabilities"],
            }).encode(), headers={"Content-Type": "application/json",
                                  "Authorization": "Bearer " + route.credential})
            response = connection.getresponse()
            safe = (response.status == 503 and route.trust_not_started and
                    response.getheader("X-Lab-Execution") == "not-started")
            if response.status != 200:
                if safe:
                    run.processing_possible = False
                run.close_upstream()
                connection.close()
                raise OpenFailure("upstream_not_started" if safe else "upstream_status_unknown", safe)
            if not response.getheader("Content-Type", "").startswith("text/event-stream"):
                run.close_upstream()
                connection.close()
                raise OpenFailure("upstream_protocol_unknown")
            return connection, response
        except (OSError, http.client.HTTPException) as exc:
            run.close_upstream()
            connection.close()
            raise OpenFailure("upstream_transport_unknown") from exc

    def serve(self, handler, run, request, candidates):
        connection = response = producer = None
        stop = threading.Event()
        emitted = False
        outcome, reason, actual = "failed", "unavailable", None
        try:
            limit = min(len(candidates), self.settings.max_attempts) if request.get("allow_failover", False) else 1
            for attempt, route in enumerate(candidates[:limit], 1):
                self.audit.emit("route_attempt", tenant=run.tenant, request_id=run.request_id,
                                route=route.name, attempt=attempt)
                try:
                    connection, response = self._open(run, route, request)
                    break
                except OpenFailure as exc:
                    reason = exc.code
                    self.audit.emit("route_failed", tenant=run.tenant, request_id=run.request_id,
                                    route=route.name, attempt=attempt, reason=reason)
                    if not exc.safe or attempt == limit:
                        raise StreamFailure(reason)
            events = queue.Queue(maxsize=self.settings.queue_events)

            def put(item):
                while not stop.is_set() and not run.cancelled.is_set():
                    try:
                        events.put(item, timeout=0.02)
                        run.metric("queue_peak", max(run.stats()["queue_peak"], events.qsize()))
                        return
                    except queue.Full:
                        run.metric("backpressure_waits")

            def produce():
                run.metric("producer_stopped", False)
                try:
                    for event in read_events(response):
                        put(event)
                        if stop.is_set() or run.cancelled.is_set():
                            return
                    put(("_eof", {}))
                except (OSError, ValueError, ProtocolError, http.client.HTTPException):
                    put(("_broken", {}))
                finally:
                    run.metric("producer_stopped", True)

            producer = threading.Thread(target=produce, daemon=True)
            producer.start()
            produced, usage = 0, None
            while True:
                if run.cancelled.is_set():
                    raise Cancelled("cancelled")
                try:
                    event, data = events.get(timeout=0.05)
                except queue.Empty:
                    continue
                if event == "token":
                    if usage is not None or set(data) != {"text"} or not isinstance(data["text"], str):
                        raise StreamFailure("invalid_token")
                    produced += len(data["text"].encode("utf-8"))
                    if produced > request["max_output_units"]:
                        raise StreamFailure("output_exceeds_reservation")
                elif event == "usage":
                    if usage is not None or set(data) != {"contract", "input_units", "output_units"}:
                        raise StreamFailure("invalid_usage")
                    if (data["contract"] != UNIT_CONTRACT or type(data["input_units"]) is not int or
                            type(data["output_units"]) is not int or
                            data["input_units"] != len(request["prompt"].encode("utf-8")) or
                            data["output_units"] != produced):
                        raise StreamFailure("untrusted_usage")
                    usage = data
                elif event == "done":
                    if usage is None or data:
                        raise StreamFailure("missing_final_usage")
                else:
                    raise StreamFailure("upstream_stream_interrupted")
                if not emitted:
                    handler.connection.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096)
                    handler.send_response(200)
                    handler.send_header("Content-Type", "text/event-stream")
                    handler.send_header("Transfer-Encoding", "chunked")
                    handler.send_header("Cache-Control", "no-store")
                    handler.send_header("Connection", "close")
                    handler.send_header("X-Request-ID", run.request_id)
                    handler.end_headers()
                    emitted = True
                send_chunk(handler, encode_event(event, data))
                if event == "done":
                    end_chunks(handler)
                    actual = usage["input_units"] + usage["output_units"]
                    outcome, reason = "completed", "final_usage_received"
                    break
        except Cancelled:
            outcome, reason = "cancelled", "cancelled"
        except StreamFailure as exc:
            reason = str(exc)  # Only locally-defined constant codes reach here.
        except (OSError, http.client.HTTPException, ProtocolError):
            outcome, reason = "cancelled", "downstream_disconnected_or_slow"
        finally:
            stop.set()
            run.close_upstream()
            if producer is not None:
                producer.join(self.settings.upstream_timeout + 0.5)
            if response is not None:
                response.close()
            if connection is not None:
                connection.close()
            record = self.ledger.finish(run.tenant, run.request_id, outcome, actual,
                                        known_unstarted=not run.processing_possible)
            self.audit.emit("finished", tenant=run.tenant, request_id=run.request_id,
                            outcome=outcome, reason=reason, usage_state=record["usage_state"],
                            actual_units=actual)
            if actual is None:
                error = {"code": reason, "usage_state": record["usage_state"]}
                if emitted:
                    try:
                        send_chunk(handler, encode_event("error", error))
                        end_chunks(handler)
                    except OSError:
                        pass
                else:
                    handler.reply(409 if outcome == "cancelled" else 503, {"error": error})

    @property
    def url(self):
        return "http://127.0.0.1:" + str(self.server.server_port)

    def run_stats(self, tenant, request_id):
        with self.lock:
            run = self.running[tenant, request_id]
        return run.stats()

    def start(self):
        self.thread.start()
        return self

    def close(self):
        with self.lock:
            runs = list(self.running.values())
        for run in runs:
            run.cancel()
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(2)

    def __enter__(self):
        return self.start()

    def __exit__(self, *args):
        self.close()

"""The book's explicit teaching protocol; this is not an OpenAI-compatible API."""
import json

MAX_FRAME_BYTES = 32 * 1024
UNIT_CONTRACT = "utf8-bytes-v1"


class ProtocolError(Exception):
    pass


def encode_event(event, data):
    wire = ("event: " + event + "\ndata: " + json.dumps(data, ensure_ascii=False,
            separators=(",", ":")) + "\n\n").encode("utf-8")
    if len(wire) > MAX_FRAME_BYTES:
        raise ProtocolError("frame_too_large")
    return wire


def read_events(response):
    """Read complete bounded SSE frames from an HTTPResponse, including chunk decoding.

    Deliberately accept only the event/data subset used by the lab. A truncated
    final frame is an error, never an event that can be executed or billed.
    """
    event, data, size = None, [], 0
    first_line = True
    for line, wire_size, terminated in _lines(response):
        size += wire_size
        if size > MAX_FRAME_BYTES:
            raise ProtocolError("frame_too_large")
        if not terminated:
            raise ProtocolError("truncated_frame")
        try:
            text = line.decode("utf-8-sig" if first_line else "utf-8")
        except UnicodeDecodeError as exc:
            raise ProtocolError("invalid_utf8") from exc
        first_line = False
        if not text:
            if event is not None or data:
                try:
                    payload = json.loads("\n".join(data))
                except (ValueError, TypeError) as exc:
                    raise ProtocolError("invalid_event_json") from exc
                if event is None or not isinstance(payload, dict):
                    raise ProtocolError("invalid_event")
                yield event, payload
            event, data, size = None, [], 0
        elif text.startswith(":"):
            pass
        else:
            field, _, value = text.partition(":")
            if value.startswith(" "):
                value = value[1:]
            if field == "event":
                event = value
            elif field == "data":
                data.append(value)
            else:
                raise ProtocolError("unsupported_sse_field")
    if size:
        raise ProtocolError("truncated_frame")


def _lines(response):
    """Incremental SSE lines: CR, LF and CRLF, including split UTF-8/CRLF.

    Charge CR as two bytes immediately, reserving a possible following LF;
    skip that LF if present. This never undercounts wire bytes and dispatches
    CR-only empty lines immediately. A CR-only frame may be rejected slightly
    before the wire limit. The caller also bounds the sum of lines in a frame.
    """
    line = bytearray()
    skip_lf = False
    while True:
        chunk = response.read1(4096)
        if not chunk:
            if line:
                yield bytes(line), len(line), False
            return
        for byte in chunk:
            if skip_lf:
                skip_lf = False
                if byte == 10:
                    continue
            if byte == 13:
                yield bytes(line), len(line) + 2, True
                line.clear()
                skip_lf = True
            elif byte == 10:
                yield bytes(line), len(line) + 1, True
                line.clear()
            else:
                line.append(byte)
                if len(line) > MAX_FRAME_BYTES:
                    raise ProtocolError("frame_too_large")


def send_chunk(handler, data):
    handler.wfile.write(format(len(data), "x").encode() + b"\r\n" + data + b"\r\n")
    handler.wfile.flush()


def end_chunks(handler):
    handler.wfile.write(b"0\r\n\r\n")
    handler.wfile.flush()

"""A real HTTP model substitute with observable execution and fault boundaries."""
from collections import deque
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import socket
import threading
import time

from .protocol import UNIT_CONTRACT, encode_event, end_chunks, send_chunk


class QuietServer(ThreadingHTTPServer):
    daemon_threads = True
    block_on_close = False
    request_queue_size = 64

    def handle_error(self, request, client_address):
        # Never print request bodies, authorization headers or exception text.
        pass


@dataclass(frozen=True)
class Plan:
    mode: str = "normal"
    chunks: tuple = ("hello", " gateway")
    delay: float = 0.01
    first_delay: float = 0.0
    raw_chunks: tuple = ()
    release_after_first_chunk: object = None


class ModelStub:
    """Loopback model service, not a simulated policy function.

    X-Lab-Execution: not-started is an explicit private protocol promise: for
    this one response the service did not start execution or accrue usage.
    The unknown_503 plan makes the opposite counterexample observable.
    """

    def __init__(self, plans=(), credential="fixture-upstream-secret"):
        self.plans = deque(plans)
        self.credential = credential
        self.lock = threading.Lock()
        self.requests = 0
        self.operations_started = 0
        self.events_written = 0
        self.disconnects = 0
        owner = self

        class Handler(BaseHTTPRequestHandler):
            protocol_version = "HTTP/1.1"

            def log_message(self, *args):
                pass

            def do_POST(self):
                self.close_connection = True
                if self.path != "/generate":
                    self.reply(404, {"error": "not_found"})
                    return
                if self.headers.get("Authorization") != "Bearer " + owner.credential:
                    self.reply(401, {"error": "unauthorized"})
                    return
                try:
                    n = int(self.headers.get("Content-Length", "0"))
                    if not 0 < n <= 65536:
                        raise ValueError
                    request = json.loads(self.rfile.read(n))
                    prompt = request["prompt"]
                    maximum = request["max_output_units"]
                    if not isinstance(prompt, str) or type(maximum) is not int or maximum < 1:
                        raise ValueError
                except (ValueError, KeyError, TypeError):
                    self.reply(400, {"error": "invalid_request"})
                    return
                with owner.lock:
                    owner.requests += 1
                    plan = owner.plans.popleft() if owner.plans else Plan()
                    if plan.mode != "safe_503":
                        owner.operations_started += 1
                if plan.mode in {"safe_503", "unknown_503"}:
                    self.reply(503, {"error": "unavailable"},
                               {"X-Lab-Execution": "not-started"} if plan.mode == "safe_503" else {})
                    return
                self.connection.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096)
                self.connection.settimeout(3)
                try:
                    self.send_response(200)
                    self.send_header("Content-Type", "text/event-stream")
                    self.send_header("Transfer-Encoding", "chunked")
                    self.send_header("Connection", "close")
                    self.end_headers()
                    if plan.first_delay:
                        time.sleep(plan.first_delay)
                    if plan.raw_chunks:
                        for index, raw in enumerate(plan.raw_chunks):
                            send_chunk(self, raw)
                            if index == 0 and plan.release_after_first_chunk is not None:
                                plan.release_after_first_chunk.wait(3)
                        end_chunks(self)
                        return
                    if plan.mode == "pre_event_disconnect":
                        return  # No terminal chunk and no not-started promise.
                    produced = 0
                    for i, text in enumerate(plan.chunks):
                        text = text.encode("utf-8")[:maximum - produced].decode("utf-8", errors="ignore")
                        if not text:
                            break
                        if plan.delay:
                            time.sleep(plan.delay)
                        send_chunk(self, encode_event("token", {"text": text}))
                        produced += len(text.encode("utf-8"))
                        with owner.lock:
                            owner.events_written += 1
                        if plan.mode == "midstream_disconnect" and i == 0:
                            return
                    if plan.mode != "omit_usage":
                        usage = {"contract": UNIT_CONTRACT,
                                 "input_units": len(prompt.encode("utf-8")),
                                 "output_units": produced}
                        if plan.mode == "invalid_usage":
                            usage["output_units"] += 1
                        send_chunk(self, encode_event("usage", usage))
                    send_chunk(self, encode_event("done", {}))
                    end_chunks(self)
                except (OSError, TimeoutError):
                    with owner.lock:
                        owner.disconnects += 1

            def reply(self, status, payload, extra=None):
                data = json.dumps(payload).encode()
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(data)))
                self.send_header("Connection", "close")
                for name, value in (extra or {}).items():
                    self.send_header(name, value)
                self.end_headers()
                self.wfile.write(data)

        self.server = QuietServer(("127.0.0.1", 0), Handler)
        self.thread = threading.Thread(target=lambda: self.server.serve_forever(poll_interval=0.02), daemon=True)

    @property
    def url(self):
        return "http://127.0.0.1:" + str(self.server.server_port)

    def start(self):
        self.thread.start()
        return self

    def close(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(2)

    def __enter__(self):
        return self.start()

    def __exit__(self, *args):
        self.close()

"""Real loopback HTTP/SSE tests: no vendor SDKs, model keys or policy-only mocks."""
from concurrent.futures import ThreadPoolExecutor
from contextlib import ExitStack
import http.client
import json
import socket
import threading
import time
import unittest
from urllib.parse import urlsplit

from lab.client import cancel, generate, open_stream
from lab.gateway import Gateway, Route, Settings, TenantPolicy
from lab.protocol import UNIT_CONTRACT, read_events
from lab.upstream import ModelStub, Plan


def request(request_id="example", **changes):
    data = {"request_id": request_id, "prompt": "hi", "capabilities": ["text"],
            "max_output_units": 100}
    data.update(changes)
    return data


def eventually(check, timeout=3):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        result = check()
        if result:
            return result
        time.sleep(0.01)
    raise AssertionError("condition did not become true")


class GatewayHTTPTests(unittest.TestCase):
    def setUp(self):
        self.stack = ExitStack()
        self.addCleanup(self.stack.close)

    def stub(self, *plans):
        return self.stack.enter_context(ModelStub(plans))

    def gateway(self, routes, **kwargs):
        return self.stack.enter_context(Gateway(routes, **kwargs))

    def finished(self, gateway, request_id, tenant="tenant-a"):
        return eventually(lambda: (r if r and not r["active"] else None)
                          if (r := gateway.ledger.record(tenant, request_id)) is not None else None)

    def test_real_http_stream_and_trusted_byte_usage(self):
        stub = self.stub(Plan(chunks=("你好", " AI")))
        gateway = self.gateway([Route("local", stub.url)])
        status, events = generate(gateway.url, request(prompt="问题", max_output_units=20))
        self.assertEqual(status, 200)
        self.assertEqual([e for e, _ in events], ["token", "token", "usage", "done"])
        self.assertEqual(events[-2][1], {"contract": UNIT_CONTRACT, "input_units": 6, "output_units": 9})
        record = self.finished(gateway, "example")
        self.assertEqual(record["usage_state"], "settled")
        self.assertEqual(record["actual_units"], 15)
        self.assertEqual(gateway.ledger.snapshot("tenant-a")["held_units"], 0)
        self.assertEqual(stub.operations_started, 1)

    def test_capability_filter_precedes_admission(self):
        text, image = self.stub(), self.stub()
        gateway = self.gateway([Route("text-only", text.url),
                                Route("vision", image.url, frozenset({"text", "image"}))])
        status, _ = generate(gateway.url, request(capabilities=["image"]))
        self.assertEqual(status, 200)
        self.assertEqual((text.requests, image.requests), (0, 1))
        status, body = generate(gateway.url, request("audio", capabilities=["audio"]))
        self.assertEqual((status, body["error"]), (422, "unsupported_capability"))
        self.assertIsNone(gateway.ledger.record("tenant-a", "audio"))

    def test_default_never_replays_post_even_trusted_not_started(self):
        first, second = self.stub(Plan("safe_503")), self.stub()
        gateway = self.gateway([Route("first", first.url, trust_not_started=True), Route("second", second.url)])
        status, _ = generate(gateway.url, request())
        self.assertEqual(status, 503)
        self.assertEqual((first.requests, second.requests), (1, 0))
        self.assertEqual(self.finished(gateway, "example")["usage_state"], "released")

    def test_explicit_failover_with_trusted_not_started_promise(self):
        first, second = self.stub(Plan("safe_503")), self.stub()
        gateway = self.gateway([Route("first", first.url, trust_not_started=True), Route("second", second.url)])
        status, events = generate(gateway.url, request(allow_failover=True))
        self.assertEqual(status, 200)
        self.assertEqual(events[-1], ("done", {}))
        self.assertEqual((first.requests, first.operations_started, second.operations_started), (1, 0, 1))
        self.assertEqual(self.finished(gateway, "example")["usage_state"], "settled")

    def test_unknown_503_never_replayed_and_keeps_reservation(self):
        first, second = self.stub(Plan("unknown_503")), self.stub()
        gateway = self.gateway([Route("first", first.url, trust_not_started=True), Route("second", second.url)])
        status, _ = generate(gateway.url, request(allow_failover=True))
        self.assertEqual(status, 503)
        self.assertEqual((first.operations_started, second.requests), (1, 0))
        self.assertEqual(self.finished(gateway, "example")["usage_state"], "unknown")
        self.assertEqual(gateway.ledger.snapshot("tenant-a")["held_units"], 102)

    def test_untrusted_route_cannot_declare_safe_replay(self):
        first, second = self.stub(Plan("safe_503")), self.stub()
        gateway = self.gateway([Route("untrusted", first.url), Route("second", second.url)])
        status, _ = generate(gateway.url, request(allow_failover=True))
        self.assertEqual(status, 503)
        self.assertEqual(second.requests, 0)
        self.assertEqual(self.finished(gateway, "example")["usage_state"], "unknown")

    def test_failover_has_finite_attempt_bound(self):
        a, b, c = self.stub(Plan("safe_503")), self.stub(Plan("safe_503")), self.stub()
        gateway = self.gateway([Route("a", a.url, trust_not_started=True),
                                Route("b", b.url, trust_not_started=True), Route("c", c.url)])
        self.assertEqual(generate(gateway.url, request(allow_failover=True))[0], 503)
        self.assertEqual((a.requests, b.requests, c.requests), (1, 1, 0))
        self.assertEqual(self.finished(gateway, "example")["usage_state"], "released")

    def test_connect_failure_has_no_automatic_new_attempt(self):
        unavailable = socket.socket()
        unavailable.bind(("127.0.0.1", 0))  # Reserved port, deliberately not listening.
        self.addCleanup(unavailable.close)
        second = self.stub()
        gateway = self.gateway([Route("unavailable", "http://127.0.0.1:" + str(unavailable.getsockname()[1])),
                                Route("backup", second.url)])
        status, _ = generate(gateway.url, request(allow_failover=True))
        self.assertEqual(status, 503)
        self.assertEqual(second.requests, 0)
        self.assertEqual(self.finished(gateway, "example")["usage_state"], "released")

    def test_sse_line_endings_multidata_and_fragmented_utf8_over_http(self):
        for ending in ["\n", "\r\n", "\r"]:
            with self.subTest(ending=repr(ending)):
                lines = [": comment", "event:token", 'data:{"text":', 'data: "你"}', "",
                         "event:usage", 'data:{"contract":"utf8-bytes-v1","input_units":2,"output_units":3}', "",
                         "event:done", "data:{}", "", ""]
                wire = ending.join(lines).encode("utf-8")
                first_byte = wire.index("你".encode())
                chunks = (wire[:first_byte + 1], wire[first_byte + 1:first_byte + 2], wire[first_byte + 2:])
                stub = self.stub(Plan(raw_chunks=chunks))
                gateway = self.gateway([Route("local", stub.url)])
                status, events = generate(gateway.url, request())
                self.assertEqual(status, 200)
                self.assertEqual(events[0], ("token", {"text": "你"}))
                self.assertEqual(events[-1][0], "done")
                self.assertEqual(self.finished(gateway, "example")["actual_units"], 5)

    def test_eof_without_empty_sse_line_does_not_dispatch_partial_token(self):
        stub = self.stub(Plan(raw_chunks=(b'event:token\ndata:{"text":"partial"}\n',)))
        gateway = self.gateway([Route("local", stub.url)])
        status, body = generate(gateway.url, request())
        self.assertEqual(status, 503)
        self.assertEqual(body["error"]["usage_state"], "unknown")
        self.assertEqual(self.finished(gateway, "example")["usage_state"], "unknown")

    def test_cr_only_complete_event_is_dispatched_before_any_later_bytes(self):
        release = threading.Event()
        first = b'event:token\rdata:{"text":"x"}\r\r'
        later = (b'event:usage\rdata:{"contract":"utf8-bytes-v1","input_units":2,"output_units":1}\r\r'
                 b'event:done\rdata:{}\r\r')
        stub = self.stub(Plan(raw_chunks=(first, later), release_after_first_chunk=release))
        gateway = self.gateway([Route("local", stub.url)])
        with ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(open_stream, gateway.url, request())
            try:
                # Stub cannot write later bytes until we release it. Getting a
                # valid token here therefore proves immediate CR dispatch.
                connection, response = future.result(timeout=1)
                try:
                    events = read_events(response)
                    self.assertEqual(next(events), ("token", {"text": "x"}))
                    self.assertFalse(release.is_set())
                    release.set()
                    self.assertEqual(list(events)[-1], ("done", {}))
                finally:
                    response.close()
                    connection.close()
            finally:
                release.set()
        self.assertEqual(self.finished(gateway, "example")["actual_units"], 3)

    def test_oversized_sse_frame_is_rejected_before_forwarding(self):
        wire = b'event:token\ndata:{"text":"' + b'x' * (33 * 1024) + b'"}\n\n'
        stub = self.stub(Plan(raw_chunks=(wire,)))
        gateway = self.gateway([Route("local", stub.url)])
        status, _ = generate(gateway.url, request())
        self.assertEqual(status, 503)
        self.assertEqual(self.finished(gateway, "example")["usage_state"], "unknown")

    def test_no_output_does_not_prove_no_execution(self):
        first, second = self.stub(Plan("pre_event_disconnect")), self.stub()
        gateway = self.gateway([Route("first", first.url, trust_not_started=True), Route("second", second.url)])
        self.assertEqual(generate(gateway.url, request(allow_failover=True))[0], 503)
        self.assertEqual((first.operations_started, second.requests), (1, 0))
        self.assertEqual(self.finished(gateway, "example")["usage_state"], "unknown")

    def test_midstream_disconnect_reports_error_without_replay(self):
        first, second = self.stub(Plan("midstream_disconnect")), self.stub()
        gateway = self.gateway([Route("first", first.url, trust_not_started=True), Route("second", second.url)])
        status, events = generate(gateway.url, request(allow_failover=True))
        self.assertEqual(status, 200)
        self.assertEqual([e for e, _ in events], ["token", "error"])
        self.assertEqual(events[-1][1]["usage_state"], "unknown")
        self.assertEqual((first.operations_started, second.requests), (1, 0))
        self.finished(gateway, "example")

    def test_first_event_timeout_is_unknown_and_not_replayed(self):
        first, second = self.stub(Plan(first_delay=0.25)), self.stub()
        gateway = self.gateway([Route("first", first.url, trust_not_started=True), Route("second", second.url)],
                               settings=Settings(upstream_timeout=0.08))
        self.assertEqual(generate(gateway.url, request(allow_failover=True))[0], 503)
        self.assertEqual(second.requests, 0)
        self.assertEqual(self.finished(gateway, "example")["usage_state"], "unknown")

    def test_missing_or_false_usage_cannot_settle(self):
        for mode in ["omit_usage", "invalid_usage"]:
            with self.subTest(mode=mode):
                stub = self.stub(Plan(mode))
                gateway = self.gateway([Route("local", stub.url)])
                status, events = generate(gateway.url, request())
                self.assertEqual(status, 200)
                self.assertEqual(events[-1][0], "error")
                self.assertEqual(self.finished(gateway, "example")["usage_state"], "unknown")

    def test_client_cannot_set_usage_or_tenant(self):
        stub = self.stub()
        gateway = self.gateway([Route("local", stub.url)])
        for extra in [{"usage": {"output_units": 0}}, {"tenant": "tenant-b"}, {"allow_failover": "true"}]:
            with self.subTest(extra=extra):
                self.assertEqual(generate(gateway.url, request(**extra))[0], 400)
        self.assertEqual(stub.requests, 0)

    def test_output_text_is_data_and_never_accounting(self):
        text = '{"usage":{"input_units":0,"output_units":0}}'
        stub = self.stub(Plan(chunks=(text,)))
        gateway = self.gateway([Route("local", stub.url)])
        status, events = generate(gateway.url, request())
        self.assertEqual(status, 200)
        self.assertEqual(events[0], ("token", {"text": text}))
        self.assertEqual(self.finished(gateway, "example")["actual_units"], len(text) + 2)

    def test_atomic_tenant_concurrency_admission(self):
        stub = self.stub(Plan(chunks=("x",) * 200, delay=0.01))
        gateway = self.gateway([Route("local", stub.url)], policies={"tenant-a": TenantPolicy(max_concurrent=1)})
        connection, response = open_stream(gateway.url, request("held", max_output_units=200))
        self.addCleanup(connection.close)
        self.addCleanup(response.close)
        self.assertEqual(next(read_events(response))[0], "token")
        self.assertEqual(generate(gateway.url, request("excess"))[0], 429)
        self.assertEqual(stub.requests, 1)
        self.assertEqual(cancel(gateway.url, "held")[0], 202)
        self.finished(gateway, "held")

    def test_parallel_reservation_cannot_overspend(self):
        stub = self.stub(Plan(first_delay=0.05))
        gateway = self.gateway([Route("local", stub.url)],
                               policies={"tenant-a": TenantPolicy(budget_units=150, max_concurrent=10)})
        barrier = threading.Barrier(6)

        def call(i):
            barrier.wait()
            return generate(gateway.url, request("parallel-" + str(i)))[0]

        with ThreadPoolExecutor(max_workers=6) as executor:
            statuses = list(executor.map(call, range(6)))
        # Initial reservation is 102; after completion actual is 15. Timing can
        # admit more later calls, but the balance must never become negative.
        self.assertIn(200, statuses)
        self.assertIn(402, statuses)
        self.assertTrue(all(s in {200, 402} for s in statuses))
        self.assertGreaterEqual(gateway.ledger.snapshot("tenant-a")["available_units"], 0)

    def test_interrupted_usage_is_held_until_authoritative_reconciliation(self):
        stub = self.stub(Plan("midstream_disconnect"))
        gateway = self.gateway([Route("local", stub.url)],
                               policies={"tenant-a": TenantPolicy(budget_units=110)})
        generate(gateway.url, request())
        self.finished(gateway, "example")
        self.assertEqual(generate(gateway.url, request("blocked"))[0], 402)
        self.assertEqual(gateway.ledger.snapshot("tenant-a")["active"], 0)
        gateway.ledger.reconcile("tenant-a", "example", 7)
        self.assertEqual(gateway.ledger.snapshot("tenant-a")["spent_units"], 7)
        self.assertEqual(gateway.ledger.snapshot("tenant-a")["held_units"], 0)
        self.assertEqual(generate(gateway.url, request("after-reconcile"))[0], 200)

    def test_duplicate_id_does_not_duplicate_upstream_side_effect(self):
        stub = self.stub()
        gateway = self.gateway([Route("local", stub.url)])
        self.assertEqual(generate(gateway.url, request())[0], 200)
        self.finished(gateway, "example")
        self.assertEqual(generate(gateway.url, request())[0], 409)
        self.assertEqual(stub.operations_started, 1)

    def test_cancellation_is_tenant_scoped_and_propagates_upstream(self):
        stub = self.stub(Plan(chunks=("x",) * 1000, delay=0.005))
        gateway = self.gateway([Route("local", stub.url)],
                               policies={"tenant-a": TenantPolicy(), "tenant-b": TenantPolicy()},
                               credentials={"tenant-a-secret": "tenant-a", "tenant-b-secret": "tenant-b"})
        connection, response = open_stream(gateway.url, request("cancel-me", max_output_units=1000))
        self.addCleanup(connection.close)
        self.addCleanup(response.close)
        self.assertEqual(next(read_events(response))[0], "token")
        self.assertEqual(cancel(gateway.url, "cancel-me", "tenant-b-secret")[0], 404)
        self.assertTrue(gateway.ledger.record("tenant-a", "cancel-me")["active"])
        self.assertEqual(cancel(gateway.url, "cancel-me")[0], 202)
        record = self.finished(gateway, "cancel-me")
        self.assertEqual((record["outcome"], record["usage_state"]), ("cancelled", "unknown"))
        eventually(lambda: stub.disconnects > 0)
        eventually(lambda: gateway.run_stats("tenant-a", "cancel-me")["producer_stopped"])
        self.assertEqual(gateway.ledger.snapshot("tenant-a")["active"], 0)

    def test_slow_consumer_has_bounded_event_queue_and_backpressure(self):
        stub = self.stub(Plan(chunks=("x" * 2048,) * 2000, delay=0))
        gateway = self.gateway([Route("local", stub.url)],
                               policies={"tenant-a": TenantPolicy(budget_units=5_000_000)},
                               settings=Settings(queue_events=2, client_write_timeout=0.5))
        parsed = urlsplit(gateway.url)
        connection = http.client.HTTPConnection(parsed.hostname, parsed.port, timeout=5)
        connection.connect()
        connection.sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024)
        connection.request("POST", "/v1/generate", body=json.dumps(request("slow", max_output_units=4_096_000)),
                           headers={"Authorization": "Bearer tenant-a-secret", "Content-Type": "application/json"})
        response = connection.getresponse()
        self.addCleanup(connection.close)
        self.addCleanup(response.close)
        self.assertEqual(response.status, 200)
        eventually(lambda: gateway.run_stats("tenant-a", "slow")["backpressure_waits"] > 0)
        self.assertLessEqual(gateway.run_stats("tenant-a", "slow")["queue_peak"], 2)
        self.assertLess(stub.events_written, 2000)
        cancel(gateway.url, "slow")
        self.assertEqual(self.finished(gateway, "slow")["usage_state"], "unknown")
        eventually(lambda: stub.disconnects > 0)

    def test_downstream_socket_close_stops_upstream_without_explicit_cancel(self):
        stub = self.stub(Plan(chunks=("x",) * 1000, delay=0.005))
        gateway = self.gateway([Route("local", stub.url)])
        connection, response = open_stream(gateway.url, request("disconnect", max_output_units=1000))
        self.assertEqual(next(read_events(response))[0], "token")
        # Force a real TCP close, not just abandoning an iterator while keeping
        # an HTTPResponse buffered file and socket alive.
        response.fp.raw._sock.shutdown(socket.SHUT_RDWR)
        response.close()
        connection.close()
        self.assertEqual(self.finished(gateway, "disconnect")["usage_state"], "unknown")
        eventually(lambda: stub.disconnects > 0)

    def test_audit_is_useful_without_prompt_tokens_or_keys(self):
        marker = "sensitive-prompt-marker"
        output = "sensitive-generated-marker"
        stub = self.stub(Plan(chunks=(output,)))
        gateway = self.gateway([Route("local", stub.url)])
        generate(gateway.url, request(prompt=marker))
        self.finished(gateway, "example")
        audit = gateway.audit.jsonl()
        self.assertIn('"event": "route_attempt"', audit)
        self.assertIn('"usage_state": "settled"', audit)
        for secret in [marker, output, "tenant-a-secret", "fixture-upstream-secret", "Authorization"]:
            self.assertNotIn(secret, audit)
        self.assertEqual(generate(gateway.url, request("unauth"), "wrong-secret")[0], 401)
        self.assertEqual(stub.requests, 1)


if __name__ == "__main__":
    unittest.main()