Add structured workflow control flow and V2 List tools (M4A)

application/single_app/config.py

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

application/single_app/functions_document_analysis_checkpoints.py

@@ -10,6 +10,7 @@
 """
 
 import hashlib
+import json
 import uuid
 from copy import deepcopy
 
@@ -118,7 +119,8 @@ def _lineage(self):
         seen = set()
         expected_child = None
         while binding is not None:
-            identity = tuple(sorted(binding.items()))
+            identity = tuple((key, json.dumps(value, sort_keys=True, separators=(',', ':')))
+                             for key, value in sorted(binding.items()))
             if identity in seen or len(seen) >= 128:
                 raise WorkflowResultIntegrityError('Analysis checkpoint retry lineage is invalid.')
             seen.add(identity)
@@ -243,12 +245,23 @@ def analysis_checkpoints_for_workflow(
     attempt_token=None, settings=None, store=None, source_authorizer=None,
     operation_request=None, operation_sources=None,
     recover_running_unit=None,
+    execution_id=None, node_id=None, attempt=None, iteration_path=None,
 ):
-    binding = _identity(workflow, run_id, task_id)
+    selectors = {}
+    if execution_id is not None:
+        selectors = {"execution_id": execution_id, "node_id": node_id, "attempt": attempt,
+                     "iteration_path": [] if iteration_path is None else iteration_path}
+    binding = _identity(workflow, run_id, task_id, **selectors)
+    result_store = store or _configured_result_store(binding, settings=settings, for_write=True)
+    resume_from = _identity(workflow, resume_run_id, task_id) if resume_run_id else None
+    if selectors and attempt > 1:
+        prior_binding = _identity(workflow, run_id, task_id, **{**selectors, "attempt": attempt - 1})
+        if result_store._analysis_guard(prior_binding) is not None:
+            resume_from = prior_binding
     return AnalysisWorkUnitCheckpoints(
-        store or _configured_result_store(binding, settings=settings, for_write=True),
+        result_store,
         binding, user_id=user_id, authorize=authorize, attempt_token=attempt_token,
-        resume_from=_identity(workflow, resume_run_id, task_id) if resume_run_id else None,
+        resume_from=resume_from,
         source_authorizer=source_authorizer,
         operation_request=operation_request, operation_sources=operation_sources,
         recover_running_unit=recover_running_unit,

application/single_app/functions_group_workflows.py

@@ -503,7 +503,10 @@ def save_group_workflow(group_id, workflow_data, actor_user_id, user_info=None):
             reference, actor_user_id=actor_user_id,
         )
     task_prompt = _normalize_text(
-        workflow_data.get('task_prompt') or (tasks[0].get('instructions') if tasks else ''),
+        workflow_data.get('task_prompt') or (
+            workflow_name if definition_fields['definition_version'] == 3
+            else tasks[0].get('instructions') if tasks else ''
+        ),
         'Task prompt',
         required=True,
     )

application/single_app/functions_personal_workflows.py

@@ -215,9 +215,15 @@ def _normalize_workflow_tasks(
 
     normalized_tasks = []
     seen_task_ids = set()
+    structured = workflow_data.get('definition_version') == 3
     for index, raw_task in enumerate(raw_tasks):
         if not isinstance(raw_task, dict):
             raise ValueError(f'Workflow task {index + 1} is invalid.')
+        if structured and raw_task.keys() - {
+            'id', 'type', 'name', 'instructions', 'order', 'runner', 'document_action', 'inputs',
+            'reference_ids', 'output_contract', 'approval', 'publication',
+        }:
+            raise ValueError('A structured task contains unsupported executable fields.')
 
         task_type = _normalize_text(raw_task.get('type') or 'instructions', 'Task type').lower()
         if task_type != 'instructions':
@@ -235,7 +241,7 @@ def _normalize_workflow_tasks(
         publication = None
         if raw_task.get('publication') is not None:
             publication = normalize_workflow_publication(raw_task['publication'])
-            if index == 0:
+            if index == 0 and not structured:
                 raise ValueError('Add an analysis task before its publication task.')
             publication_action = raw_task.get('document_action')
             if publication_action is not None and (
@@ -253,6 +259,10 @@ def _normalize_workflow_tasks(
             )
 
         raw_runner = raw_task.get('runner') if isinstance(raw_task.get('runner'), dict) else {}
+        if structured and raw_runner.keys() - {
+            'type', 'selected_agent', 'model_endpoint_id', 'model_id', 'model_provider', 'model_binding_summary',
+        }:
+            raise ValueError('A structured task runner contains unsupported executable fields.')
         runner_type = _normalize_text(raw_runner.get('type') or 'inherit', 'Task runner type').lower()
         if runner_type not in WORKFLOW_TASK_RUNNER_TYPES:
             raise ValueError(f'Workflow task {index + 1} has an unsupported runner type.')
@@ -278,7 +288,7 @@ def _normalize_workflow_tasks(
 
         if callable(task_document_action_normalizer):
             raw_document_action = raw_task.get('document_action')
-            if not isinstance(raw_document_action, dict) and index == 0:
+            if not isinstance(raw_document_action, dict) and index == 0 and not structured:
                 # Workflows saved before per-task documents kept a single workflow-level
                 # action that only ever executed on the first task.
                 raw_document_action = default_document_action
@@ -815,7 +825,10 @@ def save_personal_workflow(user_id, workflow_data, actor_user_id=None):
     for reference in definition_fields.get('reference_inputs', []):
         authorize_workflow_reference({'user_id': user_id}, reference, actor_user_id=modifying_user_id)
     task_prompt = _normalize_text(
-        workflow_data.get('task_prompt') or (tasks[0].get('instructions') if tasks else ''),
+        workflow_data.get('task_prompt') or (
+            workflow_name if definition_fields['definition_version'] == 3
+            else tasks[0].get('instructions') if tasks else ''
+        ),
         'Task prompt',
         required=True,
     )
@@ -1103,7 +1116,7 @@ def is_public_workflow_run_item(item):
     private_types = (
         "workflow_result_chunk", "chat_analysis_result_chunk",
         "orchestration_analysis_result_chunk", "analysis_work_unit_checkpoint",
-        "workflow_runtime_control",
+        "workflow_runtime_control", "workflow_runtime_journal",
     )
     return isinstance(item, dict) and not any(
         item.get(field) in private_types for field in ("type", "item_type")
@@ -1113,10 +1126,10 @@ def is_public_workflow_run_item(item):
 WORKFLOW_PUBLIC_RUN_ITEMS_FILTER = (
     'AND (NOT IS_DEFINED(c.type) OR c.type NOT IN '
     '("workflow_result_chunk", "chat_analysis_result_chunk", '
-    '"orchestration_analysis_result_chunk", "analysis_work_unit_checkpoint", "workflow_runtime_control")) '
+    '"orchestration_analysis_result_chunk", "analysis_work_unit_checkpoint", "workflow_runtime_control", "workflow_runtime_journal")) '
     'AND (NOT IS_DEFINED(c.item_type) OR c.item_type NOT IN '
     '("workflow_result_chunk", "chat_analysis_result_chunk", '
-    '"orchestration_analysis_result_chunk", "analysis_work_unit_checkpoint", "workflow_runtime_control")) '
+    '"orchestration_analysis_result_chunk", "analysis_work_unit_checkpoint", "workflow_runtime_control", "workflow_runtime_journal")) '
 )
 
 

application/single_app/functions_saved_analysis.py

@@ -18,6 +18,7 @@
 from functions_generated_file_exports import build_saved_analysis_export
 from functions_workflow_context import WorkflowContextBudgetError, calculate_workflow_context_budget
 from functions_workflow_result_store import WorkflowResultStorageUnavailableError, _quota_bytes
+from functions_workflow_runtime_store import WorkflowRuntimeConflict
 from functions_workflow_results import (
     ANALYSIS_SOURCE_ACCESS_VERSION,
     WorkflowResultNotReadyError,
@@ -172,18 +173,22 @@ def _load_authorized_workflow(user_id, binding):
     from functions_workflow_runner import _workflow_task_run_item_id
 
     group_id = binding.get("group_id")
+    item_id = (
+        _workflow_task_run_item_id(binding["run_id"], binding["task_id"], binding["execution_id"])
+        if binding.get("execution_id") else _workflow_task_run_item_id(binding["run_id"], binding["task_id"])
+    )
     if group_id:
         assert_group_role(user_id, group_id, allowed_roles=("Owner", "Admin", "DocumentManager", "User"))
         workflow = get_group_workflow(group_id, binding["workflow_id"])
         run = get_group_workflow_run(group_id, binding["run_id"])
         item = get_group_workflow_run_item(
-            binding["run_id"], _workflow_task_run_item_id(binding["run_id"], binding["task_id"]),
+            binding["run_id"], item_id,
         )
     else:
         workflow = get_personal_workflow(user_id, binding["workflow_id"])
         run = get_personal_workflow_run(user_id, binding["run_id"])
         item = get_personal_workflow_run_item(
-            binding["run_id"], _workflow_task_run_item_id(binding["run_id"], binding["task_id"]),
+            binding["run_id"], item_id,
         )
     if (
         not workflow or workflow.get("id") != binding["workflow_id"]
@@ -367,6 +372,8 @@ def workflow_saved_analysis_descriptor(summary, workflow, *, conversation_id, me
             "kind": "workflow", "workflow_id": workflow["id"],
             "run_id": producer["run_id"], "task_id": producer["task_id"],
             "group_id": workflow.get("group_id"),
+            **({key: producer[key] for key in ("node_id", "execution_id", "iteration_path", "attempt")}
+               if producer.get("execution_id") else {}),
         },
         "record_count": summary.get("record_count"),
         "source_count": summary.get("source_count"),
@@ -467,6 +474,15 @@ def analysis_artifact_metadata(producer):
         if not isinstance(value, str) or not value.strip() or len(value.encode("utf-8")) > 1024:
             raise ValueError("The analysis artifact producer is incomplete.")
         normalized[field] = value.strip()
+    if producer["kind"] == "workflow" and producer.get("execution_id"):
+        for field in ("node_id", "execution_id"):
+            value = producer.get(field)
+            if not isinstance(value, str) or not value or len(value) > 128:
+                raise ValueError("The exact analysis execution identity is incomplete.")
+            normalized[field] = value
+        if type(producer.get("attempt")) is not int or producer["attempt"] < 1 or producer.get("iteration_path") != []:
+            raise ValueError("The exact analysis attempt identity is incomplete.")
+        normalized.update(attempt=producer["attempt"], iteration_path=[])
     return {"analysis_result_required": True, "analysis_producer": normalized}
 
 
@@ -501,7 +517,10 @@ def _workflow_analysis_artifact_manifest(user_id, artifact, producer):
     from functions_personal_workflows import get_personal_workflow_run_item
     from functions_workflow_runner import _workflow_task_run_item_id
 
-    item_id = _workflow_task_run_item_id(producer["run_id"], producer["task_id"])
+    item_id = (
+        _workflow_task_run_item_id(producer["run_id"], producer["task_id"], producer["execution_id"])
+        if producer.get("execution_id") else _workflow_task_run_item_id(producer["run_id"], producer["task_id"])
+    )
     items = [
         item for item in (
             get_personal_workflow_run_item(producer["run_id"], item_id),
@@ -516,11 +535,25 @@ def _workflow_analysis_artifact_manifest(user_id, artifact, producer):
     binding = {**producer, "group_id": item.get("group_id")}
     workflow = _load_authorized_workflow(user_id, binding)
     summary = item.get("workflow_result") or {}
+    selectors = {}
+    if producer.get("execution_id"):
+        from functions_workflow_runtime_store import workflow_runtime_store
+
+        journal = workflow_runtime_store(workflow, producer["run_id"])
+        workflow = journal.run_definition()
+        row = journal.journal_read(
+            "attempt", [producer["execution_id"], producer["attempt"]],
+        )
+        if row is None or row["payload"].get("task_id") != producer["task_id"] or row["payload"].get("node_id") != producer["node_id"]:
+            raise AnalysisResultUnavailable("analysis_artifact_unbound")
+        summary = row["payload"].get("workflow_result") or {}
+        selectors = {key: producer[key] for key in ("node_id", "execution_id", "iteration_path", "attempt")}
     reference = summary.get("result_ref")
     if not isinstance(reference, Mapping):
         raise AnalysisResultUnavailable("analysis_artifact_unbound")
     manifest, _ = authorize_workflow_task_result_read(
         workflow, producer["run_id"], producer["task_id"], reference, reader_user_id=user_id,
+        **selectors,
     )
     if not any(
         value.get("artifact_message_id") == artifact.get("id")
@@ -711,11 +744,21 @@ def load_saved_analysis(
             if not isinstance(binding.get(field), str) or not binding[field]:
                 raise AnalysisResultUnavailable("analysis_lineage_invalid")
         workflow = (workflow_getter or _load_authorized_workflow)(user_id, binding)
-        loader = workflow_loader or _workflow_load
-        load = lambda ref: loader(workflow, binding["run_id"], binding["task_id"], ref)
+        selectors = {}
+        if binding.get("execution_id"):
+            from functions_workflow_result_store import load_workflow_node_result
+            from functions_workflow_runtime_store import workflow_runtime_store
+
+            workflow = workflow_runtime_store(workflow, binding["run_id"]).run_definition()
+            selectors = {key: binding[key] for key in ("node_id", "execution_id", "iteration_path", "attempt")}
+            loader = workflow_loader or load_workflow_node_result
+        else:
+            loader = workflow_loader or _workflow_load
+        load = lambda ref: loader(workflow, binding["run_id"], binding["task_id"], ref, **selectors)
         manifest, checked = authorize_workflow_task_result_read(
             workflow, binding["run_id"], binding["task_id"], reference,
             reader_user_id=user_id, load_result=loader, source_resolver=source_resolver,
+            **selectors,
         )
     elif binding.get("kind") == "orchestration":
         for field in ("user_id", "conversation_id", "run_id", "step_id"):
@@ -1647,33 +1690,44 @@ def sanitize_workflow_analysis_history(workflow, run_record, user_id, *, items=N
     tasks = list(run_record.get("task_results") or []) + list(items or [])
     denied = set()
     checked = {}
+    bound_workflow = None
     for task in tasks:
         if not isinstance(task, Mapping):
             continue
         task_id = task.get("task_id")
         summary = task.get("workflow_result") or {}
-        candidates = [summary] if isinstance(summary, Mapping) and summary.get("analysis_result") else []
+        candidates = [summary] if isinstance(summary, Mapping) and (
+            summary.get("analysis_result") or summary.get("contract_version") == "workflow-result-v2"
+        ) else []
         candidates.extend(
             consumed for consumed in task.get("consumed_inputs") or []
-            if isinstance(consumed, Mapping) and consumed.get("analysis_result")
+            if isinstance(consumed, Mapping) and (consumed.get("analysis_result") or (consumed.get("producer") or {}).get("execution_id"))
         )
         for candidate in candidates:
             producer = candidate.get("producer") or {}
             reference = candidate.get("result_ref") or {}
-            key = (producer.get("run_id"), producer.get("task_id"), reference.get("sha256"))
+            key = (producer.get("run_id"), producer.get("task_id"), producer.get("execution_id"),
+                   producer.get("attempt"), reference.get("sha256"))
             if key not in checked:
                 try:
                     if (
                         producer.get("workflow_id") != workflow.get("id")
                         or producer.get("run_id") != run_record.get("id")
-                        or not producer.get("task_id")
+                        or not producer.get("task_id") and not producer.get("execution_id")
                     ):
                         raise AnalysisResultUnavailable("analysis_lineage_invalid")
+                    selectors = {}
+                    if producer.get("execution_id"):
+                        selectors = {name: producer.get(name) for name in ("node_id", "execution_id", "iteration_path", "attempt")}
+                        if result_reader is None and bound_workflow is None:
+                            from functions_workflow_runtime_store import workflow_runtime_store
+
+                            bound_workflow = workflow_runtime_store(workflow, producer["run_id"]).run_definition()
                     read(
-                        workflow, producer["run_id"], producer["task_id"], reference,
-                        reader_user_id=user_id,
+                        bound_workflow or workflow, producer["run_id"], producer.get("task_id"), reference,
+                        reader_user_id=user_id, **selectors,
                     )
-                except (PermissionError, LookupError, ValueError, AzureError, WorkflowResultStorageUnavailableError) as exc:
+                except (PermissionError, LookupError, ValueError, AzureError, WorkflowResultStorageUnavailableError, WorkflowRuntimeConflict) as exc:
                     log_event(
                         "[DOCUMENT_ANALYSIS] Workflow analysis preview withheld.",
                         extra={"run_id": run_record.get("id"), "task_id": task_id, "error_type": type(exc).__name__},

application/single_app/functions_workflow_activity.py

@@ -117,6 +117,7 @@ def _serialize_run(run_record):
                 field: task.get(field)
                 for field in (
                     'task_id', 'task_name', 'task_order', 'status',
+                    'execution_id', 'node_id', 'iteration_path', 'attempt',
                     'workflow_result', 'context_budget', 'consumed_inputs', 'workflow_validation',
                 )
             }

application/single_app/functions_workflow_definition_store.py

@@ -36,9 +36,10 @@ def save_workflow_definition_record(container, partition_key, workflow, existing
             raise WorkflowDefinitionConflict("This workflow is being deleted. Your draft was not saved.")
         if workflow_definition_revision(current) != expected:
             raise WorkflowDefinitionConflict("This workflow changed since it was opened. Reload it before saving.")
-        if workflow.get("definition_version") == 2 and current.get("active_run_id"):
+        if workflow.get("definition_version") in {2, 3} and current.get("active_run_id"):
             raise WorkflowDefinitionConflict("An active run started while editing. Wait or cancel it before saving.")
-        body = dict(workflow)
+        body = {key: value for key, value in current.items() if not key.startswith("_")}
+        body.update(workflow)
         body.update({key: current[key] for key in WORKFLOW_RUNTIME_FIELDS if key in current})
         # A schedule edit intentionally computes a new next run; preserve the
         # live scheduler value only when its authored inputs stayed the same.