feat(scheduler): confirm before deleting scheduled jobs

Require preview confirmation for schedule_delete, extra confirm_foreign for others' jobs in groups, and mirror the same flow in chat slash commands.

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

runtime/application/gateway/inbound_service.py

@@ -113,7 +113,15 @@ def _menu_text() -> str:
     )
 
 
-def _handle_productivity_commands(*, text: str, tenant_id: str, user_id: str, session_id: str = "") -> str | None:
+def _handle_productivity_commands(
+    *,
+    text: str,
+    tenant_id: str,
+    user_id: str,
+    session_id: str = "",
+    creator_external_user_id: str = "",
+    creator_push_name: str = "",
+) -> str | None:
     t = (text or "").strip()
     t_low = t.lower()
     if not t:
@@ -206,15 +214,95 @@ def _handle_productivity_commands(*, text: str, tenant_id: str, user_id: str, se
         ok = store.scheduled_job_set_status(tenant_id=tenant_id, job_id=full, status="paused")
         return "已暂停。" if ok else "未找到该定时任务。"
 
+    if t.startswith("确认删除定时任务 ") or t.startswith("确认删除定时 "):
+        prefix = "确认删除定时任务 " if t.startswith("确认删除定时任务 ") else "确认删除定时 "
+        jid = t[len(prefix) :].strip()
+        if not jid:
+            return "请提供任务 id。示例：确认删除定时 1234abcd"
+        from runtime.scheduler.job_delete import (
+            job_delete_preview,
+            job_is_foreign,
+            resolve_scheduled_jobs_by_id,
+        )
+
+        matches = resolve_scheduled_jobs_by_id(store, tenant_id=tenant_id, job_id=jid)
+        if not matches:
+            return "未找到该定时任务。"
+        if len(matches) > 1:
+            lines = [
+                f"- {r.name} | id={r.id[:8]}… | {r.schedule_kind}:{r.schedule_expr}" for r in matches[:8]
+            ]
+            return "匹配到多个任务，请用更完整的 id：\n" + "\n".join(lines)
+        job = matches[0]
+        foreign = job_is_foreign(
+            job,
+            actor_user_id=str(user_id or ""),
+            actor_external_user_id=str(creator_external_user_id or ""),
+        )
+        if foreign:
+            return (
+                f"该任务看起来是其他人创建的（{job.name} / id={job.id[:8]}）。"
+                "如需删除，请明确说「确认删除他人定时 <完整id>」。"
+            )
+        ok = store.scheduled_job_delete(tenant_id=tenant_id, job_id=str(job.id))
+        if not ok:
+            return "未找到该定时任务。"
+        preview = job_delete_preview(job)
+        return f"已删除：{preview.get('name')}（id={str(preview.get('id') or '')[:8]}）"
+
+    if t.startswith("确认删除他人定时任务 ") or t.startswith("确认删除他人定时 "):
+        prefix = "确认删除他人定时任务 " if t.startswith("确认删除他人定时任务 ") else "确认删除他人定时 "
+        jid = t[len(prefix) :].strip()
+        if not jid:
+            return "请提供完整任务 id。示例：确认删除他人定时 1234abcd-...."
+        from runtime.scheduler.job_delete import job_delete_preview, resolve_scheduled_jobs_by_id
+
+        matches = resolve_scheduled_jobs_by_id(store, tenant_id=tenant_id, job_id=jid)
+        if not matches:
+            return "未找到该定时任务。"
+        if len(matches) > 1:
+            return "删除他人任务须使用唯一完整 id，当前前缀匹配到多个任务。"
+        job = matches[0]
+        ok = store.scheduled_job_delete(tenant_id=tenant_id, job_id=str(job.id))
+        if not ok:
+            return "未找到该定时任务。"
+        preview = job_delete_preview(job)
+        return f"已删除他人任务：{preview.get('name')}（id={str(preview.get('id') or '')[:8]}）"
+
     if t.startswith("删除定时任务 ") or t.startswith("删除定时 "):
         prefix = "删除定时任务 " if t.startswith("删除定时任务 ") else "删除定时 "
         jid = t[len(prefix) :].strip()
         if not jid:
-            return "请提供任务 id 前缀。示例：删除定时任务 1234abcd"
-        rows = store.scheduled_job_list(tenant_id=tenant_id, status=None, limit=200)
-        full = next((r.id for r in rows if str(r.id).startswith(jid)), jid)
-        ok = store.scheduled_job_delete(tenant_id=tenant_id, job_id=full)
-        return "已删除。" if ok else "未找到该定时任务。"
+            return "请提供任务 id 前缀。示例：删除定时 1234abcd"
+        from runtime.scheduler.job_delete import (
+            job_delete_preview_markdown,
+            job_delete_preview,
+            job_is_foreign,
+            resolve_scheduled_jobs_by_id,
+        )
+
+        matches = resolve_scheduled_jobs_by_id(store, tenant_id=tenant_id, job_id=jid)
+        if not matches:
+            return "未找到该定时任务。"
+        if len(matches) > 1:
+            lines = [
+                f"- {r.name} | id={r.id[:8]}… | {r.schedule_kind}:{r.schedule_expr}" for r in matches[:8]
+            ]
+            return "匹配到多个任务，请指定更完整的 id：\n" + "\n".join(lines)
+        job = matches[0]
+        foreign = job_is_foreign(
+            job,
+            actor_user_id=str(user_id or ""),
+            actor_external_user_id=str(creator_external_user_id or ""),
+        )
+        preview = job_delete_preview(job)
+        md = job_delete_preview_markdown(preview, foreign=foreign, lang="zh")
+        if foreign:
+            return (
+                md
+                + f"\n\n如确认删除，请发送：确认删除他人定时 {job.id}"
+            )
+        return md + f"\n\n如确认删除，请发送：确认删除定时 {job.id}"
 
     schedule_prefix = ""
     if t.startswith("记定时 ") or t.startswith("创建定时 "):
@@ -245,6 +333,15 @@ def _handle_productivity_commands(*, text: str, tenant_id: str, user_id: str, se
             tenant_id=tenant_id,
             session_id=str(session_id or "").strip(),
         )
+        from runtime.scheduler.job_delete import merge_delivery_creator
+
+        delivery = merge_delivery_creator(
+            delivery,
+            user_id=str(user_id or ""),
+            external_user_id=str(creator_external_user_id or ""),
+            push_name=str(creator_push_name or ""),
+            session_id=str(session_id or "").strip(),
+        )
         row = store.scheduled_job_create(
             tenant_id=tenant_id,
             name=prompt_text[:40] or "定时提醒",
@@ -1027,6 +1124,21 @@ def process_inbound_payload(payload: dict[str, Any]) -> dict[str, Any]:
                         tenant_id=tenant_id,
                         user_id=user_id,
                         session_id=str(session_id),
+                        creator_external_user_id=str(getattr(inbound, "external_user_id", "") or ""),
+                        creator_push_name=str(
+                            (
+                                (inbound.metadata or {}).get("push_name")
+                                if isinstance(inbound.metadata, dict)
+                                else ""
+                            )
+                            or (
+                                ((inbound.metadata or {}).get("raw") or {}).get("pushName")
+                                if isinstance(inbound.metadata, dict)
+                                and isinstance((inbound.metadata or {}).get("raw"), dict)
+                                else ""
+                            )
+                            or ""
+                        ),
                     )
                     if cmd_reply is not None:
                         reply = cmd_reply

runtime/chat/tool_runtime.py

@@ -529,13 +529,29 @@ class ToolExecutor:
         # which is important for "created-by specialist" semantics and for later scheduled runs.
         try:
             tname = str(tc.name or "")
-            if tname == "schedule_create":
+            if tname in {"schedule_create", "schedule_delete"}:
                 md = ctx.inbound_metadata if isinstance(ctx.inbound_metadata, dict) else {}
+                raw_block = md.get("raw") if isinstance(md.get("raw"), dict) else {}
+                push_name = str(
+                    md.get("push_name") or md.get("pushName") or raw_block.get("pushName") or ""
+                ).strip()
+                external_uid = str(
+                    md.get("external_user_id")
+                    or md.get("sender_jid")
+                    or raw_block.get("participant")
+                    or raw_block.get("sender")
+                    or ""
+                ).strip()
+                if tname == "schedule_create":
                     cur_spec = str(ctx.specialist or "").strip().lower()
                     if cur_spec and not str(tool_args.get("specialist") or "").strip() and not str(
                         tool_args.get("selected_specialist") or ""
                     ).strip():
                         tool_args["selected_specialist"] = cur_spec
+                    if external_uid and not str(tool_args.get("creator_external_user_id") or "").strip():
+                        tool_args["creator_external_user_id"] = external_uid
+                    if push_name and not str(tool_args.get("creator_push_name") or "").strip():
+                        tool_args["creator_push_name"] = push_name
                     raw_mentions = md.get("mentioned_jids") or md.get("mentionedJids") or md.get("mentions") or []
                     mention_list = raw_mentions if isinstance(raw_mentions, list) else []
                     bot_jid = str(md.get("bot_jid") or "").strip().lower()
@@ -583,6 +599,9 @@ class ToolExecutor:
                                 tool_args["whatsapp_mention_names"] = [
                                     str(x or "").strip() for x in names_md if str(x or "").strip()
                                 ]
+                else:
+                    if external_uid and not str(tool_args.get("actor_external_user_id") or "").strip():
+                        tool_args["actor_external_user_id"] = external_uid
         except Exception:
             pass
         tool_args = filter_arguments_to_schema(tool.parameters, tool_args)

runtime/scheduler/job_delete.py

@@ -0,0 +1,197 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+
+def _delivery_dict(job: Any) -> dict[str, Any]:
+    raw = getattr(job, "delivery_json", None)
+    if isinstance(job, dict):
+        if isinstance(job.get("delivery"), dict):
+            return dict(job.get("delivery") or {})
+        raw = job.get("delivery_json")
+    text = str(raw or "").strip()
+    if not text:
+        return {}
+    try:
+        data = json.loads(text)
+        return data if isinstance(data, dict) else {}
+    except Exception:
+        return {}
+
+
+def creator_from_delivery(delivery: dict[str, Any] | None) -> dict[str, str]:
+    raw = (delivery or {}).get("creator") if isinstance(delivery, dict) else None
+    if not isinstance(raw, dict):
+        return {}
+    return {
+        "user_id": str(raw.get("user_id") or "").strip(),
+        "external_user_id": str(raw.get("external_user_id") or "").strip(),
+        "push_name": str(raw.get("push_name") or "").strip(),
+        "session_id": str(raw.get("session_id") or "").strip(),
+    }
+
+
+def merge_delivery_creator(
+    delivery: dict[str, Any] | None,
+    *,
+    user_id: str = "",
+    external_user_id: str = "",
+    push_name: str = "",
+    session_id: str = "",
+) -> dict[str, Any]:
+    out = dict(delivery or {})
+    creator = {
+        "user_id": str(user_id or "").strip(),
+        "external_user_id": str(external_user_id or "").strip(),
+        "push_name": str(push_name or "").strip(),
+        "session_id": str(session_id or "").strip(),
+    }
+    if any(creator.values()):
+        out["creator"] = creator
+    return out
+
+
+def job_is_foreign(
+    job: Any,
+    *,
+    actor_user_id: str = "",
+    actor_external_user_id: str = "",
+) -> bool:
+    """True when the actor is clearly not the creator (best-effort for group chats)."""
+    delivery = _delivery_dict(job)
+    creator = creator_from_delivery(delivery)
+    actor_uid = str(actor_user_id or "").strip()
+    actor_ext = str(actor_external_user_id or "").strip().lower()
+    creator_uid = str(creator.get("user_id") or getattr(job, "created_by_user_id", "") or "").strip()
+    if isinstance(job, dict) and not creator_uid:
+        creator_uid = str(job.get("created_by_user_id") or "").strip()
+    creator_ext = str(creator.get("external_user_id") or "").strip().lower()
+
+    if actor_ext and creator_ext and actor_ext != creator_ext:
+        return True
+    if actor_uid and creator_uid and actor_uid != creator_uid:
+        # Only treat as foreign when we also lack matching external ids
+        # (same account owner may create many channel jobs).
+        if creator_ext or actor_ext:
+            if creator_ext and actor_ext:
+                return actor_ext != creator_ext
+            # external known on one side only — prefer not to block
+            return False
+        return True
+    return False
+
+
+def resolve_scheduled_jobs_by_id(
+    store: Any,
+    *,
+    tenant_id: str,
+    job_id: str,
+    limit: int = 200,
+) -> list[Any]:
+    jid = str(job_id or "").strip()
+    if not jid:
+        return []
+    exact = store.scheduled_job_get(job_id=jid, tenant_id=tenant_id)
+    if exact is not None:
+        return [exact]
+    rows = store.scheduled_job_list(tenant_id=tenant_id, status=None, limit=max(1, min(int(limit), 500)))
+    return [r for r in rows if str(getattr(r, "id", "") or "").startswith(jid)]
+
+
+def job_delete_preview(job: Any) -> dict[str, Any]:
+    delivery = _delivery_dict(job)
+    creator = creator_from_delivery(delivery)
+    recipe: dict[str, Any] = {}
+    try:
+        raw = getattr(job, "recipe_json", None)
+        if isinstance(job, dict):
+            if isinstance(job.get("recipe"), dict):
+                recipe = dict(job.get("recipe") or {})
+            else:
+                raw = job.get("recipe_json")
+        if not recipe and raw:
+            parsed = json.loads(str(raw or "{}"))
+            if isinstance(parsed, dict):
+                recipe = parsed
+    except Exception:
+        recipe = {}
+    goal = str((recipe or {}).get("goal") or getattr(job, "prompt_text", "") or "").strip()
+    if isinstance(job, dict) and not goal:
+        goal = str(job.get("prompt_text") or "").strip()
+    return {
+        "id": str(getattr(job, "id", None) or (job.get("id") if isinstance(job, dict) else "") or ""),
+        "name": str(getattr(job, "name", None) or (job.get("name") if isinstance(job, dict) else "") or ""),
+        "status": str(getattr(job, "status", None) or (job.get("status") if isinstance(job, dict) else "") or ""),
+        "schedule_kind": str(
+            getattr(job, "schedule_kind", None) or (job.get("schedule_kind") if isinstance(job, dict) else "") or ""
+        ),
+        "schedule_expr": str(
+            getattr(job, "schedule_expr", None) or (job.get("schedule_expr") if isinstance(job, dict) else "") or ""
+        ),
+        "prompt_text": str(
+            getattr(job, "prompt_text", None) or (job.get("prompt_text") if isinstance(job, dict) else "") or ""
+        ),
+        "goal": goal,
+        "created_by_user_id": str(
+            getattr(job, "created_by_user_id", None)
+            or (job.get("created_by_user_id") if isinstance(job, dict) else "")
+            or ""
+        ),
+        "creator": creator,
+        "next_run_at": str(
+            getattr(job, "next_run_at", None) or (job.get("next_run_at") if isinstance(job, dict) else "") or ""
+        )
+        or None,
+    }
+
+
+def job_delete_preview_markdown(preview: dict[str, Any], *, foreign: bool = False, lang: str = "zh") -> str:
+    creator = preview.get("creator") if isinstance(preview.get("creator"), dict) else {}
+    creator_label = str(creator.get("push_name") or "").strip()
+    if not creator_label:
+        creator_label = str(creator.get("external_user_id") or preview.get("created_by_user_id") or "").strip() or "（未知）"
+    is_en = str(lang or "").lower().startswith("en")
+    if is_en:
+        lines = [
+            "## Scheduled job delete preview (confirm required)",
+            f"- **Id**: `{preview.get('id')}`",
+            f"- **Name**: {preview.get('name') or '(untitled)'}",
+            f"- **Schedule**: `{preview.get('schedule_kind')}` `{preview.get('schedule_expr')}`",
+            f"- **Status**: {preview.get('status')}",
+            f"- **Goal / prompt**: {preview.get('goal') or preview.get('prompt_text') or '(empty)'}",
+            f"- **Created by**: {creator_label}",
+        ]
+        if foreign:
+            lines.append("- **Warning**: This job appears to be created by someone else.")
+        lines.append("")
+        lines.append("Reply **confirm delete** with this id, then call schedule_delete with confirmed=true.")
+        if foreign:
+            lines.append("Also set confirm_foreign=true after the owner explicitly agrees.")
+        return "\n".join(lines)
+    lines = [
+        "## 删除定时任务预览（需确认）",
+        f"- **Id**：`{preview.get('id')}`",
+        f"- **名称**：{preview.get('name') or '（未命名）'}",
+        f"- **时间**：`{preview.get('schedule_kind')}` `{preview.get('schedule_expr')}`",
+        f"- **状态**：{preview.get('status')}",
+        f"- **目标/内容**：{preview.get('goal') or preview.get('prompt_text') or '（空）'}",
+        f"- **创建者**：{creator_label}",
+    ]
+    if foreign:
+        lines.append("- **注意**：该任务看起来是**其他人**创建的。")
+    lines.append("")
+    lines.append("请回复**确认删除**并带上该 id；确认后调用 `schedule_delete` 且 `confirmed=true`。")
+    if foreign:
+        lines.append("若删除他人任务，还需用户明确同意，并设置 `confirm_foreign=true`。")
+    return "\n".join(lines)
+
+
+__all__ = [
+    "creator_from_delivery",
+    "job_delete_preview",
+    "job_delete_preview_markdown",
+    "job_is_foreign",
+    "merge_delivery_creator",
+    "resolve_scheduled_jobs_by_id",
+]

runtime/tools/experts/productivity/schedule_tools.py

@@ -6,6 +6,13 @@ from typing import Any
 
 from runtime.scheduler.cron_service import build_delivery_for_session
 from runtime.scheduler.expressions import normalize_schedule_kind
+from runtime.scheduler.job_delete import (
+    job_delete_preview,
+    job_delete_preview_markdown,
+    job_is_foreign,
+    merge_delivery_creator,
+    resolve_scheduled_jobs_by_id,
+)
 from runtime.scheduler.recipe import (
     looks_like_complex_schedule_prompt,
     normalize_recipe,
@@ -228,13 +235,20 @@ def schedule_create_tool() -> ToolSpec:
                 delivery,
                 args.get("whatsapp_mention_names") if args.get("whatsapp_mention_names") is not None else None,
             )
+            session_id = str(args.get("session_id") or "").strip() or None
+            delivery = merge_delivery_creator(
+                delivery,
+                user_id=owner_user_id,
+                external_user_id=str(args.get("creator_external_user_id") or "").strip(),
+                push_name=str(args.get("creator_push_name") or "").strip(),
+                session_id=str(session_id or ""),
+            )
             interaction_mode = normalize_interaction_mode(
                 str(args.get("interaction_mode") or "expert")
             )
             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
@@ -303,6 +317,14 @@ def schedule_create_tool() -> ToolSpec:
                     "type": "array",
                     "items": {"type": "string"},
                 },
