Fix orchestrated web search and deep research in V2 chat

application/single_app/config.py

@@ -100,7 +100,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.208"
+VERSION = "0.261.209"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 # Opt-out for deployments where App Service Easy Auth is active but the platform

application/single_app/functions_orchestration_bootstrap.py

@@ -1,22 +1,22 @@
 # functions_orchestration_bootstrap.py
 """Application-owned factories shared by web requests and scheduler continuations.
 
-Version: 0.261.140
+Version: 0.261.209
 
 Unlike the result/rendering services, this is an application composition root.
 Import it only after config has initialized the existing clients. Registering the
 artifact factory performs no I/O; each use rebuilds current actor/source access.
+External-source identity trusts the signed-in session's app roles, like classic
+chat, and makes no Microsoft Graph calls (0.261.209).
 """
 
 import hashlib
 import hmac
 from copy import deepcopy
-from urllib.parse import urlsplit
 
-import requests
+from flask import has_request_context
 
 import config
-import functions_authentication as authentication
 from agent_execution_context import capture_execution_identity
 from functions_generated_export_contracts import GeneratedFileExportRequest
 from functions_generated_export_registry import resolve_generated_file_export_format
@@ -94,7 +94,23 @@
 
 
 def build_external_identity_reader(actor_user_id, actor_conversation_id, *, execution_check=None):
-    """Create a lazy current-directory reader, never a saved-session role fallback."""
+    """Create a lazy current-access reader over the signed-in session's app roles.
+
+    Like classic chat, roles come from the signed-in session. They are captured
+    now, while the request context exists, because execution continues on a
+    worker thread. Without a matching session, as in a scheduler continuation,
+    every external-source read fails closed instead of restoring saved roles.
+    Conversation ownership and Control Center restrictions are still point-read
+    on each access. Nothing calls Microsoft Graph or persists roles.
+    """
+    session_roles, session_email = None, None
+    if has_request_context():
+        try:
+            identity = capture_execution_identity(actor_user_id, actor_conversation_id)
+        except PermissionError:
+            identity = None
+        if identity is not None:
+            session_roles, session_email = identity.roles, identity.email
     reader = None
     timeout = 10.0
 
@@ -107,7 +123,8 @@
             raise ResultUnavailableError("external_identity_access_denied") from None
 
     def read_user_settings(user_id):
