@@ -28,15 +28,14 @@ import { APICall, APIPromise } from "../types/async.js";
 import { Result } from "../types/fp.js";
 
 /**
- * Delete a control binding (namespace-wide)
+ * Delete a control binding
  *
  * @remarks
  * Delete a control binding by surrogate ID.
  *
- * See the GET-by-id docstring for the authorization scope: this route
- * is namespace-wide because the target identifiers are not available
- * before the binding is loaded. Use ``POST /by-key:delete`` for
- * target-scoped detach that includes the target in the request context.
+ * Target-aware authorizers use the binding's stored target identifiers.
+ * Other authorizers retain namespace-wide authorization. The deletion
+ * remains scoped to the authorized namespace.
  */
 export function controlBindingsDelete(
   client: AgentControlSDKCore,
 
@@ -28,17 +28,14 @@ import { APICall, APIPromise } from "../types/async.js";
 import { Result } from "../types/fp.js";
 
 /**
- * Get a control binding (namespace-wide)
+ * Get a control binding
  *
  * @remarks
  * Read a single control binding by surrogate ID.
  *
- * Authorization is namespace-wide: the binding's target identifiers
- * are not available until after the row is loaded.
- * Callers whose authorization model requires per-target permissions
- * should use the natural-key endpoints (``PUT /by-key``,
- * ``POST /by-key:delete``) and the target-filtered list endpoint, all
- * of which include ``(target_type, target_id)`` in the request context.
+ * Target-aware authorizers use the binding's stored target identifiers.
+ * Other authorizers retain namespace-wide authorization. The row is loaded
+ * using the authorized namespace before any binding data is returned.
  */
 export function controlBindingsGet(
   client: AgentControlSDKCore,
 
@@ -28,15 +28,14 @@ import { APICall, APIPromise } from "../types/async.js";
 import { Result } from "../types/fp.js";
 
 /**
- * Update a control binding (namespace-wide)
+ * Update a control binding
  *
  * @remarks
  * Update the ``enabled`` flag on a control binding.
  *
- * See the GET-by-id docstring for the authorization scope: this route
- * is namespace-wide because the target identifiers are not available
- * before the binding is loaded. Use ``PUT /by-key`` for target-scoped
- * upserts that include the target in the request context.
+ * Target-aware authorizers use the binding's stored target identifiers.
+ * Other authorizers retain namespace-wide authorization. The mutation
+ * remains scoped to the authorized namespace.
  */
 export function controlBindingsUpdate(
   client: AgentControlSDKCore,
 
@@ -118,15 +118,14 @@ export class ControlBindings extends ClientSDK {
   }
 
   /**
-   * Delete a control binding (namespace-wide)
+   * Delete a control binding
    *
    * @remarks
    * Delete a control binding by surrogate ID.
    *
-   * See the GET-by-id docstring for the authorization scope: this route
-   * is namespace-wide because the target identifiers are not available
-   * before the binding is loaded. Use ``POST /by-key:delete`` for
-   * target-scoped detach that includes the target in the request context.
+   * Target-aware authorizers use the binding's stored target identifiers.
+   * Other authorizers retain namespace-wide authorization. The deletion
+   * remains scoped to the authorized namespace.
    */
   async delete(
     request:
@@ -141,17 +140,14 @@ export class ControlBindings extends ClientSDK {
   }
 
   /**
-   * Get a control binding (namespace-wide)
+   * Get a control binding
    *
    * @remarks
    * Read a single control binding by surrogate ID.
    *
-   * Authorization is namespace-wide: the binding's target identifiers
-   * are not available until after the row is loaded.
-   * Callers whose authorization model requires per-target permissions
-   * should use the natural-key endpoints (``PUT /by-key``,
-   * ``POST /by-key:delete``) and the target-filtered list endpoint, all
-   * of which include ``(target_type, target_id)`` in the request context.
+   * Target-aware authorizers use the binding's stored target identifiers.
+   * Other authorizers retain namespace-wide authorization. The row is loaded
+   * using the authorized namespace before any binding data is returned.
    */
   async get(
     request:
@@ -166,15 +162,14 @@ export class ControlBindings extends ClientSDK {
   }
 
   /**
-   * Update a control binding (namespace-wide)
+   * Update a control binding
    *
    * @remarks
    * Update the ``enabled`` flag on a control binding.
    *
-   * See the GET-by-id docstring for the authorization scope: this route
-   * is namespace-wide because the target identifiers are not available
-   * before the binding is loaded. Use ``PUT /by-key`` for target-scoped
-   * upserts that include the target in the request context.
+   * Target-aware authorizers use the binding's stored target identifiers.
+   * Other authorizers retain namespace-wide authorization. The mutation
+   * remains scoped to the authorized namespace.
    */
   async update(
     request:
 
@@ -32,6 +32,7 @@
 import os
 import ssl
 from dataclasses import dataclass
+from urllib.parse import urlsplit, urlunsplit
 
 from ..config import auth_settings
 from ..logging_utils import get_logger
@@ -53,6 +54,7 @@
 # Default flow.
 _MODE_ENV = "AGENT_CONTROL_AUTH_MODE"
 _UPSTREAM_URL_ENV = "AGENT_CONTROL_AUTH_UPSTREAM_URL"
+_UPSTREAM_IDENTITY_URL_ENV = "AGENT_CONTROL_AUTH_UPSTREAM_IDENTITY_URL"
 _UPSTREAM_TIMEOUT_ENV = "AGENT_CONTROL_AUTH_UPSTREAM_TIMEOUT_SECONDS"
 _UPSTREAM_TOKEN_ENV = "AGENT_CONTROL_AUTH_UPSTREAM_SERVICE_TOKEN"
 _UPSTREAM_TOKEN_HEADER_ENV = "AGENT_CONTROL_AUTH_UPSTREAM_SERVICE_TOKEN_HEADER"
@@ -63,6 +65,8 @@
 _UPSTREAM_MAX_KEEPALIVE_CONNECTIONS_ENV = (
     "AGENT_CONTROL_AUTH_UPSTREAM_MAX_KEEPALIVE_CONNECTIONS"
 )
+_ORBIT_MANAGEMENT_PATH = "/internal/auth/agent_control/check_management_access"
+_ORBIT_IDENTITY_PATH = "/internal/auth/resolve_tenant_context"
 
 # Runtime flow.
 _RUNTIME_MODE_ENV = "AGENT_CONTROL_RUNTIME_AUTH_MODE"
@@ -110,7 +114,10 @@ def configure_auth_from_env() -> None:
       is unset, startup selects ``api_key`` only if local API-key validation is
       enabled; otherwise it selects ``none``.
     - ``AGENT_CONTROL_AUTH_MODE=http_upstream``: :class:`HttpUpstreamAuthProvider`
-      pointed at ``AGENT_CONTROL_AUTH_UPSTREAM_URL``.
+      pointed at ``AGENT_CONTROL_AUTH_UPSTREAM_URL``. By-ID binding routes use
+      stored-target authorization when an identity URL can be derived from
+      Orbit's management URL or ``AGENT_CONTROL_AUTH_UPSTREAM_IDENTITY_URL`` is
+      set. Other upstreams retain the prior namespace-wide authorization flow.
 
     Runtime flow:
 
@@ -242,9 +249,14 @@ def _build_default_provider() -> RequestAuthorizer:
             max_keepalive_connections=max_keepalive_connections,
         )
         _logger.info("Default auth provider: http_upstream url=%s", url)
+        # 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)
         try:
             upstream_config = HttpUpstreamConfig(
                 url=url,
+                identity_url=identity_url,
                 timeout_seconds=timeout,
                 service_token=token,
                 service_token_header=token_header,
@@ -272,6 +284,17 @@ def _build_default_provider() -> RequestAuthorizer:
     )
 
 
+def _derive_orbit_identity_url(url: str) -> str | None:
+    """Resolve Orbit's existing identity route from its management route."""
+    parts = urlsplit(url)
+    if not parts.path.endswith(_ORBIT_MANAGEMENT_PATH):
+        return None
+    prefix = parts.path[: -len(_ORBIT_MANAGEMENT_PATH)]
+    return urlunsplit(
+        (parts.scheme, parts.netloc, prefix + _ORBIT_IDENTITY_PATH, parts.query, parts.fragment)
+    )
+
+
 def _validate_local_api_key_mode(mode_env: str = _MODE_ENV) -> None:
     """Fail startup when local API-key mode has no local key validator."""
     if not auth_settings.api_key_enabled:
 
@@ -26,7 +26,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from enum import StrEnum
-from typing import Any, Protocol
+from typing import Any, Protocol, runtime_checkable
 
 from fastapi import Request
 
@@ -115,6 +115,22 @@ async def authorize(
     ) -> Principal: ...
 
 
+@runtime_checkable
+class IdentityResolver(Protocol):
+    """Opt-in credential and namespace lookup for target-bound authorizers.
+
+    Providers opt in only when they can resolve an identity without changing
+    the authorization behavior of existing namespace-wide binding ID routes.
+    The route then calls ``authorize`` with the stored target after the
+    namespace-scoped lookup.
+    """
+
+    @property
+    def binding_target_authorization(self) -> bool: ...
+
+    async def resolve_identity(self, request: Request, operation: Operation) -> Principal: ...
+
+
 _default_authorizer: RequestAuthorizer | None = None
 _operation_authorizers: dict[Operation, RequestAuthorizer] = {}
 
 
@@ -67,6 +67,8 @@
 _logger = get_logger(__name__)
 
 _DEFAULT_FORWARDED_HEADERS = ("X-API-Key", "Authorization", "Cookie")
+# Diagnostic label only; the identity POST has no operation payload.
+_IDENTITY_LOOKUP_LABEL = "identity.resolve"
 
 _AUTH_UPSTREAM_ATTEMPTS = Counter(
     "agent_control_server_auth_upstream_attempts_total",
@@ -180,6 +182,9 @@ class HttpUpstreamConfig:
     max_keepalive_connections: int = 20
     """Maximum idle connections retained for the auth upstream."""
 
+    identity_url: str | None = None
+    """Optional URL for credential and namespace resolution before target authorization."""
+
     def __post_init__(self) -> None:
         if self.keepalive_expiry_seconds < 0:
             raise ValueError("keepalive_expiry_seconds must be greater than or equal to 0")
@@ -213,6 +218,7 @@ def __init__(
         client: httpx.AsyncClient | None = None,
     ) -> None:
         self._config = config
+        self._identity_url = config.identity_url
         self._owns_client = client is None
         if client is not None:
             self._client = client
@@ -229,6 +235,11 @@ def __init__(
                 client_kwargs["verify"] = ssl.create_default_context(cafile=config.ca_file)
             self._client = httpx.AsyncClient(**client_kwargs)
 
+    @property
+    def binding_target_authorization(self) -> bool:
+        """Use stored-target checks only when a separate identity URL exists."""
+        return self._identity_url is not None
+
     async def aclose(self) -> None:
         """Release the HTTP client if this provider created it."""
         if self._owns_client:
@@ -245,19 +256,49 @@ async def authorize(
         if context:
             payload["context"] = context
 
-        response = await self._post_upstream(operation, payload, headers)
-        return self._handle_response(response, operation, context)
+        response = await self._post_upstream(operation.value, payload, headers)
+        return self._handle_response(response, operation.value, context)
+
+    async def resolve_identity(self, request: Request, operation: Operation) -> Principal:
+        """Authenticate with the upstream before a namespace-scoped lookup."""
+        del operation  # Target-specific permission is checked by authorize afterward.
+        if self._identity_url is None:
+            raise APIError(
+                status_code=500,
+                error_code=ErrorCode.AUTH_MISCONFIGURED,
+                reason=ErrorReason.INTERNAL_ERROR,
+                detail="Authorization identity endpoint is not configured.",
+                hint="Set AGENT_CONTROL_AUTH_UPSTREAM_IDENTITY_URL.",
+            )
+        response = await self._post_upstream(
+            _IDENTITY_LOOKUP_LABEL,
+            None,
+            self._forward_headers(request),
+            url=self._identity_url,
+        )
+        if response.status_code == 404:
+            raise APIError(
+                status_code=502,
+                error_code=ErrorCode.AUTH_UPSTREAM_REJECTED,
+                reason=ErrorReason.INTERNAL_ERROR,
+                detail="Authorization identity endpoint was not found.",
+                hint="Check the configured authorization identity URL.",
+            )
+        principal = self._handle_response(response, _IDENTITY_LOOKUP_LABEL, None)
+        return Principal(namespace_key=principal.namespace_key, caller_id=principal.caller_id)
 
     async def _post_upstream(
         self,
-        operation: Operation,
-        payload: dict[str, Any],
+        operation: str,
+        payload: dict[str, Any] | None,
         headers: dict[str, str],
+        *,
+        url: str | None = None,
     ) -> httpx.Response:
         started = perf_counter()
         try:
             response = await self._client.post(
-                self._config.url,
+                url or self._config.url,
                 json=payload,
                 headers=headers,
             )
@@ -270,7 +311,7 @@ async def _post_upstream(
             )
             _logger.warning(
                 "Auth upstream unreachable for operation %s: %s",
-                operation.value,
+                operation,
                 exc,
             )
             raise _authorization_service_unavailable_error() from exc
@@ -301,7 +342,7 @@ def _forward_headers(self, request: Request) -> dict[str, str]:
     def _handle_response(
         self,
         response: httpx.Response,
-        operation: Operation,
+        operation: str,
         context: dict[str, Any] | None,
     ) -> Principal:
         status = response.status_code
@@ -318,7 +359,7 @@ def _handle_response(
         if status == 403:
             raise ForbiddenError(
                 error_code=ErrorCode.AUTH_INSUFFICIENT_PRIVILEGES,
-                detail=f"Not authorized to perform {operation.value!r}.",
+                detail=f"Not authorized to perform {operation!r}.",
                 hint="Contact your administrator if you expected access.",
             )
         if status == 404:
@@ -343,7 +384,7 @@ def _handle_response(
                 hint = f"{hint} Retry-After: {retry_after}."
             _logger.warning(
                 "Upstream returned 429 for operation %s",
-                operation.value,
+                operation,
             )
             raise APIError(
                 status_code=503,
@@ -355,7 +396,7 @@ def _handle_response(
         if 400 <= status < 500:
             _logger.warning(
                 "Authorization upstream rejected operation %s with status %d",
-                operation.value,
+                operation,
                 status,
             )
             raise APIError(
@@ -375,7 +416,7 @@ def _handle_response(
         _logger.warning(
             "Unexpected upstream status %d for operation %s",
             status,
-            operation.value,
+            operation,
         )
         raise APIError(
             status_code=503,
@@ -420,21 +461,21 @@ def _parse_principal(self, response: httpx.Response) -> Principal:
 
 
 def _observe_upstream_attempt(
-    operation: Operation,
+    operation: str,
     duration_seconds: float,
     *,
     outcome: str,
     status_code: int | None = None,
     error: httpx.HTTPError | None = None,
 ) -> None:
     _AUTH_UPSTREAM_ATTEMPTS.labels(
-        operation=operation.value,
+        operation=operation,
         outcome=outcome,
         status_code=str(status_code) if status_code is not None else "none",
         error_type=type(error).__name__ if error is not None else "none",
     ).inc()
     _AUTH_UPSTREAM_ATTEMPT_DURATION.labels(
-        operation=operation.value,
+        operation=operation,
         outcome=outcome,
     ).observe(duration_seconds)