Add M4B serial workflow loops and exact Collect

application/single_app/admin_settings_fields.py

@@ -132,6 +132,13 @@
     normalize_terms_of_use_redirect_url,
     normalize_terms_of_use_text,
 )
+from functions_workflow_limits import (
+    WORKFLOW_LOOP_ITEMS_DEFAULT,
+    WORKFLOW_LOOP_ITEMS_MAX,
+    WORKFLOW_LOOP_ITEMS_MIN,
+    WorkflowLoopLimitError,
+    validate_workflow_max_loop_items,
+)
 
 HEX_COLOR_PATTERN = re.compile(r"^#[0-9a-fA-F]{6}$")
 
@@ -3987,6 +3994,20 @@
             "max": 100,
             "step": 1,
         },
+        {
+            "key": "workflow_max_loop_items",
+            "type": "number",
+            "label": "Workflow Loop Item Limit",
+            "help": (
+                "Maximum actual items visited by each For each loop in a new personal "
+                "or group workflow run. Authors may choose a lower maximum. Oversized "
+                "inputs are rejected, never truncated. Active runs keep their admitted limit."
+            ),
+            "default": WORKFLOW_LOOP_ITEMS_DEFAULT,
+            "min": WORKFLOW_LOOP_ITEMS_MIN,
+            "max": WORKFLOW_LOOP_ITEMS_MAX,
+            "step": 1,
+        },
     ],
     # --- Agents & Actions -------------------------------------------------
     #
@@ -6555,6 +6576,12 @@ def _normalize_field_value(key, value, field):
             else f"{key} cannot be changed through this endpoint."
         ), None
 
+    if key == "workflow_max_loop_items":
+        try:
+            return validate_workflow_max_loop_items(value), None, None
+        except WorkflowLoopLimitError as error:
+            return None, error.public_message, None
+
     if field_type == "switch":
         return _coerce_bool(value), None, None
 

application/single_app/agent_delegation_runtime.py

@@ -246,6 +246,9 @@ async def _target_messages(target, task, context, frame, settings):
 async def execute_target(target, task, context, frame):
     """Invoke precisely this canonical target; never select a default or by name."""
     from functions_settings import get_settings
+    from functions_workflow_loop_runners import assert_workflow_loop_agent_type
+
+    assert_workflow_loop_agent_type(target.get("agent_type", "local"))
 
     bridge = frame.identity.bridge(target) if frame.identity.bridge else nullcontext()
     kernel = None
@@ -514,6 +517,9 @@ def invoke_stream(self, messages, **kwargs):
 
 def prepare_agent_execution(agent, reference, *, user_id, settings, conversation_id=None,
                             cancel_requested=None, budget=None, identity=None, prevent_replay=False):
+    from functions_workflow_loop_runners import assert_workflow_loop_agent_type
+
+    assert_workflow_loop_agent_type(getattr(agent, "agent_type", "local"))
     if str(getattr(agent, "agent_type", "local") or "local").lower() != "local":
         return agent
     identity = identity or capture_execution_identity(user_id, conversation_id)

application/single_app/config.py

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

application/single_app/functions_document_access_index.py

@@ -2468,7 +2468,7 @@ def _is_backfill_state_ready_for_scope(state, source_scope):
     return source_scope in completed_scopes
 
 
-def _get_document_access_index_readiness(source_scope, settings=None):
+def _get_document_access_index_readiness(source_scope, settings=None, *, read_only=False):
     normalized_settings = get_document_access_index_settings(settings)
     if not normalized_settings.get('container_enabled'):
         return {
@@ -2483,7 +2483,7 @@ def _get_document_access_index_readiness(source_scope, settings=None):
             'settings': normalized_settings,
         }
     try:
-        state = _read_backfill_state()
+        state = _read_backfill_state(use_cache=False) if read_only else _read_backfill_state()
     except Exception as exc:
         log_event(
             '[DOCUMENT_ACCESS_INDEX] DAI read path readiness check failed; source document read should be used.',
@@ -2504,7 +2504,17 @@ def _get_document_access_index_readiness(source_scope, settings=None):
             'backfill_status': (state or {}).get('status'),
         }
 
