diff --git a/SCOPE.md b/SCOPE.md index e45500f..a3b18d3 100644 --- a/SCOPE.md +++ b/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`. --- diff --git a/contracts/functional/messages-admission.md b/contracts/functional/messages-admission.md new file mode 100644 index 0000000..8359a22 --- /dev/null +++ b/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). diff --git a/llm_connect/messages_gate.py b/llm_connect/messages_gate.py new file mode 100644 index 0000000..e371ca6 --- /dev/null +++ b/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() diff --git a/tests/test_messages_gate.py b/tests/test_messages_gate.py new file mode 100644 index 0000000..66fa167 --- /dev/null +++ b/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 + ) diff --git a/workplans/LLM-WP-0009-owner-metered-messages-transport.md b/workplans/LLM-WP-0009-owner-metered-messages-transport.md new file mode 100644 index 0000000..c4af720 --- /dev/null +++ b/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.