Base64-encode binary state token in Arrow IPC metadata for UTF-8 safety

The STATE_KEY metadata value contained raw binary data (HMAC-signed token
with serialized Arrow IPC and SHA-256 digest) that violated the Arrow IPC
requirement for UTF-8 metadata values. This broke cross-language Arrow
consumers.

- Rename STATE_KEY from vgi_rpc.stream_state to vgi_rpc.stream_state#b64
  to signal that the value is base64-encoded binary data
- Base64-encode in _pack_state_token, base64-decode in _unpack_state_token
- Add tests for UTF-8 validity and pack/unpack roundtrip
- Update wire protocol docs, README, and docstrings

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: rustyconover

CLAUDE.md

@@ -59,7 +59,7 @@ Then you can commit.
 
 - **`logging_utils.py`** — `VgiJsonFormatter`, a `logging.Formatter` subclass that serializes log records as single-line JSON. Not auto-imported; must be imported explicitly from `vgi_rpc.logging_utils`.
 
-- **`metadata.py`** — Shared helpers for `pa.KeyValueMetadata`. Centralises well-known metadata key constants (`vgi_rpc.method`, `vgi_rpc.stream_state`, `vgi_rpc.log_level`, `vgi_rpc.log_message`, `vgi_rpc.log_extra`, `vgi_rpc.server_id`, `vgi_rpc.request_version`, `vgi_rpc.location`, `vgi_rpc.shm_offset`, etc.) and provides encoding, merging, and key-stripping utilities used by `rpc/`, `http/`, `log.py`, `external.py`, `shm.py`, and `introspect.py`.
+- **`metadata.py`** — Shared helpers for `pa.KeyValueMetadata`. Centralises well-known metadata key constants (`vgi_rpc.method`, `vgi_rpc.stream_state#b64`, `vgi_rpc.log_level`, `vgi_rpc.log_message`, `vgi_rpc.log_extra`, `vgi_rpc.server_id`, `vgi_rpc.request_version`, `vgi_rpc.location`, `vgi_rpc.shm_offset`, etc.) and provides encoding, merging, and key-stripping utilities used by `rpc/`, `http/`, `log.py`, `external.py`, `shm.py`, and `introspect.py`.
 
 - **`introspect.py`** — Introspection support. Provides the built-in `__describe__` RPC method, `MethodDescription`, `ServiceDescription`, `build_describe_batch`, `parse_describe_batch`, and `introspect()`. Enabled on `RpcServer` via `enable_describe=True`.
 

README.md

@@ -1420,7 +1420,7 @@ All framework metadata keys live in the `vgi_rpc.` namespace:
 |---|---|---|
 | `vgi_rpc.method` | batch metadata | Target RPC method name |
 | `vgi_rpc.request_version` | batch metadata | Wire protocol version (`"1"`) |
-| `vgi_rpc.stream_state` | batch metadata | Serialized stream state (HTTP transport) |
+| `vgi_rpc.stream_state#b64` | batch metadata | Base64-encoded serialized stream state (HTTP transport). The `#b64` suffix indicates the value is base64-encoded binary data. |
 | `vgi_rpc.log_level` | batch metadata | Log level on zero-row log/error batches |
 | `vgi_rpc.log_message` | batch metadata | Log message text |
 | `vgi_rpc.log_extra` | batch metadata | JSON-encoded extra fields |
@@ -1480,7 +1480,7 @@ All endpoints use `Content-Type: application/vnd.apache.arrow.stream`.
 | `{prefix}/{method}/init` | POST | Stream initialization (producer and exchange) |
 | `{prefix}/{method}/exchange` | POST | Stream continuation (producer and exchange) |
 
-Over HTTP, streaming is **stateless**: each exchange carries serialized `StreamState` in a signed token in the `vgi_rpc.stream_state` batch metadata key. Producer stream init returns data batches directly; exchange stream init returns a state token.
+Over HTTP, streaming is **stateless**: each exchange carries serialized `StreamState` in a signed token in the `vgi_rpc.stream_state#b64` batch metadata key. The token is base64-encoded to ensure the metadata value is valid UTF-8. Producer stream init returns data batches directly; exchange stream init returns a state token.
 
 For streams with headers, the `/init` response body contains the header IPC stream prepended to the main output IPC stream. The `/exchange` endpoint never re-sends the header — it is only included in the initial response.
 

docs/WIRE_PROTOCOL.md

@@ -83,7 +83,7 @@ where they appear, and their semantics:
 
 | Key (bytes) | Value | Description |
 |-------------|-------|-------------|