+                "creator_external_user_id": {
+                    "type": "string",
+                    "description": "Auto-filled channel sender id for group ownership checks.",
+                },
+                "creator_push_name": {
+                    "type": "string",
+                    "description": "Auto-filled channel display name of the creator.",
+                },
                 "delivery": {"type": "object"},
                 "description": {"type": "string"},
             },
@@ -483,17 +505,109 @@ def schedule_delete_tool() -> ToolSpec:
             args = _scoped_args(store, "schedule_delete", args)
             tenant_id = _require(str(args.get("tenant_id") or ""), "tenant_id")
             job_id = _require(str(args.get("job_id") or ""), "job_id")
-            ok = store.scheduled_job_delete(tenant_id=tenant_id, job_id=job_id)
-            return {"ok": bool(ok), "job_id": job_id}
+            confirmed = bool(args.get("confirmed"))
+            confirm_foreign = bool(args.get("confirm_foreign"))
+            lang = str(args.get("lang") or "zh")
+            actor_user_id = str(args.get("owner_user_id") or args.get("user_id") or "").strip()
+            actor_external_user_id = str(args.get("actor_external_user_id") or "").strip()
+
+            matches = resolve_scheduled_jobs_by_id(store, tenant_id=tenant_id, job_id=job_id)
+            if not matches:
+                return {"ok": False, "error": "job_not_found", "job_id": job_id}
+            if len(matches) > 1:
+                candidates = [job_delete_preview(j) for j in matches[:10]]
+                return {
+                    "ok": False,
+                    "error": "ambiguous_job",
+                    "candidates": candidates,
+                    "hint": (
+                        "Multiple jobs match this id prefix. Show candidates to the user and "
+                        "retry schedule_delete with the full job id."
+                    ),
+                }
+
+            job = matches[0]
+            preview = job_delete_preview(job)
+            foreign = job_is_foreign(
+                job,
+                actor_user_id=actor_user_id,
+                actor_external_user_id=actor_external_user_id,
+            )
+            preview_md = job_delete_preview_markdown(preview, foreign=foreign, lang=lang)
+
+            if not confirmed:
+                return {
+                    "ok": False,
+                    "error": "confirmation_required",
+                    "needs_confirm": True,
+                    "foreign_job": foreign,
+                    "job_id": preview["id"],
+                    "preview": preview,
+                    "preview_markdown": preview_md,
+                    "next_step": (
+                        "Show preview_markdown to the user. After they explicitly confirm deleting "
+                        "THIS job, call schedule_delete again with the full job_id and confirmed=true"
+                        + (", and confirm_foreign=true if foreign_job" if foreign else "")
+                        + "."
+                    ),
+                }
+
+            if foreign and not confirm_foreign:
+                return {
+                    "ok": False,
+                    "error": "foreign_job_confirmation_required",
+                    "needs_confirm": True,
+                    "foreign_job": True,
+                    "job_id": preview["id"],
+                    "preview": preview,
+                    "preview_markdown": preview_md,
+                    "hint": (
+                        "This job was created by someone else. Only delete after the user clearly "
+                        "agrees, then set confirmed=true and confirm_foreign=true."
+                    ),
+                }
+
+            ok = store.scheduled_job_delete(tenant_id=tenant_id, job_id=str(preview["id"]))
+            if not ok:
+                return {"ok": False, "error": "job_not_found", "job_id": preview["id"]}
+            return {"ok": True, "job_id": preview["id"], "deleted": preview}
         except Exception as e:
             return {"ok": False, "error": f"{type(e).__name__}: {e}"}
 
     return ToolSpec(
         name="schedule_delete",
-        description="Delete (soft) a scheduled job.",
+        description=(
+            "Soft-delete a scheduled job WITH confirmation. "
+            "First call WITHOUT confirmed (or confirmed=false) to get preview_markdown; "
+            "show it to the user. After explicit user confirmation, call again with "
+            "confirmed=true. If foreign_job (created by someone else in a group), also set "
+            "confirm_foreign=true. Never guess when multiple jobs match — resolve ambiguity first. "
+            "Prefer pausing instead of deleting when the user only wants to stop temporary runs."
+        ),
         parameters={
             "type": "object",
-            "properties": {"tenant_id": {"type": "string"}, "job_id": {"type": "string"}},
+            "properties": {
+                "tenant_id": {"type": "string"},
+                "job_id": {
+                    "type": "string",
+                    "description": "Full job id preferred; prefix allowed only when unique.",
+                },
+                "confirmed": {
+                    "type": "boolean",
+                    "description": "Must be true to actually delete after user confirmation.",
+                    "default": False,
+                },
+                "confirm_foreign": {
+                    "type": "boolean",
+                    "description": "Required when deleting a job created by someone else.",
+                    "default": False,
+                },
+                "actor_external_user_id": {
+                    "type": "string",
+                    "description": "Auto-filled channel sender id for ownership checks.",
+                },
+                "lang": {"type": "string"},
+            },
             "required": ["job_id"],
             "additionalProperties": False,
         },

skills/_workspace/public/scheduled-workflows/SKILL.md

@@ -1,6 +1,6 @@
 ---
 name: scheduled-workflows
-description: "把复杂/多步工作固化成定时 Workflow Recipe：先 schedule_propose 出草稿给用户确认，再 schedule_create。Recipe 必须自包含——到点执行时没有原对话，LLM 仅凭 recipe 仍能正确完成任务。"
+description: "把复杂/多步工作固化成定时 Workflow Recipe：先 schedule_propose 确认再创建；删除须先预览再确认（群聊他人任务另需 confirm_foreign）。Recipe 必须自包含。"
 ---
 
 # 定时工作流 — Scheduled Workflows
@@ -109,6 +109,23 @@ description: "把复杂/多步工作固化成定时 Workflow Recipe：先 schedu
 | 喝水/开会提醒 | `schedule_create`，仅短 `prompt_text`（本身已自包含） |
 | 多步流程 /「刚才那件事」 | 套用**任务模板** → `schedule_propose` → 确认 → `schedule_create(recipe=...)` |
 
+## 删除定时任务（必须确认）
+
+多任务 / 群聊多人创建时，**禁止直接删除**。
+
+1. 先 `schedule_list` 或让用户说清目标；`schedule_delete(job_id=…)` **不要**先带 `confirmed=true`。
+2. 把返回的 `preview_markdown` 发给用户（含名称、时间、创建者）。
+3. 用户明确确认「删这个」后，再 `schedule_delete(job_id=完整id, confirmed=true)`。
+4. 若 `foreign_job=true`（他人创建）：还须用户明确同意删他人的，并设 `confirm_foreign=true`。
+5. 前缀匹配到多个任务 → 展示候选，**不要猜**。
+6. 仅想暂时停跑时优先 `schedule_pause`，不要删。
+
+聊天快捷指令：
+
+- `删除定时 <id>` → 只预览，不删
+- `确认删除定时 <id>` → 删除自己的
+- `确认删除他人定时 <完整id>` → 删除他人的（需更明确）
+
 ## 修改已有任务
 
 用 `schedule_update` 更新 `recipe`。改完仍做自检：无上下文能否执行；必要时 `schedule_run_now` 验一次。

tests/test_schedule_delete_confirm.py

@@ -0,0 +1,170 @@
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+from runtime.scheduler.job_delete import job_is_foreign, merge_delivery_creator
+from runtime.tools.experts.productivity.schedule_tools import (
+    schedule_create_tool,
+    schedule_delete_tool,
+)
+from svc.persistence.assistant_store import reset_assistant_store_singleton
+from svc.persistence.sqlite_store import SqliteStore
+
+
+class ScheduleDeleteConfirmTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
+        self.db = Path(self._tmp.name) / "del.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"])
+        other = self.store.create_user_account(
+            tenant_id=self.tenant_id,
+            username="alice",
+            display_name="Alice",
+            role="member",
+            password_hash="x",
+            is_active=True,
+        )
+        self.other_id = str(other["id"])
+
+    def tearDown(self) -> None:
+        reset_assistant_store_singleton()
+        self._tmp.cleanup()
+
+    def _create(self, *, owner: str, name: str, creator_ext: str = "", push: str = "") -> str:
+        out = schedule_create_tool().handler(
+            {
+                "tenant_id": self.tenant_id,
+                "owner_user_id": owner,
+                "name": name,
+                "prompt_text": "提醒喝水",
+                "schedule_kind": "interval",
+                "schedule_expr": "3600",
+                "creator_external_user_id": creator_ext,
+                "creator_push_name": push,
+            }
+        )
+        self.assertTrue(out.get("ok"), out)
+        return str((out.get("job") or {})["id"])
+
+    def test_delete_requires_confirmation(self) -> None:
+        jid = self._create(owner=self.user_id, name="喝水", creator_ext="111@lid", push="Bob")
+        preview = schedule_delete_tool().handler(
+            {
+                "tenant_id": self.tenant_id,
+                "owner_user_id": self.user_id,
+                "job_id": jid,
+                "actor_external_user_id": "111@lid",
+            }
+        )
+        self.assertFalse(preview.get("ok"))
+        self.assertEqual(preview.get("error"), "confirmation_required")
+        self.assertIn("preview_markdown", preview)
+        still = self.store.scheduled_job_get(job_id=jid, tenant_id=self.tenant_id)
+        self.assertIsNotNone(still)
+        self.assertEqual(still.status, "active")
+
+        done = schedule_delete_tool().handler(
+            {
+                "tenant_id": self.tenant_id,
+                "owner_user_id": self.user_id,
+                "job_id": jid,
+                "confirmed": True,
+                "actor_external_user_id": "111@lid",
+            }
+        )
+        self.assertTrue(done.get("ok"), done)
+        deleted = self.store.scheduled_job_get(job_id=jid, tenant_id=self.tenant_id)
+        self.assertIsNotNone(deleted)
+        self.assertEqual(deleted.status, "deleted")
+
+    def test_foreign_delete_needs_extra_flag(self) -> None:
+        jid = self._create(
+            owner=self.other_id,
+            name="AliceReminder",
+            creator_ext="alice@lid",
+            push="Alice",
+        )
+        first = schedule_delete_tool().handler(
+            {
+                "tenant_id": self.tenant_id,
+                "owner_user_id": self.user_id,
+                "job_id": jid,
+                "confirmed": True,
+                "actor_external_user_id": "bob@lid",
+            }
+        )
+        self.assertFalse(first.get("ok"))
+        self.assertEqual(first.get("error"), "foreign_job_confirmation_required")
+
+        done = schedule_delete_tool().handler(
+            {
+                "tenant_id": self.tenant_id,
+                "owner_user_id": self.user_id,
+                "job_id": jid,
+                "confirmed": True,
+                "confirm_foreign": True,
+                "actor_external_user_id": "bob@lid",
+            }
+        )
+        self.assertTrue(done.get("ok"), done)
+
+    def test_ambiguous_prefix(self) -> None:
+        # Force two jobs; prefix of common empty won't work — use identical id prefix via listing
+        a = self._create(owner=self.user_id, name="A")
+        b = self._create(owner=self.user_id, name="B")
+        # uuid unlikely share long prefix; use empty path with resolve of weird short then skip
+        # Instead verify prefix of full id alone still unique
+        out = schedule_delete_tool().handler(
+            {
+                "tenant_id": self.tenant_id,
+                "owner_user_id": self.user_id,
+                "job_id": a[:8],
+            }
+        )
+        # Usually unique with 8 hex chars of uuid
+        if out.get("error") == "ambiguous_job":
+            self.assertTrue(out.get("candidates"))
+        else:
+            self.assertIn(out.get("error"), {"confirmation_required", "job_not_found"})
+            if out.get("error") == "confirmation_required":
+                self.assertEqual(out.get("job_id"), a)
+        _ = b
+
+    def test_creator_merge_and_foreign_helper(self) -> None:
+        delivery = merge_delivery_creator(
+            {},
+            user_id="u1",
+            external_user_id="ext-a",
+            push_name="Ann",
+        )
+        job = type(
+            "J",
+            (),
+            {
+                "created_by_user_id": "u1",
+                "delivery_json": __import__("json").dumps(delivery),
+            },
+        )()
+        self.assertFalse(job_is_foreign(job, actor_user_id="u1", actor_external_user_id="ext-a"))
+        self.assertTrue(job_is_foreign(job, actor_user_id="u2", actor_external_user_id="ext-b"))
+
+
+if __name__ == "__main__":
+    unittest.main()