-    has_repair_backlog = has_document_access_index_repair_backlog()
+    if read_only:
+        # Advisory selection must not initialize or repair catalog state.
+        try:
+            backlog_state = _read_repair_backlog_state(use_cache=False)
+            has_repair_backlog = bool(
+                isinstance(backlog_state, dict) and backlog_state.get('has_repair_backlog')
+            ) or _query_repair_backlog_exists()
+        except Exception:
+            has_repair_backlog = None
+    else:
+        has_repair_backlog = has_document_access_index_repair_backlog()
     if has_repair_backlog is None:
         return {
             'ready': False,
@@ -2528,6 +2538,88 @@ def _get_document_access_index_readiness(source_scope, settings=None):
     }
 
 
+class DocumentAccessIndexEnumerationError(RuntimeError):
+    """A complete, read-only catalog enumeration could not be established."""
+
+    def __init__(self, code='document_catalog_unavailable'):
+        self.code = code
+        super().__init__('The document catalog is temporarily unavailable. Try again later.')
+
+
+def iter_document_access_index_candidates(
+    source_scope, *, user_id=None, group_ids=None, public_workspace_ids=None,
+    settings=None, page_size=100, check=None,
+):
+    """Page current candidate IDs without the preview helper's 1,001-row ceiling.
+
+    These projection rows are not permission grants. The caller must authorize
+    each requested scope before enumeration and each source before consumption.
+    """
+    if source_scope not in DOCUMENT_ACCESS_SOURCE_SCOPES:
+        raise DocumentAccessIndexEnumerationError('invalid_source_scope')
+    if type(page_size) is not int or not 1 <= page_size <= 1000:
+        raise DocumentAccessIndexEnumerationError('invalid_page_size')
+    scope_keys = list(dict.fromkeys(_build_shadow_scope(
+        source_scope, user_id=user_id, group_ids=group_ids,
+        public_workspace_ids=public_workspace_ids,
+    )))
+    if not scope_keys or not all(scope_keys):
+        raise DocumentAccessIndexEnumerationError('missing_scope_keys')
+    if len(scope_keys) > DOCUMENT_ACCESS_BOUNDED_CATALOG_MAX_SCOPES:
+        raise DocumentAccessIndexEnumerationError('scope_limit_exceeded')
+
+    def check_readiness():
+        if check is not None:
+            check()
+        readiness = _get_document_access_index_readiness(
+            source_scope, settings=settings, read_only=True,
+        )
+        if not readiness.get('ready'):
+            raise DocumentAccessIndexEnumerationError()
+
+    for scope_key in scope_keys:
+        check_readiness()
+        result = cosmos_document_access_index_container.query_items(
+            query=(
+                'SELECT c.document_id, c.source_document_id, c.version, c.revision_family_id '
+                'FROM c WHERE c.type = @type AND c.source_scope = @source_scope '
+                'AND c.scope_key = @scope_key AND c.access_granted = true '
+                'AND c.is_current_version = true AND c.projection_version = @projection_version '
+                'ORDER BY c.document_id ASC'
+            ),
+            parameters=[
+                {'name': '@type', 'value': DOCUMENT_ACCESS_INDEX_TYPE},
+                {'name': '@source_scope', 'value': source_scope},
+                {'name': '@scope_key', 'value': scope_key},
+                {'name': '@projection_version', 'value': DOCUMENT_ACCESS_INDEX_SCHEMA_VERSION},
+            ],
+            partition_key=scope_key,
+            max_item_count=page_size,
+        )
+        if not callable(getattr(result, 'by_page', None)):
+            raise DocumentAccessIndexEnumerationError('document_catalog_paging_unavailable')
+        pages = iter(result.by_page())
+        while True:
+            check_readiness()
+            page = next(pages, None)
+            if page is None:
+                if getattr(pages, 'continuation_token', None):
+                    raise DocumentAccessIndexEnumerationError('document_catalog_continuation_failed')
+                break
+            for row in page:
+                if not isinstance(row, dict) or not (
+                    row.get('source_document_id') or row.get('document_id')
+                ):
+                    raise DocumentAccessIndexEnumerationError('document_catalog_invalid')
+                yield row
+        check_readiness()
+
+
+def document_matches_list_filters(document, filters=None):
+    """Apply the shared document-list metadata semantics to a current document."""
+    return _matches_shadow_filters(document, filters)
+
+
 def query_document_access_index_documents(
     source_scope,
     user_id=None,

application/single_app/functions_group_workflows.py

@@ -68,7 +68,7 @@
 def _apply_group_document_action_scope(group_id, action_config):
     """Force a normalized document action to stay inside the owning group workspace."""
     action_config = action_config if isinstance(action_config, dict) else {'type': 'none'}
-    if action_config.get('type') == 'none':
+    if action_config.get('type') == 'none' or action_config.get('target_mode') == 'current_item':
         return action_config
 
     action_config['doc_scope'] = 'group'
@@ -488,6 +488,7 @@ def save_group_workflow(group_id, workflow_data, actor_user_id, user_info=None):
                 action_payload,
                 allow_empty_file_sync_targets=allow_empty_file_sync_targets,
                 settings=settings,
+                allow_current_item=workflow_data.get('definition_version') == 3,
             ),
         ),
         default_document_action=document_action,
