feat: managed knowledge Toolbox and optional user delegation

CHANGELOG.md

@@ -2,6 +2,31 @@
 
 ## [Unreleased]
 
+### Added
+
+- **Optional user delegation.** Reuse `HOSTED_AGENT_AUTH_MODE`, defaulting to
+  `user_delegated`, with an explicitly selected Hosted `service_identity`
+  profile. Both profiles require admitted UI users and preserve conversation,
+  citation and download ownership boundaries.
+- **Managed knowledge Toolbox.** Register and promote the project Toolbox
+  after Search setup, publish its endpoint through App Configuration, and
+  apply scoped project, caller and Search identity grants in both Bicep and
+  Terraform.
+
+### Changed
+
+- **Source-owned authorization.** Do not restrict service identity to Blob or
+  a source/capability allowlist. Preserve source permissions and ingestion
+  metadata. Sources requiring unavailable OBO fail explicitly without
+  authentication fallback, consent-link replay or service-privilege escalation.
+
+### Migration
+
+- Upgrade the coordinated platform and native UI/runtime combination.
+  Keep the default OBO profile unless service identity is explicitly required.
+  Change authentication modes through coordinated redeployment and new
+  sessions. Roll back the complete prior combination and deployment settings.
+
 ## [v7.0.0] - 2026-11-02
 
 ### Changed

config/aifoundry/toolbox.py

@@ -0,0 +1,92 @@
+"""Provision the managed Foundry IQ Toolbox after Search configures its KB."""
+
+import re
+from collections.abc import Mapping
+from dataclasses import dataclass
+
+from azure.ai.projects import AIProjectClient
+from azure.ai.projects.models import MCPToolboxTool
+from azure.identity import (
+    AzureCliCredential,
+    ChainedTokenCredential,
+    ManagedIdentityCredential,
+)
+
+from .appconfig import AppConfigClient
+
+SETTINGS = (
+    "RETRIEVAL_BACKEND",
+    "AI_FOUNDRY_PROJECT_ENDPOINT", "AI_FOUNDRY_PROJECT_RESOURCE_ID", "TOOLBOX_ENDPOINT",
+    "KNOWLEDGE_BASE_NAME", "KNOWLEDGE_BASE_ENDPOINT", "KNOWLEDGE_BASE_CONNECTION_ID", "FOUNDRY_IQ_API_VERSION",
+)
+
+
+@dataclass(frozen=True)
+class KnowledgeToolbox:
+    project_endpoint: str
+    name: str
+    server_url: str
+    connection_name: str
+
+    @classmethod
+    def read(cls, values: Mapping[str, str]) -> "KnowledgeToolbox | None":
+        if values.get("RETRIEVAL_BACKEND") != "foundry_iq":
+            return None
+        name = values.get("KNOWLEDGE_BASE_NAME", "")
+        project = values.get("AI_FOUNDRY_PROJECT_ENDPOINT", "").rstrip("/")
+        project_id = values.get("AI_FOUNDRY_PROJECT_RESOURCE_ID", "")
+        search = values.get("KNOWLEDGE_BASE_ENDPOINT", "").rstrip("/")
+        connection = values.get("KNOWLEDGE_BASE_CONNECTION_ID", "")
+        api_version = values.get("FOUNDRY_IQ_API_VERSION", "")
+        if not re.fullmatch(r"\d{4}-\d{2}-\d{2}(?:-preview)?", api_version):
+            raise ValueError("Managed Toolbox requires FOUNDRY_IQ_API_VERSION from provisioning.")
+        if not re.fullmatch(r"[A-Za-z0-9_-]+", name):
+            raise ValueError("Managed Toolbox requires a valid KNOWLEDGE_BASE_NAME.")
+        if not re.fullmatch(r"https://[a-z0-9-]+\.services\.ai\.azure\.com/api/projects/[A-Za-z0-9_.-]+", project):
+            raise ValueError("Managed Toolbox requires AI_FOUNDRY_PROJECT_ENDPOINT.")
+        if not re.fullmatch(r"https://[a-z0-9-]+\.search\.windows\.net", search):
+            raise ValueError("Managed Toolbox requires the provisioned KNOWLEDGE_BASE_ENDPOINT.")
+        project_match = re.fullmatch(
+            r"/subscriptions/[0-9a-fA-F-]{36}/resourceGroups/[^/]+/providers/"
+            r"Microsoft\.CognitiveServices/accounts/([^/]+)/projects/([^/]+)", project_id,
+        )
+        if (
+            project_match is None
+            or project != f"https://{project_match[1]}.services.ai.azure.com/api/projects/{project_match[2]}"
+            or not connection.startswith(project_id + "/connections/")
+            or not re.fullmatch(r"[A-Za-z0-9_.-]+", connection.rsplit("/", 1)[-1])
+        ):
+            raise ValueError("Managed Toolbox connection must belong to its provisioned Foundry project.")
+        if values.get("TOOLBOX_ENDPOINT") != f"{project}/toolboxes/{name}/mcp?api-version=v1":
+            raise ValueError("TOOLBOX_ENDPOINT must match the provisioned project and knowledge base.")
+        return cls(
+            project, name, f"{search}/knowledgebases/{name}/mcp?api-version={api_version}",
+            connection.rsplit("/", 1)[-1],
+        )
+
+    def provision(self, client: AIProjectClient) -> None:
+        version = client.toolboxes.create_version(
+            self.name, description="Foundry IQ knowledge base behind one managed endpoint",
+            tools=[MCPToolboxTool(
+                name="knowledge_base", server_label="knowledge_base",
+                server_url=self.server_url, project_connection_id=self.connection_name,
+                allowed_tools=["knowledge_base_retrieve"], require_approval="never",
+            )],
+        )
+        client.toolboxes.update(self.name, default_version=version.version)
+
+
+def main() -> None:
+    config = AppConfigClient()
+    binding = KnowledgeToolbox.read({key: config.get(key) for key in SETTINGS if config.get(key) is not None})
+    if binding is None:
+        return
+    with (
+        ChainedTokenCredential(AzureCliCredential(), ManagedIdentityCredential()) as credential,
+        AIProjectClient(endpoint=binding.project_endpoint, credential=credential) as client,
+    ):
+        binding.provision(client)
+
+
+if __name__ == "__main__":
+    main()

