M4C-2: publish exact saved workflow and Collect outputs

application/single_app/config.py

@@ -97,7 +97,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.118"
+VERSION = "0.261.119"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax')

application/single_app/content_screening/access.py

@@ -651,7 +651,11 @@ def public_history_messages(messages, user_id=None):
         try:
             refreshed = refresh_workspace_attachment(message, user_id)
             assert_evidence_available(refreshed, user_id, cached=True)
-            safe_messages.append(deepcopy(refreshed))
+            # The shared generated-file source dispatcher keeps private saved-output
+            # cards from bypassing the same boundary used by their downloads.
+            from functions_generated_artifact_sources import sanitize_generated_artifact_history
+
+            safe_messages.append(deepcopy(sanitize_generated_artifact_history(refreshed, user_id)))
         except (ScreeningError, LookupError, PermissionError):
             if request_context:
                 flask.g.content_screening_error = previous_error

application/single_app/functions_generated_artifact_sources.py

@@ -0,0 +1,159 @@
+# functions_generated_artifact_sources.py
+"""Shared source dispatch for generated files, independent of their renderer."""
+
+from copy import deepcopy
+import uuid
+
+from azure.core.exceptions import AzureError
+from flask import g, has_request_context
+
+from content_screening.contracts import ScreeningError
+from functions_analysis_access import AnalysisResultUnavailable
+from functions_appinsights import log_event
+from functions_workflow_result_store import WorkflowResultStorageUnavailableError
+from functions_workflow_runtime_store import RuntimeUnavailable
+
+
+_HISTORY_SOURCE_ERRORS = (PermissionError, LookupError, ValueError, AzureError,
+                          WorkflowResultStorageUnavailableError, RuntimeUnavailable, ScreeningError)
+_UNAVAILABLE_HISTORY = "Saved workflow output is unavailable because current access could not be confirmed."
+
+
+def has_generated_artifact_source(metadata):
+    return (
+        "generated_artifact_source" in metadata
+        or bool(metadata.get("generated_artifact_source_required"))
+        or str(metadata.get("generated_artifact_idempotency_key") or "").startswith("generated-export:v1:")
+    )
+
+
+def generated_chat_artifact_address(owner_id, conversation_id, file_name, idempotency_key, blob_container):
+    suffix = uuid.uuid5(
+        uuid.NAMESPACE_URL, f"simplechat-generated-artifact:{conversation_id}:{idempotency_key}",
+    ).hex if idempotency_key else uuid.uuid4().hex
+    message_id = f"{conversation_id}_generated_file_{suffix}"
+    return {
+        "conversation_id": conversation_id, "artifact_message_id": message_id,
+        "file_name": file_name, "blob_container": blob_container,
+        "blob_path": f"{owner_id}/{conversation_id}/generated/{message_id}/{file_name}",
+    }
+
+
+def generated_artifact_source_metadata(source):
+    # Workflow stores are loaded only for an explicit workflow-source adapter.
+    from functions_workflow_artifacts import validate_workflow_artifact_binding
+
+    binding = validate_workflow_artifact_binding(source)
+    return {"generated_artifact_source_required": True, "generated_artifact_source": deepcopy(binding)}
+
+
+def authorize_generated_artifact_preparation(user_id, metadata):
+    from functions_workflow_artifacts import load_workflow_artifact_binding
+
+    if metadata.get("analysis_result_required") or metadata.get("analysis_producer"):
+        raise AnalysisResultUnavailable("generated_artifact_source_conflict")
+    return load_workflow_artifact_binding(
+        user_id, metadata.get("generated_artifact_source"), require_ready=False, for_publication=True,
+    )
+
+
+def authorize_generated_artifact_source(user_id, artifact, *, for_publication=False, native_authorizer=None):
+    metadata = artifact.get("metadata") or {}
+    if has_generated_artifact_source(metadata):
+        from functions_workflow_artifacts import authorize_workflow_saved_output_artifact
+
+        if metadata.get("analysis_result_required") or metadata.get("analysis_producer"):
+            raise AnalysisResultUnavailable("generated_artifact_source_conflict")
+        if metadata.get("generated_artifact_source_required") is not True:
+            raise AnalysisResultUnavailable("generated_artifact_source_unbound")
+        return authorize_workflow_saved_output_artifact(user_id, artifact, for_publication=for_publication)
+    if native_authorizer is None:
+        # Native sources retain their existing authorization and eligibility contract.
+        from functions_saved_analysis import authorize_analysis_artifact
+
+        native_authorizer = authorize_analysis_artifact
+    return native_authorizer(user_id, artifact, **({"for_publication": True} if for_publication else {}))
+
+
+def sanitize_generated_artifact_history(message, user_id):
+    """Reauthorize saved-output cards and strip private bindings from history."""
+    metadata = message.get("metadata") or {}
+    is_file = message.get("role") == "file" and has_generated_artifact_source(metadata)
+    fields = ("generated_analysis_artifacts", "generated_tabular_outputs")
+    cards = [
+        item for field in fields for item in metadata.get(field) or []
+        if isinstance(item, dict) and item.get("source_kind") == "workflow_saved_output"
+    ]
+    if not is_file and not cards:
+        return message
+    # Reuse the complete conversation/approval/source/screening boundary, not just an opaque id.
+    from route_enhanced_citations import _get_authorized_chat_artifact_message
+
+    def authorize(conversation_id, message_id):
+        request_context = has_request_context()
+        previous_error = getattr(g, "content_screening_error", None) if request_context else None
+        previous_sources = dict(getattr(g, "content_screening_sources", {}) or {}) if request_context else {}
+        try:
+            return _get_authorized_chat_artifact_message(user_id, conversation_id, message_id)
+        except _HISTORY_SOURCE_ERRORS:
+            if request_context:
+                g.content_screening_error = previous_error
+                g.content_screening_sources = previous_sources
+            raise
+
+    def log_unavailable(exc):
+        log_event(
+            "[SIMPLE_CHAT] Saved-output artifact withheld on history read",
+            {"message_id": message.get("id"), "exception_type": type(exc).__name__},
+        )
+
+    safe = deepcopy(message)
+    if is_file:
+        try:
+            authorize(message.get("conversation_id"), message.get("id"))
+        except _HISTORY_SOURCE_ERRORS as exc:
+            log_unavailable(exc)
+            return {
+                **{key: deepcopy(message[key]) for key in (
+                    "id", "conversation_id", "timestamp", "created_at", "updated_at", "thread_id", "active_thread",
+                ) if key in message},
+                "role": "file", "content": _UNAVAILABLE_HISTORY, "content_unavailable": True,
+                "file_content": "", "extracted_text": "",
+            }
+        safe = {key: value for key, value in safe.items() if key in {
+            "id", "conversation_id", "role", "filename", "file_name", "content",
+            "timestamp", "created_at", "updated_at", "thread_id", "active_thread",
+        }}
+        safe["metadata"] = {
+            key: value for key, value in metadata.items() if key in {
+                "is_generated_chat_artifact", "generated_artifact_capability", "generated_artifact_output_format",
+                "generated_artifact_summary", "thread_info",
+            }
+        }
+        return safe
+    for field in fields:
+        if field not in metadata:
+            continue
+        projected = []
+        for card in metadata[field]:
+            if not isinstance(card, dict) or card.get("source_kind") != "workflow_saved_output":
+                projected.append(deepcopy(card))
+                continue
+            try:
+                if card.get("conversation_id") != message.get("conversation_id"):
+                    raise AnalysisResultUnavailable("generated_artifact_source_unbound")
+                authorize(card["conversation_id"], card.get("artifact_message_id"))
+            except _HISTORY_SOURCE_ERRORS as exc:
+                log_unavailable(exc)
+                projected.append({
+                    "capability": "file_export", "source_kind": "workflow_saved_output",
+                    "output_format": "json", "status": "unavailable",
+                    "summary": _UNAVAILABLE_HISTORY,
+                })
+            else:
+                projected.append({key: deepcopy(value) for key, value in card.items() if key in {
+                    "capability", "source_kind", "artifact_message_id", "conversation_id", "storage_scope",
+                    "file_name", "output_format", "summary", "suppress_assistant_text", "row_count", "row_source",
+                }})
+        safe["metadata"][field] = projected
+    return safe

