python client: Authorization bearer is sent on unary RPCs only; StreamProcessOutput, ReadFile and WriteFile go out without it

The bug lives in clients/python/src/ate_env/client.py: start with _AuthInterceptor._inject and Env._metadata. Run the hermetic four-shape script from the issue to see which call shapes lose the authorization header. Done means the bearer arrives on unary-unary, unary-stream, stream-unary and stream-stream calls, with a regression test in clients/python/tests/test_client.py.

Summary

ate_env.Client(endpoint, auth_token=…) installs _AuthInterceptor on the grpc.aio channel to add authorization: Bearer <token> to every call. The interceptor implements all four interceptor types, but the header only reaches the server on unary-unary calls. On unary-stream (StreamProcessOutput, ReadFile), stream-unary (WriteFile) and stream-stream calls the server sees no authorization metadata at all. Anything that authenticates ate-env-api (a bearer-token proxy in front of it, or a future built-in check) therefore accepts CreateEnvironment/Info/StartProcess and rejects the first shell(), read_file() or write_file() with UNAUTHENTICATED / HTTP 401.

Root cause is in how grpc.aio (grpcio 1.84.0, the version clients/python/pyproject.toml pins as the minimum) handles interceptors on streaming calls: the metadata the interceptor returns in a new ClientCallDetails is used for unary-unary calls but ignored for the three streaming shapes, which send the original client_call_details.metadata object. _AuthInterceptor._inject builds a new list and returns new details, so its header is dropped exactly on the streaming RPCs.

Impact

• Who hits it: every user of the Python client against an authenticated ate-env-api. Unauthenticated deployments do not notice (nothing checks the header).
• What breaks: env.shell() (via StreamProcessOutput), env.read_file() / read_file_bytes() (ReadFile), env.write_file() (WriteFile). create(), info(), delete(), StartProcess itself work, which makes the failure look like a server-side bug on "just the streaming endpoints".
• Severity: blocks all command execution and file transfer as soon as auth is on. P0 in that configuration.

Environment

• agent-substrate/env @ 1c08a7a (2026-10-14), clients/python (ate-env-client 0.1.0).
• grpcio 1.84.0, Python 3.12.11 (also reproduced from a Debian python:3.12-slim pod and macOS).
• Observed against ate-env-api fronted by an nginx bearer-token gate on an internal load balancer; the gate logs showed the unary calls with Authorization present and the streaming calls without it.

Steps to reproduce

Hermetic (no Substrate needed), ~30 lines — reproduces with the exact interceptor pattern from clients/python/src/ate_env/client.py:

import asyncio, collections, grpc

class _CCD(collections.namedtuple("_CCD", ["method","timeout","metadata","credentials","wait_for_ready"]),
           grpc.aio.ClientCallDetails): pass

class AuthInterceptor(grpc.aio.UnaryUnaryClientInterceptor, grpc.aio.UnaryStreamClientInterceptor,
                      grpc.aio.StreamUnaryClientInterceptor, grpc.aio.StreamStreamClientInterceptor):
    def _inj(self, d):                                  # same as _AuthInterceptor._inject
        md = list(d.metadata or []); md.append(("authorization", "Bearer TOK"))
        return _CCD(d.method, d.timeout, md, d.credentials, d.wait_for_ready)
    async def intercept_unary_unary(self, c, d, r):   return await c(self._inj(d), r)
    async def intercept_unary_stream(self, c, d, r):  return await c(self._inj(d), r)
    async def intercept_stream_unary(self, c, d, it): return await c(self._inj(d), it)
    async def intercept_stream_stream(self, c, d, it):return await c(self._inj(d), it)

def hdr(ctx): return dict(ctx.invocation_metadata()).get("authorization", "<none>").encode()
async def uu(r, c): return hdr(c)
async def us(r, c): yield hdr(c)
async def su(it, c):
    async for _ in it: pass
    return hdr(c)
async def ss(it, c):
    async for _ in it: pass
    yield hdr(c)
ident = lambda x: x

async def main():
    s = grpc.aio.server()
    s.add_generic_rpc_handlers((grpc.method_handlers_generic_handler("t", {
        "UU": grpc.unary_unary_rpc_method_handler(uu, ident, ident),
        "US": grpc.unary_stream_rpc_method_handler(us, ident, ident),
        "SU": grpc.stream_unary_rpc_method_handler(su, ident, ident),
        "SS": grpc.stream_stream_rpc_method_handler(ss, ident, ident)}),))
    p = s.add_insecure_port("127.0.0.1:0"); await s.start()
    ch = grpc.aio.insecure_channel(f"127.0.0.1:{p}", interceptors=[AuthInterceptor()])
    async def gen(): yield b""
    print("unary_unary  :", (await ch.unary_unary("/t/UU", ident, ident)(b"")).decode())
    print("unary_stream :", [x async for x in ch.unary_stream("/t/US", ident, ident)(b"")][0].decode())
    print("stream_unary :", (await ch.stream_unary("/t/SU", ident, ident)(gen())).decode())
    print("stream_stream:", [x async for x in ch.stream_stream("/t/SS", ident, ident)(gen())][0].decode())
    await ch.close(); await s.stop(0)
asyncio.run(main())

Output on grpcio 1.84.0:

unary_unary  : Bearer TOK
unary_stream : <none>
stream_unary : <none>
stream_stream: <none>

Two controls, same harness:

• Passing the header at the call site (stub.Method(req, metadata=(("authorization", …),))) reaches the server on all four shapes → per-call metadata is a reliable carrier.
• An interceptor that mutates details.metadata in place (details.metadata.add("authorization", …) on the grpc.aio.Metadata object it is given) also reaches the server on streaming calls → grpc.aio sends the original metadata object for streaming calls, not the one the interceptor returns.

Against a real deployment: put any bearer check in front of ate-env-api (we used nginx: if ($http_authorization != "Bearer <token>") { return 401; }), then Client(endpoint, auth_token=token): await client.create(...) succeeds, await env.shell("true") fails with UNAUTHENTICATED.

Expected vs actual

• Expected: auth_token is sent on every RPC the client makes.
• Actual: sent on unary-unary RPCs only; StreamProcessOutput, ReadFile, WriteFile (and any future streaming RPC) are unauthenticated.

Workaround

Carry the bearer in the per-call metadata the client already attaches (Env._metadata() adds x-env-id / x-env-atespace to every call); we run a two-hunk patch that does this plus a guard in the interceptor so the header is not added twice. PR to follow.

Suggested fix

1. Env._metadata() (and Process's use of it) appends ("authorization", f"Bearer {token}") when the client was created with auth_token — this does not depend on interceptor behaviour.
2. _AuthInterceptor._inject skips when an authorization entry is already present (so unary calls do not send it twice).
3. Regression test in clients/python/tests/test_client.py: an in-process grpc.aio server that echoes invocation_metadata()["authorization"] for all four call shapes, asserting the bearer arrives on each (this is the script above, as a pytest).

An alternative that keeps the logic in the interceptor is to mutate details.metadata in place when it is a grpc.aio.Metadata; it works on 1.84.0 but relies on grpc.aio using the original object, which is the very behaviour that differs between call shapes, so the per-call carrier seems the safer choice. The grpc.aio asymmetry itself probably deserves an upstream report at grpc/grpc; I have not found an existing one.