-| `vgi_rpc.stream_state` | Opaque binary (signed token) | Serialized stream state for stateless HTTP exchanges. |
+| `vgi_rpc.stream_state#b64` | Base64-encoded binary (signed token) | Serialized stream state for stateless HTTP exchanges. The `#b64` suffix signals that the value is base64-encoded binary data. |
 
 ### Shared memory pointer batch metadata
 
@@ -277,7 +277,7 @@ receive(batch, custom_metadata):
   IF custom_metadata contains "vgi_rpc.shm_offset":
     → SHM POINTER batch → resolve via shared memory (see Section 11)
 
-  IF custom_metadata contains "vgi_rpc.stream_state":
+  IF custom_metadata contains "vgi_rpc.stream_state#b64":
     → STATE TOKEN batch → stream continuation (see Section 10)
 
   // Zero-row batch with unrecognized metadata
@@ -578,7 +578,7 @@ Response body:
 
 All produced data batches are included inline. If the response would exceed
 `max_stream_response_bytes`, the server truncates the output and appends a
-**continuation batch**: a zero-row batch with `vgi_rpc.stream_state` in its
+**continuation batch**: a zero-row batch with `vgi_rpc.stream_state#b64` in its
 custom metadata. The client then follows up with `/exchange` requests.
 
 #### Exchange stream init response
@@ -590,7 +590,7 @@ Response body:
 ```
 
 The zero-row batch carries the signed state token in
-`vgi_rpc.stream_state` custom metadata.
+`vgi_rpc.stream_state#b64` custom metadata.
 
 ### Stream exchange (HTTP)
 
