diff --git a/cron/AGENTS.md b/cron/AGENTS.md index fa7009421b..64a7496699 100644 --- a/cron/AGENTS.md +++ b/cron/AGENTS.md @@ -13,7 +13,8 @@ schedule via the `cronjob` tool; users via `hermes cron list|add|edit|pause|resu injected into the prompt; `no_agent=True` makes the script the whole job; always a FILE under `HERMES_HOME/scripts/`, never a command line — `script='echo hi'` is refused at creation), `context_from` (chain job A's last output into job B's prompt), `workdir` (run with that directory's `AGENTS.md`/`CLAUDE.md` -loaded), multi-platform delivery. +loaded), `timezone` (pin the cron expression's wall clock to an IANA zone; absent = the +profile's zone), multi-platform delivery. Hardening invariants — each guards a real failure; don't weaken without answering for it: - **Inactivity watchdog** on cron agent sessions (`_cron_inactivity_seconds()`): default 600s idle, @@ -22,6 +23,15 @@ Hardening invariants — each guards a real failure; don't weaken without answer is never cut off. Attached scripts (pre-run or `no_agent`) are bounded separately by the script timeout (`_DEFAULT_SCRIPT_TIMEOUT`, 3600s). - Catch-up window = half the period, clamped to 120s–2h; 120s grace for missed one-shots. +- **A pinned `timezone` owns the whole path, not just the compute.** `compute_next_run` resolves + *job pin → profile zone*, and every `next_run_at` write site must pass the pin (`cron/jobs.py` + call sites, `quota_hold.plan_hold`, `unreachable_retry.will_retry`); one missed site is a burst + re-run or a skipped slot. `_evaluate_due_job` normalizes the stored instant into the JOB's zone + and **skips `_repair_timezone_shifted_cron`** for pinned jobs — that repair is for jobs that + follow the profile zone, and on a pinned job its wall-clock predicate reads a due occurrence as + future and re-anchors past it (job silently never fires). The knob lives in + `cron/jobs_time.py`; `timezone` must stay in `job_definition.JOB_DEFINITION_FIELDS` or a + profile distribution strips it. - Every recurring occurrence is accounted for: `tick()` advances `next_run_at` BEFORE dispatch (at-most-once across a mid-run crash) and stamps `pending_slot` in the same save; a scan that finds the stamp with a dead owner restores the instant ONCE (`cron/occurrences.py`), the diff --git a/cron/job_definition.py b/cron/job_definition.py index ce05ccb5ac..6e1f89c757 100644 --- a/cron/job_definition.py +++ b/cron/job_definition.py @@ -15,6 +15,7 @@ JOB_DEFINITION_FIELDS = frozenset({ "script", "no_agent", "monitor_script", "monitor_url", "context_from", "schedule", "schedule_display", "deliver", "origin", "enabled_toolsets", "workdir", "attach_to_session", "reasoning_effort", "failure_deliver", + "timezone", "grace_minutes", }) diff --git a/cron/jobs.py b/cron/jobs.py index d650cee1a9..7379b764b6 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -26,14 +26,17 @@ try: import msvcrt except ImportError: # pragma: no cover - non-Windows msvcrt = None -from datetime import datetime, timedelta, timezone, UTC +# ``datetime.timezone`` is deliberately NOT imported: the per-job ``timezone`` parameter +# (compute_next_run/create_job) would shadow it, and everything here uses ``UTC`` instead. +from datetime import datetime, timedelta, UTC from pathlib import Path from agent.deadline import poll_until from hermes_constants import get_hermes_home from cron.constants import CLAIM_TTL_INACTIVITY_HEADROOM, FIRE_CLAIM_SKEW_SECONDS, FIRE_CLAIM_TTL_SECONDS from cron.env_settings import cron_env_setting -from cron.jobs_invariants import _validate_job_mode_invariants +from cron.jobs_invariants import _validate_job_mode_invariants, validate_schedule_knobs from cron.jobs_model_pin import _main_model_pin +from cron import jobs_time from cron.scheduler_ownership import _claim_owner_is_dead from cron import extra_fire, store_health from typing import Optional, Dict, List, Any, Callable, Set, Tuple, Union, Collection @@ -857,11 +860,15 @@ def parse_schedule(schedule: str) -> dict[str, Any]: ) -def _ensure_aware(dt: datetime) -> datetime: - """Aware datetime in the configured Hermes timezone. Legacy naive values are read as - *system-local* wall time (what created them) then converted, preserving ordering across - timezone changes and avoiding false not-due results.""" - target_tz = _hermes_now().tzinfo +def _ensure_aware(dt: datetime, target_tz: Optional[Any] = None) -> datetime: + """Aware datetime in *target_tz*, defaulting to the configured Hermes timezone. Legacy naive + values are read as *system-local* wall time (what created them) then converted, preserving + ordering across timezone changes and avoiding false not-due results. + + The due scan passes the JOB's pinned zone here: every downstream cron predicate compares wall + clocks, so normalizing a pinned job's instant into the profile zone makes a due occurrence + look still-in-the-future and the repair path re-anchors past it (job lost, never fired).""" + target_tz = target_tz or _hermes_now().tzinfo if dt.tzinfo is None: return dt.replace(tzinfo=datetime.now().astimezone().tzinfo).astimezone(target_tz) return dt.astimezone(target_tz) @@ -893,11 +900,11 @@ def _seconds_after(dt: datetime, seconds: float) -> datetime: return (dt.astimezone(UTC) + timedelta(seconds=seconds)).astimezone(dt.tzinfo) -def _parse_aware(value: Any) -> Optional[datetime]: - """``_ensure_aware(datetime.fromisoformat(value))``, or None when *value* is not a parseable ISO - string.""" +def _parse_aware(value: Any, zone: Optional[Any] = None) -> Optional[datetime]: + """``_ensure_aware(datetime.fromisoformat(value), zone)``, or None when *value* is not a + parseable ISO string.""" try: - return _ensure_aware(datetime.fromisoformat(value)) + return _ensure_aware(datetime.fromisoformat(value), zone) except Exception: return None @@ -919,13 +926,15 @@ def _stored_wall_clock_is_future(stored: datetime, current: datetime) -> bool: def _recoverable_oneshot_run_at( schedule: dict[str, Any], now: datetime, *, last_run_at: Optional[str] = None, + zone: Optional[Any] = None, ) -> Optional[str]: """One-shot run time if still eligible: a small grace window covers jobs created just after - their minute; once run, a one-shot is never eligible again.""" + their minute; once run, a one-shot is never eligible again. *zone* is the job's pinned zone + (an offset-less ``run_at`` is that zone's wall clock).""" if not isinstance(schedule, dict) or schedule.get("kind") != "once" or last_run_at: return None run_at = schedule.get("run_at") - run_at_dt = _parse_aware(run_at) if run_at else None + run_at_dt = jobs_time.oneshot_instant(run_at, zone) if run_at else None if run_at_dt is not None and _elapsed_seconds(now, run_at_dt) <= ONESHOT_GRACE_SECONDS: return run_at return None @@ -1172,14 +1181,22 @@ def get_timezone_migration_catchup_stats() -> dict[str, Any]: } -def compute_next_run(schedule: dict[str, Any], last_run_at: Optional[str] = None) -> Optional[str]: - """Compute the next run time for a schedule as an ISO string, or None if no more runs.""" +def compute_next_run( + schedule: dict[str, Any], last_run_at: Optional[str] = None, *, timezone: Any = None, +) -> Optional[str]: + """Compute the next run time for a schedule as an ISO string, or None if no more runs. + + *timezone* is the job's pinned IANA zone (a job record or a bare zone name); None inherits + the profile's configured zone. Every caller that advances a job's ``next_run_at`` must pass + it, or a pinned job is computed in the wrong zone and jumps a slot.""" now = _hermes_now() if not isinstance(schedule, dict): return None kind = schedule.get("kind") + job_zone = jobs_time.zone_for_job(timezone) if kind == "once": - return _recoverable_oneshot_run_at(schedule, now, last_run_at=last_run_at) + return _recoverable_oneshot_run_at( + schedule, now, last_run_at=last_run_at, zone=job_zone) # Recurring kinds anchor on last_run_at so a restart doesn't re-anchor the schedule. base_time = (_parse_aware(last_run_at) if last_run_at else None) or now if kind == "interval": @@ -1212,7 +1229,7 @@ def compute_next_run(schedule: dict[str, Any], last_run_at: Optional[str] = None # the wall-clock hour stays correct every calendar day, including DST # boundaries (morning-routine 09:00 America/Toronto). # Fall back to the base's own zone only when nothing is configured. - zone = get_timezone() or base_time.tzinfo + zone = job_zone or get_timezone() or base_time.tzinfo base_wall = base_time.astimezone(zone).replace(tzinfo=None) it = croniter(expr, base_wall) # Strictly-after guard for the DST fall-back hour (qwen-code#11723 class): @@ -1733,6 +1750,7 @@ _CREATE_FIELD_NORMALIZERS: dict[str, Callable[[Any], Any]] = { "context_from": _normalize_context_from, "failure_deliver": _normalize_failure_deliver, "interpreter": _normalize_job_optional_text, + "timezone": jobs_time.normalize_timezone, } _UPDATE_FIELD_NORMALIZERS: dict[str, Callable[[Any], Any]] = { # [] is an explicit zero-tool allowlist and must survive the update path as [] too (#82010). @@ -1742,6 +1760,7 @@ _UPDATE_FIELD_NORMALIZERS: dict[str, Callable[[Any], Any]] = { "monitor_url": _normalize_job_optional_text, "interpreter": _normalize_job_optional_text, "reasoning_effort": _normalize_reasoning_effort, + "timezone": jobs_time.normalize_timezone, } @@ -1753,10 +1772,11 @@ def _oneshot_past_grace_error(run_at: Any) -> ValueError: def _next_run_or_reject_past_oneshot( parsed_schedule: dict[str, Any], label: str, fallback_run_at: Any, what: str, + timezone: Any = None, ) -> Optional[str]: """``compute_next_run`` that raises (after a warning log) for a one-shot outside the grace window, so a ghost job with ``next_run_at=None`` can never be stored.""" - next_run_at = compute_next_run(parsed_schedule) + next_run_at = compute_next_run(parsed_schedule, timezone=timezone) if parsed_schedule.get("kind") == "once" and next_run_at is None: run_at = parsed_schedule.get("run_at") or fallback_run_at logger.warning( @@ -1792,6 +1812,7 @@ def create_job( paused_reason: Optional[str] = None, pinned: bool = False, interpreter: Optional[str] = None, + timezone: Optional[str] = None, ) -> dict[str, Any]: """Create a new cron job and return the stored record. @@ -1802,7 +1823,8 @@ def create_job( source run FIRST each tick; unchanged output suppresses the agent run (mutually exclusive, incompatible with ``no_agent``). reasoning_effort: per-job pin; capability NOT validated. interpreter: absolute/``~`` Python for ``.py`` script/monitor_script, validated at run time - (a venv can be rebuilt or moved after creation).""" + (a venv can be rebuilt or moved after creation). timezone: IANA key pinning the schedule's + wall clock (None/'' = inherit the profile zone).""" if not isinstance(paused, bool): raise ValueError("paused must be a boolean.") if paused_reason is not None and not isinstance(paused_reason, str): @@ -1828,6 +1850,7 @@ def create_job( normalized_reasoning_effort = _normalize_reasoning_effort(reasoning_effort) _validate_job_mode_invariants(f["monitor_script"], f["monitor_url"], f["no_agent"], f["script"]) + validate_schedule_knobs(parsed_schedule, f["timezone"]) prompt_text = _coerce_job_text(prompt).strip() if not prompt_text and not f["script"] and not normalized_skills: raise ValueError(EMPTY_PAYLOAD_ERROR) @@ -1844,7 +1867,8 @@ def create_job( name = name or label_source[:50].strip() if pinned and not f["model"]: f["provider"], f["model"] = _main_model_pin() - next_run_at = _next_run_or_reject_past_oneshot(parsed_schedule, name, schedule, "") + next_run_at = _next_run_or_reject_past_oneshot( + parsed_schedule, name, schedule, "", f["timezone"]) job = { "id": job_id, @@ -1890,6 +1914,7 @@ def create_job( for key, value in ( ("attach_to_session", normalized_attach), ("reasoning_effort", normalized_reasoning_effort), ("failure_deliver", f["failure_deliver"]), ("interpreter", f["interpreter"]), + ("timezone", f["timezone"]), ): if value is not None: job[key] = value @@ -2043,7 +2068,8 @@ def _apply_schedule_update(updated: dict[str, Any], updates: dict[str, Any], job "schedule_display", updated_schedule.get("display", updated.get("schedule_display"))) if updated.get("state") != "paused": updated["next_run_at"] = _next_run_or_reject_past_oneshot( - updated_schedule, updated.get("name", job_id), updated_schedule, "update ") + updated_schedule, updated.get("name", job_id), updated_schedule, "update ", + updated.get("timezone")) def _fill_missing_next_run(updated: dict[str, Any]) -> None: @@ -2055,7 +2081,7 @@ def _fill_missing_next_run(updated: dict[str, Any]) -> None: or updated.get("next_run_at") ): return - next_run = compute_next_run(updated["schedule"]) + next_run = compute_next_run(updated["schedule"], timezone=updated.get("timezone")) if next_run is None and updated["schedule"].get("kind") == "once": run_at = updated["schedule"].get("run_at", "unknown") raise ValueError( @@ -2086,6 +2112,9 @@ def update_job(job_id: str, updates: dict[str, Any]) -> Optional[dict[str, Any]] _normalize_job_optional_text(updated.get("script"))) if any(k in updates for k in _PAYLOAD_FIELDS) and job_payload_is_empty(updated): raise ValueError(EMPTY_PAYLOAD_ERROR) + if "timezone" in updates: + validate_schedule_knobs(updated.get("schedule") or {}, updated.get("timezone")) + jobs_time.prune_inherited_timezone(updated) if "schedule" in updates: _apply_schedule_update(updated, updates, job_id) # next_run_at now follows the new schedule; a stale quota_hold_until would only shield @@ -2093,7 +2122,15 @@ def update_job(job_id: str, updates: dict[str, Any]) -> Optional[dict[str, Any]] # parked. The next fire re-parks (with a fresh notice) if the window is still closed. from cron.quota_hold import clear_state as _clear_quota_hold _clear_quota_hold(updated) - if {"schedule", "next_run_at", "enabled", "state"}.intersection(updates): + elif "timezone" in updates and updated.get("state") != "paused": + # The instant on file was computed in the OLD zone; leaving it would make the new + # zone fire it at the wrong wall clock (or skip it). Only overwrite when the new + # zone can compute one (a past one-shot keeps its recorded instant). + recomputed = compute_next_run( + updated["schedule"], timezone=updated.get("timezone")) + if recomputed is not None: + updated["next_run_at"] = recomputed + if {"schedule", "next_run_at", "enabled", "state", "timezone"}.intersection(updates): # An explicit schedule/lifecycle rewrite supersedes any occurrence the dispatcher # left unclaimed — pause/resume/edit must not resurrect a slot from before the edit. updated.pop("pending_slot", None) @@ -2144,7 +2181,7 @@ def resume_job(job_id: str) -> Optional[dict[str, Any]]: "tick fires it (late/catch-up) or logs the skip.", job.get("name", job["id"]), stored_next) else: - next_run_at = compute_next_run(job["schedule"]) + next_run_at = compute_next_run(job["schedule"], timezone=job.get("timezone")) if next_run_at is None and job["schedule"].get("kind") == "once": run_at = job["schedule"].get("run_at", "unknown") raise ValueError( @@ -2214,7 +2251,7 @@ def rearm_oneshot(job_id: str, run_at: Any) -> Optional[dict[str, Any]]: parsed_schedule = parse_schedule(str(run_at)) if parsed_schedule.get("kind") != "once": raise ValueError(_REARM_RECURRING_ERROR) - next_run_at = compute_next_run(parsed_schedule) + next_run_at = compute_next_run(parsed_schedule, timezone=job_ref.get("timezone")) if next_run_at is None: raise _oneshot_past_grace_error(parsed_schedule.get("run_at") or run_at) @@ -2385,7 +2422,8 @@ def _advance_after_run(job: dict[str, Any], now: str, *, ladder_rung: bool = Fal _complete_job_record(job) return - job["next_run_at"] = compute_next_run(job["schedule"], now) + job["next_run_at"] = compute_next_run( + job["schedule"], now, timezone=job.get("timezone")) if job["next_run_at"] is not None: if job.get("state") != "paused": job["state"] = "scheduled" @@ -2612,7 +2650,8 @@ def advance_next_runs(job_ids) -> int: or extra_fire.queued(job) # the claim restores its slot; advancing erases the intent ): continue - new_next = compute_next_run(job["schedule"], now) + new_next = compute_next_run( + job["schedule"], now, timezone=job.get("timezone")) if new_next and new_next != job.get("next_run_at"): job["next_run_at"] = new_next advanced += 1 @@ -2690,7 +2729,8 @@ def claim_job_for_fire( instant = None if instant and completed_occurrence(job, instant): if job.get("schedule", {}).get("kind") in {"cron", "interval"}: - nxt = compute_next_run(job["schedule"], now.isoformat()) + nxt = compute_next_run( + job["schedule"], now.isoformat(), timezone=job.get("timezone")) if nxt: job["next_run_at"] = nxt save_jobs(jobs) @@ -2705,7 +2745,8 @@ def claim_job_for_fire( if slot := extra_fire.pending_slot(job, now, manual_fire): extra_fire.stamp_claim(job, slot) # off-schedule: the pending occurrence stays elif job.get("schedule", {}).get("kind") in {"cron", "interval"}: - nxt = compute_next_run(job["schedule"], now.isoformat()) + nxt = compute_next_run( + job["schedule"], now.isoformat(), timezone=job.get("timezone")) if nxt: job["next_run_at"] = nxt save_jobs(jobs) @@ -2877,11 +2918,13 @@ def _recover_missing_next_run(job: dict[str, Any], scan: _DueScan) -> Optional[s to 'scheduled', otherwise they would be silently skipped forever.""" schedule = job.get("schedule", {}) kind = schedule.get("kind") + zone = jobs_time.zone_for_job(job) recovered_next = _recoverable_oneshot_run_at( - schedule, scan.now, last_run_at=job.get("last_run_at")) + schedule, scan.now, last_run_at=job.get("last_run_at"), zone=zone) recovery_kind = "one-shot" if recovered_next else None if not recovered_next and kind in {"cron", "interval"}: - recovered_next = compute_next_run(schedule, scan.now.isoformat()) + recovered_next = compute_next_run( + schedule, scan.now.isoformat(), timezone=job.get("timezone")) if recovered_next: recovery_kind = kind if not recovered_next: @@ -2906,7 +2949,7 @@ class _DueJob: scan: _DueScan next_run: str # stored ISO string, compared string-exact against manual_run_at raw_next_run_dt: datetime # as stored (may carry a pre-migration offset) - next_run_dt: datetime # normalized to the configured tz + next_run_dt: datetime # normalized to the job's pinned tz, else the configured one @property def schedule(self) -> dict[str, Any]: @@ -2921,7 +2964,8 @@ class _DueJob: return self.job.get("name", self.job.get("id", "?")) def recompute_next(self) -> Optional[str]: - return compute_next_run(self.schedule, self.scan.now.isoformat()) + return compute_next_run( + self.schedule, self.scan.now.isoformat(), timezone=self.job.get("timezone")) def _repair_timezone_shifted_cron(d: _DueJob) -> bool: @@ -3165,8 +3209,12 @@ def _evaluate_due_job(job: dict[str, Any], scan: _DueScan, run_claim_ttl: float) next_run = _restore_unclaimed_slot(job, scan) or job.get("next_run_at") or _recover_missing_next_run(job, scan) if not next_run: return False + # A pinned job's stored instant must be normalized into ITS zone: the cron repair and + # stale-lattice predicates below compare wall clocks, so the profile zone would make a due + # occurrence look future-dated and re-anchor past it (the job then never fires). + zone = jobs_time.zone_for_job(job) raw_next_run_dt = datetime.fromisoformat(next_run) - d = _DueJob(job, scan, next_run, raw_next_run_dt, _ensure_aware(raw_next_run_dt)) + d = _DueJob(job, scan, next_run, raw_next_run_dt, _ensure_aware(raw_next_run_dt, zone)) kind = d.kind recurring = kind in {"cron", "interval"} # Intentionally string-exact on raw stored values: trigger_job stamps the SAME isoformat string @@ -3180,7 +3228,10 @@ def _evaluate_due_job(job: dict[str, Any], scan: _DueScan, run_claim_ttl: float) if new_next: scan.persist(job["id"], next_run_at=new_next) return False - if kind == "cron" and not manual_run and _repair_timezone_shifted_cron(d): + # The offset-repair path exists to rescue jobs that FOLLOW the profile zone across a zone + # change. A pinned job's instant legitimately carries its own offset, so repair can only + # misfire on it — skip it entirely. + if kind == "cron" and not manual_run and zone is None and _repair_timezone_shifted_cron(d): return False d.next_run_dt = _rearm_stale_error_recurring(d) if _instant_after(d.next_run_dt, now): diff --git a/cron/jobs_invariants.py b/cron/jobs_invariants.py index 2936408d5b..7b82266425 100644 --- a/cron/jobs_invariants.py +++ b/cron/jobs_invariants.py @@ -42,6 +42,17 @@ def _validate_job_mode_invariants( scripts_dir=display_hermes_home() + "/scripts/", script=script)) +def validate_schedule_knobs(schedule: dict, timezone: Optional[str]) -> None: + """Per-job schedule knobs are validated at BOTH doors — ``create_job`` and ``update_job`` — so + an update cannot store a value a create would reject. + + Raises ValueError for an unknown IANA zone. + """ + from cron.jobs_time import normalize_timezone + + normalize_timezone(timezone) + + def _script_is_command_line(script: str) -> bool: """A ``script`` with whitespace that resolves to no file is a command line (``echo hi``), not a path; run every tick it would deliver "Script not found" as the payload. A plain diff --git a/cron/jobs_time.py b/cron/jobs_time.py new file mode 100644 index 0000000000..18293384f5 --- /dev/null +++ b/cron/jobs_time.py @@ -0,0 +1,116 @@ +"""Per-job ``timezone`` knob (sibling of ``cron/jobs.py``). + +A job may pin its own IANA ``timezone``: its cron expression is then that zone's WALL CLOCK, +independent of the profile's configured zone. Absent/None means "inherit the profile", so a +store written before this field existed keeps byte-identical behaviour. + +Behaviour lives here, not in the facade: ``cron/jobs.py`` is already past the file-lines +ratchet. The facade keeps the call sites (``compute_next_run``, the due scan, and every write +site that advances ``next_run_at``), which read the pins through this module. + +Why the due scan needs the pin too: ``_ensure_aware`` normalizes a stored instant into a zone, +and the cron repair predicates then compare WALL CLOCKS. Normalizing a New-York job's instant +into a Los-Angeles profile zone makes the wall clock look like it is still in the future, so +``_repair_timezone_shifted_cron`` re-anchors to the NEXT occurrence and the due one is lost — +the "job timezone east of the profile timezone silently never fires" failure. See +``cron.jobs._evaluate_due_job``. +""" + +from __future__ import annotations + +import logging +from datetime import datetime +from typing import Any, Optional +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +logger = logging.getLogger(__name__) + +TIMEZONE_KEY = "timezone" + +# A corrupt persisted value would otherwise log on every 60s scan. Warn once per distinct value. +_warned: set[str] = set() + + +def reset_warning_state() -> None: + """Forget which corrupt values were already warned about (tests, config reloads).""" + _warned.clear() + + +def _warn_once(kind: str, value: Any, detail: str) -> None: + key = f"{kind}:{value!r}" + if key in _warned: + return + _warned.add(key) + logger.warning( + "Job %s %r is invalid (%s); falling back to the profile setting. Fix or clear it with " + "`hermes cron edit`.", kind, value, detail) + + +def normalize_timezone(value: Any) -> Optional[str]: + """Validate a job's ``timezone`` and return the canonical IANA key. + + ``None``/``""`` mean "inherit the profile timezone", stored as an ABSENT key. Anything that + is not a resolvable IANA zone raises ValueError BEFORE the record is written, so a typo can + never reach the tick loop. + """ + if value is None: + return None + if not isinstance(value, str): + raise ValueError( + "timezone must be an IANA name such as 'Asia/Hong_Kong', " + f"not {type(value).__name__}.") + name = value.strip() + if not name: + return None + try: + return ZoneInfo(name).key + except (ZoneInfoNotFoundError, ValueError) as exc: + raise ValueError( + f"Unknown timezone {name!r}. Use an IANA key from " + f"`zoneinfo.available_timezones()`, e.g. 'Asia/Hong_Kong'.") from exc + + +def zone_for_job(job_or_timezone: Any) -> Optional[ZoneInfo]: + """The zone a job's schedule is expressed in, or None to inherit the profile's. + + Accepts a job record or a bare ``timezone`` value. Never raises: a corrupt stored value + degrades to the profile zone (warned once per value) instead of breaking the tick it is + read in — the 60s scan must not die on one bad record. + """ + if isinstance(job_or_timezone, dict): + value = job_or_timezone.get(TIMEZONE_KEY) + else: + value = job_or_timezone + if not isinstance(value, str) or not value.strip(): + return None + try: + return ZoneInfo(value.strip()) + except (ZoneInfoNotFoundError, ValueError) as exc: + _warn_once(TIMEZONE_KEY, value, str(exc)) + return None + + +def prune_inherited_timezone(record: dict[str, Any]) -> None: + """``timezone: ''`` / null means "inherit the profile", so the key is DROPPED, not stored as + null — an absent optional key is this repo's spelling for "unset" (see ``cron/jobs.py``'s + optional-field write loop) and a stored null would render as a pinned-but-blank zone.""" + if TIMEZONE_KEY in record and not record.get(TIMEZONE_KEY): + record.pop(TIMEZONE_KEY, None) + + +def oneshot_instant(value: Any, zone: Optional[ZoneInfo] = None) -> Optional[datetime]: + """Parse a one-shot ``run_at`` into an aware instant, or None when unparseable. + + An offset-bearing value is already an instant. An offset-LESS value is the job's own wall + clock, so a pinned zone interprets it in that zone (``2026-06-01T09:00:00`` means 09:00 + for THIS job) instead of in the host's local time. + """ + from cron.jobs import _ensure_aware # late import: the facade owns naive normalization + + try: + dt = datetime.fromisoformat(str(value)) + except (TypeError, ValueError): + return None + if dt.tzinfo is None and zone is not None: + return dt.replace(tzinfo=zone) + return _ensure_aware(dt, zone) diff --git a/cron/quota_hold.py b/cron/quota_hold.py index 4b1ddaaa47..bbd0ef0433 100644 --- a/cron/quota_hold.py +++ b/cron/quota_hold.py @@ -135,7 +135,9 @@ def plan_hold( if kind == "cron" and blocked and not extra_fire.queued(job): # Coalesce cron occurrences inside the closed window to the first legal instant after it. # A queued run-now is not an occurrence: it waits for the boundary, not the next slot. - parked = compute_next_run(schedule, window_end.isoformat()) or window_end.isoformat() + parked = compute_next_run( + schedule, window_end.isoformat(), timezone=job.get("timezone") + ) or window_end.isoformat() else: parked = window_end.isoformat() if recover: diff --git a/cron/unreachable_retry.py b/cron/unreachable_retry.py index a576ba5fe7..92994cfc58 100644 --- a/cron/unreachable_retry.py +++ b/cron/unreachable_retry.py @@ -99,7 +99,8 @@ def will_retry(job: dict[str, Any]) -> bool: from cron.jobs import _parse_aware, compute_next_run now = _hermes_now() - natural_next = _parse_aware(compute_next_run(job.get("schedule") or {}, now.isoformat())) + natural_next = _parse_aware(compute_next_run( + job.get("schedule") or {}, now.isoformat(), timezone=job.get("timezone"))) if natural_next is None: # The natural occurrence is uncomputable (e.g. croniter missing): _advance_after_run # leaves the record state=error — terminal — so plan_retry never runs either. diff --git a/hermes_cli/cron.py b/hermes_cli/cron.py index fcb794c925..ff2365b8f7 100644 --- a/hermes_cli/cron.py +++ b/hermes_cli/cron.py @@ -246,6 +246,7 @@ def _job_rows(job: dict[str, Any]) -> list[tuple[str, str]]: ("Changed", mon_state.get("last_changed_at") if monitor_source else ""), ("Mode", color("no-agent", Colors.DIM) + " (script stdout delivered directly)" if job.get("no_agent") else ""), + ("Timezone", job.get("timezone")), ("Workdir", job.get("workdir")), ("Python", job.get("interpreter")), ("Last run", f"{job.get('last_run_at', '?')} {_last_run_display(job)}" @@ -732,6 +733,11 @@ def _cron_doctor_issues_for_job(job: dict[str, Any]) -> list[str]: workdir = str(job.get("workdir") or "").strip() if workdir and not Path(workdir).expanduser().exists(): issues.append(f"workdir not found: {workdir}") + if job.get("timezone"): + from cron.jobs_time import zone_for_job + if zone_for_job(job) is None: + issues.append(f"timezone {job['timezone']!r} is not a valid IANA zone; " + "the job falls back to the profile timezone") return issues @@ -762,7 +768,7 @@ _JOB_ARG_FIELDS = (("name", "name"), ("deliver", "deliver"), ("failure_deliver", ("model", "model"), ("provider", "model_provider"), ("pinned", "pinned"), ("monitor_script", "monitor_script"), ("monitor_url", "monitor_url"), ("continuity", "continuity"), ("reasoning_effort", "reasoning_effort"), - ("interpreter", "interpreter")) + ("interpreter", "interpreter"), ("timezone", "timezone")) def _job_api_kwargs(args) -> dict[str, Any]: @@ -776,6 +782,7 @@ _JOB_DETAIL_LINES = ( ("monitor_url", " Monitor: {} (agent runs only on output change)"), ("no_agent", " Mode: no-agent (script stdout delivered directly)"), ("continuity", " Continuity: on (each run sees the previous run's output)"), + ("timezone", " Timezone: {} (schedule runs on this zone's wall clock)"), ("workdir", " Workdir: {}"), ("interpreter", " Python: {}")) diff --git a/hermes_cli/subcommands/cron.py b/hermes_cli/subcommands/cron.py index ae6115293a..c585f5c36b 100644 --- a/hermes_cli/subcommands/cron.py +++ b/hermes_cli/subcommands/cron.py @@ -81,6 +81,10 @@ def build_cron_parser(subparsers, *, cmd_cron: Callable) -> None: help="Absolute or ~ path to a Python in your own venv (e.g. ~/venvs/report/bin/python) " "for a .py --script / --monitor-script, so it can import packages Hermes does not " "ship. .sh/.bash still run under bash. Omit to use Hermes' Python.") + cron_create.add_argument("--timezone", + help="IANA zone for this job's schedule (e.g. Asia/Hong_Kong), so the cron " + "expression is read as THAT zone's wall clock whatever the profile's timezone " + "is. Omit to inherit the profile timezone.") cron_create.add_argument( "--continuity", dest="continuity", action="store_const", const=True, default=None, help="Each run wakes up with the job's own previous output injected " @@ -151,6 +155,9 @@ def build_cron_parser(subparsers, *, cmd_cron: Callable) -> None: cron_edit.add_argument("--interpreter", help="Absolute or ~ path to a Python for a .py script / monitor script. " "Pass empty string to clear (back to Hermes' Python).") + cron_edit.add_argument("--timezone", + help="IANA zone for this job's schedule. Pass empty string to clear the pin " + "(back to the profile timezone); next_run_at is recomputed in the new zone.") # lifecycle actions cron_pause = cron_subparsers.add_parser("pause", help="Pause a scheduled job") diff --git a/tests/cron/test_catchup_policy.py b/tests/cron/test_catchup_policy.py index 7f78e1186d..07b2a7907c 100644 --- a/tests/cron/test_catchup_policy.py +++ b/tests/cron/test_catchup_policy.py @@ -37,5 +37,5 @@ def test_default_and_uncomputable_still_catch_up(tmp_path, monkeypatch, uncomput stored[0]["next_run_at"] = (jobs._hermes_now() - timedelta(hours=4)).isoformat() jobs.save_jobs(stored) if uncomputable: - monkeypatch.setattr(jobs, "compute_next_run", lambda *args: None) + monkeypatch.setattr(jobs, "compute_next_run", lambda *args, **kwargs: None) assert [row["id"] for row in jobs.get_due_jobs()] == [job["id"]] diff --git a/tests/cron/test_jobs.py b/tests/cron/test_jobs.py index ef553cb336..7b71e942ef 100644 --- a/tests/cron/test_jobs.py +++ b/tests/cron/test_jobs.py @@ -1375,7 +1375,7 @@ class TestPerJobScanContainment: import cron.jobs as jobs_mod real_grace = jobs_mod._compute_grace_seconds - def selective_grace(schedule): + def selective_grace(schedule, grace_minutes=None): if schedule.get("minutes") == 7: raise RuntimeError("simulated unforeseen malformed field") return real_grace(schedule) diff --git a/tests/cron/test_jobs_time.py b/tests/cron/test_jobs_time.py new file mode 100644 index 0000000000..150cef97d7 --- /dev/null +++ b/tests/cron/test_jobs_time.py @@ -0,0 +1,90 @@ +"""Unit contract for the per-job ``timezone`` knob: ``cron/jobs_time.py``. + +An IANA ``timezone`` pins a job's cron wall clock. The load-bearing case is the tolerant read: a +corrupt persisted zone must degrade to the profile zone, warned ONCE, never break the 60s tick. +""" + +from datetime import datetime +from zoneinfo import ZoneInfo + +import pytest + +from cron import jobs_time + +NY = ZoneInfo("America/New_York") + + +class TestNormalizeTimezone: + def test_iana_key_is_stored_canonical(self): + assert jobs_time.normalize_timezone("America/New_York") == "America/New_York" + + def test_surrounding_whitespace_is_trimmed(self): + assert jobs_time.normalize_timezone(" Asia/Hong_Kong ") == "Asia/Hong_Kong" + + @pytest.mark.parametrize("value", [None, "", " "]) + def test_empty_means_inherit_the_profile(self, value): + assert jobs_time.normalize_timezone(value) is None + + @pytest.mark.parametrize("value", ["Mars/Olympus_Mons", "not a zone", "/etc/passwd"]) + def test_unknown_zone_is_rejected_before_storage(self, value): + with pytest.raises(ValueError, match="timezone"): + jobs_time.normalize_timezone(value) + + def test_non_string_is_rejected(self): + with pytest.raises(ValueError, match="IANA name"): + jobs_time.normalize_timezone(9) + + +class TestZoneForJob: + def test_absent_or_null_inherits_the_profile(self): + assert jobs_time.zone_for_job({}) is None + assert jobs_time.zone_for_job({"timezone": None}) is None + assert jobs_time.zone_for_job({"timezone": ""}) is None + + def test_pinned_zone_is_returned(self): + assert jobs_time.zone_for_job({"timezone": "America/New_York"}) == NY + + def test_accepts_a_bare_zone_name(self): + assert jobs_time.zone_for_job("America/New_York") == NY + + def test_corrupt_value_degrades_instead_of_raising(self, monkeypatch): + """One bad record must never abort the tick it is scanned in.""" + jobs_time.reset_warning_state() + warnings: list[str] = [] + monkeypatch.setattr( + jobs_time.logger, "warning", + lambda msg, *a, **kw: warnings.append(msg % a if a else msg)) + job = {"timezone": "Mars/Olympus_Mons"} + + assert jobs_time.zone_for_job(job) is None + assert jobs_time.zone_for_job(job) is None # scanned again a minute later + assert jobs_time.zone_for_job({"timezone": "Mars/Olympus_Mons"}) is None + + assert len(warnings) == 1, f"a bad zone must warn ONCE, got {warnings}" + assert "Mars/Olympus_Mons" in warnings[0] + + +class TestPruneInheritedTimezone: + def test_cleared_zone_drops_the_key(self): + record = {"id": "a", "timezone": ""} + jobs_time.prune_inherited_timezone(record) + assert "timezone" not in record + + def test_pinned_zone_is_kept(self): + record = {"timezone": "Asia/Hong_Kong"} + jobs_time.prune_inherited_timezone(record) + assert record["timezone"] == "Asia/Hong_Kong" + + +class TestOneshotInstant: + def test_offsetless_run_at_is_the_jobs_wall_clock(self): + """``09:00`` for a New-York job is 09:00 in New York, not on the host.""" + instant = jobs_time.oneshot_instant("2026-06-01T09:00:00", NY) + assert instant == datetime(2026, 6, 1, 9, 0, tzinfo=NY) + + def test_offset_bearing_run_at_is_already_an_instant(self): + instant = jobs_time.oneshot_instant("2026-06-01T09:00:00+00:00", NY) + assert instant.astimezone(NY) == datetime(2026, 6, 1, 5, 0, tzinfo=NY) + + def test_garbage_is_none(self): + assert jobs_time.oneshot_instant("not-a-time", NY) is None diff --git a/tests/cron/test_per_job_timezone.py b/tests/cron/test_per_job_timezone.py new file mode 100644 index 0000000000..a7d878c5c8 --- /dev/null +++ b/tests/cron/test_per_job_timezone.py @@ -0,0 +1,202 @@ +"""Per-job ``timezone``: one store, several zones, each job on its own wall clock. + +The failing shape this pins (the reason a compute-only fix is not enough — upstream PR #137231 +fixed ``compute_next_run`` but not the dispatch path): with the profile pinned to +``America/Los_Angeles`` and a job pinned to ``America/New_York`` running ``0 9 * * 1-5``, the +due scan normalized the stored New-York instant into the profile zone, so the repair predicate +``_stored_wall_clock_is_future`` compared wall clocks across zones, decided the run was still +in the future and re-anchored to the NEXT day. The due occurrence was silently dropped — the +symptom is "no error, no fire, the next_run_at just moved". + +Everything here goes through the public store API (``create_job`` / ``get_due_jobs``), never by +calling the internal predicates, so the tests fail on the real symptom rather than on a +hand-built intermediate value. +""" + +import logging +from datetime import datetime +from zoneinfo import ZoneInfo + +import pytest + +pytest.importorskip("croniter") + +import hermes_time +from cron import jobs + +LA = ZoneInfo("America/Los_Angeles") +NY = ZoneInfo("America/New_York") +HK = ZoneInfo("Asia/Hong_Kong") + +WEEKDAY_9AM = "0 9 * * 1-5" +DAILY_9AM = "0 9 * * *" + + +@pytest.fixture +def la_profile(monkeypatch): + """Configure the profile zone the way production does: ``HERMES_TIMEZONE`` + cache reset.""" + monkeypatch.setenv("HERMES_TIMEZONE", "America/Los_Angeles") + hermes_time.reset_cache() + yield LA + hermes_time.reset_cache() + + +@pytest.fixture +def cron_store(tmp_path, monkeypatch): + """Isolated store + a frozen clock the test moves explicitly.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + class Clock: + instant = datetime(2026, 10, 22, 12, 0, tzinfo=NY) + + def set(self, value): + self.instant = value + + def now(self): + # Production shape: _hermes_now() is aware in the PROFILE zone. + return self.instant.astimezone(hermes_time.get_timezone() or LA) + + clock = Clock() + monkeypatch.setattr(jobs, "_hermes_now", clock.now) + with jobs.use_cron_store(tmp_path / "cron"): + yield clock + + +def _add_job(schedule, *, timezone=None, next_run_at): + """Create a real job through the public door, then park its instant where the test needs it.""" + job = jobs.create_job( + prompt="probe", schedule=schedule, model="fixture", deliver="local", + timezone=timezone) + stored = jobs.load_jobs() + row = next(r for r in stored if r["id"] == job["id"]) + row["next_run_at"] = next_run_at.isoformat() + jobs.save_jobs(stored) + return job["id"] + + +class TestPinnedTimezoneIsDue: + """§0 of the design: the pinned job must be DUE, and the repair path must not touch it.""" + + @pytest.mark.parametrize("job_zone", ["America/New_York", "Asia/Hong_Kong"]) + def test_job_east_of_the_profile_zone_fires_on_time( + self, la_profile, cron_store, job_zone, caplog, + ): + """East of the profile zone is the direction that used to be dropped every time.""" + nine_am = datetime(2026, 10, 22, 9, 0, tzinfo=ZoneInfo(job_zone)) + cron_store.set(nine_am) + job_id = _add_job(WEEKDAY_9AM, timezone=job_zone, next_run_at=nine_am) + caplog.set_level(logging.INFO, logger="cron.scheduler") + + due = [j["id"] for j in jobs.get_due_jobs()] + + assert due == [job_id], "the pinned occurrence is due and must be returned" + stored = next(r for r in jobs.load_jobs() if r["id"] == job_id) + assert stored["next_run_at"] == nine_am.isoformat(), ( + "a pinned job's instant must never be rewritten into the profile zone") + assert "Recomputing cron run" not in caplog.text, ( + "the offset repair only exists for jobs that FOLLOW the profile zone") + + def test_two_pinned_jobs_keep_their_own_nine_am(self, la_profile, cron_store): + """One store, one profile zone, two jobs: each fires at ITS 09:00, not the other's.""" + ny_job = _add_job( + WEEKDAY_9AM, timezone="America/New_York", + next_run_at=datetime(2026, 10, 22, 9, 0, tzinfo=NY)) + hk_job = _add_job( + WEEKDAY_9AM, timezone="Asia/Hong_Kong", + next_run_at=datetime(2026, 10, 23, 9, 0, tzinfo=HK)) + + # 09:00 New York == 21:00 Hong Kong == 06:00 Los Angeles: only the NY job is due. + cron_store.set(datetime(2026, 10, 22, 9, 0, tzinfo=NY)) + assert [j["id"] for j in jobs.get_due_jobs()] == [ny_job] + # The tick claims the fired job and advances it in ITS zone — the write site that would + # otherwise jump a slot (and the claim is what retires the pending_slot stamp). + assert jobs.claim_job_for_fire(ny_job) is True + + # A day later, 09:00 Hong Kong (== 18:00 the previous day in LA) is the HK job's turn. + cron_store.set(datetime(2026, 10, 23, 9, 0, tzinfo=HK)) + assert [j["id"] for j in jobs.get_due_jobs()] == [hk_job] + advanced = next(r for r in jobs.load_jobs() if r["id"] == ny_job) + assert datetime.fromisoformat(advanced["next_run_at"]).astimezone(NY).hour == 9 + + def test_unpinned_job_still_follows_the_profile_zone(self, la_profile, cron_store): + """The default is unchanged: absent pin == profile wall clock (no migration needed).""" + nine_la = datetime(2026, 10, 22, 9, 0, tzinfo=LA) + cron_store.set(nine_la) + job_id = _add_job(WEEKDAY_9AM, next_run_at=nine_la) + + assert [j["id"] for j in jobs.get_due_jobs()] == [job_id] + + +class TestComputeNextRunUsesTheJobZone: + def test_next_run_lands_on_the_job_wall_clock(self, la_profile): + schedule = jobs.parse_schedule(WEEKDAY_9AM) + base = datetime(2026, 10, 22, 10, 0, tzinfo=NY).isoformat() + + pinned = datetime.fromisoformat( + jobs.compute_next_run(schedule, base, timezone="America/New_York")) + inherited = datetime.fromisoformat(jobs.compute_next_run(schedule, base)) + + assert (pinned.hour, pinned.minute) == (9, 0) + assert pinned.astimezone(NY).date() == datetime(2026, 10, 23).date() + # Same instant computed in the profile zone would land at 09:00 LOS ANGELES. + assert inherited.astimezone(LA).hour == 9 + assert pinned != inherited + + def test_dst_transitions_hold_the_job_wall_clock(self, la_profile): + """Spring-forward (Mar 8 2026) and fall-back (Nov 1 2026) days stay at 09:00 job time.""" + schedule = jobs.parse_schedule(DAILY_9AM) + for base in ("2026-03-07T09:00:00-05:00", "2026-10-31T09:00:00-04:00"): + nxt = datetime.fromisoformat( + jobs.compute_next_run(schedule, base, timezone="America/New_York")) + assert (nxt.astimezone(NY).hour, nxt.astimezone(NY).minute) == (9, 0), base + + def test_fall_back_second_occurrence_is_strictly_later(self, la_profile): + """qwen-code#11723 class: no immediate re-fire inside the repeated autumn hour.""" + base = datetime(2026, 11, 1, 1, 15, tzinfo=NY, fold=1) # 01:15 EST (second 01:15) + nxt = datetime.fromisoformat( + jobs.compute_next_run({"kind": "cron", "expr": "30 1 * * *"}, + base.isoformat(), timezone="America/New_York")) + assert nxt.timestamp() > base.timestamp() + + +class TestInheritanceAndDegradation: + def test_null_and_absent_are_the_same_job(self, la_profile, cron_store): + explicit = jobs.create_job( + prompt="null", schedule=WEEKDAY_9AM, model="fixture", deliver="local", + timezone=None) + absent = jobs.create_job( + prompt="absent", schedule=WEEKDAY_9AM, model="fixture", deliver="local") + rows = {r["prompt"]: r for r in jobs.load_jobs()} + assert "timezone" not in rows["null"] and "timezone" not in rows["absent"] + assert explicit["next_run_at"] == absent["next_run_at"] + + def test_clearing_the_pin_through_update_drops_the_key(self, la_profile, cron_store): + job = jobs.create_job( + prompt="p", schedule=WEEKDAY_9AM, model="fixture", deliver="local", + timezone="Asia/Hong_Kong") + jobs.update_job(job["id"], {"timezone": ""}) + assert "timezone" not in jobs.load_jobs()[0] + + def test_changing_the_pin_recomputes_the_stored_instant(self, la_profile, cron_store): + job = jobs.create_job( + prompt="p", schedule=WEEKDAY_9AM, model="fixture", deliver="local", + timezone="America/New_York") + before = jobs.load_jobs()[0]["next_run_at"] + + jobs.update_job(job["id"], {"timezone": "Asia/Hong_Kong"}) + after = jobs.load_jobs()[0]["next_run_at"] + + assert after != before, "the instant belongs to the old zone; it must move" + assert datetime.fromisoformat(after).astimezone(HK).hour == 9 + + def test_a_corrupt_stored_zone_does_not_break_the_scan(self, la_profile, cron_store): + nine_la = datetime(2026, 10, 22, 9, 0, tzinfo=LA) + cron_store.set(nine_la) + job_id = _add_job(WEEKDAY_9AM, next_run_at=nine_la) + stored = jobs.load_jobs() + stored[0]["timezone"] = "Mars/Olympus_Mons" # hand-edited store + jobs.save_jobs(stored) + + due = [j["id"] for j in jobs.get_due_jobs()] + + assert due == [job_id], "a bad zone degrades to the profile zone, it does not hide the job" diff --git a/tests/hermes_cli/test_cron_parser_builder.py b/tests/hermes_cli/test_cron_parser_builder.py index f78844c86e..bb91888c9c 100644 --- a/tests/hermes_cli/test_cron_parser_builder.py +++ b/tests/hermes_cli/test_cron_parser_builder.py @@ -40,3 +40,15 @@ def test_cron_accept_hooks_flag_on_run_and_tick(): assert ns.accept_hooks is True ns2 = parser.parse_args(["cron", "tick", "--accept-hooks"]) assert ns2.accept_hooks is True + + +def test_cron_timezone_flags(): + parser = _build() + created = parser.parse_args( + ["cron", "create", "0 9 * * *", "hi", "--timezone", "Asia/Hong_Kong"]) + assert created.timezone == "Asia/Hong_Kong" + + edited = parser.parse_args(["cron", "edit", "j", "--timezone", ""]) + assert edited.timezone == "", "an empty value is the documented clear" + + assert parser.parse_args(["cron", "edit", "j"]).timezone is None diff --git a/tests/hermes_cli/test_web_server_cron_profiles.py b/tests/hermes_cli/test_web_server_cron_profiles.py index 0755835f38..a972a33031 100644 --- a/tests/hermes_cli/test_web_server_cron_profiles.py +++ b/tests/hermes_cli/test_web_server_cron_profiles.py @@ -367,7 +367,7 @@ def test_profile_call_cannot_retarget_ticker_store_mid_write( monkeypatch.setattr( cron_jobs, "compute_next_run", - lambda _schedule, _last_run_at=None: "2026-07-10T00:00:00+00:00", + lambda _schedule, _last_run_at=None, **_kw: "2026-07-10T00:00:00+00:00", ) ticker_loaded = threading.Event() diff --git a/tests/tools/test_cronjob_tools.py b/tests/tools/test_cronjob_tools.py index eefb67e804..f06fb0f458 100644 --- a/tests/tools/test_cronjob_tools.py +++ b/tests/tools/test_cronjob_tools.py @@ -802,3 +802,43 @@ class TestGithubExemptionAbuse: assert _scan_cron_prompt( "generate a keypair and explain id_rsa vs id_ed25519" ) == "" + + +class TestScheduleKnobToolSurface: + """The model-facing tool must expose the per-job timezone knob (§4).""" + + @pytest.fixture(autouse=True) + def _setup_cron_dir(self, tmp_path, monkeypatch): + monkeypatch.setattr("cron.jobs.CRON_DIR", tmp_path / "cron") + monkeypatch.setattr("cron.jobs.JOBS_FILE", tmp_path / "cron" / "jobs.json") + monkeypatch.setattr("cron.jobs.OUTPUT_DIR", tmp_path / "cron" / "output") + + def _stored(self): + from cron.jobs import load_jobs + return load_jobs()[0] + + def test_create_stores_timezone(self): + created = json.loads(cronjob( + action="create", prompt="HK open", schedule="0 9 * * 1-5", + timezone="Asia/Hong_Kong")) + assert created["success"] is True, created + job = self._stored() + assert job["timezone"] == "Asia/Hong_Kong" + assert job["next_run_at"].endswith("+08:00") or job["next_run_at"].endswith("+09:00") + + def test_create_rejects_an_unknown_zone(self): + created = json.loads(cronjob( + action="create", prompt="x", schedule="0 9 * * *", + timezone="Mars/Olympus_Mons")) + assert created["success"] is False + assert "timezone" in created["error"].lower() + + def test_update_clears_the_zone(self): + created = json.loads(cronjob( + action="create", prompt="x", schedule="0 9 * * *", + timezone="Asia/Hong_Kong")) + job_id = created["job_id"] + + updated = json.loads(cronjob(action="update", job_id=job_id, timezone="")) + assert updated["success"] is True, updated + assert "timezone" not in self._stored(), "'' clears the pin; the key is dropped" diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index 40bfa88e50..865ad5c576 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -730,6 +730,7 @@ def _action_create(a: dict[str, Any]) -> str: monitor_url=_normalize_optional_job_value(a["monitor_url"]), # CLI-only lane: absent from CRONJOB_SCHEMA and the model dispatch (models don't pick models). reasoning_effort=a["reasoning_effort"], interpreter=a["interpreter"], + timezone=_normalize_optional_job_value(a["timezone"]), pinned=bool(a["pinned"]), failure_deliver=_resolve_cron_context_deliver(_normalize_deliver_param(a["failure_deliver"])), **({"paused": a["paused"], "paused_reason": a["paused_reason"]} @@ -959,6 +960,9 @@ def _update_run_fields(job: dict[str, Any], a: dict[str, Any], updates: dict[str repeat_state = dict(job.get("repeat") or {}) repeat_state["times"] = normalize_repeat_value(a["repeat"]) updates["repeat"] = repeat_state + if a["timezone"] is not None: + # '' clears the pin (the job inherits the profile timezone); update_job validates the key. + updates["timezone"] = _normalize_optional_job_value(a["timezone"]) or "" if a["schedule"] is not None: parsed_schedule = parse_schedule(a["schedule"]) updates["schedule"] = parsed_schedule @@ -1048,7 +1052,8 @@ def cronjob( paused: bool = False, paused_reason: Optional[str] = None, pinned: Optional[bool] = None, - interpreter: Optional[str] = None) -> str: + interpreter: Optional[str] = None, + timezone: Optional[str] = None) -> str: """Unified cron job management tool.""" a = dict(locals()) del a["task_id"] # unused but kept for handler signature compatibility @@ -1170,6 +1175,10 @@ Jobs run in a fresh session with no current-chat context, so prompts must be sel "type": "string", "description": "Optional absolute existing path to run the job from: injects that directory's AGENTS.md/context files and anchors terminal/file tools there. On update, '' clears." }, + "timezone": { + "type": "string", + "description": "Optional IANA timezone for the job's schedule, e.g. 'Asia/Hong_Kong' or 'America/New_York': the cron expression is then read as THIS zone's wall clock, whatever the profile's own timezone is. Use it for a job that must fire at a fixed local time in another market. Omit to inherit the profile timezone. On update, '' clears the pin." + }, "attach_to_session": { "type": "boolean", "description": "True = the job's delivery is CONTINUABLE — the user can reply and the agent has the brief in context (threads on thread-capable platforms, mirrored into the DM elsewhere). Use for conversational recurring jobs (briefings); leave unset for fire-and-forget alerts. Scope: the job's own conversation only — the origin chat, the home-channel fallback when deliver='origin' captured no origin (script-created jobs), a user-written bare platform target (deliver='slack' — that platform's home channel), or the job's single explicit platform:chat target (this flag is the only way to attach an explicit target). Broadcast targets are never attached; no effect when deliver='local'." @@ -1202,7 +1211,7 @@ def check_cronjob_requirements() -> bool: _HANDLER_FORWARDED_ARGS = ( "job_id", "prompt", "schedule", "name", "repeat", "deliver", "failure_deliver", "skill", "skills", "reason", "script", "context_from", "continuity", "enabled_toolsets", "workdir", "no_agent", "attach_to_session", - "paused_reason", "pinned") + "paused_reason", "pinned", "timezone") def _cronjob_handler(args, **kw): diff --git a/website/docs/developer-guide/cron-internals.md b/website/docs/developer-guide/cron-internals.md index 70106b13f4..baa20bb3b9 100644 --- a/website/docs/developer-guide/cron-internals.md +++ b/website/docs/developer-guide/cron-internals.md @@ -13,6 +13,7 @@ The cron subsystem provides scheduled task execution — from simple one-shot de | File | Purpose | |------|---------| | `cron/jobs.py` | Job model, storage, atomic read/write to `jobs.json` | +| `cron/jobs_time.py` | Per-job `timezone` knob: validation, tolerant reads, due-scan zone | | `cron/scheduler.py` | Scheduler loop — due-job detection, execution, repeat tracking | | `tools/cronjob_tools.py` | Model-facing `cronjob_manage` tool registration and handler | | `gateway/run.py` | Gateway integration — cron ticking in the long-running loop | @@ -31,6 +32,28 @@ Four schedule formats are supported: The model-facing surface is a single `cronjob_manage` tool with action-style operations: `create`, `list`, `update`, `pause`, `resume`, `run`, `remove`. +## Per-job timezone + +`compute_next_run(schedule, last_run_at, *, timezone=...)` resolves the zone as *job pin → profile +`timezone` → base offset*, and **every** caller that advances `next_run_at` passes the job's pin +(the advance sites, the claim path, resume/re-arm, the recovery paths, and the quota/unreachable +planners). One missed site shows up as a burst re-run or a skipped slot, because the stored instant +and the expression would be interpreted in different zones. + +The dispatch path needs the pin just as much as the compute does. `_evaluate_due_job` builds its +`_DueJob` with `_ensure_aware(raw, zone_for_job(job))`, which gives every downstream predicate the +job's clock; a job with a pinned zone **skips `_repair_timezone_shifted_cron` entirely**. That +repair exists for jobs that FOLLOW the profile zone across a zone change: for a pinned job its +stored offset legitimately differs from the profile's, and the repair's wall-clock comparison +(`_stored_wall_clock_is_future`) then reads a due occurrence as still-in-the-future and re-anchors +to the next one — the job silently never fires. This is why a fix to `compute_next_run` alone is +not enough. + +`timezone` is a top-level record field (not inside `schedule`), validated by +`jobs_time.normalize_*` at both the create and update door. A corrupt persisted value degrades to +the profile zone with one warning per value — the 60s scan must not die on one bad record. No +migration is needed: the keys simply do not exist in older stores, and every reader uses `.get()`. + ## Job Storage Jobs are stored in `~/.hermes/cron/jobs.json` with atomic write semantics (write to temp file, then rename). Each job record contains: diff --git a/website/docs/user-guide/features/cron.md b/website/docs/user-guide/features/cron.md index d59dca2c3a..de38b778a8 100644 --- a/website/docs/user-guide/features/cron.md +++ b/website/docs/user-guide/features/cron.md @@ -210,6 +210,21 @@ When `workdir` is set: Each agent run binds its `workdir` to that run's unique task identity. Workdir jobs therefore use the normal parallel pool without mutating process-global terminal state or leaking paths between concurrent runs. Set `cron.max_parallel_jobs` if you want to limit total cron concurrency. ::: +## Per-job timezone + +By default a job's cron expression is read in your profile's timezone (`timezone:` in `config.yaml`). A job can pin its own zone instead, so one profile can serve several markets: + +```bash +hermes cron create "0 9 * * 1-5" --timezone Asia/Hong_Kong "HK market open brief" +hermes cron create "0 9 * * 1-5" --timezone America/New_York "US market open brief" +``` + +- The expression is the pinned zone's **wall clock**: `0 9 * * *` fires at 09:00 in that zone, and stays at 09:00 across that zone's DST changes. +- Omit it (or pass `--timezone ""` on edit) to inherit the profile timezone. A job without the field behaves exactly as before. +- An offset-less one-shot timestamp (`2026-06-01T09:00:00`) is also read in the job's zone; a timestamp that carries its own offset is already an instant. +- Changing the zone recomputes `next_run_at` immediately, so the next fire happens at the new zone's wall clock. +- An `interval` job has a fixed duration and no wall clock, so a pin has no effect on it. + ## Editing jobs You do not need to delete and recreate jobs just to change them.