Add owner-metered Messages transport with conservative admission

Assistant: codex
Assistant-Model: gpt-5.6-luna
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

SCOPE.md

@@ -34,6 +34,9 @@ re-export shim pointing here.
 - TOML-based configuration resolution via `toml_config.py` and `config.py`.
 - Shared HTTP utilities, token estimation, similarity helpers, and the
   `LLMError` exception hierarchy.
+- An opt-in owner-hosted Messages transport with an injected request-admission
+  protocol; caller-owned ledgers, hosting, custody and network policy remain
+  outside this library. See `contracts/functional/messages-admission.md`.
 
 ---
 

contracts/functional/messages-admission.md

@@ -0,0 +1,83 @@
+# Owner-hosted Messages admission
+
+`llm_connect.messages_gate` supplies an opt-in, stdlib-only Messages listener.
+It is separate from `LLMServer` and is never enabled by normal `serve` mode.
+The work record is `LLM-WP-0009`; factory integration is HFACT-WP-0001-T01 and
+REINAH-WP-0003-T05/T06.
+
+## Contract
+
+The owner constructs immutable `MessagesPolicy` with an exact model, tariff
+reference, maximum admitted context/output, integer micro-USD per-token upper
+rates, explicit beta allowlist, body limit and request timeout. There is no
+built-in live price, FX source, token estimate or default beta grant. Input rates
+must conservatively cover input, both cache-write lifetimes, cache reads and all
+accepted multipliers. Accepted provider limits and tariff validity remain an
+operator policy responsibility; a fixture policy cannot establish them.
+
+`RequestMeter.reserve_request(token, policy_sha256, liability_microusd)` must
+atomically check authority and remaining parent capacity, persist the hold and
+return an opaque receipt **before** the transport opens a provider connection.
+The reserved amount is full context times the maximum input rate plus requested
+output times the maximum output rate. This deliberately trades utilization for
+a bound that does not depend on unproven local token estimation.
+
+`request_active(receipt)` checks the run and lease. `complete_request(receipt,
+observed_microusd)` records trusted terminal accounting while retaining the full
+reservation. Any exception, cancellation, non-200 response, truncated stream,
+unknown fee/model/content feature or missing usage leaves an unknown hold. A
+retry is another request and cannot be forwarded through an unresolved hold.
+The forwarder neither retries nor follows redirects; it ignores proxy variables
+and accepts only a fixed HTTPS provider origin (loopback HTTP is explicit for
+tests). Only owner headers and the accepted beta list reach that origin.
+
+The listener accepts `POST /v1/messages[?beta=true]` with an opaque `x-api-key`
+route token. No `/execute`, arbitrary URL, alternate authorization or compressed
+request route is available. Duplicate JSON/header fields and unsupported API
+features refuse. The supported subset is streaming text, custom client tools,
+ordinary thinking/effort and ephemeral cache controls. Context management is
+limited to the CLI's exact keep-all-thinking form; server compaction, server
+tools, media/URL blocks, model fallback, extended context and extra paid features
+need separate implementation and accepted bounds.
+
+SSE terminal accounting includes input, output, cache creation and cache reads;
+output deltas are cumulative. The final stream event is held until the durable
+completion returns so a following tool-loop request cannot race that write.
+Request bodies, provider bodies and credentials are not written to logs or
+request receipts. Client metadata is untrusted context, never authorization.
+
+## Owning integration and remaining admission
+
+Rein's `RequestLedger` implements this protocol as child holds in the existing
+private SQLite envelope. Parent reservations continue to account for daily/total
+capacity; child requests cannot mint a second allowance. Its explicit schema
+provisioning never runs automatically on missing state. A trusted owner binds
+one route to the admitted parent run, exact policy digest and queue lease expiry;
+the parent already binds worker, definition, project, target, grant and runtime
+digests. Route tokens are random, stored only as hashes, cannot be rebound or
+renewed by the workload, and revoke on parent terminal observation.
+
+The next integration must host the listener in the trusted owner boundary, keep
+provider credentials and ledger inaccessible to the sandbox, deliver only its
+run token/base URL, bind actual lease loss to route revocation, and prove direct
+provider and alternate-route denial. This module does not install a listener,
+configure a sandbox, resolve credentials, or promote a profile. The installed
+CLI fixture demonstrates transport/ledger behavior in a fake-provider namespace;
+it does **not** prove secret or network separation between real owner/workload
+processes. LLM-WP-0009-T03 retains this owner integration return.
+
+## Verification and primary protocol references
+
+`tests/test_messages_gate.py` exercises the policy, protocol and bounded refusal.
+Rein's `tests/test_request_admission.py` exercises actual HTTP and SQLite capacity,
+concurrency, recovery, lease loss and uncertain outcomes. Its opt-in
+`tests/test_native_cli_boundary.py` also uses the installed 2.1.266 CLI: the
+USD 0.01 counterexample is refused before any fake upstream call, while a
+permitted two-request tool session succeeds. All rates/FX/keys in those tests
+are synthetic and have no production authority.
+
+Protocol references inspected 2026-09-09:
+[Messages](https://platform.claude.com/docs/en/api/messages/create),
+[streaming](https://platform.claude.com/docs/en/build-with-claude/streaming),
+[context editing](https://platform.claude.com/docs/en/build-with-claude/context-editing),
+[prompt caching](https://platform.claude.com/docs/en/build-with-claude/prompt-caching).

llm_connect/messages_gate.py

@@ -0,0 +1,534 @@
+"""Opt-in Anthropic Messages transport with owner-supplied request admission.
+
+This listener has no /execute route, environment credential discovery, proxy,
+redirect or automatic retry. Hosting, accepted tariffs, custody and egress remain
+the caller's responsibility. No production listener is installed by this module.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import http.client
+import json
+import re
+import threading
+import time
+from dataclasses import asdict, dataclass
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Protocol
+from urllib.parse import urlsplit
+
+
+class RequestRefused(RuntimeError):
+    """Bounded refusal; never carries request content or credentials."""
+
+
+def _integer(value: object, upper: int) -> bool:
+    return type(value) is int and 0 < value <= upper
+
+
+def _unique(pairs):
+    result = {}
+    for key, value in pairs:
+        if key in result:
+            raise ValueError("duplicate field")
+        result[key] = value
+    return result
+
+
+def _json(raw: bytes):
+    def invalid(_):
+        raise ValueError("nonfinite number")
+
+    return json.loads(raw, object_pairs_hook=_unique, parse_constant=invalid)
+
+
+@dataclass(frozen=True)
+class MessagesPolicy:
+    """Owner-accepted upper bounds, not a built-in price list or token estimate.
+
+    input_microusd_per_token must cover the highest admitted input/cache rate,
+    including residency/tier multipliers. Reserve the entire admitted context.
+    One microusd = USD 0.000001; rounding is the accepting owner's responsibility.
+    """
+
+    tariff_ref: str
+    model: str
+    context_tokens: int
+    max_output_tokens: int
+    input_microusd_per_token: int
+    output_microusd_per_token: int
+    allowed_betas: tuple[str, ...] = ()
+    max_body_bytes: int = 2_000_000
+    timeout_seconds: int = 120
+
+    def __post_init__(self):
+        for value in (self.tariff_ref, self.model):
+            if not isinstance(value, str) or not re.fullmatch(
+                r"[A-Za-z0-9][A-Za-z0-9._:/@+-]{0,199}", value
+            ):
+                raise RequestRefused("invalid request policy identity")
+        for value, upper in (
+            (self.context_tokens, 10_000_000),
+            (self.max_output_tokens, 1_000_000),
+            (self.input_microusd_per_token, 1_000_000),
+            (self.output_microusd_per_token, 1_000_000),
+            (self.max_body_bytes, 2_000_000),
+            (self.timeout_seconds, 900),
+        ):
+            if not _integer(value, upper):
+                raise RequestRefused("invalid request policy bound")
+        if not isinstance(self.allowed_betas, tuple) or len(set(self.allowed_betas)) != len(
+            self.allowed_betas
+        ):
+            raise RequestRefused("invalid beta policy")
+        if any(
+            not isinstance(b, str) or not re.fullmatch(r"[a-z0-9-]{1,100}", b)
+            for b in self.allowed_betas
+        ):
+            raise RequestRefused("invalid beta policy")
+
+    @property
+    def sha256(self) -> str:
+        return hashlib.sha256(
+            json.dumps(asdict(self), sort_keys=True, separators=(",", ":")).encode()
+        ).hexdigest()
+
+    def validate(self, data: object, betas: str) -> int:
+        """Validate a narrow text/custom-tool Messages subset, then bound cost."""
+        allowed = {
+            "model",
+            "messages",
+            "max_tokens",
+            "stream",
+            "system",
+            "tools",
+            "tool_choice",
+            "metadata",
+            "thinking",
+            "output_config",
+            "temperature",
+            "top_p",
+            "top_k",
+            "stop_sequences",
+            "cache_control",
+            "context_management",
+        }
+        if not isinstance(data, dict) or set(data) - allowed:
+            raise RequestRefused("unsupported request fields")
+        if data.get("model") != self.model or data.get("stream") is not True:
+            raise RequestRefused("request model or stream not admitted")
+        output = data.get("max_tokens")
+        if not _integer(output, self.max_output_tokens):
+            raise RequestRefused("output limit not admitted")
+        if set(filter(None, (b.strip() for b in betas.split(",")))) - set(self.allowed_betas):
+            raise RequestRefused("beta feature not admitted")
+        if "context_management" in data and data["context_management"] != {
+            "edits": [{"type": "clear_thinking_20251015", "keep": "all"}]
+        }:
+            raise RequestRefused("only keep-all thinking context admitted")
+
+        def cache(value):
+            if (
+                not isinstance(value, dict)
+                or set(value) - {"type", "ttl"}
+                or value.get("type") != "ephemeral"
+                or value.get("ttl", "5m") not in ("5m", "1h")
+            ):
+                raise RequestRefused("cache mode not admitted")
+
+        def blocks(value, *, system=False, nested=False):
+            if isinstance(value, str):
+                return
+            if not isinstance(value, list) or len(value) > 10000:
+                raise RequestRefused("content not admitted")
+            for block in value:
+                if not isinstance(block, dict):
+                    raise RequestRefused("content block not admitted")
+                kind = block.get("type")
+                fields = {
+                    "text": {"type", "text", "cache_control"},
+                    "tool_use": {"type", "id", "name", "input", "cache_control"},
+                    "tool_result": {"type", "tool_use_id", "content", "is_error", "cache_control"},
+                    "thinking": {"type", "thinking", "signature"},
+                    "redacted_thinking": {"type", "data"},
+                }
+                if (
+                    kind not in fields
+                    or set(block) - fields[kind]
+                    or ((system or nested) and kind != "text")
+                ):
+                    raise RequestRefused("content feature not admitted")
+                if kind == "text" and not isinstance(block.get("text"), str):
+                    raise RequestRefused("invalid text block")
+                if kind == "tool_use" and (
+                    not isinstance(block.get("input"), dict)
+                    or not isinstance(block.get("name"), str)
+                    or not isinstance(block.get("id"), str)
+                ):
+                    raise RequestRefused("invalid custom tool use")
+                if kind == "tool_result":
+                    if (
+                        not isinstance(block.get("tool_use_id"), str)
+                        or type(block.get("is_error", False)) is not bool
+                    ):
+                        raise RequestRefused("invalid custom tool result")
+                    blocks(block.get("content", ""), nested=True)
+                if "cache_control" in block:
+                    cache(block["cache_control"])
+
+        messages = data.get("messages")
+        if not isinstance(messages, list) or not 1 <= len(messages) <= 10000:
+            raise RequestRefused("messages not admitted")
+        for message in messages:
+            if (
+                not isinstance(message, dict)
+                or set(message) != {"role", "content"}
+                or message["role"] not in ("user", "assistant")
+            ):
+                raise RequestRefused("message not admitted")
+            blocks(message["content"])
+        if "system" in data:
+            blocks(data["system"], system=True)
+        tools = data.get("tools", [])
+        if not isinstance(tools, list) or len(tools) > 100:
+            raise RequestRefused("tools not admitted")
+        for tool in tools:
+            if (
+                not isinstance(tool, dict)
+                or set(tool) - {"name", "description", "input_schema", "cache_control"}
+                or not isinstance(tool.get("name"), str)
+                or not isinstance(tool.get("input_schema"), dict)
+            ):
+                raise RequestRefused("only custom client tools admitted")
+            if "cache_control" in tool:
+                cache(tool["cache_control"])
+        if "cache_control" in data:
+            cache(data["cache_control"])
+        if "thinking" in data:
+            value = data["thinking"]
+            if (
+                not isinstance(value, dict)
+                or set(value) - {"type", "budget_tokens", "display"}
+                or value.get("type") not in ("disabled", "adaptive", "enabled")
+            ):
+                raise RequestRefused("thinking mode not admitted")
+            if value.get("type") == "enabled" and not _integer(value.get("budget_tokens"), output):
+                raise RequestRefused("thinking budget not admitted")
+            if "display" in value and value["display"] not in ("summarized", "omitted"):
+                raise RequestRefused("thinking display not admitted")
+        if "output_config" in data:
+            value = data["output_config"]
+            if (
+                not isinstance(value, dict)
+                or set(value) != {"effort"}
+                or value["effort"] not in ("low", "medium", "high", "max")
+            ):
+                raise RequestRefused("output feature not admitted")
+        if "tool_choice" in data:
+            value = data["tool_choice"]
+            if (
+                not isinstance(value, dict)
+                or set(value) - {"type", "name", "disable_parallel_tool_use"}
+                or value.get("type") not in ("auto", "any", "tool", "none")
+            ):
+                raise RequestRefused("tool choice not admitted")
+        if "metadata" in data:
+            value = data["metadata"]
+            if (
+                not isinstance(value, dict)
+                or set(value) - {"user_id"}
+                or not isinstance(value.get("user_id", ""), str)
+                or len(value.get("user_id", "")) > 1000
+            ):
+                raise RequestRefused("metadata not admitted")
+        return (
+            self.context_tokens * self.input_microusd_per_token
+            + output * self.output_microusd_per_token
+        )
+
+
+class RequestMeter(Protocol):
+    """Implemented by the trusted caller; methods must commit before returning."""
+
+    def reserve_request(self, token: str, policy_sha256: str, liability_microusd: int) -> str: ...
+    def request_active(self, receipt: str) -> bool: ...
+    def complete_request(self, receipt: str, observed_microusd: int) -> None: ...
+
+
+class _Stream:
+    """Observe terminal usage without persisting provider content."""
+
+    def __init__(self, policy: MessagesPolicy):
+        self.policy = policy
+        self.started = self.stopped = self.delta = False
+        self.usage = {}
+
+    def event(self, raw: bytes):
+        data = b"\n".join(
+            line[5:].lstrip() for line in raw.splitlines() if line.startswith(b"data:")
+        )
+        if not data:
+            return
+        value = _json(data)
+        kind = value.get("type")
+        if kind == "ping":
+            return
+        if self.stopped or kind == "error":
+            raise RequestRefused("provider stream incomplete")
+        if kind == "message_start":
+            message = value.get("message", {})
+            if self.started or message.get("model") != self.policy.model:
+                raise RequestRefused("provider model mismatch")
+            self.started = True
+            self._usage(message.get("usage", {}))
+        elif kind == "message_delta":
+            if not self.started:
+                raise RequestRefused("provider stream order invalid")
+            self._usage(value.get("usage", {}))
+            self.delta = self.delta or bool(value.get("delta", {}).get("stop_reason"))
+        elif kind == "message_stop":
+            if not self.started or not self.delta:
+                raise RequestRefused("provider stream incomplete")
+            self.stopped = True
+        elif (
+            kind not in ("content_block_start", "content_block_delta", "content_block_stop")
+            or not self.started
+        ):
+            raise RequestRefused("provider stream feature not admitted")
+        elif kind == "content_block_start" and value.get("content_block", {}).get("type") not in (
+            "text",
+            "thinking",
+            "redacted_thinking",
+            "tool_use",
+        ):
+            raise RequestRefused("provider content feature not admitted")
+
+    def _usage(self, value):
+        if not isinstance(value, dict):
+            raise RequestRefused("provider usage incomplete")
+        for key, count in value.items():
+            if key in (
+                "input_tokens",
+                "output_tokens",
+                "cache_creation_input_tokens",
+                "cache_read_input_tokens",
+            ):
+                if type(count) is not int or count < self.usage.get(key, 0):
+                    raise RequestRefused("provider usage regressed")
+            if key == "server_tool_use" and count and any(count.values()):
+                raise RequestRefused("provider fee feature not admitted")
+        self.usage.update(value)
+
+    def cost(self) -> int:
+        if not self.stopped:
+            raise RequestRefused("provider stream incomplete")
+        counts = {}
+        for key in (
+            "input_tokens",
+            "output_tokens",
+            "cache_creation_input_tokens",
+            "cache_read_input_tokens",
+        ):
+            value = self.usage.get(key, 0 if key.startswith("cache_") else None)
+            if type(value) is not int or not 0 <= value <= 100_000_000:
+                raise RequestRefused("provider usage incomplete")
+            counts[key] = value
+        inputs = sum(counts[k] for k in counts if k != "output_tokens")
+        return (
+            inputs * self.policy.input_microusd_per_token
+            + counts["output_tokens"] * self.policy.output_microusd_per_token
+        )
+
+
+class _Handler(BaseHTTPRequestHandler):
+    def log_message(self, *args):
+        pass
+
+    def _error(self, status, code):
+        raw = json.dumps(
+            {"type": "error", "error": {"type": "invalid_request_error", "message": code}}
+        ).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(raw)))
+        self.end_headers()
+        self.wfile.write(raw)
+
+    def do_POST(self):
+        owner = self.server.owner
+        upstream = None
+        sent = False
+        try:
+            if self.path not in ("/v1/messages", "/v1/messages?beta=true"):
+                self._error(404, "route_not_admitted")
+                return
+            for header in (
+                "Content-Length",
+                "x-api-key",
+                "anthropic-beta",
+                "Authorization",
+                "Content-Type",
+            ):
+                if len(self.headers.get_all(header, [])) > 1:
+                    raise RequestRefused("ambiguous headers")
+            if (
+                self.headers.get("Transfer-Encoding")
+                or self.headers.get("Content-Encoding")
+                or self.headers.get("Authorization")
+            ):
+                raise RequestRefused("unsupported request encoding or authentication")
+            if self.headers.get_content_type() != "application/json":
+                raise RequestRefused("JSON required")
+            length = int(self.headers.get("Content-Length", "0"))
+            if not 0 < length <= owner.policy.max_body_bytes:
+                raise RequestRefused("request body limit")
+            self.connection.settimeout(10)
+            raw = self.rfile.read(length)
+            if len(raw) != length:
+                raise RequestRefused("incomplete request")
+            data = _json(raw)
+            betas = self.headers.get("anthropic-beta", "")
+            liability = owner.policy.validate(data, betas)
+            receipt = owner.meter.reserve_request(
+                self.headers.get("x-api-key", ""), owner.policy.sha256, liability
+            )
+            if not owner.meter.request_active(receipt):
+                raise RequestRefused("request lease lost")
+            deadline = time.monotonic() + owner.policy.timeout_seconds
+            kind = (
+                http.client.HTTPSConnection
+                if owner.endpoint.scheme == "https"
+                else http.client.HTTPConnection
+            )
+            upstream = kind(
+                owner.endpoint.hostname, owner.endpoint.port, timeout=owner.policy.timeout_seconds
+            )
+            headers = {
+                "Content-Type": "application/json",
+                "Accept": "text/event-stream",
+                "Accept-Encoding": "identity",
+                "anthropic-version": "2023-06-01",
+                "x-api-key": owner._provider_key,
+            }
+            if betas:
+                headers["anthropic-beta"] = betas
+            upstream.request(
+                "POST",
+                "/v1/messages",
+                body=json.dumps(data, allow_nan=False).encode(),
+                headers=headers,
+            )
+            response = upstream.getresponse()
+            if (
+                response.status != 200
+                or response.headers.get_content_type() != "text/event-stream"
+                or response.getheader("Content-Encoding")
+            ):
+                raise RequestRefused("provider outcome uncertain")
+            self.send_response(200)
+            self.send_header("Content-Type", "text/event-stream")
+            self.send_header("Connection", "close")
+            self.end_headers()
+            sent = True
+            self.close_connection = True
+            stream = _Stream(owner.policy)
+            event = bytearray()
+            total = 0
+            terminal_bytes = bytearray()
+            while True:
+                if not owner.meter.request_active(receipt) or time.monotonic() >= deadline:
+                    raise RequestRefused("request lease or deadline lost")
+                # read1 avoids waiting for a full buffer and preserves streaming.
+                if upstream.sock is not None:
+                    upstream.sock.settimeout(max(0.001, deadline - time.monotonic()))
+                chunk = response.read1(16384)
+                if not chunk:
+                    break
+                total += len(chunk)
+                event.extend(chunk)
+                if total > 32_000_000 or len(event) > 1_000_000:
+                    raise RequestRefused("provider stream size exceeded")
+                # SSE accepts CRLF too. Preserve original bytes on the client wire.
+                normalized = bytes(event).replace(b"\r\n", b"\n")
+                while b"\n\n" in normalized:
+                    frame, normalized = normalized.split(b"\n\n", 1)
+                    stream.event(frame)
+                event = bytearray(normalized)
+                if stream.stopped:
+                    terminal_bytes.extend(chunk)
+                else:
+                    self.wfile.write(chunk)
+                    self.wfile.flush()
+            if bytes(event).strip():
+                raise RequestRefused("provider stream truncated")
+            owner.meter.complete_request(receipt, stream.cost())
+            # Do not expose message_stop until durable completion; the CLI may
+            # immediately issue its next tool-loop request after this event.
+            self.wfile.write(terminal_bytes)
+            self.wfile.flush()
+        except Exception:
+            # Any post-reservation failure leaves a durable unknown hold.
+            # Neither upstream errors nor caller exceptions reach logs or clients.
+            if not sent:
+                self._error(400, "request_not_admitted_or_outcome_uncertain")
+        finally:
+            if upstream is not None:
+                upstream.close()
+
+
+class MessagesServer:
+    """Explicit owner-hosted listener; default loopback, never enabled by serve mode."""
+
+    def __init__(
+        self,
+        policy: MessagesPolicy,
+        meter: RequestMeter,
+        *,
+        provider_key: str,
+        upstream_url: str = "https://api.anthropic.com",
+        host: str = "127.0.0.1",
+        port: int = 0,
+        allow_test_http: bool = False,
+    ):
+        endpoint = urlsplit(upstream_url)
+        if endpoint.scheme != "https" and not (
+            allow_test_http
+            and endpoint.scheme == "http"
+            and endpoint.hostname in ("127.0.0.1", "::1")
+        ):
+            raise RequestRefused("provider transport requires HTTPS")
+        if (
+            not endpoint.hostname
+            or endpoint.username
+            or endpoint.password
+            or endpoint.path not in ("", "/")
+            or endpoint.query
+            or endpoint.fragment
+        ):
+            raise RequestRefused("fixed provider origin required")
+        if (
+            not isinstance(provider_key, str)
+            or not provider_key
+            or any(ord(c) < 33 or ord(c) > 126 for c in provider_key)
+        ):
+            raise RequestRefused("explicit provider credential required")
+        self.policy, self.meter, self.endpoint = policy, meter, endpoint
+        self._provider_key = provider_key
+        self._httpd = ThreadingHTTPServer((host, port), _Handler)
+        self._httpd.owner = self
+        self._thread = None
+
+    @property
+    def port(self):
+        return self._httpd.server_address[1]
+
+    def start(self):
+        self._thread = threading.Thread(target=self._httpd.serve_forever, daemon=True)
+        self._thread.start()
+
+    def stop(self):
+        if self._thread is not None:
+            self._httpd.shutdown()
+            self._thread.join()
+        self._httpd.server_close()

tests/test_messages_gate.py

@@ -0,0 +1,189 @@
+"""Narrow Messages admission and wire behavior; deterministic, no provider SDK."""
+
+import http.client
+import json
+from dataclasses import FrozenInstanceError, replace
+
+import pytest
+
+from llm_connect.messages_gate import (
+    MessagesPolicy,
+    MessagesServer,
+    RequestRefused,
+    _json,
+    _Stream,
+)
+
+
+@pytest.fixture
+def policy():
+    return MessagesPolicy("fixture:no-live-tariff", "fixture-model", 1000, 100, 3, 15)
+
+
+def body():
+    return {
+        "model": "fixture-model",
+        "max_tokens": 100,
+        "stream": True,
+        "messages": [{"role": "user", "content": "fixture"}],
+    }
+
+
+def test_conservative_bound_ignores_unproven_input_estimates(policy):
+    small = body()
+    large = body()
+    large["messages"][0]["content"] = "fixture " * 10000
+    assert policy.validate(small, "") == policy.validate(large, "") == 4500
+    cached = body()
+    cached["cache_control"] = {"type": "ephemeral", "ttl": "1h"}
+    assert policy.validate(cached, "") == 4500
+    with pytest.raises(FrozenInstanceError):
+        policy.context_tokens = 1
+    assert replace(policy, context_tokens=1001).sha256 != policy.sha256
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("context_tokens", True),
+        ("context_tokens", 0),
+        ("max_output_tokens", -1),
+        ("input_microusd_per_token", 1.1),
+        ("output_microusd_per_token", 0),
+        ("timeout_seconds", 1000),
+        ("max_body_bytes", 2000001),
+        ("model", "x\nheader"),
+        ("allowed_betas", ["unknown"]),
+    ],
+)
+def test_invalid_policy_refused(policy, field, value):
+    with pytest.raises(RequestRefused):
+        replace(policy, **{field: value})
+
+
+def test_beta_policy_never_implicitly_expands_context(policy):
+    with pytest.raises(RequestRefused):
+        policy.validate(body(), "context-1m-2025-08-07")
+    admitted = replace(policy, allowed_betas=("fixture-beta",))
+    assert admitted.validate(body(), "fixture-beta") == 4500
+    with pytest.raises(RequestRefused):
+        admitted.validate(body(), "fixture-beta,unknown")
+
+
+@pytest.mark.parametrize(
+    "raw", [b'{"max_tokens":1,"max_tokens":2}', b'{"a":NaN}', b'{"a":Infinity}']
+)
+def test_ambiguous_json_refused(raw):
+    with pytest.raises(ValueError):
+        _json(raw)
+
+
+def event(stream, data):
+    stream.event(b"data: " + json.dumps(data).encode())
+
+
+def test_stream_usage_cumulative_cache_and_missing_terminal(policy):
+    stream = _Stream(policy)
+    event(
+        stream,
+        {
+            "type": "message_start",
+            "message": {
+                "model": "fixture-model",
+                "usage": {
+                    "input_tokens": 10,
+                    "output_tokens": 0,
+                    "cache_creation_input_tokens": 20,
+                    "cache_read_input_tokens": 30,
+                },
+            },
+        },
+    )
+    event(stream, {"type": "message_delta", "delta": {}, "usage": {"output_tokens": 5}})
+    event(
+        stream,
+        {
+            "type": "message_delta",
+            "delta": {"stop_reason": "end_turn"},
+            "usage": {"output_tokens": 10},
+        },
+    )
+    with pytest.raises(RequestRefused):
+        stream.cost()
+    event(stream, {"type": "message_stop"})
+    assert stream.cost() == 330
+
+
+@pytest.mark.parametrize(
+    "extra",
+    [
+        {"type": "error"},
+        {"type": "unknown"},
+        {"type": "message_delta", "usage": {"input_tokens": 1}},
+        {"type": "message_delta", "usage": {"server_tool_use": {"web_search_requests": 1}}},
+        {"type": "content_block_start", "content_block": {"type": "fallback"}},
+        {"type": "message_stop"},
+    ],
+)
+def test_stream_unknown_fees_regression_and_premature_terminal_refuse(policy, extra):
+    stream = _Stream(policy)
+    event(
+        stream,
+        {
+            "type": "message_start",
+            "message": {
+                "model": "fixture-model",
+                "usage": {"input_tokens": 10, "output_tokens": 0},
+            },
+        },
+    )
+    with pytest.raises(RequestRefused):
+        event(stream, extra)
+
+
+class DeniedMeter:
+    def __init__(self):
+        self.calls = 0
+
+    def reserve_request(self, *args):
+        self.calls += 1
+        raise RuntimeError("private credential-shaped exception must not escape")
+
+
+def test_no_execute_route_no_header_forward_and_bounded_refusal(policy):
+    meter = DeniedMeter()
+    server = MessagesServer(policy, meter, provider_key="dummy-never-used")
+    server.start()
+    try:
+        for path, extra, raw in [
+            ("/execute", {}, json.dumps(body())),
+            ("/v1/messages", {"Authorization": "Bearer dummy"}, json.dumps(body())),
+            ("/v1/messages", {}, '{"model": "x", "model": "y"}'),
+            ("/v1/messages", {}, json.dumps(body())),
+        ]:
+            conn = http.client.HTTPConnection("127.0.0.1", server.port, timeout=3)
+            conn.request("POST", path, raw, {"Content-Type": "application/json", **extra})
+            response = conn.getresponse()
+            result = response.read()
+            assert response.status in (400, 404)
+            assert b"private" not in result and b"dummy" not in result
+            conn.close()
+        assert meter.calls == 1
+    finally:
+        server.stop()
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "http://example.com",
+        "https://a:b@example.com",
+        "https://example.com/execute",
+        "https://example.com?token=x",
+    ],
+)
+def test_fixed_origin_no_redirect_proxy_or_request_url(policy, url):
+    with pytest.raises(RequestRefused):
+        MessagesServer(
+            policy, DeniedMeter(), provider_key="dummy", upstream_url=url, allow_test_http=True
+        )

workplans/LLM-WP-0009-owner-metered-messages-transport.md

@@ -0,0 +1,69 @@
+---
+id: LLM-WP-0009
+type: workplan
+title: "Owner-metered Messages transport for bounded factory execution"
+domain: agents
+repo: llm-connect
+status: active
+owner: codex
+topic_slug: llm-connect
+created: "2026-09-09"
+updated: "2026-09-09"
+related:
+  - HFACT-WP-0001
+  - REINAH-WP-0003
+  - GLAS-WP-0015
+---
+
+The factory's installed-CLI proof demonstrated native dollar-threshold overshoot.
+Implement its accepted next source slice in the transport owner, reusing rein's
+parent ledger. This workplan records implementation under the user's continued
+factory programme; it grants no operating, custody, deployment or paid authority.
+
+## Define request admission and implement the narrow transport
+
+```task
+id: LLM-WP-0009-T01
+status: done
+priority: high
+```
+
+Implemented immutable policy/upper-rate liability, explicit owner meter protocol,
+fixed-origin HTTPS Messages forwarding, strict supported features and beta
+allowlist, bounded streaming, full-charge accounting and uncertain-outcome holds.
+No retry, proxy discovery, redirect or existing /execute bypass exists on this
+listener. See `contracts/functional/messages-admission.md`.
+
+## Prove admission through the real consumer CLI with a fake provider
+
+```task
+id: LLM-WP-0009-T02
+status: done
+priority: high
+```
+
+Rein's real HTTP/SQLite tests cover exhausted capacity, concurrent requests,
+unknown prior outcomes, replay, revoked/expired leases and parent recovery.
+The installed Claude Code 2.1.266 proof refuses the USD 0.01 counterexample with
+zero upstream requests; a permitted two-request tool session creates its file.
+No actual inference, provider credential or live price/FX policy is involved.
+
+## Integrate the admitted owner route and prove production confinement
+
+```task
+id: LLM-WP-0009-T03
+status: wait
+priority: high
+blocking_reason: "Requires trusted owner hosting and actual lease/token delivery, provider custody and direct-route denial under HFACT T03/T04; accepted tariff/FX and G0 remain HFACT T01."
+```
+
+Return to HFACT-WP-0001-T01 and REINAH-WP-0003-T05/T06: integrate this transport
+inside the protected owner runtime, initialize the request extension explicitly,
+bind a run-scoped route to the real lease, inject only its base URL/token into
+the workload and revoke on lease loss. Keep the provider key and ledger outside
+the sandbox, enforce sole egress through the owner, and prove bypass denial.
+Pin accepted provider context/output and maximum tariffs with validity and FX;
+review compatibility of the exact CLI/beta combination against the actual
+provider before accepting a live profile. Local fake-provider evidence cannot
+close this task or establish a hard live EUR ceiling. Reuse GLAS-WP-0015 identity
+and native-delivery owner work; completed verifier CCRs are not reopened.