Review of the run-history fallback: - Merge agent sessions with script-only output docs per execution instead of branching on 'any sessions exist': a job converted to no_agent keeps its id (update_job supports it), so a surviving historical agent session must not hide newer script-only fires. Same-execution rows dedupe by session span. - Decode output-doc filenames and read the execution ledger inside the owner profile's home scope: hermes_time resolves the configured zone through the current HERMES_HOME, so a cross-profile request decoded with the dashboard's zone and read the dashboard's executions.db. - Attach per-run status from the execution ledger (one row per fire) matched by claim window instead of last_run_at proximity, and surface terminal attempts with no surviving doc as their own rows — the latest failed run no longer disappears behind (or relabels) an older successful document. - Read output docs with encoding='utf-8-sig' (Windows footgun: PowerShell and some editors BOM files; plain utf-8 breaks on BOM-prefixed docs). Three regressions cover the converted-job merge, the per-profile timezone decode, and the missing-newest-doc case.
839 lines
36 KiB
Python
839 lines
36 KiB
Python
"""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/<job_id>/ — 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),
|
|
}]
|
|
|
|
|
|
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()
|
|
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["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 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 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 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))
|
|
|
|
|
|
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
|
|
# Names external plugins imported from this module before the Sep 2026 decomposition.
|
|
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
|
|
# The whole block is removed by reverting the commit that added it.
|
|
import logging # noqa: F401,E402
|
|
# ---- END PLUGIN-COMPAT ----
|