-        authorize_conversation(user_id=user_id, conversation_id=actor_conversation_id)
+        if user_id != actor_user_id:
+            raise ResultUnavailableError("external_identity_access_denied")
         return _document_response(config.cosmos_user_settings_container.read_item(
             item=user_id, partition_key=user_id,
             connection_timeout=timeout, read_timeout=timeout, retry_total=0,
@@ -118,50 +135,14 @@
         if user_id != actor_user_id or conversation_id != actor_conversation_id:
             raise ResultUnavailableError("external_identity_access_denied")
         if reader is None:
-            # Optional directory authorization is initialized only when external data is accessed.
-            from functions_orchestration_external_identity import (
-                ExternalIdentityServiceError,
-                GraphExternalIdentityReader,
-            )
+            # The reader is initialized only when external data is accessed.
+            from functions_orchestration_external_identity import SessionExternalIdentityReader
 
-            graph_base = authentication.get_graph_base_url()
-            graph_origin = urlsplit(graph_base)
-            graph_scope = f"{graph_origin.scheme}://{graph_origin.netloc}/.default"
-            app_client_id = authentication.CLIENT_ID
-            application = None
-
-            def get_access_token(scope):
-                nonlocal application
-                if scope != graph_scope or authentication.CLIENT_ID != app_client_id:
-                    raise ResultUnavailableError("external_identity_access_denied")
-                if application is None:
-                    application = authentication._build_msal_app(
-                        authority_override=authentication.get_graph_authority(), timeout=timeout,
-                    )
-                result = application.acquire_token_for_client(scopes=[scope])
-                if type(result) is not dict:
-                    raise ExternalIdentityServiceError("external_identity_response_invalid")
-                token = result.get("access_token")
-                if type(token) is str and token:
-                    return token
-                error = result.get("error")
-                if error in ("server_error", "temporarily_unavailable"):
-                    raise ExternalIdentityServiceError()
-                if error == "too_many_requests":
-                    raise ExternalIdentityServiceError("external_identity_throttled")
-                if error in (
-                    "invalid_client", "invalid_grant", "unauthorized_client", "invalid_scope",
-                    "access_denied", "consent_required", "interaction_required",
-                ):
-                    raise ResultUnavailableError("external_identity_access_denied")
-                raise ExternalIdentityServiceError("external_identity_response_invalid")
-
-            reader = GraphExternalIdentityReader(
+            reader = SessionExternalIdentityReader(
                 user_id=actor_user_id, conversation_id=actor_conversation_id,
-                app_client_id=app_client_id, graph_base_url=graph_base, graph_scope=graph_scope,
-                get_access_token=get_access_token, http_get=requests.get,
+                roles=session_roles, email=session_email,
                 authorize_conversation=authorize_conversation, read_user_settings=read_user_settings,
-                execution_check=execution_check, request_timeout=timeout,
+                execution_check=execution_check,
             )
         return reader(user_id=user_id, conversation_id=conversation_id)
 

application/single_app/functions_orchestration_composition.py

@@ -1,7 +1,8 @@
 # functions_orchestration_composition.py
 """Explicit one-call content preparation from named authorized result readers.
 
-Version: 0.261.141
+Version: 0.261.209
+Refusals for a missing signed-in session reported as their own failure in: 0.261.209
 No retrieval, file-format inference, upload, publication, or implicit sibling inputs.
 
 Answer-writing steps receive saved memory, the resolved conversation references, the
@@ -42,8 +43,8 @@
 )
 from functions_orchestration_results import MAX_VALUE_BYTES, NamedOutput, ResultUnavailableError
 from functions_orchestration_schema import (
-    STEP_STATUS_COMPLETED, STEP_STATUS_FAILED, STEP_STATUS_PARTIAL, build_failure, build_step_result,
-    failure_from_exception, safe_failure, step_input_specs, validate_inline_output_schema,
+    STEP_STATUS_COMPLETED, STEP_STATUS_FAILED, STEP_STATUS_PARTIAL, access_failure, build_failure,
+    build_step_result, failure_from_exception, safe_failure, step_input_specs, validate_inline_output_schema,
 )
 from functions_orchestration_visuals import (
     build_answer_visual_guidance, build_existing_charts_note, collect_run_charts,
@@ -516,12 +517,15 @@ def prepared_values(response):
     except Exception as exc:
         raise_source_service_failure(exc)
         code = getattr(exc, 'code', '')
+        authority_reason = getattr(exc, 'authority_reason', None)
+        if type(authority_reason) is not str and isinstance(exc, ResultUnavailableError):
+            authority_reason = code
         if code == 'result_requires_streaming' or isinstance(exc, WorkflowContextBudgetError):
             failure = build_failure('result_input_too_large')
         elif isinstance(exc, OrchestrationMemoryError):
             failure = build_failure('context_unavailable')
         elif isinstance(exc, (ResultUnavailableError, PermissionError, ScreeningError)):
-            failure = build_failure('result_unavailable')
+            failure = access_failure(exc)
         elif isinstance(exc, (ResultContractError, ValueError)):
             failure = build_failure('result_invalid')
         else:
@@ -537,6 +541,7 @@ def prepared_values(response):
                 'capability_id': step['capability_id'], 'error_type': type(exc).__name__,
                 'failure_code': failure['code'],
                 **({'execution_code': code} if type(code) is str and code else {}),
+                **({'authority_reason': authority_reason} if type(authority_reason) is str and authority_reason else {}),
             },
         )
         return build_step_result(

application/single_app/functions_orchestration_executor.py

@@ -31,7 +31,8 @@
 
 A plan from the removed legacy contract is refused before anything runs.
 
-Version: 0.261.141
+Version: 0.261.209
+Refusals for a missing signed-in session reported as their own failure in: 0.261.209
 """
 
 import logging
@@ -90,6 +91,7 @@
     STEP_STATUS_WAITING,
     STEP_STATUS_PARTIAL,
     STEP_STATUS_SKIPPED,
+    access_failure,
     build_step_result,
     build_failure,
     safe_failure,
@@ -154,6 +156,14 @@ def _error_code(error):
     return code if isinstance(code, str) else None
 
 
+def _authority_reason(error):
+    """The stable reason an access refusal carries, for diagnostics only."""
+    reason = getattr(error, 'authority_reason', None)
+    if isinstance(reason, str):
+        return reason
+    return _error_code(error) if isinstance(error, ResultUnavailableError) else None
+
+
 def _partial_input_log_fields(error):
     """Which producer returned the partial result a complete-only input refused, as codes and counts."""
     if not isinstance(error, PartialInputNotAcceptedError):
@@ -893,7 +903,7 @@ def _run_dependency_step(
         if isinstance(exc, OrchestrationFilePolicyError):
             failure = build_failure('file_publication_not_allowed')
         elif isinstance(exc, (ResultUnavailableError, ElicitationContextError, PermissionError, ScreeningError)):
-            failure = build_failure('result_unavailable')
+            failure = access_failure(exc)
         elif isinstance(exc, PartialInputNotAcceptedError):
             failure = build_failure('input_partial_not_accepted')
         elif isinstance(exc, ResultContractError):
@@ -906,6 +916,7 @@ def _run_dependency_step(
             extra={
                 **_step_log_context(context, step), 'error_type': type(exc).__name__,
                 'failure_code': failure['code'], 'execution_code': _error_code(exc),
+                'authority_reason': _authority_reason(exc),
                 **_partial_input_log_fields(exc),
             },
         )
@@ -1266,15 +1277,17 @@ def save(record):
                     _raise_dependency_service_failure(step, exc)
                     if step['role'] == 'render':
                         raise
+                    failure = access_failure(exc)
                     log_event(
                         f'{_LOG_PREFIX} A saved wait could not be resumed.',
                         level=logging.WARNING,
                         extra={
                             **_step_log_context(context, step), 'error_type': type(exc).__name__,
-                            'failure_code': 'result_unavailable', 'execution_code': _error_code(exc),
+                            'failure_code': failure['code'], 'execution_code': _error_code(exc),
+                            'authority_reason': _authority_reason(exc),
                         },
                     )
-                    result = build_step_result(status=STEP_STATUS_FAILED, failure=build_failure('result_unavailable'))
+                    result = build_step_result(status=STEP_STATUS_FAILED, failure=failure)
                 task = result.get('task_result')
                 if result['status'] in (STEP_STATUS_COMPLETED, STEP_STATUS_PARTIAL, STEP_STATUS_WAITING):
                     if step['role'] == 'render':
@@ -1507,11 +1520,15 @@ def finished_stop():
             WorkflowResultIntegrityError, WorkflowResultStorageUnavailableError,
         ) as exc:
             raise_source_service_failure(exc)
+            failure = access_failure(exc)
             log_event(
                 f'{_LOG_PREFIX} Retained content could not be reauthorized for finalization.',
-                level=logging.WARNING, extra={'run_id': context.run_id, 'error_type': type(exc).__name__},
+                level=logging.WARNING, extra={
+                    'run_id': context.run_id, 'error_type': type(exc).__name__,
+                    'failure_code': failure['code'], 'authority_reason': _authority_reason(exc),
+                },
             )
-            context.failures.append(build_failure('result_unavailable'))
+            context.failures.append(failure)
             message, complete, response_reference, citations = '', False, None, []
             context.artifacts = []
     file_outputs, committed_artifacts = _dependency_file_outcomes(

application/single_app/functions_orchestration_external_configuration.py

@@ -1,7 +1,8 @@
 # functions_orchestration_external_configuration.py
 """Opaque invocation-configuration proof and capture-independent current reads.
 
-Version: 0.261.127
+Version: 0.261.209
+Research planner profile restriction removed in: 0.261.209
 
 The owning engines supply actual acquisition evidence. The application root
 supplies current metadata reads; this module discovers no settings or clients,
@@ -531,23 +532,6 @@ def _fetch_policy(settings, *, direct):
     return policy
 
 
-def is_research_acquisition_profile_supported(settings):
-    """Return whether normalized research settings avoid unattested planner calls."""
-    if type(settings) is not dict:
-        raise ResultContractError("external_configuration_invalid")
-    web_enabled = settings.get("enable_web_search", False)
-    if type(web_enabled) is not bool:
-        _invalid_metadata()
-    effective = get_source_review_config(settings)
-    return not (
-        web_enabled
-        and effective["deep_research_max_search_queries_per_turn"] > 1
-        and effective["deep_research_enable_query_planning"]
-        or effective["enable_deep_source_review"]
-        and effective["source_review_enable_llm_planning"]
-    )
-
-
 def _components(source_type, *, settings, source, selector, private_digest, require_run=False):
     source = _envelope(source)
     if source_type == "web":
@@ -901,11 +885,6 @@ def validate_acquisition(
         if type(settings) is not dict or type(current_settings) is not dict:
             raise ResultContractError("external_configuration_invalid")
         _selector(source_type, selector)
-        if source_type == "deep_research" and (
-            not is_research_acquisition_profile_supported(current_settings)
-            or not is_research_acquisition_profile_supported(settings)
-        ):
-            raise ResultUnavailableError("external_configuration_research_profile_unsupported")
         current_selector, current_parts = self._current_components(
             source_type, producer=producer, settings=current_settings, source=current_source,
         )