http: worker-visible response budgets + framework strict-fail enforcement

Phase B of the response-cap rework, building on the operator knobs and
conformance scaffolding from 086e3ef.

Worker-visible budgets on OutputCollector:

* ``out.remaining_response_bytes: int | None`` — wire body bytes the
  framework will accept this iteration before triggering a continuation
  token (producer) or strict-fail (unary/exchange).
* ``out.remaining_externalized_response_bytes: int | None`` — external
  channel bytes left this iteration.
* ``out.externalization_enabled: bool`` — whether the server has a
  storage backend wired up.

All snapshot semantics — fixed at collector construction, not live.
Wire bytes include IPC framing (slightly conservative for a worker
computing payload size).  Defaults to None/False so non-HTTP transports
and existing call sites are unchanged.

Framework strict-fail across all HTTP methods, with method-type-aware
hardness:

* Unary: both caps are *hard*.  Overshoot raises
  ``_RpcHttpError(500)``; ``_set_http_status`` rewrites to
  200 + ``X-VGI-RPC-Error`` carrying the EXCEPTION batch.
* Stream-exchange: both caps are *hard*.  Overshoot replaces the
  response body with a fresh IPC stream containing the EXCEPTION batch.
* Stream-producer: wire cap is *soft* — continuation tokens cover
  overshoots.  External cap stays *hard*: externalised uploads have
  no escape valve, so a producer pushing more than
  ``max_externalized_response_bytes`` writes an EXCEPTION batch into
  the in-progress IPC stream and breaks the loop.

A new ``_enforce_response_budgets`` helper in ``_responses.py``
centralises the cap check and the error message format.  Cross-language
ports must produce error messages containing the tokens
``max_response_bytes`` and ``max_externalized_response_bytes`` —
the conformance tests assert on those substrings.

Conformance tests (HTTP-only via ``transports=("http",)``):

* ``http_response_cap.unary_strict_fail``
* ``http_response_cap.exchange_strict_fail``
* ``http_response_cap.producer_external_strict_fail``
* ``http_response_cap.externalized_strict_fail``

Tests probe the active server's caps via ``http_capabilities()``
(URL stashed on ``LogCollector.http_base_url`` by the CLI runner) and
self-skip when caps aren't configured or when the wrong combination of
caps + externalisation is in effect.  ``LogCollector`` gained the
``http_base_url`` field; ``run_conformance`` gained a ``transports=``
parameter that gates incompatible tests with a clear skip reason.

A ``TestHttpResponseCap`` / ``TestHttpResponseCapSoftWire`` pytest
class pair in ``_pytest_suite.py`` exercises the same contracts via
the strict-cap fixture (``conformance_http_strict_cap_port`` from
Phase A).

Strict-cap conformance worker bumped from 64 KiB to 1 MiB so the
existing conformance suite runs cleanly against it (smaller streams
like ``cancellable_producer`` no longer trip the wire cap).

Documentation in ``docs/porting-guide.md`` and ``CLAUDE.md`` covers
the two-cap design, capability discovery via response headers, the
``transports=`` field, and the strict-fail contract for ports.

Verification:

* ruff format / ruff check / mypy / ty: clean.
* HTTP conformance against regular worker (no caps): 4 strict-fail
  tests skip cleanly; 97 existing tests pass.
* HTTP conformance against strict-cap worker (no externalisation):
  unary + exchange strict-fail PASS; externalised pair SKIP.
* HTTP conformance against strict-cap + externalisation: producer
  external + externalized PASS; unary + exchange SKIP (wire rescued).
* Pipe conformance: 4 strict-fail tests skip via transports filter
  with clear reason; existing 97 tests pass.
* Full pytest: 2986 passed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: rustyconover

CLAUDE.md

@@ -82,6 +82,16 @@ The full process before committing code is
 
 - **`http/`** *(optional — `pip install vgi-rpc[http]`)* — HTTP transport package using Falcon (server) and httpx (client). Exposes `make_wsgi_app()` to serve an `RpcServer` as a Falcon WSGI app, `serve_http()` as a convenience wrapper that combines `make_wsgi_app` + automatic free-port selection + `waitress.serve` (prints `PORT:<port>` to stdout for machine-readable discovery), and `http_connect()` for the client side. Streaming is stateless: each exchange carries serialized `StreamState` in a signed token in Arrow custom metadata. Supports pluggable authentication via an `authenticate` callback and `_AuthMiddleware`. Includes `_testing.py` with `make_sync_client()` for in-process testing without a real HTTP server.
 
