feat(scheduler): workflow recipes with confirm-before-create

Store self-contained playbook recipes on scheduled jobs, add schedule_propose draft gate, and run playbook instructions at fire time instead of only short reminders.

Co-authored-by: Cursor <cursoragent@cursor.com>

runtime/scheduler/recipe.py

@@ -0,0 +1,339 @@
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+
+COMPLEX_PROMPT_HINTS = (
+    "刚才",
+    "之前",
+    "继续",
+    "按刚才",
+    "同样的",
+    "那件事",
+    "那套",
+    "流程",
+    "步骤",
+    "生成",
+    "报告",
+    "文档",
+    "pdf",
+    "xlsx",
+    "附件",
+    "发群",
+    "发给",
+    "执行",
+    "整理",
+    "汇总",
+    "拉取",
+    "爬取",
+    "监控",
+    "same as",
+    "continue",
+    "as before",
+    "workflow",
+    "report",
+    "generate",
+    "attach",
+)
+
+
+def _as_str_list(raw: Any) -> list[str]:
+    if raw is None:
+        return []
+    if isinstance(raw, str):
+        item = raw.strip()
+        return [item] if item else []
+    if not isinstance(raw, list):
+        return []
+    out: list[str] = []
+    for item in raw:
+        text = str(item or "").strip()
+        if text:
+            out.append(text)
+    return out
+
+
+def _as_str_dict(raw: Any) -> dict[str, str]:
+    if not isinstance(raw, dict):
+        return {}
+    out: dict[str, str] = {}
+    for key, value in raw.items():
+        k = str(key or "").strip()
+        if not k:
+            continue
+        out[k] = str(value if value is not None else "").strip()
+    return out
+
+
+def parse_recipe_arg(raw: Any) -> dict[str, Any] | None:
+    if raw is None:
+        return None
+    if isinstance(raw, dict):
+        return raw
+    if isinstance(raw, str) and raw.strip():
+        try:
+            data = json.loads(raw)
+            return data if isinstance(data, dict) else None
+        except Exception:
+            return None
+    return None
+
+
+def normalize_recipe(raw: Any) -> dict[str, Any]:
+    data = parse_recipe_arg(raw) or {}
+    goal = str(data.get("goal") or "").strip()
+    steps = _as_str_list(data.get("steps"))
+    constraints = _as_str_list(data.get("constraints"))
+    success_criteria = _as_str_list(data.get("success_criteria") or data.get("successCriteria"))
+    inputs_raw = data.get("inputs") if isinstance(data.get("inputs"), dict) else {}
+    constants = _as_str_dict(inputs_raw.get("constants"))
+    from_context = _as_str_list(inputs_raw.get("from_context") or inputs_raw.get("fromContext"))
+    output_raw = data.get("output") if isinstance(data.get("output"), dict) else {}
+    source_raw = data.get("source") if isinstance(data.get("source"), dict) else {}
+    version = int(data.get("version") or 1)
+    recipe: dict[str, Any] = {
+        "version": max(1, version),
+        "goal": goal,
+        "steps": steps,
+        "constraints": constraints,
+        "success_criteria": success_criteria,
+        "inputs": {
+            "constants": constants,
+            "from_context": from_context,
+        },
+        "output": {
+            "style": str(output_raw.get("style") or "channel_update").strip() or "channel_update",
+            "need_attachments": bool(output_raw.get("need_attachments") or output_raw.get("needAttachments")),
+        },
+        "source": {
+            "session_id": str(source_raw.get("session_id") or source_raw.get("sessionId") or "").strip(),
+            "compiled_at": str(source_raw.get("compiled_at") or source_raw.get("compiledAt") or "").strip(),
+        },
+    }
+    return recipe
+
+
+def recipe_is_empty(recipe: dict[str, Any] | None) -> bool:
+    if not recipe:
+        return True
+    norm = normalize_recipe(recipe)
+    return not (norm.get("goal") or norm.get("steps") or norm.get("success_criteria") or norm.get("constraints"))
+
+
+def recipe_has_playbook(recipe: dict[str, Any] | None) -> bool:
+    if not recipe:
+        return False
+    norm = normalize_recipe(recipe)
+    steps = list(norm.get("steps") or [])
+    goal = str(norm.get("goal") or "").strip()
+    return bool(goal) and len(steps) >= 2
+
+
+def recipe_missing_fields(recipe: dict[str, Any] | None) -> list[str]:
+    norm = normalize_recipe(recipe or {})
+    missing: list[str] = []
+    if not str(norm.get("goal") or "").strip():
+        missing.append("goal")
+    steps = list(norm.get("steps") or [])
+    if len(steps) < 2:
+        missing.append("steps")
+    if not list(norm.get("success_criteria") or []):
+        missing.append("success_criteria")
+    return missing
+
+
+def looks_like_complex_schedule_prompt(prompt_text: str, *, recipe: dict[str, Any] | None = None) -> bool:
+    """Heuristic: multi-step / referential prompts need a recipe."""
+    if recipe_has_playbook(recipe):
+        return True
+    text = str(prompt_text or "").strip()
+    if not text:
+        return False
+    low = text.lower()
+    if any(hint in low or hint in text for hint in COMPLEX_PROMPT_HINTS):
+        # Short pure reminders like "提醒喝水" should stay simple.
+        if len(text) <= 12 and ("提醒" in text or "remind" in low) and "步骤" not in text:
+            return False
+        return True
+    if len(text) >= 80:
+        return True
+    if text.count("\n") >= 2:
+        return True
+    if re.search(r"(1[\.\)]|第一步|step\s*1)", text, flags=re.I):
+        return True
+    return False
+
+
+def prompt_summary_from_recipe(recipe: dict[str, Any] | None, *, fallback: str = "") -> str:
+    norm = normalize_recipe(recipe or {})
+    goal = str(norm.get("goal") or "").strip()
+    if goal:
+        return goal
+    steps = list(norm.get("steps") or [])
+    if steps:
+        return steps[0]
+    return str(fallback or "").strip()
+
+
+def preview_markdown(
+    *,
+    name: str,
+    schedule_kind: str,
+    schedule_expr: str,
+    timezone_name: str,
+    recipe: dict[str, Any],
+    lang: str = "zh",
+) -> str:
+    norm = normalize_recipe(recipe)
+    is_en = str(lang or "").lower().startswith("en")
+    lines: list[str] = []
+    if is_en:
+        lines.append("## Scheduled workflow draft (confirm before create)")
+        lines.append(f"- **Name**: {name or '(untitled)'}")
+        lines.append(f"- **Schedule**: `{schedule_kind}` `{schedule_expr}` ({timezone_name or 'system'})")
+        lines.append(f"- **Goal**: {norm.get('goal') or '(missing)'}")
+        lines.append("### Steps")
+        for i, step in enumerate(list(norm.get("steps") or []), start=1):
+            lines.append(f"{i}. {step}")
+        constraints = list(norm.get("constraints") or [])
+        if constraints:
+            lines.append("### Constraints")
+            for item in constraints:
+                lines.append(f"- {item}")
+        criteria = list(norm.get("success_criteria") or [])
+        if criteria:
+            lines.append("### Success criteria")
+            for item in criteria:
+                lines.append(f"- {item}")
+        constants = dict((norm.get("inputs") or {}).get("constants") or {})
+        if constants:
+            lines.append("### Fixed inputs")
+            for key, value in constants.items():
+                lines.append(f"- `{key}`: {value}")
+        lines.append("")
+        lines.append("Reply **confirm** to create, or list the edits you want.")
+    else:
+        lines.append("## 定时工作流草稿（请确认后再创建）")
+        lines.append(f"- **名称**：{name or '（未命名）'}")
+        lines.append(f"- **时间**：`{schedule_kind}` `{schedule_expr}`（{timezone_name or '系统时区'}）")
+        lines.append(f"- **目标**：{norm.get('goal') or '（缺失）'}")
+        lines.append("### 步骤")
+        for i, step in enumerate(list(norm.get("steps") or []), start=1):
+            lines.append(f"{i}. {step}")
+        constraints = list(norm.get("constraints") or [])
+        if constraints:
+            lines.append("### 约束")
+            for item in constraints:
+                lines.append(f"- {item}")
+        criteria = list(norm.get("success_criteria") or [])
+        if criteria:
+            lines.append("### 成功标准")
+            for item in criteria:
+                lines.append(f"- {item}")
+        constants = dict((norm.get("inputs") or {}).get("constants") or {})
+        if constants:
+            lines.append("### 固定输入")
+            for key, value in constants.items():
+                lines.append(f"- `{key}`：{value}")
+        lines.append("")
+        lines.append("请回复**确认**以创建，或说明要修改的地方。")
+    return "\n".join(lines).strip()
+
+
+def compile_playbook_instruction(*, recipe: dict[str, Any], lang: str = "zh") -> str:
+    norm = normalize_recipe(recipe)
+    is_en = str(lang or "").lower().startswith("en")
+    steps = list(norm.get("steps") or [])
+    constraints = list(norm.get("constraints") or [])
+    criteria = list(norm.get("success_criteria") or [])
+    constants = dict((norm.get("inputs") or {}).get("constants") or {})
+    from_context = list((norm.get("inputs") or {}).get("from_context") or [])
+    need_attachments = bool((norm.get("output") or {}).get("need_attachments"))
+
+    if is_en:
+        lines = [
+            "[Scheduled playbook — internal instruction, not a user message]",
+            f"Goal: {norm.get('goal') or '(unspecified)'}",
+            "Execute this recurring playbook end-to-end. Use tools as needed.",
+            "Do not reply with only a short reminder unless the playbook is truly reminder-only.",
+            "Steps:",
+        ]
+        for i, step in enumerate(steps, start=1):
+            lines.append(f"{i}. {step}")
+        if constraints:
+            lines.append("Constraints:")
+            lines.extend(f"- {c}" for c in constraints)
+        if criteria:
+            lines.append("Success criteria:")
+            lines.extend(f"- {c}" for c in criteria)
+        if constants:
+            lines.append("Fixed inputs:")
+            lines.extend(f"- {k}: {v}" for k, v in constants.items())
+        if from_context:
+            lines.append("Pull from context when needed:")
+            lines.extend(f"- {item}" for item in from_context)
+        if need_attachments:
+            lines.append(
+                "If files are produced, call save_deliverable_attachment so channel delivery includes them."
+            )
+        lines.append("Deliver a useful channel update that reflects completed work.")
+        return "\n".join(lines)
+
+    lines = [
+        "【定时工作流·内部指令，不是用户发言】",
+        f"目标：{norm.get('goal') or '（未指定）'}",
+        "请按下方 playbook 完整执行本轮定时任务；按需调用工具。",
+        "除非任务本身只是提醒，否则不要只回一句短提醒。",
+        "步骤：",
+    ]
+    for i, step in enumerate(steps, start=1):
+        lines.append(f"{i}. {step}")
+    if constraints:
+        lines.append("约束：")
+        lines.extend(f"- {c}" for c in constraints)
+    if criteria:
+        lines.append("成功标准：")
+        lines.extend(f"- {c}" for c in criteria)
+    if constants:
+        lines.append("固定输入：")
+        lines.extend(f"- {k}：{v}" for k, v in constants.items())
+    if from_context:
+        lines.append("需要时从上下文获取：")
+        lines.extend(f"- {item}" for item in from_context)
+    if need_attachments:
+        lines.append("若产生文件，必须调用 save_deliverable_attachment，渠道才会随消息发送附件。")
+    lines.append("完成后向渠道发送能体现已完成工作的更新消息。")
+    return "\n".join(lines)
+
+
+def load_recipe_from_job(job: Any) -> dict[str, Any]:
+    raw = getattr(job, "recipe_json", None)
+    if raw is None and isinstance(job, dict):
+        raw = job.get("recipe_json") or job.get("recipe")
+    if isinstance(raw, dict):
+        return normalize_recipe(raw)
+    text = str(raw or "").strip()
+    if not text:
+        return normalize_recipe({})
+    try:
+        data = json.loads(text)
+    except Exception:
+        return normalize_recipe({})
+    return normalize_recipe(data if isinstance(data, dict) else {})
+
+
+__all__ = [
+    "COMPLEX_PROMPT_HINTS",
+    "compile_playbook_instruction",
+    "load_recipe_from_job",
+    "looks_like_complex_schedule_prompt",
+    "normalize_recipe",
+    "parse_recipe_arg",
+    "preview_markdown",
+    "prompt_summary_from_recipe",
+    "recipe_has_playbook",
+    "recipe_is_empty",
+    "recipe_missing_fields",
+]

