diff --git a/application/single_app/config.py b/application/single_app/config.py index 122ab0ac4..906d30955 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.261.118" +VERSION = "0.261.119" IS_DEVELOPMENT = is_development_env_enabled() SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax') diff --git a/application/single_app/content_screening/access.py b/application/single_app/content_screening/access.py index c04362fcb..a90ce56b3 100644 --- a/application/single_app/content_screening/access.py +++ b/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 diff --git a/application/single_app/functions_artifact_publication.py b/application/single_app/functions_artifact_publication.py index e71855b12..489bdb2a1 100644 --- a/application/single_app/functions_artifact_publication.py +++ b/application/single_app/functions_artifact_publication.py @@ -2,6 +2,7 @@ """Explicit workspace publication of existing artifacts, with scoped retry receipts.""" from copy import deepcopy +from contextlib import ExitStack from datetime import datetime, timezone import hashlib import json @@ -28,6 +29,7 @@ from functions_collaboration import build_conversation_participation_context from functions_documents import allowed_file, create_document, update_document from functions_generated_file_approvals import assert_generated_file_approval_allows_download +from functions_generated_artifact_sources import authorize_generated_artifact_source, has_generated_artifact_source from functions_group import assert_group_role, check_group_status_allows_operation, find_group_by_id from functions_notifications import create_group_notification, create_notification, create_public_workspace_notification from functions_personal_workflows import normalize_workflow_publication @@ -69,7 +71,9 @@ def _authorize_artifact(user_id, conversation_id, message_id): raise LookupError("Generated artifact is unavailable.") assert_generated_file_approval_allows_download(user_id, artifact) assert_generated_chat_artifact_is_published_for_user(user_id, artifact) - authorize_analysis_artifact(user_id, artifact, for_publication=True) + authorize_generated_artifact_source( + user_id, artifact, for_publication=True, native_authorizer=authorize_analysis_artifact, + ) return artifact @@ -109,7 +113,7 @@ def _authorize_destination(user_id, destination): def _artifact_identity(artifact): metadata = artifact.get("metadata") or {} - return { + identity = { "conversation_id": artifact["conversation_id"], "message_id": artifact["id"], "blob_container": artifact["blob_container"], @@ -118,6 +122,36 @@ def _artifact_identity(artifact): "contexts": metadata.get("analysis_result_contexts"), "content_sha256": metadata.get("generated_artifact_content_sha256"), } + if has_generated_artifact_source(metadata): + identity["generated_source"] = deepcopy(metadata.get("generated_artifact_source")) + identity["source_required"] = metadata.get("generated_artifact_source_required") + return identity + + +def _artifact_producer(artifact): + metadata = artifact.get("metadata") or {} + if has_generated_artifact_source(metadata): + return {"kind": "workflow_saved_output", **metadata["generated_artifact_source"]["producer"]} + return metadata.get("analysis_producer") + + +def _publication_destination(publication): + return {key: value for key, value in publication.items() + if key not in {"artifact_format", "completion_policy", "source_kind"}} + + +def _read_publication_artifact_content(artifact, resources, expected_digest, *, check=None): + if has_generated_artifact_source(artifact.get("metadata") or {}): + # The existing transport verifies into bounded private storage before any handoff. + from functions_simplechat_operations import open_generated_chat_artifact_stream + + if (artifact.get("metadata") or {}).get("generated_artifact_content_sha256") != expected_digest: + raise ValueError("The generated artifact bytes changed.") + return resources.enter_context(open_generated_chat_artifact_stream(artifact, check=check)) + content = download_blob_content(artifact["blob_container"], artifact["blob_path"]) + if hashlib.sha256(content).hexdigest() != expected_digest: + raise ValueError("The generated artifact bytes changed.") + return content def _receipt_change(artifact, key, change): @@ -352,6 +386,19 @@ def _completion_response(receipt, document): def publish_generated_chat_artifact_for_user( user_id, *, conversation_id, message_id, destination, request_id, requester_display_name="", file_name="", completion_policy=None, source_receipt=None, execution_check=None, +): + with ExitStack() as resources: + return _publish_generated_chat_artifact_for_user( + user_id, conversation_id=conversation_id, message_id=message_id, destination=destination, + request_id=request_id, requester_display_name=requester_display_name, file_name=file_name, + completion_policy=completion_policy, source_receipt=source_receipt, + execution_check=execution_check, resources=resources, + ) + + +def _publish_generated_chat_artifact_for_user( + user_id, *, conversation_id, message_id, destination, request_id, requester_display_name="", file_name="", + completion_policy=None, source_receipt=None, execution_check=None, resources, ): """Copy or request approval once; uncertain external work is never blindly repeated.""" user_id = _text(user_id, "Acting user") @@ -377,6 +424,10 @@ def reauthorize(): _authorize_destination(user_id, destination) metadata = artifact.get("metadata") or {} + bound_source = has_generated_artifact_source(metadata) + recheck_effect = completion_policy is not None or bound_source + if bound_source and source_receipt is None: + source_receipt = deepcopy(metadata["generated_artifact_source"]["source_receipt"]) name = str(file_name or artifact.get("filename") or "generated-artifact.json").replace("\\", "/").split("/")[-1] output_format = str(metadata.get("generated_artifact_output_format") or "").lower() extension = {"markdown": ".md", "md": ".md", "csv": ".csv", "json": ".json"}.get(output_format) @@ -396,8 +447,10 @@ def reauthorize(): key = hashlib.sha256(json.dumps(identity, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest() if key not in (metadata.get(RECEIPTS_FIELD) or {}): if artifact_bytes is None: - artifact_bytes = download_blob_content(artifact["blob_container"], artifact["blob_path"]) - if hashlib.sha256(artifact_bytes).hexdigest() != content_sha256: + artifact_bytes = _read_publication_artifact_content( + artifact, resources, content_sha256, check=execution_check, + ) + if not bound_source and hashlib.sha256(artifact_bytes).hexdigest() != content_sha256: raise ValueError("The generated artifact bytes changed.") if destination["workspace_scope"] != "personal": stem, suffix = os.path.splitext(name) @@ -407,12 +460,14 @@ def reauthorize(): "content_sha256": content_sha256, "document_id": str(uuid.uuid5(uuid.NAMESPACE_URL, f"simplechat-publication:{key}")), "file_name": name, "created_at": datetime.now(timezone.utc).isoformat(), "stages": {}, } - if completion_policy is not None: + if recheck_effect: receipt.update( - completion_policy=completion_policy, source_receipt=deepcopy(source_receipt), + source_receipt=deepcopy(source_receipt), artifact_reference={"conversation_id": conversation_id, "artifact_message_id": message_id}, source_identity=_artifact_identity(artifact), ) + if completion_policy is not None: + receipt["completion_policy"] = completion_policy if len(metadata.get(RECEIPTS_FIELD) or {}) >= MAX_ARTIFACT_PUBLICATION_REQUESTS and key not in metadata[RECEIPTS_FIELD]: raise ValueError("This artifact has reached its publication request limit.") reauthorize() @@ -427,7 +482,7 @@ def reauthorize(): if document is None: reauthorize() if document is None and _stage(artifact, receipt, "create"): - if completion_policy: + if recheck_effect: reauthorize() try: create_document( @@ -440,7 +495,7 @@ def reauthorize(): if document is None: return _publication_response(artifact, receipt, container) _stage(artifact, receipt, "create", complete=True) - if completion_policy and "document_version" not in receipt: + if recheck_effect and "document_version" not in receipt: if type(document.get("version")) is not int or document["version"] < 1: raise ValueError("The publication destination has no valid native revision.") reauthorize() @@ -451,10 +506,10 @@ def reauthorize(): if not document.get("generated_artifact_publication_receipt_id"): reauthorize() if not document.get("generated_artifact_publication_receipt_id") and _stage(artifact, receipt, "prepare"): - if completion_policy: + if recheck_effect: reauthorize() updates = {"generated_artifact_publication_receipt_id": key} - if completion_policy: + if recheck_effect: updates[PUBLICATION_BINDING] = { "version": 1, "receipt_id": key, "document_version": receipt["document_version"], "content_sha256": content_sha256, "conversation_id": conversation_id, @@ -487,12 +542,14 @@ def reauthorize(): if scope == "personal": if not (receipt.get("stages") or {}).get("queue"): if artifact_bytes is None: - artifact_bytes = download_blob_content(artifact["blob_container"], artifact["blob_path"]) - if hashlib.sha256(artifact_bytes).hexdigest() != receipt["content_sha256"]: + artifact_bytes = _read_publication_artifact_content( + artifact, resources, receipt["content_sha256"], check=execution_check, + ) + if not bound_source and hashlib.sha256(artifact_bytes).hexdigest() != receipt["content_sha256"]: raise ValueError("The generated artifact bytes changed.") reauthorize() if _stage(artifact, receipt, "queue"): - if completion_policy: + if recheck_effect: reauthorize() try: queue_generated_document_processing( @@ -524,13 +581,13 @@ def workspace_notice(): "message": f"{requester_display_name or 'A workspace member'} requested approval for {name} in {workspace_name}.", "link_url": link_url, "link_context": link_context, "metadata": notification_metadata, } - if completion_policy: + if recheck_effect: return create_notification(**scope_args, **kwargs, idempotency_key=f"publication:{key}:workspace") return notify_workspace(destination[target_field], kwargs.pop("notification_type"), kwargs.pop("title"), kwargs.pop("message"), **kwargs) _notify_once( artifact, receipt, "workspace_notification", "approval_request_pending", - workspace_notice, before=reauthorize if completion_policy else None, + workspace_notice, before=reauthorize if recheck_effect else None, ) reauthorize() _notify_once( @@ -540,18 +597,18 @@ def workspace_notice(): title="Generated artifact submitted for approval", message=f"{name} is waiting for approval in {workspace_name}.", link_url=link_url, link_context=link_context, metadata=notification_metadata, - **({"idempotency_key": f"publication:{key}:submitter"} if completion_policy else {}), + **({"idempotency_key": f"publication:{key}:submitter"} if recheck_effect else {}), ), - before=reauthorize if completion_policy else None, + before=reauthorize if recheck_effect else None, ) if scope == "group": invalidate_group_search_cache(destination["group_id"]) - if completion_policy: + if recheck_effect: reauthorize() return _publication_response(artifact, receipt, container) -def publish_workflow_analysis_artifact( +def publish_workflow_artifact( user_id, *, publication, artifact_reference, request_id, source_receipt=None, execution_check=None, ): """Dispatch only a configured publication task using an actual upstream artifact address.""" @@ -564,9 +621,19 @@ def publish_workflow_analysis_artifact( message_id = _text(artifact_reference.get("artifact_message_id"), "Upstream artifact message id") artifact = _authorize_artifact(user_id, conversation_id, message_id) metadata = artifact.get("metadata") or {} - if not metadata.get("analysis_result_required"): + saved_output = publication.get("source_kind") == "saved_output" + if saved_output: + if not has_generated_artifact_source(metadata): + raise ValueError("The upstream artifact is not bound to saved workflow records.") + bound_receipt = metadata["generated_artifact_source"]["source_receipt"] + if not isinstance(source_receipt, dict) or any( + source_receipt.get(key) != bound_receipt[key] + for key in ("producer", "result_ref", "output_name", "output_ref") + ): + raise ValueError("Publication requires the exact saved-output source receipt.") + elif not metadata.get("analysis_result_required") or has_generated_artifact_source(metadata): raise ValueError("The upstream artifact is not bound to a saved final analysis.") - if artifact_reference.get("producer") is not None and artifact_reference["producer"] != metadata.get("analysis_producer"): + if (saved_output or artifact_reference.get("producer") is not None) and artifact_reference.get("producer") != _artifact_producer(artifact): raise ValueError("The upstream artifact belongs to a different analysis result.") output_format = str(metadata.get("generated_artifact_output_format") or "").lower() if output_format == "markdown": @@ -574,19 +641,19 @@ def publish_workflow_analysis_artifact( if output_format != publication["artifact_format"]: raise ValueError("The requested format is not available. Publish an existing upstream artifact.") if "completion_policy" in publication: - producer = metadata.get("analysis_producer") or {} + producer = _artifact_producer(artifact) or {} if ( - producer.get("kind") != "workflow" or not producer.get("execution_id") + producer.get("kind") != ("workflow_saved_output" if saved_output else "workflow") or not producer.get("execution_id") or not isinstance(source_receipt, dict) or source_receipt.get("producer") != {key: value for key, value in producer.items() if key != "kind"} or not isinstance(source_receipt.get("result_ref"), dict) or not isinstance(source_receipt.get("output_ref"), dict) or not source_receipt.get("output_name") ): - raise ValueError("Publication completion needs the exact native workflow result and attempt.") + raise ValueError("Publication completion needs the exact saved workflow result and attempt.") result = publish_generated_chat_artifact_for_user( user_id, conversation_id=conversation_id, message_id=message_id, - destination={key: value for key, value in publication.items() if key not in {"artifact_format", "completion_policy"}}, + destination=_publication_destination(publication), request_id=request_id, completion_policy=publication.get("completion_policy"), source_receipt=source_receipt, execution_check=execution_check, @@ -606,6 +673,16 @@ def publish_workflow_analysis_artifact( return response +def publish_workflow_analysis_artifact( + user_id, *, publication, artifact_reference, request_id, source_receipt=None, execution_check=None, +): + """Compatibility entry point; native definitions retain their original source checks.""" + return publish_workflow_artifact( + user_id, publication=publication, artifact_reference=artifact_reference, request_id=request_id, + source_receipt=source_receipt, execution_check=execution_check, + ) + + def read_workflow_artifact_publication( user_id, request, *, reconcile=False, execution_check=None, authorization_only=False, ): @@ -620,10 +697,10 @@ def read_workflow_artifact_publication( ("artifact_reference", {key: address[key] for key in ("conversation_id", "artifact_message_id")}), )): raise PermissionError("The publication receipt does not match this workflow request.") - destination = {key: value for key, value in publication.items() if key not in {"artifact_format", "completion_policy"}} + destination = _publication_destination(publication) if ( receipt["destination"] != destination or receipt.get("source_identity") != _artifact_identity(artifact) - or address.get("producer") != (artifact.get("metadata") or {}).get("analysis_producer") + or address.get("producer") != _artifact_producer(artifact) ): raise PermissionError("The publication receipt belongs to a different producer or destination.") _, _, container = _authorize_destination(user_id, destination) @@ -724,6 +801,11 @@ def decide_artifact_publication(user_id, document, choice): def _decide_artifact_publication(user_id, document, choice): + with ExitStack() as resources: + return _decide_artifact_publication_with_content(user_id, document, choice, resources=resources) + + +def _decide_artifact_publication_with_content(user_id, document, choice, *, resources): if choice not in {"approved", "rejected", "cancelled"}: raise ValueError("Invalid publication decision.") roles = ("Owner", "Admin", "DocumentManager", "User") if choice == "cancelled" else ("Owner", "Admin", "DocumentManager") @@ -769,9 +851,7 @@ def authorize_decision(): raise ValueError("The original publication artifact changed.") _authorize_artifact(receipt["actor_user_id"], artifact["conversation_id"], artifact["id"]) _authorize_destination(receipt["actor_user_id"], destination) - source_bytes = download_blob_content(artifact["blob_container"], artifact["blob_path"]) - if hashlib.sha256(source_bytes).hexdigest() != receipt["content_sha256"]: - raise ValueError("The original publication artifact changed.") + source_bytes = _read_publication_artifact_content(artifact, resources, receipt["content_sha256"]) _authorize_artifact(receipt["actor_user_id"], artifact["conversation_id"], artifact["id"]) _authorize_destination(receipt["actor_user_id"], destination) authorize_decision() @@ -791,6 +871,9 @@ def record_decision(current): container = cosmos_group_documents_container if scope == "group" else cosmos_public_documents_container scope_args = {key: destination[key] for key in ("group_id", "public_workspace_id") if key in destination} if choice == "approved": + if has_generated_artifact_source(artifact.get("metadata") or {}): + _authorize_artifact(receipt["actor_user_id"], artifact["conversation_id"], artifact["id"]) + _authorize_destination(receipt["actor_user_id"], destination) authorize_decision() _replace_publication_destination(container, receipt, { "generated_artifact_promotion_status": "approved", diff --git a/application/single_app/functions_generated_artifact_sources.py b/application/single_app/functions_generated_artifact_sources.py new file mode 100644 index 000000000..b6d4b3c08 --- /dev/null +++ b/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 diff --git a/application/single_app/functions_generated_file_exports.py b/application/single_app/functions_generated_file_exports.py index 44850f17b..f4e6e0e2c 100644 --- a/application/single_app/functions_generated_file_exports.py +++ b/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,6 +749,23 @@ 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, @@ -633,8 +773,31 @@ def build_generated_file_export( 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, ) diff --git a/application/single_app/functions_personal_workflows.py b/application/single_app/functions_personal_workflows.py index d3d9e1d53..2de9d8d56 100644 --- a/application/single_app/functions_personal_workflows.py +++ b/application/single_app/functions_personal_workflows.py @@ -42,7 +42,8 @@ from functions_workflow_bindings import authorize_workflow_reference from functions_workflow_definition_store import save_workflow_definition_record, update_workflow_runtime_record from functions_workflow_definitions import ( - normalize_publication_completion_policy, normalize_workflow_definition, workflow_definition_for_editor, + normalize_publication_completion_policy, normalize_publication_source_kind, + normalize_workflow_definition, workflow_definition_for_editor, ) from functions_workflow_runtime_store import workflow_runtime_store @@ -165,11 +166,11 @@ def _normalize_alert_priority(value): def normalize_workflow_publication(publication): - """Normalize an explicit existing-artifact request, never a workspace preference.""" + """Normalize an explicit file publication request, never a workspace preference.""" if publication is None: return None if not isinstance(publication, dict) or set(publication) - { - 'artifact_format', 'workspace_scope', 'group_id', 'public_workspace_id', 'completion_policy', + 'artifact_format', 'workspace_scope', 'group_id', 'public_workspace_id', 'completion_policy', 'source_kind', }: raise ValueError('Task publication must specify an artifact format and destination.') output_format = _normalize_text(publication.get('artifact_format'), 'Artifact format', required=True).lower() @@ -181,6 +182,10 @@ def normalize_workflow_publication(publication): if scope not in {'personal', 'group', 'public'}: raise ValueError('Publication destination must be personal, group, or public.') normalized = {'artifact_format': output_format, 'workspace_scope': scope} + if 'source_kind' in publication: + normalized['source_kind'] = normalize_publication_source_kind(publication['source_kind']) + if normalized['source_kind'] == 'saved_output' and output_format != 'json': + raise ValueError('Saved-output publication currently supports exact JSON records only.') if 'completion_policy' in publication: normalized['completion_policy'] = normalize_publication_completion_policy(publication['completion_policy']) target_field = {'group': 'group_id', 'public': 'public_workspace_id'}.get(scope) diff --git a/application/single_app/functions_simplechat_operations.py b/application/single_app/functions_simplechat_operations.py index f2f421b07..197452ce8 100644 --- a/application/single_app/functions_simplechat_operations.py +++ b/application/single_app/functions_simplechat_operations.py @@ -9,12 +9,14 @@ import re import tempfile import uuid +from contextlib import contextmanager from datetime import datetime, timezone from typing import Any, Dict, Iterable, List, Optional, Tuple from urllib.parse import quote import requests -from azure.cosmos.exceptions import CosmosResourceNotFoundError +from azure.core.exceptions import ResourceExistsError +from azure.cosmos.exceptions import CosmosResourceExistsError, CosmosResourceNotFoundError from flask import current_app, has_app_context, session from collaboration_models import normalize_collaboration_user @@ -68,6 +70,13 @@ requires_generated_file_approval, user_can_approve_generated_file, ) +from functions_generated_artifact_sources import ( + authorize_generated_artifact_preparation, + authorize_generated_artifact_source, + generated_artifact_source_metadata, + generated_chat_artifact_address, + has_generated_artifact_source, +) from functions_chat_bootstrap_cache import bump_chat_bootstrap_global_cache_version from functions_group import ( assert_group_role, @@ -1449,6 +1458,30 @@ def upload_generated_analysis_artifact_stream_for_user( artifact_idempotency_key: str = "", artifact_lifecycle_metadata: Optional[Dict[str, Any]] = None, analysis_producer: Optional[Dict[str, Any]] = None, +) -> Dict[str, Any]: + """Compatibility entry point for real native Analyze producers.""" + return upload_generated_file_artifact_stream_for_user( + current_user_id, conversation_id, file_name, file_stream, file_size, + capability=capability, output_format=output_format, summary=summary, + artifact_idempotency_key=artifact_idempotency_key, + artifact_lifecycle_metadata=artifact_lifecycle_metadata, analysis_producer=analysis_producer, + ) + + +def upload_generated_file_artifact_stream_for_user( + current_user_id: str, + conversation_id: str, + file_name: str, + file_stream: Any, + file_size: int, + capability: str = "analysis", + output_format: str = "", + summary: str = "", + artifact_idempotency_key: str = "", + artifact_lifecycle_metadata: Optional[Dict[str, Any]] = None, + analysis_producer: Optional[Dict[str, Any]] = None, + generated_artifact_source: Optional[Dict[str, Any]] = None, + execution_check=None, ) -> Dict[str, Any]: """Upload a bounded-memory generated artifact stream for an authorized user.""" normalized_user_id = str(current_user_id or "").strip() @@ -1466,6 +1499,8 @@ def upload_generated_analysis_artifact_stream_for_user( raise ValueError("file_stream must be seekable and readable") if not allowed_file(normalized_file_name): raise ValueError("Generated file type is not supported") + if generated_artifact_source is not None and analysis_producer is not None: + raise ValueError("A generated artifact must have exactly one producer binding.") normalized_file_size = max(0, int(file_size or 0)) if normalized_file_size <= 0: @@ -1496,8 +1531,11 @@ def upload_generated_analysis_artifact_stream_for_user( "summary": normalized_summary, **(artifact_lifecycle_metadata if isinstance(artifact_lifecycle_metadata, dict) else {}), **analysis_artifact_metadata(analysis_producer), + **(generated_artifact_source_metadata(generated_artifact_source) + if generated_artifact_source is not None else {}), }, artifact_idempotency_key=artifact_idempotency_key, + execution_check=execution_check, ) @@ -2373,12 +2411,29 @@ def _write_temp_markdown_file(markdown_content: str) -> str: return temp_file.name -def _write_temp_generated_file(file_content_bytes: bytes, suffix: str) -> str: +def _write_temp_generated_file(file_content_bytes: Any, suffix: str) -> str: sc_temp_files_dir = "/sc-temp-files" if os.path.exists("/sc-temp-files") else None normalized_suffix = suffix if suffix.startswith('.') else f'.{suffix}' if suffix else '.json' with tempfile.NamedTemporaryFile(delete=False, suffix=normalized_suffix, dir=sc_temp_files_dir) as temp_file: - temp_file.write(file_content_bytes) - return temp_file.name + path = temp_file.name + complete = False + try: + if hasattr(file_content_bytes, "read") and hasattr(file_content_bytes, "seek"): + file_content_bytes.seek(0) + for chunk in iter(lambda: file_content_bytes.read(1024 * 1024), b""): + if not isinstance(chunk, bytes): + raise ValueError("A generated file stream must contain bytes.") + temp_file.write(chunk) + else: + temp_file.write(file_content_bytes) + if temp_file.tell() == 0: + raise ValueError("Generated file content is empty.") + complete = True + return path + finally: + if not complete: + temp_file.close() + os.remove(path) def _queue_document_upload_background_task( @@ -2443,12 +2498,14 @@ def queue_generated_document_processing( if not normalized_name: raise ValueError("normalized_file_name is required") - if isinstance(file_content_bytes, bytes): + if hasattr(file_content_bytes, "read") and hasattr(file_content_bytes, "seek"): + normalized_file_content_bytes = file_content_bytes + elif isinstance(file_content_bytes, bytes): normalized_file_content_bytes = file_content_bytes else: normalized_file_content_bytes = str(file_content_bytes or "").encode("utf-8") - if not normalized_file_content_bytes.strip(): + if isinstance(normalized_file_content_bytes, bytes) and not normalized_file_content_bytes.strip(): raise ValueError("file_content_bytes is required") file_extension = os.path.splitext(normalized_name)[1].lower() or ".json" @@ -3008,6 +3065,51 @@ def auto_deny_expired_generated_file_approvals() -> int: return denied_count +def _verify_generated_artifact_blob(blob_client, expected_digest, expected_size, *, check=None): + digest, size = hashlib.sha256(), 0 + for chunk in blob_client.download_blob().chunks(): + digest.update(chunk) + size += len(chunk) + if size > expected_size: + raise ValueError("The existing generated artifact contains different bytes.") + if check is not None: + check() + if size != expected_size or digest.hexdigest() != expected_digest: + raise ValueError("The existing generated artifact contains different bytes.") + + +@contextmanager +def open_generated_chat_artifact_stream(artifact, *, check=None): + """Verify an entire bound file before a caller hands off or returns its bytes.""" + metadata = artifact.get("metadata") or {} + expected_size = metadata.get("generated_artifact_size_bytes") + expected_digest = metadata.get("generated_artifact_content_sha256") + if type(expected_size) is not int or expected_size < 1 or not expected_digest: + raise ValueError("The generated artifact byte binding is unavailable.") + blob_service_client = CLIENTS.get("storage_account_office_docs_client") + if not blob_service_client: + raise RuntimeError("Blob storage client not available") + blob_client = blob_service_client.get_blob_client(container=artifact["blob_container"], blob=artifact["blob_path"]) + with tempfile.TemporaryFile(mode="w+b") as content: + digest, size = hashlib.sha256(), 0 + if check is not None: + check() + for chunk in blob_client.download_blob().chunks(): + size += len(chunk) + if size > expected_size: + raise ValueError("The generated artifact bytes changed.") + content.write(chunk) + digest.update(chunk) + if check is not None: + check() + if size != expected_size or digest.hexdigest() != expected_digest: + raise ValueError("The generated artifact bytes changed.") + if check is not None: + check() + content.seek(0) + yield content + + def _upload_generated_chat_artifact_for_current_user( current_user_id: str, conversation_id: str, @@ -3015,6 +3117,7 @@ def _upload_generated_chat_artifact_for_current_user( file_content_bytes: bytes, artifact_metadata: Optional[Dict[str, Any]] = None, artifact_idempotency_key: str = "", + execution_check=None, ) -> Dict[str, Any]: try: conversation_item = cosmos_conversations_container.read_item( @@ -3034,15 +3137,32 @@ def _upload_generated_chat_artifact_for_current_user( analysis_metadata = analysis_artifact_metadata(artifact_metadata.get("analysis_producer")) if artifact_metadata.get("analysis_result_required") and not analysis_metadata: raise ValueError("The analysis artifact has no producer binding.") + source_context = ( + authorize_generated_artifact_preparation(current_user_id, artifact_metadata) + if has_generated_artifact_source(artifact_metadata) else None + ) + source_metadata = ( + generated_artifact_source_metadata(source_context["binding"]) if source_context else {} + ) content_digest = hashlib.sha256() + content_size = 0 if hasattr(file_content_bytes, "read") and hasattr(file_content_bytes, "seek"): file_content_bytes.seek(0) for block in iter(lambda: file_content_bytes.read(1024 * 1024), b""): content_digest.update(block) + content_size += len(block) + if execution_check is not None: + execution_check() file_content_bytes.seek(0) else: content_digest.update(file_content_bytes) + content_size = len(file_content_bytes) content_sha256 = content_digest.hexdigest() + if source_context and ( + content_sha256 != source_context["descriptor"]["content_sha256"] + or content_size != source_context["descriptor"]["size_bytes"] + ): + raise ValueError("The prepared generated artifact contains different bytes.") approval_metadata = {} if requires_generated_file_approval( access_context, @@ -3059,22 +3179,49 @@ def _upload_generated_chat_artifact_for_current_user( raise RuntimeError("Blob storage client not available") normalized_idempotency_key = str(artifact_idempotency_key or "").strip() - if normalized_idempotency_key: - artifact_suffix = uuid.uuid5( - uuid.NAMESPACE_URL, - f"simplechat-generated-artifact:{conversation_id}:{normalized_idempotency_key}", - ).hex - else: - artifact_suffix = uuid.uuid4().hex - artifact_message_id = f"{conversation_id}_generated_file_{artifact_suffix}" - blob_path = ( - f"{current_user_id}/{conversation_id}/generated/" - f"{artifact_message_id}/{normalized_file_name}" - ) + address = source_context["descriptor"]["artifact"] if source_context else generated_chat_artifact_address( + current_user_id, conversation_id, normalized_file_name, normalized_idempotency_key, + storage_account_personal_chat_container_name, + ) + if source_context and ( + address["conversation_id"] != conversation_id or address["file_name"] != normalized_file_name + or normalized_idempotency_key != f"generated-export:v1:{source_context['binding']['export_key']}" + or artifact_metadata.get("output_format") != "json" + ): + raise ValueError("The generated artifact does not match its prepared address.") + artifact_message_id, blob_path = address["artifact_message_id"], address["blob_path"] + blob_container = address["blob_container"] blob_client = blob_service_client.get_blob_client( - container=storage_account_personal_chat_container_name, + container=blob_container, blob=blob_path, ) + + def reauthorize_write(): + if execution_check is not None: + execution_check() + if source_context: + current_conversation = cosmos_conversations_container.read_item( + item=conversation_id, partition_key=conversation_id, + ) + build_conversation_participation_context(current_user_id, current_conversation) + authorize_generated_artifact_preparation(current_user_id, artifact_metadata) + + def check_existing(existing): + metadata = existing.get("metadata") or {} + if ( + existing.get("role") != "file" or existing.get("conversation_id") != conversation_id + or existing.get("filename") != normalized_file_name or existing.get("blob_path") != blob_path + or existing.get("blob_container") != blob_container or existing.get("file_content_source") != "blob" + or any(metadata.get(key) != value for key, value in source_metadata.items()) + or metadata.get("generated_artifact_content_sha256") != content_sha256 + or metadata.get("generated_artifact_size_bytes") != content_size + or metadata.get("generated_artifact_idempotency_key") != normalized_idempotency_key + or metadata.get("generated_artifact_output_format") != "json" + ): + raise ValueError("The existing generated artifact has a different immutable binding.") + _verify_generated_artifact_blob(blob_client, content_sha256, content_size, check=execution_check) + reauthorize_write() + if normalized_idempotency_key: try: existing_message = cosmos_messages_container.read_item( @@ -3083,6 +3230,8 @@ def _upload_generated_chat_artifact_for_current_user( ) except CosmosResourceNotFoundError: existing_message = None + if source_context and existing_message is not None: + check_existing(existing_message) if ( isinstance(existing_message, dict) and existing_message.get("role") == "file" @@ -3099,23 +3248,27 @@ def _upload_generated_chat_artifact_for_current_user( "message": { "id": artifact_message_id, "file_name": normalized_file_name, - "blob_container": storage_account_personal_chat_container_name, + "blob_container": blob_container, "blob_path": blob_path, "capability": existing_metadata.get("generated_artifact_capability") or "analysis", "output_format": existing_metadata.get("generated_artifact_output_format") or "", }, "conversation_id": conversation_id, } - blob_client.upload_blob( - file_content_bytes, - overwrite=True, - metadata={ - "conversation_id": conversation_id, - "user_id": current_user_id, - "generated_artifact": "true", - "idempotent_artifact": str(bool(normalized_idempotency_key)).lower(), - }, - ) + reauthorize_write() + try: + blob_client.upload_blob( + file_content_bytes, overwrite=not bool(source_context), + metadata={ + "conversation_id": conversation_id, "user_id": current_user_id, + "generated_artifact": "true", + "idempotent_artifact": str(bool(normalized_idempotency_key)).lower(), + }, + ) + except ResourceExistsError: + if not source_context: + raise + _verify_generated_artifact_blob(blob_client, content_sha256, content_size, check=execution_check) timestamp = datetime.now(timezone.utc).isoformat() current_thread_id = str(uuid.uuid4()) @@ -3137,7 +3290,7 @@ def _upload_generated_chat_artifact_for_current_user( "filename": normalized_file_name, "is_table": file_extension in TABULAR_EXTENSIONS, "file_content_source": "blob", - "blob_container": storage_account_personal_chat_container_name, + "blob_container": blob_container, "blob_path": blob_path, "timestamp": timestamp, "model_deployment_name": None, @@ -3150,6 +3303,8 @@ def _upload_generated_chat_artifact_for_current_user( "generated_artifact_idempotency_key": normalized_idempotency_key or None, "generated_artifact_content_sha256": content_sha256, **analysis_metadata, + **source_metadata, + **({"generated_artifact_size_bytes": content_size} if source_context else {}), **lifecycle_metadata, **approval_metadata, "thread_info": { @@ -3160,9 +3315,19 @@ def _upload_generated_chat_artifact_for_current_user( }, }, } - cosmos_messages_container.upsert_item(message_doc) + reauthorize_write() + created = True + if source_context: + try: + cosmos_messages_container.create_item(message_doc) + except CosmosResourceExistsError: + message_doc = cosmos_messages_container.read_item(item=artifact_message_id, partition_key=conversation_id) + check_existing(message_doc) + created = False + else: + cosmos_messages_container.upsert_item(message_doc) - if approval_metadata: + if approval_metadata and created: _notify_generated_file_approval_requested(message_doc, access_context) log_event( @@ -3184,7 +3349,7 @@ def _upload_generated_chat_artifact_for_current_user( "message": { "id": artifact_message_id, "file_name": normalized_file_name, - "blob_container": storage_account_personal_chat_container_name, + "blob_container": blob_container, "blob_path": blob_path, "capability": artifact_capability, "output_format": artifact_output_format, @@ -3272,8 +3437,8 @@ def _generated_artifact_has_lifecycle_contract(metadata: Dict[str, Any]) -> bool def assert_generated_chat_artifact_is_published_for_user(current_user_id: str, message_item: Dict[str, Any]) -> None: """Reauthorize a generated artifact against its committed artifact-set manifest.""" metadata = message_item.get("metadata") if isinstance(message_item.get("metadata"), dict) else {} - if metadata.get("analysis_result_required") or metadata.get("analysis_result_contexts"): - authorize_analysis_artifact(current_user_id, message_item) + if has_generated_artifact_source(metadata) or metadata.get("analysis_result_required") or metadata.get("analysis_result_contexts"): + authorize_generated_artifact_source(current_user_id, message_item, native_authorizer=authorize_analysis_artifact) if not _generated_artifact_has_lifecycle_contract(metadata): return diff --git a/application/single_app/functions_workflow_artifacts.py b/application/single_app/functions_workflow_artifacts.py new file mode 100644 index 000000000..bf2af6a1d --- /dev/null +++ b/application/single_app/functions_workflow_artifacts.py @@ -0,0 +1,333 @@ +# functions_workflow_artifacts.py +"""Authorized workflow record sources and fenced shared-export materialization.""" + +from copy import deepcopy +import re + +from azure.core.exceptions import ResourceNotFoundError +from azure.cosmos.exceptions import CosmosResourceNotFoundError + +from functions_analysis_access import AnalysisResultUnavailable +from functions_appinsights import log_event +from functions_generated_artifact_sources import generated_chat_artifact_address +from functions_generated_file_exports import GeneratedFileExportRequest, build_generated_file_export +from functions_workflow_definitions import workflow_definition_revision +from functions_workflow_identity import canonical_digest, workflow_execution_id, workflow_node_identity +from functions_workflow_node_results import open_workflow_record_input, result_selectors +from functions_workflow_result_store import _quota_bytes, load_workflow_node_result +from functions_workflow_runtime_store import WorkflowRuntimeConflict, workflow_runtime_store + + +EXPORT_CONTRACT = "generated-file-export-v1" +EXPORT_PROFILE = "exact_records_v1" +_DIGEST = re.compile(r"[0-9a-f]{64}\Z") +_RECEIPT_FIELDS = ("producer", "result_ref", "output_name", "output_ref") + + +def _source_receipt(value): + if ( + not isinstance(value, dict) or not all(name in value for name in _RECEIPT_FIELDS) + or not all(isinstance(value[name], dict) for name in ("producer", "result_ref", "output_ref")) + or not isinstance(value["output_name"], str) or not value["output_name"] + ): + raise AnalysisResultUnavailable("generated_artifact_source_unbound") + return {name: deepcopy(value[name]) for name in _RECEIPT_FIELDS} + + +def _scope(workflow): + return {"type": "group" if workflow.get("group_id") else "personal", + "id": workflow.get("group_id") or workflow["user_id"]} + + +def _export_key(workflow, receipt, allow_partial): + return canonical_digest({ + "contract": EXPORT_CONTRACT, "workflow_scope": _scope(workflow), + "definition_revision": workflow_definition_revision(workflow), + **_source_receipt(receipt), "allow_partial": allow_partial, + "profile": EXPORT_PROFILE, "output_format": "json", + }) + + +def _root_selectors(workflow, run_id): + root = workflow["flow"]["id"] + return {"node_id": root, "execution_id": workflow_execution_id(workflow, run_id, root), + "iteration_path": [], "attempt": 1} + + +def validate_workflow_artifact_binding(value): + if ( + not isinstance(value, dict) or value.keys() != { + "version", "kind", "scope", "producer", "source_receipt", "allow_partial", + "export_key", "profile", "output_format", "materialization", + } + or type(value.get("version")) is not int or value["version"] != 1 + or value.get("kind") != "workflow_saved_output" + or value.get("profile") != EXPORT_PROFILE or value.get("output_format") != "json" + or type(value.get("allow_partial")) is not bool + or not isinstance(value.get("export_key"), str) or not _DIGEST.fullmatch(value["export_key"]) + ): + raise AnalysisResultUnavailable("generated_artifact_source_unbound") + scope, materialization = value["scope"], value["materialization"] + if ( + not isinstance(scope, dict) or scope.keys() != {"type", "id"} + or scope.get("type") not in {"personal", "group"} + or not isinstance(scope.get("id"), str) or not scope["id"] or len(scope["id"]) > 1024 + or not isinstance(materialization, dict) or materialization.keys() != {"descriptor_ref"} + or not isinstance(materialization["descriptor_ref"], dict) + or value["producer"] != _source_receipt(value["source_receipt"])["producer"] + ): + raise AnalysisResultUnavailable("generated_artifact_source_unbound") + producer = value["producer"] + if not all(isinstance(producer.get(key), str) and producer[key] for key in ( + "workflow_id", "run_id", "node_id", "execution_id", + )): + raise AnalysisResultUnavailable("generated_artifact_source_unbound") + return deepcopy(value) + + +def _authorize_workflow_scope(user_id, binding): + # Application stores and current membership are resolved only at the access boundary. + from functions_group import assert_group_role, check_group_status_allows_operation, find_group_by_id + from functions_group_workflows import get_group_workflow, get_group_workflow_run + from functions_personal_workflows import get_personal_workflow, get_personal_workflow_run + + scope, producer = binding["scope"], binding["producer"] + if scope["type"] == "group": + assert_group_role(user_id, scope["id"], allowed_roles=("Owner", "Admin", "DocumentManager", "User")) + allowed, _ = check_group_status_allows_operation(find_group_by_id(scope["id"]), "view") + if not allowed: + raise AnalysisResultUnavailable("generated_artifact_source_unavailable") + workflow = get_group_workflow(scope["id"], producer["workflow_id"]) + run = get_group_workflow_run(scope["id"], producer["run_id"]) + else: + if scope["id"] != user_id: + raise AnalysisResultUnavailable("generated_artifact_source_unavailable") + workflow = get_personal_workflow(user_id, producer["workflow_id"]) + run = get_personal_workflow_run(user_id, producer["run_id"]) + if ( + not workflow or workflow.get("id") != producer["workflow_id"] or _scope(workflow) != scope + or not run or run.get("id") != producer["run_id"] or run.get("workflow_id") != workflow["id"] + ): + raise AnalysisResultUnavailable("generated_artifact_source_unavailable") + return workflow, run + + +class WorkflowRecordExportSource: + kind = "records" + + def __init__(self, workflow, run_id, receipt, *, actor_user_id, store, allow_partial=False, + load_result=load_workflow_node_result, inspection=False): + self.workflow, self.run_id, self.store = workflow, run_id, store + self.actor_user_id = actor_user_id + self.receipt = _source_receipt(receipt) + self.allow_partial = allow_partial + identity = self.receipt["producer"] + expected = workflow_node_identity( + workflow, run_id, identity.get("node_id"), identity.get("execution_id"), identity.get("attempt"), + task_id=identity.get("task_id"), iteration_path=identity.get("iteration_path"), + ) + if identity != expected: + raise AnalysisResultUnavailable("generated_artifact_source_unbound") + self._check_scope() + self._check_attempt() + self.reader = open_workflow_record_input( + workflow, run_id, identity, self.receipt["result_ref"], output_name=self.receipt["output_name"], + reader_user_id=actor_user_id, allow_partial=allow_partial, load_result=load_result, inspection=inspection, + ) + if self.reader.kind != "records" or _source_receipt(self.reader.receipt) != self.receipt: + raise AnalysisResultUnavailable("generated_artifact_source_unbound") + self.record_count = self.reader.record_count + + def _check_attempt(self): + identity = self.receipt["producer"] + row = self.store.journal_read("attempt", [identity["execution_id"], identity["attempt"]]) + payload = row["payload"] if row else {} + summary = payload.get("workflow_result") or {} + if ( + payload.get("state") not in {"completed", "succeeded"} + or any(payload.get(key) != identity.get(key) for key in ( + "node_id", "execution_id", "iteration_path", "attempt", "task_id", + )) + or summary.get("producer") != identity or summary.get("result_ref") != self.receipt["result_ref"] + ): + raise AnalysisResultUnavailable("generated_artifact_source_uncommitted") + + def _check_scope(self): + _authorize_workflow_scope(self.actor_user_id, { + "scope": _scope(self.workflow), "producer": self.receipt["producer"], + }) + + def recheck(self) -> None: + self._check_scope() + self._check_attempt() + self.reader.recheck() + + def iter_records(self): + for index, record in enumerate(self.reader.iter_records()): + if index % 100 == 0: + self._check_scope() + self._check_attempt() + yield record + + +def load_workflow_artifact_binding(user_id, value, *, require_ready=True, for_publication=False): + try: + return _load_workflow_artifact_binding( + user_id, value, require_ready=require_ready, for_publication=for_publication, + ) + except (ValueError, WorkflowRuntimeConflict, ResourceNotFoundError, CosmosResourceNotFoundError) as exc: + log_event( + "[SIMPLE_CHAT] Saved-output artifact source unavailable", + {"exception_type": type(exc).__name__}, + ) + raise AnalysisResultUnavailable("generated_artifact_source_unavailable") from exc + + +def _load_workflow_artifact_binding(user_id, value, *, require_ready, for_publication): + binding = validate_workflow_artifact_binding(value) + workflow, run = _authorize_workflow_scope(user_id, binding) + store = workflow_runtime_store(workflow, binding["producer"]["run_id"]) + workflow = store.run_definition() + run_id, key = binding["producer"]["run_id"], binding["export_key"] + if _scope(workflow) != binding["scope"] or key != _export_key(workflow, binding["source_receipt"], binding["allow_partial"]): + raise AnalysisResultUnavailable("generated_artifact_source_unbound") + source = WorkflowRecordExportSource( + workflow, run_id, binding["source_receipt"], actor_user_id=user_id, store=store, + allow_partial=binding["allow_partial"], inspection=not for_publication, + ) + selectors = _root_selectors(workflow, run_id) + reference = binding["materialization"]["descriptor_ref"] + prepared = store.journal_read("unit", ["generated-file-prepare", key]) + expected_prepared = {"state": "completed", "selectors": selectors, "result_ref": reference} + if not prepared or prepared["payload"] != expected_prepared: + raise AnalysisResultUnavailable("generated_artifact_source_uncommitted") + descriptor = load_workflow_node_result(workflow, run_id, None, reference, **selectors) + if not isinstance(descriptor, dict): + raise AnalysisResultUnavailable("generated_artifact_source_unbound") + expected_source = {name: field for name, field in binding.items() if name != "materialization"} + address = descriptor.get("artifact") or {} + if ( + descriptor.get("contract_version") != EXPORT_CONTRACT or descriptor.get("source") != expected_source + or address.get("conversation_id") != run.get("conversation_id") + or not address.get("blob_container") or not address.get("conversation_id") + or descriptor.get("record_count") != source.record_count + or type(descriptor.get("size_bytes")) is not int or descriptor["size_bytes"] < 2 + or not isinstance(descriptor.get("content_sha256"), str) or not _DIGEST.fullmatch(descriptor["content_sha256"]) + or address != generated_chat_artifact_address( + workflow["user_id"], address.get("conversation_id"), f"workflow-output-{key}.json", + f"generated-export:v1:{key}", address.get("blob_container"), + ) + ): + raise AnalysisResultUnavailable("generated_artifact_source_unbound") + if require_ready: + ready = store.journal_read("unit", ["generated-file-ready", key]) + if not ready or ready["payload"] != {"state": "completed", "descriptor_ref": reference}: + raise AnalysisResultUnavailable("generated_artifact_source_uncommitted") + return {"binding": binding, "descriptor": descriptor, "source": source} + + +def authorize_workflow_saved_output_artifact(user_id, artifact, *, for_publication=False): + metadata = artifact.get("metadata") or {} + context = load_workflow_artifact_binding( + user_id, metadata.get("generated_artifact_source"), for_publication=for_publication, + ) + descriptor, binding = context["descriptor"], context["binding"] + expected = descriptor["artifact"] + if ( + artifact.get("id") != expected["artifact_message_id"] + or artifact.get("filename") != expected["file_name"] or artifact.get("role") != "file" + or any(artifact.get(key) != expected[key] for key in ("conversation_id", "blob_container", "blob_path")) + or metadata.get("generated_artifact_content_sha256") != descriptor["content_sha256"] + or metadata.get("generated_artifact_size_bytes") != descriptor["size_bytes"] + or metadata.get("generated_artifact_output_format") != "json" + or metadata.get("generated_artifact_idempotency_key") != f"generated-export:v1:{binding['export_key']}" + ): + raise AnalysisResultUnavailable("generated_artifact_source_unbound") + return context + + +def materialize_workflow_saved_output(execution, receipt, *, actor_user_id, conversation_id, + allow_partial=False, upload=None, artifact_container=None): + """Commit a shared exact export without creating a task, scheduler, or publication ledger.""" + workflow, run_id = execution.workflow, execution.run_id + execution.check() + source = WorkflowRecordExportSource( + workflow, run_id, receipt, actor_user_id=actor_user_id, store=execution.store, + allow_partial=allow_partial, load_result=execution.load_result, + ) + key = _export_key(workflow, source.receipt, allow_partial) + prepared_key, ready_key = ["generated-file-prepare", key], ["generated-file-ready", key] + prepared = execution.store.journal_read("unit", prepared_key) + selectors = _root_selectors(workflow, run_id) + descriptor, reference = None, None + if prepared: + if ( + prepared["payload"].get("state") != "completed" or prepared["payload"].get("selectors") != selectors + or not isinstance(prepared["payload"].get("result_ref"), dict) + ): + raise AnalysisResultUnavailable("generated_artifact_source_unbound") + reference = prepared["payload"]["result_ref"] + descriptor = execution.load_result(workflow, run_id, None, reference, **selectors) + if descriptor is None or execution.store.journal_read("unit", ready_key) is None: + if artifact_container is None: + # Reuse the configured private generated-file transport at materialization time. + from config import storage_account_personal_chat_container_name + + artifact_container = storage_account_personal_chat_container_name + if upload is None: + from functions_simplechat_operations import upload_generated_file_artifact_stream_for_user + + upload = upload_generated_file_artifact_stream_for_user + base_binding = { + "version": 1, "kind": "workflow_saved_output", "scope": _scope(workflow), + "producer": deepcopy(source.receipt["producer"]), "source_receipt": source.receipt, + "allow_partial": allow_partial, "export_key": key, "profile": EXPORT_PROFILE, "output_format": "json", + } + with build_generated_file_export( + source=source, export_request=GeneratedFileExportRequest("json", EXPORT_PROFILE), + max_output_bytes=_quota_bytes(execution.settings), check=execution.check, + ) as exported: + candidate = { + "contract_version": EXPORT_CONTRACT, "source": base_binding, + "artifact": generated_chat_artifact_address( + workflow["user_id"], conversation_id, f"workflow-output-{key}.json", + f"generated-export:v1:{key}", artifact_container, + ), + "content_sha256": exported.content_sha256, "size_bytes": exported.size_bytes, + "record_count": exported.record_count, + "validation_status": (source.reader.manifest.get("workflow_validation") or {}).get("status"), + } + if descriptor is not None and candidate != descriptor: + raise AnalysisResultUnavailable("generated_artifact_content_changed") + if descriptor is None: + descriptor = candidate + reference = execution.save_result( + workflow, run_id, None, descriptor, settings=execution.settings, **selectors, + ) + execution.store.journal_commit(execution.lease.token, "unit", prepared_key, { + "state": "completed", "selectors": selectors, "result_ref": reference, + }, immutable=True) + binding = {**base_binding, "materialization": {"descriptor_ref": reference}} + execution.check() + source.recheck() + upload( + actor_user_id, conversation_id, descriptor["artifact"]["file_name"], + exported.file_content, exported.size_bytes, capability="file_export", output_format="json", + summary=f"{exported.record_count} exact saved records ({descriptor['validation_status']}).", + artifact_idempotency_key=f"generated-export:v1:{key}", generated_artifact_source=binding, + execution_check=execution.check, + ) + execution.check() + source.recheck() + execution.store.journal_commit(execution.lease.token, "unit", ready_key, { + "state": "completed", "descriptor_ref": reference, + }, immutable=True) + binding = {**descriptor["source"], "materialization": {"descriptor_ref": reference}} + load_workflow_artifact_binding(actor_user_id, binding, for_publication=True) + source.recheck() + execution.check() + return { + **descriptor["artifact"], "output_format": "json", "capability": "file_export", + "record_count": descriptor["record_count"], "content_sha256": descriptor["content_sha256"], + "validation_status": descriptor["validation_status"], "source_binding": binding, + } diff --git a/application/single_app/functions_workflow_definitions.py b/application/single_app/functions_workflow_definitions.py index e349a5943..5b45c3b8c 100644 --- a/application/single_app/functions_workflow_definitions.py +++ b/application/single_app/functions_workflow_definitions.py @@ -15,6 +15,7 @@ WORKFLOW_OUTPUT_KINDS = frozenset({"any", "text", "records", "json", "document_results"}) WORKFLOW_INPUT_PROCESSING_MODES = frozenset({"full", "saved_record_report"}) WORKFLOW_PUBLICATION_COMPLETION_POLICIES = ("submitted", "approved", "indexed_ready") +WORKFLOW_PUBLICATION_SOURCE_KINDS = ("native_analysis", "saved_output") WORKFLOW_FLOW_TASK_FIELDS = frozenset({"inputs", "reference_ids", "output_contract", "approval", "input_processing"}) WORKFLOW_DEFINITION_FIELDS = ( "name", "description", "task_prompt", "tasks", "runner_type", "chat_capabilities_enabled", @@ -52,12 +53,27 @@ def normalize_publication_completion_policy(value): return value +def normalize_publication_source_kind(value): + if not isinstance(value, str) or value not in WORKFLOW_PUBLICATION_SOURCE_KINDS: + raise WorkflowDefinitionError("Publication source must be native_analysis or saved_output.") + return value + + def validate_workflow_publication_completion(workflow): for task in workflow.get("tasks") or []: if not isinstance(task, dict): raise WorkflowDefinitionError("A workflow task must be an object.") publication = task.get("publication") - if isinstance(publication, dict) and "completion_policy" in publication: + if not isinstance(publication, dict): + continue + if "source_kind" in publication: + normalize_publication_source_kind(publication["source_kind"]) + if publication.get("source_kind") == "saved_output": + if workflow.get("definition_version") != 3 or workflow.get("durable_execution") is not True: + raise WorkflowDefinitionError("Saved-output publication requires a version-3 durable workflow.") + if publication.get("artifact_format") != "json": + raise WorkflowDefinitionError("Saved-output publication currently supports exact JSON records only.") + if "completion_policy" in publication: normalize_publication_completion_policy(publication["completion_policy"]) if workflow.get("definition_version") != 3 or workflow.get("durable_execution") is not True: raise WorkflowDefinitionError("Publication completion policies require a version-3 durable workflow.") diff --git a/application/single_app/functions_workflow_editor.py b/application/single_app/functions_workflow_editor.py index d32cfa70e..1a785c37d 100644 --- a/application/single_app/functions_workflow_editor.py +++ b/application/single_app/functions_workflow_editor.py @@ -2,6 +2,7 @@ """Non-secret editor choices and trusted loop-runner eligibility.""" from functions_ai_connections import supports_model_capability +from functions_generated_file_exports import GENERATED_RECORD_EXPORT_FORMATS from functions_workflow_definitions import ( WORKFLOW_DEFINITION_VERSION, WORKFLOW_INPUT_PROCESSING_MODES, WORKFLOW_PUBLICATION_COMPLETION_POLICIES, ) @@ -63,6 +64,12 @@ def build_workflow_editor_options(*, scope_type, scope_id, can_manage, max_tasks "supported_binding_sources": ["node_output", "loop_item"], "supported_input_processing_modes": sorted(WORKFLOW_INPUT_PROCESSING_MODES), "supported_publication_completion_policies": list(WORKFLOW_PUBLICATION_COMPLETION_POLICIES), + "publication_source_capabilities": [ + {"source_kind": "native_analysis", "output_kinds": ["records", "json", "text", "document_results"], + "artifact_formats": ["md", "csv", "json"]}, + {"source_kind": "saved_output", "output_kinds": ["records"], + "artifact_formats": list(GENERATED_RECORD_EXPORT_FORMATS["exact_records_v1"])}, + ], "flow_limits": { **FLOW_LIMITS, "max_loop_items": validate_workflow_max_loop_items(max_loop_items), diff --git a/application/single_app/functions_workflow_flow.py b/application/single_app/functions_workflow_flow.py index 4c142c7c9..ca4f676ab 100644 --- a/application/single_app/functions_workflow_flow.py +++ b/application/single_app/functions_workflow_flow.py @@ -641,8 +641,17 @@ def analyze(current, initial, *, branch=False): after_possible.update({(node["id"], output) for output in WORKFLOW_BINDABLE_OUTPUTS}) if "run_when" not in node: after_definite.update(keys) - if catalogue[node["task_id"]].get("publication") is not None and len(bindings) != 1: - raise WorkflowDefinitionError("A v3 publication task requires exactly one explicit upstream input.") + publication = catalogue[node["task_id"]].get("publication") + if publication is not None: + if len(bindings) != 1: + raise WorkflowDefinitionError("A v3 publication task requires exactly one explicit upstream input.") + if isinstance(publication, dict) and publication.get("source_kind") == "saved_output" and ( + bindings[0]["source"]["kind"] != "node_output" + or bindings[0]["required"] is not True + or bindings[0]["expected_kind"] != "records" + or collection_kind(bindings[0]["source"]) != "records" + ): + raise WorkflowDefinitionError("Saved-output publication requires one required records node-output input.") elif kind == "if": ends = {name: analyze(node[name], (set(definite), set(possible)), branch=True) for name in ("then", "else")} diff --git a/application/single_app/functions_workflow_runner.py b/application/single_app/functions_workflow_runner.py index 13a1409f2..4be576747 100644 --- a/application/single_app/functions_workflow_runner.py +++ b/application/single_app/functions_workflow_runner.py @@ -10423,14 +10423,21 @@ def raise_if_cancelled(): if publication_completion and (not structured_definition or durable is None): raise WorkflowResultNotReadyError('Publication completion requires a version-3 durable workflow.') publication_inputs = flow_runner.resolve(task['inputs'], metadata_only=True) if flow_runner else None + # Completed loops are not yielded on replay; their display ordinal is + # not part of a saved-output publication's immutable authored input. + publication_task = ( + {key: value for key, value in task.items() if key != 'order'} + if task['publication'].get('source_kind') == 'saved_output' else task + ) task_result, consumed_inputs = workflow_unit( task_unit_key, lambda: _execute_workflow_analysis_publication( workflow, run_id, task, previous_task_id, previous_result_ref, actor_user_id=actor_user_id, + conversation_id=conversation_id, explicit_inputs=publication_inputs['bound_inputs'] if publication_inputs else None, ), - inputs=({'task': task, 'consumed_inputs': publication_inputs['consumed_inputs']} + inputs=({'task': publication_task, 'consumed_inputs': publication_inputs['consumed_inputs']} if publication_inputs is not None else {'task': task, 'producer_task_id': previous_task_id, 'result_ref': previous_result_ref}), approval=task.get('approval'), @@ -10989,7 +10996,7 @@ def save_analysis_section(bound_workflow, bound_run, bound_task, section, **_kwa def _execute_workflow_analysis_publication( workflow, run_id, task, previous_task_id, previous_result_ref, *, actor_user_id=None, - result_reader=None, publish=None, explicit_inputs=None, + result_reader=None, publish=None, explicit_inputs=None, conversation_id=None, ): """Select only a committed ancestor artifact, before resolving a model or its context.""" publication = task.get('publication') @@ -11002,6 +11009,63 @@ def _execute_workflow_analysis_publication( publication = normalize_workflow_publication(publication) reader = result_reader or authorize_workflow_task_result_read actor = actor_user_id or workflow.get('user_id') + + def complete_response(result, request_id): + state = (result.get('publication') or {}).get('state') + if 'completion_policy' in publication: + execution = current_workflow_execution() + result['_publication_reference'] = { + 'kind': 'artifact_publication', 'version': 1, 'execution_id': execution.execution_id(), + 'attempt': execution.selectors()['attempt'], 'request_id': request_id, + 'receipt_id': result['publication']['id'], + } + elif state == 'pending_approval': + result['execution_status'] = 'pending' + elif state in {'uncertain', 'approval_failed'}: + result['execution_status'] = 'blocked' + return result + + if publication.get('source_kind') == 'saved_output': + from functions_artifact_publication import publish_workflow_artifact + from functions_generated_file_exports import build_generated_file_artifact_metadata + from functions_workflow_artifacts import materialize_workflow_saved_output + + execution = current_workflow_execution() + bindings = task.get('inputs') or [] + if ( + workflow.get('definition_version') != 3 or execution is None or not conversation_id + or not isinstance(explicit_inputs, list) or len(explicit_inputs) != 1 or len(bindings) != 1 + or bindings[0].get('required') is not True or bindings[0].get('expected_kind') != 'records' + or (bindings[0].get('source') or {}).get('kind') != 'node_output' + ): + raise WorkflowResultNotReadyError('Saved-output publication requires one exact records input and the current durable execution.') + receipt = explicit_inputs[0] + artifact = materialize_workflow_saved_output( + execution, receipt, actor_user_id=actor, conversation_id=conversation_id, + allow_partial=bindings[0].get('allow_partial', False), + ) + producer = receipt['producer'] + request_id = f"workflow-publication:v3:{execution.execution_id()}:{producer['execution_id']}:{producer['attempt']}" + result = (publish or publish_workflow_artifact)( + actor, publication=publication, + artifact_reference={ + 'conversation_id': conversation_id, 'artifact_message_id': artifact['artifact_message_id'], + 'producer': {'kind': 'workflow_saved_output', **producer}, + }, + request_id=request_id, source_receipt=receipt, execution_check=execution.check, + ) + public_artifact = build_generated_file_artifact_metadata( + {'file_name': artifact['file_name'], 'output_format': 'json', 'capability': 'file_export', + 'row_source': 'saved_records', 'row_count': artifact['record_count'], + 'summary': f"{artifact['record_count']} exact saved records ({artifact['validation_status']})."}, + {'message': {'id': artifact['artifact_message_id'], 'file_name': artifact['file_name']}}, + conversation_id, + ) + public_artifact.update( + row_count=artifact['record_count'], suppress_assistant_text=False, source_kind='workflow_saved_output', + ) + result['generated_tabular_outputs'] = [public_artifact] + return complete_response(result, request_id), [receipt] if workflow.get('definition_version') == 3: if not isinstance(explicit_inputs, list) or len(explicit_inputs) != 1: raise WorkflowResultNotReadyError('Publication requires exactly one explicit saved analysis input.') @@ -11092,18 +11156,7 @@ def _execute_workflow_analysis_publication( request_id=publication_request_id, **completion_options, ) - state = (result.get('publication') or {}).get('state') - if 'completion_policy' in publication: - result['_publication_reference'] = { - 'kind': 'artifact_publication', 'version': 1, 'execution_id': execution.execution_id(), - 'attempt': execution.selectors()['attempt'], 'request_id': publication_request_id, - 'receipt_id': result['publication']['id'], - } - elif state == 'pending_approval': - result['execution_status'] = 'pending' - elif state in {'uncertain', 'approval_failed'}: - result['execution_status'] = 'blocked' - return result, ( + return complete_response(result, publication_request_id), ( [*explicit_inputs, native_receipt] if workflow.get('definition_version') == 3 and explicit_inputs[0]['producer'] != native_receipt['producer'] else [native_receipt] ) diff --git a/application/single_app/route_backend_conversation_export.py b/application/single_app/route_backend_conversation_export.py index b79e614ba..44e91c84d 100644 --- a/application/single_app/route_backend_conversation_export.py +++ b/application/single_app/route_backend_conversation_export.py @@ -2193,7 +2193,10 @@ def _load_export_message_for_user(user_id: str, conversation_id: str, message_id for analysis_context in analysis_result_contexts(message): load_saved_analysis(user_id, analysis_context) if message.get('role') == 'file': - authorize_analysis_artifact(user_id, message) + # File exports retain the same source boundary as direct artifact downloads. + from functions_generated_artifact_sources import authorize_generated_artifact_source + + authorize_generated_artifact_source(user_id, message, native_authorizer=authorize_analysis_artifact) if isinstance(message.get('agent_citations'), list) and any( isinstance(citation, dict) and citation.get('artifact_id') diff --git a/application/single_app/route_enhanced_citations.py b/application/single_app/route_enhanced_citations.py index 10a3b7118..ce2e50f79 100644 --- a/application/single_app/route_enhanced_citations.py +++ b/application/single_app/route_enhanced_citations.py @@ -2,6 +2,7 @@ # Backend endpoints for enhanced citations supporting different media types from flask import jsonify, request, Response +from contextlib import ExitStack from datetime import datetime, timedelta, timezone import hashlib import logging @@ -37,6 +38,7 @@ from functions_public_workspaces import get_user_visible_public_workspace_ids_from_settings from functions_collaboration import build_conversation_participation_context from functions_generated_file_approvals import assert_generated_file_approval_allows_download +from functions_generated_artifact_sources import authorize_generated_artifact_source, has_generated_artifact_source from functions_saved_analysis import authorize_analysis_artifact from functions_simplechat_operations import ( assert_generated_chat_artifact_is_published_for_user, @@ -215,34 +217,48 @@ def _build_content_disposition(disposition, file_name, fallback='download'): def _serve_chat_artifact_download(user_id, conversation_id, message_id): """Read an authorized chat artifact, not a workspace document's representation.""" artifact = _get_authorized_chat_artifact_message(user_id, conversation_id, message_id) - active_document = None - if artifact.get('workspace_document_id'): - active_document, content = read_available_document_bytes( - artifact['workspace_document_id'], user_id=user_id, purpose='chat_file', + with ExitStack() as resources: + active_document = None + streamed = False + if artifact.get('workspace_document_id'): + active_document, content = read_available_document_bytes( + artifact['workspace_document_id'], user_id=user_id, purpose='chat_file', + ) + elif has_generated_artifact_source(artifact.get('metadata') or {}): + # This explicit source can exceed the old in-memory artifact path. + from functions_simplechat_operations import open_generated_chat_artifact_stream + + content = resources.enter_context(open_generated_chat_artifact_stream(artifact)) + streamed = True + else: + content = download_blob_content(artifact['blob_container'], artifact['blob_path']) + current = _get_authorized_chat_artifact_message(user_id, conversation_id, message_id) + identity_fields = ('id', 'conversation_id', 'workspace_document_id', 'blob_container', 'blob_path', 'filename', '_etag') + digest = (artifact.get('metadata') or {}).get('generated_artifact_content_sha256') + current_digest = (current.get('metadata') or {}).get('generated_artifact_content_sha256') + if any(artifact.get(field) != current.get(field) for field in identity_fields) or digest != current_digest: + raise LookupError('The artifact changed during download.') + if not streamed and active_document is None and digest and hashlib.sha256(content).hexdigest() != digest: + raise LookupError('The artifact content changed.') + + file_name = (active_document or {}).get('file_name') or _resolve_generated_artifact_file_name(current) + content_type = { + '.csv': 'text/csv; charset=utf-8', + '.md': 'text/markdown; charset=utf-8', + '.json': 'application/json', + }.get(os.path.splitext(file_name)[1].lower()) or mimetypes.guess_type(file_name)[0] or 'application/octet-stream' + response = Response( + iter(lambda: content.read(65536), b'') if streamed else content, + content_type=content_type, headers={ + 'Content-Length': str(current['metadata']['generated_artifact_size_bytes'] if streamed else len(content)), + 'Content-Disposition': _build_content_disposition('attachment', file_name), + 'Cache-Control': 'private, no-store', + 'X-Content-Type-Options': 'nosniff', + }, ) - else: - content = download_blob_content(artifact['blob_container'], artifact['blob_path']) - current = _get_authorized_chat_artifact_message(user_id, conversation_id, message_id) - identity_fields = ('id', 'conversation_id', 'workspace_document_id', 'blob_container', 'blob_path', 'filename', '_etag') - digest = (artifact.get('metadata') or {}).get('generated_artifact_content_sha256') - current_digest = (current.get('metadata') or {}).get('generated_artifact_content_sha256') - if any(artifact.get(field) != current.get(field) for field in identity_fields) or digest != current_digest: - raise LookupError('The artifact changed during download.') - if active_document is None and digest and hashlib.sha256(content).hexdigest() != digest: - raise LookupError('The artifact content changed.') - - file_name = (active_document or {}).get('file_name') or _resolve_generated_artifact_file_name(current) - content_type = { - '.csv': 'text/csv; charset=utf-8', - '.md': 'text/markdown; charset=utf-8', - '.json': 'application/json', - }.get(os.path.splitext(file_name)[1].lower()) or mimetypes.guess_type(file_name)[0] or 'application/octet-stream' - return Response(content, content_type=content_type, headers={ - 'Content-Length': str(len(content)), - 'Content-Disposition': _build_content_disposition('attachment', file_name), - 'Cache-Control': 'private, no-store', - 'X-Content-Type-Options': 'nosniff', - }) + if streamed: + response.call_on_close(resources.pop_all().close) + return response def _log_enhanced_citations_debug(message, **details): @@ -735,7 +751,9 @@ def promote_chat_artifact_to_workspace(): if not payload.get("workspace_scope"): raise ValueError("Choose an explicit destination for this analysis artifact.") - authorize_analysis_artifact(user_id, message_item, for_publication=True) + authorize_generated_artifact_source( + user_id, message_item, for_publication=True, native_authorizer=authorize_analysis_artifact, + ) file_name = _resolve_generated_artifact_file_name(message_item) requester_display_name = ( str(current_user_info.get("displayName") or "").strip() diff --git a/application/v2_ui/src/components/chat/GeneratedArtifactCard.tsx b/application/v2_ui/src/components/chat/GeneratedArtifactCard.tsx index dbbd35ded..14165329e 100644 --- a/application/v2_ui/src/components/chat/GeneratedArtifactCard.tsx +++ b/application/v2_ui/src/components/chat/GeneratedArtifactCard.tsx @@ -294,6 +294,7 @@ export function GeneratedArtifactCard({ const summary = text(artifact.summary); const running = Boolean(artifact.background_export); const compact = isCompletedTabularArtifact(artifact); + const savedRecordsExport = artifact.capability === 'file_export' && artifact.row_source === 'saved_records'; const downloadUrl = generatedArtifactDownloadUrl(artifact, conversationId); const sourceNote = [ @@ -424,6 +425,9 @@ export function GeneratedArtifactCard({ {summary &&

