Plan replay: repeat an approved chat plan as a personal workflow (#1550)

application/single_app/admin_settings_fields.py

@@ -4382,6 +4382,29 @@
                 {"key": "allow_user_workflows", "equals": True},
             ],
         },
+        {
+            "key": "enable_workflow_plan_replay",
+            "type": "switch",
+            "label": "Repeat Chat Plans On A Schedule",
+            "help": (
+                "Lets a user save a completed plan from their own private conversation as a personal "
+                "workflow that repeats the same steps on a schedule, such as every Monday. The user "
+                "sees every step and the schedule before creating it. Each run replays the saved "
+                "steps exactly, as that user, in the workflow's own conversation, and checks the "
+                "user's access again first. Only steps that search, analyze, compare or merge "
+                "documents, inspect tabular files, write the answer or generate images can be "
+                "repeated. Web search and other steps that need a signed-in session can't be "
+                "repeated. Off by default, because a saved plan "
+                "keeps running with the user's access. Turning it off makes existing repeat "
+                "workflows stop with a clear reason instead of deleting them. Requires Chat "
+                "Orchestration and Enable Personal Workflows."
+            ),
+            "default": False,
+            "depends_on": [
+                {"key": "enable_chat_orchestration", "equals": True},
+                {"key": "allow_user_workflows", "equals": True},
+            ],
+        },
         {
             "key": "chat_orchestration_enabled_capabilities",
             "type": "checkbox_set",

application/single_app/config.py

@@ -101,7 +101,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.307"
+VERSION = "0.261.308"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 # Opt-out for deployments where App Service Easy Auth is active but the platform

application/single_app/functions_group_workflows.py

@@ -483,6 +483,14 @@ def build_group_workflow_document(group_id, workflow_data, actor_user_id, user_i
     and ``origin`` belong to a server create path, never to a save payload.
     """
     workflow_data = workflow_data if isinstance(workflow_data, dict) else {}
+    if workflow_data.get('plan_replay') is not None or any(
+        isinstance(task, dict) and task.get('type') == 'plan_replay'
+        for task in (workflow_data.get('tasks') if isinstance(workflow_data.get('tasks'), list) else [])
+    ):
+        # Plan replay is personal-only; importing here avoids a workflow-store import cycle.
+        from functions_workflow_plan_replay import PlanReplaySaveError
+
+        raise PlanReplaySaveError('group_not_supported')
     settings_options = {'settings': settings} if settings is not None else {}
     settings = get_settings() if settings is None else settings
     now_iso = _utc_now_iso()

application/single_app/functions_orchestration_registry.py

@@ -129,6 +129,10 @@ def effective_capability_ids(values):
 # workflow. Hand-off also needs proposals, runs and results turned on, because it creates a
 # workflow, starts its one run and posts that run's result back into the chat.
 WORKFLOW_HANDOFF_SETTING = 'enable_chat_orchestration_workflow_handoff'
+# The settings key that must be exactly True before a completed plan can be saved as a personal
+# workflow that replays it on a schedule, and before an existing replay workflow may run. It is not a
+# capability: a replay is never a step the planner can choose, only a user's explicit save.
+WORKFLOW_PLAN_REPLAY_SETTING = 'enable_workflow_plan_replay'
 # The tasks a workflow proposal may hold, and the action kinds the planner may say a task needs.
 # The workflow planning context and the deliverables import these, so each has one definition.
 # The draft service's own task limit (functions_workflow_drafts.BLUEPRINT_MAX_TASKS) stays a

application/single_app/functions_orchestration_scheduler.py

@@ -70,6 +70,8 @@ def enumerate_due_runs(container, *, now=None, limit=64):
         "AND (NOT IS_DEFINED(c.checkpoints_deleted) OR c.checkpoints_deleted != true) "
         "AND (NOT IS_DEFINED(c.outputs_deleted) OR c.outputs_deleted != true) "
         "AND (NOT IS_DEFINED(c.superseded_by_run_id) OR IS_NULL(c.superseded_by_run_id)) "
+        # A saved-workflow plan replay is driven by its workflow run, never by this scheduler.
+        "AND (NOT IS_DEFINED(c.workflow_replay) OR IS_NULL(c.workflow_replay)) "
         "AND (NOT IS_DEFINED(c.latest_attempt_run_id) OR IS_NULL(c.latest_attempt_run_id) "
         'OR c.latest_attempt_run_id = "") '
         "AND (NOT IS_DEFINED(c.execution_lease) OR IS_NULL(c.execution_lease) "
@@ -454,6 +456,10 @@ def _process_run(self, selector):
         lease = None
         try:
             record = self.read_run(selector)
+            if record.get("workflow_replay"):
+                # The owning workflow run drives, cancels and fences its plan replay.
+                self.result["runs"].append({**selector, "action": "deferred", "state": "workflow_replay"})
+                return
             states = {row["step_id"]: row["status"] for row in record.get("execution_steps") or []}
             execution_work = any(
                 step.get("enabled", True) and step["role"] != "render"

application/single_app/functions_personal_workflows.py

@@ -900,7 +900,7 @@ def get_due_personal_workflows(limit=20):
 
 def build_personal_workflow_document(user_id, workflow_data, actor_user_id=None, *, settings=None, workflow_id=None,
                                      origin=None, user_settings_reader=None, resolve_document=None,
-                                     sanitize_source=None, one_time=False):
+                                     sanitize_source=None, one_time=False, plan_replay_task=None):
     """Normalize and authorize a personal workflow exactly as saving it would, without writing.
 
     Returns ``(workflow, existing_workflow)``: the document a save persists and the stored workflow
@@ -937,6 +937,37 @@ def build_personal_workflow_document(user_id, workflow_data, actor_user_id=None,
         existing_workflow = get_personal_workflow(user_id, workflow_id) if workflow_id else None
         refuse_save_of_deleted_workflow(cosmos_personal_workflows_container, user_id, workflow_data, existing_workflow)
 
+    incoming_tasks = workflow_data.get('tasks') if isinstance(workflow_data.get('tasks'), list) else []
+    existing_plan_replay_task = next((
+        task for task in (existing_workflow or {}).get('tasks') or []
+        if isinstance(task, dict) and task.get('type') == 'plan_replay' and isinstance(task.get('plan_replay'), dict)
+    ), None)
+    incoming_replay_task = any(
+        isinstance(task, dict) and task.get('type') == 'plan_replay'
+        for task in incoming_tasks
+    )
+    has_replay_payload = (
+        plan_replay_task is not None
+        or existing_plan_replay_task is not None
+        or workflow_data.get('plan_replay') is not None
+        or incoming_replay_task
+    )
+    attach_plan_replay = None
+    if has_replay_payload:
+        # Plan replay is a server-owned task envelope; importing here avoids a workflow-store import cycle.
+        from functions_workflow_plan_replay import (
+            PlanReplaySaveError,
+            attach_plan_replay,
+            normalize_plan_replay_update,
+        )
+
+        if plan_replay_task is not None and existing_workflow is not None:
+            raise PlanReplaySaveError('plan_replay_read_only')
+        if existing_plan_replay_task:
+            workflow_data = normalize_plan_replay_update(existing_workflow, workflow_data)
+        elif plan_replay_task is None:
+            raise PlanReplaySaveError('plan_replay_read_only')
+
     workflow_name = _normalize_text(workflow_data.get('name'), 'Workflow name', required=True)
     description = _normalize_text(workflow_data.get('description'), 'Description')
     file_sync = _normalize_file_sync_config(
@@ -970,6 +1001,10 @@ def build_personal_workflow_document(user_id, workflow_data, actor_user_id=None,
         ),
         default_document_action=document_action,
     )
+    if plan_replay_task is not None:
+        tasks = [attach_plan_replay(tasks[0], plan_replay_task)]
+    elif existing_plan_replay_task:
+        tasks = [attach_plan_replay(tasks[0], existing_plan_replay_task['plan_replay'])]
     definition_fields = normalize_workflow_definition(
         workflow_data, existing_workflow, tasks, user_id=user_id,
     )
@@ -1068,7 +1103,8 @@ def build_personal_workflow_document(user_id, workflow_data, actor_user_id=None,
             workflow_data.get('schedule'),
             existing_workflow=existing_workflow,
             settings=settings,
-            **({'orchestration': True} if origin is not None else {}),
+            # A saved chat plan keeps the chat-workflow floor whenever its schedule is saved.
+            **({'orchestration': True} if origin is not None or existing_plan_replay_task is not None else {}),
         )
 
     workflow = {
@@ -1166,7 +1202,7 @@ def save_personal_workflow(user_id, workflow_data, actor_user_id=None):
 
 def create_personal_workflow_if_absent(user_id, workflow_data, *, workflow_id, origin, actor_user_id=None,
                                        settings=None, user_settings_reader=None, resolve_document=None,
-                                       sanitize_source=None, one_time=False):
+                                       sanitize_source=None, one_time=False, plan_replay_task=None):
     """Create a personal workflow under a server-chosen id and origin, at most once.
 
     Chat orchestration derives ``workflow_id`` from the proposal the user accepted, so accepting the
@@ -1190,7 +1226,7 @@ def create_personal_workflow_if_absent(user_id, workflow_data, *, workflow_id, o
         user_id, {} if workflow_data is None else workflow_data, actor_user_id,
         settings=settings, workflow_id=workflow_id, origin=origin,
         user_settings_reader=user_settings_reader, resolve_document=resolve_document,
-        sanitize_source=sanitize_source, one_time=one_time,
+        sanitize_source=sanitize_source, one_time=one_time, plan_replay_task=plan_replay_task,
     )
     record, created = create_workflow_definition_record_if_absent(
         cosmos_personal_workflows_container, user_id, workflow,

application/single_app/functions_saved_analysis.py

@@ -1950,6 +1950,8 @@
         for field in ("output_summary", "response_preview", "reply"):
             if field in redacted:
                 redacted[field] = UNVERIFIED_WORKFLOW_OUTPUT_MESSAGE
+        # A saved chat plan's typed projection carries the answer text, so it is withheld with the preview.
+        redacted.pop("plan_replay", None)
         if redacted.get("error"):
             redacted["error"] = UNVERIFIED_WORKFLOW_OUTPUT_MESSAGE
         redacted["analysis_access_available"] = False

application/single_app/functions_settings.py

@@ -1648,6 +1648,10 @@ def get_settings(use_cosmos=False, include_source=False):
         # user approves on the plan and again on a hand-off card. Off by default and separate from
         # proposals and runs, so an upgrade never turns it on by itself.
         'enable_chat_orchestration_workflow_handoff': False,
+        # Lets a user save a completed chat plan as a personal workflow that replays the frozen plan
+        # as that user on a schedule. Off by default; turning it off makes existing replay
+        # workflows refuse to run with a fixed reason rather than deleting them.
+        'enable_workflow_plan_replay': False,
         # Workflows that a chat plan creates for a user: how many one user may hold, and the
         # shortest interval one may run on (also never shorter than the general workflow minimum).
         'chat_orchestration_max_workflows_per_user': CHAT_ORCHESTRATION_MAX_WORKFLOWS_DEFAULT,
@@ -2476,6 +2480,12 @@ def update_settings(new_settings, *, expected_etag=None):
                 new_settings['enable_chat_orchestration_workflow_handoff'] is True
             ),
         }
+    if isinstance(new_settings, dict) and 'enable_workflow_plan_replay' in new_settings:
+        # Only a real boolean true lets chat plans be repeated on a schedule; anything else saves as off.
+        new_settings = {
+            **new_settings,
+            'enable_workflow_plan_replay': new_settings['enable_workflow_plan_replay'] is True,
+        }
     if isinstance(new_settings, dict) and 'chat_orchestration_max_workflow_handoffs_per_day' in new_settings:
         new_settings = {
             **new_settings,