+  **HTTP response caps**. Two independent operator knobs gate response size:
+  - `max_response_bytes` caps the HTTP body (what literally lands on the wire). Default `None` = unbounded. For producer streams this is *soft* — continuation tokens cover overshoot; for unary and stream-exchange it is *hard* and surfaces as `RpcError` (200 + `X-VGI-RPC-Error: true` + EXCEPTION batch).
+  - `max_externalized_response_bytes` caps total bytes uploaded to external storage during one HTTP response. Always *hard* — externalised uploads have no escape valve. Strict-fail surfaces the same way.
+
+  Externalised payloads are NOT charged against `max_response_bytes` (they leave only tiny pointer batches on the wire). The two knobs answer different operator questions: HTTP body size (proxy/gateway limit) vs per-call data volume.
+
+  Both are surfaced via response headers (`VGI-Max-Response-Bytes`, `VGI-Max-Externalized-Response-Bytes`, `VGI-Externalization-Enabled`) so `http_capabilities()` and conformance tests can probe them. Workers can read `out.remaining_response_bytes` / `out.remaining_externalized_response_bytes` / `out.externalization_enabled` on `OutputCollector` to size emits within budget.
+
+  The deprecated alias `max_stream_response_bytes` (constructor kwarg, `--max-stream-response-bytes` CLI flag, `VGI_RPC_MAX_STREAM_RESPONSE_BYTES` env) is retained for one release cycle; emits a DeprecationWarning when set. The cap is no longer stream-only — it now governs all HTTP method responses.
+
 ### Wire protocol
 
 Multiple IPC streams are written sequentially on the same pipe. Every request batch carries `vgi_rpc.request_version` in custom metadata; the server validates this before dispatch and rejects mismatches with `VersionError`. Each method call writes one request stream and reads one response stream:

docs/porting-guide.md

@@ -82,6 +82,105 @@ In order, smallest-blast-radius first:
 - **Java**: aligned for the stdio transport (120 access-log entries validate). HTTP transport does not yet wire the dispatch hook; stream_id stability across HTTP continuations is a follow-up — see `vgi-rpc-go` for the reference state-token plumbing.
 - **Rust**: aligned for the stdio transport (120 access-log entries validate). HTTP transport does not yet wire the new DispatchInfo fields (remote_addr, request_data, stable stream_id via state-token round-trip) — see `vgi-rpc-go` for reference plumbing.
 
