Query-farm
diff --git a/‎CLAUDE.md‎
Lines changed: 95 additions & 0 deletions b/‎CLAUDE.md‎
Lines changed: 95 additions & 0 deletions
diff --git a/‎pyproject.toml‎
Lines changed: 2 additions & 2 deletions b/‎pyproject.toml‎
Lines changed: 2 additions & 2 deletions
diff --git a/‎tests/test_serve.py‎
Lines changed: 197 additions & 0 deletions b/‎tests/test_serve.py‎
Lines changed: 197 additions & 0 deletions
@@ -297,6 +297,11 @@ vgi-client --input data.parquet --function sum_all_columns --worker vgi-fixture-
 | `VGI_OAUTH_DEVICE_CODE_CLIENT_ID` | Client ID for device-code flow (optional, URL-safe chars only) |
 | `VGI_OAUTH_DEVICE_CODE_CLIENT_SECRET` | Client secret for device-code flow (optional, URL-safe chars only) |
 | `VGI_OAUTH_USE_ID_TOKEN` | When `1`/`true`/`yes`, clients use OIDC `id_token` as Bearer instead of `access_token` |
+| `VGI_INTROSPECT_PRINCIPALS` | Comma-separated principals permitted to call `__introspect_token__`. Required — no permissive default — whenever the worker implements `resolve_token()` (see below) |
+| `VGI_INTROSPECT_RATE_LIMIT` | Introspection requests allowed per caller per second (default 20) |
+| `VGI_WORKER_ACCESS_LOG_SAMPLE` | Fraction of *successful* calls to keep in the access log, `0.0`–`1.0`. Errors are always kept; the decision is per call, so every record of one stream shares a fate |
+| `VGI_WORKER_ACCESS_LOG_ASYNC` | When `1`/`true`/`yes`, emit access-log records from a listener thread. Bounded queue; full means drop, and a crash loses whatever is queued |
+| `VGI_WORKER_ACCESS_LOG_QUEUE_SIZE` | Bound on the async access-log queue (default 10000) |
 | `VGI_OTEL_ENABLED` | Enable OpenTelemetry instrumentation (`1`/`true`/`yes`) |
 | `VGI_OTEL_CUSTOM_ATTRIBUTES` | Comma-separated `key=value` pairs for custom span/metric attributes |
 | `VGI_OTEL_CLAIM_ATTRIBUTES` | Comma-separated `claim_key=span_attr_name` pairs for claim extraction |
@@ -392,6 +397,96 @@ Set `SENTRY_DSN` (and install `vgi[sentry]`) to forward unhandled exceptions to
 
 The same enrichment applies to OTel spans when `VGI_OTEL_ENABLED=1` — both backends read from the same `VgiTracer.set_current_span_attributes()` call sites in `vgi/otel.py`. Either, neither, or both can be active in a process.
 