config/deployment/hosted_access.py

@@ -1,10 +1,10 @@
 """Plan (default) or apply the hosted instance's minimum runtime RBAC.
 
 Only configuration metadata, Foundry agent metadata and ARM are read. No secret
-values, inference requests or continuity activation are used. Configured Search
-retrieval adds an exact-service Search Index Data Reader data-plane query grant
-for transport. It is not a Blob grant or permission to omit delegated source
-authorization; no end-user Search RBAC or impersonation grant is added here.
+values, inference requests or continuity activation are used. Managed KB access
+adds Foundry User at the exact project; standalone Search uses an exact-service
+Search Index Data Reader grant. Neither grants Blob access or permission to
+omit delegated source authorization.
 """
 
 from __future__ import annotations
@@ -27,7 +27,9 @@
 from config.deployment.hosted import FOUNDRY_TOKEN_AUDIENCE
 from config.deployment.composition import HOSTED_AGENT_NAME_DEFAULT, resolve_mode
 from config.deployment.resource_groups import resource_group_for
-from config.retrieval.settings import SEARCH_BACKENDS, validate_auth_mode
+from config.retrieval.settings import (
+    SEARCH_BACKENDS, hosted_auth_mode, validate_auth_mode,
+)
 from util.azure_cli import resolve_az_command
 
 
@@ -74,6 +76,7 @@ class Role(str, Enum):
     SECRET = "4633458b-17de-408a-b874-0445c86b69e6"
     MODEL = "5e0bd9bd-7b93-4f28-af87-19fc36ad61bd"
     SEARCH = "1407120a-92aa-4202-b7e9-c0e197c71c8f"
+    TOOLBOX = "53ca6127-db72-4b80-b1b0-d745d6d5456d"
 
 
 GUID = r"[0-9a-fA-F]{8}-(?:[0-9a-fA-F]{4}-){3}[0-9a-fA-F]{12}"
@@ -84,6 +87,7 @@ class Role(str, Enum):
     Role.MODEL: rf"Microsoft\.CognitiveServices/accounts/{NAME}",
     Role.SECRET: r"Microsoft\.KeyVault/vaults/[A-Za-z0-9-]+/secrets/[A-Za-z0-9-]{1,127}",
     Role.SEARCH: rf"Microsoft\.Search/searchServices/{NAME}",
+    Role.TOOLBOX: rf"Microsoft\.CognitiveServices/accounts/{NAME}/projects/{NAME}",
 }
 PROJECT_PATH = rf"Microsoft\.CognitiveServices/accounts/({NAME})/projects/({NAME})"
 KV_REFERENCE = "application/vnd.microsoft.appconfig.keyvaultref+json"
@@ -194,10 +198,11 @@ class AccessPlan:
     skipped: bool = False
     foundry_resource_group: str = ""
     search_transport: bool = False
+    toolbox: bool = False
 
     def validate(self) -> None:
         if self.skipped:
-            if self.grants or self.principal_id or self.search_transport:
+            if self.grants or self.principal_id or self.search_transport or self.toolbox:
                 raise AccessError("Inapplicable Hosted bootstrap cannot contain grants.")
             return
         _guid(self.subscription)