runtime/scheduler/service.py

@@ -7,6 +7,7 @@ import uuid
 from datetime import datetime, timezone
 from typing import Any
 
+from runtime.scheduler.recipe import load_recipe_from_job, recipe_has_playbook
 from runtime.scheduler.session_resolver import resolve_scheduled_session, resolve_scheduled_viewer_username
 from runtime.scheduler.turn_text import build_scheduled_turn_instruction
 from runtime.worker import ensure_worker_started
@@ -76,7 +77,14 @@ def enqueue_scheduled_job_run(
     agent_run_id = uuid.uuid4().hex
     prompt_text = str(getattr(job, "prompt_text", "") or "").strip()
     lang = str(getattr(job, "lang", "") or "zh")
-    user_text = build_scheduled_turn_instruction(prompt_text=prompt_text, mode=mode, lang=lang)
+    recipe = load_recipe_from_job(job)
+    playbook = recipe_has_playbook(recipe)
+    user_text = build_scheduled_turn_instruction(
+        prompt_text=prompt_text,
+        mode=mode,
+        lang=lang,
+        recipe=recipe if playbook else None,
+    )
     viewer_username = resolve_scheduled_viewer_username(
         store,
         tenant_id=tenant_id,
@@ -95,6 +103,7 @@ def enqueue_scheduled_job_run(
         "lang": lang,
         "text": user_text,
         "prompt_text": prompt_text,
+        "recipe": recipe if playbook else {},
         "attachments": [],
         "metadata": {
             "scheduled_job_id": job_id,
@@ -103,6 +112,7 @@ def enqueue_scheduled_job_run(
             "selected_specialist": str(getattr(job, "specialist", "") or "generalist"),
             "scheduled_mode": mode,
             "scheduled_proactive": True,
+            "scheduled_playbook": playbook,
         },
         "interaction_mode": str(getattr(job, "interaction_mode", "") or "expert"),
         "requested_specialist": str(getattr(job, "specialist", "") or "generalist"),

runtime/scheduler/turn_text.py

@@ -1,5 +1,9 @@
 from __future__ import annotations
 
+from typing import Any
+
+from runtime.scheduler.recipe import compile_playbook_instruction, recipe_has_playbook
+
 
 def format_scheduled_user_reminder(prompt_text: str) -> str:
     body = str(prompt_text or "").strip()
@@ -10,10 +14,19 @@ def format_scheduled_user_reminder(prompt_text: str) -> str:
     return f"⏰ 提醒：{body}"
 
 
-def build_scheduled_turn_instruction(*, prompt_text: str, mode: str, lang: str) -> str:
-    """Internal LLM instruction for proactive scheduled reminders (not user-facing)."""
-    intent = str(prompt_text or "").strip()
+def build_scheduled_turn_instruction(
+    *,
+    prompt_text: str,
+    mode: str,
+    lang: str,
+    recipe: dict[str, Any] | None = None,
+) -> str:
+    """Internal LLM instruction for proactive scheduled reminders/playbooks (not user-facing)."""
     _ = str(mode or "scheduled").strip()
+    if recipe_has_playbook(recipe):
+        return compile_playbook_instruction(recipe=recipe or {}, lang=lang)
+
+    intent = str(prompt_text or "").strip()
     is_en = str(lang or "").lower().startswith("en")
     if is_en:
         return (
@@ -30,8 +43,21 @@ def build_scheduled_turn_instruction(*, prompt_text: str, mode: str, lang: str)
     )
 
 
-def scheduled_turn_system_suffix(*, lang: str) -> str:
+def scheduled_turn_system_suffix(*, lang: str, playbook: bool = False) -> str:
     is_en = str(lang or "").lower().startswith("en")
+    if playbook:
+        if is_en:
+            return (
+                "\n\n[Scheduled playbook mode] You are executing a recurring workflow for the user. "
+                "Follow the playbook steps, use tools as needed, and deliver a useful update "
+                "(including save_deliverable_attachment for generated files). "
+                "Do not pretend the user just messaged you."
+            )
+        return (
+            "\n\n【定时工作流模式】你正在执行周期性工作流。"
+            "按 playbook 步骤完成任务，按需调用工具；若生成文件须 save_deliverable_attachment。"
+            "不要假装用户刚刚发了消息，不要只回一句空提醒。"
+        )
     if is_en:
         return (
             "\n\n[Scheduled job mode] You are sending a proactive reminder to the user. "

runtime/tools/experts/productivity/schedule_tools.py

@@ -1,17 +1,27 @@
 from __future__ import annotations
 
 import json
+from datetime import datetime, timezone
 from typing import Any
 
 from runtime.scheduler.cron_service import build_delivery_for_session
-from runtime.scheduler.whatsapp_mentions import merge_whatsapp_mention_jids, merge_whatsapp_mention_names
 from runtime.scheduler.expressions import normalize_schedule_kind
-from runtime.scheduler.system_timezone import default_system_timezone
+from runtime.scheduler.recipe import (
+    looks_like_complex_schedule_prompt,
+    normalize_recipe,
+    parse_recipe_arg,
+    preview_markdown,
+    prompt_summary_from_recipe,
+    recipe_has_playbook,
+    recipe_missing_fields,
+)
 from runtime.scheduler.service import run_scheduled_job_now
-from runtime.types import normalize_interaction_mode, normalize_requested_specialist
-from svc.persistence.assistant_store import get_assistant_store
+from runtime.scheduler.system_timezone import default_system_timezone
+from runtime.scheduler.whatsapp_mentions import merge_whatsapp_mention_jids, merge_whatsapp_mention_names
 from runtime.tools.base import ToolSpec
 from runtime.tools.context_inject import enrich_tool_arguments
+from runtime.types import normalize_interaction_mode, normalize_requested_specialist
+from svc.persistence.assistant_store import get_assistant_store
 
 
 def _require(s: str, name: str) -> str:
@@ -35,6 +45,124 @@ def _parse_delivery_arg(raw: Any) -> dict[str, Any] | None:
     return None
 
 
+def _scoped_args(store: Any, tool_name: str, args: dict[str, Any]) -> dict[str, Any]:
+    return enrich_tool_arguments(
+        store=store,
+        session_id=str(args.get("session_id") or ""),
+        tool_name=tool_name,
+        arguments=args,
+    )
+
+
+_RECIPE_PARAM = {
+    "type": "object",
+    "description": (
+        "Self-contained workflow recipe (playbook). Required for complex/multi-step jobs. "
+        "Must be understandable WITHOUT prior chat context: no '继续刚才/按上面'; "
+        "put concrete paths, commands, time windows, and params in goal/steps/inputs.constants. "
+        "Fields: goal, steps (>=2), success_criteria, optional constraints/inputs/output/source."
+    ),
+    "properties": {
+        "version": {"type": "integer"},
+        "goal": {"type": "string"},
+        "steps": {"type": "array", "items": {"type": "string"}},
+        "constraints": {"type": "array", "items": {"type": "string"}},
+        "success_criteria": {"type": "array", "items": {"type": "string"}},
+        "inputs": {"type": "object"},
+        "output": {"type": "object"},
+        "source": {"type": "object"},
+    },
+}
+
+
+def schedule_propose_tool() -> ToolSpec:
+    def handler(args: dict[str, Any]) -> dict[str, Any]:
+        try:
+            store = get_assistant_store()
+            args = _scoped_args(store, "schedule_propose", args)
+            name = str(args.get("name") or "").strip() or "Scheduled workflow"
+            schedule_kind = normalize_schedule_kind(str(args.get("schedule_kind") or "cron"))
+            schedule_expr = _require(str(args.get("schedule_expr") or ""), "schedule_expr")
+            timezone_name = str(args.get("timezone") or default_system_timezone()).strip() or default_system_timezone()
+            lang = str(args.get("lang") or "zh")
+            session_id = str(args.get("session_id") or "").strip()
+            recipe = normalize_recipe(parse_recipe_arg(args.get("recipe")))
+            if session_id and not str((recipe.get("source") or {}).get("session_id") or "").strip():
+                recipe["source"]["session_id"] = session_id
+            if not str((recipe.get("source") or {}).get("compiled_at") or "").strip():
+                recipe["source"]["compiled_at"] = datetime.now(timezone.utc).isoformat()
+
+            missing = recipe_missing_fields(recipe)
+            if missing:
+                return {
+                    "ok": False,
+                    "error": "recipe_incomplete",
+                    "missing_fields": missing,
+                    "hint": (
+                        "Fill goal, at least 2 steps, and success_criteria from the recent conversation, "
+                        "then call schedule_propose again. Do not create the job yet."
+                    ),
+                    "recipe": recipe,
+                }
+
+            preview = preview_markdown(
+                name=name,
+                schedule_kind=schedule_kind,
+                schedule_expr=schedule_expr,
+                timezone_name=timezone_name,
+                recipe=recipe,
+                lang=lang,
+            )
+            return {
+                "ok": True,
+                "draft": True,
+                "name": name,
+                "schedule_kind": schedule_kind,
+                "schedule_expr": schedule_expr,
+                "timezone": timezone_name,
+                "recipe": recipe,
+                "prompt_text": prompt_summary_from_recipe(recipe, fallback=name),
+                "preview_markdown": preview,
+                "next_step": (
+                    "Show preview_markdown to the user. After they confirm (or request edits), "
+                    "call schedule_create with the same recipe and schedule fields."
+                ),
+            }
+        except Exception as e:
+            return {"ok": False, "error": f"{type(e).__name__}: {e}"}
+
+    return ToolSpec(
+        name="schedule_propose",
+        description=(
+            "Draft a scheduled workflow recipe WITHOUT creating the job. "
+            "Use when the user wants to schedule a multi-step task they just guided "
+            "(e.g. '做成定时/每周跑刚才那套'). "
+            "CRITICAL: the recipe must be self-contained — at fire time there is no prior chat; "
+            "a new LLM must understand the task from recipe alone (no '继续刚才', embed paths/params in steps/constants). "
+            "Compile goal/steps(>=2)/success_criteria/constraints from the conversation into `recipe`, "
+            "then show preview_markdown and wait for confirmation before schedule_create."
+        ),
+        parameters={
+            "type": "object",
+            "properties": {
+                "tenant_id": {"type": "string", "description": "Auto-filled from session; do not guess."},
+                "owner_user_id": {"type": "string", "description": "Auto-filled from session."},
+                "session_id": {"type": "string", "description": "Auto-filled from session."},
+                "name": {"type": "string"},
+                "recipe": _RECIPE_PARAM,
+                "schedule_kind": {"type": "string", "enum": ["cron", "once", "interval"]},
+                "schedule_expr": {"type": "string"},
+                "timezone": {"type": "string"},
+                "lang": {"type": "string"},
+            },
+            "required": ["recipe", "schedule_kind", "schedule_expr"],
+            "additionalProperties": False,
+        },
+        handler=handler,
+        tags=frozenset({"productivity", "schedule"}),
+    )
+
+
 def schedule_create_tool() -> ToolSpec:
     def handler(args: dict[str, Any]) -> dict[str, Any]:
         try:
@@ -51,7 +179,37 @@ def schedule_create_tool() -> ToolSpec:
                 "owner_user_id",
             )
             name = _require(str(args.get("name") or ""), "name")
-            prompt_text = _require(str(args.get("prompt_text") or ""), "prompt_text")
+            prompt_text = str(args.get("prompt_text") or "").strip()
+            recipe_raw = parse_recipe_arg(args.get("recipe"))
+            recipe = normalize_recipe(recipe_raw) if recipe_raw is not None else {}
+            if recipe_has_playbook(recipe):
+                prompt_text = prompt_summary_from_recipe(recipe, fallback=prompt_text or name)
+            if not prompt_text:
+                raise ValueError("prompt_text is required")
+
+            needs_recipe = looks_like_complex_schedule_prompt(prompt_text, recipe=recipe)
+            if needs_recipe and not recipe_has_playbook(recipe):
+                missing = recipe_missing_fields(recipe) or ["goal", "steps", "success_criteria"]
+                return {
+                    "ok": False,
+                    "error": "recipe_required",
+                    "missing_fields": missing,
+                    "hint": (
+                        "This looks like a complex/multi-step job. Call schedule_propose first, "
+                        "show the draft to the user, then schedule_create with a full recipe "
+                        "(goal + >=2 steps + success_criteria). Do not store vague prompts like "
+                        "'继续刚才那个'."
+                    ),
+                }
+            if recipe and not recipe_has_playbook(recipe) and recipe_missing_fields(recipe):
+                # Explicit but incomplete recipe → reject rather than silently drop.
+                return {
+                    "ok": False,
+                    "error": "recipe_incomplete",
+                    "missing_fields": recipe_missing_fields(recipe),
+                    "hint": "Complete the recipe or omit it for a simple reminder.",
+                }
+
             schedule_kind = normalize_schedule_kind(str(args.get("schedule_kind") or "cron"))
             schedule_expr = _require(str(args.get("schedule_expr") or ""), "schedule_expr")
             delivery = _parse_delivery_arg(args.get("delivery"))
@@ -76,6 +234,15 @@ def schedule_create_tool() -> ToolSpec:
             specialist = normalize_requested_specialist(
                 str(args.get("specialist") or args.get("selected_specialist") or "generalist")
             )
+            session_id = str(args.get("session_id") or "").strip() or None
+            if recipe_has_playbook(recipe):
+                if session_id and not str((recipe.get("source") or {}).get("session_id") or "").strip():
+                    recipe["source"]["session_id"] = session_id
+                if not str((recipe.get("source") or {}).get("compiled_at") or "").strip():
+                    recipe["source"]["compiled_at"] = datetime.now(timezone.utc).isoformat()
+            else:
+                recipe = {}
+
             row = store.scheduled_job_create(
                 tenant_id=tenant_id,
                 name=name,
@@ -88,7 +255,8 @@ def schedule_create_tool() -> ToolSpec:
                 specialist=specialist,
                 lang=str(args.get("lang") or "zh"),
                 delivery=delivery,
-                source_session_id=str(args.get("session_id") or "").strip() or None,
+                recipe=recipe,
+                source_session_id=session_id,
                 created_by_user_id=owner_user_id,
                 source="chat",
             )
@@ -98,7 +266,14 @@ def schedule_create_tool() -> ToolSpec:
 
     return ToolSpec(
         name="schedule_create",
-        description="Create a scheduled job (cron, once, or interval). Delivery follows the current chat channel (WhatsApp vs WeChat) unless delivery is set explicitly. For WhatsApp group @mentions, set whatsapp_mention_jids with explicit JIDs.",
+        description=(
+            "Create a scheduled job after the user confirmed the draft. "
+            "Simple reminders may use prompt_text only. "
+            "Complex / multi-step / '刚才那件事做成定时' jobs MUST include a self-contained recipe "
+            "(goal + >=2 concrete steps + success_criteria; no chat-dependent phrasing); "
+            "call schedule_propose and get user confirmation first. "
+            "Delivery follows the current chat channel unless delivery is set explicitly."
+        ),
         parameters={
             "type": "object",
             "properties": {
@@ -106,7 +281,11 @@ def schedule_create_tool() -> ToolSpec:
                 "owner_user_id": {"type": "string", "description": "Auto-filled from session."},
                 "session_id": {"type": "string", "description": "Auto-filled from session."},
                 "name": {"type": "string"},
-                "prompt_text": {"type": "string"},
+                "prompt_text": {
+                    "type": "string",
+                    "description": "Short summary / reminder intent. For playbooks, prefer recipe.goal.",
+                },
+                "recipe": _RECIPE_PARAM,
                 "schedule_kind": {"type": "string", "enum": ["cron", "once", "interval"]},
                 "schedule_expr": {"type": "string"},
                 "timezone": {"type": "string", "description": "IANA timezone; defaults to the host system timezone."},
@@ -120,6 +299,10 @@ def schedule_create_tool() -> ToolSpec:
                     "items": {"type": "string"},
                     "description": "WhatsApp JIDs to @mention on delivery (e.g. 628...@s.whatsapp.net). Stored in delivery.whatsapp.mention_jids.",
                 },
+                "whatsapp_mention_names": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                },
                 "delivery": {"type": "object"},
                 "description": {"type": "string"},
             },
@@ -167,15 +350,6 @@ def schedule_list_tool() -> ToolSpec:
     )
 
 
-def _scoped_args(store: Any, tool_name: str, args: dict[str, Any]) -> dict[str, Any]:
-    return enrich_tool_arguments(
-        store=store,
-        session_id=str(args.get("session_id") or ""),
-        tool_name=tool_name,
-        arguments=args,
-    )
-
-
 def schedule_update_tool() -> ToolSpec:
     def handler(args: dict[str, Any]) -> dict[str, Any]:
         try:
@@ -200,6 +374,21 @@ def schedule_update_tool() -> ToolSpec:
             delivery = _parse_delivery_arg(args.get("delivery"))
             if delivery is not None:
                 patch["delivery"] = delivery
+            recipe_raw = parse_recipe_arg(args.get("recipe"))
+            if recipe_raw is not None:
+                recipe = normalize_recipe(recipe_raw)
+                if recipe_has_playbook(recipe):
+                    patch["recipe"] = recipe
+                    if "prompt_text" not in patch:
+                        patch["prompt_text"] = prompt_summary_from_recipe(recipe)
+                elif recipe_missing_fields(recipe):
+                    return {
+                        "ok": False,
+                        "error": "recipe_incomplete",
+                        "missing_fields": recipe_missing_fields(recipe),
+                    }
+                else:
+                    patch["recipe"] = {}
             row = store.scheduled_job_update(tenant_id=tenant_id, job_id=job_id, patch=patch)
             if not row:
                 return {"ok": False, "error": "job_not_found"}
@@ -209,7 +398,7 @@ def schedule_update_tool() -> ToolSpec:
 
     return ToolSpec(
         name="schedule_update",
-        description="Update a scheduled job.",
+        description="Update a scheduled job (including recipe playbook fields).",
         parameters={
             "type": "object",
             "properties": {
@@ -217,6 +406,7 @@ def schedule_update_tool() -> ToolSpec:
                 "job_id": {"type": "string"},
                 "name": {"type": "string"},
                 "prompt_text": {"type": "string"},
+                "recipe": _RECIPE_PARAM,
                 "schedule_kind": {"type": "string"},
                 "schedule_expr": {"type": "string"},
                 "timezone": {"type": "string"},
@@ -340,6 +530,7 @@ def schedule_run_now_tool() -> ToolSpec:
 
 __all__ = [
     "schedule_create_tool",
+    "schedule_propose_tool",
     "schedule_list_tool",
     "schedule_update_tool",
     "schedule_pause_tool",

runtime/tools/public/schedule_tools.py

@@ -13,6 +13,7 @@ from runtime.tools.experts.productivity.schedule_tools import (
     schedule_delete_tool,
     schedule_list_tool,
     schedule_pause_tool,
+    schedule_propose_tool,
     schedule_resume_tool,
     schedule_run_now_tool,
     schedule_update_tool,
@@ -23,6 +24,7 @@ __all__ = [
     "schedule_delete_tool",
     "schedule_list_tool",
     "schedule_pause_tool",
+    "schedule_propose_tool",
     "schedule_resume_tool",
     "schedule_run_now_tool",
     "schedule_update_tool",

runtime/worker.py

@@ -286,9 +286,16 @@ def _worker_loop(*, store: Any, worker_id: str, poll_interval_s: float) -> None:
             if not system_prompt:
                 system_prompt = str(getattr(executor, "system_prompt", "") or "")
             if is_scheduled_turn:
+                from runtime.scheduler.recipe import recipe_has_playbook
                 from runtime.scheduler.turn_text import scheduled_turn_system_suffix
 
-                system_prompt = str(system_prompt or "") + scheduled_turn_system_suffix(lang=lang)
+                playbook = bool((metadata or {}).get("scheduled_playbook")) or recipe_has_playbook(
+                    payload.get("recipe") if isinstance(payload.get("recipe"), dict) else None
+                )
+                system_prompt = str(system_prompt or "") + scheduled_turn_system_suffix(
+                    lang=lang,
+                    playbook=playbook,
+                )
 
             max_messages = int(store.get_setting("AIA_TURN_MAX_CONTEXT_MESSAGES") or 80)
             max_tool_rounds = int(store.get_setting("AIA_TURN_MAX_TOOL_ROUNDS") or 100)