+### Serving Behind a Load Balancer
+
+`--max-externalized-response-bytes` caps a single *externalized* response — the
+payload uploaded to blob storage and replaced on the wire by a pointer. Set it
+to whatever the load balancer, API gateway, or object-store policy in front of
+the worker will actually carry.
+
+```bash
+vgi-serve my.worker:MyWorker --http --max-externalized-response-bytes 67108864
+```
+
+Unlike `--max-stream-response-bytes`, which governs the wire and is *soft* for
+producer streams (a continuation token carries the overshoot to the next turn),
+this cap is **hard on every method type with no continuation escape** — bytes
+already uploaded cannot be un-uploaded. Default is no cap. When set, the value
+is advertised as `VGI-Max-Externalized-Response-Bytes` so clients can size their
+own expectations.
+
+Also available on `Worker.main --http` and the fixture server. Under
+`--server granian` it travels to worker processes through `export_serve_config`.
+
+### Token Introspection (`resolve_token`)
+
+A worker may optionally expose `POST {prefix}/__introspect_token__`, which
+resolves an opaque bearer credential to a principal — for a reverse proxy that
+terminates the only public listener and must know the caller's identity before
+it can authorize anything.
+
+```python
+from vgi.auth import AuthUnavailableError, TokenIdentity
+from vgi.worker import Worker
+
+class MyWorker(Worker):
+    functions = [...]
+
+    @classmethod
+    def resolve_token(cls, token: str) -> TokenIdentity | None:
+        try:
+            row = api_keys.lookup(token)          # your own store
+        except ConnectionError as exc:
+            # "I could not find out" — 503 + Retry-After, not 401.
+            raise AuthUnavailableError(str(exc)) from exc
+        if row is None:
+            return None                           # "the credential is unknown"
+        return TokenIdentity(principal=row.principal, token_name=row.label)
+```
+
+**The route does not exist until `resolve_token` is overridden** — absent, not
+routed-and-refusing. That is what keeps a dependency upgrade from growing a
+credential-to-identity oracle on every existing worker.
+
+Enabling it also requires an allowlist of principals permitted to ask, via
+`--introspect-principals` or `VGI_INTROSPECT_PRINCIPALS`. There is no permissive
+default and a worker that overrides the hook without one **refuses to start**:
+authenticating and introspecting are different capabilities, and "any
+authenticated caller" lets any user resolve any other user's credential to its
+owner. `--introspect-rate-limit` (default 20/caller/second) bounds, rather than
+closes, the oracle an allowlisted-but-compromised caller still has.
+
+Return `None` for "the store answered and this credential is unknown"; raise
+`AuthUnavailableError` for "the answer is not knowable". A caller that
+negative-caches the first must not cache the second. Never return claims — a
+pass-through claims field would let a worker choose its caller's tenant routing
+and policy branch.
+
+`AuthUnavailableError` is worth reaching for in any custom `authenticate`
+callback too, not just here: it is deliberately **not** a `ValueError`, because
+`chain_authenticate` advances to the next authenticator on `ValueError` — so a
+sidecar outage raised as one is read as "not my credential, try the next" and
+ends up a 401 from the end of the chain, restarting every session in the fleet
+over a thirty-second blip.
+
+### Access Log Sampling and Async Emission
+
+`--access-log-sample 0.05` keeps 5% of *successful* calls. Errors are always
+kept, and the decision is made per call so every record belonging to one stream
+shares a fate. `--access-log-async` moves formatting and writing to a listener
+thread so disk latency stays off the request path; the queue is bounded
+(`--access-log-queue-size`, default 10000), full means drop, and the next record
+through carries `dropped_records` so a gap is never silent. A crash loses
+whatever is still queued — that is the trade.
+
+Both apply only to `vgi_rpc.access`, which gets its own handler for the purpose.
+They are deliberately not applied to the diagnostic loggers: dropping half of a
+traceback is not a saving.
+
+**Trace correlation needs no configuration.** vgi-rpc reads whatever span is
+current when it emits a record, so `trace_id` / `span_id` appear on every access
+record as soon as `VGI_OTEL_ENABLED=1`.
+
 ### Key Constraints for Scalar Functions:
 - **1:1 row mapping**: Output must have exactly the same number of rows as input
 - **Single column output**: Output schema has exactly one column named "result"
 
@@ -1,6 +1,6 @@
 [project]
 name = "vgi-python"
