"""Tests for cron/jobs.py — schedule parsing, job CRUD, and due-job detection.""" import threading import pytest from datetime import datetime, timedelta, timezone from cron.jobs import ( parse_duration, parse_schedule, compute_next_run, create_job, load_jobs, save_jobs, get_job, list_jobs, update_job, pause_job, resume_job, remove_job, mark_job_run, advance_next_run, claim_dispatch, claim_job_for_fire, heartbeat_run_claim, get_due_jobs, save_job_output, _hermes_now, ) # ========================================================================= # parse_duration # ========================================================================= class TestParseDuration: def test_minutes(self): assert parse_duration("30m") == 30 assert parse_duration("1min") == 1 assert parse_duration("5mins") == 5 assert parse_duration("10minute") == 10 assert parse_duration("120minutes") == 120 def test_bare_units_default_to_one(self): assert parse_duration("hour") == 60 assert parse_duration("hr") == 60 assert parse_duration("h") == 60 assert parse_duration("minute") == 1 assert parse_duration("m") == 1 assert parse_duration("day") == 1440 assert parse_duration("d") == 1440 def test_every_bare_unit_schedule(self): result = parse_schedule("every hour") assert result["kind"] == "interval" assert result["minutes"] == 60 result = parse_schedule("every day") assert result["kind"] == "interval" assert result["minutes"] == 1440 def test_invalid_raises(self): with pytest.raises(ValueError): parse_duration("abc") with pytest.raises(ValueError): parse_duration("hourx") with pytest.raises(ValueError): parse_duration("30x") with pytest.raises(ValueError): parse_duration("") with pytest.raises(ValueError): parse_duration("m30") # ========================================================================= # parse_schedule # ========================================================================= class TestParseSchedule: def test_bare_duration_becomes_recurring_interval(self): """Contract: bare '30m' means EVERY 30 minutes (tool schema says so). Regression for the cron contract bug (2026-08-04): parse_schedule returned kind='once' for bare durations, so an agent passing '30m' for 'every 30 minutes' silently got a one-shot job that ran once and died. The documented tool contract (tools/cronjob_tools.py) says '30m' (every 30 minutes) — the code now honors it. """ result = parse_schedule("30m") assert result["kind"] == "interval" assert result["minutes"] == 30 assert "run_at" not in result def test_in_duration_becomes_once(self): """Explicit one-shot by duration: 'in 30m' fires once in 30 minutes.""" result = parse_schedule("in 30m") assert result["kind"] == "once" run_at_str = result["run_at"] assert isinstance(run_at_str, str) run_at = datetime.fromisoformat(run_at_str) now = datetime.now().astimezone() assert run_at > now assert run_at < now + timedelta(minutes=31) def test_every_becomes_interval(self): result = parse_schedule("every 2h") assert result["kind"] == "interval" assert result["minutes"] == 120 # ---- Natural-language weekday/daily phrases → cron (issue: documented # "every monday 9am" format was rejected because the "every" branch only # accepted durations). ---- def test_every_weekday_time_becomes_cron(self): pytest.importorskip("croniter") result = parse_schedule("every monday 9am") assert result["kind"] == "cron" assert result["expr"] == "0 9 * * 1" # Display preserves the user's natural phrasing. assert result["display"] == "every monday 9am" def test_every_sunday_maps_to_zero(self): pytest.importorskip("croniter") # Cron weekday numbering puts Sunday at 0. assert parse_schedule("every sunday 9am")["expr"] == "0 9 * * 0" def test_every_weekday_abbreviations(self): pytest.importorskip("croniter") assert parse_schedule("every mon 9am")["expr"] == "0 9 * * 1" assert parse_schedule("every fri 5pm")["expr"] == "0 17 * * 5" def test_every_day_keyword_is_daily(self): pytest.importorskip("croniter") assert parse_schedule("every day at 9am")["expr"] == "0 9 * * *" assert parse_schedule("every day 7am")["expr"] == "0 7 * * *" def test_every_weekday_keyword_is_business_days(self): pytest.importorskip("croniter") assert parse_schedule("every weekday at 9am")["expr"] == "0 9 * * 1-5" def test_every_weekend_keyword(self): pytest.importorskip("croniter") assert parse_schedule("every weekend at 10am")["expr"] == "0 10 * * 0,6" def test_no_every_prefix_natural_forms(self): # The Desktop dialog advertises these WITHOUT the "every" prefix # (#51975 repro): they must parse identically. pytest.importorskip("croniter") assert parse_schedule("weekdays at 9am")["expr"] == "0 9 * * 1-5" assert parse_schedule("monday at 9:30")["expr"] == "30 9 * * 1" assert parse_schedule("daily at 7am")["expr"] == "0 7 * * *" def test_weekday_list_forms(self): # Comma/"and"-separated day lists from the #51975 repro. pytest.importorskip("croniter") assert parse_schedule("Monday, Wednesday at 9am")["expr"] == "0 9 * * 1,3" assert parse_schedule("monday and friday at 5pm")["expr"] == "0 17 * * 1,5" assert parse_schedule("every tue, thu 8am")["expr"] == "0 8 * * 2,4" def test_natural_form_negatives_still_reject(self): pytest.importorskip("croniter") with pytest.raises(ValueError): parse_schedule("monday banana at 9am") with pytest.raises(ValueError): parse_schedule("funday at 9am") def test_every_time_formats(self): pytest.importorskip("croniter") # 24-hour, explicit minutes, noon/midnight, bare hour. assert parse_schedule("every monday 14:30")["expr"] == "30 14 * * 1" assert parse_schedule("every monday 9:05am")["expr"] == "5 9 * * 1" assert parse_schedule("every monday noon")["expr"] == "0 12 * * 1" assert parse_schedule("every monday midnight")["expr"] == "0 0 * * 1" assert parse_schedule("every monday at 7")["expr"] == "0 7 * * 1" def test_every_12_hour_boundaries(self): pytest.importorskip("croniter") # 12am is midnight (00:00), 12pm is noon (12:00). assert parse_schedule("every monday 12am")["expr"] == "0 0 * * 1" assert parse_schedule("every monday 12pm")["expr"] == "0 12 * * 1" def test_every_weekday_time_is_case_insensitive(self): pytest.importorskip("croniter") assert parse_schedule("Every Monday 9AM")["expr"] == "0 9 * * 1" def test_every_weekday_schedule_computes_next_run(self): pytest.importorskip("croniter") # End-to-end: the produced cron schedule is usable by compute_next_run. schedule = parse_schedule("every monday 9am") next_run = compute_next_run(schedule) assert next_run is not None dt = datetime.fromisoformat(next_run) assert dt.weekday() == 0 # Python: Monday == 0 assert (dt.hour, dt.minute) == (9, 0) def test_every_duration_still_interval(self): # The interval path must keep working unchanged. assert parse_schedule("every 30m")["kind"] == "interval" assert parse_schedule("every 1d")["minutes"] == 1440 def test_every_weekday_without_time_raises(self): # A weekday with no time is ambiguous — reject rather than guess. with pytest.raises(ValueError): parse_schedule("every monday") def test_every_invalid_time_raises(self): with pytest.raises(ValueError): parse_schedule("every monday 25am") with pytest.raises(ValueError): parse_schedule("every monday 9pm pizza") def test_cron_expression(self): pytest.importorskip("croniter") result = parse_schedule("0 9 * * *") assert result["kind"] == "cron" assert result["expr"] == "0 9 * * *" def test_cron_named_weekdays_and_months(self): # Named months/weekdays (and ranges/lists) are valid cron and must # route to croniter, not be rejected as "Invalid schedule". pytest.importorskip("croniter") for expr in ( "0 9 * * MON", "*/15 9-17 * * MON-FRI", "0 9 1 JAN *", "0 9 * * MON,WED,FRI", ): result = parse_schedule(expr) assert result["kind"] == "cron", expr assert result["expr"] == expr def test_invalid_named_cron_still_rejected(self): pytest.importorskip("croniter") with pytest.raises(ValueError): parse_schedule("0 9 * * FUNDAY") def test_iso_timestamp(self): result = parse_schedule("2030-01-15T14:00:00") assert result["kind"] == "once" assert "2030-01-15" in result["run_at"] def test_naive_iso_anchors_to_configured_tz_not_server_local(self, monkeypatch): """A naive ISO timestamp must be interpreted in the CONFIGURED Hermes timezone, NOT the server's local timezone (#51021). Regression: when the configured zone differs from the server's local zone (common on cloud hosts running UTC), parse_schedule used ``dt.astimezone()`` (server-local), baking in the wrong offset. The due-check compares against ``_hermes_now()`` (configured zone), so the stored instant landed hours off the user's wall-clock intent — far enough that one-shots never became due. This asserts the parsed offset matches the configured-now offset, the invariant that keeps the stored instant on the same clock the scheduler checks against. """ configured_now = datetime(2026, 6, 22, 20, 0, 0, tzinfo=timezone(timedelta(hours=5, minutes=30))) monkeypatch.setattr("cron.jobs._hermes_now", lambda: configured_now) result = parse_schedule("2026-06-22T20:07:00") # naive, user wall-clock assert result["kind"] == "once" parsed = datetime.fromisoformat(result["run_at"]) assert parsed.utcoffset() == configured_now.utcoffset() # Same wall-clock the user typed, on the configured clock. assert parsed.replace(tzinfo=None) == datetime(2026, 6, 22, 20, 7, 0) # ========================================================================= # Timezone-divergence regression (#51021) # ========================================================================= class TestNaiveScheduleTimezoneDivergence: """End-to-end: a one-shot created with a naive recent-past timestamp must become due even when the configured Hermes timezone differs from the server's local timezone. Before #51021 the naive value was anchored to server-local, so the job never fired.""" def test_recent_past_oneshot_is_due_under_diverging_tz(self, tmp_cron_dir, monkeypatch): # Configured zone: a fixed +05:30 offset. The server's actual local # zone is irrelevant to the parse now — that is the whole point. configured = timezone(timedelta(hours=5, minutes=30)) now = datetime(2026, 6, 22, 20, 7, 30, tzinfo=configured) monkeypatch.setattr("cron.jobs._hermes_now", lambda: now) # 30s ago in the configured wall clock, supplied as a NAIVE string. naive_str = (now - timedelta(seconds=30)).replace(tzinfo=None).isoformat() job = create_job(prompt="test message", schedule=naive_str, deliver="local") due = get_due_jobs() assert any(d["id"] == job["id"] for d in due), ( f"one-shot should be due; next_run_at={job['next_run_at']}" ) # ========================================================================= # compute_next_run # ========================================================================= class TestComputeNextRun: def test_once_future_returns_time(self): future = (datetime.now() + timedelta(hours=1)).isoformat() schedule = {"kind": "once", "run_at": future} assert compute_next_run(schedule) == future def test_once_recent_past_within_grace_returns_time(self, monkeypatch): now = datetime(2026, 3, 18, 4, 22, 3, tzinfo=timezone.utc) run_at = "2026-03-18T04:22:00+00:00" monkeypatch.setattr("cron.jobs._hermes_now", lambda: now) schedule = {"kind": "once", "run_at": run_at} assert compute_next_run(schedule) == run_at def test_once_with_last_run_returns_none_even_within_grace(self, monkeypatch): now = datetime(2026, 3, 18, 4, 22, 3, tzinfo=timezone.utc) run_at = "2026-03-18T04:22:00+00:00" monkeypatch.setattr("cron.jobs._hermes_now", lambda: now) schedule = {"kind": "once", "run_at": run_at} assert compute_next_run(schedule, last_run_at=now.isoformat()) is None def test_interval_first_run(self): schedule = {"kind": "interval", "minutes": 60} result = compute_next_run(schedule) next_dt = datetime.fromisoformat(result) # Should be ~60 minutes from now assert next_dt > datetime.now().astimezone() + timedelta(minutes=59) # ========================================================================= # Job CRUD (with tmp file storage) # ========================================================================= @pytest.fixture() def tmp_cron_dir(tmp_path, monkeypatch): """Redirect cron storage to a temp directory.""" 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") return tmp_path class TestJobCRUD: def test_cjk_and_emoji_round_trip_readable_in_jobs_json(self, tmp_cron_dir): """CJK/emoji job text must round-trip AND stay human-readable on disk. With json.dump's default ensure_ascii=True, every non-ASCII char in jobs.json is written as \\uXXXX escapes, which users reported as unreadable garbage when inspecting their job store (#52302, #29754). ensure_ascii=False + the existing encoding="utf-8" writer keeps the text literal; the utf-8-sig reader must parse it back identically. """ name = "日次レポート 🎉 café" job = create_job(prompt=f"Summarize {name}", schedule="30m", name=name) # Round-trip through save/load is lossless. fetched = get_job(job["id"]) assert fetched["name"] == name assert name in fetched["prompt"] # On-disk representation is literal UTF-8, not \uXXXX escapes. from cron.jobs import JOBS_FILE raw = JOBS_FILE.read_text(encoding="utf-8") assert "日次レポート" in raw assert "🎉" in raw assert "\\u65e5" not in raw def test_create_and_get(self, tmp_cron_dir): job = create_job(prompt="Check server status", schedule="30m") assert job["id"] assert job["prompt"] == "Check server status" assert job["enabled"] is True assert job["schedule"]["kind"] == "interval" fetched = get_job(job["id"]) assert fetched is not None assert fetched["prompt"] == "Check server status" def test_list_jobs(self, tmp_cron_dir): create_job(prompt="Job 1", schedule="every 1h") create_job(prompt="Job 2", schedule="every 2h") jobs = list_jobs() assert len(jobs) == 2 def test_remove_job(self, tmp_cron_dir): job = create_job(prompt="Temp job", schedule="30m") assert remove_job(job["id"]) is True assert get_job(job["id"]) is None def test_auto_repeat_for_once(self, tmp_cron_dir): job = create_job(prompt="One-shot", schedule="in 1h") assert job["repeat"]["times"] == 1 def test_repeat_string_forms_coerced(self, tmp_cron_dir): """Agents pass 'forever'/'once' as repeat — must coerce, not TypeError. Regression for #66824/#64520/#7142: repeat='forever' died with "'<=' not supported between instances of 'str' and 'int'". The tool schema documents repeat as an integer but user-facing forms are strings; coerce at create_job so every entry point inherits it. """ forever = create_job(prompt="Str forever", schedule="every 1h", repeat="forever") assert forever["repeat"]["times"] is None # None = infinite once = create_job(prompt="Str once", schedule="every 1h", repeat="once") assert once["repeat"]["times"] == 1 three = create_job(prompt="Str 3", schedule="every 1h", repeat="3") assert three["repeat"]["times"] == 3 with pytest.raises(ValueError, match="Invalid repeat"): create_job(prompt="Bad", schedule="every 1h", repeat="banana") def test_update_repeat_string_forms_coerced(self, tmp_cron_dir): """The UPDATE path must coerce repeat the same way create does. Before this fix, update_job({"repeat": "forever"}) stored the raw string, and the next mark_job_run died with "'str' object has no attribute 'get'". Same class as the create-path TypeError (#66824/#64520/#7142/#71987/#95706) — the bare-value and dict shapes both route through normalize_repeat_value now. """ from cron.jobs import mark_job_run, update_job job = create_job(prompt="t", schedule="every 1h", repeat=2) mark_job_run(job["id"], success=True) # completed=1 updated = update_job(job["id"], {"repeat": "forever"}) assert updated["repeat"]["times"] is None assert updated["repeat"]["completed"] == 1 # counter preserved mark_job_run(job["id"], success=True) # must not raise updated = update_job(job["id"], {"repeat": {"times": "3"}}) assert updated["repeat"]["times"] == 3 assert updated["repeat"]["completed"] == 2 with pytest.raises(ValueError, match="Invalid repeat"): update_job(job["id"], {"repeat": "banana"}) def test_rejects_stale_past_one_shot_at_creation(self, tmp_cron_dir, monkeypatch): now = datetime(2026, 3, 18, 4, 30, 0, tzinfo=timezone.utc) monkeypatch.setattr("cron.jobs._hermes_now", lambda: now) stale = (now - timedelta(minutes=5)).isoformat() with pytest.raises(ValueError, match="past and cannot be scheduled"): create_job(prompt="Too late", schedule=stale) assert load_jobs() == [] def test_default_delivery_origin(self, tmp_cron_dir): job = create_job( prompt="Test", schedule="30m", origin={"platform": "telegram", "chat_id": "123"}, ) assert job["deliver"] == "origin" class TestUpdateJob: def test_update_name(self, tmp_cron_dir): job = create_job(prompt="Check server status", schedule="every 1h", name="Old Name") assert job["name"] == "Old Name" updated = update_job(job["id"], {"name": "New Name"}) assert updated is not None assert isinstance(updated, dict) assert updated["name"] == "New Name" # Verify other fields are preserved assert updated["prompt"] == "Check server status" assert updated["id"] == job["id"] assert updated["schedule"] == job["schedule"] # Verify persisted to disk fetched = get_job(job["id"]) assert fetched["name"] == "New Name" class TestPauseResumeJob: def test_pause_sets_state(self, tmp_cron_dir): job = create_job(prompt="Pause me", schedule="every 1h") paused = pause_job(job["id"], reason="user paused") assert paused is not None assert paused["enabled"] is False assert paused["state"] == "paused" assert paused["paused_reason"] == "user paused" assert paused.get("paused_at") def test_pause_is_authoritative_due_jobs_do_not_fire(self, tmp_cron_dir): """Behavioural invariant: after pause, a past-due job must not be due. Checks that last_run_at cannot advance via the scheduler path — not merely that pause() returned success. Regression for the 07-30 outage where state=paused coexisted with enabled=true and jobs kept firing. """ job = create_job(prompt="Must not fire while paused", schedule="every 1h") past = (_hermes_now() - timedelta(hours=2)).isoformat() # Force the job overdue, then pause. updated = update_job(job["id"], {"next_run_at": past}) assert updated["enabled"] is True assert job["id"] in {j["id"] for j in get_due_jobs()} paused = pause_job(job["id"], reason="outage freeze") assert paused["enabled"] is False assert paused["state"] == "paused" assert paused.get("paused_at") # Scheduler-honoured flag and pause markers must never contradict. assert not (paused.get("enabled") and paused.get("paused_at")) due_ids = {j["id"] for j in get_due_jobs()} assert job["id"] not in due_ids before = get_job(job["id"]) assert before["last_run_at"] is None or before["last_run_at"] == job.get("last_run_at") # claim path also closed assert claim_job_for_fire(job["id"]) is False after = get_job(job["id"]) assert after["last_run_at"] == before.get("last_run_at") assert after["enabled"] is False def test_contradictory_half_pause_self_disables_and_does_not_fire(self, tmp_cron_dir): """enabled=true + paused_at must not fire; scan heals enabled=false.""" now = _hermes_now() job = { "id": "half-paused-1", "name": "half-paused", "prompt": "should never run", "schedule": {"kind": "interval", "minutes": 5, "display": "every 5m"}, "schedule_display": "every 5m", "repeat": {"times": None, "completed": 0}, # The contradiction from the 07-30 outage: "enabled": True, "state": "paused", "paused_at": (now - timedelta(hours=20)).isoformat(), "paused_reason": "operator thought this was frozen", "next_run_at": (now - timedelta(hours=1)).isoformat(), "last_run_at": None, "last_status": None, "last_error": None, "last_delivery_error": None, "created_at": (now - timedelta(days=1)).isoformat(), "deliver": "local", } save_jobs([job]) # Display must NOT say paused while enabled (honest list). from cron.jobs import effective_job_state, list_jobs assert effective_job_state(job) == "scheduled" listed = {j["id"]: j for j in list_jobs(include_disabled=True)} # Honest list: enabled=true half-pause must not render as paused. assert listed["half-paused-1"]["enabled"] is True assert listed["half-paused-1"]["state"] != "paused" assert claim_job_for_fire("half-paused-1") is False due = get_due_jobs() assert "half-paused-1" not in {j["id"] for j in due} healed = get_job("half-paused-1") assert healed is not None assert healed["enabled"] is False assert healed["state"] == "paused" assert healed.get("paused_at") # Still not due after heal assert "half-paused-1" not in {j["id"] for j in get_due_jobs()} def test_resume_rejects_past_oneshot(self, tmp_cron_dir, monkeypatch): """Resuming a paused one-shot whose time is now in the past must raise ValueError — the revived job would silently never fire.""" now = datetime(2026, 7, 6, 12, 0, 0, tzinfo=timezone.utc) monkeypatch.setattr("cron.jobs._hermes_now", lambda: now) # Create directly — bypass create_job's past-oneshot guard so we can # test the resume path independently. job = { "id": "test-resume-past", "name": "test-resume-past", "prompt": "Past one-shot", "schedule": {"kind": "once", "run_at": (now - timedelta(minutes=5)).isoformat(), "display": "once"}, "repeat": {"times": 1, "completed": 0}, "enabled": False, "state": "paused", "paused_at": now.isoformat(), "paused_reason": "test", "next_run_at": None, "last_run_at": None, "last_status": None, "last_error": None, "last_delivery_error": None, "created_at": (now - timedelta(hours=1)).isoformat(), "deliver": "local", } save_jobs([job]) with pytest.raises(ValueError, match="in the past"): resume_job("test-resume-past") class TestResolveJobRef: """Name-based job lookup for CLI/tool callers (PR #2627, @buntingszn).""" def test_resolve_by_exact_id(self, tmp_cron_dir): from cron.jobs import resolve_job_ref job = create_job(prompt="A", schedule="1h", name="alpha") assert resolve_job_ref(job["id"])["id"] == job["id"] def test_mutations_refuse_ambiguous_name(self, tmp_cron_dir): """pause/resume/trigger/remove must refuse to act on an ambiguous name.""" from cron.jobs import AmbiguousJobReference, trigger_job create_job(prompt="A", schedule="1h", name="dup") create_job(prompt="B", schedule="1h", name="dup") for fn in (pause_job, resume_job, trigger_job): with pytest.raises(AmbiguousJobReference): fn("dup") with pytest.raises(AmbiguousJobReference): remove_job("dup") class TestMarkJobRun: def test_increments_completed(self, tmp_cron_dir): job = create_job(prompt="Test", schedule="every 1h") mark_job_run(job["id"], success=True) updated = get_job(job["id"]) assert updated["repeat"]["completed"] == 1 assert updated["last_status"] == "ok" def test_repeat_limit_retains_completed_record(self, tmp_cron_dir): """A finished one-shot must stay inspectable, not vanish from the store.""" job = create_job(prompt="Once", schedule="in 30m", repeat=1) mark_job_run(job["id"], success=True) updated = get_job(job["id"]) assert updated is not None, "completed one-shot was deleted from jobs.json" assert updated["state"] == "completed" assert updated["enabled"] is False assert updated["next_run_at"] is None assert updated["last_status"] == "ok" def test_repeat_limit_retains_delivery_error(self, tmp_cron_dir): """A one-shot whose delivery failed must keep the error on its record.""" job = create_job(prompt="Once", schedule="in 30m", repeat=1) mark_job_run( job["id"], success=True, delivery_error="platform 'telegram' not configured", ) updated = get_job(job["id"]) assert updated is not None assert updated["state"] == "completed" assert updated["last_delivery_error"] == "platform 'telegram' not configured" # A terminal completion that never reached the user is not a success. assert updated["last_status"] == "delivery_failed" def test_completed_oneshot_visible_in_list(self, tmp_cron_dir): """list_jobs(include_disabled=True) surfaces the completed record.""" job = create_job(prompt="Once", schedule="in 30m", repeat=1) mark_job_run(job["id"], success=True, delivery_error="send failed: 502") listed = {j["id"]: j for j in list_jobs(include_disabled=True)} assert job["id"] in listed assert listed[job["id"]]["state"] == "completed" assert listed[job["id"]]["last_delivery_error"] == "send failed: 502" assert listed[job["id"]]["last_status"] == "delivery_failed" # Default (enabled-only) listing hides it, matching paused/disabled jobs. assert job["id"] not in {j["id"] for j in list_jobs()} def test_completed_oneshot_not_due(self, tmp_cron_dir): """A retained completed one-shot must never be dispatched again.""" job = create_job(prompt="Once", schedule="in 30m", repeat=1) mark_job_run(job["id"], success=True) assert job["id"] not in {j["id"] for j in get_due_jobs()} def test_error_status(self, tmp_cron_dir): job = create_job(prompt="Fail", schedule="every 1h") mark_job_run(job["id"], success=False, error="timeout") updated = get_job(job["id"]) assert updated["last_status"] == "error" assert updated["last_error"] == "timeout" def test_delivery_error_tracked_separately(self, tmp_cron_dir): """Agent succeeds but delivery fails — surfaced, not hidden behind ok. Regression guard for #83993: recording ``last_status="ok"`` made a run the user never received look like a quiet success everywhere that keys off "ok". The agent error stays independent of the delivery error, and the delivery failure is not an agent failure (no streak). """ job = create_job(prompt="Report", schedule="every 1h") mark_job_run(job["id"], success=True, delivery_error="send failed: 502") updated = get_job(job["id"]) assert updated["last_status"] == "delivery_failed" assert updated["last_error"] is None assert updated["last_delivery_error"] == "send failed: 502" assert updated["failure_streak"] == 0 def test_success_without_delivery_error_stays_ok(self, tmp_cron_dir): """A fully successful run is still plain "ok".""" job = create_job(prompt="Report", schedule="every 1h") mark_job_run(job["id"], success=True) assert get_job(job["id"])["last_status"] == "ok" # An empty delivery error is no error at all. mark_job_run(job["id"], success=True, delivery_error="") assert get_job(job["id"])["last_status"] == "ok" def test_agent_failure_still_error_with_delivery_error(self, tmp_cron_dir): """An agent failure outranks delivery: still "error", still a streak.""" job = create_job(prompt="Report", schedule="every 1h") mark_job_run( job["id"], success=False, error="timeout", delivery_error="send failed: 502", ) updated = get_job(job["id"]) assert updated["last_status"] == "error" assert updated["last_error"] == "timeout" assert updated["failure_streak"] == 1 def test_explicit_status_override_wins_over_delivery_failed(self, tmp_cron_dir): """An explicit terminal status (T1-26 blocked_config) still wins.""" job = create_job(prompt="Report", schedule="every 1h") mark_job_run( job["id"], success=True, delivery_error="send failed: 502", status="blocked_config", ) updated = get_job(job["id"]) assert updated["last_status"] == "blocked_config" assert updated["last_delivery_error"] == "send failed: 502" def test_failure_streak_increments_and_resets(self, tmp_cron_dir): """failure_streak counts consecutive agent failures; success resets.""" job = create_job(prompt="Flaky", schedule="every 1h") assert get_job(job["id"])["failure_streak"] == 0 mark_job_run(job["id"], success=False, error="timeout") mark_job_run(job["id"], success=False, error="timeout") assert get_job(job["id"])["failure_streak"] == 2 mark_job_run(job["id"], success=True) assert get_job(job["id"])["failure_streak"] == 0 def test_failure_streak_ignores_delivery_errors(self, tmp_cron_dir): """A successful run with a delivery error must not count as a failure.""" job = create_job(prompt="Report", schedule="every 1h") mark_job_run(job["id"], success=False, error="timeout") mark_job_run(job["id"], success=True, delivery_error="send failed: 502") assert get_job(job["id"])["failure_streak"] == 0 def test_failure_streak_backcompat_missing_field(self, tmp_cron_dir): """Jobs persisted before the field existed increment from 0.""" job = create_job(prompt="Old", schedule="every 1h") # Simulate a pre-field record on disk. jobs = load_jobs() for j in jobs: j.pop("failure_streak", None) save_jobs(jobs) mark_job_run(job["id"], success=False, error="boom") assert get_job(job["id"])["failure_streak"] == 1 def test_recurring_cron_not_disabled_when_croniter_missing(self, tmp_cron_dir, monkeypatch): """Regression test for issue #16265. If the gateway runs in an env where `croniter` went missing after a recurring cron job was persisted, `compute_next_run()` returns None. `mark_job_run()` must NOT treat that as terminal completion — the job has to stay enabled with state=error so the user notices, rather than silently flipping to enabled=false, state=completed. """ pytest.importorskip("croniter") # need it to create the job job = create_job(prompt="Recurring", schedule="0 7,15,23 * * *") assert job["schedule"]["kind"] == "cron" # Simulate the runtime env having lost croniter between job creation # and this run. monkeypatch.setattr("cron.jobs.HAS_CRONITER", False) mark_job_run(job["id"], success=True) updated = get_job(job["id"]) assert updated is not None, "recurring cron job was deleted" assert updated["enabled"] is True, ( "recurring cron job was disabled despite croniter-missing being " "a runtime dep issue, not a terminal completion" ) assert updated["state"] == "error" assert updated["state"] != "completed" assert updated["next_run_at"] is None assert updated["last_error"] assert "croniter" in updated["last_error"].lower() class TestAdvanceNextRun: """Tests for advance_next_run() — crash-safety for recurring jobs.""" def test_advances_interval_job(self, tmp_cron_dir): """Interval jobs should have next_run_at bumped to the next future occurrence.""" job = create_job(prompt="Recurring check", schedule="every 1h") # Force next_run_at to 5 minutes ago (i.e. the job is due) jobs = load_jobs() old_next = (datetime.now() - timedelta(minutes=5)).isoformat() jobs[0]["next_run_at"] = old_next save_jobs(jobs) result = advance_next_run(job["id"]) assert result is True updated = get_job(job["id"]) from cron.jobs import _ensure_aware, _hermes_now new_next_dt = _ensure_aware(datetime.fromisoformat(updated["next_run_at"])) assert new_next_dt > _hermes_now(), "next_run_at should be in the future after advance" def test_skips_oneshot_job(self, tmp_cron_dir): """One-shot jobs should NOT be advanced — they need to retry on restart.""" job = create_job(prompt="Run once", schedule="in 30m") original_next = get_job(job["id"])["next_run_at"] result = advance_next_run(job["id"]) assert result is False updated = get_job(job["id"]) assert updated["next_run_at"] == original_next, "one-shot next_run_at should be unchanged" def test_crash_safety_scenario(self, tmp_cron_dir): """Simulate the crash-loop scenario: after advance, the job should NOT be due.""" job = create_job(prompt="Crash test", schedule="every 1h") # Force next_run_at to 5 minutes ago (job is due) jobs = load_jobs() jobs[0]["next_run_at"] = (datetime.now() - timedelta(minutes=5)).isoformat() save_jobs(jobs) # Job should be due before advance due_before = get_due_jobs() assert len(due_before) == 1 # Advance (simulating what tick() does before run_job) advance_next_run(job["id"]) # Now the job should NOT be due (simulates restart after crash) due_after = get_due_jobs() assert len(due_after) == 0, "Job should not be due after advance_next_run" class TestGetDueJobs: def test_past_due_within_window_returned(self, tmp_cron_dir): """Jobs within the dynamic grace window are still considered due (not stale). For an hourly job, grace = 30 min (half the period, clamped to [120s, 2h]). """ job = create_job(prompt="Due now", schedule="every 1h") # Force next_run_at to 10 minutes ago (within the 30-min grace for hourly) jobs = load_jobs() jobs[0]["next_run_at"] = (datetime.now() - timedelta(minutes=10)).isoformat() save_jobs(jobs) due = get_due_jobs() assert len(due) == 1 assert due[0]["id"] == job["id"] def test_stale_past_due_runs_once_and_fast_forwards(self, tmp_cron_dir): """Recurring jobs past their grace window run once now and fast-forward next_run_at. For an hourly job, grace = 30 min. Setting 35 min late exceeds the window. The job should be returned as due (execute once) with next_run_at in the future. """ job = create_job(prompt="Stale", schedule="every 1h") # Force next_run_at to 35 minutes ago (beyond the 30-min grace for hourly) jobs = load_jobs() jobs[0]["next_run_at"] = (datetime.now() - timedelta(minutes=35)).isoformat() save_jobs(jobs) due = get_due_jobs() # Job is returned as due — execute once now instead of skipping assert len(due) == 1 assert due[0]["id"] == job["id"] # next_run_at should be fast-forwarded to the future (accumulated slots skipped) updated = get_job(job["id"]) from cron.jobs import _ensure_aware, _hermes_now next_dt = _ensure_aware(datetime.fromisoformat(updated["next_run_at"])) assert next_dt > _hermes_now() def test_idless_job_does_not_crash_or_block_sibling_jobs(self, tmp_cron_dir): """A job missing its 'id' key must not crash the tick or freeze siblings. Regression: jobs authored by a direct jobs.json edit (bypassing create_job) sometimes used the key 'job_id' instead of 'id'. The logging helpers evaluated ``job.get("name", job["id"])`` -- Python evaluates the default argument ``job["id"]`` eagerly, so an id-less job raised ``KeyError: 'id'`` mid-tick. That exception aborted ``_get_due_jobs_locked()`` BEFORE ``save_jobs()`` ran, so every healthy job's fast-forwarded next_run_at was computed in memory then discarded -- the whole profile's scheduler froze in a per-minute loop. """ healthy = create_job(prompt="Healthy", schedule="every 1h") jobs = load_jobs() # Push the healthy job beyond its grace window so the fast-forward path # (one of the id-less-crash sites) runs. jobs[0]["next_run_at"] = (datetime.now() - timedelta(minutes=35)).isoformat() # A malformed record: no 'id' key, mirroring the real corruption. jobs.append({ "name": "idless-job", "schedule": {"kind": "cron", "expr": "0 4 * * *"}, "enabled": True, "no_agent": True, "next_run_at": None, }) save_jobs(jobs) # Must not raise KeyError. due = get_due_jobs() # The healthy sibling is still discovered despite the malformed neighbor. assert any(d.get("id") == healthy["id"] for d in due) def test_long_execution_does_not_perpetually_defer(self, tmp_cron_dir, monkeypatch): """#33315: a recurring job whose runtime exceeds interval+grace must still run once when the tick comes back, not skip forever. Reproduces the production loop: a 5-min interval job whose previous run overran the interval, leaving next_run_at ~11 min in the past — beyond the 150s grace for a 5m interval. The job must be returned as due (run once) AND have next_run_at fast-forwarded (so accumulated missed slots don't all fire).""" from cron.jobs import _ensure_aware, _hermes_now job = create_job(prompt="Long job", schedule="every 5m") jobs = load_jobs() # 11 minutes ago: > grace (150s for a 5m interval) — the "still running" miss. stale = (_hermes_now() - timedelta(minutes=11)).isoformat() jobs[0]["next_run_at"] = stale jobs[0]["last_run_at"] = (_hermes_now() - timedelta(minutes=1)).isoformat() save_jobs(jobs) due = get_due_jobs() assert [j["id"] for j in due] == [job["id"]], "long-execution job was skipped (perpetual-defer bug)" # next_run_at fast-forwarded into the future (no burst of missed slots). nxt = _ensure_aware(datetime.fromisoformat(get_job(job["id"])["next_run_at"])) assert nxt > _hermes_now() def test_broken_recent_one_shot_without_next_run_is_recovered(self, tmp_cron_dir, monkeypatch): now = datetime(2026, 3, 18, 4, 22, 30, tzinfo=timezone.utc) monkeypatch.setattr("cron.jobs._hermes_now", lambda: now) run_at = "2026-03-18T04:22:00+00:00" save_jobs( [{ "id": "oneshot-recover", "name": "Recover me", "prompt": "Word of the day", "schedule": {"kind": "once", "run_at": run_at, "display": "once at 2026-03-18 04:22"}, "schedule_display": "once at 2026-03-18 04:22", "repeat": {"times": 1, "completed": 0}, "enabled": True, "state": "scheduled", "paused_at": None, "paused_reason": None, "created_at": "2026-03-18T04:21:00+00:00", "next_run_at": None, "last_run_at": None, "last_status": None, "last_error": None, "deliver": "local", "origin": None, }] ) due = get_due_jobs() assert [job["id"] for job in due] == ["oneshot-recover"] # Recovery restores next_run_at to the original run time; the # cross-process double-exec guard (#59229) is a separate run_claim # stamped under the lock, not a next_run_at mutation. recovered = get_job("oneshot-recover") assert recovered["next_run_at"] == run_at assert recovered.get("run_claim") is not None assert recovered["run_claim"]["at"] == now.isoformat() def test_one_shot_not_redispatched_while_running(self, tmp_cron_dir, monkeypatch): """#59229: two concurrent schedulers must not double-execute a one-shot. Reproduces the reported failure with a job whose run OUTLIVES the tick interval (a ~2.5-min research prompt). Process A's tick returns it as due and stamps a run_claim; while A is still running, every later tick (process B, or A's own next tick) must see the fresh claim and skip — not just for one tick window but for the whole run. """ from cron.jobs import _hermes_now t0 = _hermes_now() run_at = (t0 - timedelta(seconds=5)).isoformat() save_jobs([{ "id": "long-oneshot", "name": "R", "prompt": "2.5min research", "schedule": {"kind": "once", "run_at": run_at}, "next_run_at": run_at, "enabled": True, "state": "scheduled", }]) # Process A tick: picks it up + claims it. dueA = get_due_jobs() assert [j["id"] for j in dueA] == ["long-oneshot"] assert get_job("long-oneshot").get("run_claim") is not None # Process B (and A's own subsequent ticks) while A is still running: # 28s later (the exact gap in the report) AND 61s later (past any # fixed +60s window) — both must skip. for gap in (28, 61, 130): monkeypatch.setattr("cron.jobs._hermes_now", lambda t0=t0, g=gap: t0 + timedelta(seconds=g)) assert get_due_jobs() == [], f"double-dispatched at +{gap}s" def test_run_claim_heartbeat_keeps_long_run_claimed_past_ttl( self, tmp_cron_dir, monkeypatch ): """#62002 cross-process leg: a heartbeat-refreshed claim never expires while the run is alive, so no other tick re-dispatches or stale-removes the job even when the run outlives the original TTL horizon.""" monkeypatch.delenv("HERMES_CRON_TIMEOUT", raising=False) from cron.jobs import _hermes_now, _oneshot_run_claim_ttl_seconds ttl = _oneshot_run_claim_ttl_seconds() t0 = _hermes_now() run_at = (t0 - timedelta(seconds=5)).isoformat() save_jobs([{ "id": "slowrun", "name": "R", "prompt": "x", "schedule": {"kind": "once", "run_at": run_at}, "next_run_at": run_at, "enabled": True, "state": "scheduled", "repeat": {"times": 1, "completed": 0}, }]) # Tick claims + dispatches the job. assert [j["id"] for j in get_due_jobs()] == ["slowrun"] assert claim_dispatch("slowrun") is True # Mid-run heartbeat before the TTL horizon refreshes the claim. monkeypatch.setattr("cron.jobs._hermes_now", lambda: t0 + timedelta(seconds=ttl - 60)) owner = get_job("slowrun")["run_claim"]["by"] assert heartbeat_run_claim("slowrun", expected_owner=owner) is True # Past the ORIGINAL claim's TTL horizon: without the heartbeat this # tick would stale-remove the maxed one-shot; with it the claim is # fresh, so the job is skipped and the record survives. monkeypatch.setattr("cron.jobs._hermes_now", lambda: t0 + timedelta(seconds=ttl + 10)) assert get_due_jobs() == [] assert get_job("slowrun") is not None # Run completes → outcome lands on a record that still exists # (times=1 reached, so mark_job_run retires the job as a terminal # completed record instead of deleting it). mark_job_run("slowrun", True) retired = get_job("slowrun") assert retired is not None assert retired["state"] == "completed" assert retired["enabled"] is False assert retired["last_status"] == "ok" def test_heartbeat_run_claim_rejects_replaced_owner(self, tmp_cron_dir): """A resumed stale runner must not keep a newer owner's claim alive.""" future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat() original_at = datetime.now(timezone.utc).isoformat() save_jobs([{ "id": "reclaimed", "name": "R", "prompt": "x", "schedule": {"kind": "once", "run_at": future}, "next_run_at": future, "enabled": True, "state": "scheduled", "run_claim": {"at": original_at, "by": "new-owner"}, }]) assert heartbeat_run_claim("reclaimed", expected_owner="old-owner") is False assert get_job("reclaimed")["run_claim"] == { "at": original_at, "by": "new-owner", } class TestEnabledToolsets: def test_enabled_toolsets_stored(self, tmp_cron_dir): job = create_job(prompt="monitor", schedule="every 1h", enabled_toolsets=["web", "terminal"]) assert job["enabled_toolsets"] == ["web", "terminal"] class TestMarkJobRunConcurrency: """Regression tests for concurrent parallel job state writes. tick() dispatches multiple jobs to separate threads simultaneously. Without _jobs_file_lock protecting the load→modify→save cycle in mark_job_run(), concurrent writes can clobber each other's updates (last-writer-wins), leaving some jobs with stale last_status / last_run_at. """ def test_three_concurrent_mark_job_run_no_overwrites(self, tmp_cron_dir): """Run mark_job_run() for 3 jobs in parallel threads; all must land correctly.""" # Create 3 distinct recurring jobs job_a = create_job(prompt="Job A", schedule="every 1h") job_b = create_job(prompt="Job B", schedule="every 1h") job_c = create_job(prompt="Job C", schedule="every 1h") errors: list = [] def run_mark(job_id: str, success: bool, error_msg=None): try: mark_job_run(job_id, success=success, error=error_msg) except Exception as exc: # pragma: no cover errors.append(exc) # Fire all three concurrently threads = [ threading.Thread(target=run_mark, args=(job_a["id"], True)), threading.Thread(target=run_mark, args=(job_b["id"], False, "timeout")), threading.Thread(target=run_mark, args=(job_c["id"], True)), ] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Unexpected exceptions in worker threads: {errors}" # Verify each job has the correct state — no overwrites a = get_job(job_a["id"]) b = get_job(job_b["id"]) c = get_job(job_c["id"]) assert a is not None, "Job A was unexpectedly deleted" assert b is not None, "Job B was unexpectedly deleted" assert c is not None, "Job C was unexpectedly deleted" assert a["last_status"] == "ok", f"Job A last_status wrong: {a['last_status']}" assert a["last_run_at"] is not None, "Job A last_run_at not set" assert a["repeat"]["completed"] == 1, f"Job A completed count wrong: {a['repeat']['completed']}" assert b["last_status"] == "error", f"Job B last_status wrong: {b['last_status']}" assert b["last_error"] == "timeout", f"Job B last_error wrong: {b['last_error']}" assert b["last_run_at"] is not None, "Job B last_run_at not set" assert b["repeat"]["completed"] == 1, f"Job B completed count wrong: {b['repeat']['completed']}" assert c["last_status"] == "ok", f"Job C last_status wrong: {c['last_status']}" assert c["last_run_at"] is not None, "Job C last_run_at not set" assert c["repeat"]["completed"] == 1, f"Job C completed count wrong: {c['repeat']['completed']}" class TestBadNextRunAtRecovery: """Regression: malformed next_run_at must not crash the due scan or starve siblings. Mirrors the id-less and non-dict-schedule patterns: a single bad persisted record in jobs.json must not abort _get_due_jobs_locked before save. """ def test_bad_next_run_at_does_not_crash_or_block_sibling_jobs(self, tmp_cron_dir): """One job with unparseable next_run_at + one healthy due sibling. get_due_jobs must succeed and return the healthy job; the bad record must be repaired (next_run_at cleared so recovery can set a sane value). """ from datetime import timezone, timedelta as td now = datetime.now(timezone.utc) past = (now - td(seconds=30)).isoformat() future = (now + td(days=1)).isoformat() # Bad record: next_run_at is not a valid ISO string (e.g. from hand-edit or corruption) # Healthy sibling is past due with good schedule. bad_job = { "id": "bad-next", "schedule": {"kind": "interval", "minutes": 60}, "next_run_at": "not-a-valid-iso-timestamp!!!", "enabled": True, "created_at": past, } good_job = { "id": "good-sibling", "schedule": {"kind": "interval", "minutes": 5}, "next_run_at": past, "enabled": True, "created_at": past, } save_jobs([bad_job, good_job]) # Must not raise due = get_due_jobs() # The healthy job must still be returned ids = [j["id"] for j in due] assert "good-sibling" in ids, f"healthy sibling missing from due jobs: {ids}" assert "bad-next" not in ids # bad one may be repaired and/or not yet due after repair # Bad job should have been auto-repaired (next_run_at stripped or fixed) repaired = get_job("bad-next") assert repaired is not None nr = repaired.get("next_run_at") if nr is not None: # If still present it must now be parseable datetime.fromisoformat(nr) # Calling again must remain stable (no crash on re-scan) due2 = get_due_jobs() assert any(j["id"] == "good-sibling" for j in due2) class TestPerJobScanContainment: """Structural guard: ANY per-job exception in the due scan must degrade to skipping that one job for the tick — never abort the scan and starve healthy siblings (the freeze class behind bad id / schedule / next_run_at). """ def test_unforeseen_per_job_exception_does_not_starve_siblings(self, tmp_cron_dir): """Simulate a FUTURE malformed-field variant none of the shape normalizers repair, by making grace computation raise for one job only. The per-job guard must skip it and still return the sibling.""" from datetime import timezone, timedelta as td from unittest.mock import patch as mock_patch now = datetime.now(timezone.utc) past = (now - td(seconds=30)).isoformat() poison = { "id": "poison", # minutes=7 tags this schedule so the patched helper can target it "schedule": {"kind": "interval", "minutes": 7}, "next_run_at": past, "enabled": True, "created_at": past, } good = { "id": "good-sibling", "schedule": {"kind": "interval", "minutes": 5}, "next_run_at": past, "enabled": True, "created_at": past, } save_jobs([poison, good]) import cron.jobs as jobs_mod real_grace = jobs_mod._compute_grace_seconds def selective_grace(schedule): if schedule.get("minutes") == 7: raise RuntimeError("simulated unforeseen malformed field") return real_grace(schedule) with mock_patch.object(jobs_mod, "_compute_grace_seconds", selective_grace): due = get_due_jobs() # must not raise ids = [j["id"] for j in due] assert "good-sibling" in ids, f"healthy sibling starved: {ids}" assert "poison" not in ids # Scheduler stays alive on subsequent ticks too. with mock_patch.object(jobs_mod, "_compute_grace_seconds", selective_grace): due2 = get_due_jobs() assert any(j["id"] == "good-sibling" for j in due2) class TestSaveJobOutput: def test_creates_output_file(self, tmp_cron_dir): output_file = save_job_output("test123", "# Results\nEverything ok.") assert output_file.exists() assert output_file.read_text() == "# Results\nEverything ok." assert "test123" in str(output_file) class TestCronOutputRetention: """Per-run cron output must self-prune so long deploys don't fill the disk (#52383).""" @staticmethod def _seed(d, count): d.mkdir(parents=True, exist_ok=True) names = [f"2026-06-25_10-00-{i:02d}.md" for i in range(count)] for n in names: (d / n).write_text("x", encoding="utf-8") return names def test_prune_keeps_newest_n(self, tmp_path): from cron.jobs import _prune_job_output d = tmp_path / "job" names = self._seed(d, 10) assert _prune_job_output(d, keep=3) == 7 assert sorted(p.name for p in d.glob("*.md")) == names[-3:] # ========================================================================= # claim_dispatch — pre-run one-shot crash safety (issue #38758) # ========================================================================= class TestClaimDispatch: """One-shot jobs must commit their dispatch BEFORE the side effect runs, so a tick that dies mid-execution (gateway kill, OOM, hard-timeout) can re-fire the job at most ``repeat.times`` times instead of infinitely.""" def _oneshot(self, times=1, completed=0): return { "id": "os1", "name": "one-shot", "enabled": True, "schedule": {"kind": "once", "run_at": "2026-01-01T00:00:00+00:00"}, "repeat": {"times": times, "completed": completed}, } def test_claim_increments_and_persists(self, tmp_cron_dir): save_jobs([self._oneshot(times=1, completed=0)]) assert claim_dispatch("os1") is True # Persisted BEFORE any side effect — survives a crash. assert load_jobs()[0]["repeat"]["completed"] == 1 def test_already_dispatched_oneshot_is_removed(self, tmp_cron_dir): # A prior tick claimed (completed==times) then died before mark_job_run # could remove the job. The next claim must refuse AND clean up. save_jobs([self._oneshot(times=1, completed=1)]) assert claim_dispatch("os1") is False assert load_jobs() == [] # removed, will not re-fire def test_mark_job_run_does_not_double_count_preclaimed_oneshot(self, tmp_cron_dir): # Full lifecycle: claim bumps completed to times, then mark_job_run must # NOT increment again — it recognizes the pre-claim and retires the job # as a terminal completed record (retained for inspection, not re-fired). save_jobs([self._oneshot(times=1, completed=0)]) assert claim_dispatch("os1") is True assert load_jobs()[0]["repeat"]["completed"] == 1 mark_job_run("os1", success=True) retired = load_jobs() assert len(retired) == 1 # completed once, retired — not fired twice assert retired[0]["repeat"]["completed"] == 1 # no double count assert retired[0]["state"] == "completed" assert retired[0]["enabled"] is False def test_get_due_jobs_removes_stale_maxed_oneshot(self, tmp_cron_dir): # A claimed one-shot whose tick died leaves completed>=times with # last_run_at still unset, so the recovery helper re-arms it as due. # get_due_jobs must drop it instead of returning it for another fire. past = (datetime.now(timezone.utc) - timedelta(seconds=5)).isoformat() save_jobs([{ "id": "os1", "name": "one-shot", "enabled": True, "schedule": {"kind": "once", "run_at": past}, "repeat": {"times": 1, "completed": 1}, "next_run_at": None, }]) due = get_due_jobs() assert due == [] assert load_jobs() == [] # cleaned up class TestLateEnvRepointScopesStore: """A HERMES_HOME set AFTER cron.jobs import must scope the store even without use_cron_store(): fixtures that patch the environment too late previously read/wrote the import-time jobs.json — the user's real file.""" def test_late_env_repoint_scopes_store(self, tmp_path, monkeypatch): import cron.jobs as jobs monkeypatch.setenv("HERMES_HOME", str(tmp_path)) store = jobs._current_cron_store() expected = tmp_path.resolve() / "cron" assert store.cron_dir == expected assert store.jobs_file == expected / "jobs.json" assert store.output_dir == expected / "output" # the import-time compatibility constants are untouched assert jobs.JOBS_FILE != store.jobs_file def test_use_cron_store_override_still_wins(self, tmp_path, monkeypatch): import cron.jobs as jobs monkeypatch.setenv("HERMES_HOME", str(tmp_path / "env-home")) with jobs.use_cron_store(tmp_path / "override-home"): store = jobs._current_cron_store() assert store.jobs_file == (tmp_path / "override-home").resolve() / "cron" / "jobs.json" def test_heartbeat_does_not_recreate_deleted_named_profile(self, tmp_path): import cron.jobs as jobs profiles_dir = tmp_path / "profiles" profiles_dir.mkdir() deleted_home = profiles_dir / "deleted" with jobs.use_cron_store(deleted_home): jobs.record_ticker_heartbeat() assert not deleted_home.exists() def test_heartbeat_initializes_existing_named_profile(self, tmp_path): import cron.jobs as jobs profile_home = tmp_path / "profiles" / "active" profile_home.mkdir(parents=True) with jobs.use_cron_store(profile_home): jobs.record_ticker_heartbeat() assert (profile_home / "cron" / "ticker_heartbeat").is_file() def test_public_io_after_late_env_repoint_leaves_old_file_untouched( self, tmp_path, monkeypatch ): """The public API, not the store internals: save_jobs()/load_jobs() called after a post-import HERMES_HOME repoint must operate on the NEW home's jobs.json and leave the import-time file byte-identical. The "import-time home" is SIMULATED at a tmp location by patching the module constants and the import-time snapshot together (so they still compare equal and the deliberate-repoint branch does not fire). The test must never touch the real import-time jobs.json: if this module was first imported before the suite's env isolation applied, that path IS the developer's live file — writing a sentinel there is exactly the incident this PR exists to prevent.""" import cron.jobs as jobs sim_old_home = tmp_path / "import-time-home" sim_cron = sim_old_home / "cron" monkeypatch.setattr(jobs, "HERMES_DIR", sim_old_home) monkeypatch.setattr(jobs, "CRON_DIR", sim_cron) monkeypatch.setattr(jobs, "JOBS_FILE", sim_cron / "jobs.json") monkeypatch.setattr(jobs, "OUTPUT_DIR", sim_cron / "output") monkeypatch.setattr( jobs, "_IMPORT_STORE", jobs._CronStorePaths(jobs.CRON_DIR, jobs.JOBS_FILE, jobs.OUTPUT_DIR), ) # Plant a sentinel at the (simulated) import-time location — the file # a late-patching fixture used to clobber. old_file = jobs.JOBS_FILE old_file.parent.mkdir(parents=True, exist_ok=True) sentinel = '[{"id": "sentinel-do-not-touch"}]' old_file.write_text(sentinel, encoding="utf-8") new_home = tmp_path / "late-home" monkeypatch.setenv("HERMES_HOME", str(new_home)) job = { "id": "lateenvjob01", "name": "late-env", "prompt": None, "schedule_display": None, "schedule": {"kind": "interval", "minutes": 60, "display": "every 60m"}, "enabled": True, } save_jobs([job]) # public read round-trips from the NEW home... loaded = load_jobs() assert [j["id"] for j in loaded] == ["lateenvjob01"] new_file = new_home.resolve() / "cron" / "jobs.json" assert new_file.is_file() # ...and the import-time file is byte-identical to the sentinel. assert old_file.read_text(encoding="utf-8") == sentinel # ========================================================================= # UTF-8 BOM on jobs.json (Windows Notepad / PowerShell 5.1) # ========================================================================= class TestJobsJsonShapes: def test_load_jobs_normalizes_id_keyed_jobs_mapping(self, tmp_cron_dir): import json from cron.jobs import JOBS_FILE job_a = { "id": "cron1234abcd", "name": "daily briefing", "enabled": True, "prompt": "Summarize overnight incidents", "schedule": {"kind": "interval", "minutes": 1440, "display": "every 24h"}, } job_b = { "id": "cron5678efgh", "name": "disabled cleanup", "enabled": False, "prompt": "Clean stale scratch files", "schedule": {"kind": "once", "run_at": "2030-01-15T14:00:00+00:00"}, } payload = { "jobs": { job_a["id"]: job_a, job_b["id"]: job_b, }, "updated_at": "2026-08-23T00:00:00+00:00", } JOBS_FILE.parent.mkdir(parents=True, exist_ok=True) JOBS_FILE.write_text(json.dumps(payload), encoding="utf-8") loaded = load_jobs() assert isinstance(loaded, list) assert {job["id"] for job in loaded} == {job_a["id"], job_b["id"]} listed = {job["id"]: job for job in list_jobs(include_disabled=True)} assert set(listed) == {job_a["id"], job_b["id"]} for expected in (job_a, job_b): actual = listed[expected["id"]] assert actual["id"] == expected["id"] assert actual["name"] == expected["name"] assert actual["prompt"] == expected["prompt"] assert actual["schedule"] == expected["schedule"] assert actual["enabled"] is expected["enabled"] class TestJobsJsonUtf8Bom: """jobs.json readers must accept a leading UTF-8 BOM. Matching the env-class dialect (utf-8-sig): a BOM from Windows editors must not raise JSONDecodeError / RuntimeError on load_jobs(). """ def test_load_jobs_accepts_utf8_bom(self, tmp_cron_dir): """BOM'd jobs.json loads — the pre-fix crash repro.""" import json from pathlib import Path from cron.jobs import JOBS_FILE, load_jobs payload = { "jobs": [ { "id": "bomjob01", "name": "bom-test", "enabled": True, "prompt": "hello", "schedule": {"kind": "interval", "minutes": 60, "display": "every 60m"}, } ] } JOBS_FILE.parent.mkdir(parents=True, exist_ok=True) JOBS_FILE.write_bytes( b"\xef\xbb\xbf" + json.dumps(payload).encode("utf-8") ) loaded = load_jobs() assert [j["id"] for j in loaded] == ["bomjob01"] assert loaded[0]["name"] == "bom-test" def test_load_jobs_bomless_regression(self, tmp_cron_dir): """BOM-less UTF-8 jobs.json must keep loading after utf-8-sig.""" import json from cron.jobs import JOBS_FILE, load_jobs payload = { "jobs": [ { "id": "plainjob01", "name": "plain", "enabled": True, "prompt": "hi", "schedule": {"kind": "interval", "minutes": 30, "display": "every 30m"}, } ] } JOBS_FILE.parent.mkdir(parents=True, exist_ok=True) JOBS_FILE.write_text(json.dumps(payload), encoding="utf-8") loaded = load_jobs() assert [j["id"] for j in loaded] == ["plainjob01"] # ========================================================================= # ID-keyed jobs map on jobs.json (external tools / hand edits) — #92935 # ========================================================================= class TestJobsJsonIdKeyedMap: """load_jobs() must flatten an ID-keyed ``jobs`` map to the list contract. A store written as ``{"jobs": {"": {...}, ...}}`` (external tool or hand edit — Hermes' own save_jobs() only ever writes a list) made load_jobs() return a dict. Every consumer iterates it as a list, so ``list_jobs()`` → ``_normalize_job_record`` → ``dict()`` raised ``ValueError: dictionary update sequence element #0 has length 1; 2 is required`` and took down ``hermes cron list``, the ``cronjob(action= "list")`` tool, and the Dashboard cron view. The values already carry their own ``id`` matching the map key, so flattening is lossless. """ _ID_KEYED = { "jobs": { "cron1234abcd": { "id": "cron1234abcd", "name": "Example job", "enabled": True, "prompt": "do a thing", "schedule": {"kind": "interval", "minutes": 60, "display": "every 60m"}, }, "cron5678efgh": { "id": "cron5678efgh", "name": "Second job", "enabled": True, "prompt": "do another", "schedule": {"kind": "interval", "minutes": 30, "display": "every 30m"}, }, }, "updated_at": "2026-08-23T10:10:12+08:00", } def test_load_jobs_flattens_id_keyed_map(self, tmp_cron_dir): """The pre-fix repro: load_jobs() returns a list, not the raw dict.""" import json from cron.jobs import JOBS_FILE, load_jobs JOBS_FILE.parent.mkdir(parents=True, exist_ok=True) JOBS_FILE.write_text(json.dumps(self._ID_KEYED), encoding="utf-8") loaded = load_jobs() assert isinstance(loaded, list) assert {j["id"] for j in loaded} == {"cron1234abcd", "cron5678efgh"} assert all(isinstance(j, dict) for j in loaded) def test_list_jobs_survives_id_keyed_map(self, tmp_cron_dir): """The reported traceback path (hermes cron list / cronjob list tool).""" import json from cron.jobs import JOBS_FILE, list_jobs JOBS_FILE.parent.mkdir(parents=True, exist_ok=True) JOBS_FILE.write_text(json.dumps(self._ID_KEYED), encoding="utf-8") # Pre-fix this raised ValueError from _normalize_job_record(dict()). jobs = list_jobs(include_disabled=True) assert {j["id"] for j in jobs} == {"cron1234abcd", "cron5678efgh"} def test_id_keyed_map_repaired_to_list_on_disk(self, tmp_cron_dir): """Loading rewrites the store into the canonical {"jobs": [...]} form.""" import json from cron.jobs import JOBS_FILE, load_jobs JOBS_FILE.parent.mkdir(parents=True, exist_ok=True) JOBS_FILE.write_text(json.dumps(self._ID_KEYED), encoding="utf-8") load_jobs() on_disk = json.loads(JOBS_FILE.read_text(encoding="utf-8")) assert isinstance(on_disk["jobs"], list) assert {j["id"] for j in on_disk["jobs"]} == {"cron1234abcd", "cron5678efgh"} # A second load reads the repaired list unchanged (idempotent). reloaded = load_jobs() assert {j["id"] for j in reloaded} == {"cron1234abcd", "cron5678efgh"} def test_empty_id_keyed_map_returns_empty_list(self, tmp_cron_dir): """An empty ``jobs`` map must not crash and yields no jobs.""" import json from cron.jobs import JOBS_FILE, load_jobs JOBS_FILE.parent.mkdir(parents=True, exist_ok=True) JOBS_FILE.write_text(json.dumps({"jobs": {}}), encoding="utf-8") assert load_jobs() == [] def test_map_value_without_inline_id_adopts_key(self, tmp_cron_dir): """A value lacking an inline "id" gets the map key as its id.""" import json from cron.jobs import JOBS_FILE, load_jobs payload = { "jobs": { "cronkeyonly1": { "name": "keyed only", "enabled": True, "prompt": "no inline id here", "schedule": {"kind": "interval", "minutes": 15, "display": "every 15m"}, }, "cron-ignored-key": { "id": "croninline99", "name": "inline id wins", "enabled": True, "prompt": "inline id present", "schedule": {"kind": "interval", "minutes": 5, "display": "every 5m"}, }, } } JOBS_FILE.parent.mkdir(parents=True, exist_ok=True) JOBS_FILE.write_text(json.dumps(payload), encoding="utf-8") loaded = {j["id"]: j for j in load_jobs()} # Key adopted when the value has no inline id. assert "cronkeyonly1" in loaded assert loaded["cronkeyonly1"]["name"] == "keyed only" # Inline id wins over a differing map key. assert "croninline99" in loaded assert "cron-ignored-key" not in loaded # Self-heal persisted the id-merged records. on_disk = json.loads(JOBS_FILE.read_text(encoding="utf-8")) assert isinstance(on_disk["jobs"], list) assert {j["id"] for j in on_disk["jobs"]} == {"cronkeyonly1", "croninline99"} def test_non_dict_map_values_skipped_with_warning(self, tmp_cron_dir, caplog): """Junk (non-dict) values in the map are skipped, never crash.""" import json import logging from cron.jobs import JOBS_FILE, list_jobs, load_jobs payload = { "jobs": { "goodjob1": { "name": "survivor", "enabled": True, "prompt": "keep me", "schedule": {"kind": "interval", "minutes": 60, "display": "every 60m"}, }, "junk-string": "i am not a job", "junk-number": 42, "junk-null": None, } } JOBS_FILE.parent.mkdir(parents=True, exist_ok=True) JOBS_FILE.write_text(json.dumps(payload), encoding="utf-8") with caplog.at_level(logging.WARNING, logger="cron.jobs"): loaded = load_jobs() assert [j["id"] for j in loaded] == ["goodjob1"] assert any("non-dict" in rec.getMessage() for rec in caplog.records) # The reported traceback path also survives the junk. jobs = list_jobs(include_disabled=True) assert {j["id"] for j in jobs} == {"goodjob1"} # Self-heal wrote only the valid record, canonical list shape. on_disk = json.loads(JOBS_FILE.read_text(encoding="utf-8")) assert isinstance(on_disk["jobs"], list) assert [j["id"] for j in on_disk["jobs"]] == ["goodjob1"] def test_all_junk_map_values_yield_empty_list(self, tmp_cron_dir): """A map of only junk values flattens to [] without crashing.""" import json from cron.jobs import JOBS_FILE, load_jobs JOBS_FILE.parent.mkdir(parents=True, exist_ok=True) JOBS_FILE.write_text( json.dumps({"jobs": {"a": "junk", "b": 1}}), encoding="utf-8" ) assert load_jobs() == [] class TestAdvanceNextRuns: """Tests for advance_next_runs() — the batched due-set advance. The scheduler's pre-dispatch loop advanced each due job individually: N due jobs = N full load_jobs() + N full save_jobs() of the jobs file (~110 ms at N=50, measured). The batch form does one load + at most one save (~2 ms). Imported inside test bodies so the pre-fix tree fails with a real test failure (ImportError), not a collection error. """ def _make_due(self, tmp_cron_dir, n_recurring=3, n_oneshot=1): rec = [create_job(prompt=f"rec {i}", schedule="every 1h") for i in range(n_recurring)] one = [create_job(prompt=f"one {i}", schedule="in 30m") for i in range(n_oneshot)] jobs = load_jobs() old = (datetime.now() - timedelta(minutes=5)).isoformat() for j in jobs: j["next_run_at"] = old save_jobs(jobs) return [j["id"] for j in rec], [j["id"] for j in one] def test_batch_advances_recurring_skips_oneshots(self, tmp_cron_dir): from cron.jobs import advance_next_runs rec_ids, one_ids = self._make_due(tmp_cron_dir) advanced = advance_next_runs(rec_ids + one_ids) assert advanced == len(rec_ids) from cron.jobs import _ensure_aware, _hermes_now for jid in rec_ids: nxt = _ensure_aware(datetime.fromisoformat(get_job(jid)["next_run_at"])) assert nxt > _hermes_now() for jid in one_ids: # one-shots keep their (past) next_run_at for restart retry assert datetime.fromisoformat(get_job(jid)["next_run_at"]) < datetime.now() def test_batch_single_load_and_save(self, tmp_cron_dir, monkeypatch): """I/O pin: the whole due set costs one load + one save, not N+N. Fails pre-fix (function absent) and would fail on any regression back to per-job I/O.""" from cron.jobs import advance_next_runs rec_ids, _ = self._make_due(tmp_cron_dir, n_recurring=10, n_oneshot=0) import cron.jobs as cj counts = {"load": 0, "save": 0} real_load, real_save = cj.load_jobs, cj.save_jobs monkeypatch.setattr(cj, "load_jobs", lambda *a, **k: ( counts.__setitem__("load", counts["load"] + 1), real_load(*a, **k))[1]) monkeypatch.setattr(cj, "save_jobs", lambda *a, **k: ( counts.__setitem__("save", counts["save"] + 1), real_save(*a, **k))[1]) advance_next_runs(rec_ids) assert counts == {"load": 1, "save": 1} def test_batch_no_save_when_nothing_advances(self, tmp_cron_dir, monkeypatch): from cron.jobs import advance_next_runs rec_ids, one_ids = self._make_due(tmp_cron_dir, n_recurring=0, n_oneshot=2) import cron.jobs as cj saves = [0] real_save = cj.save_jobs monkeypatch.setattr(cj, "save_jobs", lambda *a, **k: ( saves.__setitem__(0, saves[0] + 1), real_save(*a, **k))[1]) assert advance_next_runs(one_ids + ["missing-id"]) == 0 assert saves[0] == 0 def test_wrapper_semantics_unchanged(self, tmp_cron_dir): """advance_next_run keeps its per-job contract over the batch.""" rec_ids, one_ids = self._make_due(tmp_cron_dir) assert advance_next_run(rec_ids[0]) is True assert advance_next_run(one_ids[0]) is False assert advance_next_run("missing-id") is False # ========================================================================= # Completed one-shot retention sweep # ========================================================================= class TestCompletedOneshotRetentionSweep: """Completed one-shots are retained for inspection, then pruned by age.""" def _completed_oneshot(self, age_days: float): """Create a one-shot, complete it, and backdate its last_run_at.""" job = create_job(prompt="Once", schedule="in 30m", repeat=1) mark_job_run(job["id"], success=True, delivery_error="boom") stamp = ( datetime.now(timezone.utc) - timedelta(days=age_days) ).isoformat() jobs = load_jobs() for j in jobs: if j["id"] == job["id"]: j["last_run_at"] = stamp save_jobs(jobs) return job["id"] def test_sweep_prunes_old_completed_oneshot(self, tmp_cron_dir): old_id = self._completed_oneshot(age_days=30) get_due_jobs() # sweep runs as part of the due scan assert get_job(old_id) is None def test_sweep_keeps_recent_completed_oneshot(self, tmp_cron_dir): recent_id = self._completed_oneshot(age_days=1) get_due_jobs() kept = get_job(recent_id) assert kept is not None assert kept["state"] == "completed" assert kept["last_delivery_error"] == "boom" def test_sweep_ignores_recurring_jobs(self, tmp_cron_dir): """Old recurring jobs are never candidates, whatever their history.""" job = create_job(prompt="Recurring", schedule="every 1h") stamp = ( datetime.now(timezone.utc) - timedelta(days=365) ).isoformat() jobs = load_jobs() for j in jobs: if j["id"] == job["id"]: j["last_run_at"] = stamp save_jobs(jobs) get_due_jobs() assert get_job(job["id"]) is not None def test_sweep_disabled_by_nonpositive_retention(self, tmp_cron_dir, monkeypatch): monkeypatch.setattr( "cron.jobs._completed_oneshot_retention_days", lambda: 0.0 ) old_id = self._completed_oneshot(age_days=30) get_due_jobs() assert get_job(old_id) is not None def test_recurring_jobs_unaffected_by_retention_change(self, tmp_cron_dir): """A recurring job still cycles normally alongside retained one-shots.""" recurring = create_job(prompt="Recurring", schedule="every 1h") self._completed_oneshot(age_days=1) mark_job_run(recurring["id"], success=True) updated = get_job(recurring["id"]) assert updated["enabled"] is True assert updated["state"] == "scheduled" assert updated["next_run_at"] is not None class TestEnsureCronDirWidened: """Tests for the widened _ensure_cron_dir covering all cron mkdir sites.""" def test_ensure_cron_dir_named_profile_subdir_fails_closed(self, tmp_path): """A subdir under a deleted named profile's cron/ must not recreate it.""" import cron.jobs as jobs profiles_dir = tmp_path / "profiles" profiles_dir.mkdir() deleted_home = profiles_dir / "deleted" # cron_dir doesn't exist because the profile was deleted output_dir = deleted_home / "cron" / "output" / "job_123" import pytest with pytest.raises(FileNotFoundError): jobs._ensure_cron_dir(output_dir) assert not deleted_home.exists() def test_ensure_cron_dir_default_home_creates_subdir(self, tmp_path): """A subdir under a default home's cron/ should be created normally.""" import cron.jobs as jobs default_home = tmp_path / "default_home" default_home.mkdir() output_dir = default_home / "cron" / "output" / "job_123" jobs._ensure_cron_dir(output_dir) assert output_dir.is_dir() def test_ensure_cron_dir_named_profile_cron_dir_fails_closed(self, tmp_path): """The cron dir of a deleted named profile must not be recreated.""" import cron.jobs as jobs profiles_dir = tmp_path / "profiles" profiles_dir.mkdir() deleted_home = profiles_dir / "deleted" cron_dir = deleted_home / "cron" import pytest with pytest.raises(FileNotFoundError): jobs._ensure_cron_dir(cron_dir) assert not deleted_home.exists() def test_ensure_cron_dir_existing_named_profile_cron_dir_works(self, tmp_path): """An existing named profile's cron dir should be created normally.""" import cron.jobs as jobs profiles_dir = tmp_path / "profiles" active_home = profiles_dir / "active" active_home.mkdir(parents=True) cron_dir = active_home / "cron" jobs._ensure_cron_dir(cron_dir) assert cron_dir.is_dir() def test_ensure_cron_dir_scripts_dir_under_named_profile_fails_closed(self, tmp_path): """A scripts dir under a deleted named profile must not be recreated.""" import cron.jobs as jobs profiles_dir = tmp_path / "profiles" profiles_dir.mkdir() deleted_home = profiles_dir / "deleted" scripts_dir = deleted_home / "scripts" import pytest with pytest.raises(FileNotFoundError): jobs._ensure_cron_dir(scripts_dir) assert not deleted_home.exists()