@@ -208,6 +213,10 @@ def validate(self) -> None:
             raise AccessError("Invalid Search transport configuration status.")
         if self.search_transport:
             expected.add(Role.SEARCH)
+        if type(self.toolbox) is not bool:
+            raise AccessError("Invalid managed Toolbox configuration status.")
+        if self.toolbox:
+            expected.add(Role.TOOLBOX)
         if self.audit == "configured":
             expected.add(Role.SECRET)
         elif self.audit != "not-configured":
@@ -217,7 +226,7 @@ def validate(self) -> None:
         for grant in self.grants:
             group = (
                 self.foundry_resource_group or self.resource_group
-                if grant.role in {Role.MODEL, Role.SEARCH} else self.resource_group
+                if grant.role in {Role.MODEL, Role.SEARCH, Role.TOOLBOX} else self.resource_group
             )
             _bound_scope(grant.scope, RESOURCE_PATHS[grant.role], self.subscription, group)
 
@@ -452,7 +461,8 @@ def discover(environment: Mapping[str, str], *, run: RunAzure = run_az) -> Acces
         sub, group, principal, name, version, tuple(grants),
         "configured" if audit else "not-configured",
         foundry_resource_group=foundry_group,
-        search_transport=search is not None,
+        search_transport=search is not None and search.role is Role.SEARCH,
+        toolbox=search is not None and search.role is Role.TOOLBOX,
     )
     plan.validate()
     return plan