-version = "0.24.0"
+version = "0.25.0"
 description = "Vector Gateway Interface - Connect DuckDB to external programs via Apache Arrow"
 readme = "README.md"
 keywords = [
@@ -40,7 +40,7 @@ dependencies = [
     "pyarrow",
     "typer>=0.9",
     "platformdirs",
-    "vgi-rpc>=0.33.0",
+    "vgi-rpc>=0.39.1",
     "httpx>=0.24",
 ]
 
 
@@ -270,6 +270,203 @@ def test_signing_key_passed(self) -> None:
         assert isinstance(app, falcon.App)
 
 
+def _capability_headers(app: object) -> dict[str, str]:
+    """Read the capability headers a served app advertises on ``/health``.
+
+    Args:
+        app: The Falcon app under test.
+
+    Returns:
+        Response headers, lowercased.
+
+    """
+    import falcon.testing
+
+    resp = falcon.testing.TestClient(app).simulate_get("/vgi/health")
+    assert resp.status_code == 200
+    return {k.lower(): v for k, v in resp.headers.items()}
+
+
+class TestMaxExternalizedResponseBytes:
+    """The externalized-response cap, for workers behind a size-limited proxy."""
+
+    def test_advertised_when_set(self) -> None:
+        """The cap reaches the wire, so a client can size its own expectations."""
+        app = create_app(_SingleWorker, prefix="/vgi", describe=False, max_externalized_response_bytes=65536)
+        assert _capability_headers(app)["vgi-max-externalized-response-bytes"] == "65536"
+
+    def test_absent_by_default(self) -> None:
+        """No cap unless one is configured.
+
+        The header's absence is what tells a client there is no ceiling; a
+        default value here would be a ceiling nobody chose.
+        """
+        app = create_app(_SingleWorker, prefix="/vgi", describe=False)
+        assert "vgi-max-externalized-response-bytes" not in _capability_headers(app)
+
+
+class _IntrospectingWorker(Worker):
+    """Worker that implements the optional token-introspection hook."""
+
+    functions = [_DoubleFunc]
+
+    @classmethod
+    def resolve_token(cls, token: str) -> object | None:
+        from vgi.auth import TokenIdentity
+
+        if token == "good-token":
+            return TokenIdentity(principal="alice", token_name="deploy-key-1")
+        return None
+
+
+class TestIntrospectResolverDetection:
+    """``resolve_token`` presence is detected by override, not by a flag."""
+
+    def test_base_worker_has_no_resolver(self) -> None:
+        """A worker that never wrote the lookup gets no endpoint."""
+        assert _SingleWorker._introspect_resolver() is None
+
+    def test_override_is_detected(self) -> None:
+        """Implementing the hook is what enables it."""
+        resolver = _IntrospectingWorker._introspect_resolver()
+        assert resolver is not None
+        assert resolver("good-token") is not None
+        assert resolver("nope") is None
+
+
+class TestTokenIntrospection:
+    """``POST /__introspect_token__`` — absent unless a worker implements it."""
+
+    @staticmethod
+    def _client(worker_cls: type[Worker], **env: str) -> object:
+        import falcon.testing
+
+        from vgi.serve import _resolve_authenticate
+
+        with pytest.MonkeyPatch.context() as mp:
+            for key in ("VGI_INTROSPECT_PRINCIPALS", "VGI_INTROSPECT_RATE_LIMIT", "VGI_BEARER_TOKENS"):
+                mp.delenv(key, raising=False)
+            for key, value in env.items():
+                mp.setenv(key, value)
+            app = create_app(
+                worker_cls,
+                prefix="/vgi",
+                describe=False,
+                authenticate=_resolve_authenticate(),
+            )
+        return falcon.testing.TestClient(app)
+
+    def test_route_absent_without_hook(self) -> None:
+        """Not "routed and refusing" — absent.
+
+        This is the property that keeps a dependency upgrade from silently
+        growing a credential-to-identity oracle on every existing worker.
+        """
+        client = self._client(_SingleWorker)
+        resp = client.simulate_post("/vgi/__introspect_token__", json={"token": "x"})  # type: ignore[attr-defined]
+        assert resp.status_code == 404
+
+    def test_capability_not_advertised_without_hook(self) -> None:
+        """A proxy preflighting at boot learns this worker cannot answer."""
+        app = create_app(_SingleWorker, prefix="/vgi", describe=False)
+        assert "vgi-token-introspection" not in _capability_headers(app)
+
+    def test_capability_advertised_with_hook(self) -> None:
+        """A proxy can discover support at boot rather than at first login."""
+        with pytest.MonkeyPatch.context() as mp:
+            mp.setenv("VGI_INTROSPECT_PRINCIPALS", "proxy-a")
+            app = create_app(_IntrospectingWorker, prefix="/vgi", describe=False)
+        assert _capability_headers(app)["vgi-token-introspection"] == "true"
+
+    def test_allowlisted_caller_resolves(self) -> None:
+        """The endpoint answers with a principal and never with claims."""
+        client = self._client(
+            _IntrospectingWorker,
+            VGI_INTROSPECT_PRINCIPALS="proxy-a",
+            VGI_BEARER_TOKENS="ptok=proxy-a",
+        )
+        resp = client.simulate_post(  # type: ignore[attr-defined]
+            "/vgi/__introspect_token__",
+            json={"token": "good-token"},
+            headers={"Authorization": "Bearer ptok"},
+        )
+        assert resp.status_code == 200
+        assert resp.json == {"principal": "alice", "token_name": "deploy-key-1", "ttl_seconds": 300}
+        assert "claims" not in resp.json
+
+    def test_unresolvable_token_is_404(self) -> None:
+        """Unknown credential and unknown route answer alike, on purpose."""
+        client = self._client(
+            _IntrospectingWorker,
+            VGI_INTROSPECT_PRINCIPALS="proxy-a",
+            VGI_BEARER_TOKENS="ptok=proxy-a",
+        )
+        resp = client.simulate_post(  # type: ignore[attr-defined]
+            "/vgi/__introspect_token__",
+            json={"token": "nope"},
+            headers={"Authorization": "Bearer ptok"},
+        )
+        assert resp.status_code == 404
+
+    def test_non_allowlisted_caller_refused(self) -> None:
+        """Authenticating is not the same capability as introspecting.
+
+        Without this, any user holding a valid credential could resolve any
+        other user's credential to its owner.
+        """
+        client = self._client(
+            _IntrospectingWorker,
+            VGI_INTROSPECT_PRINCIPALS="proxy-a",
+            VGI_BEARER_TOKENS="ptok=proxy-a,utok=user-b",
+        )
+        resp = client.simulate_post(  # type: ignore[attr-defined]
+            "/vgi/__introspect_token__",
+            json={"token": "good-token"},
+            headers={"Authorization": "Bearer utok"},
+        )
+        assert resp.status_code == 403
+
+    def test_hook_without_allowlist_refuses_to_start(self) -> None:
+        """Fail closed and loud, rather than defaulting to an open oracle."""
+        with pytest.MonkeyPatch.context() as mp:
+            mp.delenv("VGI_INTROSPECT_PRINCIPALS", raising=False)
+            with pytest.raises(SystemExit) as exc:
+                create_app(_IntrospectingWorker, prefix="/vgi", describe=False)
+        assert exc.value.code == 1
+
+    @pytest.mark.parametrize("bad", ["nonsense", "0", "-5"])
+    def test_bad_rate_limit_refuses_to_start(self, bad: str) -> None:
+        """A typo must not silently become "refuse everything" or "no bound"."""
+        with pytest.MonkeyPatch.context() as mp:
+            mp.setenv("VGI_INTROSPECT_PRINCIPALS", "proxy-a")
+            mp.setenv("VGI_INTROSPECT_RATE_LIMIT", bad)
+            with pytest.raises(SystemExit):
+                create_app(_IntrospectingWorker, prefix="/vgi", describe=False)
+
+
+class TestAuthUnavailableReExport:
+    """``AuthUnavailableError`` is reachable without naming a vgi_rpc module."""
+
+    def test_importable_from_vgi_auth(self) -> None:
+        """Worker authors reach it from ``vgi.auth`` like every other auth type."""
+        from vgi_rpc.http import AuthUnavailableError as Upstream
+
+        from vgi.auth import AuthUnavailableError
+
+        assert AuthUnavailableError is Upstream
+
+    def test_is_not_a_value_error(self) -> None:
+        """The whole point: ``chain_authenticate`` must not swallow it.
+
+        The chain advances to the next authenticator on ``ValueError``, so an
+        outage raised as one is read as "not my credential, try the next" and
+        ends up a 401 — restarting every session in the fleet over a blip.
+        """
+        from vgi.auth import AuthUnavailableError
+
+        assert not issubclass(AuthUnavailableError, ValueError)
+
+
 # ---------------------------------------------------------------------------
 # Tests: env var helpers
 # ---------------------------------------------------------------------------