feat(cron): optional per-job grace_minutes override

The catch-up window (how late a missed cron occurrence may run instead of being skipped) was
half the job's period, clamped to 120s-2h, for every job. A job can now override it outright:
0-1440 minutes, `0` meaning never catch up.

`grace_minutes` is a real 0, not "unset" — and the obvious write pattern would eat it:
`v in {None, "", False}` (the `workdir` update lambda) and `if value:` both treat `0` as
False in Python. Both doors therefore share `jobs_time.normalize_grace_minutes`, reads test
`is None`, and the renderer formats before testing truthiness.

The override moves the dispatch stamp too: `_classify_dispatch_lateness` and the dispatch
health metric read the same window, so a job the scheduler calls "late" is never reported
overdue by the dashboard.

Refused on `once`/`interval` schedules: the window describes a missed wall clock, and a
duration has none.

Split from the timezone commit so it can be reverted alone.

agent/monitoring/cron_health.py

@@ -120,7 +120,10 @@ def _is_overdue(job: dict[str, Any], now: datetime) -> bool:
     try:
         if next_run.tzinfo is None and now.tzinfo is not None:
             next_run = next_run.replace(tzinfo=now.tzinfo)
-        return (now - next_run).total_seconds() > _compute_grace_seconds(schedule)
+        # A per-job grace override moves the scheduler's own overdue line; the metric must
+        # agree with it or the dashboard flags jobs the ticker considers on time.
+        return (now - next_run).total_seconds() > _compute_grace_seconds(
+            schedule, job.get("grace_minutes"))
     except (TypeError, ValueError):
         return False
 

cron/AGENTS.md

@@ -13,8 +13,9 @@ 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), `timezone` (pin the cron expression's wall clock to an IANA zone; absent = the
-profile's zone), multi-platform delivery.
+loaded), `timezone` (pin the cron expression's wall clock to an IANA zone; absent = the profile's
+zone), `grace_minutes` (override the catch-up window; `0` = never catch up, cron only),
+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,16 +23,18 @@ Hardening invariants — each guards a real failure; don't weaken without answer
   session is hard-interrupted so it cannot monopolise the scheduler, while a long-but-active job
   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.
+- Catch-up window = half the period, clamped to 120s–2h, or the job's `grace_minutes` when set
+  (`0` = never catch up — a real value, never "unset": `0 == False`); 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.
+  future and re-anchors past it (job silently never fires). Knobs live in `cron/jobs_time.py`;
+  `timezone`/`grace_minutes` must stay in `job_definition.JOB_DEFINITION_FIELDS` or a profile
+  distribution strips them.
 - 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/jobs.py

@@ -944,9 +944,15 @@ _MIN_GRACE_SECONDS = 120
 _MAX_GRACE_SECONDS = 7200
 
 
-def _compute_grace_seconds(schedule: dict) -> int:
+def _compute_grace_seconds(schedule: dict, grace_minutes: Any = None) -> int:
     """How late a job can be and still catch up rather than fast-forward: half the period, clamped
-    to [120s, 2h], so daily jobs catch up but frequent jobs fast-forward quickly."""
+    to [120s, 2h], so daily jobs catch up but frequent jobs fast-forward quickly.
+
+    A per-job ``grace_minutes`` overrides that window outright (0 = never catch up: a missed
+    occurrence is skipped, not run late)."""
+    override = jobs_time.grace_seconds_override(grace_minutes)
+    if override is not None:
+        return override
     period_seconds = _schedule_cadence_seconds(schedule)
     if not period_seconds:
         return _MIN_GRACE_SECONDS
@@ -1751,6 +1757,7 @@ _CREATE_FIELD_NORMALIZERS: dict[str, Callable[[Any], Any]] = {
     "failure_deliver": _normalize_failure_deliver,
     "interpreter": _normalize_job_optional_text,
     "timezone": jobs_time.normalize_timezone,
+    "grace_minutes": jobs_time.normalize_grace_minutes,
 }
 _UPDATE_FIELD_NORMALIZERS: dict[str, Callable[[Any], Any]] = {
     # [] is an explicit zero-tool allowlist and must survive the update path as [] too (#82010).
@@ -1760,7 +1767,10 @@ _UPDATE_FIELD_NORMALIZERS: dict[str, Callable[[Any], Any]] = {
     "monitor_url": _normalize_job_optional_text,
     "interpreter": _normalize_job_optional_text,
     "reasoning_effort": _normalize_reasoning_effort,
+    # ``0`` is a real grace value (never catch up), so this normalizer must never use an
+    # ``in {None, "", False}`` test — ``0 == False`` would silently clear it.
     "timezone": jobs_time.normalize_timezone,
+    "grace_minutes": jobs_time.normalize_grace_minutes,
 }
 
 
@@ -1813,6 +1823,7 @@ def create_job(
     pinned: bool = False,
     interpreter: Optional[str] = None,
     timezone: Optional[str] = None,
+    grace_minutes: Optional[int] = None,
 ) -> dict[str, Any]:
     """Create a new cron job and return the stored record.
 
