feat(review): isolate OpenCode, Noema, and Strix gateway identities

contextual_orchestrator/review_gateway.py

@@ -17,14 +17,15 @@
 
 import argparse
 import os
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, replace
-from typing import Any, Mapping, Sequence
+from typing import Any
 
 from .credentials import NotConfigured, get_credential, register_credential
 from .debug_logging import LOG_LEVEL_NAMES, configure_logging
 from .model_discovery import (
-    DiscoveredModel,
     PROVIDER_MODEL_SOURCES,
+    DiscoveredModel,
     agent_from_discovered,
     discover_all_models,
     free_image_chat_serving_candidates,
@@ -329,7 +330,11 @@ def _build_parser() -> argparse.ArgumentParser:
     )
     parser.add_argument("--auth-token-key", default=REVIEW_AUTH_CREDENTIAL_NAME)
     parser.add_argument("--admin-token-key", default=REVIEW_ADMIN_CREDENTIAL_NAME)
-    parser.add_argument("--inference-token-key", default=REVIEW_INFERENCE_CREDENTIAL_NAME)
+    parser.add_argument("--inference-token-key")
+    parser.add_argument(
+        "--workload-token-key", action="append", default=[], metavar="NAME=KV_KEY",
+        help="Separate production inference credential for opencode, noema, or strix; repeat per workload.",
+    )
     parser.add_argument("--production", action="store_true")
     parser.add_argument("--allow-public-bind", action="store_true")
     parser.add_argument(
@@ -361,6 +366,23 @@ def main() -> None:
         parser.error("--allow-public-bind requires --production")
     if args.preseeded_kv and args.auth_token:
         parser.error("--preseeded-kv requires a KV auth token, not --auth-token")
+    if args.workload_token_key and not args.production:
+        parser.error("--workload-token-key requires --production")
+    if args.workload_token_key and args.inference_token_key:
+        parser.error("--workload-token-key cannot be combined with --inference-token-key")
+    if args.inference_token_key is not None and not args.inference_token_key.strip():
+        parser.error("--inference-token-key requires a KV key name")
+    workload_keys = {}
+    for specification in args.workload_token_key:
+        workload, separator, credential_key = specification.partition("=")
+        if (
+            not separator or workload not in {"opencode", "noema", "strix"}
+            or not credential_key.isascii() or not credential_key.isidentifier()
+            or workload in workload_keys or credential_key == args.admin_token_key
+            or credential_key in workload_keys.values()
+        ):
+            parser.error("--workload-token-key requires a unique opencode, noema, or strix KV key")
+        workload_keys[workload] = credential_key
     try:
         SecurityConfig().check_bind(args.host, allow_public_bind=args.allow_public_bind)
     except ValueError as exc:
@@ -375,29 +397,43 @@ def main() -> None:
             parser.error(str(exc))
     try:
         if args.production:
-            admin_token = get_credential(args.admin_token_key)
-            inference_token = get_credential(args.inference_token_key)
-            if not admin_token or not inference_token or admin_token == inference_token:
+            credential_keys = {"admin": args.admin_token_key, **(
+                workload_keys or {"review": (
+                    args.inference_token_key if args.inference_token_key is not None
+                    else REVIEW_INFERENCE_CREDENTIAL_NAME
+                )}
+            )}
+            initial_tokens = {name: get_credential(key) for name, key in credential_keys.items()}
+            values = list(initial_tokens.values())
+            if not all(isinstance(value, str) and value.strip() for value in values) or len(set(values)) != len(values):
                 parser.error("--production requires distinct admin and inference KV credentials")
 
-            def current_scope(token: str) -> str | None:
-                admin = get_credential(args.admin_token_key)
-                inference = get_credential(args.inference_token_key)
-                if not token or not admin or not inference or admin == inference:
+            def current_identity(token: str) -> str | None:
+                if not token:
+                    return None
+                values = {name: get_credential(key) for name, key in credential_keys.items()}
+                if not workload_keys and not all(values.values()):
                     return None
-                if SecurityConfig._constant_time_token_match(token, admin):
-                    return "admin"
-                if SecurityConfig._constant_time_token_match(token, inference):
-                    return "inference"
-                return None
+                matches = [
+                    name for name, value in values.items()
+                    if value and SecurityConfig._constant_time_token_match(token, value)
+                ]
+                return matches[0] if len(matches) == 1 else None
+
+            def current_scope(token: str) -> str | None:
+                identity = current_identity(token)
+                return "admin" if identity == "admin" else "inference" if identity else None
 
             security = SecurityConfig(
-                admin_token=admin_token,
-                inference_token=inference_token,
+                admin_token="" if workload_keys else initial_tokens["admin"],
+                inference_token="" if workload_keys else initial_tokens["review"],
                 bearer_verifier=lambda token, scope: current_scope(token) == scope,
-                principal_resolver=lambda token: "review-gateway" if current_scope(token) else None,
+                principal_resolver=lambda token: (
+                    f"review-workload:{identity}" if workload_keys else "review-gateway"
+                ) if (identity := current_identity(token)) else None,
                 allow_public_bind=args.allow_public_bind,
                 max_body_bytes=REVIEW_MAX_BODY_BYTES,
+                review_only=True,
             )
         else:
             auth_token = args.auth_token or get_credential(args.auth_token_key)

contextual_orchestrator/server.py

@@ -438,6 +438,7 @@ class SecurityConfig:
     # Optional companion seam for external verifiers that can expose a stable,
     # tenant-scoped principal key without exposing the bearer itself.
     principal_resolver: Callable[[str], str | None] | None = None
+    review_only: bool = False
     _rate_buckets: dict[str, tuple[int, float]] = field(default_factory=dict, init=False, repr=False)
     _rate_lock: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False)
     _run_semaphore: threading.BoundedSemaphore = field(init=False, repr=False)
