diff --git a/CHANGELOG.md b/CHANGELOG.md index 4b5691f..6b3f8e3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,17 @@ ## [Unreleased] +### Added + +- **Optional-ingestion upload capability.** Honor + `DEPLOY_INGESTION_CONTAINER_APP` independently of the chat backend: `false` + hides standalone and Entra Copilot upload controls and rejects attachments + before ingestion URL or Dapr resolution. Missing preserves legacy upload + eligibility; `true` retains existing authentication and conversation checks; + malformed boolean values fail with a configuration error. Anonymous Copilot + uploads remain disabled. Text chat and retrieval are unchanged; no native + Foundry IQ conversation upload path is introduced. + ## [v3.2.2] - 2026-10-14 ### Fixed diff --git a/README.md b/README.md index dcffd75..2458cad 100644 --- a/README.md +++ b/README.md @@ -165,6 +165,16 @@ rejects first-turn uploads rather than fabricating an ID. A resumed Chainlit thread restores context from its ordered messages and obtains a fresh managed ID from the hosted runtime. +Uploads also require the independent `DEPLOY_INGESTION_CONTAINER_APP` capability +from the existing App Configuration provider. `false` hides upload controls in +standalone and Entra Copilot modes and rejects attachments before ingestion URL +or Dapr resolution with “Attachments are unavailable in this deployment.” +Missing preserves legacy eligibility; `true` retains existing authentication and +conversation checks. Invalid boolean values fail with a configuration error. +Text chat and retrieval are unchanged; this does not add native Foundry IQ +conversation uploads. Supported ingestion sources and deployment combinations +remain owned by the [platform documentation](https://azure.github.io/AI-Landing-Zones/agent-landing-zone/). + Live acceptance still requires Basic and isolated topology validation, two-turn managed conversation continuity, and a two-user negative document-authorization test. These deployment and identity checks cannot be diff --git a/docs/copilot-embedding.md b/docs/copilot-embedding.md index 9c340d9..d957494 100644 --- a/docs/copilot-embedding.md +++ b/docs/copilot-embedding.md @@ -13,6 +13,16 @@ Embedding is disabled by default. Enabling it without a valid mode from missing Entra configuration and never downgrades an Entra failure to anonymous access. +Entra Copilot upload controls also require the independent +`DEPLOY_INGESTION_CONTAINER_APP` capability read through the existing App +Configuration provider. A missing setting preserves legacy upload eligibility; +`true` keeps the existing Entra and conversation checks. `false` disables the +controls and server-side attachment processing, including ingestion URL/Dapr +fallbacks, with “Attachments are unavailable in this deployment.” Malformed +boolean values fail with an actionable configuration error. Anonymous Copilot +uploads remain disabled regardless of this flag. Text chat, authentication and +retrieval are unchanged; no native conversation upload path is introduced. + ## Rendering and token flow Chainlit 2.12.0 Copilot mode renders a floating widget in an open Shadow DOM. It diff --git a/src/gpt_rag_ui/auth/embed_security.py b/src/gpt_rag_ui/auth/embed_security.py index 26ee738..8edb36f 100644 --- a/src/gpt_rag_ui/auth/embed_security.py +++ b/src/gpt_rag_ui/auth/embed_security.py @@ -21,6 +21,8 @@ reset_request_copilot_session, ) from gpt_rag_ui.config.embed_config import EmbedSettings +from gpt_rag_ui.config.appconfig import load_ingestion_uploads_enabled +from gpt_rag_ui.config.dependencies import get_config from gpt_rag_ui.telemetry.boundary_diagnostics import log_boundary_failure @@ -654,6 +656,7 @@ def _inject_chainlit_cookie( def _copilot_project_settings_send(send, auth_mode: str): """Expose only the capabilities supported by this Copilot auth mode.""" + ingestion_uploads_enabled = load_ingestion_uploads_enabled(get_config()) pending_start = None async def send_response(message): @@ -678,7 +681,7 @@ async def send_response(message): .get("spontaneous_file_upload", {}) ) if isinstance(upload, dict): - upload["enabled"] = auth_mode == "entra" + upload["enabled"] = ingestion_uploads_enabled and auth_mode == "entra" body = json.dumps( payload, separators=(",", ":"), diff --git a/src/gpt_rag_ui/bootstrap.py b/src/gpt_rag_ui/bootstrap.py index 12b13c8..0ad9da0 100644 --- a/src/gpt_rag_ui/bootstrap.py +++ b/src/gpt_rag_ui/bootstrap.py @@ -12,7 +12,7 @@ from fastapi import FastAPI, Response from fastapi.responses import JSONResponse, StreamingResponse -from gpt_rag_ui.config.appconfig import AppConfigClient +from gpt_rag_ui.config.appconfig import AppConfigClient, load_ingestion_uploads_enabled from gpt_rag_ui.telemetry.boundary_diagnostics import log_boundary_failure from gpt_rag_ui.clients.blob import BlobClient from gpt_rag_ui.config.dependencies import get_config @@ -154,21 +154,27 @@ def _format_release_value(value: str | None, missing_message: str) -> str: return missing_message -def _want_chainlit_spontaneous_file_upload(auth_state: AuthState) -> bool: - """True whenever the effective allow_anonymous is false. +def _want_chainlit_spontaneous_file_upload( + auth_state: AuthState, *, ingestion_uploads_enabled: bool = True, +) -> bool: + """Enable uploads only with ingestion capability and authenticated callers. Per-conversation uploads require an authenticated caller (each file is bound to the - user's conversation), so we enable the Chainlit paperclip whenever auth is in force, + user's conversation), so the Chainlit paperclip also requires auth in force, regardless of whether ALLOW_ANONYMOUS came from env, App Config, or the default. """ - return not auth_state.allow_anonymous + return ingestion_uploads_enabled and not auth_state.allow_anonymous -def _sync_chainlit_spontaneous_file_upload(auth_state: AuthState) -> None: +def _sync_chainlit_spontaneous_file_upload( + auth_state: AuthState, *, ingestion_uploads_enabled: bool = True, +) -> None: """Write [features.spontaneous_file_upload].enabled before `import chainlit` (Chainlit reads TOML on import).""" - want_enabled = _want_chainlit_spontaneous_file_upload(auth_state) + want_enabled = _want_chainlit_spontaneous_file_upload( + auth_state, ingestion_uploads_enabled=ingestion_uploads_enabled, + ) path = get_asset_root() / ".chainlit" / "config.toml" if not path.is_file(): logger.warning("Chainlit config not found at %s; skipping spontaneous file upload sync", path) @@ -728,10 +734,13 @@ def _create_chainlit_app( """ embed_settings = embed_settings or EmbedSettings() + ingestion_uploads_enabled = load_ingestion_uploads_enabled(config) _configure_embed_environment(embed_settings) _configure_auth_environment(config, auth_state) effective_auth = auth_state or _evaluate_auth_state(config, embed_settings) - _sync_chainlit_spontaneous_file_upload(effective_auth) + _sync_chainlit_spontaneous_file_upload( + effective_auth, ingestion_uploads_enabled=ingestion_uploads_enabled, + ) download_tokens = None if embed_settings.enabled: @@ -757,6 +766,14 @@ def _create_chainlit_app( # origins can be applied in memory without modifying the deployment files. from chainlit.config import config as chainlit_config + # Also enforce the capability if TOML could not be written or Chainlit + # was imported earlier in this process. + if chainlit_config.features.spontaneous_file_upload is not None: + chainlit_config.features.spontaneous_file_upload.enabled = ( + _want_chainlit_spontaneous_file_upload( + effective_auth, ingestion_uploads_enabled=ingestion_uploads_enabled, + ) + ) configure_chainlit_allowed_origins(embed_settings, chainlit_config) import chainlit.server as chainlit_server diff --git a/src/gpt_rag_ui/clients/ingestion_client.py b/src/gpt_rag_ui/clients/ingestion_client.py index 98bd56f..f819735 100644 --- a/src/gpt_rag_ui/clients/ingestion_client.py +++ b/src/gpt_rag_ui/clients/ingestion_client.py @@ -8,6 +8,11 @@ import httpx +from gpt_rag_ui.config.appconfig import ( + ATTACHMENTS_UNAVAILABLE_MESSAGE, + load_ingestion_uploads_enabled, +) +from gpt_rag_ui.config.dependencies import get_config from gpt_rag_ui.clients.orchestrator_client import ( _format_outgoing_request_debug, _get_config_value, @@ -189,6 +194,9 @@ async def _build_ingest_documents_payload(conversation_id: str, question_id: str async def ingest_files_session(conversation_id: str, question_id: str, auth_info: dict, files: list[dict]) -> bool: + if not load_ingestion_uploads_enabled(get_config()): + raise RuntimeError(ATTACHMENTS_UNAVAILABLE_MESSAGE) + url, target_context = _build_ingestion_url() headers = {"Content-Type": "application/json"} diff --git a/src/gpt_rag_ui/config/appconfig.py b/src/gpt_rag_ui/config/appconfig.py index 4c46019..fcefa16 100644 --- a/src/gpt_rag_ui/config/appconfig.py +++ b/src/gpt_rag_ui/config/appconfig.py @@ -1,7 +1,7 @@ import os import logging -from typing import Dict, Any +from typing import Dict, Any, Protocol from azure.identity import ChainedTokenCredential, ManagedIdentityCredential, AzureCliCredential from azure.identity.aio import ChainedTokenCredential as AsyncChainedTokenCredential, ManagedIdentityCredential as AsyncManagedIdentityCredential, AzureCliCredential as AsyncAzureCliCredential from azure.appconfiguration import AzureAppConfigurationClient @@ -20,6 +20,29 @@ APP_LABEL = 'agent-app-ui' LEGACY_APP_LABEL = 'gpt-rag-ui' +ATTACHMENTS_UNAVAILABLE_MESSAGE = "Attachments are unavailable in this deployment." + + +class IngestionUploadConfig(Protocol): + def get(self, key: str, default: str, type: type) -> str: ... + + +def load_ingestion_uploads_enabled(config: IngestionUploadConfig) -> bool: + """Read the independent ingestion capability without changing provider precedence. + + Missing preserves legacy upload eligibility; a present malformed value is + an operator error, never an implicit opt-in. + """ + value = config.get("DEPLOY_INGESTION_CONTAINER_APP", "true", str).strip().lower() + if value in {"true", "1", "yes", "y", "on"}: + return True + if value in {"false", "0", "no", "n", "off"}: + return False + raise ConfigurationError( + "DEPLOY_INGESTION_CONTAINER_APP must be a boolean (true or false). " + "Set false for deployments without the ingestion service." + ) + def candidate_keys(key: str) -> list[str]: """Return the lookup keys for ``key``; only the exact key name is read.""" diff --git a/src/gpt_rag_ui/services/chat.py b/src/gpt_rag_ui/services/chat.py index 5a8d414..9269edd 100644 --- a/src/gpt_rag_ui/services/chat.py +++ b/src/gpt_rag_ui/services/chat.py @@ -20,6 +20,10 @@ replace_source_reference_links, ) from gpt_rag_ui.config.dependencies import get_config +from gpt_rag_ui.config.appconfig import ( + ATTACHMENTS_UNAVAILABLE_MESSAGE, + load_ingestion_uploads_enabled, +) from gpt_rag_ui.clients.blob import BlobClient from gpt_rag_ui.services.conversation_security import get_owned_conversation from gpt_rag_ui.clients.ingestion_client import ingest_files_session @@ -526,12 +530,18 @@ def _trim_for_log(value: str, limit: int = 400) -> str: rejected: list[str] = [] total_declared_bytes = 0 file_reply_parts: list[str] = [] + ingestion_uploads_enabled = True if message.elements: + ingestion_uploads_enabled = load_ingestion_uploads_enabled(config) for element in message.elements: if not isinstance(element, cl.File): continue + if not ingestion_uploads_enabled: + rejected.append(ATTACHMENTS_UNAVAILABLE_MESSAGE) + continue + mime = (getattr(element, "mime", "") or "").lower() name = getattr(element, "name", "upload") path = getattr(element, "path", None) @@ -661,7 +671,10 @@ def _trim_for_log(value: str, limit: int = 400) -> str: return user_ask = (message.content or "").strip() - if CHAT_BACKEND == "hosted_agent" and not user_ask: + if not user_ask and ( + CHAT_BACKEND == "hosted_agent" + or (rejected and not ingestion_uploads_enabled) + ): final_text = "".join(file_reply_parts).strip() if not final_text: final_text = "Enter a question to start or continue the hosted conversation." @@ -670,7 +683,7 @@ def _trim_for_log(value: str, limit: int = 400) -> str: response_msg.content = final_text await response_msg.update() logger.info( - "Skipping hosted agent (empty ask): conversation=%s question_id=%s", + "Skipping chat backend (empty ask): conversation=%s question_id=%s", hosted_conversation_id or "new", message.id, ) diff --git a/tests/test_optional_ingestion_uploads.py b/tests/test_optional_ingestion_uploads.py new file mode 100644 index 0000000..9769598 --- /dev/null +++ b/tests/test_optional_ingestion_uploads.py @@ -0,0 +1,297 @@ +"""Offline behavioral coverage for the independent ingestion upload capability.""" + +from contextlib import ExitStack +import importlib +import json +import os +from pathlib import Path +import tempfile +import tomllib +from types import SimpleNamespace +import unittest +from unittest.mock import AsyncMock, MagicMock, Mock, patch + +from gpt_rag_ui.config.appconfig import ( + AppConfigClient, + ATTACHMENTS_UNAVAILABLE_MESSAGE, + load_ingestion_uploads_enabled, +) +from gpt_rag_ui.config.errors import ConfigurationError + + +def configuration(value=None): + # Exercise the real provider lookup/conversion, without creating Azure clients. + config = AppConfigClient.__new__(AppConfigClient) + config.client = {} if value is None else {"DEPLOY_INGESTION_CONTAINER_APP": value} + config.connected = False + return config + + +class UploadConfigurationTests(unittest.TestCase): + def test_real_config_missing_and_valid_boolean_values(self): + for value, expected in ( + (None, True), ("true", True), (True, True), (" false ", False), + (False, False), ("1", True), ("0", False), ("yes", True), ("off", False), + ): + with self.subTest(value=value): + self.assertIs(expected, load_ingestion_uploads_enabled(configuration(value))) + + def test_malformed_values_fail_actionably(self): + for value in ("", " ", "enabled", "tru", 2, [], {}): + with self.subTest(value=value), self.assertRaisesRegex( + ConfigurationError, "DEPLOY_INGESTION_CONTAINER_APP.*true or false", + ): + load_ingestion_uploads_enabled(configuration(value)) + + def test_startup_writes_real_chainlit_controls_independent_of_backend(self): + with patch("gpt_rag_ui.config.dependencies.get_config", return_value=configuration()): + bootstrap = importlib.import_module("gpt_rag_ui.bootstrap") + for backend in ("orchestrator", "hosted_agent"): + for value in (None, "true", "false"): + for anonymous in (False, True): + with self.subTest(backend=backend, value=value, anonymous=anonymous): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / ".chainlit").mkdir() + path = root / ".chainlit" / "config.toml" + path.write_text( + "[features.spontaneous_file_upload]\nenabled = true\n", + encoding="utf-8", + ) + with ( + patch.dict(os.environ, {"CHAT_BACKEND": backend}), + patch.object(bootstrap, "get_asset_root", return_value=root), + ): + bootstrap._sync_chainlit_spontaneous_file_upload( + SimpleNamespace( + allow_anonymous=anonymous, + allow_anonymous_source="test", + ), + ingestion_uploads_enabled=load_ingestion_uploads_enabled( + configuration(value), + ), + ) + settings = tomllib.loads(path.read_text(encoding="utf-8")) + self.assertIs( + value != "false" and not anonymous, + settings["features"]["spontaneous_file_upload"]["enabled"], + ) + + def test_invalid_flag_stops_classic_hosted_and_copilot_startup(self): + with patch("gpt_rag_ui.config.dependencies.get_config", return_value=configuration()): + bootstrap = importlib.import_module("gpt_rag_ui.bootstrap") + from gpt_rag_ui.config.embed_config import EmbedSettings + + for backend in ("orchestrator", "hosted_agent"): + for copilot in (False, True): + with ( + self.subTest(backend=backend, copilot=copilot), + patch.dict(os.environ, {"CHAT_BACKEND": backend}), + patch.object(bootstrap, "_configure_auth_environment") as auth, + self.assertRaisesRegex(ConfigurationError, "DEPLOY_INGESTION_CONTAINER_APP"), + ): + bootstrap._create_chainlit_app( + configuration("invalid"), embed_settings=EmbedSettings(enabled=copilot), + ) + auth.assert_not_called() + + def test_startup_true_restores_controls_after_disabled(self): + with patch("gpt_rag_ui.config.dependencies.get_config", return_value=configuration()): + bootstrap = importlib.import_module("gpt_rag_ui.bootstrap") + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / ".chainlit").mkdir() + path = root / ".chainlit" / "config.toml" + path.write_text("[features.spontaneous_file_upload]\nenabled = false\n", encoding="utf-8") + with patch.object(bootstrap, "get_asset_root", return_value=root): + bootstrap._sync_chainlit_spontaneous_file_upload( + SimpleNamespace(allow_anonymous=False, allow_anonymous_source="test"), + ingestion_uploads_enabled=load_ingestion_uploads_enabled(configuration("true")), + ) + self.assertTrue(tomllib.loads(path.read_text())["features"]["spontaneous_file_upload"]["enabled"]) + + def test_startup_enforces_in_memory_controls_even_without_toml_write(self): + with patch("gpt_rag_ui.config.dependencies.get_config", return_value=configuration()): + bootstrap = importlib.import_module("gpt_rag_ui.bootstrap") + from chainlit.config import config as chainlit_config + + class StartupObserved(RuntimeError): + pass + + for backend in ("orchestrator", "hosted_agent"): + for value in (None, "true", "false"): + expected = value != "false" + + def observe_controls(*args): + self.assertIs(expected, chainlit_config.features.spontaneous_file_upload.enabled) + raise StartupObserved() + + with ( + self.subTest(backend=backend, value=value), + patch.dict(os.environ, {"CHAT_BACKEND": backend}), + patch.object(chainlit_config.features.spontaneous_file_upload, "enabled", not expected), + patch.object(bootstrap, "_configure_embed_environment"), + patch.object(bootstrap, "_configure_auth_environment"), + patch.object(bootstrap, "_sync_chainlit_spontaneous_file_upload"), + patch.object(bootstrap, "configure_chainlit_allowed_origins", side_effect=observe_controls), + self.assertRaises(StartupObserved), + ): + bootstrap._create_chainlit_app( + configuration(value), auth_state=SimpleNamespace(allow_anonymous=False), + ) + + +class UploadBoundaryTests(unittest.IsolatedAsyncioTestCase): + async def test_copilot_project_settings_do_not_override_capability(self): + from gpt_rag_ui.auth import embed_security + + for value in (None, "true", "false"): + for mode in ("entra", "anonymous"): + for originally_enabled in (False, True): + with self.subTest(value=value, mode=mode, original=originally_enabled): + send = AsyncMock() + with patch.object(embed_security, "get_config", return_value=configuration(value)): + wrapped = embed_security._copilot_project_settings_send(send, mode) + await wrapped({"type": "http.response.start", "status": 200, "headers": []}) + await wrapped({ + "type": "http.response.body", + "body": json.dumps({ + "features": {"spontaneous_file_upload": {"enabled": originally_enabled}}, + "dataPersistence": True, + }).encode(), + }) + payload = json.loads(send.await_args_list[-1].args[0]["body"]) + self.assertIs( + mode == "entra" and value != "false", + payload["features"]["spontaneous_file_upload"]["enabled"], + ) + self.assertIs(mode == "entra", payload["dataPersistence"]) + + with ( + patch.object(embed_security, "get_config", return_value=configuration("invalid")), + self.assertRaisesRegex(ConfigurationError, "DEPLOY_INGESTION_CONTAINER_APP"), + ): + embed_security._copilot_project_settings_send(AsyncMock(), "entra") + + async def test_client_false_and_invalid_reject_before_both_target_fallbacks(self): + from gpt_rag_ui.clients import ingestion_client as client + + for value in ("false", "invalid"): + # Both derived Container App URL and Dapr fallback configurations. + for orchestrator_url in ("https://example-orchestrator.example.com", None): + with self.subTest(value=value, orchestrator_url=orchestrator_url), ExitStack() as stack: + stack.enter_context(patch.object(client, "get_config", return_value=configuration(value))) + stack.enter_context(patch.dict(os.environ, {}, clear=True)) + build_url = stack.enter_context(patch.object(client, "_build_ingestion_url")) + base_url = stack.enter_context(patch.object(client, "_get_ingestion_base_url")) + orchestrator = stack.enter_context(patch.object( + client, "_get_orchestrator_base_url", return_value=orchestrator_url, + )) + config_lookup = stack.enter_context(patch.object(client, "_get_config_value")) + payload = stack.enter_context(patch.object(client, "_build_ingest_documents_payload")) + http = stack.enter_context(patch.object(client.httpx, "AsyncClient")) + with self.assertRaisesRegex( + RuntimeError if value == "false" else ConfigurationError, + "Attachments are unavailable" if value == "false" else "DEPLOY_INGESTION_CONTAINER_APP", + ): + await client.ingest_files_session("conversation", "question", {}, [{"path": "forged.pdf"}]) + for mock in (build_url, base_url, orchestrator, config_lookup, payload, http): + mock.assert_not_called() + + async def test_client_missing_and_true_keep_existing_resolution(self): + from gpt_rag_ui.clients import ingestion_client as client + + for value in (None, "true"): + with ( + self.subTest(value=value), + patch.object(client, "get_config", return_value=configuration(value)), + patch.object(client, "_build_ingestion_url", side_effect=RuntimeError("target reached")) as target, + self.assertRaisesRegex(RuntimeError, "target reached"), + ): + await client.ingest_files_session("conversation", "question", {}, []) + target.assert_called_once() + + async def test_handler_attachment_matrix_preserves_auth_and_conversation_checks(self): + with ( + patch.dict(os.environ, {"CHAT_BACKEND": "orchestrator", "CHAINLIT_AUTH_SECRET": "test-secret"}), + patch("gpt_rag_ui.config.dependencies.get_config", return_value=configuration()), + patch("gpt_rag_ui.telemetry.monitoring.Telemetry.configure_monitoring"), + ): + chat = importlib.import_module("gpt_rag_ui.services.chat") + + for backend in ("orchestrator", "hosted_agent"): + for mode in (None, "entra", "anonymous"): + for value in (None, "true", "false"): + for has_conversation in (False, True): + with self.subTest( + backend=backend, mode=mode, value=value, conversation=has_conversation, + ), ExitStack() as stack: + state = {} + if has_conversation: + state.update(conversation_id="classic-id", hosted_agent_conversation_id="hosted-id") + response = AsyncMock() + response.content = "" + ingest = AsyncMock(return_value=True) + owner = AsyncMock(return_value={"id": "classic-id"}) + turns = [] + + async def stream(*args, **kwargs): + turns.append((args, kwargs)) + if False: + yield + + for name, replacement in { + "cl": SimpleNamespace( + File=SimpleNamespace, Message=Mock(return_value=response), + user_session=SimpleNamespace( + get=state.get, set=lambda key, val: state.__setitem__(key, val), + ), + chat_context=SimpleNamespace(to_openai=lambda: []), + ), + "config": configuration(value), "tracer": MagicMock(), "CHAT_BACKEND": backend, + "HOSTED_CONTINUITY_ENABLED": False, "SHOW_STATISTICS": False, "ENABLE_FEEDBACK": False, + "get_auth_info": AsyncMock(return_value={"authorized": True, "copilot_auth_mode": mode}), + "ingest_files_session": ingest, "get_owned_conversation": owner, + "call_orchestrator_stream": stream, "call_hosted_agent_stream": stream, + "build_invocation_messages": Mock(return_value=[]), + }.items(): + stack.enter_context(patch.object(chat, name, replacement, create=True)) + await chat.handle_message(SimpleNamespace( + id="upload-reference", content="A text question", + elements=[SimpleNamespace( + mime="application/pdf", name="file.pdf", path="forged.pdf", size=1, + )], + )) + eligible = value != "false" and mode != "anonymous" and ( + backend == "orchestrator" or has_conversation + ) + self.assertEqual(int(eligible), ingest.await_count) + self.assertEqual( + int(eligible and backend == "orchestrator" and has_conversation), + owner.await_count, + ) + self.assertEqual(1, len(turns), "Text chat must remain available") + notices = "".join(call.args[0] for call in response.stream_token.await_args_list) + if value == "false": + self.assertIn(ATTACHMENTS_UNAVAILABLE_MESSAGE, notices) + self.assertNotIn("uploaded_docs", state) + if eligible: + self.assertEqual(["file.pdf"], state["uploaded_docs"]) + if has_conversation: + self.assertEqual( + "hosted-id" if backend == "hosted_agent" else "classic-id", + ingest.await_args.kwargs["conversation_id"], + ) + + # Attachment-only forged messages must not reach either chat backend. + if value == "false": + turns.clear() + await chat.handle_message(SimpleNamespace( + id="forged-only", content="", elements=[SimpleNamespace()], + )) + self.assertEqual([], turns) + self.assertIn(ATTACHMENTS_UNAVAILABLE_MESSAGE, response.content) + + +if __name__ == "__main__": + unittest.main()