@@ -1824,7 +1835,8 @@ def create_job(
     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). timezone: IANA key pinning the schedule's
-    wall clock (None/'' = inherit the profile zone)."""
+    wall clock (None/'' = inherit the profile zone). grace_minutes: catch-up window override
+    (0 = never catch up)."""
     if not isinstance(paused, bool):
         raise ValueError("paused must be a boolean.")
     if paused_reason is not None and not isinstance(paused_reason, str):
@@ -1850,7 +1862,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"])
+    validate_schedule_knobs(parsed_schedule, f["timezone"], f["grace_minutes"])
     prompt_text = _coerce_job_text(prompt).strip()
     if not prompt_text and not f["script"] and not normalized_skills:
         raise ValueError(EMPTY_PAYLOAD_ERROR)
@@ -1914,7 +1926,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"]),
+        ("timezone", f["timezone"]), ("grace_minutes", f["grace_minutes"]),
     ):
         if value is not None:
             job[key] = value
@@ -2112,9 +2124,11 @@ 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 {"timezone", "grace_minutes"}.intersection(updates):
+            validate_schedule_knobs(
+                updated.get("schedule") or {}, updated.get("timezone"),
+                updated.get("grace_minutes"))
+            jobs_time.prune_inherited_knobs(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
@@ -3242,7 +3256,7 @@ def _evaluate_due_job(job: dict[str, Any], scan: _DueScan, run_claim_ttl: float)
 
     if not manual_run and kind == "cron" and _reanchor_stale_cron(d):
         return False
-    grace = _compute_grace_seconds(d.schedule)
+    grace = _compute_grace_seconds(d.schedule, job.get("grace_minutes"))
     if not manual_run and recurring and _fast_forward_missed_recurring(d, grace):
         return False
     if kind == "once":

cron/jobs_invariants.py

@@ -42,15 +42,29 @@ 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.
+def validate_schedule_knobs(
+    schedule: dict, timezone: Optional[str], grace_minutes: Optional[int],
+) -> None:
+    """Per-job schedule knobs (``timezone``/``grace_minutes``) are validated at BOTH doors —
+    ``create_job`` and ``update_job`` — so an update cannot store a schedule knob a create
+    would reject.
 
-    Raises ValueError for an unknown IANA zone.
+    Raises ValueError for an unknown IANA zone, and for ``grace_minutes`` on a schedule that has
+    no wall clock to be late for (``once``/``interval``): the catch-up window only means
+    something to a recurring cron cadence, so accepting it silently would suggest an effect it
+    cannot have.
     """
-    from cron.jobs_time import normalize_timezone
+    from cron.jobs_time import normalize_grace_minutes, normalize_timezone
 
     normalize_timezone(timezone)
+    minutes = normalize_grace_minutes(grace_minutes)
+    if minutes is None:
+        return
+    kind = (schedule or {}).get("kind")
+    if kind != "cron":
+        raise ValueError(
+            f"grace_minutes only applies to cron schedules (a fixed wall clock that can be "
+            f"missed); this job's schedule kind is {kind!r}.")
 
 
 def _script_is_command_line(script: str) -> bool:

cron/jobs_time.py

@@ -1,8 +1,9 @@
-"""Per-job ``timezone`` knob (sibling of ``cron/jobs.py``).
+"""Per-job ``timezone`` and ``grace_minutes`` knobs (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.
+store written before this field existed keeps byte-identical behaviour. ``grace_minutes``
+overrides the computed half-period catch-up window for one job (``0`` = never catch up).
 
 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
@@ -26,6 +27,12 @@ from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 logger = logging.getLogger(__name__)
 
 TIMEZONE_KEY = "timezone"
+GRACE_KEY = "grace_minutes"
+
+# ``0`` is a real value ("skip a missed occurrence instead of running it late"); a whole day is
+# the most a late run can usefully be.
+MIN_GRACE_MINUTES = 0
+MAX_GRACE_MINUTES = 24 * 60
 
 # A corrupt persisted value would otherwise log on every 60s scan. Warn once per distinct value.
 _warned: set[str] = set()
@@ -70,6 +77,40 @@ def normalize_timezone(value: Any) -> Optional[str]:
             f"`zoneinfo.available_timezones()`, e.g. 'Asia/Hong_Kong'.") from exc
 
 
+def normalize_grace_minutes(value: Any) -> Optional[int]:
+    """Validate a job's ``grace_minutes`` override; None means "use the computed window".
+
+    ``0`` is a real value (never catch up), so it must never be collapsed into absent:
+    ``0 == False`` in Python, which is exactly the trap the ``workdir`` update lambda
+    (``v in {None, "", False}``) would spring. Bools are therefore rejected explicitly.
+    """
+    if value is None:
+        return None
+    if isinstance(value, bool):
+        raise ValueError(
+            "grace_minutes must be a whole number of minutes (0 = never catch up), "
+            "not a boolean.")
+    if isinstance(value, str):
+        if not value.strip():
+            return None
+        if not value.strip().lstrip("+-").isdigit():
+            raise ValueError(
+                f"grace_minutes must be a whole number of minutes, not {value!r}.")
+    try:
+        minutes = int(value)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(
+            f"grace_minutes must be a whole number of minutes, not {value!r}.") from exc
+    if isinstance(value, float) and value != minutes:
+        raise ValueError(
+            f"grace_minutes must be a whole number of minutes, not {value!r}.")
+    if not MIN_GRACE_MINUTES <= minutes <= MAX_GRACE_MINUTES:
+        raise ValueError(
+            f"grace_minutes must be between {MIN_GRACE_MINUTES} and "
+            f"{MAX_GRACE_MINUTES} (0 = never catch up), got {minutes}.")
+    return minutes
+
+
 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.
 
@@ -90,12 +131,33 @@ def zone_for_job(job_or_timezone: Any) -> Optional[ZoneInfo]:
         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."""
+def grace_seconds_override(grace_minutes: Any) -> Optional[int]:
+    """Catch-up window in seconds from an explicit override, or None to compute it from cadence.
+
+    Tolerant read (the record may have been hand-edited): a bad value is ignored rather than
+    crashing the scan.
+    """
+    try:
+        minutes = normalize_grace_minutes(grace_minutes)
+    except ValueError as exc:
+        _warn_once(GRACE_KEY, grace_minutes, str(exc))
+        return None
+    return None if minutes is None else minutes * 60
+
+
+def prune_inherited_knobs(record: dict[str, Any]) -> None:
+    """An explicitly cleared knob is stored by DROPPING the key, not 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.
+
+    ``grace_minutes`` is tested with ``is None`` — ``0`` is the real "never catch up" value and
+    must survive, which is exactly the ``0 == False`` trap an ``in {None, "", False}`` test
+    would spring.
+    """
     if TIMEZONE_KEY in record and not record.get(TIMEZONE_KEY):
         record.pop(TIMEZONE_KEY, None)
+    if GRACE_KEY in record and record.get(GRACE_KEY) is None:
+        record.pop(GRACE_KEY, None)
 
 
 def oneshot_instant(value: Any, zone: Optional[ZoneInfo] = None) -> Optional[datetime]:

hermes_cli/cron.py

@@ -247,6 +247,7 @@ def _job_rows(job: dict[str, Any]) -> list[tuple[str, str]]:
         ("Mode", color("no-agent", Colors.DIM) + " (script stdout delivered directly)"
          if job.get("no_agent") else ""),
         ("Timezone", job.get("timezone")),
+        ("Grace", f"{job['grace_minutes']} min" if job.get("grace_minutes") is not None else ""),
         ("Workdir", job.get("workdir")),
         ("Python", job.get("interpreter")),
         ("Last run", f"{job.get('last_run_at', '?')}  {_last_run_display(job)}"
@@ -738,6 +739,15 @@ def _cron_doctor_issues_for_job(job: dict[str, Any]) -> list[str]:
         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")
+    if job.get("grace_minutes") is not None:
+        from cron.jobs_time import normalize_grace_minutes
+        try:
+            normalize_grace_minutes(job.get("grace_minutes"))
+        except ValueError as exc:
+            issues.append(f"grace_minutes is unusable ({exc})")
+        else:
+            if (job.get("schedule") or {}).get("kind") != "cron":
+                issues.append("grace_minutes has no effect on a non-cron schedule")
     return issues
 
 
@@ -768,7 +778,8 @@ _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"), ("timezone", "timezone"))
+                   ("interpreter", "interpreter"), ("timezone", "timezone"),
+                   ("grace_minutes", "grace_minutes"))
 
 
 def _job_api_kwargs(args) -> dict[str, Any]:

hermes_cli/subcommands/cron.py

@@ -85,6 +85,10 @@ def build_cron_parser(subparsers, *, cmd_cron: Callable) -> None:
         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("--grace-minutes", dest="grace_minutes",
+        help="Catch-up window override in minutes (0-1440) for a cron schedule. "
+            "Default: half the period, clamped to 2-120 minutes. 0 = never catch up "
+            "(a missed occurrence is skipped, not run late).")
     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 "
@@ -158,6 +162,9 @@ def build_cron_parser(subparsers, *, cmd_cron: Callable) -> None:
     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.")
+    cron_edit.add_argument("--grace-minutes", dest="grace_minutes",
+        help="Catch-up window override in minutes (0-1440). 0 = never catch up. "
+            "Pass empty string to restore the computed window.")
 
     # lifecycle actions
     cron_pause = cron_subparsers.add_parser("pause", help="Pause a scheduled job")

tests/cron/test_jobs_time.py

@@ -1,7 +1,10 @@
-"""Unit contract for the per-job ``timezone`` knob: ``cron/jobs_time.py``.
+"""Unit contract for the per-job schedule knobs: ``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.
+Covers the two values a job may pin — an IANA ``timezone`` for its cron wall clock and a
+``grace_minutes`` override for the catch-up window — including the two traps that cost a
+silent misbehaviour rather than an exception: ``0 == False`` (a grace of 0 means "never catch
+up", not "unset") and a corrupt persisted zone (must degrade to the profile zone, warned once,
+never break the 60s tick).
 """
 
 from datetime import datetime
@@ -35,6 +38,43 @@ class TestNormalizeTimezone:
             jobs_time.normalize_timezone(9)
 
 
+class TestGraceMinutesNormalization:
+    """The ``0 == False`` trap: 0 must survive as a real override."""
+
+    @pytest.mark.parametrize("value", [None, "", "  "])
+    def test_empty_means_use_the_computed_window(self, value):
+        assert jobs_time.normalize_grace_minutes(value) is None
+
+    def test_zero_is_a_real_value_not_unset(self):
+        assert jobs_time.normalize_grace_minutes(0) == 0
+        assert jobs_time.normalize_grace_minutes("0") == 0
+        # The whole point: 0 is falsy but must NOT read as "absent".
+        assert jobs_time.normalize_grace_minutes(0) is not None
+
+    def test_upper_bound_is_one_day(self):
+        assert jobs_time.normalize_grace_minutes(24 * 60) == 24 * 60
+        with pytest.raises(ValueError, match="between"):
+            jobs_time.normalize_grace_minutes(24 * 60 + 1)
+
+    def test_negative_is_rejected(self):
+        with pytest.raises(ValueError, match="between"):
+            jobs_time.normalize_grace_minutes(-1)
+
+    @pytest.mark.parametrize("value", [True, False])
+    def test_booleans_are_rejected(self, value):
+        with pytest.raises(ValueError, match="boolean"):
+            jobs_time.normalize_grace_minutes(value)
+
+    @pytest.mark.parametrize("value", ["soon", "1.5", [30]])
+    def test_non_numeric_is_rejected(self, value):
+        with pytest.raises(ValueError, match="grace_minutes"):
+            jobs_time.normalize_grace_minutes(value)
+
+    def test_fractional_float_is_rejected(self):
+        with pytest.raises(ValueError, match="whole number"):
+            jobs_time.normalize_grace_minutes(1.5)
+
+
 class TestZoneForJob:
     def test_absent_or_null_inherits_the_profile(self):
         assert jobs_time.zone_for_job({}) is None
@@ -64,15 +104,38 @@ class TestZoneForJob:
         assert "Mars/Olympus_Mons" in warnings[0]
 
 
-class TestPruneInheritedTimezone:
+class TestGraceSecondsOverride:
+    def test_absent_override_defers_to_the_cadence_window(self):
+        assert jobs_time.grace_seconds_override(None) is None
+
+    def test_minutes_become_seconds(self):
+        assert jobs_time.grace_seconds_override(30) == 1800
+
+    def test_zero_override_means_never_catch_up(self):
+        assert jobs_time.grace_seconds_override(0) == 0
+
+    def test_corrupt_override_is_ignored_with_one_warning(self, monkeypatch):
+        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))
+
+        assert jobs_time.grace_seconds_override("tomorrow") is None
+        assert jobs_time.grace_seconds_override("tomorrow") is None
+        assert len(warnings) == 1
+
+
+class TestPruneInheritedKnobs:
     def test_cleared_zone_drops_the_key(self):
