From d036190195c49df3485b45e7e7bd6908806b4ca5 Mon Sep 17 00:00:00 2001 From: oliver Date: Tue, 14 Jul 2026 09:58:39 +0800 Subject: [PATCH] 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 --- interfaces/admin/routes.py | 2 + runtime/scheduler/recipe.py | 339 ++++++++++++++++++ runtime/scheduler/service.py | 12 +- runtime/scheduler/turn_text.py | 34 +- .../experts/productivity/schedule_tools.py | 227 +++++++++++- runtime/tools/public/schedule_tools.py | 2 + runtime/worker.py | 9 +- .../public/scheduled-workflows/SKILL.md | 114 ++++++ svc/persistence/scheduled_job_store.py | 95 +++-- tests/test_schedule_recipe.py | 204 +++++++++++ 10 files changed, 989 insertions(+), 49 deletions(-) create mode 100644 runtime/scheduler/recipe.py create mode 100644 skills/_workspace/public/scheduled-workflows/SKILL.md create mode 100644 tests/test_schedule_recipe.py diff --git a/interfaces/admin/routes.py b/interfaces/admin/routes.py index 00d3644b..3230c7d1 100644 --- a/interfaces/admin/routes.py +++ b/interfaces/admin/routes.py @@ -2376,6 +2376,7 @@ def build_admin_router() -> APIRouter: if delivery is None: wa_chat = str((payload.get("whatsapp") or {}).get("chat_id") if isinstance(payload.get("whatsapp"), dict) else payload.get("whatsapp_chat_id") or "") delivery = build_default_delivery(store=store, tenant_id=tenant_id, whatsapp_chat_id=wa_chat) + recipe = payload.get("recipe") if isinstance(payload.get("recipe"), dict) else None row = store.scheduled_job_create( tenant_id=tenant_id, name=name, @@ -2388,6 +2389,7 @@ def build_admin_router() -> APIRouter: specialist=normalize_requested_specialist(payload.get("specialist") or "generalist"), lang=str(payload.get("lang") or "zh"), delivery=delivery, + recipe=recipe, source_session_id=str(payload.get("source_session_id") or "").strip() or None, created_by_user_id=str(ctx.get("user_id") or ""), source="admin", diff --git a/runtime/scheduler/recipe.py b/runtime/scheduler/recipe.py new file mode 100644 index 00000000..2fa8f245 --- /dev/null +++ b/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", +] diff --git a/runtime/scheduler/service.py b/runtime/scheduler/service.py index 31fe3c14..83da0d6e 100644 --- a/runtime/scheduler/service.py +++ b/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"), diff --git a/runtime/scheduler/turn_text.py b/runtime/scheduler/turn_text.py index 5d6daac8..3058f148 100644 --- a/runtime/scheduler/turn_text.py +++ b/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. " diff --git a/runtime/tools/experts/productivity/schedule_tools.py b/runtime/tools/experts/productivity/schedule_tools.py index d1b67e93..60bf81f2 100644 --- a/runtime/tools/experts/productivity/schedule_tools.py +++ b/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", diff --git a/runtime/tools/public/schedule_tools.py b/runtime/tools/public/schedule_tools.py index c45c7d30..8bf3b2eb 100644 --- a/runtime/tools/public/schedule_tools.py +++ b/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", diff --git a/runtime/worker.py b/runtime/worker.py index 5771e211..f4e2a1dc 100644 --- a/runtime/worker.py +++ b/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) diff --git a/skills/_workspace/public/scheduled-workflows/SKILL.md b/skills/_workspace/public/scheduled-workflows/SKILL.md new file mode 100644 index 00000000..f740cbc1 --- /dev/null +++ b/skills/_workspace/public/scheduled-workflows/SKILL.md @@ -0,0 +1,114 @@ +--- +name: scheduled-workflows +description: "把复杂/多步工作固化成定时 Workflow Recipe：先 schedule_propose 出草稿给用户确认，再 schedule_create。Recipe 必须自包含——到点执行时没有原对话，LLM 仅凭 recipe 仍能正确完成任务。" +--- + +# 定时工作流 — Scheduled Workflows + +## 何时使用 + +用户想把**刚才引导完成的多步任务**做成周期执行（例如「做成每周一 9 点的定时」「按刚才那套流程每天跑」），或创建会产出文件/报告的定时任务。 + +简单一句提醒（「每三小时提醒我喝水」）**不要**用本流程，直接 `schedule_create` + 短 `prompt_text` 即可。 + +## 最重要：自包含（无上下文仍可执行） + +到点触发时，worker **不会**把「当时那通聊天」当成任务说明。LLM 几乎只看到 `recipe` 编译出的内部指令（外加可能截断的会话噪声）。 + +因此写入的 recipe / 步骤文案必须满足： + +> **一个陌生的 LLM 在没有原对话、没有「刚才」「那个」指代的情况下，仍能正确理解并完成任务。** + +### 禁止写入（指代 / 依赖上文） + +- 「继续刚才那个」「按上面流程」「和上次一样」 +- 「用用户给的那个文件」「那个模板」——却不写清路径/文件名/字段 +- 「参考本会话里刚生成的报告」——却不把关键结论、参数、路径抄进 recipe +- 步骤短到只有动词：「处理数据」「发一下」 + +### 必须写入（自包含要素） + +- **做什么**：具体目标与交付物（文件类型、发给谁、消息长什么样） +- **怎么做**：可逐步执行的步骤（工具/命令/数据源/命名规则） +- **用什么**：路径、URL、表名、模板内容或固定参数 → 放进 `inputs.constants` +- **边界**：不要动什么、失败怎么处理 → `constraints` +- **怎样算完成**：可检查的成功标准 → `success_criteria` + +写完后自检一句：**「若只有这份 recipe、清空聊天记录，新模型能否一次做对？」** 不能则补字段，禁止提交。 + +## 任务模板（标准样例，照此展开） + +新建复杂定时任务时，**按下面模板填满**再 `schedule_propose`。括号内是说明，写入时换成真实内容。 + +```json +{ + "version": 1, + "goal": "【一句话完整目标】例如：每周一 09:00 汇总上一自然周值班告警，生成 PDF 发到当前 WhatsApp 群并 @值班人", + "steps": [ + "【步骤1·取数】说明数据从哪来：接口/命令/文件路径、时间范围如何算（上周一 00:00 至周日 23:59，时区 Asia/Shanghai）", + "【步骤2·处理】说明过滤规则、分组字段、统计口径；需要生成的中间文件路径（如 data/workspace/tmp/duty_weekly.csv）", + "【步骤3·产出】说明最终交付物格式与路径（如 data/workspace/tmp/duty_weekly_YYYYMMDD.pdf）；标题/章节结构写清", + "【步骤4·投递】调用 save_deliverable_attachment 发送 PDF；渠道消息须含：周期范围、总数、Top3 风险、是否需人工跟进；若需 @人写清展示名与用途" + ], + "constraints": [ + "【边界】只读该数据源，不修改生产配置/历史文件", + "【失败】取数失败则渠道只发失败原因摘要，不假装成功、不发空 PDF" + ], + "success_criteria": [ + "群内收到本周 PDF 附件", + "消息正文含统计周期与至少一条结论或风险点", + "无未说明的错误/半成品路径糊弄用户" + ], + "inputs": { + "constants": { + "timezone": "Asia/Shanghai", + "data_source": "【写死：命令、API、或 workspace 固定路径】", + "output_dir": "data/workspace/tmp", + "report_title_template": "值班周报 {week_range}" + }, + "from_context": [] + }, + "output": { + "style": "channel_update", + "need_attachments": true + } +} +``` + +说明： + +- `from_context` 默认保持 `[]`。凡是执行要用的信息，**尽量写进 `constants` 或 steps 正文**，不要指望「到点还能翻到建任务时那通聊天」。 +- `goal` 也可作为 `prompt_text` 短摘要，但**不能只有 goal、steps 仍含糊**。 + +### 反例 → 正例 + +| 反例（依赖上下文） | 正例（自包含） | +|-------------------|----------------| +| 步骤：生成报告并发群 | 用 `data/workspace/templates/duty.md` 模板，把上周告警按 NE 聚合，写 PDF 到 `data/workspace/tmp/duty_weekly.pdf`，再 `save_deliverable_attachment` 发当前渠道 | +| constants: `{}`，口头说「用刚才那个 CSV」 | `"source_csv": "data/workspace/uploads/alarms_export.csv"`（或写入导出口径与再拉取命令） | +| success: 搞定 | 渠道出现 PDF；正文含「告警总数 / 严重占比」 | + +## 标准流程（必须遵守） + +1. **`schedule_propose`**（只草稿，不落库） + 按上方**任务模板**从对话整理自包含 `recipe`（禁止原样塞「继续刚才」）。 +2. **把 `preview_markdown` 发给用户**，并口头强调：到点不会带着本次聊天，请核对步骤是否自包含。 +3. 用户确认后 **`schedule_create`**，带上同一份 `recipe` + 时间字段。 + `prompt_text` = `recipe.goal`（完整句，不要用「同上」）。 + +禁止：复杂任务直接 `schedule_create(prompt_text="继续刚才那个")`。 + +## 与渠道附件 + +若周期任务会生成文件，设 `"output": {"need_attachments": true}`，步骤里写明 `save_deliverable_attachment`（见 **channel-file-delivery**）。 + +## 简单 vs 复杂 + +| 类型 | 做法 | +|------|------| +| 喝水/开会提醒 | `schedule_create`，仅短 `prompt_text`（本身已自包含） | +| 多步流程 /「刚才那件事」 | 套用**任务模板** → `schedule_propose` → 确认 → `schedule_create(recipe=...)` | + +## 修改已有任务 + +用 `schedule_update` 更新 `recipe`。改完仍做自检：无上下文能否执行；必要时 `schedule_run_now` 验一次。 diff --git a/svc/persistence/scheduled_job_store.py b/svc/persistence/scheduled_job_store.py index 52d93775..ad85e812 100644 --- a/svc/persistence/scheduled_job_store.py +++ b/svc/persistence/scheduled_job_store.py @@ -20,6 +20,7 @@ CREATE TABLE IF NOT EXISTS scheduled_job ( schedule_expr TEXT NOT NULL, timezone TEXT NOT NULL DEFAULT 'Asia/Shanghai', prompt_text TEXT NOT NULL, + recipe_json TEXT NOT NULL DEFAULT '{}', interaction_mode TEXT NOT NULL DEFAULT 'expert', specialist TEXT NOT NULL DEFAULT 'generalist', lang TEXT NOT NULL DEFAULT 'zh', @@ -81,6 +82,7 @@ class ScheduledJob: schedule_expr: str timezone: str prompt_text: str + recipe_json: str interaction_mode: str specialist: str lang: str @@ -95,6 +97,18 @@ class ScheduledJob: updated_at: str +def _row_field(row: Any, key: str, default: str = "") -> str: + try: + if hasattr(row, "keys") and key in row.keys(): + return str(row[key] if row[key] is not None else default) + except Exception: + pass + try: + return str(row[key] if row[key] is not None else default) + except Exception: + return default + + @dataclass(frozen=True) class ScheduledJobRun: id: str @@ -115,27 +129,28 @@ class ScheduledJobRun: def _row_to_job(row: Any) -> ScheduledJob: return ScheduledJob( - id=str(row["id"] or ""), - tenant_id=str(row["tenant_id"] or ""), - name=str(row["name"] or ""), - description=str(row["description"] or ""), - status=str(row["status"] or ""), - schedule_kind=str(row["schedule_kind"] or ""), - schedule_expr=str(row["schedule_expr"] or ""), - timezone=str(row["timezone"] or default_system_timezone()), - prompt_text=str(row["prompt_text"] or ""), - interaction_mode=str(row["interaction_mode"] or "expert"), - specialist=str(row["specialist"] or "generalist"), - lang=str(row["lang"] or "zh"), - delivery_json=str(row["delivery_json"] or "{}"), - source_session_id=str(row["source_session_id"] or "") or None, - created_by_user_id=str(row["created_by_user_id"] or ""), - source=str(row["source"] or ""), - next_run_at=str(row["next_run_at"] or "") or None, - last_run_at=str(row["last_run_at"] or "") or None, - last_run_status=str(row["last_run_status"] or ""), - created_at=str(row["created_at"] or ""), - updated_at=str(row["updated_at"] or ""), + id=_row_field(row, "id"), + tenant_id=_row_field(row, "tenant_id"), + name=_row_field(row, "name"), + description=_row_field(row, "description"), + status=_row_field(row, "status"), + schedule_kind=_row_field(row, "schedule_kind"), + schedule_expr=_row_field(row, "schedule_expr"), + timezone=_row_field(row, "timezone", default_system_timezone()) or default_system_timezone(), + prompt_text=_row_field(row, "prompt_text"), + recipe_json=_row_field(row, "recipe_json", "{}") or "{}", + interaction_mode=_row_field(row, "interaction_mode", "expert") or "expert", + specialist=_row_field(row, "specialist", "generalist") or "generalist", + lang=_row_field(row, "lang", "zh") or "zh", + delivery_json=_row_field(row, "delivery_json", "{}") or "{}", + source_session_id=_row_field(row, "source_session_id") or None, + created_by_user_id=_row_field(row, "created_by_user_id"), + source=_row_field(row, "source"), + next_run_at=_row_field(row, "next_run_at") or None, + last_run_at=_row_field(row, "last_run_at") or None, + last_run_status=_row_field(row, "last_run_status"), + created_at=_row_field(row, "created_at"), + updated_at=_row_field(row, "updated_at"), ) @@ -160,7 +175,7 @@ def _row_to_run(row: Any) -> ScheduledJobRun: _JOB_SELECT = """ SELECT id, tenant_id, name, description, status, schedule_kind, schedule_expr, timezone, - prompt_text, interaction_mode, specialist, lang, delivery_json, source_session_id, + prompt_text, recipe_json, interaction_mode, specialist, lang, delivery_json, source_session_id, created_by_user_id, source, next_run_at, last_run_at, last_run_status, created_at, updated_at FROM scheduled_job """ @@ -172,6 +187,19 @@ class ScheduledJobStoreMixin: conn.execute(SCHEDULED_JOB_INDEX_DDL) conn.execute(SCHEDULED_JOB_RUN_DDL) conn.execute(SCHEDULED_JOB_RUN_INDEX_DDL) + self._ensure_scheduled_job_recipe_column(conn) + + def _ensure_scheduled_job_recipe_column(self, conn: Any) -> None: + if bool(getattr(self, "_use_pg", False)): + conn.execute( + "ALTER TABLE scheduled_job ADD COLUMN IF NOT EXISTS recipe_json TEXT NOT NULL DEFAULT '{}'" + ) + return + cols = {row[1] for row in conn.execute("PRAGMA table_info(scheduled_job)").fetchall()} + if "recipe_json" not in cols: + conn.execute( + "ALTER TABLE scheduled_job ADD COLUMN recipe_json TEXT NOT NULL DEFAULT '{}'" + ) def scheduled_job_create( self, @@ -187,6 +215,7 @@ class ScheduledJobStoreMixin: specialist: str = "generalist", lang: str = "zh", delivery: dict[str, Any] | None = None, + recipe: dict[str, Any] | None = None, source_session_id: str | None = None, created_by_user_id: str = "", source: str = "admin", @@ -203,14 +232,15 @@ class ScheduledJobStoreMixin: from_dt=None, ) delivery_json = json.dumps(delivery or {}, ensure_ascii=False) + recipe_json = json.dumps(recipe or {}, ensure_ascii=False) with self._connect() as conn: # type: ignore[attr-defined] conn.execute( """ INSERT INTO scheduled_job (id, tenant_id, name, description, status, schedule_kind, schedule_expr, timezone, - prompt_text, interaction_mode, specialist, lang, delivery_json, source_session_id, + prompt_text, recipe_json, interaction_mode, specialist, lang, delivery_json, source_session_id, created_by_user_id, source, next_run_at, last_run_at, last_run_status, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, '', ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, '', ?, ?) """, ( jid, @@ -222,6 +252,7 @@ class ScheduledJobStoreMixin: str(schedule_expr or "").strip(), tz, str(prompt_text or "").strip(), + recipe_json, str(interaction_mode or "expert"), str(specialist or "generalist"), str(lang or "zh"), @@ -302,6 +333,11 @@ class ScheduledJobStoreMixin: delivery_json = json.dumps(p["delivery"], ensure_ascii=False) elif "delivery_json" in p: delivery_json = str(p.get("delivery_json") or "{}") + recipe_json = cur.recipe_json + if "recipe" in p and isinstance(p.get("recipe"), dict): + recipe_json = json.dumps(p["recipe"], ensure_ascii=False) + elif "recipe_json" in p: + recipe_json = str(p.get("recipe_json") or "{}") source_session_id = cur.source_session_id if "source_session_id" in p: raw_sid = str(p.get("source_session_id") or "").strip() @@ -323,7 +359,7 @@ class ScheduledJobStoreMixin: conn.execute( """ UPDATE scheduled_job SET - name = ?, description = ?, prompt_text = ?, interaction_mode = ?, specialist = ?, + name = ?, description = ?, prompt_text = ?, recipe_json = ?, interaction_mode = ?, specialist = ?, lang = ?, schedule_kind = ?, schedule_expr = ?, timezone = ?, delivery_json = ?, source_session_id = ?, status = ?, next_run_at = ?, updated_at = ? WHERE id = ? AND tenant_id = ? @@ -332,6 +368,7 @@ class ScheduledJobStoreMixin: name, description, prompt_text, + recipe_json, interaction_mode, specialist, lang, @@ -576,6 +613,13 @@ class ScheduledJobStoreMixin: delivery = raw except Exception: delivery = {} + recipe: dict[str, Any] = {} + try: + raw_recipe = json.loads(job.recipe_json or "{}") + if isinstance(raw_recipe, dict): + recipe = raw_recipe + except Exception: + recipe = {} return { "id": job.id, "tenant_id": job.tenant_id, @@ -586,6 +630,7 @@ class ScheduledJobStoreMixin: "schedule_expr": job.schedule_expr, "timezone": job.timezone, "prompt_text": job.prompt_text, + "recipe": recipe, "interaction_mode": job.interaction_mode, "specialist": job.specialist, "lang": job.lang, diff --git a/tests/test_schedule_recipe.py b/tests/test_schedule_recipe.py new file mode 100644 index 00000000..da63e783 --- /dev/null +++ b/tests/test_schedule_recipe.py @@ -0,0 +1,204 @@ +from __future__ import annotations + +import os +import tempfile +import unittest +from pathlib import Path + +from runtime.scheduler.recipe import ( + compile_playbook_instruction, + looks_like_complex_schedule_prompt, + normalize_recipe, + preview_markdown, + prompt_summary_from_recipe, + recipe_has_playbook, + recipe_missing_fields, +) +from runtime.scheduler.turn_text import build_scheduled_turn_instruction, scheduled_turn_system_suffix +from runtime.tools.experts.productivity.schedule_tools import ( + schedule_create_tool, + schedule_propose_tool, +) +from svc.persistence.assistant_store import reset_assistant_store_singleton +from svc.persistence.sqlite_store import SqliteStore + + +class RecipeHelpersTests(unittest.TestCase): + def test_normalize_and_missing_fields(self) -> None: + recipe = normalize_recipe( + { + "goal": "发值班报告", + "steps": ["拉数据"], + "success_criteria": [], + } + ) + self.assertEqual(recipe["goal"], "发值班报告") + self.assertEqual(recipe_missing_fields(recipe), ["steps", "success_criteria"]) + self.assertFalse(recipe_has_playbook(recipe)) + + def test_playbook_ready(self) -> None: + recipe = normalize_recipe( + { + "goal": "发值班报告", + "steps": ["拉数据", "写摘要", "发群"], + "success_criteria": ["群内收到 PDF"], + "output": {"need_attachments": True}, + } + ) + self.assertTrue(recipe_has_playbook(recipe)) + self.assertEqual(recipe_missing_fields(recipe), []) + preview = preview_markdown( + name="值班报告", + schedule_kind="cron", + schedule_expr="0 9 * * 1", + timezone_name="Asia/Shanghai", + recipe=recipe, + lang="zh", + ) + self.assertIn("拉数据", preview) + self.assertIn("确认", preview) + instr = compile_playbook_instruction(recipe=recipe, lang="zh") + self.assertIn("定时工作流", instr) + self.assertIn("save_deliverable_attachment", instr) + self.assertEqual(prompt_summary_from_recipe(recipe), "发值班报告") + + def test_complex_prompt_heuristic(self) -> None: + self.assertTrue(looks_like_complex_schedule_prompt("把刚才那件事做成每周一定时")) + self.assertTrue(looks_like_complex_schedule_prompt("继续刚才那个生成 PDF 流程")) + self.assertFalse(looks_like_complex_schedule_prompt("提醒喝水")) + + def test_turn_instruction_modes(self) -> None: + reminder = build_scheduled_turn_instruction(prompt_text="喝水", mode="scheduled", lang="zh") + self.assertIn("提醒意图", reminder) + recipe = { + "goal": "发报告", + "steps": ["A", "B"], + "success_criteria": ["完成"], + } + playbook = build_scheduled_turn_instruction( + prompt_text="发报告", + mode="scheduled", + lang="zh", + recipe=recipe, + ) + self.assertIn("步骤", playbook) + self.assertIn("A", playbook) + self.assertIn("工作流", scheduled_turn_system_suffix(lang="zh", playbook=True)) + self.assertIn("提醒", scheduled_turn_system_suffix(lang="zh", playbook=False)) + + +class ScheduleRecipeToolTests(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.db = Path(self._tmp.name) / "recipe.sqlite" + os.environ["OPS_ASSISTANT_DB_PATH"] = str(self.db) + os.environ["AIA_ASSISTANT_DB_BACKEND"] = "sqlite" + reset_assistant_store_singleton() + self.store = SqliteStore(str(self.db)) + t = self.store.create_tenant("Team") + self.tenant_id = str(t["id"]) + user = self.store.create_user_account( + tenant_id=self.tenant_id, + username="administrator", + display_name="Admin", + role="owner", + password_hash="x", + is_active=True, + ) + self.user_id = str(user["id"]) + + def tearDown(self) -> None: + reset_assistant_store_singleton() + self._tmp.cleanup() + + def test_propose_rejects_incomplete_recipe(self) -> None: + out = schedule_propose_tool().handler( + { + "tenant_id": self.tenant_id, + "owner_user_id": self.user_id, + "name": "报告", + "schedule_kind": "cron", + "schedule_expr": "0 9 * * 1", + "recipe": {"goal": "发报告", "steps": ["一步"]}, + } + ) + self.assertFalse(out.get("ok")) + self.assertEqual(out.get("error"), "recipe_incomplete") + + def test_propose_and_create_playbook(self) -> None: + recipe = { + "goal": "每周发值班 PDF", + "steps": ["拉取数据", "生成 PDF", "发群"], + "success_criteria": ["群内收到 PDF"], + "constraints": ["不改历史文件"], + "output": {"need_attachments": True}, + } + proposed = schedule_propose_tool().handler( + { + "tenant_id": self.tenant_id, + "owner_user_id": self.user_id, + "session_id": "sess-1", + "name": "值班报告", + "schedule_kind": "cron", + "schedule_expr": "0 9 * * 1", + "recipe": recipe, + } + ) + self.assertTrue(proposed.get("ok"), proposed) + self.assertTrue(proposed.get("draft")) + self.assertIn("preview_markdown", proposed) + + created = schedule_create_tool().handler( + { + "tenant_id": self.tenant_id, + "owner_user_id": self.user_id, + "session_id": "sess-1", + "name": "值班报告", + "prompt_text": "每周发值班 PDF", + "schedule_kind": "cron", + "schedule_expr": "0 9 * * 1", + "recipe": recipe, + } + ) + self.assertTrue(created.get("ok"), created) + job = created.get("job") or {} + self.assertEqual((job.get("recipe") or {}).get("goal"), "每周发值班 PDF") + self.assertEqual(len((job.get("recipe") or {}).get("steps") or []), 3) + + got = self.store.scheduled_job_get(job_id=str(job["id"]), tenant_id=self.tenant_id) + assert got is not None + self.assertIn("每周发值班 PDF", got.recipe_json) + + def test_create_rejects_vague_complex_prompt(self) -> None: + out = schedule_create_tool().handler( + { + "tenant_id": self.tenant_id, + "owner_user_id": self.user_id, + "name": "bad", + "prompt_text": "继续刚才那个", + "schedule_kind": "interval", + "schedule_expr": "3600", + } + ) + self.assertFalse(out.get("ok")) + self.assertEqual(out.get("error"), "recipe_required") + + def test_simple_reminder_still_works(self) -> None: + out = schedule_create_tool().handler( + { + "tenant_id": self.tenant_id, + "owner_user_id": self.user_id, + "name": "喝水", + "prompt_text": "提醒喝水", + "schedule_kind": "interval", + "schedule_expr": "10800", + } + ) + self.assertTrue(out.get("ok"), out) + job = out.get("job") or {} + self.assertEqual(job.get("prompt_text"), "提醒喝水") + self.assertEqual(job.get("recipe") or {}, {}) + + +if __name__ == "__main__": + unittest.main()