fix(server): validate Orbit binding-list targets locally

sdks/typescript/scripts/normalize_generated_sdk.py

@@ -62,14 +62,11 @@ def _normalize_sdk_metadata_version(file_path: Path) -> None:
         file_path.write_text(normalized, encoding="utf-8")
 
 
-def _strip_unstable_type_comments(file_path: Path) -> None:
+def _strip_unstable_type_comments(file_path: Path, type_name: str, description: str) -> None:
     original = file_path.read_text(encoding="utf-8")
     normalized = original.replace(
-        "/**\n"
-        " * Template-backed input payload for control create/update requests.\n"
-        " */\n"
-        "export type TemplateControlInput = {",
-        "export type TemplateControlInput = {",
+        f"/**\n * {description}\n */\nexport type {type_name} = {{",
+        f"export type {type_name} = {{",
     )
 
     if normalized != original:
@@ -82,9 +79,19 @@ def normalize_generated_sdk(_schema_path: Path, generated_dir: Path) -> None:
     if config_file.exists():
         _normalize_sdk_metadata_version(config_file)
 
-    template_control_input_file = generated_dir / "models" / "template-control-input.ts"
-    if template_control_input_file.exists():
-        _strip_unstable_type_comments(template_control_input_file)
+    # Speakeasy output varies in whether these type-level descriptions are present.
+    # Normalize only the known unstable comments; retain field documentation.
+    for filename, type_name, description in (
+        (
+            "template-control-input.ts",
+            "TemplateControlInput",
+            "Template-backed input payload for control create/update requests.",
+        ),
+        ("regex-template-parameter.ts", "RegexTemplateParameter", "RE2 regex template parameter."),
+    ):
+        model_file = generated_dir / "models" / filename
+        if model_file.exists():
+            _strip_unstable_type_comments(model_file, type_name, description)
 
     validation_error_file = generated_dir / "models" / "validation-error.ts"
     if not validation_error_file.exists():

sdks/typescript/src/generated/models/regex-template-parameter.ts

@@ -9,9 +9,6 @@ import { Result as SafeParseResult } from "../types/fp.js";
 import * as types from "../types/primitives.js";
 import { SDKValidationError } from "./errors/sdk-validation-error.js";
 
-/**
- * RE2 regex template parameter.
- */
 export type RegexTemplateParameter = {
   /**
    * Optional default regex pattern

sdks/typescript/tests/test_normalize_generated_sdk.py

@@ -5,6 +5,7 @@
 import sys
 from pathlib import Path
 
+import pytest
 
 SCRIPT_PATH = (
     Path(__file__).resolve().parent.parent / "scripts" / "normalize_generated_sdk.py"
@@ -262,3 +263,39 @@ def test_normalizer_removes_unstable_template_control_input_type_comment(
 
     normalized = template_control_input.read_text(encoding="utf-8")
     assert normalized == TEMPLATE_CONTROL_INPUT_WITHOUT_UNSTABLE_TYPE_COMMENT
+
+
+@pytest.mark.parametrize("include_type_comment", [True, False])
+def test_normalizer_stabilizes_regex_type_comment(
+    tmp_path: Path, include_type_comment: bool
+) -> None:
+    schema_path = _write_schema(tmp_path, include_context_fields=False)
+    generated_dir = tmp_path / "generated"
+    models_dir = generated_dir / "models"
+    models_dir.mkdir(parents=True)
+    model_file = models_dir / "regex-template-parameter.ts"
+    expected = (
+        "export type RegexTemplateParameter = {\n"
+        "  /** Optional default regex pattern */\n"
+        "  default?: string | undefined;\n"
+        "};\n"
+    )
+    type_comment = "/**\n * RE2 regex template parameter.\n */\n"
+    model_file.write_text(
+        (type_comment if include_type_comment else "") + expected, encoding="utf-8"
+    )
+
+    # Both generator variants converge, and a second normalization is a no-op.
+    for _ in range(2):
+        subprocess.run(
+            [
+                sys.executable,
+                str(SCRIPT_PATH),
+                "--schema",
+                str(schema_path),
+                "--generated-dir",
+                str(generated_dir),
+            ],
+            check=True,
+        )
+        assert model_file.read_text(encoding="utf-8") == expected

server/README.md

@@ -57,3 +57,21 @@ psycopg's bundled binary package.
 Server configuration is driven by environment variables (database, auth, observability, evaluators). For the full list and examples, see the docs.
 
 Full guide: https://docs.agentcontrol.dev/components/server
+
+### Binding lists with Orbit authorization
+
+When `AGENT_CONTROL_AUTH_UPSTREAM_URL` points to Orbit's
+`/internal/auth/agent_control/check_management_access` route, binding lists need
+both `target_type` and `target_id`:
+
+```http
+GET /api/v1/control-bindings?target_type=log_stream&target_id=<log-stream-uuid>&limit=100
+```
+
+Add `control_id` to filter bindings within that target. Include both target
+parameters on every pagination request. A missing, blank, or incomplete target
+returns `422 VALIDATION_ERROR` before calling the authorization service.
+
+This requirement follows Orbit's target-scoped authorization contract. Local
+authentication and generic HTTP upstreams keep their namespace-wide list behavior.
+Unexpected upstream rejections still return `502 AUTH_UPSTREAM_REJECTED`.

server/src/agent_control_server/auth_framework/config.py

@@ -252,11 +252,13 @@ def _build_default_provider() -> RequestAuthorizer:
         # Only the known Orbit management route has a predictable identity route.
         # Other upstreams keep their existing namespace-wide authorization flow.
         explicit_identity_url = (os.environ.get(_UPSTREAM_IDENTITY_URL_ENV) or "").strip()