-        record = {"id": "a", "timezone": ""}
-        jobs_time.prune_inherited_timezone(record)
+        record = {"id": "a", "timezone": "", "grace_minutes": 0}
+        jobs_time.prune_inherited_knobs(record)
         assert "timezone" not in record
+        assert record["grace_minutes"] == 0, "grace 0 is a value, not an unset key"
 
     def test_pinned_zone_is_kept(self):
         record = {"timezone": "Asia/Hong_Kong"}
-        jobs_time.prune_inherited_timezone(record)
+        jobs_time.prune_inherited_knobs(record)
         assert record["timezone"] == "Asia/Hong_Kong"
 
 

tests/cron/test_per_job_timezone.py

@@ -62,11 +62,11 @@ def cron_store(tmp_path, monkeypatch):
         yield clock
 
 
-def _add_job(schedule, *, timezone=None, next_run_at):
+def _add_job(schedule, *, timezone=None, grace_minutes=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)
+        timezone=timezone, grace_minutes=grace_minutes)
     stored = jobs.load_jobs()
     row = next(r for r in stored if r["id"] == job["id"])
     row["next_run_at"] = next_run_at.isoformat()
@@ -200,3 +200,70 @@ class TestInheritanceAndDegradation:
         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"
+
+
+class TestGraceMinutesOverride:
+    """``grace_minutes`` replaces the cadence-derived window for ONE job.
+
+    Each case uses an hourly job 45 minutes late: the computed window for a 1h cadence is 30min
+    (half the period, floored at 120s... capped at 2h), so the late run is a ``catch_up``;
+    a 120-minute override puts the same run inside the window and it is merely ``late``. The
+    dispatch stamp is what the dashboard and health check read, so it must follow the override.
+    """
+
+    def _stamp_for(self, la_profile, cron_store, *, grace_minutes):
+        now = datetime(2026, 10, 22, 12, 45, tzinfo=LA)
+        cron_store.set(now)
+        job_id = _add_job(
+            "0 * * * *", timezone="America/Los_Angeles", grace_minutes=grace_minutes,
+            next_run_at=datetime(2026, 10, 22, 12, 0, tzinfo=LA))
+        due = [j["id"] for j in jobs.get_due_jobs()]
+        assert due == [job_id], "the missed occurrence still fires; only its classification moves"
+        stored = next(r for r in jobs.load_jobs() if r["id"] == job_id)
+        return stored["last_dispatch"]["kind"]
+
+    def test_computed_window_classifies_a_45_min_late_hourly_run_as_catch_up(
+        self, la_profile, cron_store,
+    ):
+        assert self._stamp_for(la_profile, cron_store, grace_minutes=None) == "catch_up"
+
+    def test_override_widens_the_window_so_the_same_run_is_only_late(
+        self, la_profile, cron_store,
+    ):
+        assert self._stamp_for(la_profile, cron_store, grace_minutes=120) == "late"
+
+    def test_zero_grace_skips_outright_when_catch_up_is_disabled(
+        self, la_profile, cron_store, tmp_path,
+    ):
+        """``0`` = never catch up. Hermes still runs a beyond-window miss once by default
+        (``cron.catch_up_missed``); with catch-up disabled the zero window is what makes the
+        skip, while a job with a wide override still fires."""
+        (tmp_path / "config.yaml").write_text(
+            "cron:\n  catch_up_missed: false\n", encoding="utf-8")
+        now = datetime(2026, 10, 22, 12, 0, tzinfo=LA)
+        cron_store.set(now)
+        instant = datetime(2026, 10, 22, 11, 0, tzinfo=LA)  # one hour late, on the lattice
+        never = _add_job("0 11 * * *", timezone="America/Los_Angeles", grace_minutes=0,
+                         next_run_at=instant)
+        tolerant = _add_job("0 11 * * *", timezone="America/Los_Angeles", grace_minutes=120,
+                            next_run_at=instant)
+
+        due = {j["id"] for j in jobs.get_due_jobs()}
+
+        assert tolerant in due, "a 2h window covers a 1h-late run"
+        assert never not in due, "grace 0 + catch-up disabled means the miss is skipped"
+
+    def test_grace_is_refused_on_a_non_cron_schedule(self, la_profile, cron_store):
+        with pytest.raises(ValueError, match="grace_minutes only applies to cron"):
+            jobs.create_job(
+                prompt="p", schedule="every 5m", model="fixture", deliver="local",
+                grace_minutes=30)
+
+    @pytest.mark.parametrize("minutes, expected", [(0, 0), (45, 45 * 60)])
+    def test_stored_grace_round_trips(self, la_profile, cron_store, minutes, expected):
+        job = jobs.create_job(
+            prompt="p", schedule=WEEKDAY_9AM, model="fixture", deliver="local",
+            grace_minutes=minutes)
+        stored = next(r for r in jobs.load_jobs() if r["id"] == job["id"])
+        assert stored["grace_minutes"] == minutes, "0 must not be dropped as falsy"
+        assert jobs._compute_grace_seconds(stored["schedule"], stored["grace_minutes"]) == expected

