diff --git a/application/single_app/config.py b/application/single_app/config.py index a182549ff..7cd617af2 100644 --- a/application/single_app/config.py +++ b/application/single_app/config.py @@ -97,7 +97,7 @@ EXECUTOR_TYPE = 'thread' EXECUTOR_MAX_WORKERS = 30 SESSION_TYPE = 'filesystem' -VERSION = "0.250.061" +VERSION = "0.250.062" IS_DEVELOPMENT = is_development_env_enabled() SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax') diff --git a/application/single_app/functions_mixed_source_orchestration.py b/application/single_app/functions_mixed_source_orchestration.py new file mode 100644 index 000000000..11d4934ba --- /dev/null +++ b/application/single_app/functions_mixed_source_orchestration.py @@ -0,0 +1,684 @@ +# functions_mixed_source_orchestration.py +"""Authorization-safe source manifest and evidence contracts for mixed sources.""" + +import json +import logging +import math +import os +import time + +from functions_appinsights import log_event + + +SOURCE_KIND_TABULAR = "tabular" +SOURCE_KIND_NARRATIVE = "narrative" +SOURCE_KIND_UNSUPPORTED = "unsupported" +SOURCE_KIND_UNRESOLVED = "unresolved" +SOURCE_KINDS = frozenset({ + SOURCE_KIND_TABULAR, + SOURCE_KIND_NARRATIVE, + SOURCE_KIND_UNSUPPORTED, + SOURCE_KIND_UNRESOLVED, +}) + +SOURCE_SCOPE_PERSONAL = "personal" +SOURCE_SCOPE_GROUP = "group" +SOURCE_SCOPE_PUBLIC = "public" +SOURCE_SCOPE_CHAT = "chat" +SOURCE_SCOPES = frozenset({ + SOURCE_SCOPE_PERSONAL, + SOURCE_SCOPE_GROUP, + SOURCE_SCOPE_PUBLIC, + SOURCE_SCOPE_CHAT, +}) + +AUTHORIZATION_STATUS_AUTHORIZED = "authorized" +AUTHORIZATION_STATUS_UNRESOLVED = "unresolved" +SOURCE_MANIFEST_MAX_SOURCES = 100 + +SELECTION_MODE_SELECTED = "selected" +SELECTION_MODE_ALL = "all" +SELECTION_MODE_HISTORY = "history" +SELECTION_MODE_RELEVANCE = "relevance" +SELECTION_MODES = frozenset({ + SELECTION_MODE_SELECTED, + SELECTION_MODE_ALL, + SELECTION_MODE_HISTORY, + SELECTION_MODE_RELEVANCE, +}) + +TABULAR_SOURCE_EXTENSIONS = frozenset({".csv", ".xls", ".xlsx", ".xlsm"}) +NARRATIVE_SOURCE_EXTENSIONS = frozenset({ + ".txt", ".doc", ".docm", ".docx", ".html", ".htm", ".md", ".markdown", + ".json", ".xml", ".yaml", ".yml", ".log", ".pdf", ".ppt", ".pptx", + ".msg", ".vsdx", ".jpg", ".jpeg", ".png", ".bmp", ".tiff", ".tif", + ".heif", ".heic", ".3ga", ".aac", ".ac3", ".aif", ".aifc", ".aiff", + ".amr", ".ape", ".au", ".caf", ".dts", ".f4a", ".flac", ".m4a", + ".m4b", ".m4r", ".mka", ".mp2", ".mp3", ".mpa", ".oga", ".ogg", + ".opus", ".spx", ".wav", ".weba", ".wma", ".wv", ".mp4", ".mov", + ".avi", ".mkv", ".flv", ".mxf", ".gxf", ".ts", ".ps", ".3gp", + ".3gpp", ".mpg", ".wmv", ".asf", ".m4v", ".isma", ".ismv", + ".dvr-ms", ".webm", ".mpeg", +}) + +EVIDENCE_ENGINE_TABULAR_TOOLS = "tabular_tools" +EVIDENCE_ENGINE_DOCUMENT_ANALYSIS = "document_analysis" +EVIDENCE_ENGINE_HYBRID_SEARCH = "hybrid_search" +EVIDENCE_ENGINES = frozenset({ + EVIDENCE_ENGINE_TABULAR_TOOLS, + EVIDENCE_ENGINE_DOCUMENT_ANALYSIS, + EVIDENCE_ENGINE_HYBRID_SEARCH, +}) + +EVIDENCE_STATUS_COMPLETED = "completed" +EVIDENCE_STATUS_PARTIAL = "partial" +EVIDENCE_STATUS_FAILED = "failed" +EVIDENCE_STATUS_SKIPPED = "skipped" +EVIDENCE_STATUSES = frozenset({ + EVIDENCE_STATUS_COMPLETED, + EVIDENCE_STATUS_PARTIAL, + EVIDENCE_STATUS_FAILED, + EVIDENCE_STATUS_SKIPPED, +}) + +EVIDENCE_ENVELOPE_MAX_BYTES = 65536 +EVIDENCE_SUMMARY_MAX_BYTES = 4096 +EVIDENCE_ERROR_MAX_BYTES = 1024 +EVIDENCE_LIST_MAX_ITEMS = 10 +EVIDENCE_ITEM_MAX_BYTES = 1536 +EVIDENCE_COVERAGE_MAX_BYTES = 4096 +EVIDENCE_JSON_MAX_DEPTH = 4 +EVIDENCE_JSON_MAX_COLLECTION_ITEMS = 20 +EVIDENCE_JSON_MAX_STRING_BYTES = 1024 + + +def normalize_selection_mode(selection_mode, default=SELECTION_MODE_SELECTED): + """Return a supported selection mode or raise for an invalid explicit value.""" + normalized_default = str(default or "").strip().lower() + if normalized_default not in SELECTION_MODES: + raise ValueError("Invalid default selection_mode") + + normalized_mode = str(selection_mode or "").strip().lower() + if not normalized_mode: + return normalized_default + if normalized_mode not in SELECTION_MODES: + raise ValueError( + f"selection_mode must be one of: {', '.join(sorted(SELECTION_MODES))}" + ) + return normalized_mode + + +def classify_source_kind(file_name, document_item=None): + """Classify a resolved source by native capability without reading its content.""" + normalized_file_name = str(file_name or "").strip() + extension = os.path.splitext(normalized_file_name)[1].lower() + if extension in TABULAR_SOURCE_EXTENSIONS: + return SOURCE_KIND_TABULAR + if extension in NARRATIVE_SOURCE_EXTENSIONS: + return SOURCE_KIND_NARRATIVE + + document_item = document_item if isinstance(document_item, dict) else {} + if any( + document_item.get(field_name) + for field_name in ( + "num_file_chunks", + "comparison_text", + "extracted_text", + "vision_analysis", + ) + ): + return SOURCE_KIND_NARRATIVE + return SOURCE_KIND_UNSUPPORTED + + +def _normalize_document_id(requested_source): + if isinstance(requested_source, dict): + requested_source = requested_source.get("document_id") or requested_source.get("id") + return str(requested_source or "").strip() + + +def _normalize_identifier_list(values): + if values is None: + return [] + if isinstance(values, (str, int)): + values = [values] + return [ + normalized_value + for normalized_value in ( + str(value or "").strip() + for value in list(values) + ) + if normalized_value + ] + + +def _safe_file_name(file_name): + return str(file_name or "").replace("\\", "/").split("/")[-1].strip() + + +def _unresolved_manifest_entry(document_id): + return { + "document_id": document_id, + "display_name": None, + "file_name": None, + "extension": None, + "source_kind": SOURCE_KIND_UNRESOLVED, + "scope": None, + "scope_id": None, + "group_id": None, + "public_workspace_id": None, + "conversation_id": None, + "source_version": None, + "authorization_status": AUTHORIZATION_STATUS_UNRESOLVED, + } + + +def _build_authorized_manifest_entry(document_id, user_id, document_context): + if not isinstance(document_context, dict): + return _unresolved_manifest_entry(document_id) + + document_item = document_context.get("document") + if not isinstance(document_item, dict): + return _unresolved_manifest_entry(document_id) + + resolved_document_id = str(document_item.get("id") or "").strip() + if resolved_document_id != document_id: + return _unresolved_manifest_entry(document_id) + + scope = str(document_context.get("scope") or "").strip().lower() + if scope not in SOURCE_SCOPES: + return _unresolved_manifest_entry(document_id) + + group_id = None + public_workspace_id = None + conversation_id = str( + document_context.get("conversation_id") + or document_item.get("conversation_id") + or "" + ).strip() or None + + if scope == SOURCE_SCOPE_PERSONAL: + scope_id = str(document_item.get("user_id") or user_id or "").strip() + elif scope == SOURCE_SCOPE_GROUP: + group_id = str(document_context.get("group_id") or "").strip() or None + scope_id = group_id + elif scope == SOURCE_SCOPE_PUBLIC: + public_workspace_id = str( + document_context.get("public_workspace_id") or "" + ).strip() or None + scope_id = public_workspace_id + else: + scope_id = conversation_id + + if not scope_id: + return _unresolved_manifest_entry(document_id) + + file_name = _safe_file_name( + document_item.get("file_name") + or document_item.get("filename") + or document_item.get("title") + ) + display_name = str(document_item.get("title") or file_name or document_id).strip() + extension = os.path.splitext(file_name)[1].lower() or None + source_version = document_item.get("version") + if source_version is None: + source_version = document_item.get("source_version") + if source_version is not None and not isinstance(source_version, (str, int, float)): + source_version = str(source_version) + + return { + "document_id": document_id, + "display_name": display_name, + "file_name": file_name or None, + "extension": extension, + "source_kind": classify_source_kind(file_name, document_item=document_item), + "scope": scope, + "scope_id": scope_id, + "group_id": group_id, + "public_workspace_id": public_workspace_id, + "conversation_id": conversation_id, + "source_version": source_version, + "authorization_status": AUTHORIZATION_STATUS_AUTHORIZED, + } + + +def _default_document_context_batch_resolver(**resolver_arguments): + # Imported lazily so this contract module remains usable by startup code and isolated tests. + from functions_search_service import resolve_document_contexts + + resolver_arguments["include_content"] = False + return resolve_document_contexts(**resolver_arguments) + + +def resolve_authorized_source_manifest( + requested_sources, + user_id, + selection_mode=SELECTION_MODE_SELECTED, + conversation_id=None, + active_group_ids=None, + active_public_workspace_ids=None, + context_resolver=None, +): + """Resolve each unique requested ID once into an ordered, authorized manifest.""" + normalized_selection_mode = normalize_selection_mode(selection_mode) + normalized_user_id = str(user_id or "").strip() + if not normalized_user_id: + raise ValueError("user_id is required") + + if isinstance(requested_sources, (str, int, dict)): + requested_source_list = [requested_sources] + else: + requested_source_list = list(requested_sources or []) + if len(requested_source_list) > SOURCE_MANIFEST_MAX_SOURCES: + log_event( + "[MixedSourceManifest] Rejected over-limit source manifest request.", + extra={ + "selection_mode": normalized_selection_mode, + "requested_source_count": len(requested_source_list), + "source_limit": SOURCE_MANIFEST_MAX_SOURCES, + }, + level=logging.WARNING, + ) + raise ValueError( + f"A source manifest supports at most {SOURCE_MANIFEST_MAX_SOURCES} requested sources" + ) + + unique_document_ids = [] + seen_document_ids = set() + duplicate_ids_removed = 0 + for requested_source in requested_source_list: + document_id = _normalize_document_id(requested_source) + if not document_id: + continue + if document_id in seen_document_ids: + duplicate_ids_removed += 1 + continue + seen_document_ids.add(document_id) + unique_document_ids.append(document_id) + + if context_resolver is not None and not callable(context_resolver): + raise TypeError("context_resolver must be callable") + + started_at = time.perf_counter() + manifest = [] + resolution_error_count = 0 + normalized_active_group_ids = _normalize_identifier_list(active_group_ids) + normalized_public_workspace_ids = _normalize_identifier_list( + active_public_workspace_ids + ) + normalized_conversation_id = str(conversation_id or "").strip() or None + + resolved_contexts = None + if context_resolver is None: + try: + resolved_contexts = _default_document_context_batch_resolver( + document_ids=unique_document_ids, + user_id=normalized_user_id, + doc_scope="all", + active_group_ids=normalized_active_group_ids, + active_public_workspace_id=normalized_public_workspace_ids, + conversation_id=normalized_conversation_id, + ) + except Exception: + resolved_contexts = [None] * len(unique_document_ids) + resolution_error_count = len(unique_document_ids) + if ( + not isinstance(resolved_contexts, list) + or len(resolved_contexts) != len(unique_document_ids) + ): + resolved_contexts = [None] * len(unique_document_ids) + resolution_error_count = len(unique_document_ids) + + for document_index, document_id in enumerate(unique_document_ids): + document_context = None + if resolved_contexts is not None: + document_context = resolved_contexts[document_index] + else: + try: + document_context = context_resolver( + document_id=document_id, + user_id=normalized_user_id, + doc_scope="all", + active_group_ids=normalized_active_group_ids, + active_public_workspace_id=normalized_public_workspace_ids, + conversation_id=normalized_conversation_id, + ) + except Exception: + resolution_error_count += 1 + manifest.append( + _build_authorized_manifest_entry( + document_id, + normalized_user_id, + document_context, + ) + ) + + source_kind_counts = {source_kind: 0 for source_kind in SOURCE_KINDS} + scope_distribution = {scope: 0 for scope in SOURCE_SCOPES} + for entry in manifest: + source_kind_counts[entry["source_kind"]] += 1 + if entry["scope"] in scope_distribution: + scope_distribution[entry["scope"]] += 1 + + duration_ms = round((time.perf_counter() - started_at) * 1000, 3) + resolved_source_count = len(manifest) - source_kind_counts[SOURCE_KIND_UNRESOLVED] + log_event( + "[MixedSourceManifest] Resolved authorized source manifest.", + extra={ + "selection_mode": normalized_selection_mode, + "requested_source_count": len(requested_source_list), + "unique_source_count": len(unique_document_ids), + "resolved_source_count": resolved_source_count, + "tabular_source_count": source_kind_counts[SOURCE_KIND_TABULAR], + "narrative_source_count": source_kind_counts[SOURCE_KIND_NARRATIVE], + "unsupported_source_count": source_kind_counts[SOURCE_KIND_UNSUPPORTED], + "unresolved_or_unauthorized_count": source_kind_counts[SOURCE_KIND_UNRESOLVED], + "duplicate_ids_removed": duplicate_ids_removed, + "resolution_error_count": resolution_error_count, + "scope_distribution": scope_distribution, + "manifest_resolution_duration_ms": duration_ms, + }, + level=logging.INFO, + ) + return manifest + + +def partition_source_manifest(manifest): + """Partition a manifest by capability while preserving order within each cohort.""" + partitions = { + "tabular_sources": [], + "narrative_sources": [], + "unsupported_sources": [], + "unresolved_sources": [], + } + partition_key_by_source_kind = { + SOURCE_KIND_TABULAR: "tabular_sources", + SOURCE_KIND_NARRATIVE: "narrative_sources", + SOURCE_KIND_UNSUPPORTED: "unsupported_sources", + SOURCE_KIND_UNRESOLVED: "unresolved_sources", + } + + for raw_entry in list(manifest or []): + entry = raw_entry if isinstance(raw_entry, dict) else {} + document_id = _normalize_document_id(entry) + if entry.get("authorization_status") != AUTHORIZATION_STATUS_AUTHORIZED: + partitions["unresolved_sources"].append( + _unresolved_manifest_entry(document_id) + ) + continue + + partition_key = partition_key_by_source_kind.get( + entry.get("source_kind"), + "unsupported_sources", + ) + partitions[partition_key].append(entry) + + return partitions + + +def _truncate_utf8(value, max_bytes): + normalized_value = str(value or "") + encoded_value = normalized_value.encode("utf-8") + if len(encoded_value) <= max_bytes: + return normalized_value + if max_bytes <= 3: + return encoded_value[:max_bytes].decode("utf-8", errors="ignore") + return ( + encoded_value[:max_bytes - 3].decode("utf-8", errors="ignore").rstrip() + + "..." + ) + + +def _make_json_safe(value, depth=0): + if value is None or isinstance(value, (bool, int)): + return value, False + if isinstance(value, float): + return (value, False) if math.isfinite(value) else (None, True) + if isinstance(value, str): + bounded_value = _truncate_utf8(value, EVIDENCE_JSON_MAX_STRING_BYTES) + return bounded_value, bounded_value != value + if depth >= EVIDENCE_JSON_MAX_DEPTH: + return _truncate_utf8(str(value), EVIDENCE_JSON_MAX_STRING_BYTES), True + if isinstance(value, dict): + source_items = list(value.items()) + bounded_value = {} + was_truncated = len(source_items) > EVIDENCE_JSON_MAX_COLLECTION_ITEMS + for key, item_value in source_items[:EVIDENCE_JSON_MAX_COLLECTION_ITEMS]: + normalized_key = _truncate_utf8(key, 128) + bounded_item, item_was_truncated = _make_json_safe( + item_value, + depth + 1, + ) + if ( + not isinstance(key, str) + or normalized_key != key + or normalized_key in bounded_value + ): + was_truncated = True + bounded_value[normalized_key] = bounded_item + was_truncated = was_truncated or item_was_truncated + return bounded_value, was_truncated + if isinstance(value, (list, tuple, set)): + source_items = list(value) + bounded_value = [] + was_truncated = len(source_items) > EVIDENCE_JSON_MAX_COLLECTION_ITEMS + for item in source_items[:EVIDENCE_JSON_MAX_COLLECTION_ITEMS]: + bounded_item, item_was_truncated = _make_json_safe(item, depth + 1) + bounded_value.append(bounded_item) + was_truncated = was_truncated or item_was_truncated + return bounded_value, was_truncated + return _truncate_utf8(str(value), EVIDENCE_JSON_MAX_STRING_BYTES), True + + +def _json_size_bytes(value): + return len( + json.dumps( + value, + allow_nan=False, + ensure_ascii=False, + separators=(",", ":"), + ).encode("utf-8") + ) + + +def _bound_json_value(value, max_bytes): + safe_value, safe_value_was_truncated = _make_json_safe(value) + if _json_size_bytes(safe_value) <= max_bytes: + return safe_value, safe_value_was_truncated + + serialized_preview = json.dumps( + safe_value, + allow_nan=False, + ensure_ascii=False, + separators=(",", ":"), + ) + preview_max_bytes = max(16, max_bytes // 3) + while preview_max_bytes > 0: + bounded_value = { + "truncated": True, + "preview": _truncate_utf8(serialized_preview, preview_max_bytes), + } + if _json_size_bytes(bounded_value) <= max_bytes: + return bounded_value, True + preview_max_bytes //= 2 + return {"truncated": True}, True + + +def _bound_json_list(values): + if values is None: + return [], False + if not isinstance(values, (list, tuple)): + raise ValueError("Evidence collection values must be lists") + + source_values = list(values) + bounded_values = [] + was_truncated = len(source_values) > EVIDENCE_LIST_MAX_ITEMS + for value in source_values[:EVIDENCE_LIST_MAX_ITEMS]: + bounded_value, value_was_truncated = _bound_json_value( + value, + EVIDENCE_ITEM_MAX_BYTES, + ) + bounded_values.append(bounded_value) + was_truncated = was_truncated or value_was_truncated + return bounded_values, was_truncated + + +def _build_truncated_coverage(coverage, coverage_was_truncated=False): + normalized_coverage = dict(coverage or {}) + normalized_coverage["evidence_envelope_truncated"] = True + if coverage_was_truncated: + normalized_coverage["coverage_truncated"] = True + + bounded_coverage, additional_truncation = _bound_json_value( + normalized_coverage, + EVIDENCE_COVERAGE_MAX_BYTES, + ) + if additional_truncation: + return { + "evidence_envelope_truncated": True, + "coverage_truncated": True, + } + return bounded_coverage + + +def build_evidence_envelope( + document_id, + source_kind, + engine, + status, + summary="", + evidence=None, + citations=None, + generated_artifacts=None, + coverage=None, + error=None, +): + """Build a bounded, JSON-safe evidence envelope for later synthesis phases.""" + normalized_document_id = str(document_id or "").strip() + if not normalized_document_id: + raise ValueError("document_id is required") + + normalized_source_kind = str(source_kind or "").strip().lower() + if normalized_source_kind not in {SOURCE_KIND_TABULAR, SOURCE_KIND_NARRATIVE}: + raise ValueError("Evidence source_kind must be tabular or narrative") + + normalized_engine = str(engine or "").strip().lower() + if normalized_engine not in EVIDENCE_ENGINES: + raise ValueError(f"Unsupported evidence engine: {normalized_engine}") + + normalized_status = str(status or "").strip().lower() + if normalized_status not in EVIDENCE_STATUSES: + raise ValueError(f"Unsupported evidence status: {normalized_status}") + + if coverage is not None and not isinstance(coverage, dict): + raise ValueError("coverage must be a dictionary") + + bounded_evidence, evidence_was_truncated = _bound_json_list(evidence) + bounded_citations, citations_were_truncated = _bound_json_list(citations) + bounded_artifacts, artifacts_were_truncated = _bound_json_list(generated_artifacts) + normalized_summary = _truncate_utf8(summary, EVIDENCE_SUMMARY_MAX_BYTES) + normalized_error = ( + _truncate_utf8(error, EVIDENCE_ERROR_MAX_BYTES) + if error is not None + else None + ) + bounds_applied = bool( + evidence_was_truncated + or citations_were_truncated + or artifacts_were_truncated + or len(str(summary or "").encode("utf-8")) > EVIDENCE_SUMMARY_MAX_BYTES + or ( + error is not None + and len(str(error).encode("utf-8")) > EVIDENCE_ERROR_MAX_BYTES + ) + ) + normalized_coverage = dict(coverage or {}) + if bounds_applied: + normalized_coverage["evidence_envelope_truncated"] = True + bounded_coverage, coverage_was_truncated = _bound_json_value( + normalized_coverage, + EVIDENCE_COVERAGE_MAX_BYTES, + ) + if coverage_was_truncated: + bounded_coverage = _build_truncated_coverage( + {}, + coverage_was_truncated=True, + ) + + envelope = { + "document_id": normalized_document_id, + "source_kind": normalized_source_kind, + "engine": normalized_engine, + "status": normalized_status, + "summary": normalized_summary, + "evidence": bounded_evidence, + "citations": bounded_citations, + "generated_artifacts": bounded_artifacts, + "coverage": bounded_coverage, + "error": normalized_error, + } + + while _json_size_bytes(envelope) > EVIDENCE_ENVELOPE_MAX_BYTES: + candidate_field = max( + ("evidence", "citations", "generated_artifacts"), + key=lambda field_name: len(envelope[field_name]), + ) + if envelope[candidate_field]: + envelope[candidate_field].pop() + envelope["coverage"] = _build_truncated_coverage( + envelope["coverage"], + ) + continue + envelope["summary"] = _truncate_utf8( + envelope["summary"], + max(128, len(envelope["summary"].encode("utf-8")) // 2), + ) + if len(envelope["summary"].encode("utf-8")) <= 128: + raise ValueError("Unable to bound evidence envelope") + + return envelope + + +def serialize_evidence_envelope(envelope): + """Validate and serialize a bounded evidence envelope.""" + if not isinstance(envelope, dict): + raise ValueError("Evidence envelope must be a dictionary") + + required_fields = { + "document_id", + "source_kind", + "engine", + "status", + "summary", + "evidence", + "citations", + "generated_artifacts", + "coverage", + "error", + } + if set(envelope) != required_fields: + raise ValueError("Evidence envelope fields do not match the contract") + + bounded_envelope = build_evidence_envelope( + document_id=envelope["document_id"], + source_kind=envelope["source_kind"], + engine=envelope["engine"], + status=envelope["status"], + summary=envelope["summary"], + evidence=envelope["evidence"], + citations=envelope["citations"], + generated_artifacts=envelope["generated_artifacts"], + coverage=envelope["coverage"], + error=envelope["error"], + ) + + serialized_envelope = json.dumps( + bounded_envelope, + allow_nan=False, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + if len(serialized_envelope.encode("utf-8")) > EVIDENCE_ENVELOPE_MAX_BYTES: + raise ValueError("Evidence envelope exceeds its serialized size bound") + return serialized_envelope \ No newline at end of file diff --git a/application/single_app/functions_search_service.py b/application/single_app/functions_search_service.py index 58b9793b0..becbe4502 100644 --- a/application/single_app/functions_search_service.py +++ b/application/single_app/functions_search_service.py @@ -12,7 +12,12 @@ from azure.cosmos.exceptions import CosmosResourceNotFoundError from openai import AzureOpenAI -from config import CLIENTS, cognitive_services_scope, cosmos_messages_container +from config import ( + CLIENTS, + cognitive_services_scope, + cosmos_conversations_container, + cosmos_messages_container, +) from functions_appinsights import log_event from functions_debug import debug_print from functions_documents import get_document_record, get_ordered_document_chunks @@ -238,34 +243,101 @@ def _build_chat_upload_chunks(text_content, max_chunks=None): return chunks -def _resolve_chat_upload_context(document_id, conversation_id=None): +def _authorize_chat_upload_conversation(user_id, conversation_id): + normalized_user_id = str(user_id or "").strip() + normalized_conversation_id = str(conversation_id or "").strip() + if not normalized_user_id or not normalized_conversation_id: + return False + + try: + conversation_item = cosmos_conversations_container.read_item( + item=normalized_conversation_id, + partition_key=normalized_conversation_id, + ) + except CosmosResourceNotFoundError: + return False + except Exception as exc: + log_event( + "[SearchService] Failed to authorize chat upload conversation.", + extra={"exception_type": type(exc).__name__}, + level=logging.WARNING, + exceptionTraceback=True, + debug_only=True, + ) + return False + + return str(conversation_item.get("user_id") or "").strip() == normalized_user_id + + +def _resolve_chat_upload_context( + document_id, + user_id=None, + conversation_id=None, + include_content=True, + authorization_prechecked=False, +): normalized_conversation_id = str(conversation_id or "").strip() normalized_document_id = str(document_id or "").strip() if not normalized_conversation_id or not normalized_document_id: return None + if ( + not authorization_prechecked + and not _authorize_chat_upload_conversation(user_id, normalized_conversation_id) + ): + return None try: - message_item = cosmos_messages_container.read_item( - item=normalized_document_id, - partition_key=normalized_conversation_id, - ) + if include_content: + message_item = cosmos_messages_container.read_item( + item=normalized_document_id, + partition_key=normalized_conversation_id, + ) + else: + metadata_items = list(cosmos_messages_container.query_items( + query=""" + SELECT TOP 1 + c.id, + c.role, + c.filename, + c.title, + c.version, + c.metadata.is_user_upload AS is_user_upload, + c.metadata.is_generated_chat_artifact AS is_generated_chat_artifact, + c.metadata.generated_artifact_capability AS generated_artifact_capability, + c.metadata.generated_artifact_output_format AS generated_artifact_output_format + FROM c + WHERE c.id = @document_id + """, + parameters=[ + {"name": "@document_id", "value": normalized_document_id}, + ], + partition_key=normalized_conversation_id, + )) + if not metadata_items: + return None + message_item = metadata_items[0] except CosmosResourceNotFoundError: return None except Exception as exc: - debug_print( - "[SearchService] Failed to resolve chat upload context | " - f"document_id={normalized_document_id} | conversation_id={normalized_conversation_id} | error={exc}" + log_event( + "[SearchService] Failed to resolve authorized chat upload context.", + extra={"exception_type": type(exc).__name__}, + level=logging.WARNING, + exceptionTraceback=True, + debug_only=True, ) return None role_name = str(message_item.get("role") or "").strip().lower() metadata = message_item.get("metadata", {}) or {} - is_uploaded_image = role_name == "image" and bool((message_item.get("metadata") or {}).get("is_user_upload")) + is_uploaded_image = role_name == "image" and bool( + metadata.get("is_user_upload") or message_item.get("is_user_upload") + ) if role_name not in {"file", "image"} or (role_name == "image" and not is_uploaded_image): return None - comparison_text = _coerce_chat_upload_text(message_item) - if not comparison_text: + comparison_text = _coerce_chat_upload_text(message_item) if include_content else "" + if include_content and not comparison_text: return None message_title = str(message_item.get("filename") or message_item.get("title") or normalized_document_id).strip() or normalized_document_id @@ -275,11 +347,24 @@ def _resolve_chat_upload_context(document_id, conversation_id=None): "title": message_title, "conversation_id": normalized_conversation_id, "source_type": "chat_upload", - "source_subtype": "generated_chat_artifact" if metadata.get("is_generated_chat_artifact") else "chat_upload", - "artifact_capability": str(metadata.get("generated_artifact_capability") or "").strip().lower() or None, - "artifact_output_format": str(metadata.get("generated_artifact_output_format") or "").strip().lower() or None, - "comparison_text": comparison_text, + "source_subtype": "generated_chat_artifact" if ( + metadata.get("is_generated_chat_artifact") + or message_item.get("is_generated_chat_artifact") + ) else "chat_upload", + "artifact_capability": str( + metadata.get("generated_artifact_capability") + or message_item.get("generated_artifact_capability") + or "" + ).strip().lower() or None, + "artifact_output_format": str( + metadata.get("generated_artifact_output_format") + or message_item.get("generated_artifact_output_format") + or "" + ).strip().lower() or None, + "version": message_item.get("version"), } + if include_content: + resolved_document["comparison_text"] = comparison_text return { "scope": "chat", "group_id": None, @@ -289,6 +374,59 @@ def _resolve_chat_upload_context(document_id, conversation_id=None): } +def _resolve_personal_document_context(document_id, user_id): + personal_document = get_document_record( + user_id=user_id, + document_id=document_id, + ) + if not personal_document: + return None + return { + "scope": "personal", + "group_id": None, + "public_workspace_id": None, + "document": personal_document, + } + + +def _resolve_group_document_context(document_id, user_id, authorized_group_ids): + for group_id in authorized_group_ids or []: + group_document = get_document_record( + user_id=user_id, + document_id=document_id, + group_id=group_id, + ) + if group_document: + return { + "scope": "group", + "group_id": group_id, + "public_workspace_id": None, + "document": group_document, + } + return None + + +def _resolve_public_document_context( + document_id, + user_id, + authorized_public_workspace_ids, +): + for public_workspace_id in authorized_public_workspace_ids or []: + public_document = get_document_record( + user_id=user_id, + document_id=document_id, + public_workspace_id=public_workspace_id, + ) + if public_document: + return { + "scope": "public", + "group_id": None, + "public_workspace_id": public_workspace_id, + "document": public_document, + } + return None + + def resolve_document_context( document_id, user_id, @@ -296,59 +434,45 @@ def resolve_document_context( active_group_ids=None, active_public_workspace_id=None, conversation_id=None, + include_content=True, ): normalized_scope = normalize_search_scope(doc_scope) if normalized_scope in ("all", "personal"): - personal_document = get_document_record(user_id=user_id, document_id=document_id) - if personal_document: - return { - "scope": "personal", - "group_id": None, - "public_workspace_id": None, - "document": personal_document, - } + personal_context = _resolve_personal_document_context(document_id, user_id) + if personal_context: + return personal_context if normalized_scope in ("all", "group"): - for group_id in _resolve_active_group_ids( + group_context = _resolve_group_document_context( + document_id, user_id, - active_group_ids=active_group_ids, - fallback_to_memberships=True, - ): - group_document = get_document_record( - user_id=user_id, - document_id=document_id, - group_id=group_id, - ) - if group_document: - return { - "scope": "group", - "group_id": group_id, - "public_workspace_id": None, - "document": group_document, - } + _resolve_active_group_ids( + user_id, + active_group_ids=active_group_ids, + fallback_to_memberships=True, + ), + ) + if group_context: + return group_context if normalized_scope in ("all", "public"): - for public_workspace_id in _resolve_public_workspace_ids( + public_context = _resolve_public_document_context( + document_id, user_id, - active_public_workspace_id=active_public_workspace_id, - ): - public_document = get_document_record( - user_id=user_id, - document_id=document_id, - public_workspace_id=public_workspace_id, - ) - if public_document: - return { - "scope": "public", - "group_id": None, - "public_workspace_id": public_workspace_id, - "document": public_document, - } + _resolve_public_workspace_ids( + user_id, + active_public_workspace_id=active_public_workspace_id, + ), + ) + if public_context: + return public_context chat_upload_context = _resolve_chat_upload_context( document_id=document_id, + user_id=user_id, conversation_id=conversation_id, + include_content=include_content, ) if chat_upload_context: return chat_upload_context @@ -356,6 +480,68 @@ def resolve_document_context( return None +def resolve_document_contexts( + document_ids, + user_id, + doc_scope="all", + active_group_ids=None, + active_public_workspace_id=None, + conversation_id=None, + include_content=True, +): + """Resolve ordered document contexts using one current authorization snapshot.""" + normalized_scope = normalize_search_scope(doc_scope) + normalized_document_ids = normalize_search_id_list(document_ids) + authorized_group_ids = [] + if normalized_scope in ("all", "group"): + authorized_group_ids = _resolve_active_group_ids( + user_id, + active_group_ids=active_group_ids, + fallback_to_memberships=True, + ) + authorized_public_workspace_ids = [] + if normalized_scope in ("all", "public"): + authorized_public_workspace_ids = _resolve_public_workspace_ids( + user_id, + active_public_workspace_id=active_public_workspace_id, + ) + + normalized_conversation_id = str(conversation_id or "").strip() + chat_conversation_authorized = bool( + normalized_conversation_id + and _authorize_chat_upload_conversation(user_id, normalized_conversation_id) + ) + + resolved_contexts = [] + for document_id in normalized_document_ids: + document_context = None + if normalized_scope in ("all", "personal"): + document_context = _resolve_personal_document_context(document_id, user_id) + if not document_context and normalized_scope in ("all", "group"): + document_context = _resolve_group_document_context( + document_id, + user_id, + authorized_group_ids, + ) + if not document_context and normalized_scope in ("all", "public"): + document_context = _resolve_public_document_context( + document_id, + user_id, + authorized_public_workspace_ids, + ) + if not document_context and chat_conversation_authorized: + document_context = _resolve_chat_upload_context( + document_id=document_id, + user_id=user_id, + conversation_id=normalized_conversation_id, + include_content=include_content, + authorization_prechecked=True, + ) + resolved_contexts.append(document_context) + + return resolved_contexts + + def build_search_request( query, user_id, diff --git a/application/single_app/functions_settings.py b/application/single_app/functions_settings.py index d54d3824f..6db9782cf 100644 --- a/application/single_app/functions_settings.py +++ b/application/single_app/functions_settings.py @@ -210,6 +210,11 @@ def is_tabular_processing_enabled(settings): return bool((settings or {}).get('enable_enhanced_citations', False)) +def is_mixed_source_manifest_enabled(settings): + """Return whether Phase 1 mixed-source manifest diagnostics are enabled.""" + return bool((settings or {}).get('enable_mixed_source_manifest', False)) + + CHAT_FILE_UPLOAD_APP_ROLE = "ChatFileUploadUser" WORKFLOW_USER_APP_ROLE = "WorkflowUser" DOCUMENT_INTELLIGENCE_PDF_IMAGE_EXTRACTION_MODES = {"read", "layout", "auto"} @@ -759,6 +764,7 @@ def get_settings(use_cosmos=False, include_source=False): 'enable_fact_memory_plugin': True, 'enable_tabular_processing_plugin': False, 'enable_multi_agent_orchestration': False, + 'enable_mixed_source_manifest': False, 'max_rounds_per_agent': 1, 'workflow_max_auto_invoke_attempts': 60, 'enable_semantic_kernel': False, diff --git a/application/single_app/functions_workflow_runner.py b/application/single_app/functions_workflow_runner.py index aff2f3848..4087561fa 100644 --- a/application/single_app/functions_workflow_runner.py +++ b/application/single_app/functions_workflow_runner.py @@ -78,6 +78,7 @@ build_agent_citation_artifact_documents, make_json_serializable, ) +from functions_mixed_source_orchestration import resolve_authorized_source_manifest from model_endpoint_clients import ( MODEL_ENDPOINT_PROTOCOL_AZURE_OPENAI, ) @@ -88,7 +89,14 @@ from functions_search_service import resolve_document_context, search_documents from functions_search import normalize_search_id_list, normalize_search_scope, normalize_search_top_n from functions_simplechat_operations import upload_generated_analysis_artifact_for_current_user -from functions_settings import get_settings, get_user_settings, is_tabular_processing_enabled, normalize_model_endpoints, resolve_model_endpoint_foundry_scope +from functions_settings import ( + get_settings, + get_user_settings, + is_mixed_source_manifest_enabled, + is_tabular_processing_enabled, + normalize_model_endpoints, + resolve_model_endpoint_foundry_scope, +) from functions_source_review import ( URL_ACCESS_CONTEXT_WORKFLOW, compact_source_review_result_for_metadata, @@ -1562,7 +1570,7 @@ def _normalize_tabular_source_hint(scope): return 'workspace' -def _resolve_tabular_document_action_documents(action_config, user_id, conversation_id=''): +def _get_document_action_source_ids(action_config): action_config = action_config if isinstance(action_config, dict) else {} action_type = str(action_config.get('type') or '').strip().lower() @@ -1588,6 +1596,16 @@ def _resolve_tabular_document_action_documents(action_config, user_id, conversat document_ids.append(document_id) role_by_document_id[document_id] = 'right' + return document_ids, role_by_document_id + + +def _resolve_tabular_document_action_documents( + action_config, + user_id, + conversation_id='', +): + action_config = action_config if isinstance(action_config, dict) else {} + document_ids, role_by_document_id = _get_document_action_source_ids(action_config) if not document_ids: return [] @@ -2002,13 +2020,36 @@ def _maybe_execute_tabular_document_action( ): if action_type not in {DOCUMENT_ACTION_TYPE_ANALYZE, DOCUMENT_ACTION_TYPE_COMPARISON}: return None - if not callable(invoke_prompt) or not is_tabular_processing_enabled(settings): - return None user_id = str(workflow.get('user_id') or '').strip() if not user_id: return None + if is_mixed_source_manifest_enabled(settings): + requested_source_ids, _ = _get_document_action_source_ids(action_config) + if requested_source_ids: + try: + resolve_authorized_source_manifest( + requested_source_ids, + user_id=user_id, + selection_mode='selected', + conversation_id=conversation_id, + active_group_ids=action_config.get('active_group_ids'), + active_public_workspace_ids=action_config.get('active_public_workspace_id'), + ) + except Exception: + log_event( + '[MixedSourceManifest] Workflow shadow resolution failed.', + extra={ + 'requested_source_count': len(requested_source_ids), + 'selection_mode': 'selected', + }, + level=logging.WARNING, + ) + + if not callable(invoke_prompt) or not is_tabular_processing_enabled(settings): + return None + tabular_documents = _resolve_tabular_document_action_documents( action_config, user_id, diff --git a/application/single_app/route_backend_chats.py b/application/single_app/route_backend_chats.py index 495bd9475..03c967be5 100644 --- a/application/single_app/route_backend_chats.py +++ b/application/single_app/route_backend_chats.py @@ -37,6 +37,7 @@ build_model_endpoint_sync_chat_client, build_semantic_kernel_chat_service_for_model, ) +from functions_mixed_source_orchestration import resolve_authorized_source_manifest import builtins import asyncio, types import ast @@ -289,6 +290,44 @@ def _normalize_capability_action(document_action_type): return ASSIGNED_KNOWLEDGE_USER_ACTION_SEARCH +def _maybe_resolve_chat_source_manifest( + settings, + user_id, + conversation_id, + selected_document_ids, + scope_context, +): + if not is_mixed_source_manifest_enabled(settings): + return [] + + requested_source_ids = _normalize_conversation_task_document_ids( + selected_document_ids + ) + if not requested_source_ids: + return [] + + scope_context = scope_context if isinstance(scope_context, dict) else {} + try: + return resolve_authorized_source_manifest( + requested_source_ids, + user_id=user_id, + selection_mode='selected', + conversation_id=conversation_id, + active_group_ids=scope_context.get('active_group_ids'), + active_public_workspace_ids=scope_context.get('active_public_workspace_ids'), + ) + except Exception: + log_event( + '[MixedSourceManifest] Chat shadow resolution failed.', + extra={ + 'requested_source_count': len(requested_source_ids), + 'selection_mode': 'selected', + }, + level=logging.WARNING, + ) + return [] + + def _source_review_metadata_used(source_review_result): if not isinstance(source_review_result, dict): return False @@ -13527,6 +13566,14 @@ def result_requires_message_reload(result: Any) -> bool: selected_document_id = effective_selected_document_id document_scope = effective_document_scope + _maybe_resolve_chat_source_manifest( + settings, + user_id, + conversation_id, + effective_selected_document_ids, + scope_context, + ) + # Clear plugin invocations at start of message processing to ensure # each message only shows citations for tools executed during that specific interaction plugin_logger = get_plugin_logger() @@ -17199,6 +17246,14 @@ def build_streaming_capability_usage(): selected_document_id = effective_selected_document_id document_scope = effective_document_scope + _maybe_resolve_chat_source_manifest( + settings, + user_id, + conversation_id, + effective_selected_document_ids, + scope_context, + ) + # Determine chat type actual_chat_type = 'personal_single_user' if conversation_item.get('chat_type'): diff --git a/docs/explanation/features/MIXED_SOURCE_MANIFEST_AND_EVIDENCE_CONTRACTS.md b/docs/explanation/features/MIXED_SOURCE_MANIFEST_AND_EVIDENCE_CONTRACTS.md new file mode 100644 index 000000000..d5d7aec74 --- /dev/null +++ b/docs/explanation/features/MIXED_SOURCE_MANIFEST_AND_EVIDENCE_CONTRACTS.md @@ -0,0 +1,115 @@ +# Mixed-Source Manifest and Evidence Contracts + +Implemented in version: **0.250.062** + +GitHub issue: [#1056](https://github.com/microsoft/simplechat/issues/1056) + +Parent initiative: [#1055](https://github.com/microsoft/simplechat/issues/1055) + +## Overview + +SimpleChat now has a shared, authorization-safe contract for describing mixed document selections before any processing engine is selected. The ordered manifest classifies each authorized source as tabular, narrative, or unsupported, while sources that cannot be resolved or authorized receive the same scrubbed unresolved shape. + +This is Phase 1 of the mixed-source orchestration initiative. It establishes internal contracts and diagnostics without changing Chat, Search, Analyze, Compare, conversation follow-up, or rollout behavior. + +## Purpose + +The manifest removes the need for later orchestration phases to repeatedly resolve the same document IDs into incompatible shapes. Its pure partition helper also preserves valid tabular and narrative cohorts when another selected source is unsupported or unresolved. + +The bounded evidence envelope gives later native engines one JSON-safe result shape without placing exhaustive rows or unbounded content into synthesis context. + +## Dependencies + +- Existing personal document ownership checks in `functions_documents.get_document_record(...)` +- Existing group membership checks used by `functions_search_service.resolve_document_context(...)` +- Existing public workspace visibility checks in `functions_public_workspaces.py` +- Personal conversation ownership checks for chat-upload message resolution +- Structured telemetry through `functions_appinsights.log_event(...)` + +No new authorization model, tabular runner, export subsystem, route, database container, or persisted migration is introduced. + +## Technical Specifications + +### Architecture + +`functions_mixed_source_orchestration.py` provides four internal contracts: + +- `resolve_authorized_source_manifest(...)` resolves each unique requested document ID once, preserves first-occurrence order, and ignores caller-supplied scope or identity metadata. +- `partition_source_manifest(...)` returns independent tabular, narrative, unsupported, and unresolved cohorts while preserving order inside each cohort. +- `normalize_selection_mode(...)` validates `selected`, `all`, `history`, and `relevance` modes for later phases. +- `build_evidence_envelope(...)` and `serialize_evidence_envelope(...)` validate engine/status values and enforce deterministic item, string, collection, and serialized-size limits. + +Authorized manifest entries include normalized document identity, display/file names, extension, source kind, canonical scope and scope ID, applicable group/public/conversation IDs, source version when available, and authorization status. + +Unresolved and unauthorized requests are deliberately indistinguishable. Their entries retain only the caller-requested document ID and return null source metadata with `source_kind` and `authorization_status` set to `unresolved`. + +### Authorization Boundaries + +- Personal sources are returned only when the current user owns the document or has an existing approved share. +- Group source candidates are restricted to current group memberships before document lookup. +- Public source candidates are restricted to currently visible public workspaces before document lookup. +- Chat-upload metadata is queried only after the personal conversation record is loaded and its owner matches the current user. The manifest query projects identity, filename/title, version, role, and inert artifact capability fields without loading embedded file content, extracted text, vision output, or blob data. +- Requested scope, scope IDs, owner IDs, group IDs, public workspace IDs, and conversation IDs embedded in source payloads are not accepted as authorization decisions. + +Current group memberships, public workspace visibility, and chat conversation ownership are resolved once per manifest request and reused for its bounded document lookups. Authorization is still revalidated on every new manifest request. + +Manifest diagnostics contain aggregate counts, scope distribution, duplicate count, error count, and resolution duration only. They do not contain document IDs, filenames, content, blob paths, credentials, or raw configuration. + +### Evidence Bounds + +The evidence envelope has a maximum serialized size of 65,536 bytes. Summary text, error text, collection counts, individual structured values, nesting depth, and coverage metadata are bounded independently. When limits are applied, coverage records `evidence_envelope_truncated`; exhaustive output remains the responsibility of generated artifacts or durable checkpoints. + +Source manifests accept at most 100 requested entries. Over-limit requests fail before document resolution and emit count-only diagnostics; sources are never silently truncated. + +### API Endpoints + +No API endpoints are added or changed in Phase 1. + +### Configuration Options + +- `enable_mixed_source_manifest`: internal, default-off flag for producing shadow manifests in Chat and workflow requests. + +The flag is intentionally not exposed in the admin UI in this phase. Disabling it restores the previous caller path with no data rollback because manifests are request-scoped and not persisted. + +### File Structure + +- `application/single_app/functions_mixed_source_orchestration.py` +- `application/single_app/functions_search_service.py` +- `application/single_app/functions_workflow_runner.py` +- `application/single_app/route_backend_chats.py` +- `functional_tests/test_mixed_source_manifest_contracts.py` +- `functional_tests/test_tabular_document_actions_workflow.py` + +## Usage Instructions + +This phase has no user workflow or UI changes. Internal callers may enable `enable_mixed_source_manifest` to produce authorization-safe shadow manifests for selected Chat or workflow sources while legacy execution remains unchanged. + +Later phases can consume the shared partition and evidence contracts instead of resolving document IDs again. They must continue to reauthorize sources at the object boundary and must not treat a persisted manifest as proof of current access. + +## Testing and Validation + +- Executable functional coverage: `functional_tests/test_mixed_source_manifest_contracts.py` +- Updated workflow regression: `functional_tests/test_tabular_document_actions_workflow.py` +- Coverage includes PDF plus XLSX, DOCX plus CSV in both orders, duplicate IDs, duplicate filenames across scopes, unresolved and unsupported sources among valid sources, personal/group/public/chat authorization, authorization loss, ordering, partitioning, evidence serialization/bounds, selection modes, and privacy-safe diagnostics. +- Python compilation, editor diagnostics, broken-access-control checks, XSS checks, route-policy checks, and whitespace validation are part of the Phase 1 validation gate. + +## Performance Considerations + +- Duplicate requested IDs are removed before resolution, preserving the first occurrence. +- Each unique requested ID is looked up once within a request-scoped authorization snapshot. +- Requests are capped at 100 source entries before authorization or document reads begin. +- Classification uses normalized metadata and does not read source content. +- Evidence bounding occurs before serialization so synthesis payloads remain predictable. + +## Known Limitations + +- Phase 1 does not activate document retrieval from explicit Chat selections. +- Phase 1 does not run mixed Analyze engines or synthesize their outputs. +- Phase 1 does not implement cross-format Compare. +- Phase 1 does not persist or reuse source context across follow-up turns. +- Phase 1 does not enumerate an Analyze All Documents catalog. +- Rollout and native-engine behavior changes remain scoped to #1057 through #1061. + +## Related Version Updates + +- `application/single_app/config.py` was updated from **0.250.061** to **0.250.062** for #1056. \ No newline at end of file diff --git a/docs/explanation/release_notes.md b/docs/explanation/release_notes.md index 5a25204aa..4281f0087 100644 --- a/docs/explanation/release_notes.md +++ b/docs/explanation/release_notes.md @@ -2,6 +2,16 @@ For feature-focused and fix-focused drill-downs by version, see [Features by Version](/explanation/features/) and [Fixes by Version](/explanation/fixes/). +### **(v0.250.062)** + +#### New Features + +* **Authorized Mixed-Source Manifest and Evidence Contracts** + * Added one ordered, authorization-safe source manifest that classifies selected personal, group, public, and chat-upload documents as tabular, narrative, unsupported, or unresolved without exposing inaccessible source metadata. + * Added independent capability partitions, validated selection modes, bounded engine-neutral evidence envelopes, and aggregate privacy-safe diagnostics for later Chat, Search, Analyze, and Compare phases. + * Kept Phase 1 behavior-neutral through a default-off internal shadow-manifest flag; native mixed-source execution remains scoped to follow-up issues #1057-#1061. + * (Ref: microsoft/simplechat#1056, parent microsoft/simplechat#1055, `functions_mixed_source_orchestration.py`, `MIXED_SOURCE_MANIFEST_AND_EVIDENCE_CONTRACTS.md`) + ### **(v0.250.061)** #### User Interface Enhancements diff --git a/functional_tests/test_mixed_source_manifest_contracts.py b/functional_tests/test_mixed_source_manifest_contracts.py new file mode 100644 index 000000000..42b10cf55 --- /dev/null +++ b/functional_tests/test_mixed_source_manifest_contracts.py @@ -0,0 +1,694 @@ +#!/usr/bin/env python3 +# test_mixed_source_manifest_contracts.py +""" +Functional test for authorized mixed-source manifest and evidence contracts. +Version: 0.250.062 +Implemented in: 0.250.062 + +This test ensures Phase 1 of #1056 resolves requested sources once through +current authorization boundaries, preserves ordering, partitions mixed source +types, and bounds engine-neutral evidence without implementing #1057-#1061. +Parent initiative: #1055. +""" + +import importlib.util +import json +import sys +import types +from pathlib import Path + +from azure.cosmos.exceptions import CosmosResourceNotFoundError + + +REPO_ROOT = Path(__file__).resolve().parents[1] +APP_ROOT = REPO_ROOT / "application" / "single_app" +SEARCH_SERVICE_PATH = APP_ROOT / "functions_search_service.py" +sys.path.insert(0, str(APP_ROOT)) + +import functions_mixed_source_orchestration as orchestration + +ORIGINAL_ORCHESTRATION_LOG_EVENT = orchestration.log_event + + +def setup_module(module=None): + orchestration.log_event = lambda *args, **kwargs: None + + +def teardown_module(module=None): + orchestration.log_event = ORIGINAL_ORCHESTRATION_LOG_EVENT + + +class FakeItemContainer: + def __init__(self, items=None): + self.items = dict(items or {}) + self.read_calls = [] + self.query_calls = [] + + def read_item(self, item, partition_key): + self.read_calls.append((partition_key, item)) + key = (partition_key, item) + if key not in self.items: + raise CosmosResourceNotFoundError(status_code=404, message="Not found") + return dict(self.items[key]) + + def query_items(self, query, parameters, partition_key): + self.query_calls.append({ + "query": query, + "parameters": list(parameters or []), + "partition_key": partition_key, + }) + parameter_values = { + parameter.get("name"): parameter.get("value") + for parameter in list(parameters or []) + } + document_id = parameter_values.get("@document_id") + message_item = self.items.get((partition_key, document_id)) + if not message_item: + return [] + metadata = message_item.get("metadata", {}) or {} + return [{ + "id": message_item.get("id"), + "role": message_item.get("role"), + "filename": message_item.get("filename"), + "title": message_item.get("title"), + "version": message_item.get("version"), + "is_user_upload": metadata.get("is_user_upload"), + "is_generated_chat_artifact": metadata.get("is_generated_chat_artifact"), + "generated_artifact_capability": metadata.get("generated_artifact_capability"), + "generated_artifact_output_format": metadata.get("generated_artifact_output_format"), + }] + + +def _normalize_id_list(values): + if values is None: + return [] + if isinstance(values, str): + values = [values] + normalized_values = [] + for value in list(values): + normalized_value = str(value or "").strip() + if normalized_value and normalized_value not in normalized_values: + normalized_values.append(normalized_value) + return normalized_values + + +def load_isolated_search_service(): + config_stub = types.ModuleType("config") + config_stub.CLIENTS = {} + config_stub.cognitive_services_scope = "https://example.invalid/.default" + config_stub.cosmos_conversations_container = FakeItemContainer() + config_stub.cosmos_messages_container = FakeItemContainer() + + appinsights_stub = types.ModuleType("functions_appinsights") + appinsights_stub.log_event = lambda *args, **kwargs: None + + debug_stub = types.ModuleType("functions_debug") + debug_stub.debug_print = lambda *args, **kwargs: None + + documents_stub = types.ModuleType("functions_documents") + documents_stub.get_document_record = lambda **kwargs: None + documents_stub.get_ordered_document_chunks = lambda **kwargs: [] + + group_stub = types.ModuleType("functions_group") + group_stub.get_user_groups = lambda user_id: [] + + public_stub = types.ModuleType("functions_public_workspaces") + public_stub.get_user_visible_public_workspace_ids_from_settings = lambda user_id: [] + + search_stub = types.ModuleType("functions_search") + search_stub.SEARCH_DEFAULT_TOP_N = 12 + search_stub.SEARCH_MAX_TOP_N = 500 + search_stub.hybrid_search = lambda **kwargs: [] + search_stub.normalize_search_id_list = _normalize_id_list + search_stub.normalize_search_scope = ( + lambda value: str(value or "all").strip().lower() + if str(value or "all").strip().lower() in {"all", "personal", "group", "public"} + else "all" + ) + search_stub.normalize_search_top_n = ( + lambda value, default_value, max_value: default_value if value is None else int(value) + ) + + settings_stub = types.ModuleType("functions_settings") + settings_stub.get_settings = lambda: {} + settings_stub.get_user_settings = lambda user_id: {"settings": {}} + + module_stubs = { + "config": config_stub, + "functions_appinsights": appinsights_stub, + "functions_debug": debug_stub, + "functions_documents": documents_stub, + "functions_group": group_stub, + "functions_public_workspaces": public_stub, + "functions_search": search_stub, + "functions_settings": settings_stub, + } + previous_modules = { + module_name: sys.modules.get(module_name) + for module_name in module_stubs + } + sys.modules.update(module_stubs) + + try: + module_spec = importlib.util.spec_from_file_location( + "functions_search_service_mixed_source_test", + SEARCH_SERVICE_PATH, + ) + search_service = importlib.util.module_from_spec(module_spec) + module_spec.loader.exec_module(search_service) + finally: + for module_name, previous_module in previous_modules.items(): + if previous_module is None: + sys.modules.pop(module_name, None) + else: + sys.modules[module_name] = previous_module + + return search_service + + +def build_authorized_resolver_fixture(): + search_service = load_isolated_search_service() + state = { + "group_ids": {"group-a"}, + "public_workspace_ids": {"public-a"}, + "group_authorization_count": 0, + "public_authorization_count": 0, + } + personal_documents = { + "personal-pdf": { + "id": "personal-pdf", + "user_id": "user-1", + "title": "Personal report", + "file_name": "report.pdf", + "version": 3, + }, + "personal-xlsx": { + "id": "personal-xlsx", + "user_id": "user-1", + "title": "Personal workbook", + "file_name": "data.xlsx", + "version": 4, + }, + "personal-docx": { + "id": "personal-docx", + "user_id": "user-1", + "title": "Personal narrative", + "file_name": "narrative.docx", + "version": 1, + }, + "personal-csv": { + "id": "personal-csv", + "user_id": "user-1", + "title": "Personal data", + "file_name": "data.csv", + "version": 2, + }, + "personal-unsupported": { + "id": "personal-unsupported", + "user_id": "user-1", + "title": "Unsupported archive", + "file_name": "archive.zip", + }, + } + group_documents = { + ("group-a", "group-csv"): { + "id": "group-csv", + "group_id": "group-a", + "title": "Shared group data", + "file_name": "shared.csv", + "version": 5, + }, + } + public_documents = { + ("public-a", "public-csv"): { + "id": "public-csv", + "public_workspace_id": "public-a", + "title": "Shared public data", + "file_name": "shared.csv", + "version": 6, + }, + } + + def get_document_record(user_id, document_id, group_id=None, public_workspace_id=None): + if group_id is not None: + return group_documents.get((group_id, document_id)) + if public_workspace_id is not None: + return public_documents.get((public_workspace_id, document_id)) + document_item = personal_documents.get(document_id) + if document_item and document_item.get("user_id") == user_id: + return dict(document_item) + return None + + search_service.get_document_record = get_document_record + def get_user_groups(user_id): + state["group_authorization_count"] += 1 + return ( + [{"id": group_id} for group_id in sorted(state["group_ids"])] + if user_id == "user-1" + else [] + ) + + def get_visible_public_workspace_ids(user_id): + state["public_authorization_count"] += 1 + return ( + sorted(state["public_workspace_ids"]) + if user_id == "user-1" + else [] + ) + + search_service.get_user_groups = get_user_groups + search_service.get_user_visible_public_workspace_ids_from_settings = ( + get_visible_public_workspace_ids + ) + search_service.get_user_settings = lambda user_id: {"settings": {}} + search_service.cosmos_conversations_container = FakeItemContainer({ + ("conversation-1", "conversation-1"): { + "id": "conversation-1", + "user_id": "user-1", + }, + }) + search_service.cosmos_messages_container = FakeItemContainer({ + ("conversation-1", "chat-csv"): { + "id": "chat-csv", + "role": "file", + "filename": "chat.csv", + "file_content": "name,value\nalpha,1", + }, + }) + + resolver_calls = [] + + def resolver(**resolver_arguments): + resolver_calls.append(resolver_arguments["document_id"]) + return search_service.resolve_document_context(**resolver_arguments) + + return search_service, state, resolver, resolver_calls + + +def resolve_manifest(document_ids, resolver, user_id="user-1", conversation_id="conversation-1"): + return orchestration.resolve_authorized_source_manifest( + document_ids, + user_id=user_id, + conversation_id=conversation_id, + context_resolver=resolver, + ) + + +def test_mixed_classification_order_and_partition(): + _, _, resolver, _ = build_authorized_resolver_fixture() + + pdf_xlsx_manifest = resolve_manifest( + ["personal-pdf", "personal-xlsx"], + resolver, + ) + assert [entry["document_id"] for entry in pdf_xlsx_manifest] == [ + "personal-pdf", + "personal-xlsx", + ] + assert [entry["source_kind"] for entry in pdf_xlsx_manifest] == [ + "narrative", + "tabular", + ] + + for document_ids in ( + ["personal-docx", "personal-csv"], + ["personal-csv", "personal-docx"], + ): + manifest = resolve_manifest(document_ids, resolver) + assert [entry["document_id"] for entry in manifest] == document_ids + partitions = orchestration.partition_source_manifest(manifest) + assert [entry["document_id"] for entry in partitions["tabular_sources"]] == [ + "personal-csv", + ] + assert [entry["document_id"] for entry in partitions["narrative_sources"]] == [ + "personal-docx", + ] + + +def test_duplicates_and_cross_scope_filename_identity(): + _, _, resolver, resolver_calls = build_authorized_resolver_fixture() + manifest = resolve_manifest( + [ + "personal-xlsx", + "personal-xlsx", + "group-csv", + "public-csv", + ], + resolver, + ) + + assert resolver_calls.count("personal-xlsx") == 1 + assert [entry["document_id"] for entry in manifest] == [ + "personal-xlsx", + "group-csv", + "public-csv", + ] + duplicate_name_entries = [ + entry for entry in manifest if entry["file_name"] == "shared.csv" + ] + assert len(duplicate_name_entries) == 2 + assert { + (entry["scope"], entry["scope_id"], entry["document_id"]) + for entry in duplicate_name_entries + } == { + ("group", "group-a", "group-csv"), + ("public", "public-a", "public-csv"), + } + + +def test_unresolved_and_unsupported_do_not_erase_valid_sources(): + _, _, resolver, _ = build_authorized_resolver_fixture() + manifest = resolve_manifest( + ["personal-csv", "missing-source", "personal-unsupported", "personal-pdf"], + resolver, + ) + partitions = orchestration.partition_source_manifest(manifest) + + assert [entry["document_id"] for entry in manifest] == [ + "personal-csv", + "missing-source", + "personal-unsupported", + "personal-pdf", + ] + assert [entry["document_id"] for entry in partitions["tabular_sources"]] == [ + "personal-csv", + ] + assert [entry["document_id"] for entry in partitions["narrative_sources"]] == [ + "personal-pdf", + ] + assert [entry["document_id"] for entry in partitions["unsupported_sources"]] == [ + "personal-unsupported", + ] + assert [entry["document_id"] for entry in partitions["unresolved_sources"]] == [ + "missing-source", + ] + unresolved_entry = partitions["unresolved_sources"][0] + assert unresolved_entry["authorization_status"] == "unresolved" + assert unresolved_entry["file_name"] is None + assert unresolved_entry["scope"] is None + assert unresolved_entry["scope_id"] is None + + +def test_personal_group_public_and_chat_authorization(): + search_service, state, resolver, _ = build_authorized_resolver_fixture() + manifest = resolve_manifest( + ["personal-pdf", "group-csv", "public-csv", "chat-csv"], + resolver, + ) + assert [entry["scope"] for entry in manifest] == [ + "personal", + "group", + "public", + "chat", + ] + assert manifest[3]["conversation_id"] == "conversation-1" + assert manifest[3]["scope_id"] == "conversation-1" + + original_search_service_module = sys.modules.get("functions_search_service") + original_content_coercer = search_service._coerce_chat_upload_text + search_service.cosmos_messages_container.read_calls.clear() + search_service.cosmos_messages_container.query_calls.clear() + search_service._coerce_chat_upload_text = lambda message_item: (_ for _ in ()).throw( + AssertionError("Manifest resolution must not load chat-upload content") + ) + sys.modules["functions_search_service"] = search_service + try: + metadata_only_chat_manifest = orchestration.resolve_authorized_source_manifest( + ["chat-csv"], + user_id="user-1", + conversation_id="conversation-1", + ) + finally: + search_service._coerce_chat_upload_text = original_content_coercer + if original_search_service_module is None: + sys.modules.pop("functions_search_service", None) + else: + sys.modules["functions_search_service"] = original_search_service_module + assert metadata_only_chat_manifest[0]["source_kind"] == "tabular" + assert metadata_only_chat_manifest[0]["authorization_status"] == "authorized" + assert search_service.cosmos_messages_container.read_calls == [] + assert len(search_service.cosmos_messages_container.query_calls) == 1 + assert "c.file_content" not in search_service.cosmos_messages_container.query_calls[0]["query"] + assert "c.extracted_text" not in search_service.cosmos_messages_container.query_calls[0]["query"] + + caller_scope_payload = [{ + "document_id": "personal-pdf", + "scope": "public", + "public_workspace_id": "caller-controlled-workspace", + }] + caller_scope_manifest = resolve_manifest(caller_scope_payload, resolver) + assert caller_scope_manifest[0]["scope"] == "personal" + assert caller_scope_manifest[0]["public_workspace_id"] is None + + state["group_ids"].clear() + state["public_workspace_ids"].clear() + search_service.cosmos_conversations_container.items[ + ("conversation-1", "conversation-1") + ]["user_id"] = "different-user" + search_service.cosmos_messages_container.read_calls.clear() + search_service.cosmos_messages_container.query_calls.clear() + + authorization_loss_manifest = resolve_manifest( + ["group-csv", "public-csv", "chat-csv"], + resolver, + ) + assert all( + entry["source_kind"] == "unresolved" + and entry["authorization_status"] == "unresolved" + and entry["file_name"] is None + and entry["scope"] is None + for entry in authorization_loss_manifest + ) + assert search_service.cosmos_messages_container.read_calls == [] + assert search_service.cosmos_messages_container.query_calls == [] + + personal_authorization_loss = resolve_manifest( + ["personal-pdf"], + resolver, + user_id="different-user", + ) + assert personal_authorization_loss[0]["source_kind"] == "unresolved" + assert personal_authorization_loss[0]["display_name"] is None + + +def test_selection_mode_normalization(): + for selection_mode in ("selected", "all", "history", "relevance"): + assert orchestration.normalize_selection_mode( + f" {selection_mode.upper()} " + ) == selection_mode + assert orchestration.normalize_selection_mode(None) == "selected" + + try: + orchestration.normalize_selection_mode("everything") + except ValueError: + pass + else: + raise AssertionError("Invalid selection_mode must fail validation") + + +def test_batch_authorization_snapshot_and_source_limit(): + search_service, state, _, _ = build_authorized_resolver_fixture() + state["group_authorization_count"] = 0 + state["public_authorization_count"] = 0 + search_service.cosmos_conversations_container.read_calls.clear() + + original_search_service_module = sys.modules.get("functions_search_service") + sys.modules["functions_search_service"] = search_service + try: + manifest = orchestration.resolve_authorized_source_manifest( + ["personal-pdf", "group-csv", "public-csv", "chat-csv"], + user_id="user-1", + conversation_id="conversation-1", + ) + finally: + if original_search_service_module is None: + sys.modules.pop("functions_search_service", None) + else: + sys.modules["functions_search_service"] = original_search_service_module + + assert [entry["scope"] for entry in manifest] == [ + "personal", + "group", + "public", + "chat", + ] + assert state["group_authorization_count"] == 1 + assert state["public_authorization_count"] == 1 + assert search_service.cosmos_conversations_container.read_calls == [ + ("conversation-1", "conversation-1"), + ] + + over_limit_resolver_calls = [] + over_limit_sources = [ + f"source-{source_index}" + for source_index in range(orchestration.SOURCE_MANIFEST_MAX_SOURCES + 1) + ] + try: + orchestration.resolve_authorized_source_manifest( + over_limit_sources, + user_id="user-1", + context_resolver=lambda **kwargs: over_limit_resolver_calls.append(kwargs), + ) + except ValueError: + pass + else: + raise AssertionError("Over-limit source manifests must fail validation") + assert over_limit_resolver_calls == [] + + +def test_evidence_envelope_serialization_and_bounds(): + oversized_item = { + "rows": [ + {"column": "x" * 5000, "value": row_number} + for row_number in range(50) + ] + } + envelope = orchestration.build_evidence_envelope( + document_id="personal-xlsx", + source_kind="tabular", + engine="tabular_tools", + status="partial", + summary="s" * 20000, + evidence=[oversized_item for _ in range(25)], + citations=[oversized_item for _ in range(25)], + generated_artifacts=[oversized_item for _ in range(25)], + coverage={"requested_rows": 1000000, "processed_rows": 500000}, + error="e" * 5000, + ) + serialized_envelope = orchestration.serialize_evidence_envelope(envelope) + round_tripped_envelope = json.loads(serialized_envelope) + + assert len(serialized_envelope.encode("utf-8")) <= ( + orchestration.EVIDENCE_ENVELOPE_MAX_BYTES + ) + assert len(round_tripped_envelope["evidence"]) <= ( + orchestration.EVIDENCE_LIST_MAX_ITEMS + ) + assert len(round_tripped_envelope["citations"]) <= ( + orchestration.EVIDENCE_LIST_MAX_ITEMS + ) + assert len(round_tripped_envelope["generated_artifacts"]) <= ( + orchestration.EVIDENCE_LIST_MAX_ITEMS + ) + assert round_tripped_envelope["coverage"]["evidence_envelope_truncated"] is True + assert round_tripped_envelope["summary"].endswith("...") + assert round_tripped_envelope["error"].endswith("...") + + direct_envelope = { + "document_id": "personal-xlsx", + "source_kind": "tabular", + "engine": "tabular_tools", + "status": "completed", + "summary": "direct", + "evidence": [ + {"score": float("nan"), "value": item_number} + for item_number in range(orchestration.EVIDENCE_LIST_MAX_ITEMS + 5) + ], + "citations": [], + "generated_artifacts": [], + "coverage": {}, + "error": None, + } + direct_serialized = orchestration.serialize_evidence_envelope(direct_envelope) + direct_round_trip = json.loads(direct_serialized) + assert len(direct_round_trip["evidence"]) == orchestration.EVIDENCE_LIST_MAX_ITEMS + assert direct_round_trip["evidence"][0]["score"] is None + assert direct_round_trip["coverage"]["evidence_envelope_truncated"] is True + assert "NaN" not in direct_serialized + + nested_bound_envelope = orchestration.build_evidence_envelope( + document_id="personal-xlsx", + source_kind="tabular", + engine="tabular_tools", + status="completed", + evidence=[{ + "values": list( + range(orchestration.EVIDENCE_JSON_MAX_COLLECTION_ITEMS + 1) + ), + }], + coverage={"bounded": True}, + ) + assert len(nested_bound_envelope["evidence"][0]["values"]) == ( + orchestration.EVIDENCE_JSON_MAX_COLLECTION_ITEMS + ) + assert nested_bound_envelope["coverage"]["evidence_envelope_truncated"] is True + assert len(json.dumps(nested_bound_envelope["coverage"]).encode("utf-8")) <= ( + orchestration.EVIDENCE_COVERAGE_MAX_BYTES + ) + + try: + orchestration.serialize_evidence_envelope({ + **direct_envelope, + "extra_content": "not part of the contract", + }) + except ValueError: + pass + else: + raise AssertionError("Evidence serializer must reject undeclared fields") + + +def test_manifest_diagnostics_are_aggregate_only(): + _, _, resolver, _ = build_authorized_resolver_fixture() + captured_events = [] + original_log_event = orchestration.log_event + orchestration.log_event = lambda message, **kwargs: captured_events.append( + {"message": message, **kwargs} + ) + try: + resolve_manifest( + ["personal-pdf", "personal-pdf", "group-csv", "missing-source"], + resolver, + ) + finally: + orchestration.log_event = original_log_event + + assert len(captured_events) == 1 + diagnostics = captured_events[0]["extra"] + assert diagnostics["requested_source_count"] == 4 + assert diagnostics["unique_source_count"] == 3 + assert diagnostics["duplicate_ids_removed"] == 1 + assert diagnostics["narrative_source_count"] == 1 + assert diagnostics["tabular_source_count"] == 1 + assert diagnostics["unresolved_or_unauthorized_count"] == 1 + serialized_diagnostics = json.dumps(diagnostics, sort_keys=True) + for sensitive_value in ( + "personal-pdf", + "group-csv", + "missing-source", + "report.pdf", + "shared.csv", + "conversation-1", + ): + assert sensitive_value not in serialized_diagnostics + + +def run_tests(): + tests = [ + test_mixed_classification_order_and_partition, + test_duplicates_and_cross_scope_filename_identity, + test_unresolved_and_unsupported_do_not_erase_valid_sources, + test_personal_group_public_and_chat_authorization, + test_selection_mode_normalization, + test_batch_authorization_snapshot_and_source_limit, + test_evidence_envelope_serialization_and_bounds, + test_manifest_diagnostics_are_aggregate_only, + ] + results = [] + setup_module() + try: + for test in tests: + try: + test() + print(f"PASS {test.__name__}") + results.append(True) + except Exception as exc: + print(f"FAIL {test.__name__}: {exc}") + results.append(False) + finally: + teardown_module() + print(f"Results: {sum(results)}/{len(results)} tests passed") + return all(results) + + +if __name__ == "__main__": + raise SystemExit(0 if run_tests() else 1) \ No newline at end of file diff --git a/functional_tests/test_tabular_document_actions_workflow.py b/functional_tests/test_tabular_document_actions_workflow.py index e08f554ad..eed5c9140 100644 --- a/functional_tests/test_tabular_document_actions_workflow.py +++ b/functional_tests/test_tabular_document_actions_workflow.py @@ -2,21 +2,39 @@ # test_tabular_document_actions_workflow.py """ Functional test for tabular document-action workflow support. -Version: 0.241.038 -Implemented in: 0.241.038 +Version: 0.250.062 +Implemented in: 0.241.038; mixed-source manifest coverage added in 0.250.062 This test ensures tabular document actions reuse the shared tabular analysis path for Analyze and comparison workflows instead of relying only on the search-grounded chat path, including row-linked related-document evidence and -live tabular activity thoughts. +live tabular activity thoughts. It also ensures the Phase 1 contract from +#1056 preserves valid tabular sources in a mixed selection. Parent: #1055. """ +import ast +import logging from pathlib import Path +import sys import traceback ROOT = Path(__file__).resolve().parents[1] +APP_ROOT = ROOT / "application" / "single_app" WORKFLOW_RUNNER_FILE = ROOT / "application" / "single_app" / "functions_workflow_runner.py" +sys.path.insert(0, str(APP_ROOT)) + +import functions_mixed_source_orchestration as orchestration + +ORIGINAL_ORCHESTRATION_LOG_EVENT = orchestration.log_event + + +def setup_module(module=None): + orchestration.log_event = lambda *args, **kwargs: None + + +def teardown_module(module=None): + orchestration.log_event = ORIGINAL_ORCHESTRATION_LOG_EVENT def read_text(path: Path) -> str: @@ -37,6 +55,12 @@ def test_shared_tabular_document_action_helper_exists() -> None: assert 'def _resolve_tabular_document_action_documents(' in workflow_runner_content, ( "Expected functions_workflow_runner.py to resolve selected tabular documents before dispatching analysis or comparison." ) + assert 'resolve_authorized_source_manifest(' in workflow_runner_content, ( + "Expected workflow document actions to support the authorized Phase 1 source manifest." + ) + assert 'is_mixed_source_manifest_enabled(settings)' in workflow_runner_content, ( + "Expected Phase 1 workflow manifest production to remain behind its internal flag." + ) assert 'augment_tabular_invocations_with_related_document_evidence(' in workflow_runner_content, ( "Expected the shared helper to reuse row-linked related-document augmentation for tabular workflows." ) @@ -98,24 +122,184 @@ def test_tabular_document_actions_stream_live_activity() -> None: print("Tabular document-action live thought plumbing checks passed") +def test_mixed_sources_preserve_valid_tabular_partition() -> None: + print("Testing mixed-source tabular partition behavior...") + + source_records = { + "narrative-doc": { + "scope": "personal", + "document": { + "id": "narrative-doc", + "user_id": "user-1", + "title": "Narrative", + "file_name": "narrative.docx", + }, + }, + "tabular-doc": { + "scope": "personal", + "document": { + "id": "tabular-doc", + "user_id": "user-1", + "title": "Table", + "file_name": "table.csv", + }, + }, + } + resolver = lambda **kwargs: source_records.get(kwargs["document_id"]) + manifest = orchestration.resolve_authorized_source_manifest( + ["narrative-doc", "tabular-doc"], + user_id="user-1", + context_resolver=resolver, + ) + partitions = orchestration.partition_source_manifest(manifest) + + assert [entry["document_id"] for entry in manifest] == [ + "narrative-doc", + "tabular-doc", + ] + assert [entry["document_id"] for entry in partitions["narrative_sources"]] == [ + "narrative-doc", + ] + assert [entry["document_id"] for entry in partitions["tabular_sources"]] == [ + "tabular-doc", + ] + + print("Mixed-source tabular partition checks passed") + + +def test_manifest_flag_does_not_change_workflow_dispatch() -> None: + print("Testing workflow manifest flag behavior equivalence...") + + workflow_runner_tree = ast.parse(read_text(WORKFLOW_RUNNER_FILE)) + helper_node = next( + node + for node in workflow_runner_tree.body + if isinstance(node, ast.FunctionDef) + and node.name == "_maybe_execute_tabular_document_action" + ) + helper_module = ast.Module(body=[helper_node], type_ignores=[]) + ast.fix_missing_locations(helper_module) + + legacy_resolver_calls = [] + manifest_calls = [] + namespace = { + "DOCUMENT_ACTION_TYPE_ANALYZE": "analyze", + "DOCUMENT_ACTION_TYPE_COMPARISON": "comparison", + "is_tabular_processing_enabled": lambda settings: True, + "is_mixed_source_manifest_enabled": lambda settings: bool( + settings.get("enable_mixed_source_manifest") + ), + "_get_document_action_source_ids": lambda action_config: ( + list(action_config.get("document_ids") or []), + {}, + ), + "resolve_authorized_source_manifest": lambda *args, **kwargs: ( + manifest_calls.append((args, kwargs)) or [] + ), + "_resolve_tabular_document_action_documents": lambda *args, **kwargs: ( + legacy_resolver_calls.append((args, kwargs)) or [{"document_id": "table-1"}] + ), + "_resolve_tabular_document_action_model_name": lambda workflow, settings: "", + "log_event": lambda *args, **kwargs: None, + "logging": logging, + } + exec(compile(helper_module, str(WORKFLOW_RUNNER_FILE), "exec"), namespace) + helper = namespace["_maybe_execute_tabular_document_action"] + action_config = {"type": "analyze", "document_ids": ["table-1"]} + workflow = {"user_id": "user-1"} + + disabled_result = helper( + "analyze", + workflow, + action_config, + {"enable_mixed_source_manifest": False}, + conversation_id="conversation-1", + invoke_prompt=lambda *args, **kwargs: None, + ) + disabled_legacy_call = legacy_resolver_calls[-1] + assert manifest_calls == [] + + enabled_result = helper( + "analyze", + workflow, + action_config, + {"enable_mixed_source_manifest": True}, + conversation_id="conversation-1", + invoke_prompt=lambda *args, **kwargs: None, + ) + enabled_legacy_call = legacy_resolver_calls[-1] + + assert disabled_result == enabled_result is None + assert disabled_legacy_call == enabled_legacy_call + assert len(manifest_calls) == 1 + assert manifest_calls[0][0] == (["table-1"],) + + namespace["is_tabular_processing_enabled"] = lambda settings: False + manifest_calls.clear() + legacy_resolver_calls.clear() + disabled_tabular_result = helper( + "analyze", + workflow, + action_config, + {"enable_mixed_source_manifest": True}, + conversation_id="conversation-1", + invoke_prompt=lambda *args, **kwargs: None, + ) + assert disabled_tabular_result is None + assert len(manifest_calls) == 1 + assert legacy_resolver_calls == [] + + print("Workflow manifest flag behavior equivalence checks passed") + + +def test_document_action_chat_does_not_duplicate_shadow_manifest() -> None: + print("Testing document-action Chat manifest ownership...") + + route_tree = ast.parse( + read_text(ROOT / "application" / "single_app" / "route_backend_chats.py") + ) + document_action_function = next( + node + for node in ast.walk(route_tree) + if isinstance(node, ast.FunctionDef) + and node.name == "execute_document_action_chat_request" + ) + manifest_calls = [ + node + for node in ast.walk(document_action_function) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "_maybe_resolve_chat_source_manifest" + ] + assert manifest_calls == [] + + print("Document-action Chat manifest ownership checks passed") + + def run_tests() -> bool: tests = [ test_shared_tabular_document_action_helper_exists, test_analyze_and_compare_dispatch_use_tabular_helper, test_tabular_document_actions_stream_live_activity, + test_mixed_sources_preserve_valid_tabular_partition, + test_manifest_flag_does_not_change_workflow_dispatch, + test_document_action_chat_does_not_duplicate_shadow_manifest, ] results = [] - - for test in tests: - print(f"\nRunning {test.__name__}...") - try: - test() - print("PASS") - results.append(True) - except Exception as exc: - print(f"FAIL: {exc}") - traceback.print_exc() - results.append(False) + setup_module() + try: + for test in tests: + print(f"\nRunning {test.__name__}...") + try: + test() + print("PASS") + results.append(True) + except Exception as exc: + print(f"FAIL: {exc}") + traceback.print_exc() + results.append(False) + finally: + teardown_module() success = all(results) print(f"\nResults: {sum(results)}/{len(results)} tests passed")