diff --git a/interfaces/admin/static/app.js b/interfaces/admin/static/app.js index ad688624..3e8adff6 100644 --- a/interfaces/admin/static/app.js +++ b/interfaces/admin/static/app.js @@ -27,12 +27,17 @@ const I18N = { "scheduledJobs.count": "{count} 个任务", "scheduledJobs.colName": "名称", "scheduledJobs.colSchedule": "计划", + "scheduledJobs.colPlaybook": "剧本", "scheduledJobs.colStatus": "状态", "scheduledJobs.colNextRun": "下次运行", "scheduledJobs.colLastRun": "上次运行", "scheduledJobs.colSpecialist": "专家", "scheduledJobs.colDelivery": "投递", "scheduledJobs.colActions": "操作", + "scheduledJobs.playbookYes": "playbook · {n} 步", + "scheduledJobs.playbookPartial": "有 recipe · {n} 步", + "scheduledJobs.playbookNo": "提醒", + "scheduledJobs.skippedOverlap": "已跳过（叠跑）", "scheduledJobs.menuTitle": "任务操作", "scheduledJobs.viewRuns": "查看运行记录", "scheduledJobs.pause": "暂停", @@ -505,12 +510,17 @@ const I18N = { "scheduledJobs.count": "{count} job(s)", "scheduledJobs.colName": "name", "scheduledJobs.colSchedule": "schedule", + "scheduledJobs.colPlaybook": "playbook", "scheduledJobs.colStatus": "status", "scheduledJobs.colNextRun": "next_run", "scheduledJobs.colLastRun": "last_run", "scheduledJobs.colSpecialist": "specialist", "scheduledJobs.colDelivery": "delivery", "scheduledJobs.colActions": "actions", + "scheduledJobs.playbookYes": "playbook · {n} steps", + "scheduledJobs.playbookPartial": "recipe · {n} steps", + "scheduledJobs.playbookNo": "reminder", + "scheduledJobs.skippedOverlap": "Skipped (overlapping)", "scheduledJobs.menuTitle": "Job actions", "scheduledJobs.viewRuns": "View runs", "scheduledJobs.pause": "Pause", @@ -9692,6 +9702,13 @@ function formatScheduledJobDelivery(job) { return parts.length ? parts.join(" + ") : "—"; } +function formatScheduledJobPlaybook(job) { + const steps = Number(job && job.steps_n != null ? job.steps_n : 0) || 0; + if (job && job.playbook) return tf("scheduledJobs.playbookYes", { n: String(steps) }); + if (job && job.has_recipe) return tf("scheduledJobs.playbookPartial", { n: String(steps) }); + return t("scheduledJobs.playbookNo"); +} + function buildScheduledJobDeliveryPayload(job, waChatId) { const existing = (job && job.delivery && typeof job.delivery === "object") ? job.delivery : {}; const wa = existing.whatsapp && typeof existing.whatsapp === "object" ? { ...existing.whatsapp } : {}; @@ -9884,8 +9901,13 @@ async function renderScheduledJobs() { text: t("scheduledJobs.runNow"), onclick: async () => { closeScheduledJobMenu(); - await apiPost(`/admin/api/scheduled-jobs/${encodeURIComponent(jobId)}/run-now`, {}); - msg.textContent = t("scheduledJobs.triggered"); + const out = await apiPost(`/admin/api/scheduled-jobs/${encodeURIComponent(jobId)}/run-now`, {}); + if (out && out.skipped) { + msg.textContent = t("scheduledJobs.skippedOverlap"); + } else { + msg.textContent = t("scheduledJobs.triggered"); + } + await loadJobs(); }, }), ...(canWrite @@ -9935,6 +9957,7 @@ async function renderScheduledJobs() { const tr = el("tr", {}, [ tdCell(job.name || "", 24), tdCell(`${job.schedule_kind}:${job.schedule_expr}`, 28), + tdCell(formatScheduledJobPlaybook(job), 18), tdCell(job.status || "", 10), tdCell(job.next_run_at || "—", 20), tdCell(lastRun, 24), @@ -10011,12 +10034,13 @@ async function renderScheduledJobs() { el("div", { class: "table-wrap" }, [ el("table", { class: "table table--compact" }, [ el("colgroup", {}, [ - el("col", { style: "width:16%" }), - el("col", { style: "width:18%" }), + el("col", { style: "width:14%" }), + el("col", { style: "width:14%" }), + el("col", { style: "width:12%" }), + el("col", { style: "width:8%" }), + el("col", { style: "width:12%" }), + el("col", { style: "width:12%" }), el("col", { style: "width:8%" }), - el("col", { style: "width:14%" }), - el("col", { style: "width:14%" }), - el("col", { style: "width:10%" }), el("col", { style: "width:10%" }), el("col", { style: "width:10%" }), ]), @@ -10024,6 +10048,7 @@ async function renderScheduledJobs() { el("tr", {}, [ el("th", { text: t("scheduledJobs.colName") }), el("th", { text: t("scheduledJobs.colSchedule") }), + el("th", { text: t("scheduledJobs.colPlaybook") }), el("th", { text: t("scheduledJobs.colStatus") }), el("th", { text: t("scheduledJobs.colNextRun") }), el("th", { text: t("scheduledJobs.colLastRun") }), diff --git a/runtime/application/gateway/inbound_service.py b/runtime/application/gateway/inbound_service.py index ea50f08a..c4ecc8f3 100644 --- a/runtime/application/gateway/inbound_service.py +++ b/runtime/application/gateway/inbound_service.py @@ -203,10 +203,16 @@ def _handle_productivity_commands( rows = store.scheduled_job_list(tenant_id=tenant_id, status=None, limit=10) if not rows: return "当前没有定时任务。" - lines = [ - f"- {r.name} | {r.schedule_kind}:{r.schedule_expr} | {r.status} | id={r.id[:8]}" - for r in rows - ] + lines = [] + for r in rows: + d = store.scheduled_job_to_dict(r) + playbook = "playbook" if d.get("playbook") else ("recipe" if d.get("has_recipe") else "reminder") + steps_n = int(d.get("steps_n") or 0) + last = str(d.get("last_run_status") or "").strip() or "—" + lines.append( + f"- {r.name} | {r.schedule_kind}:{r.schedule_expr} | {r.status} | " + f"{playbook} steps={steps_n} | last={last} | id={r.id[:8]}" + ) return "定时任务：\n" + "\n".join(lines) if t.startswith("暂停定时任务 ") or t.startswith("暂停定时 "): diff --git a/runtime/scheduler/cron_service.py b/runtime/scheduler/cron_service.py index ddf27246..04fccf0e 100644 --- a/runtime/scheduler/cron_service.py +++ b/runtime/scheduler/cron_service.py @@ -127,6 +127,11 @@ class CronService: "timezone": row.get("timezone"), "nextRunAt": row.get("next_run_at"), "lastRunAt": row.get("last_run_at"), + "lastRunStatus": row.get("last_run_status"), + "playbook": bool(row.get("playbook")), + "hasRecipe": bool(row.get("has_recipe")), + "stepsN": int(row.get("steps_n") or 0), + "recipeGoal": row.get("recipe_goal") or "", "specialist": row.get("specialist"), "interaction_mode": row.get("interaction_mode"), "delivery": row.get("delivery"), diff --git a/runtime/scheduler/recipe.py b/runtime/scheduler/recipe.py index e093c80b..baf6c744 100644 --- a/runtime/scheduler/recipe.py +++ b/runtime/scheduler/recipe.py @@ -319,6 +319,20 @@ def prompt_summary_from_recipe(recipe: dict[str, Any] | None, *, fallback: str = return str(fallback or "").strip() +def recipe_list_summary(recipe: dict[str, Any] | None) -> dict[str, Any]: + """Compact playbook signals for schedule_list / admin tables.""" + norm = normalize_recipe(recipe or {}) + steps = list(norm.get("steps") or []) + goal = str(norm.get("goal") or "").strip() + playbook = bool(goal) and len(steps) >= 2 + return { + "playbook": playbook, + "has_recipe": not recipe_is_empty(norm), + "steps_n": len(steps), + "recipe_goal": goal[:240], + } + + def preview_markdown( *, name: str, @@ -602,6 +616,7 @@ __all__ = [ "prompt_summary_from_recipe", "recipe_has_playbook", "recipe_is_empty", + "recipe_list_summary", "recipe_missing_fields", "resolve_effective_playbook_recipe", "resolve_ops_recipe_template", diff --git a/runtime/scheduler/service.py b/runtime/scheduler/service.py index 144060de..ae7dcc48 100644 --- a/runtime/scheduler/service.py +++ b/runtime/scheduler/service.py @@ -14,13 +14,15 @@ from runtime.scheduler.recipe import ( resolve_effective_playbook_recipe, ) 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.scheduler.turn_text import build_scheduled_turn_instruction, format_scheduled_skip_summary from runtime.worker import ensure_worker_started _LOCK = threading.Lock() _THREAD: threading.Thread | None = None _RUNNING = False +_OVERLAP_ACTIVE_STATUSES = frozenset({"queued", "running"}) + def _tick_interval_seconds() -> float: import os @@ -32,6 +34,162 @@ def _tick_interval_seconds() -> float: return 30.0 +def _overlap_stale_seconds() -> float: + import os + + raw = str(os.getenv("AIA_SCHEDULER_OVERLAP_STALE_SECONDS") or "10800").strip() + try: + return max(600.0, min(float(raw), 86400.0)) + except Exception: + return 10800.0 + + +def _parse_run_ts(raw: Any) -> datetime | None: + text = str(raw or "").strip() + if not text: + return None + try: + return datetime.fromisoformat(text.replace("Z", "+00:00")) + except Exception: + return None + + +def _find_blocking_scheduled_run( + store: Any, + *, + job_id: str, + tenant_id: str, + exclude_run_id: str = "", +) -> Any | None: + """Return an active (queued/running) run that should block a new enqueue. + + Stale active runs (older than AIA_SCHEDULER_OVERLAP_STALE_SECONDS) are marked + failed so a stuck worker cannot block the job forever. + """ + lister = getattr(store, "scheduled_job_run_list", None) + if not callable(lister): + return None + try: + rows = lister(job_id=str(job_id), tenant_id=str(tenant_id), limit=12) or [] + except Exception: + return None + exclude = str(exclude_run_id or "").strip() + stale_sec = _overlap_stale_seconds() + now = datetime.now(timezone.utc) + for row in rows: + rid = str(getattr(row, "id", "") or "") + if exclude and rid == exclude: + continue + status = str(getattr(row, "status", "") or "").strip().lower() + if status not in _OVERLAP_ACTIVE_STATUSES: + continue + started = _parse_run_ts(getattr(row, "started_at", None)) or _parse_run_ts( + getattr(row, "created_at", None) + ) + age = (now - started).total_seconds() if started is not None else 0.0 + if started is not None and age > stale_sec: + updater = getattr(store, "scheduled_job_run_update", None) + if callable(updater): + try: + updater( + run_id=rid, + tenant_id=str(tenant_id), + patch={ + "status": "failed", + "finished_at": now.isoformat(), + "error": "stale_running_cleared", + }, + ) + except Exception: + pass + continue + return row + return None + + +def _notify_overlapping_skip( + store: Any, + *, + job: Any, + overlapping_run_id: str, + reply_text: str, +) -> dict[str, Any]: + from runtime.scheduler.channel_delivery import deliver_scheduled_reply + + delivery_json = str(getattr(job, "delivery_json", "") or "{}") + try: + return deliver_scheduled_reply( + store, + tenant_id=str(getattr(job, "tenant_id", "") or ""), + reply_text=reply_text, + delivery_json=delivery_json, + ) + except Exception as exc: + return {"ok": False, "error": f"{type(exc).__name__}: {exc}"} + + +def _skip_overlapping_job_run( + store: Any, + *, + job: Any, + blocking: Any, + mode: str, +) -> dict[str, Any]: + tenant_id = str(getattr(job, "tenant_id", "") or "") + job_id = str(getattr(job, "id") or "") + lang = str(getattr(job, "lang", "") or "en") + overlapping_id = str(getattr(blocking, "id", "") or "") + summary = format_scheduled_skip_summary( + job_name=str(getattr(job, "name", "") or ""), + job_id=job_id, + overlapping_run_id=overlapping_id, + lang=lang, + ) + run = store.scheduled_job_run_create( + job_id=job_id, + tenant_id=tenant_id, + scheduled_at=str(getattr(job, "next_run_at", "") or datetime.now(timezone.utc).isoformat()), + status="skipped", + ) + delivery_status = _notify_overlapping_skip( + store, + job=job, + overlapping_run_id=overlapping_id, + reply_text=summary, + ) + store.scheduled_job_run_update( + run_id=run.id, + tenant_id=tenant_id, + patch={ + "status": "skipped", + "finished_at": datetime.now(timezone.utc).isoformat(), + "reply_text": summary, + "error": "overlapping_run", + "delivery_status": delivery_status, + }, + ) + # Advance the schedule so the tick does not re-fire every 30s while blocked. + if str(mode or "") == "scheduled": + try: + store.scheduled_job_reserve_next_run(job_id=job_id, tenant_id=tenant_id) + except Exception: + pass + store.scheduled_job_mark_run( + job_id=job_id, + tenant_id=tenant_id, + last_run_status="skipped", + pause_after=False, + ) + return { + "ok": True, + "skipped": True, + "reason": "overlapping_run", + "run_id": run.id, + "overlapping_run_id": overlapping_id, + "reply_text": summary, + } + + def _load_previous_run_context( store: Any, *, @@ -101,9 +259,19 @@ def enqueue_scheduled_job_run( *, job: Any, mode: str = "scheduled", + force_overlap: bool = False, ) -> dict[str, Any]: tenant_id = str(getattr(job, "tenant_id", "") or "") job_id = str(getattr(job, "id") or "") + if not force_overlap: + blocking = _find_blocking_scheduled_run(store, job_id=job_id, tenant_id=tenant_id) + if blocking is not None: + return _skip_overlapping_job_run( + store, + job=job, + blocking=blocking, + mode=mode, + ) run = store.scheduled_job_run_create( job_id=job_id, tenant_id=tenant_id, @@ -241,26 +409,46 @@ def enqueue_scheduled_job_run( def scheduler_tick(store: Any) -> dict[str, Any]: due = store.scheduled_job_list_due(limit=20) triggered = 0 + skipped = 0 errors: list[str] = [] for job in due: try: out = enqueue_scheduled_job_run(store, job=job, mode="scheduled") - if out.get("ok"): + if out.get("skipped"): + skipped += 1 + elif out.get("ok"): triggered += 1 else: errors.append(str(out.get("error") or "enqueue_failed")) except Exception as exc: errors.append(f"{getattr(job, 'id', '')}: {type(exc).__name__}: {exc}") - return {"ok": True, "due": len(due), "triggered": triggered, "errors": errors} + return { + "ok": True, + "due": len(due), + "triggered": triggered, + "skipped": skipped, + "errors": errors, + } -def run_scheduled_job_now(store: Any, *, tenant_id: str, job_id: str) -> dict[str, Any]: +def run_scheduled_job_now( + store: Any, + *, + tenant_id: str, + job_id: str, + force_overlap: bool = False, +) -> dict[str, Any]: job = store.scheduled_job_get(job_id=job_id, tenant_id=tenant_id) if not job: return {"ok": False, "error": "job_not_found"} if str(job.status or "") != "active": return {"ok": False, "error": "job_not_active"} - return enqueue_scheduled_job_run(store, job=job, mode="manual") + return enqueue_scheduled_job_run( + store, + job=job, + mode="manual", + force_overlap=force_overlap, + ) def _scheduler_loop(*, store: Any) -> None: diff --git a/runtime/scheduler/turn_text.py b/runtime/scheduler/turn_text.py index 1a51b7f4..83e74580 100644 --- a/runtime/scheduler/turn_text.py +++ b/runtime/scheduler/turn_text.py @@ -69,6 +69,33 @@ def format_scheduled_failure_summary( return f"[Scheduled job failed] {name}\nError: {err}\nCheck the job config or retry manually." +def format_scheduled_skip_summary( + *, + job_name: str = "", + job_id: str = "", + overlapping_run_id: str = "", + lang: str = "en", +) -> str: + """User-facing notice when a due tick is skipped because a prior run is still active.""" + name = str(job_name or "").strip() or str(job_id or "").strip() or "scheduled job" + oid = str(overlapping_run_id or "").strip() + if str(lang or "").lower().startswith("zh"): + lines = [ + f"[定时任务跳过] {name}", + "原因：上一轮仍在运行（overlapping），本轮已跳过以免叠跑。", + ] + if oid: + lines.append(f"进行中的 run：{oid[:12]}") + return "\n".join(lines) + lines = [ + f"[Scheduled job skipped] {name}", + "Reason: previous run still active (overlapping); this tick was skipped.", + ] + if oid: + lines.append(f"Active run: {oid[:12]}") + return "\n".join(lines) + + def build_scheduled_turn_instruction( *, prompt_text: str, @@ -181,6 +208,7 @@ __all__ = [ "append_previous_run_context", "build_scheduled_turn_instruction", "format_scheduled_failure_summary", + "format_scheduled_skip_summary", "format_scheduled_success_summary", "format_scheduled_user_reminder", "scheduled_turn_system_suffix", diff --git a/runtime/tools/experts/productivity/schedule_tools.py b/runtime/tools/experts/productivity/schedule_tools.py index 95da5a9c..54c2e3e2 100644 --- a/runtime/tools/experts/productivity/schedule_tools.py +++ b/runtime/tools/experts/productivity/schedule_tools.py @@ -470,13 +470,22 @@ def schedule_list_tool() -> ToolSpec: status = str(args.get("status") or "").strip() or None limit = int(args.get("limit") or 50) rows = store.scheduled_job_list(tenant_id=tenant_id, status=status, limit=limit) - return {"ok": True, "items": [store.scheduled_job_to_dict(r) for r in rows]} + items: list[dict[str, Any]] = [] + for r in rows: + d = store.scheduled_job_to_dict(r) + # List view: keep playbook signals, drop bulky recipe bodies for the LLM. + d.pop("recipe", None) + items.append(d) + return {"ok": True, "items": items} except Exception as e: return {"ok": False, "error": f"{type(e).__name__}: {e}"} return ToolSpec( name="schedule_list", - description="List scheduled jobs for a tenant.", + description=( + "List scheduled jobs for a tenant. Each item includes playbook/has_recipe/steps_n/" + "recipe_goal/last_run_status (full recipe bodies omitted; open the job detail for the full recipe)." + ), parameters={ "type": "object", "properties": { diff --git a/svc/persistence/scheduled_job_store.py b/svc/persistence/scheduled_job_store.py index ad85e812..24c9672b 100644 --- a/svc/persistence/scheduled_job_store.py +++ b/svc/persistence/scheduled_job_store.py @@ -620,6 +620,9 @@ class ScheduledJobStoreMixin: recipe = raw_recipe except Exception: recipe = {} + from runtime.scheduler.recipe import recipe_list_summary + + summary = recipe_list_summary(recipe) return { "id": job.id, "tenant_id": job.tenant_id, @@ -631,6 +634,10 @@ class ScheduledJobStoreMixin: "timezone": job.timezone, "prompt_text": job.prompt_text, "recipe": recipe, + "playbook": bool(summary.get("playbook")), + "has_recipe": bool(summary.get("has_recipe")), + "steps_n": int(summary.get("steps_n") or 0), + "recipe_goal": str(summary.get("recipe_goal") or ""), "interaction_mode": job.interaction_mode, "specialist": job.specialist, "lang": job.lang, diff --git a/tests/test_scheduler_overlap_and_list.py b/tests/test_scheduler_overlap_and_list.py new file mode 100644 index 00000000..ed543883 --- /dev/null +++ b/tests/test_scheduler_overlap_and_list.py @@ -0,0 +1,186 @@ +from __future__ import annotations + +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from runtime.scheduler.service import enqueue_scheduled_job_run +from runtime.scheduler.turn_text import format_scheduled_skip_summary +from runtime.tools.experts.productivity.schedule_tools import schedule_list_tool +from svc.persistence.assistant_store import reset_assistant_store_singleton +from svc.persistence.sqlite_store import SqliteStore + + +class OverlapSkipTextTests(unittest.TestCase): + def test_skip_summary_english(self) -> None: + text = format_scheduled_skip_summary( + job_name="Hourly congestion", + overlapping_run_id="abcdef12-3456", + lang="en", + ) + self.assertIn("[Scheduled job skipped]", text) + self.assertIn("overlapping", text) + self.assertIn("abcdef12-345", text) + + +class ScheduledOverlapAndListTests(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.db = Path(self._tmp.name) / "overlap.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 _make_job(self, *, recipe: dict | None = None, prompt: str = "remind drink"): + return self.store.scheduled_job_create( + tenant_id=self.tenant_id, + name="job-a", + prompt_text=prompt, + schedule_kind="interval", + schedule_expr="3600", + timezone_name="UTC", + lang="en", + delivery={"whatsapp": {"enabled": True, "target_type": "group", "chat_id": "x@g.us"}}, + recipe=recipe or {}, + created_by_user_id=self.user_id, + ) + + def test_enqueue_skips_when_prior_run_active(self) -> None: + job = self._make_job() + active = self.store.scheduled_job_run_create( + job_id=job.id, + tenant_id=self.tenant_id, + status="running", + ) + self.store.scheduled_job_run_update( + run_id=active.id, + tenant_id=self.tenant_id, + patch={"status": "running", "started_at": "2026-08-10T23:00:00+00:00"}, + ) + + with patch( + "runtime.scheduler.service.deliver_scheduled_reply", + create=True, + ): + # Patch via channel_delivery import path used inside helper. + with patch( + "runtime.scheduler.channel_delivery.deliver_scheduled_reply", + return_value={"ok": True, "channel": "whatsapp"}, + ) as deliver: + out = enqueue_scheduled_job_run(self.store, job=job, mode="scheduled") + + self.assertTrue(out.get("ok"), out) + self.assertTrue(out.get("skipped"), out) + self.assertEqual(out.get("reason"), "overlapping_run") + self.assertEqual(out.get("overlapping_run_id"), active.id) + deliver.assert_called_once() + skipped = self.store.scheduled_job_run_get(run_id=str(out["run_id"]), tenant_id=self.tenant_id) + assert skipped is not None + self.assertEqual(skipped.status, "skipped") + self.assertIn("overlapping", skipped.reply_text.lower()) + refreshed = self.store.scheduled_job_get(job_id=job.id, tenant_id=self.tenant_id) + assert refreshed is not None + self.assertEqual(refreshed.last_run_status, "skipped") + + def test_force_overlap_bypasses_skip(self) -> None: + job = self._make_job() + active = self.store.scheduled_job_run_create( + job_id=job.id, + tenant_id=self.tenant_id, + status="running", + ) + self.store.scheduled_job_run_update( + run_id=active.id, + tenant_id=self.tenant_id, + patch={"status": "running", "started_at": "2026-08-10T23:00:00+00:00"}, + ) + + from runtime.scheduler import service as sched_service + from runtime.scheduler.session_resolver import ResolvedSession + + resolved = ResolvedSession( + session_id="sess-1", + tenant_id=self.tenant_id, + user_id=self.user_id, + channel="admin_chat", + account_id="", + external_chat_id="", + external_user_id="", + is_group=False, + ) + self.store.oclaw_task_create = MagicMock(return_value=MagicMock(id="task-1")) # type: ignore[method-assign] + original_resolve = sched_service.resolve_scheduled_session + original_worker = sched_service.ensure_worker_started + try: + sched_service.resolve_scheduled_session = MagicMock(return_value=resolved) + sched_service.ensure_worker_started = MagicMock(return_value="worker-1") + out = enqueue_scheduled_job_run(self.store, job=job, mode="manual", force_overlap=True) + finally: + sched_service.resolve_scheduled_session = original_resolve + sched_service.ensure_worker_started = original_worker + + self.assertTrue(out.get("ok"), out) + self.assertFalse(bool(out.get("skipped"))) + self.assertIn("task_id", out) + + def test_schedule_list_exposes_playbook_signals(self) -> None: + recipe = { + "goal": "Hourly bandwidth check", + "steps": ["Query alarms", "Filter areas", "Deliver xlsx"], + "success_criteria": ["Report sent"], + } + self._make_job(recipe=recipe, prompt="Hourly bandwidth check") + self._make_job(prompt="提醒喝水") + + out = schedule_list_tool().handler( + {"tenant_id": self.tenant_id, "owner_user_id": self.user_id, "limit": 20} + ) + self.assertTrue(out.get("ok"), out) + items = out.get("items") or [] + self.assertGreaterEqual(len(items), 2) + playbook_items = [i for i in items if i.get("playbook")] + self.assertTrue(playbook_items) + hit = playbook_items[0] + self.assertEqual(hit.get("steps_n"), 3) + self.assertIn("Hourly bandwidth", str(hit.get("recipe_goal") or "")) + self.assertNotIn("recipe", hit) + + reminder = next(i for i in items if not i.get("playbook") and not i.get("has_recipe")) + self.assertEqual(int(reminder.get("steps_n") or 0), 0) + + def test_job_to_dict_includes_summary(self) -> None: + job = self._make_job( + recipe={ + "goal": "License report", + "steps": ["Query", "Attach"], + "success_criteria": ["Done"], + }, + prompt="License report", + ) + d = self.store.scheduled_job_to_dict(job) + self.assertTrue(d.get("playbook")) + self.assertEqual(d.get("steps_n"), 2) + self.assertTrue(d.get("has_recipe")) + self.assertIn("License", str(d.get("recipe_goal") or "")) + + +if __name__ == "__main__": + unittest.main()