tests/hermes_cli/test_cron_parser_builder.py

@@ -42,13 +42,16 @@ def test_cron_accept_hooks_flag_on_run_and_tick():
     assert ns2.accept_hooks is True
 
 
-def test_cron_timezone_flags():
+def test_cron_timezone_and_grace_flags():
     parser = _build()
     created = parser.parse_args(
-        ["cron", "create", "0 9 * * *", "hi", "--timezone", "Asia/Hong_Kong"])
+        ["cron", "create", "0 9 * * *", "hi", "--timezone", "Asia/Hong_Kong",
+         "--grace-minutes", "0"])
     assert created.timezone == "Asia/Hong_Kong"
+    assert created.grace_minutes == "0", "an int-ish string must survive to the normalizer"
 
-    edited = parser.parse_args(["cron", "edit", "j", "--timezone", ""])
+    edited = parser.parse_args(["cron", "edit", "j", "--timezone", "", "--grace-minutes", ""])
     assert edited.timezone == "", "an empty value is the documented clear"
+    assert edited.grace_minutes == ""
 
     assert parser.parse_args(["cron", "edit", "j"]).timezone is None

tests/tools/test_cronjob_tools.py

@@ -4,6 +4,7 @@ import json
 import pytest
 
 from tools.cronjob_tools import (
+    CRONJOB_SCHEMA,
     _scan_cron_prompt,
     check_cronjob_requirements,
     cronjob,
@@ -805,7 +806,7 @@ class TestGithubExemptionAbuse:
 
 
 class TestScheduleKnobToolSurface:
-    """The model-facing tool must expose the per-job timezone knob (§4)."""
+    """The model-facing tool must expose the per-job timezone and grace knobs (§4/§5)."""
 
     @pytest.fixture(autouse=True)
     def _setup_cron_dir(self, tmp_path, monkeypatch):
@@ -817,13 +818,14 @@ class TestScheduleKnobToolSurface:
         from cron.jobs import load_jobs
         return load_jobs()[0]
 
-    def test_create_stores_timezone(self):
+    def test_create_stores_timezone_and_grace(self):
         created = json.loads(cronjob(
             action="create", prompt="HK open", schedule="0 9 * * 1-5",
-            timezone="Asia/Hong_Kong"))
+            timezone="Asia/Hong_Kong", grace_minutes=45))
         assert created["success"] is True, created
         job = self._stored()
         assert job["timezone"] == "Asia/Hong_Kong"