@@ -476,6 +486,17 @@ def _search_transport_grant(
     backend = _consistent(environment, ("RETRIEVAL_BACKEND",), backend_row.get("value"))
     if backend not in SEARCH_BACKENDS:
         raise AccessError("Configured retrieval backend has no supported Search transport contract.")
+    auth_row = _setting(run, config_endpoint, "HOSTED_AGENT_AUTH_MODE")
+    if auth_row is None and environment.get("HOSTED_AGENT_AUTH_MODE") == "service_identity":
+        raise AccessError("Publish HOSTED_AGENT_AUTH_MODE (agent-lz) before planning service_identity access.")
+    configured_auth = _consistent(
+        environment, ("HOSTED_AGENT_AUTH_MODE",),
+        auth_row.get("value") if auth_row else "user_delegated",
+    )
+    try:
+        hosted_auth_mode({"HOSTED_AGENT_AUTH_MODE": configured_auth})
+    except ValueError as error:
+        raise AccessError(str(error)) from None
     mode_row = _setting(run, config_endpoint, "RETRIEVAL_AUTH_MODE")
     if mode_row is not None:
         try:
@@ -502,9 +523,42 @@ def _search_transport_grant(
         or not _same(resource.get("type"), "Microsoft.Search/searchServices")
     ):
         raise AccessError("Search ARM resource identity/type mismatch.")
+    if backend == "foundry_iq":
+        return _toolbox_grant(
+            environment, run=run, config_endpoint=config_endpoint,
+            subscription=subscription, foundry_group=foundry_group,
+        )
     return Grant(Role.SEARCH, scope)
 
 
+def _toolbox_grant(
+    environment: Mapping[str, str], *, run: RunAzure, config_endpoint: str,
+    subscription: str, foundry_group: str,
+) -> Grant:
+    scope = _consistent(
+        environment, ("AZURE_AI_PROJECT_RESOURCE_ID",),
+        _config_value(run, config_endpoint, "AI_FOUNDRY_PROJECT_RESOURCE_ID"),
+    )
+    _bound_scope(scope, RESOURCE_PATHS[Role.TOOLBOX], subscription, foundry_group)
+    match = _scope_match(scope, PROJECT_PATH)
+    endpoint = _consistent(
+        environment, ("AZURE_AI_PROJECT_ENDPOINT",),
+        _config_value(run, config_endpoint, "AI_FOUNDRY_PROJECT_ENDPOINT"),
+    ).rstrip("/")
+    if endpoint != f"https://{match[3].lower()}.services.ai.azure.com/api/projects/{match[4]}":
+        raise AccessError("Managed Toolbox project endpoint/resource scope mismatch.")
+    name = _config_value(run, config_endpoint, "KNOWLEDGE_BASE_NAME")
+    if not re.fullmatch(r"[A-Za-z0-9_-]+", name):
+        raise AccessError("Managed Toolbox requires a valid knowledge base name.")
+    toolbox = _consistent(
+        environment, ("TOOLBOX_ENDPOINT",), _config_value(run, config_endpoint, "TOOLBOX_ENDPOINT"),
+    )
+    if toolbox != f"{endpoint}/toolboxes/{name}/mcp?api-version=v1":
+        raise AccessError("Managed Toolbox endpoint must match the configured project and knowledge base.")
+    # The hosted instance calls Toolbox; its RemoteTool project MI calls Search.
+    return Grant(Role.TOOLBOX, scope)
+
+
 def _has_exact(run: RunAzure, plan: AccessPlan, grant: Grant) -> bool:
     rows = run([
         "role", "assignment", "list", "--assignee-object-id", plan.principal_id,

config/deployment/topology.py

@@ -12,11 +12,14 @@
     describe_application, materialized_settings, validate_hosted_prerequisites,
 )
 from config.deployment.native import NativeApplication, OrchestrationHost, resolve_application
+from config.retrieval.settings import public_settings
 
 
 def resolve_environment_plan(environment: Mapping[str, str]) -> NativeApplication:
     """Explicit current-product intent only; no persisted-state classification."""
-    return resolve_application(environment)
+    app = resolve_application(environment)
+    public_settings(environment, hosted=app.host is OrchestrationHost.HOSTED)
+    return app
 
 
 def main() -> int:

config/requirements.txt

@@ -10,6 +10,7 @@ azure-mgmt-appcontainers==5.1.0
 # aifoundry
 
 azure-mgmt-cognitiveservices==13.6.0
+azure-ai-projects==2.7.0
 
 # search
 

config/retrieval/settings.py

@@ -1,4 +1,4 @@
-"""Public settings for the implemented native Search delegation route.
+"""Public settings for the native Search route and its structural auth profiles.
 
 An empty mode denotes no Hosted route (the Container Apps host).
 It is not a service-identity fallback. Source credentials remain request-local in
@@ -7,7 +7,7 @@
 
 from __future__ import annotations
 
-from typing import Mapping
+from typing import Any, Mapping
 from uuid import UUID
 
 from config.deployment.composition import is_truthy
@@ -25,6 +25,13 @@
 )
 
 
+def hosted_auth_mode(settings: Mapping[str, Any]) -> str:
+    value = settings.get("HOSTED_AGENT_AUTH_MODE") or "user_delegated"
+    if not isinstance(value, str) or value.strip() not in {"user_delegated", "service_identity"}:
+        raise ValueError("HOSTED_AGENT_AUTH_MODE must be user_delegated or service_identity.")
+    return value.strip()
+
+
 def source_public_settings(environment: Mapping[str, str]) -> dict[str, str]:
     """Validate/publish only explicitly supplied nonsecret source identity IDs.
 
@@ -64,19 +71,17 @@ def public_settings(
 ) -> dict[str, str]:
     """Resolve the route before discovery/publication, preserving request auth."""
     validate_auth_mode(environment)
+    mode = hosted_auth_mode(environment)
     if not hosted:
+        if mode == "service_identity":
+            raise ValueError("HOSTED_AGENT_AUTH_MODE=service_identity requires ORCHESTRATION_HOST=hosted.")
         return {RETRIEVAL_AUTH_MODE: ""}
-    if (environment.get("HOSTED_AGENT_AUTH_MODE") or "user_delegated").strip() != "user_delegated":
-        raise ValueError(
-            "Native protected retrieval requires HOSTED_AGENT_AUTH_MODE=user_delegated; "
-            "service credentials authenticate Search transport only."
-        )
     if is_truthy(environment.get("ALLOW_ANONYMOUS")):
         raise ValueError("Native protected retrieval requires ALLOW_ANONYMOUS=false.")
     backend = (environment.get("RETRIEVAL_BACKEND") or "foundry_iq").strip()
     if backend not in SEARCH_BACKENDS:
         raise ValueError("RETRIEVAL_BACKEND must be foundry_iq or ai_search for native Search retrieval.")
-    if backend == "foundry_iq":
+    if backend == "foundry_iq" and mode == "user_delegated":
         forward = environment.get("FOUNDRY_IQ_FORWARD_SOURCE_AUTH")
         if forward is not None and not is_truthy(forward):
             raise ValueError(
@@ -87,5 +92,5 @@ def public_settings(
     # never change source flags, invent app audiences or remove a KB member here.
     return {
         RETRIEVAL_AUTH_MODE: NATIVE_SEARCH,
-        "HOSTED_AGENT_AUTH_MODE": "user_delegated",
+        "HOSTED_AGENT_AUTH_MODE": mode,
     }

config/search/setup.py

@@ -1061,7 +1061,8 @@ def execute_setup(defs: Optional[dict], context: dict):
     )
 
     # Step 1: Clean up knowledge base resources in correct order (KB -> KS)
-    cleanup_knowledge_resources(defs, context, cred, search_endpoint)
+    if context.get("HOSTED_AGENT_AUTH_MODE", "user_delegated") != "service_identity":
+        cleanup_knowledge_resources(defs, context, cred, search_endpoint)
 
     # Step 2: Reconcile the remaining standard Search resources.
     provision_datasources(defs, context, cred, ds_to_indexers, search_endpoint, api_version)