-        identity_url = explicit_identity_url or _derive_orbit_identity_url(url)
+        orbit_identity_url = _derive_orbit_identity_url(url)
+        identity_url = explicit_identity_url or orbit_identity_url
         try:
             upstream_config = HttpUpstreamConfig(
                 url=url,
                 identity_url=identity_url,
+                binding_list_requires_target=orbit_identity_url is not None,
                 timeout_seconds=timeout,
                 service_token=token,
                 service_token_header=token_header,

server/src/agent_control_server/auth_framework/providers/http_upstream.py

@@ -63,7 +63,13 @@
     model_validator,
 )
 
-from ...errors import APIError, AuthenticationError, ForbiddenError, NotFoundError
+from ...errors import (
+    APIError,
+    APIValidationError,
+    AuthenticationError,
+    ForbiddenError,
+    NotFoundError,
+)
 from ...logging_utils import get_logger
 from ..core import Operation, Principal, RequestAuthorizer
 
@@ -289,6 +295,13 @@ class HttpUpstreamConfig:
     identity_url: str | None = None
     """Optional URL for credential and namespace resolution before target authorization."""
 
+    binding_list_requires_target: bool = False
+    """Require a complete target for binding reads when the upstream requires one.
+
+    Generic upstreams retain namespace-wide reads by default. Environment
+    configuration enables this for Orbit's management authorization route.
+    """
+
     def __post_init__(self) -> None:
         if self.keepalive_expiry_seconds < 0:
             raise ValueError("keepalive_expiry_seconds must be greater than or equal to 0")
@@ -355,6 +368,30 @@ async def authorize(
         operation: Operation,
         context: dict[str, Any] | None = None,
     ) -> Principal:
+        if (
+            self._config.binding_list_requires_target
+            and operation == Operation.CONTROL_BINDINGS_READ
+        ):
+            target_context = context or {}
+            target_type = target_context.get("target_type")
+            target_id = target_context.get("target_id")
+            has_complete_target = (
+                isinstance(target_type, str)
+                and bool(target_type.strip())
+                and isinstance(target_id, str)
+                and bool(target_id.strip())
+            )
+            if not has_complete_target:
+                raise APIValidationError(
+                    error_code=ErrorCode.VALIDATION_ERROR,
+                    detail=(
+                        "Binding lists require both target_type and target_id in this deployment."
+                    ),
+                    hint=(
+                        "Pass both target_type and target_id query parameters. "
+                        "Add control_id only as a filter within that target."
+                    ),
+                )
         headers = self._forward_headers(request)
         payload: dict[str, Any] = {"operation": operation.value}
         if context:

server/tests/test_auth_framework.py

@@ -283,6 +283,49 @@ def factory(request: httpx.Request) -> httpx.Response:
     assert captured["headers"]["x-api-key"] == "caller-key"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "operation", [Operation.AGENTS_CREATE, Operation.RUNTIME_TOKEN_EXCHANGE, Operation.RUNTIME_USE]
+)
+async def test_binding_target_requirement_preserves_other_operations(operation: Operation) -> None:
+    # Given: only binding lists require a target in this upstream deployment.
+    payloads: list[dict[str, Any]] = []
+
+    def factory(request: httpx.Request) -> httpx.Response:
+        payloads.append(json.loads(request.content))
+        return httpx.Response(200, json={"namespace_key": "org-7"})
+
+    provider = _build_upstream(factory, config_overrides={"binding_list_requires_target": True})
+
+    # When: another operation reaches the provider without binding-list context.
+    principal = await provider.authorize(_build_request(), operation)
+
+    # Then: the existing upstream flow decides that operation's authorization.
+    assert principal.namespace_key == "org-7"
+    assert payloads == [{"operation": operation.value}]
+
+
+@pytest.mark.asyncio
+async def test_complete_binding_target_preserves_unexpected_upstream_rejection() -> None:
+    # Given: a complete target still receives an unexpected upstream 400.
+    provider = _build_upstream(
+        lambda request: httpx.Response(400, json={"detail": "Unexpected contract mismatch"}),
+        config_overrides={"binding_list_requires_target": True},
+    )
+
+    # When: the provider delegates a complete binding-read request.
+    with pytest.raises(APIError) as exc_info:
+        await provider.authorize(
+            _build_request(),
+            Operation.CONTROL_BINDINGS_READ,
+            {"target_type": "log_stream", "target_id": "stream-42"},
+        )
+
+    # Then: the upstream contract failure keeps its existing 502 signal.
+    assert exc_info.value.status_code == 502
+    assert exc_info.value.error_code == "AUTH_UPSTREAM_REJECTED"
+
+
 @pytest.mark.asyncio
 async def test_http_upstream_resolves_identity_at_configured_url():
     captured: dict[str, Any] = {}
@@ -2207,6 +2250,7 @@ async def test_configure_http_upstream_selects_identity_url(
         assert isinstance(provider, HttpUpstreamAuthProvider)
         assert provider._config.identity_url == expected_identity_url
         assert provider.binding_target_authorization is (expected_identity_url is not None)
+        assert provider._config.binding_list_requires_target is (expected_identity_url is not None)
     finally:
         await auth_config.teardown_auth()
 
@@ -2236,6 +2280,9 @@ async def test_configure_http_upstream_identity_url_env(monkeypatch, upstream_ur
         assert isinstance(provider, HttpUpstreamAuthProvider)
         assert provider._config.identity_url == "https://identity.example.test/resolve"
         assert provider.binding_target_authorization is True
+        assert provider._config.binding_list_requires_target is upstream_url.endswith(
+            "/internal/auth/agent_control/check_management_access"
+        )
     finally:
         await auth_config.teardown_auth()