@@ -658,6 +659,10 @@ def save_group_workflow(group_id, workflow_data, actor_user_id, user_info=None):
         workflow['next_run_at'] = None
 
     workflow.update(definition_fields)
+    if workflow.get('definition_version') == 3:
+        from functions_workflow_loop_runners import validate_workflow_loop_runners
+
+        validate_workflow_loop_runners(workflow, actor_user_id=actor_user_id, settings=settings)
     result = save_workflow_definition_record(
         cosmos_group_workflows_container, group_id, workflow, existing_workflow,
     )

application/single_app/functions_personal_workflows.py

@@ -221,7 +221,7 @@ def _normalize_workflow_tasks(
             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',
+            'reference_ids', 'output_contract', 'approval', 'publication', 'input_processing',
         }:
             raise ValueError('A structured task contains unsupported executable fields.')
 
@@ -362,7 +362,8 @@ def _normalize_document_action_config(workflow_data, existing_workflow=None, all
     )
 
 
-def _normalize_task_document_action_config(action_payload, allow_empty_file_sync_targets=False, settings=None):
+def _normalize_task_document_action_config(action_payload, allow_empty_file_sync_targets=False, settings=None,
+                                           allow_current_item=False):
     """Normalize a single workflow task's document action payload."""
     source_settings = settings if isinstance(settings, dict) else get_settings()
     action_payload = action_payload if isinstance(action_payload, dict) else {'type': 'none'}
@@ -371,6 +372,21 @@ def _normalize_task_document_action_config(action_payload, allow_empty_file_sync
         settings=source_settings,
     )
     allowed_action_types = get_enabled_document_action_types(settings=source_settings)
+    if action_payload.get('target_mode') == 'current_item':
+        if (
+            not allow_current_item or action_payload.get('type') != DOCUMENT_ACTION_TYPE_ANALYZE
+            or DOCUMENT_ACTION_TYPE_ANALYZE not in allowed_action_types
+        ):
+            raise ValueError('Current-item Analyze requires an enabled structured document loop.')
+        if action_payload.keys() - {'type', 'target_mode', 'loop_id', 'analysis_mode'}:
+            raise ValueError('Current-item Analyze cannot supply another document selection.')
+        if action_payload.get('analysis_mode', 'combined') != 'combined':
+            raise ValueError('Current-item Analyze produces the current document as one analysis.')
+        return {
+            'type': DOCUMENT_ACTION_TYPE_ANALYZE, 'target_mode': 'current_item',
+            'loop_id': _normalize_text(action_payload.get('loop_id'), 'Loop id', required=True),
+            'analysis_mode': 'combined',
+        }
 
     if allow_empty_file_sync_targets:
         action_type = str(action_payload.get('type') or '').strip().lower()
@@ -815,6 +831,7 @@ def save_personal_workflow(user_id, workflow_data, actor_user_id=None):
             action_payload,
             allow_empty_file_sync_targets=allow_empty_file_sync_targets,
             settings=settings,
+            allow_current_item=workflow_data.get('definition_version') == 3,
         ),
         default_document_action=document_action,
     )
@@ -978,6 +995,10 @@ def save_personal_workflow(user_id, workflow_data, actor_user_id=None):
         workflow['next_run_at'] = None
 
     workflow.update(definition_fields)
+    if workflow.get('definition_version') == 3:
+        from functions_workflow_loop_runners import validate_workflow_loop_runners
+
+        validate_workflow_loop_runners(workflow, actor_user_id=modifying_user_id, settings=settings)
     result = save_workflow_definition_record(
         cosmos_personal_workflows_container, user_id, workflow, existing_workflow,
     )