+## HTTP response-cap conformance
+
+The conformance suite includes HTTP-only tests under category
+`http_response_cap.*` that verify the framework refuses to emit
+oversize responses when the operator has configured caps and there is
+no externalisation escape valve.
+
+### Two operator knobs
+
+- **`max_response_bytes`** — caps the HTTP body size (the bytes that
+  literally land on the wire).  Externalised payloads do not count
+  against this; their pointer batches are tiny.
+- **`max_externalized_response_bytes`** — caps the total bytes
+  uploaded to external storage during one HTTP response.  Bounds how
+  much data the client will end up fetching for one RPC, regardless
+  of how the framework chose to deliver it.
+
+Both default to `None` (unbounded) and are configurable via
+constructor kwargs, CLI flags, or env vars (see `make_wsgi_app`,
+`serve_http`, and `run_server` for the full set; the deprecated
+`max_stream_response_bytes` alias remains for one release cycle).
+
+### Capability discovery (HTTP response headers)
+
+Servers advertise the configured caps so capability-aware clients and
+conformance tests can probe without a separate handshake:
+
+| Header | Value when set |
+|---|---|
+| `VGI-Max-Response-Bytes` | integer body cap |
+| `VGI-Max-Externalized-Response-Bytes` | integer external cap |
+| `VGI-Externalization-Enabled` | `true` or `false` (always present) |
+
+Cross-language ports must emit these headers on every response when
+the corresponding knob is configured.
+
+### Strict-fail behaviour by method type
+
+| Method type | Wire cap (`max_response_bytes`) | External cap |
+|---|---|---|
+| Unary | hard — strict-fail | hard — strict-fail |
+| Stream-exchange | hard — strict-fail | hard — strict-fail |
+| Stream-producer | **soft** — continuation tokens cover overshoot | hard — strict-fail |
+
+Strict-fail surfaces as the existing 200 + EXCEPTION-batch shape
+(`_set_http_status` rewrites 500 → 200 with `X-VGI-RPC-Error: true`
+for unary; producer/exchange append a zero-row EXCEPTION batch to the
+in-progress IPC stream).  The client sees a normal `RpcError`.
+
+The error message is one of:
+
+- `HTTP body exceeds max_response_bytes (...) for method '<name>'`
+- `Externalised payload exceeds max_externalized_response_bytes (...) for method '<name>'`
+
+Cross-language ports must produce error messages containing the
+token `max_response_bytes` or `max_externalized_response_bytes`
+respectively — the conformance tests assert on those substrings.
+
+### `transports` field on conformance tests
+
+The Python catalog gained a `transports: tuple[Literal["pipe","http","unix"], ...]`
+field on `@_conformance_test`, defaulting to all three.  The CLI
+runner (`vgi-rpc-test`) detects the active transport from the user's
+flag (`--cmd` → `pipe`, `--unix` → `unix`, `--url` → `http`) and
+skips tests whose `transports` tuple excludes it.  Ports must
+honour this for the four `http_response_cap.*` tests:
+
+- `http_response_cap.unary_strict_fail` (HTTP only)
+- `http_response_cap.exchange_strict_fail` (HTTP only)
+- `http_response_cap.producer_external_strict_fail` (HTTP only;
+  also requires externalisation enabled and an external cap)
+- `http_response_cap.externalized_strict_fail` (HTTP only; same
+  preconditions)
+
+The tests self-skip when caps aren't configured, so a port can run
+the full suite against any worker without these failing.  To
+exercise them, boot a strict-cap worker — see
+`tests/serve_conformance_http_strict.py` for the Python reference
+(defaults to 1 MiB body + 1 MiB external).
+
+### Worker visibility (optional)
+
+`OutputCollector` exposes three new properties so worker code can
+size its emit to the available budget:
+
+- `out.remaining_response_bytes: int | None` — wire body bytes left
+  this iteration.
+- `out.remaining_externalized_response_bytes: int | None` — external
+  channel bytes left.
+- `out.externalization_enabled: bool` — whether the server has a
+  storage backend wired up.
+
+Snapshot semantic: each value is fixed at collector construction;
+within one `state.process()` call it does not update as the worker
+emits.  Wire bytes include IPC framing (slightly conservative for a
+worker computing payload size).  Optional surface; ports that don't
+expose it are still conformant — strict-fail catches workers that
+ignore the budget.
+
 ## Gotchas
 
 - **Arrow dictionary encoding.** Across language Arrow libraries, the placement of dictionary messages in IPC streams differs. The schema's `request_data` round-trip rule was chosen specifically to absorb this — don't try to byte-match Python.

tests/serve_conformance_http_strict.py

