From 6d6114c8fcb6ebcaea594341e72ec8dca9d1fc7e Mon Sep 17 00:00:00 2001 From: EMRG Evolution Date: Thu, 22 Oct 2026 18:47:10 +0800 Subject: [PATCH 1/3] emrg: a scheduled task's frequency can vary by the time of day --- DEVELOPMENT.md | 14 ++ emrg/server/daemon.py | 3 +- emrg/server/scheduler.py | 314 ++++++++++++++++++++++++++- tests/test_schedule_window.py | 397 ++++++++++++++++++++++++++++++++++ tests/test_ws_e2e.py | 43 +++- 5 files changed, 757 insertions(+), 14 deletions(-) create mode 100644 tests/test_schedule_window.py diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 4db03707..45b4bbec 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -145,6 +145,20 @@ The new path's boundary exists on **macOS (Seatbelt)**, **Linux (bubblewrap)** a A confined run cannot write your home directory, which is where `uv`, `pip` and `npm` keep their caches by default: measured at `workspace-write`, `uv run` **fails** (`Failed to initialize cache at ~/.cache/uv`), `npm`'s `~/.npm` cannot be created, and `pip` quietly disables its cache. So the tool points those four variables (`XDG_CACHE_HOME`, `UV_CACHE_DIR`, `PIP_CACHE_DIR`, `npm_config_cache`) at `/emrg-confined-cache` for the confined child β€” a directory inside a root the policy already grants, so nothing is widened. **A variable you set yourself is never overridden**, which is how you keep a warm cache: point it at a directory inside the session's workspace. +**A task's frequency by time of day** (`schedule` in `~/.emrg/tasks.yml`): a record's single `interval` answers at every instant unless it states `schedule`, an optional list of windows: + +```yaml +- name: feishu-inbox + interval: 10800 # the frequency outside every window + schedule: # optional; without it the task behaves exactly as before + - days: mon-fri # optional = every day; mon, mon-fri, mon,fri, fri-mon + from: "09:00" # quoted: unquoted, YAML reads 09:00 as a number + to: "18:00" # 24:00 allowed, and later than `from` + interval: 900 # the frequency inside this window +``` + +Inside a window that window's `interval` answers; outside every window and between windows the top-level one does, so one instant never has two answers. Times are **local time only** β€” read with `datetime.now()`, with no timezone field and no DST rule β€” so the windows follow the host's clock; there are no public holidays and no make-up workdays, only the weekdays a window names (`fri-mon` wraps around the week). A wait that would run past the next edge is **cut to it**: a 3-hour weekend cadence that begins at 06:59 wakes at 09:00 rather than sleeping through the window it was about to enter, and a restart inside a window does not drop the task to the out-of-window frequency. Two windows of one day may not overlap, a window crossing midnight is written as two windows, and every field is validated when the file is read β€” a bad window is refused with the value it refused and the shape it expects. The daemon's create/update path and the task listing carry the same shape (`effective_interval` and `next_window_boundary_in_seconds` in `list_tasks`); until a client can edit windows, hand-editing this file is how to set one, and the scheduler picks the change up at the task's next wake. + --- ## πŸ—οΈ Architecture diff --git a/emrg/server/daemon.py b/emrg/server/daemon.py index 2b2e3656..2aa09f8f 100644 --- a/emrg/server/daemon.py +++ b/emrg/server/daemon.py @@ -3101,6 +3101,7 @@ async def _process_message( repo=msg.get("repo"), description=msg.get("description"), sandbox=msg.get("sandbox"), + schedule=msg.get("schedule"), ) if not ok: await self._send(ws, {"type": "task_result", "error": res}) @@ -3116,7 +3117,7 @@ async def _process_message( if not self._scheduler: await self._send(ws, {"type": "task_result", "error": "scheduler not running"}) return - fields = {k: msg[k] for k in ("task_type", "project", "interval", "enabled", "repo", "description", "sandbox") if k in msg} + fields = {k: msg[k] for k in ("task_type", "project", "interval", "enabled", "repo", "description", "sandbox", "schedule") if k in msg} if "task_type" in fields: fields["type"] = fields.pop("task_type") ok, res = self._scheduler.task_update(msg.get("name", "").strip(), **fields) diff --git a/emrg/server/scheduler.py b/emrg/server/scheduler.py index ee0747fc..33a18688 100644 --- a/emrg/server/scheduler.py +++ b/emrg/server/scheduler.py @@ -10,6 +10,10 @@ Task config schema: name, type, enabled, interval β€” common base fields. + schedule β€” optional; time-of-day windows that answer with their own interval while + open. Read in **local time only** (`datetime.now()`), so it follows the host's + clock and whatever its zone does with DST. Weekdays only: no holidays, no make-up + workdays. config β€” type-specific config, and the mapping the templates see as ``task``. For evolution: config.project links to projects.yml name. sandbox β€” optional; also accepted under config, where the record-level value wins. @@ -28,7 +32,7 @@ import os import re import time -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path import yaml @@ -205,6 +209,230 @@ def recovery_recipe(stash_message: str) -> str: MIN_INTERVAL = 60 DEFAULT_INTERVAL = 1800 +# ── Scheduling windows (issue #2486) ───────────────────────────────────────── +# +# A task carries one frequency β€” `interval` β€” and that stays the only knob unless it +# states `schedule`: optional windows that answer with a different frequency while +# they are open. Inside a window, that window's `interval`; outside every window and +# between windows, the top-level one. Time is read with `datetime.now()` and nothing +# else β€” no timezone argument, no DST rule, no holiday table β€” so the windows follow +# the host's own clock, which is the clock the host works by. +# +# schedule: +# - days: mon-fri # optional = every day; mon, mon-fri, mon,fri, fri-mon +# from: "09:00" # quoted: unquoted, YAML 1.1 reads 09:00 as a number +# to: "18:00" # 24:00 allowed; to must be later than from +# interval: 900 # the frequency inside this window +# +# Two windows of the same day may not overlap: one instant cannot have two answers. +# `effective_interval` is the one producer of "how often, right now" β€” the heartbeat, +# the connection backoff, the wait and every reported row read it and none re-derives +# the rule. `seconds_to_next_boundary` is what keeps a long wait from sleeping through +# a window: a wait that would run past an edge is cut to it. +WEEKDAY_NAMES = ("mon", "tue", "wed", "thu", "fri", "sat", "sun") +_WINDOW_TIME_RE = re.compile(r"^(\d{1,2}):(\d{2})$") + + +def _minutes_to_hhmm(minutes: int) -> str: + """`540` β†’ `09:00` β€” the one spelling of a window edge inside a message.""" + return f"{minutes // 60:02d}:{minutes % 60:02d}" + + +def _days_label(days) -> str: + """A sorted `mon,tue` label for a set of weekday indices.""" + return ",".join(WEEKDAY_NAMES[day] for day in sorted(days)) + + +def _parse_window_time(text) -> int | None: + """`HH:MM` β†’ minutes since midnight (`24:00` is 1440), or None when unparsable.""" + if not isinstance(text, str): + return None + match = _WINDOW_TIME_RE.match(text.strip()) + if match is None: + return None + hour, minute = int(match.group(1)), int(match.group(2)) + if hour > 24 or minute > 59: + return None + return hour * 60 + minute + + +def _parse_window_days(text) -> set[int] | None: + """`mon`, `mon-fri`, `mon,fri`, `fri-mon` β†’ weekday indices (0 = Monday), or None. + + Case and surrounding whitespace are irrelevant, and a range expands **around the + ring**, so `fri-mon` is Fri, Sat, Sun, Mon. Absent or null means every day: a window + naming no weekday is a time-of-day window, not a weekly one. + """ + if text is None: + return set(range(7)) + if not isinstance(text, str): + return None + days: set[int] = set() + for part in text.split(","): + part = part.strip().lower() + if not part: + continue + if "-" in part: + first, _, last = part.partition("-") + first, last = first.strip(), last.strip() + if first not in WEEKDAY_NAMES or last not in WEEKDAY_NAMES: + return None + start, end = WEEKDAY_NAMES.index(first), WEEKDAY_NAMES.index(last) + for step in range((end - start) % 7 + 1): + days.add((start + step) % 7) + else: + if part not in WEEKDAY_NAMES: + return None + days.add(WEEKDAY_NAMES.index(part)) + return days or None + + +def _normalise_schedule(schedule) -> list[dict]: + """The windows as parsed numbers; `[]` when the field is absent or unusable. + + Unusable is not an exception path here: `validate_schedule` is what refuses a bad + window, and it runs in the loader (`validate_task_record`) and in both CRUD write + paths, so a record that reaches the loop has already been accepted. A hand edit that + slipped past every writer falls back to the top-level interval β€” a cadence that is + wrong until the next wake is recoverable; a raise on the event loop is not. + """ + if not isinstance(schedule, list): + return [] + windows: list[dict] = [] + for entry in schedule: + if not isinstance(entry, dict): + return [] + days = _parse_window_days(entry.get("days")) + start = _parse_window_time(entry.get("from")) + end = _parse_window_time(entry.get("to")) + interval = entry.get("interval") + if days is None or start is None or end is None: + return [] + if not isinstance(interval, int) or isinstance(interval, bool): + return [] + windows.append({"days": days, "start": start, "end": end, "interval": interval}) + return windows + + +def validate_schedule(schedule) -> str | None: + """Return why a `schedule` field cannot be run, or None when it is usable. + + The one validator for the field: the loader and the daemon's create/update path + both call it, so a window means the same thing to the task that reads it and to the + form that writes it. Every message names the value it refused and the shape it + expects, because "invalid schedule" leaves a host with a file and no next step. + """ + if schedule is None: + return None + if not isinstance(schedule, list): + return f"schedule of type {type(schedule).__name__}, not a list of windows" + if not schedule: + return "an empty schedule list - drop the field for one frequency all day" + parsed: list[dict] = [] + for index, entry in enumerate(schedule, 1): + where = f"schedule window {index}" + if not isinstance(entry, dict): + return ( + f"{where} of type {type(entry).__name__}, not a mapping of " + "days/from/to/interval" + ) + days = _parse_window_days(entry.get("days")) + if days is None: + return ( + f"{where} has days {entry.get('days')!r} - expected weekdays such as " + "`mon`, `mon-fri` or `mon,fri` (a range may wrap, e.g. `fri-mon`)" + ) + start = _parse_window_time(entry.get("from")) + if start is None: + return ( + f"{where} has from {entry.get('from')!r} - expected `HH:MM` written as a " + "quoted string (`\"09:00\"`; unquoted, YAML reads it as a number)" + ) + end = _parse_window_time(entry.get("to")) + if end is None: + return ( + f"{where} has to {entry.get('to')!r} - expected `HH:MM` written as a " + "quoted string, with 24:00 allowed" + ) + if end <= start: + return ( + f"{where} runs {_minutes_to_hhmm(start)}-{_minutes_to_hhmm(end)}, which " + "does not move forward - a window crossing midnight is two windows" + ) + interval = entry.get("interval") + if not isinstance(interval, int) or isinstance(interval, bool) or interval < MIN_INTERVAL: + return f"{where} has interval {interval!r} (must be an integer >= {MIN_INTERVAL})" + parsed.append({"days": days, "start": start, "end": end, "interval": interval}) + for index, first in enumerate(parsed): + for second in parsed[index + 1:]: + shared = first["days"] & second["days"] + if not shared: + continue + if first["start"] < second["end"] and second["start"] < first["end"]: + return ( + f"schedule windows overlap on {_days_label(shared)}: " + f"{_minutes_to_hhmm(first['start'])}-{_minutes_to_hhmm(first['end'])} " + f"and {_minutes_to_hhmm(second['start'])}-" + f"{_minutes_to_hhmm(second['end'])} - one instant cannot have two " + "frequencies" + ) + return None + + +def schedule_window_at(schedule, now: datetime) -> dict | None: + """The window `now` falls inside, or None when it falls outside every one.""" + minutes = now.hour * 60 + now.minute + weekday = now.weekday() + for window in _normalise_schedule(schedule): + if weekday in window["days"] and window["start"] <= minutes < window["end"]: + return window + return None + + +def effective_interval(record: dict, now: datetime | None = None) -> int: + """The frequency in force at `now` β€” the only place that answers it. + + Inside a window, that window's interval; outside every window and between windows, + the record's top-level one. Without `schedule` the answer is the same at every + instant, which is exactly the cadence a task had before windows existed. + """ + now = now or datetime.now() + window = schedule_window_at(record.get("schedule"), now) + if window is None: + return record.get("interval", DEFAULT_INTERVAL) + return window["interval"] + + +def seconds_to_next_boundary(record: dict, now: datetime | None = None) -> float | None: + """Seconds until the effective interval next changes, or None without windows. + + Every window edge β€” a start or an end, on every weekday that window names β€” is a + boundary, and a wait that would run past one is cut to it: a three-hour weekend wait + beginning at 06:59 must wake at 09:00 rather than sleep through the window it was + about to enter. Strictly in the future, so waking at an edge recomputes under the + frequency that edge selects instead of finding the same edge zero seconds away. + """ + windows = _normalise_schedule(record.get("schedule")) + if not windows: + return None + now = now or datetime.now() + midnight = now.replace(hour=0, minute=0, second=0, microsecond=0) + soonest: datetime | None = None + # Nine days reach every weekday at least once more, whichever day `now` falls on. + for offset in range(9): + day = midnight + timedelta(days=offset) + weekday = day.weekday() + for window in windows: + if weekday not in window["days"]: + continue + for edge_minutes in (window["start"], window["end"]): + edge = day + timedelta(minutes=edge_minutes) + if edge > now and (soonest is None or edge < soonest): + soonest = edge + if soonest is None: + return None + return (soonest - now).total_seconds() + # ── tasks.yml is the only source of truth (rant 2026-09-28T09:54:10) ── # # The host's principle, verbatim: γ€Œtasks.yml ζœ¬θΊ«δΈεšδ»»δ½•ε†…ε­˜ηΌ“ε­˜γ€‚δ»€δΉˆζ—Άε€™η”¨οΌŒ @@ -632,6 +860,9 @@ def validate_task_record(record: dict) -> str | None: interval = record.get("interval", DEFAULT_INTERVAL) if not isinstance(interval, int) or isinstance(interval, bool) or interval < MIN_INTERVAL: return f"task {name!r} has interval {interval!r} (must be an integer >= {MIN_INTERVAL})" + schedule_error = validate_schedule(record.get("schedule")) + if schedule_error: + return f"task {name!r} has {schedule_error}" cfg = record.get("config") if cfg is not None and not isinstance(cfg, dict): return f"task {name!r} has config of type {type(cfg).__name__}, not a mapping" @@ -2284,7 +2515,8 @@ async def run(self) -> None: continue if not _started_logged: self._logger.info( - "TaskHandler[%s] started β€” every %ds", self.name, self.interval + "TaskHandler[%s] started β€” every %ds", + self.name, self._effective_interval(), ) _started_logged = True @@ -2305,6 +2537,17 @@ async def run(self) -> None: # shorten to the remainder, or run immediately when the slot # already passed during downtime). wait_timeout = self._resume_wait_timeout(wait_timeout) + # Issue #2486 ④⑨: cut the wait to the next window edge. A wait that spans + # an edge would sleep through the whole window it was about to enter β€” the + # 3-hour weekend wait that begins at 06:59 must wake at 09:00, not at + # 09:59 β€” and waking at the edge recomputes under the frequency that edge + # selects, so this re-arms rather than spins. The resume slot folded in just + # above is clipped by the same rule, which is why a daemon restart inside a + # window does not drop the task to the out-of-window frequency. The cut only + # ever shortens: the slowdown and the backoff still set the wait it shortens. + _boundary = seconds_to_next_boundary(self._task) + if _boundary is not None and _boundary < wait_timeout: + wait_timeout = _boundary # Diagnostic log (rant 2026-08-18T20:48:45): expose which # scheduling mode drove the wait β€” normal | heartbeat | backoff β€” # so the saturation/backoff state machine is traceable end-to-end. @@ -2315,8 +2558,8 @@ async def run(self) -> None: else: _mode = "normal" self._logger.debug( - "TaskHandler[%s]: next run in %ds (mode=%s)", - self.name, int(wait_timeout), _mode, + "TaskHandler[%s]: next run in %ds (mode=%s, interval %ds)", + self.name, int(wait_timeout), _mode, self._effective_interval(), ) # Wait for interval or manual trigger (interruptible) self._next_run_at = time.time() + wait_timeout @@ -2502,12 +2745,19 @@ def status(self) -> dict: # start time to confuse it with any more: it existed only to render # `{{ uptime }}` and went with that placeholder, rant 2026-09-29T09:23:55.) started_at = self._cycle_start_time if self._cycle_running else None + # Issue #2486 ⑦: the frequency in force right now and when it next changes, so + # a client renders what the scheduler is actually doing instead of re-deriving + # it from the windows (and getting a different answer at a different instant). + boundary = seconds_to_next_boundary(self._task) return { "name": self.name, "running": self._cycle_running, "started_at": started_at, "next_run_in_seconds": remaining, "interval": self.interval, + "effective_interval": self._effective_interval(), + "schedule": self._task.get("schedule"), + "next_window_boundary_in_seconds": None if boundary is None else int(boundary), "last_run_at": last_run_at, "recent_runs": recent_runs, "saturation": saturation, @@ -2525,8 +2775,23 @@ def _heartbeat_interval(self) -> int: heartbeat = max(task_interval, min(task_interval * 8, 8 hours)). Protection = slow down, never stop. Long-interval tasks (>= 8h) keep their original cadence (the 8x/8h caps don't apply). + + The base is the interval **in force right now** (`_effective_interval`), so a + window's frequency gets its own heartbeat instead of the out-of-window one + being slowed eightfold while the host is at work (issue #2486 ⑧). + """ + interval = self._effective_interval() + return max(interval, min(interval * 8, 8 * 3600)) + + def _effective_interval(self) -> int: + """How often this task runs right now, from its own record's windows. + + The one producer of that answer: `_heartbeat_interval`, `_connect_backoff`, the + wait and the reported rows all take their base from here and none re-derives + the rule, so the slowdown and the backoff apply to the frequency actually in + force rather than fighting the window (issue #2486 ⑒⑧). """ - return max(self.interval, min(self.interval * 8, 8 * 3600)) + return effective_interval(self._task) def _connect_backoff(self) -> float: """Exponential backoff while the daemon is unreachable. @@ -2538,11 +2803,15 @@ def _connect_backoff(self) -> float: Backoff = max(30s, interval * 2^n) capped at 10 minutes, where n is the consecutive-failure count. Returns the normal interval when there are no consecutive failures. + The base is the interval in force right now (`_effective_interval`), for the + same reason the heartbeat's is: the backoff grows from the cadence the window + has selected, never from the one it overrode (issue #2486 ⑧). """ + interval = self._effective_interval() if self._connect_failures <= 0: - return float(self.interval) + return float(interval) n = min(self._connect_failures, 10) # cap the exponent growth - backoff = max(30.0, float(self.interval) * (2 ** n)) + backoff = max(30.0, float(interval) * (2 ** n)) return min(backoff, 600.0) # never wait longer than 10 minutes def _saturation_heartbeat_active(self) -> bool: @@ -3314,12 +3583,19 @@ def _idle_status(self, record: dict) -> dict: """ cfg = record.get("config") if isinstance(record.get("config"), dict) else {} project = cfg.get("project", "") + # Issue #2486 ⑦: the same three cadence keys the live handler reports β€” a + # client must not have to know whether a task has been started before it can + # render the frequency that is in force, or when it next changes. + boundary = seconds_to_next_boundary(record) return { "name": record.get("name"), "running": False, "started_at": None, "next_run_in_seconds": None, "interval": record.get("interval", DEFAULT_INTERVAL), + "effective_interval": effective_interval(record), + "schedule": record.get("schedule"), + "next_window_boundary_in_seconds": None if boundary is None else int(boundary), "last_run_at": None, "recent_runs": [], "saturation": {}, @@ -3419,6 +3695,7 @@ def _ensure_emrg_project_entry(self) -> None: @staticmethod def _validate_task_fields( name: str, task_type: str, project: str, interval: int, + schedule: list | None = None, ) -> str | None: """Return an error string, or None when the fields are valid.""" if not TASK_NAME_RE.match(name) or len(name) > TASK_NAME_MAX: @@ -3429,17 +3706,18 @@ def _validate_task_fields( return f"project {project!r} is not registered in projects.yml" if not isinstance(interval, int) or isinstance(interval, bool) or interval < MIN_INTERVAL: return f"interval must be an integer >= {MIN_INTERVAL}" - return None + # The same validator the loader runs: one shape, one set of messages. + return validate_schedule(schedule) def task_create( self, name: str, task_type: str, project: str, interval: int | None = None, enabled: bool = True, repo: str | None = None, description: str | None = None, - sandbox: str | None = None, + sandbox: str | None = None, schedule: list | None = None, ) -> tuple[bool, str | dict]: """Create a task. Returns (ok, error) or (ok, task-dict).""" interval = DEFAULT_INTERVAL if interval is None else interval - err = self._validate_task_fields(name, task_type, project, interval) + err = self._validate_task_fields(name, task_type, project, interval, schedule) if err: return False, err if sandbox is not None and sandbox not in SANDBOX_MODES: @@ -3466,6 +3744,8 @@ def task_create( } if sandbox is not None: task["sandbox"] = sandbox + if schedule is not None: + task["schedule"] = schedule if description: task["description"] = description tasks.append(task) @@ -3485,7 +3765,10 @@ def task_update(self, name: str, **fields) -> tuple[bool, str | dict]: cfg = task.get("config") if isinstance(task.get("config"), dict) else {} new_project = cfg.get("project", "") new_interval = fields.get("interval", task.get("interval", DEFAULT_INTERVAL)) - err = self._validate_task_fields(name, new_type, new_project, new_interval) + new_schedule = fields.get("schedule", task.get("schedule")) + err = self._validate_task_fields( + name, new_type, new_project, new_interval, new_schedule, + ) if err: return False, err if "type" in fields: @@ -3502,6 +3785,15 @@ def task_update(self, name: str, **fields) -> tuple[bool, str | dict]: task["config"] = cfg if "interval" in fields: task["interval"] = fields["interval"] + if "schedule" in fields: + # Only the keys a payload carries are written, and this field keeps that + # rule two ways: a payload that omits it leaves the windows alone (which is + # what lets a client that predates the field keep editing a task), while an + # explicit null means "one frequency all day", so the key leaves the file. + if fields["schedule"] is None: + task.pop("schedule", None) + else: + task["schedule"] = fields["schedule"] if "enabled" in fields: task["enabled"] = bool(fields["enabled"]) if "description" in fields: diff --git a/tests/test_schedule_window.py b/tests/test_schedule_window.py new file mode 100644 index 00000000..d44a4504 --- /dev/null +++ b/tests/test_schedule_window.py @@ -0,0 +1,397 @@ +"""Scheduling windows on a task record β€” issue #2486. + +A task carries one frequency; `schedule` adds windows that answer with their own +frequency while they are open. Acceptance is asserted, not demonstrated: + + * both edges of a window answer with their own frequency (17:59 β†’ 15 min, + 18:00 β†’ 3 h); + * a wait is cut to the next edge on its own β€” a weekend band entered at 06:59 + wakes at 09:00, not at 09:59 and not three hours later; + * a record without `schedule` waits exactly as long as it does today, measured + through the same loop; + * four rejected inputs, each with a readable error; + * the day rule on its own, and the cut wired into `run()` rather than only into + the helper that computes it β€” so deleting either rule reddens a test. + +The edge answer, the day rule and the cut are separate rules on purpose: a mutation +of any one of them has to redden something here. +""" + +from __future__ import annotations + +import asyncio +import time +from datetime import datetime, timedelta + +import pytest +import yaml + +from emrg.protocol import InstanceIdentity +from emrg.server import scheduler as mod +from emrg.server.scheduler import ( + DEFAULT_INTERVAL, + MIN_INTERVAL, + TaskHandler, + TaskScheduler, + effective_interval, + seconds_to_next_boundary, + validate_schedule, + validate_task_record, +) + +# A day whose weekday the test itself states, so no assertion here rests on a calendar +# the reader has to take on trust. +MONDAY = datetime(2026, 10, 19) +SATURDAY = MONDAY + timedelta(days=5) + + +def _record(**overrides) -> dict: + record = { + "name": "windowed", + "type": "evolution", + "config": {"project": "emrg"}, + "interval": 10800, + "enabled": True, + "schedule": [ + {"days": "mon-fri", "from": "09:00", "to": "18:00", "interval": 900}, + ], + } + record.update(overrides) + return record + + +# ── the day rule (deleting it reddens these) ───────────────────────────────── + + +def test_the_calendar_of_this_test_is_stated_not_assumed(): + assert MONDAY.weekday() == 0, "MONDAY is a Monday" + assert SATURDAY.weekday() == 5, "SATURDAY is a Saturday" + + +def test_days_parse_case_whitespace_and_wrap_around_the_ring(): + assert mod._parse_window_days("MON") == {0} + assert mod._parse_window_days(" mon , fri ") == {0, 4} + assert mod._parse_window_days("mon-fri") == {0, 1, 2, 3, 4} + # A range wraps: fri-mon is Fri, Sat, Sun, Mon β€” not the empty set, and not fri..sun. + assert mod._parse_window_days("fri-mon") == {4, 5, 6, 0} + assert mod._parse_window_days("sat-sun") == {5, 6} + assert mod._parse_window_days(None) == set(range(7)), "absent = every day" + assert mod._parse_window_days("funday") is None + + +def test_a_window_applies_only_on_its_own_weekdays(): + """The day rule, read through the answer it selects β€” a weekday and a weekend.""" + record = _record() + inside_a_weekday = effective_interval(record, MONDAY.replace(hour=10, minute=0)) + inside_a_saturday = effective_interval(record, SATURDAY.replace(hour=10, minute=0)) + assert inside_a_weekday == 900 + assert inside_a_saturday == 10800, "mon-fri says nothing about Saturday" + + +def test_a_wrapping_range_does_open_on_the_weekend(): + record = _record(schedule=[ + {"days": "fri-mon", "from": "09:00", "to": "18:00", "interval": 900}, + ]) + assert effective_interval(record, SATURDAY.replace(hour=10, minute=0)) == 900 + + +# ── both edges answer with their own frequency ─────────────────────────────── + + +@pytest.mark.parametrize("hour, minute, expected", [ + (8, 59, 10800), # one minute before the window opens + (9, 0, 900), # the opening edge selects the window's frequency + (17, 59, 900), # one minute before the window closes + (18, 0, 10800), # the closing edge gives the frequency back +]) +def test_both_window_edges_answer_with_their_own_frequency(hour, minute, expected): + record = _record() + assert effective_interval(record, MONDAY.replace(hour=hour, minute=minute)) == expected + + +def test_between_two_windows_the_top_level_frequency_is_answered(): + record = _record(schedule=[ + {"days": "mon", "from": "09:00", "to": "12:00", "interval": 900}, + {"days": "mon", "from": "14:00", "to": "18:00", "interval": 600}, + ]) + assert effective_interval(record, MONDAY.replace(hour=11, minute=0)) == 900 + assert effective_interval(record, MONDAY.replace(hour=13, minute=0)) == 10800 + assert effective_interval(record, MONDAY.replace(hour=15, minute=0)) == 600 + + +# ── the cut is a rule of its own ───────────────────────────────────────────── + + +def test_a_weekend_band_wakes_at_the_window_edge_not_after_the_whole_interval(): + """06:59 on Saturday, 3-hour weekend frequency, a band opening at 09:00. + + The wait must be 121 minutes: at 09:59 it would be 180 minutes (121 to the edge + plus a full interval inside it) and at 12:00 a whole 3 hours, sleeping through the + window it was about to enter. + """ + record = _record(schedule=[ + {"days": "sat-sun", "from": "09:00", "to": "18:00", "interval": 900}, + ]) + at_0659 = SATURDAY.replace(hour=6, minute=59) + assert effective_interval(record, at_0659) == 10800, "outside the band" + assert seconds_to_next_boundary(record, at_0659) == 7260.0, "121 minutes to 09:00" + assert 7260.0 < 10800, "the plain interval would have slept through the band" + + +def test_the_next_boundary_is_always_in_the_future(): + record = _record() + for hour in (8, 9, 17, 18, 23): + at = MONDAY.replace(hour=hour, minute=0) + boundary = seconds_to_next_boundary(record, at) + assert boundary is not None and boundary > 0, ( + f"at {hour}:00 the boundary must be ahead, never zero seconds away" + ) + + +def test_waking_at_the_edge_recomputes_instead_of_finding_the_same_edge(): + record = _record() + at_edge = MONDAY.replace(hour=9, minute=0) + assert seconds_to_next_boundary(record, at_edge) == 9 * 3600, "to 18:00" + assert effective_interval(record, at_edge) == 900, "and the edge changed the answer" + + +def test_no_schedule_means_no_boundary_and_todays_frequency(): + plain = _record() + del plain["schedule"] + for hour in (0, 9, 12, 17, 23): + at = MONDAY.replace(hour=hour, minute=30) + assert effective_interval(plain, at) == 10800 + assert seconds_to_next_boundary(plain, MONDAY) is None + + +def test_a_default_record_without_a_schedule_answers_the_documented_default(): + assert effective_interval({}) == DEFAULT_INTERVAL + + +# ── four rejected inputs, each with a readable error ───────────────────────── + + +def test_an_unparsable_days_value_is_refused_by_name(): + error = validate_schedule([ + {"days": "mon-funday", "from": "09:00", "to": "18:00", "interval": 900}, + ]) + assert error and "`mon`" in error and "mon-funday" in error + + +def test_a_window_that_does_not_move_forward_is_refused(): + error = validate_schedule([ + {"days": "mon", "from": "18:00", "to": "09:00", "interval": 900}, + ]) + assert error and "does not move forward" in error and "two windows" in error + + +def test_a_window_interval_below_the_minimum_is_refused(): + error = validate_schedule([ + {"days": "mon", "from": "09:00", "to": "18:00", "interval": MIN_INTERVAL - 1}, + ]) + assert error and str(MIN_INTERVAL) in error + + +def test_two_overlapping_windows_of_one_day_are_refused(): + error = validate_schedule([ + {"days": "mon-fri", "from": "09:00", "to": "12:00", "interval": 900}, + {"days": "mon", "from": "11:00", "to": "14:00", "interval": 600}, + ]) + assert error and "overlap" in error and "mon" in error, error + + +def test_windows_on_disjoint_days_do_not_overlap(): + """The overlap rule reads the day intersection, not the clock alone.""" + assert validate_schedule([ + {"days": "mon-fri", "from": "09:00", "to": "12:00", "interval": 900}, + {"days": "sat-sun", "from": "10:00", "to": "14:00", "interval": 600}, + ]) is None + + +def test_touching_windows_are_not_an_overlap(): + assert validate_schedule([ + {"days": "mon", "from": "09:00", "to": "12:00", "interval": 900}, + {"days": "mon", "from": "12:00", "to": "14:00", "interval": 600}, + ]) is None + + +@pytest.mark.parametrize("schedule, needle", [ + ("mon-fri", "not a list"), + ([], "empty schedule list"), + (["09:00-18:00"], "not a mapping"), + ([{"days": "mon", "from": 540, "to": "18:00", "interval": 900}], "quoted string"), + ([{"days": "mon", "from": "09:00", "to": "18:00"}], "must be an integer"), +]) +def test_every_rejected_shape_names_the_value_and_the_expectation(schedule, needle): + error = validate_schedule(schedule) + assert error and needle in error, error + + +def test_an_absent_schedule_is_not_an_error(): + assert validate_schedule(None) is None + + +def test_the_loader_refuses_a_bad_window_before_the_task_starts(): + """`validate_task_record` is the reader the loader and the CRUD path share.""" + record = _record(schedule=[ + {"days": "mon", "from": "18:00", "to": "09:00", "interval": 900}, + ]) + error = validate_task_record(record) + assert error and "does not move forward" in error, error + assert validate_task_record(_record()) is None + + +# ── the record is the one source of truth for the field ────────────────────── + + +@pytest.fixture +def sched(tmp_path, monkeypatch): + """A TaskScheduler over a tasks.yml in `tmp_path`, with config_dir redirected.""" + original = mod.config_dir + monkeypatch.setattr(mod, "config_dir", lambda: tmp_path) + # A registered project, so `task_create` has something to be valid against. + (tmp_path / "projects.yml").write_text( + yaml.safe_dump([{"name": "emrg", "path": str(tmp_path)}]) + ) + try: + scheduler = TaskScheduler(InstanceIdentity()) + scheduler._tasks_file = tmp_path / "tasks.yml" + yield scheduler + finally: + mod.config_dir = original + + +WINDOW = [{"days": "mon-fri", "from": "09:00", "to": "18:00", "interval": 900}] + + +def _row(tasks_yml, name: str) -> dict: + """One record out of a written tasks.yml by name (the file is seeded, so not [0]).""" + return next(r for r in yaml.safe_load(tasks_yml.read_text()) if r.get("name") == name) + + +def test_create_and_update_carry_the_windows_through_the_file(sched): + ok, created = sched.task_create( + name="windowed", task_type="evolution", project="emrg", + interval=10800, schedule=WINDOW, + ) + assert ok, created + written = _row(sched._tasks_file, "windowed") + assert written["schedule"] == WINDOW + + ok, updated = sched.task_update("windowed", interval=7200) + assert ok, updated + # A payload that omits the field leaves the windows alone β€” which is what lets a + # client that predates the field keep editing the task (issue #2486 β‘©). + assert updated["schedule"] == WINDOW + + ok, cleared = sched.task_update("windowed", schedule=None) + assert ok, cleared + assert "schedule" not in _row(sched._tasks_file, "windowed") + + +def test_a_bad_window_is_refused_on_both_write_paths(sched): + bad = [{"days": "mon", "from": "18:00", "to": "09:00", "interval": 900}] + ok, error = sched.task_create( + name="bad", task_type="evolution", project="emrg", interval=10800, schedule=bad, + ) + assert not ok and "does not move forward" in error, error + + assert sched.task_create( + name="good", task_type="evolution", project="emrg", interval=10800, + )[0] + ok, error = sched.task_update("good", schedule=bad) + assert not ok and "does not move forward" in error, error + + +# ── the reported rows carry the frequency in force ─────────────────────────── + + +def test_list_tasks_reports_the_effective_interval_and_the_next_edge(sched): + """A record with no handler reports the same cadence keys a live one does.""" + all_day = [{"days": None, "from": "00:00", "to": "24:00", "interval": 900}] + sched._tasks_file.write_text(yaml.safe_dump([ + {"name": "always", "type": "evolution", "config": {"project": "emrg"}, + "interval": 10800, "enabled": True, "schedule": all_day}, + ])) + row = sched.list_tasks()[0] + assert row["interval"] == 10800, "the declared frequency is still reported" + assert row["effective_interval"] == 900, "and the one in force at this instant" + assert row["schedule"] == all_day + boundary = row["next_window_boundary_in_seconds"] + assert isinstance(boundary, int) and 0 < boundary <= 86400 + + +def test_a_live_handler_reports_the_same_three_keys(): + handler = _handler_with(_record()) + status = handler.status() + assert status["interval"] == 10800 + assert status["schedule"] == _record()["schedule"] + assert status["effective_interval"] == effective_interval(_record()) + assert status["next_window_boundary_in_seconds"] is not None + + +def test_the_heartbeat_and_the_backoff_take_the_frequency_in_force(): + """⑧: neither layer re-derives the rule, and neither bypasses the other.""" + handler = _handler_with(_record()) + in_force = handler._effective_interval() + assert in_force in (900, 10800), "one of the two answers the record can give" + assert handler._heartbeat_interval() == max(in_force, min(in_force * 8, 8 * 3600)) + handler._connect_failures = 0 + assert handler._connect_backoff() == float(in_force) + + +# ── the cut is wired into the loop, not only into the helper ──────────────── + + +def _handler_with(record: dict) -> TaskHandler: + handler = TaskHandler(name=record["name"], identity=InstanceIdentity()) + handler._apply_record(record) + return handler + + +async def _first_wait(handler: TaskHandler) -> float: + """Run the loop until it arms its first wait, then stop it and return that wait.""" + task = asyncio.ensure_future(handler.run()) + try: + for _ in range(300): + if handler._next_run_at is not None: + break + await asyncio.sleep(0.01) + if handler._next_run_at is None: + return -1.0 + return handler._next_run_at - time.time() + finally: + handler._running = False + handler._trigger_event.set() + try: + await asyncio.wait_for(task, 5) + except (asyncio.TimeoutError, asyncio.CancelledError): + task.cancel() + + +def _loop_wait(tmp_path, monkeypatch, boundary): + """The wait one real `run()` arms, over a real tasks.yml, with `boundary` injected.""" + original = mod.config_dir + monkeypatch.setattr(mod, "config_dir", lambda: tmp_path) + monkeypatch.setattr(mod, "seconds_to_next_boundary", lambda record, now=None: boundary) + try: + tasks_yml = tmp_path / "tasks.yml" + # An hour, so the two loop tests differ only in the boundary they inject. + tasks_yml.write_text(yaml.safe_dump([_record(interval=3600)])) + handler = TaskHandler(name="windowed", identity=InstanceIdentity()) + handler._tasks_file = tasks_yml + return asyncio.run(_first_wait(handler)) + finally: + mod.config_dir = original + + +def test_run_cuts_the_wait_to_the_next_boundary(tmp_path, monkeypatch): + """Delete the cut in `run()` and this reads ~3600 instead of ~1: it reddens.""" + wait = _loop_wait(tmp_path, monkeypatch, boundary=1.0) + assert 0 < wait <= 2, f"the wait must be cut to the 1s boundary, not the interval: {wait}" + + +def test_run_without_a_boundary_waits_the_whole_interval(tmp_path, monkeypatch): + """The same loop and the same tasks.yml, no schedule: today's wait, unchanged.""" + wait = _loop_wait(tmp_path, monkeypatch, boundary=None) + assert 3500 < wait <= 3600, f"a task with no windows waits its interval: {wait}" diff --git a/tests/test_ws_e2e.py b/tests/test_ws_e2e.py index ad56b1e1..eb0a01c5 100644 --- a/tests/test_ws_e2e.py +++ b/tests/test_ws_e2e.py @@ -3170,7 +3170,7 @@ async def _test(): try: calls = {} - def fake_create(name, task_type, project, interval=None, enabled=True, repo=None, description=None, sandbox=None): + def fake_create(name, task_type, project, interval=None, enabled=True, repo=None, description=None, sandbox=None, schedule=None): calls.update(name=name, task_type=task_type, project=project) return True, {"name": name, "type": task_type} @@ -3200,7 +3200,7 @@ async def _test(): try: calls = {} - def fake_create(name, task_type, project, interval=None, enabled=True, repo=None, description=None, sandbox=None): + def fake_create(name, task_type, project, interval=None, enabled=True, repo=None, description=None, sandbox=None, schedule=None): calls.update(name=name, task_type=task_type, project=project) return True, {"name": name, "type": task_type} @@ -3218,6 +3218,45 @@ def fake_create(name, task_type, project, interval=None, enabled=True, repo=None await cleanup() asyncio.run(_test()) + def test_task_create_passes_the_schedule_windows_through(self): + """The windows arrive at the scheduler in the shape the client sent them. + + Issue #2486 β‘©: the loader, the daemon's create path and the client all read and + write one shape, so the wire must not reshape, filter or drop the field. + """ + async def _test(): + with tempfile.TemporaryDirectory() as tmp: + cwd = Path(tmp) + server, _, cleanup = await _boot_server(cwd) + try: + calls = {} + + def fake_create(name, task_type, project, interval=None, enabled=True, + repo=None, description=None, sandbox=None, schedule=None): + calls.update(interval=interval, schedule=schedule) + return True, {"name": name, "type": task_type} + + server._scheduler = self._mock_scheduler(create_side_effect=fake_create) + ws = await connect_to_server() + window = [ + {"days": "mon-fri", "from": "09:00", "to": "18:00", "interval": 900}, + ] + try: + resp = await self._cmd(ws, { + "type": "task_create", "name": "daily", + "task_type": "evolution", "project": "emrg", + "interval": 10800, "schedule": window, + }) + assert resp["type"] == "task_result" + assert resp.get("ok") is True + assert calls["interval"] == 10800 + assert calls["schedule"] == window, "the wire must not reshape it" + finally: + await ws.close() + finally: + await cleanup() + asyncio.run(_test()) + def test_task_update_maps_task_type_to_type(self): """task_update with task_type β†’ scheduler receives fields['type'] (its internal name).""" async def _test(): From 568e16626f9e47c744ed906a6ca2494486a924c7 Mon Sep 17 00:00:00 2001 From: EMRG Evolution Date: Thu, 22 Oct 2026 18:52:23 +0800 Subject: [PATCH 2/3] emrg: the task row shows the frequency in force --- emrg/client/widgets.py | 7 ++++++- tests/test_app_task_session.py | 20 ++++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/emrg/client/widgets.py b/emrg/client/widgets.py index e5b49610..4bbbe3ea 100644 --- a/emrg/client/widgets.py +++ b/emrg/client/widgets.py @@ -905,7 +905,12 @@ def render(self, ctx): name = t.get("name", "?") running = t.get("running", False) next_in = t.get("next_run_in_seconds") - interval = t.get("interval", 0) + # Issue #2486 ⑦: the frequency in force, not the declared one β€” a + # task inside a window runs at that window's interval, and a client + # that printed the record's `interval` would describe a cadence the + # scheduler is not using. Older rows carry no `effective_interval`, so + # the declared value stays the fallback. + interval = t.get("effective_interval", t.get("interval", 0)) if running: status_str = "● RUNNING" elif next_in is not None: diff --git a/tests/test_app_task_session.py b/tests/test_app_task_session.py index 9bb6b517..47bdad68 100644 --- a/tests/test_app_task_session.py +++ b/tests/test_app_task_session.py @@ -57,6 +57,26 @@ def test_the_default_intent_is_trigger_and_says_so(): assert "trigger" in title_of(TaskSelector([make_task()])) +def test_the_task_row_shows_the_frequency_in_force_not_the_declared_one(): + """Issue #2486 ⑦: inside a window the row must show the window's interval. + + A task whose record says 3 hours and whose window says 15 minutes runs every 15 + minutes; printing the declared value would describe a cadence nothing is using. + """ + inside = TaskSelector([make_task(effective_interval=900)]) + outside = TaskSelector([make_task(effective_interval=10800)]) + older = TaskSelector([make_task()]) # a row from a daemon that predates the field + assert "(every 900s)" in " ".join( + s.text for line in inside.render(RenderContext(width=120)) for s in line.spans + ) + assert "(every 10800s)" in " ".join( + s.text for line in outside.render(RenderContext(width=120)) for s in line.spans + ) + assert "(every 300s)" in " ".join( + s.text for line in older.render(RenderContext(width=120)) for s in line.spans + ) + + def test_a_session_picker_does_not_offer_to_trigger(): """The other direction, and the point of the rant: the same list serves two commands, so a host with both open must be able to tell which one Enter is From 82375bd0ebeeb638a83bdb2fb0937369f8450bf6 Mon Sep 17 00:00:00 2001 From: EMRG Evolution Date: Thu, 22 Oct 2026 19:04:43 +0800 Subject: [PATCH 3/3] emrg: the wait tests state their record instead of reading the runner's clock --- tests/test_schedule_window.py | 61 ++++++++++++++++++++++++++++++----- 1 file changed, 53 insertions(+), 8 deletions(-) diff --git a/tests/test_schedule_window.py b/tests/test_schedule_window.py index d44a4504..2bf1ed9f 100644 --- a/tests/test_schedule_window.py +++ b/tests/test_schedule_window.py @@ -321,20 +321,24 @@ def test_list_tasks_reports_the_effective_interval_and_the_next_edge(sched): assert isinstance(boundary, int) and 0 < boundary <= 86400 -def test_a_live_handler_reports_the_same_three_keys(): +def test_a_live_handler_reports_the_same_three_keys(monkeypatch): + # The clock is frozen inside the band: `status()` reads `now()` itself, and a test + # that compared it with a second `now()` would describe the instant the suite ran at. + monkeypatch.setattr(mod, "datetime", _frozen_at(datetime(2026, 10, 19, 10, 0))) handler = _handler_with(_record()) status = handler.status() assert status["interval"] == 10800 assert status["schedule"] == _record()["schedule"] - assert status["effective_interval"] == effective_interval(_record()) + assert status["effective_interval"] == 900, "the window answers, not the record" assert status["next_window_boundary_in_seconds"] is not None -def test_the_heartbeat_and_the_backoff_take_the_frequency_in_force(): +def test_the_heartbeat_and_the_backoff_take_the_frequency_in_force(monkeypatch): """⑧: neither layer re-derives the rule, and neither bypasses the other.""" + monkeypatch.setattr(mod, "datetime", _frozen_at(datetime(2026, 10, 19, 10, 0))) handler = _handler_with(_record()) in_force = handler._effective_interval() - assert in_force in (900, 10800), "one of the two answers the record can give" + assert in_force == 900, "inside the band the frequency in force is the window's" assert handler._heartbeat_interval() == max(in_force, min(in_force * 8, 8 * 3600)) handler._connect_failures = 0 assert handler._connect_backoff() == float(in_force) @@ -369,15 +373,40 @@ async def _first_wait(handler: TaskHandler) -> float: task.cancel() -def _loop_wait(tmp_path, monkeypatch, boundary): +def _frozen_at(instant: datetime): + """A `datetime` whose `now()` answers `instant`, so a run does not read the clock. + + A test of the wait reads whatever frequency `now()` selects. Left to the real clock + a window is open or shut depending on the hour the suite happens to run at β€” which + is how this file first reddened: measured on CI at 10:54 **UTC** on a Thursday the + `mon-fri 09:00-18:00` band was open, so a record that still carried it waited its + window's 900s, while the same suite was green locally at 18:54 +08:00 where the band + was shut. The instant is stated here instead. + """ + class _Frozen(datetime): + @classmethod + def now(cls, tz=None): + return instant + return _Frozen + + +def _plain_record(interval: int = 3600) -> dict: + """A record with **no** `schedule` key at all β€” absent, not a stored null.""" + record = _record(interval=interval) + del record["schedule"] + return record + + +def _loop_wait(tmp_path, monkeypatch, boundary, record=None, freeze_at=None): """The wait one real `run()` arms, over a real tasks.yml, with `boundary` injected.""" original = mod.config_dir monkeypatch.setattr(mod, "config_dir", lambda: tmp_path) monkeypatch.setattr(mod, "seconds_to_next_boundary", lambda record, now=None: boundary) + if freeze_at is not None: + monkeypatch.setattr(mod, "datetime", _frozen_at(freeze_at)) try: tasks_yml = tmp_path / "tasks.yml" - # An hour, so the two loop tests differ only in the boundary they inject. - tasks_yml.write_text(yaml.safe_dump([_record(interval=3600)])) + tasks_yml.write_text(yaml.safe_dump([record if record is not None else _plain_record()])) handler = TaskHandler(name="windowed", identity=InstanceIdentity()) handler._tasks_file = tasks_yml return asyncio.run(_first_wait(handler)) @@ -392,6 +421,22 @@ def test_run_cuts_the_wait_to_the_next_boundary(tmp_path, monkeypatch): def test_run_without_a_boundary_waits_the_whole_interval(tmp_path, monkeypatch): - """The same loop and the same tasks.yml, no schedule: today's wait, unchanged.""" + """The same loop over a record with **no** `schedule`: today's wait, unchanged.""" wait = _loop_wait(tmp_path, monkeypatch, boundary=None) assert 3500 < wait <= 3600, f"a task with no windows waits its interval: {wait}" + + +def test_run_waits_at_the_window_the_record_states(tmp_path, monkeypatch): + """The dual of the test above: the record's window is read, not merely carried. + + Same helper, same `interval: 3600`, same injected `boundary=None` β€” only the record + differs (it keeps the `mon-fri 09:00-18:00 / 900` band) and the clock is frozen + inside that band. The wait must therefore be the window's frequency. Make + `_effective_interval` answer the declared interval again and this reads ~3600. + """ + inside = datetime(2026, 10, 19, 10, 0) # a Monday, inside the band + wait = _loop_wait( + tmp_path, monkeypatch, boundary=None, + record=_record(interval=3600), freeze_at=inside, + ) + assert 800 < wait <= 900, f"inside the window the wait is the window's interval: {wait}"