@@ -5923,7 +5924,19 @@ def do_GET(self) -> None:  # noqa: N802
                     # OpenAI model discovery is inference-scope (same bearer as chat).
                     self._authorize("inference")
                     if path == "/v1/models":
-                        self._send(orchestrator.list_openai_models())
+                        models = orchestrator.list_openai_models()
+                        if security.review_only:
+                            free = next(
+                                (row for row in models["data"] if row["id"] == TaskOrchestrator.FREE_MODEL),
+                                None,
+                            )
+                            if free is None and any(
+                                not agent.disabled and orchestrator._is_free_agent(agent)
+                                for agent in orchestrator.agents
+                            ):
+                                free = {**models["data"][0], "id": TaskOrchestrator.FREE_MODEL}
+                            models["data"] = [free] if free is not None else []
+                        self._send(models)
                         return
                     raw_model_id = path[len("/v1/models/") :]
                     if not raw_model_id or "/" in raw_model_id:
@@ -5949,6 +5962,8 @@ def do_GET(self) -> None:  # noqa: N802
                     raw_refresh = (query.get("refresh") or ["false"])[0].lower()
                     if raw_refresh not in {"true", "false"}:
                         raise ValueError("refresh must be true or false")
+                    if security.review_only and raw_refresh == "true":
+                        raise RequestError(403, "route_not_allowed", "review credentials cannot refresh readiness")
                     self._send(orchestrator.inference_readiness_report(refresh=raw_refresh == "true"))
                     return
                 if path == "/v1/files" or path.startswith("/v1/files/"):
@@ -7259,6 +7274,10 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di
                     # Strip+writeback model before tools/response_format passthrough so
                     # proxy_completion pool match sees the same id as form/JS padded names.
                     model_name = _validate_chat_model(body)
+                    if security.review_only and model_name != TaskOrchestrator.FREE_MODEL:
+                        raise RequestError(400, "invalid_model", "review requests require orchestrator/free")
+                    if security.review_only and "routing" in body:
+                        raise RequestError(400, "invalid_routing", "review requests cannot select routing")
                     mode = _validate_mode(
                         next(
                             (
@@ -8526,6 +8545,14 @@ def _authorize(
                 effective_purpose = security.authorize(
                     self.headers, scope, self.client_address[0], purpose=purpose
                 )
+                if scope == "inference" and security.review_only and (
+                    self.command, urllib.parse.urlparse(self.path).path
+                ) not in {
+                    ("GET", "/v1/models"),
+                    ("GET", "/v1/readiness"),
+                    ("POST", "/v1/chat/completions"),
+                }:
+                    raise RequestError(403, "route_not_allowed", "review credentials cannot access this route")
                 if state_changing and scope == "admin":
                     security.validate_admin_session_origin(self.headers)
             except RequestError as exc:

docs/kv-credentials.md

@@ -405,6 +405,29 @@ keys from the gateway process environment. Keep the admin token on the owner;
 give CI consumers only the inference token. Public binding is explicit and
 requires a trusted TLS ingress and network controls. The CLI does not create
 that ingress, publish an immutable release, or establish hosted acceptance.
+Production review credentials can discover and request only `orchestrator/free`
+through the chat API; provider models, caller routing hints, file routes, and
+provider-refresh probes are outside this review contract.
+
+To give OpenCode, Noema, and Strix separate review identities, store a distinct
+credential for each workload in the owner KV and replace the shared inference
+token with repeated workload mappings:
+
+```bash
+python -m contextual_orchestrator.review_gateway --preseeded-kv --production \
+  --workload-token-key opencode=REVIEW_OPENCODE_TOKEN \
+  --workload-token-key noema=REVIEW_NOEMA_TOKEN \
+  --workload-token-key strix=REVIEW_STRIX_TOKEN
+```
+
+Each mapping grants inference access only; the admin token stays separate.
+Deleting or rotating one KV credential takes effect on the next request and
+does not revoke the other workloads. This mode does not accept the shared
+inference token. Consumers still need a brokered, short-lived credential
+before an LLM-controlled process can use the remote gateway safely.
+The revocation check follows [RFC 7662](https://www.rfc-editor.org/rfc/rfc7662)'s
+freshness principle; [RFC 6750](https://www.rfc-editor.org/rfc/rfc6750) explains
+why a bearer exposed to a model-controlled process remains reusable.
 
 ## Gateway direction