application/single_app/functions_generated_file_exports.py

@@ -1,14 +1,16 @@
 # functions_generated_file_exports.py
 """Format-neutral planning and rendering for generated chat file exports."""
 
+import hashlib
 import html
 import io
 import json
 import os
 import re
 import tempfile
+from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple
+from typing import Any, BinaryIO, Callable, Dict, Iterable, Iterator, List, Optional, Protocol, Sequence, Tuple, overload
 from xml.etree import ElementTree
 
 from defusedxml import ElementTree as DefusedElementTree
@@ -36,6 +38,127 @@
 ASSISTANT_TEXT_SUPPRESSING_FORMATS = {'json', 'xml'}
 GENERATED_FILE_PREVIEW_ROWS = 3
 REQUESTED_ARTIFACT_FORMATS = ('csv', 'json', 'xml', 'md', 'docx', 'pdf')
+GENERATED_RECORD_EXPORT_FORMATS = {'exact_records_v1': ('json',)}
+
+
+@dataclass(frozen=True)
+class GeneratedFileExportRequest:
+    output_format: str
+    profile: str = 'exact_records_v1'
+
+
+class GeneratedRecordExportSource(Protocol):
+    kind: str
+    record_count: int
+
+    def iter_records(self) -> Iterator[Dict[str, Any]]: ...
+
+    def recheck(self) -> None: ...
+
+
+@dataclass
+class GeneratedFileExportStream:
+    file_content: BinaryIO
+    output_format: str
+    media_type: str
+    size_bytes: int
+    content_sha256: str
+    record_count: int
+    profile: str
+
+    def close(self) -> None:
+        self.file_content.close()
+
+    def __enter__(self) -> "GeneratedFileExportStream":
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback) -> None:
+        self.close()
+
+
+def _validate_exact_json_value(value):
+    if type(value) is dict:
+        for key, child in value.items():
+            if type(key) is not str:
+                raise ValueError('Saved record object keys must be strings.')
+            _validate_exact_json_value(child)
+    elif type(value) is list:
+        for child in value:
+            _validate_exact_json_value(child)
+    elif type(value) not in {str, int, float, bool, type(None)}:
+        raise ValueError('Saved records must contain only finite JSON values.')
+
+
+def _build_generated_record_export(
+    source: GeneratedRecordExportSource, request: GeneratedFileExportRequest, *,
+    max_output_bytes: int, check: Optional[Callable[[], Any]],
+) -> GeneratedFileExportStream:
+    if (
+        not isinstance(request, GeneratedFileExportRequest)
+        or request.output_format not in GENERATED_RECORD_EXPORT_FORMATS.get(request.profile, ())
+        or source.kind != 'records'
+    ):
+        raise ValueError('This saved-output source and export format are not supported.')
+    if type(source.record_count) is not int or source.record_count < 0:
+        raise ValueError('The saved record count is invalid.')
+    if type(max_output_bytes) is not int or max_output_bytes < 1:
+        raise ValueError('A positive saved-output byte limit is required.')
+    stream = tempfile.TemporaryFile(mode='w+b')
+    digest = hashlib.sha256()
+    size = 0
+    checked_size = 0
+    count = 0
+    completed = False
+    encoder = json.JSONEncoder(sort_keys=True, separators=(',', ':'), ensure_ascii=True, allow_nan=False)
+
+    def write(fragment):
+        nonlocal size, checked_size
+        for offset in range(0, len(fragment), 65536):
+            chunk = fragment[offset:offset + 65536].encode('ascii')
+            if size + len(chunk) > max_output_bytes:
+                raise ValueError('The complete saved-output file exceeds the configured artifact size limit.')
+            stream.write(chunk)
+            digest.update(chunk)
+            size += len(chunk)
+            if check is not None and size - checked_size >= 65536:
+                check()
+                checked_size = size
+
+    try:
+        if check is not None:
+            check()
+        source.recheck()
+        write('[')
+        for record in source.iter_records():
+            if type(record) is not dict or count >= source.record_count:
+                raise ValueError('The saved record collection does not match its declared shape or count.')
+            _validate_exact_json_value(record)
+            if count:
+                write(',')
+            for fragment in encoder.iterencode(record):
+                write(fragment)
+            count += 1
+            if check is not None and count % 100 == 0:
+                check()
+        if count != source.record_count:
+            raise ValueError('The complete saved record count does not match the exported file.')
+        write(']')
+        source.recheck()
+        if check is not None:
+            check()
+        stream.seek(0)
+        completed = True
+        return GeneratedFileExportStream(
+            file_content=stream, output_format=request.output_format, media_type='application/json',
+            size_bytes=size, content_sha256=digest.hexdigest(), record_count=count, profile=request.profile,
+        )
+    except (TypeError, RecursionError) as exc:
+        raise ValueError('Saved records must contain only finite, bounded JSON values.') from exc
+    finally:
+        if not completed:
+            stream.close()
+
+
 STRUCTURED_ARTIFACT_FORMAT_MARKERS = {
     'json': (
         'json artifact',
@@ -626,15 +749,55 @@ def build_saved_analysis_export(analysis_result, output_format):
     }
 
 
+@overload
+def build_generated_file_export(
+    user_question: str = '',
+    assistant_content: str = '',
+    function_results: Optional[List[Dict[str, Any]]] = None,
+    prior_function_results_loader: Optional[Callable[[], Optional[List[Dict[str, Any]]]]] = None,
+    pending_output_format: Optional[str] = None,
+    analysis_result: Optional[Dict[str, Any]] = None,
+    *,
+    source: GeneratedRecordExportSource,
+    export_request: GeneratedFileExportRequest,
+    max_output_bytes: int,
+    check: Optional[Callable[[], Any]] = None,
+) -> GeneratedFileExportStream: ...
+
+
+@overload
 def build_generated_file_export(
     user_question: str,
     assistant_content: str,
     function_results: Optional[List[Dict[str, Any]]] = None,
     prior_function_results_loader: Optional[Callable[[], Optional[List[Dict[str, Any]]]]] = None,
     pending_output_format: Optional[str] = None,
     analysis_result: Optional[Dict[str, Any]] = None,
-) -> Optional[Dict[str, Any]]:
-    """Build a generated file payload from final assistant content and function-result evidence."""
+) -> Optional[Dict[str, Any]]: ...
+
+
+def build_generated_file_export(
+    user_question: str = '',
+    assistant_content: str = '',
+    function_results: Optional[List[Dict[str, Any]]] = None,
+    prior_function_results_loader: Optional[Callable[[], Optional[List[Dict[str, Any]]]]] = None,
+    pending_output_format: Optional[str] = None,
+    analysis_result: Optional[Dict[str, Any]] = None,
+    *,
+    source: Optional[GeneratedRecordExportSource] = None,
+    export_request: Optional[GeneratedFileExportRequest] = None,
+    max_output_bytes: Optional[int] = None,
+    check: Optional[Callable[[], Any]] = None,
+) -> Optional[Dict[str, Any]] | GeneratedFileExportStream:
+    """Render an explicit complete source, or preserve the existing response-export policy."""
+    if source is not None or export_request is not None:
+        if source is None or export_request is None or analysis_result is not None:
+            raise ValueError('An explicit export requires exactly one saved source and format request.')
+        if type(max_output_bytes) is not int or max_output_bytes < 1:
+            raise ValueError('A positive saved-output byte limit is required.')
+        return _build_generated_record_export(
+            source, export_request, max_output_bytes=max_output_bytes, check=check,
+        )
     output_format = get_requested_generated_file_format(user_question) or _normalize_pending_output_format(
         pending_output_format,
     )