@@ -39,14 +39,19 @@ def main() -> None:
     parser.add_argument(
         "--max-response-bytes",
         type=int,
-        default=64 * 1024,
-        help="HTTP body cap (default: 64 KiB).",
+        # 1 MiB is large enough that incidental tests (e.g.
+        # ``cancellable_producer`` running for ~1s before being cancelled)
+        # don't trip the cap, while still being small enough that the
+        # ``http_response_cap.*`` tests' 4x target overshoots provably
+        # (4 MiB > 1 MiB).
+        default=1024 * 1024,
+        help="HTTP body cap (default: 1 MiB).",
     )
     parser.add_argument(
         "--max-externalized-response-bytes",
         type=int,
-        default=64 * 1024,
-        help="External-channel cap per HTTP response (default: 64 KiB).",
+        default=1024 * 1024,
+        help="External-channel cap per HTTP response (default: 1 MiB).",
     )
     parser.add_argument(
         "--fake-storage",

tests/test_conformance_runner.py

@@ -33,7 +33,12 @@ def test_full_suite_all_pass(self) -> None:
             suite = run_conformance(proxy, log_collector)
         assert suite.success, f"Failed tests: {[r.name for r in suite.results if not r.passed]}"
         assert suite.total > 0
-        assert suite.passed == suite.total
+        # When ``transport`` is not specified, ``run_conformance`` runs every
+        # registered test regardless of the per-test ``transports`` filter.
+        # HTTP-only tests (``http_response_cap.*``) self-skip via
+        # ``_ConformanceSkip`` because the LogCollector has no
+        # ``http_base_url``; they appear as skipped, not failed.
+        assert suite.passed + suite.skipped == suite.total
         assert suite.failed == 0
 
     def test_filter_mechanism(self) -> None:

vgi_rpc/conformance/_pytest_suite.py

@@ -1767,3 +1767,84 @@ def test_different_seed(self, conformance_conn: ConnFactory) -> None:
             with session:
                 out = session.exchange(AnnotatedBatch.from_pydict({"value": [7.0]}))
                 assert out.batch.column("value")[0].as_py() == pytest.approx(7.0)
+
+
+# ---------------------------------------------------------------------------
+# HTTP response cap (strict-fail) tests — HTTP-only
+# ---------------------------------------------------------------------------
+#
+# These tests verify the framework's strict-fail behaviour for HTTP
+# responses that overshoot the operator-configured caps.  They require a
+# server booted with ``max_response_bytes`` (and optionally
+# ``max_externalized_response_bytes``) set, so they bypass the
+# ``conformance_conn`` matrix and use the strict-cap fixture directly.
+
+
+class TestHttpResponseCap:
+    """HTTP-only strict-fail tests for response-size caps.
+
+    Mirror the catalog-based tests in ``_runner.py``'s ``http_response_cap``
+    category.  Use ``conformance_http_strict_cap_port`` (a session-scoped
+    fixture booting a worker with tight caps) so the overshoot is provably
+    triggered.
+    """
+
+    def _connect(self, port: int) -> Any:
+        """Open an HTTP connection to the strict-cap conformance worker."""
+        from vgi_rpc.http import http_connect
+
+        return http_connect(ConformanceService, f"http://127.0.0.1:{port}")
+
+    def test_unary_strict_fail(self, conformance_http_strict_cap_port: int) -> None:
+        """Unary returning more bytes than ``max_response_bytes`` allows surfaces RpcError."""
+        from vgi_rpc.http import http_capabilities
+
+        caps = http_capabilities(base_url=f"http://127.0.0.1:{conformance_http_strict_cap_port}")
+        assert caps.max_response_bytes is not None, "strict-cap fixture must advertise a wire cap"
+        with (
+            self._connect(conformance_http_strict_cap_port) as proxy,
+            pytest.raises(RpcError, match=r"max_response_bytes"),
+        ):
+            proxy.oversized_unary(target_bytes=caps.max_response_bytes * 4)
+
+    def test_exchange_strict_fail(self, conformance_http_strict_cap_port: int) -> None:
+        """Exchange returning oversize output surfaces RpcError."""
+        from vgi_rpc.http import http_capabilities
+
+        caps = http_capabilities(base_url=f"http://127.0.0.1:{conformance_http_strict_cap_port}")
+        assert caps.max_response_bytes is not None
+        target_rows = max(1024, (caps.max_response_bytes * 4) // 16)
+        with (
+            self._connect(conformance_http_strict_cap_port) as proxy,
+            pytest.raises(RpcError, match=r"max_response_bytes"),
+            proxy.exchange_oversized(rows_per_batch=target_rows) as session,
+        ):
+            session.exchange(AnnotatedBatch.from_pydict({"value": [1.0]}))
+
+
+class TestHttpResponseCapSoftWire:
+    """Producer streams have a *soft* wire cap.
+
+    Verifies the design choice that a producer emitting more than
+    ``max_response_bytes`` does **not** strict-fail when there's no
+    externalisation — continuation tokens cover the overshoot
+    transparently.  The opposite direction (strict-fail) is exercised
+    in :class:`TestHttpResponseCap` for unary and exchange.
+    """
+
+    def test_producer_overshoot_uses_continuation(self, conformance_http_strict_cap_port: int) -> None:
+        """Oversize producer emit splits across continuation tokens, not RpcError."""
+        from vgi_rpc.http import http_capabilities, http_connect
+
+        caps = http_capabilities(base_url=f"http://127.0.0.1:{conformance_http_strict_cap_port}")
+        assert caps.max_response_bytes is not None
+        # Emit ~2x the wire cap of int64+int64 rows in a single batch.
+        # A single oversized iteration overflows the cap; the framework
+        # emits the body and mints a continuation token.  No RpcError.
+        target_rows = max(1024, (caps.max_response_bytes * 2) // 16)
+        with http_connect(ConformanceService, f"http://127.0.0.1:{conformance_http_strict_cap_port}") as proxy:
+            batches = list(proxy.produce_oversized_batch(rows_per_batch=target_rows))
+            # Single emit + finish; the framework may have split it into
+            # the data batch + a continuation token + trailing finish, but
+            # the client-visible batches are: 1 data batch.
+            assert sum(b.batch.num_rows for b in batches) == target_rows