332 lines
13 KiB
Python
332 lines
13 KiB
Python
"""Scheduled tasks — automatic agent actions, type cron (#170).
|
|
|
|
Tasks are persisted in ``data/scheduled_tasks.json`` (guarded by an RLock,
|
|
same pattern as the other JSON stores). Supported actions reuse the existing
|
|
mutation/notification services — no new write path:
|
|
|
|
* ``create_file`` → ``backend.services.mutations.create_file``;
|
|
* ``append_to_file`` → ``backend.services.mutations.append_to_file``;
|
|
* ``notify`` → ``backend.notify.broadcast`` (trigger ``manual``).
|
|
|
|
Supported schedules:
|
|
|
|
* ``interval_hours`` — every N hours (N >= 0.25);
|
|
* ``daily_time`` — once a day at ``HH:MM`` (local server time);
|
|
* ``once_at`` — one shot at an ISO-8601 datetime (past = due immediately).
|
|
|
|
On failure the task records ``last_error`` and a ``schedule_failure``
|
|
broadcast is emitted to the notification channels (#168) — best effort,
|
|
never recursive (a failing ``notify`` action does not rebroadcast).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import threading
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger("obsigate.scheduler")
|
|
|
|
DATA_DIR = Path(os.environ.get("OBSIGATE_DATA_DIR", "data"))
|
|
TASKS_FILE = DATA_DIR / "scheduled_tasks.json"
|
|
|
|
ACTION_KINDS = ("create_file", "append_to_file", "notify")
|
|
SCHEDULE_KINDS = ("interval_hours", "daily_time", "once_at")
|
|
|
|
_lock = threading.RLock()
|
|
|
|
|
|
# ── Store ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _read_tasks() -> list[dict[str, Any]]:
|
|
if not TASKS_FILE.exists():
|
|
return []
|
|
try:
|
|
data = json.loads(TASKS_FILE.read_text(encoding="utf-8"))
|
|
return data if isinstance(data, list) else []
|
|
except (json.JSONDecodeError, OSError):
|
|
return []
|
|
|
|
|
|
def _write_tasks(tasks: list[dict[str, Any]]) -> None:
|
|
TASKS_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = TASKS_FILE.with_suffix(".tmp")
|
|
tmp.write_text(json.dumps(tasks, indent=2, default=str), encoding="utf-8")
|
|
tmp.replace(TASKS_FILE)
|
|
|
|
|
|
def list_tasks() -> list[dict[str, Any]]:
|
|
"""Return all scheduled tasks (newest first)."""
|
|
return sorted(_read_tasks(), key=lambda t: t.get("created_at", ""), reverse=True)
|
|
|
|
|
|
def get_task(task_id: str) -> dict[str, Any] | None:
|
|
"""Return one task by id, or None."""
|
|
for task in _read_tasks():
|
|
if task.get("id") == task_id:
|
|
return task
|
|
return None
|
|
|
|
|
|
# ── Validation ─────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _validate_action(action: dict[str, Any]) -> dict[str, Any]:
|
|
kind = action.get("kind")
|
|
if kind not in ACTION_KINDS:
|
|
raise ValueError(f"Action inconnue : {kind} (attendu : {', '.join(ACTION_KINDS)})")
|
|
params = dict(action.get("params") or {})
|
|
if kind in ("create_file", "append_to_file"):
|
|
if not str(params.get("vault") or "").strip():
|
|
raise ValueError("params.vault requis pour create_file/append_to_file")
|
|
if not str(params.get("path") or "").strip():
|
|
raise ValueError("params.path requis pour create_file/append_to_file")
|
|
if kind == "append_to_file" and not str(params.get("content") or ""):
|
|
raise ValueError("params.content requis pour append_to_file")
|
|
elif kind == "notify":
|
|
if not str(params.get("title") or "").strip():
|
|
raise ValueError("params.title requis pour notify")
|
|
if not str(params.get("message") or "").strip():
|
|
raise ValueError("params.message requis pour notify")
|
|
return {"kind": kind, "params": params}
|
|
|
|
|
|
def _validate_schedule(schedule: dict[str, Any]) -> dict[str, Any]:
|
|
kind = schedule.get("kind")
|
|
if kind not in SCHEDULE_KINDS:
|
|
raise ValueError(f"Planification inconnue : {kind} (attendu : {', '.join(SCHEDULE_KINDS)})")
|
|
if kind == "interval_hours":
|
|
hours = float(schedule.get("hours") or 0)
|
|
if hours < 0.25:
|
|
raise ValueError("hours doit être >= 0.25")
|
|
return {"kind": kind, "hours": hours}
|
|
if kind == "daily_time":
|
|
at = str(schedule.get("at") or "").strip()
|
|
try:
|
|
datetime.strptime(at, "%H:%M")
|
|
except ValueError:
|
|
raise ValueError("at doit être au format HH:MM (ex. 08:30)") from None
|
|
return {"kind": kind, "at": at}
|
|
# once_at
|
|
at = str(schedule.get("at") or "").strip()
|
|
try:
|
|
parsed = datetime.fromisoformat(at)
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
raise ValueError("at doit être une date ISO-8601 (ex. 2026-10-05T08:30:00)") from None
|
|
return {"kind": kind, "at": parsed.isoformat()}
|
|
|
|
|
|
# ── CRUD ───────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def create_task(
|
|
name: str,
|
|
action: dict[str, Any],
|
|
schedule: dict[str, Any],
|
|
*,
|
|
created_by: str = "api",
|
|
enabled: bool = True,
|
|
) -> dict[str, Any]:
|
|
"""Create a scheduled task. Raises ValueError on invalid action/schedule."""
|
|
validated_action = _validate_action(action)
|
|
validated_schedule = _validate_schedule(schedule)
|
|
now = datetime.now(timezone.utc)
|
|
with _lock:
|
|
tasks = _read_tasks()
|
|
task = {
|
|
"id": str(uuid.uuid4()),
|
|
"name": (name or validated_action["kind"]).strip() or validated_action["kind"],
|
|
"action": validated_action,
|
|
"schedule": validated_schedule,
|
|
"enabled": bool(enabled),
|
|
"created_by": created_by,
|
|
"created_at": now.isoformat(),
|
|
"last_run_at": None,
|
|
"last_status": None,
|
|
"last_error": None,
|
|
"run_count": 0,
|
|
"next_run_at": compute_next_run(
|
|
{"schedule": validated_schedule, "last_run_at": None}, now
|
|
).isoformat(),
|
|
}
|
|
tasks.append(task)
|
|
_write_tasks(tasks)
|
|
logger.info(f"Scheduled task created: '{task['name']}' ({validated_schedule['kind']})")
|
|
return task
|
|
|
|
|
|
def update_task(task_id: str, updates: dict[str, Any]) -> dict[str, Any] | None:
|
|
"""Update name/enabled/action/schedule. Returns None when unknown."""
|
|
with _lock:
|
|
tasks = _read_tasks()
|
|
for task in tasks:
|
|
if task.get("id") != task_id:
|
|
continue
|
|
if updates.get("name"):
|
|
task["name"] = str(updates["name"])
|
|
if "enabled" in updates:
|
|
task["enabled"] = bool(updates["enabled"])
|
|
if "action" in updates:
|
|
task["action"] = _validate_action(updates["action"])
|
|
if "schedule" in updates:
|
|
task["schedule"] = _validate_schedule(updates["schedule"])
|
|
task["next_run_at"] = compute_next_run(task).isoformat()
|
|
_write_tasks(tasks)
|
|
return task
|
|
return None
|
|
|
|
|
|
def delete_task(task_id: str) -> bool:
|
|
"""Delete a task. Returns False when unknown."""
|
|
with _lock:
|
|
tasks = _read_tasks()
|
|
remaining = [t for t in tasks if t.get("id") != task_id]
|
|
if len(remaining) == len(tasks):
|
|
return False
|
|
_write_tasks(remaining)
|
|
return True
|
|
|
|
|
|
# ── Scheduling ─────────────────────────────────────────────────────────────
|
|
|
|
|
|
def compute_next_run(task: dict[str, Any], now: datetime | None = None) -> datetime:
|
|
"""Compute the next due datetime for *task*."""
|
|
now = now or datetime.now(timezone.utc)
|
|
if now.tzinfo is None:
|
|
now = now.replace(tzinfo=timezone.utc)
|
|
schedule = task.get("schedule", {})
|
|
kind = schedule.get("kind")
|
|
last_run_at = task.get("last_run_at")
|
|
last = None
|
|
if last_run_at:
|
|
try:
|
|
last = datetime.fromisoformat(str(last_run_at))
|
|
if last.tzinfo is None:
|
|
last = last.replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
last = None
|
|
if kind == "interval_hours":
|
|
hours = float(schedule.get("hours", 24))
|
|
base = last or now
|
|
nxt = base + timedelta(hours=hours)
|
|
# Première planification : due dès maintenant + intervalle ? Non —
|
|
# la tâche démarre au prochain intervalle, sauf retard déjà accumulé.
|
|
if last is None:
|
|
nxt = now + timedelta(hours=hours)
|
|
return max(now, nxt)
|
|
if kind == "daily_time":
|
|
hour, minute = (str(schedule.get("at", "08:00")) + ":00").split(":")[:2]
|
|
candidate = now.replace(hour=int(hour), minute=int(minute), second=0, microsecond=0)
|
|
if candidate <= now:
|
|
candidate += timedelta(days=1)
|
|
return candidate
|
|
if kind == "once_at":
|
|
try:
|
|
at = datetime.fromisoformat(str(schedule.get("at")))
|
|
if at.tzinfo is None:
|
|
at = at.replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
return now
|
|
if task.get("last_run_at"):
|
|
return datetime.max.replace(tzinfo=timezone.utc) # déjà exécutée
|
|
return at
|
|
return now + timedelta(hours=24)
|
|
|
|
|
|
def _execute_action(task: dict[str, Any]) -> dict[str, Any]:
|
|
action = task["action"]
|
|
kind = action["kind"]
|
|
params = action["params"]
|
|
if kind == "create_file":
|
|
from backend.services.mutations import create_file
|
|
|
|
return create_file(
|
|
params["vault"],
|
|
params["path"],
|
|
params.get("content", ""),
|
|
overwrite=bool(params.get("overwrite", False)),
|
|
)
|
|
if kind == "append_to_file":
|
|
from backend.services.mutations import append_to_file
|
|
|
|
return append_to_file(params["vault"], params["path"], params.get("content", ""))
|
|
if kind == "notify":
|
|
from backend.notify import broadcast
|
|
|
|
results = broadcast("manual", str(params["title"]), str(params.get("message", "")))
|
|
return {"broadcast": results}
|
|
raise ValueError(f"Action inconnue : {kind}")
|
|
|
|
|
|
def run_task(task_id: str, *, manual: bool = False) -> dict[str, Any]:
|
|
"""Execute one task now (manual or due). Records status; notifies on failure."""
|
|
with _lock:
|
|
tasks = _read_tasks()
|
|
task = next((t for t in tasks if t.get("id") == task_id), None)
|
|
if task is None:
|
|
raise KeyError(task_id)
|
|
if not task.get("enabled", True) and not manual:
|
|
return {"task_id": task_id, "skipped": True, "reason": "disabled"}
|
|
try:
|
|
result = _execute_action(task)
|
|
task["last_run_at"] = datetime.now(timezone.utc).isoformat()
|
|
task["last_status"] = "ok"
|
|
task["last_error"] = None
|
|
task["run_count"] = int(task.get("run_count", 0)) + 1
|
|
if task.get("schedule", {}).get("kind") == "once_at":
|
|
task["enabled"] = False # one-shot consommé
|
|
task["next_run_at"] = compute_next_run(task).isoformat()
|
|
_write_tasks(tasks)
|
|
if manual:
|
|
from backend.notify import broadcast
|
|
|
|
broadcast("schedule_success", f"Tâche « {task['name']} » OK", "Exécution manuelle réussie.")
|
|
return {"task_id": task_id, "ok": True, "result": result}
|
|
except Exception as e:
|
|
task["last_run_at"] = datetime.now(timezone.utc).isoformat()
|
|
task["last_status"] = "error"
|
|
task["last_error"] = str(e)
|
|
task["run_count"] = int(task.get("run_count", 0)) + 1
|
|
task["next_run_at"] = compute_next_run(task).isoformat()
|
|
_write_tasks(tasks)
|
|
logger.warning(f"Scheduled task '{task.get('name')}' failed: {e}")
|
|
if task["action"]["kind"] != "notify":
|
|
try:
|
|
from backend.notify import broadcast
|
|
|
|
broadcast(
|
|
"schedule_failure",
|
|
f"Échec tâche « {task.get('name')} »",
|
|
f"{e}",
|
|
)
|
|
except Exception:
|
|
logger.debug("Failure notification broadcast failed", exc_info=True)
|
|
return {"task_id": task_id, "ok": False, "error": str(e)}
|
|
|
|
|
|
def tick(now: datetime | None = None) -> list[dict[str, Any]]:
|
|
"""Run every due task. Returns per-task outcomes (empty when idle)."""
|
|
now = now or datetime.now(timezone.utc)
|
|
outcomes: list[dict[str, Any]] = []
|
|
for task in _read_tasks():
|
|
if not task.get("enabled", True):
|
|
continue
|
|
try:
|
|
next_run = datetime.fromisoformat(str(task.get("next_run_at") or ""))
|
|
if next_run.tzinfo is None:
|
|
next_run = next_run.replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
next_run = compute_next_run(task, now)
|
|
if next_run <= now:
|
|
outcomes.append(run_task(task["id"]))
|
|
return outcomes
|