+        assert job["grace_minutes"] == 45
         assert job["next_run_at"].endswith("+08:00") or job["next_run_at"].endswith("+09:00")
 
     def test_create_rejects_an_unknown_zone(self):
@@ -833,12 +835,32 @@ class TestScheduleKnobToolSurface:
         assert created["success"] is False
         assert "timezone" in created["error"].lower()
 
-    def test_update_clears_the_zone(self):
+    def test_grace_zero_is_stored_not_dropped(self):
+        cronjob(action="create", prompt="x", schedule="0 9 * * *", grace_minutes=0)
+        assert self._stored()["grace_minutes"] == 0
+
+    def test_update_clears_the_zone_and_the_override(self):
         created = json.loads(cronjob(
             action="create", prompt="x", schedule="0 9 * * *",
-            timezone="Asia/Hong_Kong"))
+            timezone="Asia/Hong_Kong", grace_minutes=30))
         job_id = created["job_id"]
 
-        updated = json.loads(cronjob(action="update", job_id=job_id, timezone=""))
+        updated = json.loads(cronjob(
+            action="update", job_id=job_id, timezone="", grace_minutes=""))
         assert updated["success"] is True, updated
-        assert "timezone" not in self._stored(), "'' clears the pin; the key is dropped"
+        job = self._stored()
+        assert "timezone" not in job, "'' clears the pin; the key is dropped"
+        assert "grace_minutes" not in job
+
+    def test_grace_schema_accepts_a_string_so_the_model_can_clear_it(self):
+        """``''`` means "restore the computed window"; an integer-only schema would leave the
+        model no way to express it (the sanitizer normalizes ``type: [...]`` per provider)."""
+        prop = CRONJOB_SCHEMA["parameters"]["properties"]["grace_minutes"]
+        assert "string" in prop["type"] and "integer" in prop["type"]
+        assert "''" in prop["description"]
+
+    def test_grace_on_a_non_cron_schedule_is_an_error(self):
+        created = json.loads(cronjob(
+            action="create", prompt="x", schedule="every 2h", grace_minutes=30))
+        assert created["success"] is False
+        assert "grace_minutes" in created["error"]