@@ -601,8 +601,8 @@ Request body:  IPC stream (input_schema, 1 input batch with state token in metad
 Response body: IPC stream (output_schema, 0..N log batches, 1 data batch with updated state token, EOS)
 ```
 
-The request batch's custom metadata MUST contain `vgi_rpc.stream_state`
-with the current state token.
+The request batch's custom metadata MUST contain `vgi_rpc.stream_state#b64`
+with the current state token (base64-encoded).
 
 For **producer continuation**, the input is a zero-row batch on empty schema
 with the state token. The response may contain multiple data batches and
@@ -611,12 +611,13 @@ may end with another continuation token.
 For **exchange**, the input carries real data plus the state token. The
 response data batch carries an updated state token for the next exchange.
 
-The client MUST strip `vgi_rpc.stream_state` from the batch metadata before
+The client MUST strip `vgi_rpc.stream_state#b64` from the batch metadata before
 exposing it to application code.
 
 ### State token binary format
 
-The state token is an opaque signed blob with the following wire format (v2):
+The state token is an opaque signed blob, base64-encoded for UTF-8 safe
+metadata storage. After base64-decoding, the binary wire format (v2) is:
 
 ```
 Offset      Size     Field

tests/test_http.py

@@ -280,6 +280,7 @@ def test_tampered_token_400(self, resumable_client: _SyncTestClient) -> None:
 
     def test_wrong_token_version_400(self, resumable_client: _SyncTestClient) -> None:
         """Token with an unsupported version byte returns 400."""
+        import base64
         import hashlib
         import hmac as hmac_mod
         import struct
@@ -305,7 +306,7 @@ def test_wrong_token_version_400(self, resumable_client: _SyncTestClient) -> Non
             + input_bytes
         )
         mac = hmac_mod.new(b"test-key", payload, hashlib.sha256).digest()
-        token = payload + mac
+        token = base64.b64encode(payload + mac)
 
         req_buf = BytesIO()
         state_md = pa.KeyValueMetadata({STATE_KEY: token})
@@ -360,6 +361,7 @@ def test_union_state_uses_numeric_tag(self) -> None:
 
     def test_expired_token_400(self) -> None:
         """Token with a timestamp 2 hours in the past is rejected as expired."""
+        import base64
         import hashlib
         import hmac as hmac_mod
         import struct
@@ -386,7 +388,7 @@ def test_expired_token_400(self) -> None:
             + input_bytes
         )
         mac = hmac_mod.new(b"test-key", payload, hashlib.sha256).digest()
-        token = payload + mac
+        token = base64.b64encode(payload + mac)
 
         req_buf = BytesIO()
         state_md = pa.KeyValueMetadata({STATE_KEY: token})
@@ -411,6 +413,7 @@ def test_expired_token_400(self) -> None:
 
     def test_token_ttl_zero_disables_expiry(self) -> None:
         """Token with old timestamp is accepted when token_ttl=0."""
+        import base64
         import hashlib
         import hmac as hmac_mod
         import struct
@@ -439,7 +442,7 @@ def test_token_ttl_zero_disables_expiry(self) -> None:
             + input_bytes
         )
         mac = hmac_mod.new(b"test-key", payload, hashlib.sha256).digest()
-        token = payload + mac
+        token = base64.b64encode(payload + mac)
 
         req_buf = BytesIO()
         state_md = pa.KeyValueMetadata({STATE_KEY: token})
@@ -465,6 +468,33 @@ def test_token_ttl_zero_disables_expiry(self) -> None:
             assert "expired" not in err.error_message.lower()
         c.close()
 
+    def test_pack_token_is_valid_utf8(self) -> None:
+        """Packed state token is valid UTF-8 (base64-encoded)."""
+        from vgi_rpc.http._server import _pack_state_token
+
+        # Use bytes with high bytes that are invalid UTF-8 on their own
+        token = _pack_state_token(b"\x00\xff\xfe", b"\x80\x81", b"\x90\x91", b"key", 0)
+        # Token must be valid UTF-8 (base64 produces only ASCII)
+        token.decode("utf-8")  # Should not raise
+
+    def test_pack_unpack_roundtrip_with_base64(self) -> None:
+        """Pack and unpack produce consistent results through base64 encoding."""
+        from vgi_rpc.http._server import _pack_state_token, _unpack_state_token
+
+        state = b"test-state-data"
+        schema = b"test-schema-data"
+        input_schema = b"test-input-schema"
+        key = b"signing-key"
+
+        token = _pack_state_token(state, schema, input_schema, key, 1000)
+        # Verify it's valid UTF-8
+        token.decode("utf-8")
+
+        s, sch, inp = _unpack_state_token(token, key)
+        assert s == state
+        assert sch == schema
+        assert inp == input_schema
+
 
 # ---------------------------------------------------------------------------
 # Tests: ExternalStorage over HTTP transport

vgi_rpc/http/__init__.py

@@ -15,7 +15,7 @@
 - **Stream Exchange**: ``POST /vgi/{method}/exchange``
 
 Streaming is implemented statelessly: each exchange is a separate HTTP POST
-carrying serialized state in Arrow custom metadata (``vgi_rpc.stream_state``).
+carrying serialized state in Arrow custom metadata (``vgi_rpc.stream_state#b64``).
 When ``max_stream_response_bytes`` is set, producer-stream responses are
 split across multiple exchanges; the client transparently resumes via
 ``POST /vgi/{method}/exchange``.

vgi_rpc/http/_server.py

@@ -10,6 +10,7 @@
 
 from __future__ import annotations
 
+import base64
 import hashlib
 import hmac
 import html as _html
@@ -189,7 +190,7 @@ def _pack_state_token(
         created_at: Token creation time as seconds since epoch.
 
     Returns:
-        The opaque signed token.
+        The opaque signed token, base64-encoded for UTF-8 safe metadata.
 
     """
     payload = (
@@ -203,7 +204,7 @@ def _pack_state_token(
         + input_schema_bytes
     )
     mac = hmac.new(signing_key, payload, hashlib.sha256).digest()
-    return payload + mac
+    return base64.b64encode(payload + mac)
 
 
 def _unpack_state_token(token: bytes, signing_key: bytes, token_ttl: int = 0) -> tuple[bytes, bytes, bytes]:
@@ -222,6 +223,14 @@ def _unpack_state_token(token: bytes, signing_key: bytes, token_ttl: int = 0) ->
         _RpcHttpError: On malformed, tampered, or expired tokens (HTTP 400).
 
     """
+    try:
+        token = base64.b64decode(token, validate=True)
+    except Exception:
+        raise _RpcHttpError(
+            RuntimeError("Malformed state token"),
+            status_code=HTTPStatus.BAD_REQUEST,
+        )
+
     if len(token) < _MIN_TOKEN_LEN:
         raise _RpcHttpError(
             RuntimeError("Malformed state token"),

vgi_rpc/metadata.py

@@ -46,7 +46,7 @@
 # ---------------------------------------------------------------------------
 
 RPC_METHOD_KEY = b"vgi_rpc.method"
-STATE_KEY = b"vgi_rpc.stream_state"
+STATE_KEY = b"vgi_rpc.stream_state#b64"
 LOG_LEVEL_KEY = b"vgi_rpc.log_level"
 LOG_MESSAGE_KEY = b"vgi_rpc.log_message"
 LOG_EXTRA_KEY = b"vgi_rpc.log_extra"

vgi_rpc/rpc/__init__.py

@@ -51,7 +51,7 @@
 
 Over HTTP, streaming is stateless: each exchange is a separate
 ``POST /vgi/{method}/exchange`` carrying the input batch and serialized
-``StreamState`` in Arrow custom metadata (``vgi_rpc.stream_state``).
+``StreamState`` in Arrow custom metadata (``vgi_rpc.stream_state#b64``).
 
 State-Based Stream Model
 -------------------------