fix(cron): per-job timezone drives schedule compute and due-scan

A job may pin its own IANA timezone; its cron expression is then that zone's wall clock,
independent of the profile's configured timezone. Absent/None inherits the profile, so an
existing jobs.json keeps byte-identical behaviour and needs no migration.

The dispatch path needs the pin as much as the compute does. `compute_next_run()` resolves
the zone as job pin -> profile zone -> base offset, and every site that writes `next_run_at`
now passes it (advance, claim, resume, re-arm, recovery, and the quota/unreachable
planners); one missed site re-fires or skips a slot because the stored instant and the
expression would be read in different zones.

`_evaluate_due_job()` normalizes the stored instant into the JOB's zone, which hands every
downstream cron predicate the job's clock, and a pinned job skips
`_repair_timezone_shifted_cron` outright. That repair exists for jobs that FOLLOW the profile
zone across a zone change: on a pinned job its wall-clock comparison reads a due occurrence
as still-in-the-future and re-anchors past it.

Failing shape this fixes (profile America/Los_Angeles, job 0 9 * * 1-5 pinned to
America/New_York, scanned at 09:00 New York): `get_due_jobs()` returned [] and the job was
re-anchored to the next day, i.e. job zones east of the profile zone silently never fired.
The regression test drives the public store API and asserts both the fire and the absence of
the repair, and is proven red on main ([] and a rewritten instant).

Knobs live in the new sibling `cron/jobs_time.py`; `cron/jobs.py` stays over the file-lines
ratchet and only gained the call-site plumbing.

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

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",
 })
 
 

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):

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

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)

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:

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.

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: {}"))
 

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")

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"]]

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)

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

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"

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

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()

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"

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):

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:

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.