tools/cronjob_tools.py

@@ -731,6 +731,7 @@ def _action_create(a: dict[str, Any]) -> str:
             # 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"]),
+            grace_minutes=a["grace_minutes"],
             pinned=bool(a["pinned"]),
             failure_deliver=_resolve_cron_context_deliver(_normalize_deliver_param(a["failure_deliver"])),
             **({"paused": a["paused"], "paused_reason": a["paused_reason"]}
@@ -963,6 +964,15 @@ def _update_run_fields(job: dict[str, Any], a: dict[str, Any], updates: dict[str
     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["grace_minutes"] is not None:
+        # '' / null clears the override; 0 is a real value ("never catch up") and passes through.
+        updates["grace_minutes"] = (
+            None if isinstance(a["grace_minutes"], str) and not a["grace_minutes"].strip()
+            else a["grace_minutes"])
+        if _pick(updates, job, "grace_minutes") is not None:
+            kind = (_pick(updates, job, "schedule") or {}).get("kind")
+            if kind != "cron":
+                return "grace_minutes only applies to cron schedules."
     if a["schedule"] is not None:
         parsed_schedule = parse_schedule(a["schedule"])
         updates["schedule"] = parsed_schedule
@@ -1053,7 +1063,8 @@ def cronjob(
     paused_reason: Optional[str] = None,
     pinned: Optional[bool] = None,
     interpreter: Optional[str] = None,
-    timezone: Optional[str] = None) -> str:
+    timezone: Optional[str] = None,
+    grace_minutes: Optional[int | str] = None) -> str:
     """Unified cron job management tool."""
     a = dict(locals())
     del a["task_id"]  # unused but kept for handler signature compatibility
@@ -1179,6 +1190,10 @@ Jobs run in a fresh session with no current-chat context, so prompts must be sel
                 "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."
             },
+            "grace_minutes": {
+                "type": ["integer", "string"],
+                "description": "Optional catch-up window override for a cron job, in minutes (0-1440). The default is half the job's period, clamped to 2-120 minutes. 0 means never catch up: a missed occurrence is skipped instead of being run late. Cron schedules only. On update, '' (empty string) restores the computed window."
+            },
             "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'."
@@ -1211,7 +1226,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", "timezone")
+    "paused_reason", "pinned", "timezone", "grace_minutes")
 
 
 def _cronjob_handler(args, **kw):

website/docs/developer-guide/cron-internals.md

@@ -13,7 +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/jobs_time.py` | Per-job `timezone` / `grace_minutes` knobs: 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 |
@@ -49,7 +49,7 @@ stored offset legitimately differs from the profile's, and the repair's wall-clo
 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
+`timezone` and `grace_minutes` are top-level record fields (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()`.

website/docs/user-guide/features/cron.md

@@ -225,6 +225,17 @@ hermes cron create "0 9 * * 1-5" --timezone America/New_York "US market open bri
 - 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.
 
+## Catch-up window
+
+A missed cron occurrence is run late only inside a catch-up window: half the job's period, clamped to 2 minutes–2 hours. `--grace-minutes` overrides that per job (0–1440):
+
+```bash
+hermes cron edit market-open --grace-minutes 0    # never catch up: skip a missed occurrence
+hermes cron edit market-open --grace-minutes 180  # run up to 3h late
+```
+
+`0` means never catch up; `--grace-minutes ""` restores the computed window. It only applies to cron expressions — `once` and `interval` schedules have no missed wall clock to catch up on.
+
 ## Editing jobs
 
 You do not need to delete and recreate jobs just to change them.