{summary}

} )} + {compact && !running && savedRecordsExport && summary && ( +

{summary}

+ )} {running && ( ; allowLoopItems?: boolean; + recordsOnly?: boolean; }) { const available = availableIds ?? analyzeWorkflowFlow(workflow).available.get(nodeId) ?? new Set(); - const producers = flowProducers(workflow).filter((producer) => available.has(producer.id)); - const loops = allowLoopItems ? enclosingFlowLoops(workflow, nodeId) : []; + const producers = flowProducers(workflow).filter((producer) => available.has(producer.id)) + .map((producer) => recordsOnly ? { ...producer, outputs: producer.outputs.filter(isRecordsFlowOutput) } : producer) + .filter((producer) => !recordsOnly || producer.outputs.length > 0); + const loops = allowLoopItems && !recordsOnly ? enclosingFlowLoops(workflow, nodeId) : []; const update = (index: number, binding: WorkflowFlowBinding) => onChange(bindings.map((current, position) => position === index ? binding : current)); const add = () => { @@ -65,6 +70,7 @@ export function WorkflowFlowInputs({ {label}

Only these named final outputs are consumed. A skipped producer never falls back to another task. + {recordsOnly ? ' File export requires exactly one required records output. Partial output still needs this binding’s explicit acceptance and an eligible producer.' : ''}

{bindings.map((binding, index) => { const source = binding.source; @@ -85,7 +91,7 @@ export function WorkflowFlowInputs({ : { ...flowBinding(binding.name, producers[0]?.id ?? '', producers[0]?.outputs[0]?.name ?? ''), expected_kind: producers[0]?.outputs[0]?.kind ?? 'any' })}> - + : null} {source.kind === 'node_output' ? <> @@ -100,7 +106,7 @@ export function WorkflowFlowInputs({ expected_kind: selected.outputs[0]?.kind ?? 'any', }); }}> - {!producer ? : null} + {!producer ? : null} {producers.map((item) => )} @@ -115,7 +121,7 @@ export function WorkflowFlowInputs({ }); }}> {!producer?.outputs.some((item) => item.name === source.output) ? ( - + ) : null} {producer?.outputs.map((output) => )} @@ -136,7 +142,8 @@ export function WorkflowFlowInputs({