"""Cron dashboard routes. The ``*_sync`` workers, profile resolution and the threadpool wrapper (``_run_cron_dashboard_io``) live in web_server_cron and are reached through the late-binding seam so ``monkeypatch.setattr(web_server_cron, ...)`` keeps working. """ import asyncio import functools import re import time from contextlib import contextmanager from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional from fastapi import APIRouter, HTTPException, Request from fastapi.responses import JSONResponse from hermes_cli.web_deps import late from hermes_cli.config import cfg_get from hermes_cli.web_server_cron import ( _create_cron_job_sync, _cron_optional_text, _cron_string_list, _mutate_cron_for_profile, _normalize_dashboard_cron_script, _raise_if_cron_registration_error, _run_cron_dashboard_io, _validate_dashboard_cron_context_from, _validate_dashboard_cron_effective_job, ) from hermes_cli.web_models import AutomationBlueprintInstantiate, CronJobCreate, CronJobUpdate from hermes_cli.web_routers._common import log as _log from hermes_time import get_timezone as _get_timezone from hermes_constants import ( get_hermes_home as _get_hermes_home, reset_hermes_home_override as _reset_hermes_home_override, set_hermes_home_override as _set_hermes_home_override, ) router = APIRouter() _find_cron_job_profile = late("_find_cron_job_profile", "hermes_cli.web_server_cron") _fire_cron_job_for_profile = late("_fire_cron_job_for_profile", "hermes_cli.web_server_cron") _forward_cron_fire_to_gateway = late("_forward_cron_fire_to_gateway", "hermes_cli.web_server_cron") _gateway_intentionally_stopped = late("_gateway_intentionally_stopped", "hermes_cli.web_server_cron") _notify_cron_provider_for_profile = late("_notify_cron_provider_for_profile", "hermes_cli.web_server_cron") _call_cron_for_profile = late("_call_cron_for_profile", "hermes_cli.web_server_cron") load_config = late("load_config", "hermes_cli.config") _cron_profile_dicts = late("_cron_profile_dicts", "hermes_cli.web_server_cron") _cron_profile_home = late("_cron_profile_home", "hermes_cli.web_server_cron") _open_session_db_for_profile = late("_open_session_db_for_profile", "hermes_cli.web_server_sessions") _config_profile_scope = late("_config_profile_scope", "hermes_cli.web_server_profiles") def _job_not_found() -> HTTPException: return HTTPException(status_code=404, detail="Job not found") def _normalize_dashboard_cron_updates(updates: Dict[str, Any], profile_home: Path) -> Dict[str, Any]: """Normalize dashboard JSON into cron.jobs.update_job's storage shape. Stays in the dashboard adapter layer on purpose: cron/jobs.py is the source of truth for scheduling; this only translates form payloads into shapes the core functions already accept. """ normalized = dict(updates or {}) for key in ("model", "provider", "workdir"): if key in normalized: normalized[key] = _cron_optional_text(normalized[key]) if "script" in normalized: normalized["script"] = _normalize_dashboard_cron_script(normalized["script"], profile_home) if "base_url" in normalized: normalized["base_url"] = _cron_optional_text(normalized["base_url"], strip_trailing_slash=True) if "deliver" in normalized: normalized["deliver"] = _cron_optional_text(normalized["deliver"]) or "local" if "failure_deliver" in normalized: # Same normalization as deliver, but empty CLEARS the override (failures # fall back to deliver) rather than coalescing — the field is optional. normalized["failure_deliver"] = _cron_optional_text(normalized["failure_deliver"]) for key in ("context_from", "enabled_toolsets"): if key in normalized: normalized[key] = _cron_string_list(normalized[key]) return normalized def _job_owner_profile(job_id: str, profile: Optional[str]) -> Optional[str]: """Profile that holds ``job_id`` (its jobs.json and the state.db with its run sessions). ``profile`` is a caller *hint*, not proof of ownership: the Desktop lists jobs cross-profile (``?profile=all``) while its per-item calls carry the ambient active profile. Trusting that hint opened a profile that does not hold the job, so the ``cron_{job_id}_*`` id-range scan matched nothing and answered ``200 {"runs": []}`` — the UI rendered "No runs yet" for a job that had run many times (#115345). A hint that does hold the job still wins, so deliberately scoped lookups (the same job id in two profiles, e.g. a copied jobs.json) keep reading the named profile. """ if profile: jobs = _call_cron_for_profile(profile, "list_jobs", True) if any(j.get("id") == job_id or j.get("name") == job_id for j in jobs): return profile return _find_cron_job_profile(job_id) def _job_profile(job_id: str, profile: Optional[str]) -> str: """Owning profile for the get/update/pause/resume/trigger/delete family; 404 when no profile holds the job. Same hint validation as the run lookup, so a wrong-profile hint from the cross-profile list cannot 404 (or act on the wrong store for) a job the server can locate.""" selected = _job_owner_profile(job_id, profile) if not selected: raise _job_not_found() return selected def _found(job): if not job: raise _job_not_found() return job def _list_cron_jobs_sync(profile: str = "all"): requested = (profile or "all").strip() if requested.lower() != "all": return _call_cron_for_profile(requested, "list_jobs", True) # Aggregating across profiles can surface the SAME job id more than once — # e.g. a job copied into a second profile's cron/jobs.json during profile # creation. Deduplicate by id, deterministically preferring the default # profile's copy over per-iteration order (#51721): collect all jobs first, # then resolve duplicates by id with default-profile priority, rather than # keeping whichever copy happened to be seen first during the profile loop. all_jobs: List[Dict[str, Any]] = [] for item in _cron_profile_dicts(): name = str(item.get("name") or "") if not name: continue try: all_jobs.extend(_call_cron_for_profile(name, "list_jobs", True)) except Exception: _log.exception("Failed to list cron jobs for profile %s", name) by_id: Dict[str, Dict[str, Any]] = {} unkeyed: List[Dict[str, Any]] = [] for job in all_jobs: if not isinstance(job, dict): continue jid = job.get("id") or job.get("job_id") # cron.jobs._normalize_job_record() fills a missing id with the literal # sentinel string "unknown" — which is truthy, so a plain `if not jid` # check would NOT catch it and two genuinely different id-less jobs from # different profiles would collapse into one under this shared sentinel # key. Treat the sentinel the same as no id: never deduplicated. if not jid or jid == "unknown": unkeyed.append(job) continue existing = by_id.get(jid) if existing is None or ( job.get("is_default_profile") and not existing.get("is_default_profile") ): by_id[jid] = job return list(by_id.values()) + unkeyed def _get_cron_job_sync(job_id: str, profile: Optional[str] = None): return _found(_call_cron_for_profile(_job_profile(job_id, profile), "get_job", job_id)) _CRON_OUTPUT_FILENAME_FORMAT = "%Y-%m-%d_%H-%M-%S" def _cron_output_runs_dir(profile: Optional[str], job_id: str) -> Path: """Output docs live under the job's home — resolve it even without a hint.""" if profile: try: _, profile_home = _cron_profile_home(profile) except Exception: profile_home = _get_hermes_home() else: profile_home = _get_hermes_home() return Path(profile_home) / "cron" / "output" / job_id @contextmanager def _owner_home_scope(profile: Optional[str]): """Keep reads in the owner's profile context for the whole run-history build. Filename stems, the execution ledger and ``hermes_time``'s configured zone all resolve through the *current* ``get_hermes_home()``; without this scope a cross-profile listing decodes another profile's docs with the dashboard's zone (an hours-off timestamp) and reads the dashboard's executions.db. The dashboard's own profile needs no override (its home is already current). """ if not profile: yield None return try: _, profile_home = _cron_profile_home(profile) except Exception: yield None return if Path(profile_home).resolve() == _get_hermes_home().resolve(): yield None return token = _set_hermes_home_override(str(profile_home)) try: yield profile_home finally: _reset_hermes_home_override(token) def _cron_output_run_timestamp(path: Path) -> Optional[float]: """Epoch seconds for an output filename's wall time. save_job_output writes the stem with hermes_time.now() — the configured IANA timezone (HERMES_TIMEZONE / config), falling back to server-local. Read it back with that same ZONE, not a fixed offset snapshot of today's local offset: a snapshot is wrong when the configured zone differs from the server's, and off by an hour for a historical file across a DST transition. With no configured zone, astimezone() on the naive datetime resolves the offset in effect on the filename's own date. """ try: naive = datetime.strptime(path.stem, _CRON_OUTPUT_FILENAME_FORMAT) except ValueError: return None tz = _get_timezone() if tz is not None: return naive.replace(tzinfo=tz).timestamp() return naive.astimezone().timestamp() def _cron_output_run_preview(path: Path, max_chars: int = 180) -> str: try: raw = path.read_text(encoding="utf-8-sig", errors="replace") except OSError: return "" preview = re.sub(r"\s+", " ", raw).strip() if len(preview) <= max_chars: return preview return preview[: max_chars - 1].rstrip() + "…" def _cron_job_last_run_timestamp(job: Optional[Dict[str, Any]]) -> Optional[float]: if not isinstance(job, dict): return None raw = job.get("last_run_at") if isinstance(raw, (int, float)): return float(raw) if isinstance(raw, str): text = raw.strip() if not text: return None try: return datetime.fromisoformat(text.replace("Z", "+00:00")).timestamp() except ValueError: return None return None def _cron_output_status_label(job: Optional[Dict[str, Any]]) -> str: if not isinstance(job, dict): return "" status = str(job.get("last_status") or "").strip() if not status: return "" return status.replace("_", " ").upper() def _cron_output_run_row(started_at: float, title: str, preview: Optional[str]) -> Dict[str, Any]: return { "title": title, "preview": preview or None, "source": "cron_output", "started_at": started_at, "last_active": started_at, "ended_at": started_at, "input_tokens": 0, "output_tokens": 0, "message_count": 0, "tool_call_count": 0, "model": None, "cwd": None, "archived": False, "is_active": False, } def _iso_to_epoch(text: Any) -> Optional[float]: if not isinstance(text, str) or not text.strip(): return None try: return datetime.fromisoformat(text.strip().replace("Z", "+00:00")).timestamp() except ValueError: return None def _owner_profile_executions(canonical_job_id: str) -> List[Dict[str, Any]]: """Terminal execution-ledger rows for the job, newest first. Each script-only fire creates exactly one ledger row (claimed → completed / failed), so the ledger — not filename proximity to ``last_run_at`` — is the per-attempt record that pairs an output doc with its real status. Read inside the owner-home scope so it opens the OWNER's cron/executions.db. """ try: from cron.executions import list_executions rows = list_executions(job_id=canonical_job_id, limit=100) except Exception: return [] terminal: List[Dict[str, Any]] = [] for row in rows: if str(row.get("status") or "") not in ("completed", "failed", "unknown"): continue terminal.append({ "claimed_at": _iso_to_epoch(row.get("claimed_at")) or 0.0, "finished_at": _iso_to_epoch(row.get("finished_at")), "status": str(row.get("status") or ""), "error": str(row.get("error") or "").strip(), }) terminal.sort(key=lambda r: r["claimed_at"], reverse=True) return terminal def _execution_contains( attempt: Dict[str, Any], started_at: float, grace_seconds: float = 300.0, ) -> bool: """Whether an output doc's timestamp falls inside a ledger attempt's window. ``save_job_output`` runs after the script finishes but before ``finish_execution`` closes the attempt, and both clocks are ``hermes_time.now()`` in the owner's zone, so a doc written by an attempt lands within ``[claimed_at, finished_at + grace]`` of THAT attempt (a still- open attempt has no ``finished_at``). Older attempts closed before this doc was written, which is what keeps a fast-firing job's status from bleeding onto an earlier run's row. """ if started_at + 0.001 < attempt["claimed_at"]: return False finish = attempt["finished_at"] return finish is None or started_at <= finish + grace_seconds def _execution_status_title(status: str, error: str, fallback: str) -> str: label = (status or "").replace("_", " ").upper() if label == "UNKNOWN": return fallback if not label: return fallback return f"{label} · {error}" if (label == "FAILED" and error) else f"{label} · {fallback}" def _list_cron_output_runs( job: Optional[Dict[str, Any]], canonical_job_id: str, profile: Optional[str], limit: int, ) -> List[Dict[str, Any]]: """SessionDB-less run history for jobs that never create agent sessions. Script-only (no_agent) jobs deliberately skip SessionDB (cron/scheduler.run_job), so their completed runs are invisible to the run-history endpoint. Their output docs — one .md per fire under cron/output// — are the only per-run record. Rows mirror /api/sessions shape with source='cron_output' so the frontend reuses SessionInfo; ids use a cron_output: prefix that can never collide with SessionDB cron_{job_id}_* session ids. Status comes from the execution ledger (one row per fire), matched to a doc by claim window — never from ``last_run_at`` proximity, which mislabels an older doc when the newest run's doc is missing. Terminal attempts with no surviving doc get their own metadata rows. """ output_dir = _cron_output_runs_dir(profile, canonical_job_id) try: files = sorted( (path for path in output_dir.glob("*.md") if path.is_file()), key=lambda path: path.name, reverse=True, ) except OSError: files = [] executions = _owner_profile_executions(canonical_job_id) represented: set = set() runs: List[Dict[str, Any]] = [] for path in files[:limit]: started_at = _cron_output_run_timestamp(path) if started_at is None: try: started_at = path.stat().st_mtime except OSError: started_at = 0.0 preview = _cron_output_run_preview(path) title = preview or "Script-only run" for index, attempt in enumerate(executions): if attempt["status"] not in ("completed", "failed"): continue if _execution_contains(attempt, started_at): represented.add(index) title = _execution_status_title(attempt["status"], attempt["error"], title) break runs.append({ "id": f"cron_output:{canonical_job_id}:{path.stem}", **_cron_output_run_row(started_at, title, preview), }) # Terminal attempts whose output doc is gone (pruned, or the run never # wrote one — e.g. a failed fire) are still real executions: give each its # own metadata row from the ledger instead of letting the latest attempt's # status bleed onto an older surviving document. for index, attempt in enumerate(executions): if index in represented or attempt["status"] not in ("completed", "failed"): continue if len(runs) >= limit: break when = attempt["finished_at"] or attempt["claimed_at"] error = attempt["error"] runs.append({ "id": f"cron_output:{canonical_job_id}:exec:{index}", **_cron_output_run_row( when, _execution_status_title(attempt["status"], error, "Script-only run"), error or None, ), # Internal: the attempt's claim window, so the caller can drop this # row when a session already represents the same execution. "_claim_window": (attempt["claimed_at"], attempt["finished_at"]), }) runs.sort(key=lambda r: float(r.get("started_at") or 0), reverse=True) if runs: return runs # No output docs survived (pruned or never written) but the job HAS run: # surface one metadata-only row from last_run_at/last_status instead of the # bare "No runs" the issue reports. latest_ts = _cron_job_last_run_timestamp(job) if latest_ts is None: return [] preview = "" if isinstance(job, dict): preview = str(job.get("last_error") or "").strip() title = _cron_output_status_label(job) or "Script-only run" if preview: title = f"{title} · {preview}" return [{ "id": f"cron_output:{canonical_job_id}:latest", **_cron_output_run_row(latest_ts, title, preview), }] _CRON_RUN_SESSION_ID = re.compile(r"^cron_(.+)_\d{8}_\d{6}$") # The run's session row is written just AFTER its execution claim; the grace only # absorbs float/ISO rounding between the two clocks on one host. _OWNERSHIP_CLAIM_GRACE_SECONDS = 5.0 def _live_inflight_execution(canonical_job_id: str) -> Optional[Dict[str, Any]]: """The job's claimed/running ledger attempt under a live owner, or None (fail closed). Must run inside the owner-home scope so it reads the OWNER's executions.db. """ try: from cron.executions import live_inflight_execution return live_inflight_execution(canonical_job_id) except Exception: return None def _run_owned_by(session: Dict[str, Any], inflight: Optional[Dict[str, Any]]) -> bool: """Whether the scheduler still OWNS this never-closed run session (#88443). ``is_active`` is a 300s activity window, so a live run inside a long tool call (no heartbeat, no message) reads inactive exactly like a zombie whose process died. Ownership is the durable answer: the job's in-flight attempt is held by a live process and this session started under that claim. An older never-closed run of the same job predates the claim, so it stays a zombie. """ if not inflight or session.get("ended_at") is not None: return False claimed_at = _iso_to_epoch(inflight.get("claimed_at")) try: started_at = float(session.get("started_at") or 0) except (TypeError, ValueError): return False return claimed_at is not None and started_at >= claimed_at - _OWNERSHIP_CLAIM_GRACE_SECONDS def cron_run_scheduler_owned(session: Dict[str, Any], profile: Optional[str] = None) -> Optional[bool]: """``scheduler_owned`` for one session row, or None when it is not a cron run session. The session-detail endpoint stamps this so a client re-checking a run it already has open (a restored tab, a send) gets the same answer as the runs list. """ if session.get("source") != "cron": return None match = _CRON_RUN_SESSION_ID.match(str(session.get("id") or "")) if not match: return None if session.get("ended_at") is not None: return False with _owner_home_scope(profile): return _run_owned_by(session, _live_inflight_execution(match.group(1))) def _list_cron_job_runs_sync(job_id: str, profile: Optional[str] = None, limit: int = 20): """Run history for a cron job, newest first: agent sessions PLUS script-only fires. Agent runs are ordinary sessions with id ``cron_{job_id}_{timestamp}`` (see cron/scheduler.run_job); ``source='cron'`` plus the id prefix binds them to this job. Backed by ``SessionDB.list_cron_job_runs`` — a bounded id-range scan, so cost scales with the requested window, not total cron history. Script-only (no_agent) jobs never write sessions; their per-fire output docs (``_list_cron_output_runs``) fill the gaps. The two representations are reconciled PER EXECUTION, not by branch: a job can switch modes without changing its id (update_job supports no_agent on an existing record), so a surviving historical agent session must not hide newer script-only fires and a recent session must not hide older script output. Rows carrying the same wall-clock execution (an agent fire writes BOTH a session and an output doc) are de-duplicated by timestamp window; script-only rows that match no session are added in. All owner-scoped reads (SessionDB, output docs, execution ledger, filename timezone) run inside the owner's profile scope so the dashboard's own zone/store is never borrowed cross-profile. """ selected = _job_owner_profile(job_id, profile) # job_id may be a human name; resolve to the canonical id used in run-session ids. canonical = job_id job = None if selected: job = _call_cron_for_profile(selected, "get_job", job_id) if job and job.get("id"): canonical = str(job["id"]) try: limit_n = max(1, min(int(limit), 100)) except (TypeError, ValueError): limit_n = 20 with _owner_home_scope(selected): db = _open_session_db_for_profile(selected, read_only=True) try: session_runs = db.list_cron_job_runs(canonical, limit=limit_n, offset=0) finally: db.close() now = time.time() inflight = _live_inflight_execution(canonical) for s in session_runs: s["is_active"] = s.get("ended_at") is None and (now - s.get("last_active", s.get("started_at", 0))) < 300 s["scheduler_owned"] = _run_owned_by(s, inflight) s["archived"] = bool(s.get("archived")) if selected: s["profile"] = selected doc_runs = _list_cron_output_runs(job, canonical, selected, limit_n) return {"runs": _reconcile_cron_runs(session_runs, doc_runs, limit_n), "limit": limit_n} def _reconcile_cron_runs( session_runs: List[Dict[str, Any]], doc_runs: List[Dict[str, Any]], limit: int, ) -> List[Dict[str, Any]]: """Merge session rows and output-doc rows per execution, newest first. An agent fire writes BOTH a session and an output doc for the same execution; keeping both would double-render the run. Sessions carry the richer record (tokens, message counts, chat navigation), so a doc row within ``_SESSION_DOC_MATCH_SECONDS`` of a session row is the same execution and the session wins. Doc rows matching no session are distinct executions — script-only fires — and are kept. A session row must never be dropped in favor of a doc: the doc is the fallback representation. """ _SESSION_DOC_MATCH_SECONDS = 300.0 merged = list(session_runs) for doc in doc_runs: doc = {k: v for k, v in doc.items() if not k.startswith("_")} doc_ts = float(doc.get("started_at") or 0) if any(_doc_matches_session(doc_ts, s, _SESSION_DOC_MATCH_SECONDS) for s in session_runs): continue merged.append(doc) merged.sort(key=lambda r: float(r.get("started_at") or 0), reverse=True) return merged[:limit] def _doc_matches_session(doc_ts: float, session: Dict[str, Any], grace_seconds: float) -> bool: """Whether an output doc belongs to a session's run. The doc is written when the run FINISHES, so its filename timestamp sits at the run's end — compare against the session's full span, not its start: ``[started_at - grace, last_active + grace]`` (``last_active`` covers the still-open case where ``ended_at`` is None). """ start = float(session.get("started_at") or 0) end = float(session.get("ended_at") or session.get("last_active") or start) return start - grace_seconds <= doc_ts <= end + grace_seconds _EXECUTION_FIELDS = {"prompt", "skill", "skills", "script", "no_agent"} def _update_cron_job_sync(job_id: str, body: CronJobUpdate, profile: Optional[str] = None): selected = _job_profile(job_id, profile) try: profile_name, profile_home = _cron_profile_home(selected) existing = _found(_call_cron_for_profile(profile_name, "get_job", job_id)) updates = _normalize_dashboard_cron_updates(body.updates, profile_home) if "context_from" in updates: _validate_dashboard_cron_context_from(updates.get("context_from"), profile_name) if _EXECUTION_FIELDS.intersection(updates): effective = {**existing, **updates} if "skills" in updates and "skill" not in updates: effective["skill"] = None _validate_dashboard_cron_effective_job(effective) job = _mutate_cron_for_profile(profile_name, "update_job", job_id, updates) except HTTPException: raise except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc return _found(job) def _pause_cron_job_sync(job_id: str, profile: Optional[str] = None): return _found(_mutate_cron_for_profile(_job_profile(job_id, profile), "pause_job", job_id)) def _resume_cron_job_sync(job_id: str, profile: Optional[str] = None): return _found(_mutate_cron_for_profile(_job_profile(job_id, profile), "resume_job", job_id)) def _trigger_cron_job_sync(job_id: str, profile: Optional[str] = None): selected = _job_profile(job_id, profile) job = _found(_call_cron_for_profile(selected, "resolve_job_ref", job_id)) # Never expose the job as due before claiming it: the built-in ticker and # external/manual fire paths share one durable claim, so only one executes # this run even racing across processes. Active jobs keep the legacy call # shape; paused jobs need the explicit force flag to resume + claim atomically. force = not job.get("enabled", True) or job.get("state") == "paused" ran = _fire_cron_job_for_profile(selected, job["id"], force=force) refreshed = _call_cron_for_profile(selected, "get_job", job["id"]) if refreshed and refreshed.get("last_run_at") != job.get("last_run_at"): return refreshed if not ran: raise HTTPException(status_code=409, detail="Job is already running or was claimed by another scheduler") if refreshed: return refreshed # A one-shot may remove itself after exhausting repeat=1: keep the response # shape without inventing an outcome the store no longer holds; the list # refresh removes the completed row. return {**job, "enabled": False, "state": "completed"} def _delete_cron_job_sync(job_id: str, profile: Optional[str] = None): selected = _job_profile(job_id, profile) try: removed = _mutate_cron_for_profile(selected, "remove_job", job_id) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc if not removed: raise _job_not_found() return {"ok": True} # Retry-After (seconds) on retryable cron-fire 503s: sized to clear a # scale-to-zero wake or gateway restart so a scheduler that honors it spaces its # next attempt past the outage instead of burning its retry budget in it. _CRON_FIRE_RETRY_AFTER_SECONDS = 60 @router.get("/api/cron/jobs") async def list_cron_jobs(profile: str = "all"): return await _run_cron_dashboard_io(_list_cron_jobs_sync, profile) @router.get("/api/cron/jobs/{job_id}") async def get_cron_job(job_id: str, profile: Optional[str] = None): return await _run_cron_dashboard_io(_get_cron_job_sync, job_id, profile) @router.get("/api/cron/jobs/{job_id}/runs") async def list_cron_job_runs(job_id: str, profile: Optional[str] = None, limit: int = 20): return await _run_cron_dashboard_io(_list_cron_job_runs_sync, job_id, profile, limit) @router.post("/api/cron/jobs") async def create_cron_job(body: CronJobCreate, profile: Optional[str] = None): return await _run_cron_dashboard_io(_create_cron_job_sync, body, profile) @router.get("/api/cron/delivery-targets") async def get_cron_delivery_targets(profile: Optional[str] = None): """Delivery targets for the cron dropdown: implicit ``local`` plus the configured gateway platforms (a platform without a cron home channel is still listed with ``home_target_set: false`` so the UI can say so). ``cron_delivery_targets()`` reads each platform's home channel through ``get_secret``, which fails closed once this process hosts more than one profile home (the dashboard/desktop ``serve`` backend flips multi-profile hosting on the first ``?profile=`` request). The read must therefore run inside the profile scope, exactly like the sibling cron routes — otherwise the poll raises ``UnscopedSecretError`` on every tick and the dropdown silently loses every configured platform.""" targets = [{"id": "local", "name": "Local (save only)", "home_target_set": True, "home_env_var": None}] try: from cron.scheduler_delivery import cron_delivery_targets with _config_profile_scope(profile): targets.extend(cron_delivery_targets()) except HTTPException: raise # an unknown ?profile= is the scope's 404, not a missing platform list except Exception: _log.exception("GET /api/cron/delivery-targets failed") return {"targets": targets} @router.put("/api/cron/jobs/{job_id}") async def update_cron_job(job_id: str, body: CronJobUpdate, profile: Optional[str] = None): return await _run_cron_dashboard_io(_update_cron_job_sync, job_id, body, profile) @router.post("/api/cron/jobs/{job_id}/pause") async def pause_cron_job(job_id: str, profile: Optional[str] = None): return await _run_cron_dashboard_io(_pause_cron_job_sync, job_id, profile) @router.post("/api/cron/jobs/{job_id}/resume") async def resume_cron_job(job_id: str, profile: Optional[str] = None): return await _run_cron_dashboard_io(_resume_cron_job_sync, job_id, profile) @router.post("/api/cron/jobs/{job_id}/trigger") async def trigger_cron_job(job_id: str, profile: Optional[str] = None): return await _run_cron_dashboard_io(_trigger_cron_job_sync, job_id, profile) @router.delete("/api/cron/jobs/{job_id}") async def delete_cron_job(job_id: str, profile: Optional[str] = None): return await _run_cron_dashboard_io(_delete_cron_job_sync, job_id, profile) @router.post("/api/cron/fire") async def cron_fire_webhook(request: Request): """Chronos managed-cron fire webhook (NAS -> agent) — gateway forwarder. Gated by the NAS-minted JWT (path is in ``PUBLIC_API_PATHS``), not the dashboard cookie. Execution belongs to the GATEWAY process (it owns the live platform adapters relay-fronted and E2EE targets need), so the fire is forwarded to the gateway api_server's own ``/api/cron/fire`` on loopback and its response passed through (the gateway re-verifies the JWT). Gateway unreachable -> 503 so NAS retries; deliberately NO local-execution fallback. """ from plugins.cron_providers.chronos.verify import get_fire_verifier auth = request.headers.get("Authorization", "") token = auth[7:].strip() if auth.startswith("Bearer ") else "" cfg = await asyncio.to_thread(load_config) claims = get_fire_verifier()( token=token, expected_audience=cfg_get(cfg, "cron", "chronos", "expected_audience", default=""), jwks_or_key=cfg_get(cfg, "cron", "chronos", "nas_jwks_url", default="") or None, issuer=cfg_get(cfg, "cron", "chronos", "portal_url", default="") or None, ) if claims is None: return JSONResponse({"error": "invalid fire token"}, status_code=401) try: body = await request.json() except Exception: body = {} job_id = (body or {}).get("job_id") if isinstance(body, dict) else None if not job_id: return JSONResponse({"error": "missing job_id"}, status_code=400) # Walks every profile's job list (file I/O) — off the event loop. profile = await _run_cron_dashboard_io(_find_cron_job_profile, job_id) if not profile: # job is gone (cancelled / completed): 200 so NAS does not retry return JSONResponse({"status": "gone", "job_id": job_id}, status_code=200) forwarded = await _forward_cron_fire_to_gateway(profile, job_id, auth) if forwarded is None: # Stamp the miss on the job record (last_fire_error) so the dead hop is # visible in `cronjob list` / the dashboard. Best-effort: visibility # must never break the retry contract below. try: await _run_cron_dashboard_io( _call_cron_for_profile, profile, "note_fire_forward_failure", job_id, "scheduled fire could not be forwarded to the gateway " "api_server (127.0.0.1 loopback unreachable); the gateway " "process may be down or its api_server adapter not bound " "(missing API_SERVER_KEY)", ) except Exception: _log.debug("could not stamp last_fire_error for %s", job_id, exc_info=True) # Split by operator intent: a deliberately stopped gateway (durable # desired_state == "stopped") can never be reached by retrying, so drop # with 200 + a structured log line — the Chronos provider re-arms every # job on the next gateway start. A transient window (wake, restart, # crash loop) keeps the retryable 503 with a Retry-After hint. if await _run_cron_dashboard_io(_gateway_intentionally_stopped, profile): _log.info( "cron fire dropped: gateway for profile %r is deliberately " "stopped (desired_state=stopped); job %s will resume via " "Chronos reconcile on next gateway start", profile, job_id, ) return JSONResponse( { "status": "gateway_stopped", "detail": "gateway deliberately stopped; fire dropped, jobs re-arm on next gateway start", "job_id": job_id, "profile": profile, }, status_code=200, ) return JSONResponse( {"error": "gateway unreachable; retry", "job_id": job_id, "profile": profile}, status_code=503, headers={"Retry-After": str(_CRON_FIRE_RETRY_AFTER_SECONDS)}, ) status_code, gateway_body = forwarded if isinstance(gateway_body, dict): gateway_body.setdefault("job_id", job_id) # The gateway's own 503s (draining, admission failure) are equally transient. headers = {"Retry-After": str(_CRON_FIRE_RETRY_AFTER_SECONDS)} if status_code == 503 else None return JSONResponse(gateway_body, status_code=status_code, headers=headers) @router.get("/api/cron/blueprints") async def list_cron_blueprints(profile: Optional[str] = None): """Blueprint catalog as form schemas; the ``deliver`` slot's options are rewritten from the actually configured gateway platforms.""" try: from cron.blueprint_catalog import CATALOG, blueprint_catalog_entry deliver_options = None try: from cron.scheduler_delivery import cron_delivery_targets with _config_profile_scope(profile): platforms = [t["id"] for t in cron_delivery_targets() if t.get("id")] deliver_options = ["origin", "local", *platforms] except HTTPException: raise # an unknown ?profile= is the scope's 404, not a reason for static options except Exception: _log.debug("cron_delivery_targets unavailable; using static deliver options", exc_info=True) entries = [] for r in CATALOG: entry = blueprint_catalog_entry(r) if deliver_options: for f in entry.get("fields", []): if f.get("name") == "deliver": f["options"] = deliver_options entries.append(entry) return {"blueprints": entries} except HTTPException: raise except Exception as e: _log.exception("GET /api/cron/blueprints failed") raise HTTPException(status_code=500, detail=str(e)) @router.post("/api/cron/blueprints/instantiate") async def instantiate_blueprint(body: AutomationBlueprintInstantiate, profile: str = "default"): """Fill a blueprint's slots and create the cron job (form-submit path).""" try: from cron.blueprint_catalog import BlueprintFillError, fill_blueprint, get_blueprint blueprint = get_blueprint(body.blueprint) if blueprint is None: raise HTTPException(status_code=404, detail=f"Unknown blueprint: {body.blueprint}") try: spec = fill_blueprint(blueprint, body.values) except BlueprintFillError as exc: # field-level error — 422 so the form shows it inline raise HTTPException(status_code=422, detail=str(exc)) from exc # Blueprint jobs deliver to the dashboard's configured target by default; # the form's deliver slot overrides via spec["deliver"]. spec.pop("origin", None) # Off-loop like the siblings; partial keeps **spec keys from colliding # with the wrapper's own parameters. _create = functools.partial(_call_cron_for_profile, profile, "create_job", **spec) created = await _run_cron_dashboard_io(_create) # Reconcile the profile-scoped provider (file I/O + NAS calls) off-loop. await _run_cron_dashboard_io(_notify_cron_provider_for_profile, profile) return created except HTTPException: raise except Exception as e: _raise_if_cron_registration_error(e) _log.exception("POST /api/cron/blueprints/instantiate failed") raise HTTPException(status_code=400, detail=str(e))