diff --git a/Agent.md b/Agent.md index 5062c351..fe16f4d3 100644 --- a/Agent.md +++ b/Agent.md @@ -42,7 +42,7 @@ What EMRG does today — capability statements, not a changelog. - **TUI client** — markdown/diffs, slash commands, autocomplete, pickers, clipboard images (`vision` per model), ESC interrupt, CJK-aware input - **Electron GUI** — streaming chat, tool-status cards, sessions, models, settings, onboarding, reconnect with daemon respawn; React + TypeScript; vitest -- **Scheduled tasks** — `TaskHandler` for any registered project: builtin and custom types, CRUD, hot reload +- **Scheduled tasks** — `TaskHandler` for any registered project: builtin and custom types, CRUD, hot reload, `when` early wake - **Rant-driven evolution** — host feedback through `/rant` drives the self-improvement cycles - **Memory** — project and session memory (YAML frontmatter + a Markdown index), compact/clear, a capped prompt footprint - **Skills** — `.emrg/skills/` with progressive disclosure + an installable-skill catalog diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 45b4bbec..40c0a6a2 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -159,6 +159,23 @@ A confined run cannot write your home directory, which is where `uv`, `pip` and 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. +**A task that runs when there is something to do** (`when` in `~/.emrg/tasks.yml`): `interval` says how often a task runs on its own initiative, and a fixed cadence is the wrong instrument for a pulse — a task polling an inbox every four hours is hours late for a message that arrived a minute after its last run, and a task on a short interval burns a whole agent round to discover that nothing arrived. `when` is an optional mapping from a built-in condition name to a number of seconds — "while this holds, run the task again at most this long after its last run": + +```yaml +- name: feishu-inbox + interval: 14400 + when: + feishu: 300 # an @, an @all, or a direct message needs the host +- name: some-evolution-task + interval: 14400 + when: + rant: 600 # this task's agent carries a rant that is not completed +``` + +It only ever advances a run: `interval` and `schedule` still decide the task's own cadence, `when` never suppresses a run the task would have taken anyway, and a record without `when` behaves exactly as before. The condition names are built in (`emrg/server/wake.py` holds the table) — a host writes a name and a number of seconds, never a command, a script or a function name, and an unknown name, a `0`, a negative number or a value under the `60`-second floor is refused with the value it refused and the shape it expects. A condition answers *holds*, *absent* or *cannot tell*; *cannot tell* — an unauthenticated `lark-cli`, an unreadable rant log — leaves the task asleep and logs the reason, because "could not read it" says nothing about whether the host is being asked for something. + +Unlike `schedule`, this is a **poll**: the scheduler already sleeps for the task's own wait, so the shortest `when` value takes a slot inside that wait and each condition is consulted on its own clock. A poll that finds nothing re-arms the remainder of the task's own cadence rather than restarting it, so `next_run_at` keeps reporting the real run and a condition cannot push that run out. A wake that would land while the task is already running is harmless — the poll happens outside the cycle, so it neither interrupts the cycle nor marks anything read, and the condition is judged again next time. `feishu` keeps a watermark under `~/.emrg/wake-state/.json` because Feishu reports message times to the minute and a time cannot be a cursor: consecutive polls overlap deliberately and a message id is what decides. The daemon's create/update path and its wire handlers carry the field, so editing another field in a client cannot drop it, and hand-editing the file remains how to set one. + --- ## 🏗️ Architecture diff --git a/emrg/server/daemon.py b/emrg/server/daemon.py index 3c6e897e..93cd3796 100644 --- a/emrg/server/daemon.py +++ b/emrg/server/daemon.py @@ -3039,6 +3039,7 @@ async def _process_message( description=msg.get("description"), sandbox=msg.get("sandbox"), schedule=msg.get("schedule"), + when=msg.get("when"), ) if not ok: await self._send(ws, {"type": "task_result", "error": res}) @@ -3054,7 +3055,12 @@ 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", "agent", "interval", "enabled", "repo", "description", "sandbox", "schedule") if k in msg} + # This list is the wire's whole vocabulary: a key it omits is a field a + # client cannot edit, whatever the scheduler supports — which is how + # `schedule` was once silently dropped (#2490). `when` is here for the same + # reason `schedule` is (issue #2592), and a key the message omits still + # leaves that field alone. + fields = {k: msg[k] for k in ("task_type", "agent", "interval", "enabled", "repo", "description", "sandbox", "schedule", "when") 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/rants.py b/emrg/server/rants.py index 12ba601f..d08e063d 100644 --- a/emrg/server/rants.py +++ b/emrg/server/rants.py @@ -38,6 +38,20 @@ } +def default_rants_log() -> Path: + """The ledger this installation writes — the one derivation of its path. + + Two readers had spelled `config_dir() / "rants.jsonl"` for themselves: the + ``submit_rant`` tool and the path policy. A third reader is arriving with the + scheduler's ``rant`` wake condition (issue #2592), and three copies of a path is how + a reader comes to watch a file nobody writes. The import is local because + ``emrg.config`` is imported by nearly everything and this module is imported early. + """ + from emrg.config import config_dir + + return config_dir() / "rants.jsonl" + + def _normalize_rant(raw) -> dict | None: """Normalize one parsed line to the canonical 6-field dict. diff --git a/emrg/server/scheduler.py b/emrg/server/scheduler.py index d565cc47..358eab84 100644 --- a/emrg/server/scheduler.py +++ b/emrg/server/scheduler.py @@ -14,6 +14,13 @@ 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. + when — optional; a mapping from a built-in condition name to seconds. While a + condition holds, the task is woken again at most that long after its last run, so a + task can keep a long `interval` and still answer a pulse. It only ever advances a + run: `interval` and `schedule` decide the task's own cadence and `when` never + suppresses one. The vocabulary is `emrg.server.wake.CONDITIONS` (`rant`, `feishu`), + and unlike `schedule` it is a **poll** — each condition is asked once per its own + seconds, inside the wait this scheduler already performs. config — type-specific config, and the mapping the templates see as ``task``. For evolution: config.agent links to agents.yml name. sandbox — optional; also accepted under config, where the record-level value wins. @@ -57,6 +64,12 @@ # every pattern has a minimum length, so a value windowed first cannot match a credential # that crosses the cut. from emrg.server.redaction import redact_prefix, redact_string +# The `when` conditions live in their own module: this file already carries the field +# schema, the windows and the loop, and a condition shells out to a CLI — a different +# kind of code from everything else here. `validate_when` is called from the one +# validator both the loader and the CRUD path use, so a record cannot be accepted by one +# and dropped by the other. +from emrg.server.wake import shortest_when_seconds, validate_when, wakes from emrg.server.git_utils import ( _detect_git_remote, ensure_local_exclude, @@ -923,6 +936,9 @@ def validate_task_record(record: dict) -> str | None: schedule_error = validate_schedule(record.get("schedule")) if schedule_error: return f"task {name!r} has {schedule_error}" + when_error = validate_when(record.get("when")) + if when_error: + return f"task {name!r} has {when_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" @@ -1161,6 +1177,14 @@ def __init__( self._next_run_file = self._next_run_dir / f"{self.name}.json" self._resume_next_run_at: float | None = self._load_next_run_state() + # ── `when` watermarks (issue #2592) ── + # ~/.emrg/wake-state/.json holds what a condition has already seen — + # when each was last consulted, and the Feishu message ids already woken on. + # It is written by `emrg.server.wake`, not here: this class only says where the + # file lives, the way `_tasks_file` says where the truth lives and not what it + # says. Created on first write, because a task with no `when` writes nothing. + self._wake_state_dir = config_dir() / "wake-state" + # ── The record this handler runs under (rant 2026-09-28T09:54:10) ── def _read_own_record(self) -> dict | None: @@ -2537,10 +2561,13 @@ async def _refresh_record(self) -> str: def _clear_runtime_state(self) -> None: """Remove this handler's own runtime files (it is not coming back). - Without this a deleted task leaves a heartbeat and a next-run slot behind, and - both read as "something is scheduled here" to anyone who looks. + Without this a deleted task leaves a heartbeat, a next-run slot and its `when` + watermark behind, and all three read as "something is scheduled here" to anyone + who looks. The watermark is `emrg.server.wake`'s file, so its path comes from + that module rather than being spelled again here. """ - for path in (self._heartbeat_file, self._next_run_file): + wake_state = self._wake_state_dir / f"{self.name}.json" + for path in (self._heartbeat_file, self._next_run_file, wake_state): try: path.unlink(missing_ok=True) except OSError as exc: @@ -2570,6 +2597,11 @@ async def run(self) -> None: # (rant 2026-09-29T09:23:55), and this loop only needs to know whether it # has already said this once. _started_logged = False + #: When this task's own cadence next comes due, as an instant. A `when` poll + #: shortens one wait but must not restart the interval it shortens: without this + #: the interval would be recomputed from now at every poll, and the task's own + #: run would never arrive at all (issue #2592). + natural_due: float | None = None while self._running: outcome = await self._refresh_record() @@ -2585,34 +2617,23 @@ async def run(self) -> None: ) _started_logged = True - # Slowdown → wait at the heartbeat interval (low-frequency full - # cycle, rant 2026-08-09T09:35:55; flag set by vibe check - # recommend_slowdown, rant 2026-08-20T10:58:55); otherwise the - # normal interval. Manual trigger wakes immediately either way. - # Never skip a cycle. - wait_timeout = ( - self._heartbeat_interval() - if self._slowdown_active - # Rant 2026-08-09T13:16:36: exponential backoff while the - # daemon is unreachable — stops the retry/window storm. - else self._connect_backoff() - ) - # Rant 2026-08-25T09:25:32 ④: a daemon restart must not reset the - # next run — fold the persisted slot into the wait (one-shot: - # 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 + # A poll re-arm resumes what is left of the task's own cadence instead of + # recomputing it. The cuts inside `_natural_wait` already shaped that + # instant; recomputing here would restart the interval at every poll and push + # the task's own run out for ever (issue #2592). + if natural_due is not None: + wait_timeout = max(1.0, natural_due - time.time()) + else: + wait_timeout = self._natural_wait() + natural_due = time.time() + wait_timeout + # `when` (issue #2592): a condition may wake this task ahead of its own + # cadence, so the condition poll takes a slot in the same wait. It only ever + # shortens that wait, and a poll that finds nothing re-arms instead of + # running — so the task's own interval is untouched in either direction. + _poll = self._wake_poll_seconds() + _polled = _poll is not None and _poll < wait_timeout + if _polled: + wait_timeout = _poll # 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. @@ -2627,7 +2648,9 @@ async def run(self) -> None: self.name, int(wait_timeout), _mode, self._effective_interval(), ) # Wait for interval or manual trigger (interruptible) - self._next_run_at = time.time() + wait_timeout + # A poll is internal: the next run reported stays the task's own cadence, so + # `next_run_at` never claims the task runs more often than it does. + self._next_run_at = natural_due if _polled else time.time() + wait_timeout self._save_next_run_state() # survives daemon restarts (rant 2026-08-25T09:25:32 ④) manual_trigger = False try: @@ -2642,8 +2665,21 @@ async def run(self) -> None: "TaskHandler[%s] manually triggered", self.name ) except asyncio.TimeoutError: - # Normal scheduled run - pass + # Normal scheduled run — unless this wait was a `when` poll, in which case + # the run has to be earned: a holding condition wakes the task, and one + # that does not re-arms the remainder of its own cadence (issue #2592). + # The poll lives here, outside the cycle, which is what makes a wake that + # arrives while the task is already running harmless — it neither + # interrupts the cycle nor marks anything read, and the condition is + # judged again at the next poll. + if _polled: + _woken, _notes = await self._condition_wake() + for _note in _notes: + self._logger.debug( + "TaskHandler[%s]: when %s", self.name, _note + ) + if not _woken: + continue if not self._running: break @@ -2669,6 +2705,10 @@ async def run(self) -> None: self._slowdown_active = False self._save_saturation_state() + # This run consumes the cadence: the next wait is computed fresh from the + # record rather than resumed from the instant this one was made for + # (issue #2592). + natural_due = None self._logger.debug("TaskHandler[%s] tick", self.name) self._cycle_running = True self._cycle_start_time = time.time() # per-cycle elapsed base (rant 2026-08-22T07:18:35) @@ -2848,6 +2888,71 @@ def _heartbeat_interval(self) -> int: interval = self._effective_interval() return max(interval, min(interval * 8, 8 * 3600)) + # ── `when`: what may wake this task ahead of its cadence (issue #2592) ── + + def _natural_wait(self) -> float: + """How long this task would wait if nothing woke it early. + + The whole cadence chain in one place: the slowdown's heartbeat, the connection + backoff, the persisted slot a restart must not reset, and the cut to the next + window edge. `run` takes this once per cycle and keeps the instant it makes, so a + `when` poll can shorten a wait without restarting the interval behind it. + + Slowdown → wait at the heartbeat interval (low-frequency full cycle, rant + 2026-08-09T09:35:55; flag set by vibe check `recommend_slowdown`, rant + 2026-08-20T10:58:55); otherwise the normal interval. A manual trigger wakes + immediately either way, and a cycle is never skipped. + """ + wait_timeout = ( + self._heartbeat_interval() + if self._slowdown_active + # Rant 2026-08-09T13:16:36: exponential backoff while the daemon is + # unreachable — stops the retry/window storm. + else self._connect_backoff() + ) + # Rant 2026-08-25T09:25:32 ④: a daemon restart must not reset the next run — fold + # the persisted slot into the wait (one-shot: 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 + return wait_timeout + + def _wake_poll_seconds(self) -> int | None: + """The tightest `when` this task states, or None when it states none. + + The wait is sized for the shortest one; each condition is still consulted on its + own clock (the watermark `wakes` keeps), so one condition's cadence does not + become another's. + """ + return shortest_when_seconds(self._task) + + async def _condition_wake(self) -> tuple[bool, list[str]]: + """Ask this record's `when` conditions whether the task should run now. + + In a worker thread: a condition reads a store or runs a CLI, and this coroutine + runs on the daemon's event loop — the same reason `_refresh_record` does. + """ + if not self._task: + return False, [] + agent = str(self._derived.get("agent_name") or self.name) + return await asyncio.to_thread( + wakes, + self._task, + task_name=self.name, + agent=agent, + state_dir=self._wake_state_dir, + ) + def _effective_interval(self) -> int: """How often this task runs right now, from its own record's windows. @@ -3766,6 +3871,7 @@ def _ensure_emrg_agent_entry(self) -> None: def _validate_task_fields( name: str, task_type: str, agent: str, interval: int, schedule: list | None = None, + when: dict | 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: @@ -3776,18 +3882,24 @@ def _validate_task_fields( return f"agent {agent!r} is not registered in agents.yml" if not isinstance(interval, int) or isinstance(interval, bool) or interval < MIN_INTERVAL: return f"interval must be an integer >= {MIN_INTERVAL}" - # The same validator the loader runs: one shape, one set of messages. - return validate_schedule(schedule) + # The same validators the loader runs: one shape, one set of messages. + schedule_error = validate_schedule(schedule) + if schedule_error: + return schedule_error + return validate_when(when) def task_create( self, name: str, task_type: str, agent: str, interval: int | None = None, enabled: bool = True, repo: str | None = None, description: str | None = None, sandbox: str | None = None, schedule: list | None = None, + when: dict | 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, agent, interval, schedule) + err = self._validate_task_fields( + name, task_type, agent, interval, schedule, when, + ) if err: return False, err if sandbox is not None and sandbox not in SANDBOX_MODES: @@ -3816,6 +3928,8 @@ def task_create( task["sandbox"] = sandbox if schedule is not None: task["schedule"] = schedule + if when is not None: + task["when"] = when if description: task["description"] = description tasks.append(task) @@ -3836,8 +3950,9 @@ def task_update(self, name: str, **fields) -> tuple[bool, str | dict]: new_agent = cfg.get("agent", "") new_interval = fields.get("interval", task.get("interval", DEFAULT_INTERVAL)) new_schedule = fields.get("schedule", task.get("schedule")) + new_when = fields.get("when", task.get("when")) err = self._validate_task_fields( - name, new_type, new_agent, new_interval, new_schedule, + name, new_type, new_agent, new_interval, new_schedule, new_when, ) if err: return False, err @@ -3864,6 +3979,18 @@ def task_update(self, name: str, **fields) -> tuple[bool, str | dict]: task.pop("schedule", None) else: task["schedule"] = fields["schedule"] + if "when" in fields: + # `when` follows `schedule` exactly, and for the same reason (issue #2490): + # an omitted key leaves the conditions alone, so a form that does not know + # the field cannot drop it; an explicit null removes it. Without this branch + # a client that *did* send `when` would be answered "ok" and lose it. + when_error = validate_when(fields["when"]) + if when_error: + return False, when_error + if fields["when"] is None: + task.pop("when", None) + else: + task["when"] = fields["when"] if "enabled" in fields: task["enabled"] = bool(fields["enabled"]) if "description" in fields: diff --git a/emrg/server/wake.py b/emrg/server/wake.py new file mode 100644 index 00000000..d268470e --- /dev/null +++ b/emrg/server/wake.py @@ -0,0 +1,614 @@ +"""The `when` conditions — what may wake a scheduled task ahead of its own cadence. + +A task carries one frequency: `interval`, or the windows `schedule` states. That is how +often it runs **on its own initiative**, and nothing here changes it. `when` adds an +optional mapping from a condition name to a number of seconds, and it only ever makes a +task run *sooner*: while a condition holds, the task is woken again at most that many +seconds later. A task without `when` is unchanged, and a `when` never suppresses a run +the task would have taken anyway (issue #2592). + +Conditions are **built in**. A host writes a condition name and a number of seconds — +never a command, a script or a function name — so `CONDITIONS` is the whole vocabulary, +and an unknown name is refused with the list of the known ones rather than ignored. + +Every evaluator answers one of three things, and the third one is not the second: + +* `HOLDS` — the condition is true; wake the task. +* `ABSENT` — the condition was read, and it is not true. +* `UNKNOWN` — it could not be read at all. This is **not** "nothing to do": an + unauthenticated CLI says nothing about whether a message is waiting. A wake policy + still has to act, so `UNKNOWN` leaves the task asleep and logs the reason. It never + raises, never blocks another condition, and never stops the task's own rhythm. + +The watch is a poll: each condition is consulted once per its own configured seconds, +inside the wait the scheduler already performs. There is no second, event-push path, and +no writer anywhere is modified to announce anything — that is what keeps the pop-up of a +rant out of the scheduler's business (issue #2592, defaults 3, 7). +""" + +from __future__ import annotations + +import json +import logging +import subprocess +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Callable, Sequence + +from emrg.server.rants import default_rants_log, list_rants +from emrg.tools.base import non_regular_kind + +logger = logging.getLogger("emrg.server.wake") + +#: The floor on a `when` value, and the same bound `scheduler.MIN_INTERVAL` puts on +#: `interval`: a condition may not be polled faster than a task is allowed to run, or +#: the field would become a way around that bound. `tests/test_scheduler.py` asserts the +#: two numbers agree, so the second carrier cannot drift from the first in silence. +MIN_WHEN_SECONDS = 60 + +HOLDS = "holds" +ABSENT = "absent" +UNKNOWN = "unknown" + +#: The CLI that speaks the Feishu IM API, and its own patience. Resolved from PATH at +#: call time, never stored: a deployment that has never installed it is a condition that +#: answers `UNKNOWN`, not an error at import. +LARK_CLI = "lark-cli" +_CLI_TIMEOUT = 120 + +#: How many of the host's most recently active conversations one poll reads. Each chat +#: is one CLI call in the foreground, so this is a budget on wall-clock time, not on +#: correctness: a chat outside the window is read on a poll where it is more recent. +FEISHU_MAX_CHATS = 5 + +#: The window each Feishu poll re-reads. Feishu reports message times to the minute, so +#: a time can never be a cursor: consecutive polls overlap deliberately and the message +#: id is what decides, which is why `seen` exists at all (issue #2592, default 4). +FEISHU_LOOKBACK_SECONDS = 900 + +#: How many message ids the watermark remembers. Bounded because the file is read and +#: rewritten on every poll: older ids fall out, and the lookback window is what makes +#: that safe — an id older than the window is never re-read in the first place. +FEISHU_SEEN_CAP = 500 + + +class ConditionUnavailable(Exception): + """A condition could not be read. Never fatal, never a wake, never silent.""" + + +@dataclass(frozen=True) +class Verdict: + """What one condition answered, and why — the reason is what the log carries.""" + + state: str + reason: str = "" + + @property + def holds(self) -> bool: + return self.state == HOLDS + + +# ── the field's own rules ──────────────────────────────────────────────── + + +def _is_seconds(value) -> bool: + """A `when` value: a positive whole number of seconds at or above the floor.""" + return ( + isinstance(value, int) + and not isinstance(value, bool) + and value >= MIN_WHEN_SECONDS + ) + + +def validate_when(when) -> str | None: + """Return why a `when` field cannot be run, or None when it is usable. + + One function for both the loader and the CRUD path, like `validate_schedule`: a + record refused by one and accepted by the other is how a task ends up in the file + that the daemon silently drops. Every refusal names the value it refused and the + shape it expects, because "invalid when" leaves a host with a file and no next step. + """ + if when is None: + return None + if not isinstance(when, dict): + return ( + f"when of type {type(when).__name__}, not a mapping of condition to seconds" + ) + if not when: + return "an empty when mapping - drop the field for one cadence all day" + known = ", ".join(sorted(CONDITIONS)) + for name, seconds in when.items(): + if not isinstance(name, str) or name not in CONDITIONS: + return f"unknown when condition {name!r} (known: {known})" + if not _is_seconds(seconds): + return ( + f"when condition {name!r} has {seconds!r} seconds " + f"(must be an integer >= {MIN_WHEN_SECONDS})" + ) + return None + + +def normalised_when(record) -> dict[str, int]: + """The usable pairs of a record's `when` — what scheduling may act on. + + Unusable is not an exception path here: `validate_when` is what refuses a bad field + with a readable error, and this is what a record that slipped past every writer + falls back to. Dropping one bad pair is the choice that keeps the other conditions + working, and it matches `_normalise_schedule`'s treatment of a broken window. + """ + when = record.get("when") if isinstance(record, dict) else None + if not isinstance(when, dict): + return {} + return { + name: seconds + for name, seconds in when.items() + if isinstance(name, str) and name in CONDITIONS and _is_seconds(seconds) + } + + +def shortest_when_seconds(record) -> int | None: + """The tightest `when` in this record, or None when it states none. + + The scheduler waits for the shortest one and consults each condition on its own + clock inside that wait, so a slow condition never holds up a fast one. + """ + seconds = list(normalised_when(record).values()) + return min(seconds) if seconds else None + + +# ── the watermark these conditions keep ────────────────────────────────── + + +def state_path(state_dir: Path, task_name: str) -> Path: + return Path(state_dir) / f"{task_name}.json" + + +def _refuse_non_regular(path: Path) -> str | None: + """The kind of `path` when it is not a regular file, else None. + + Asked **before** opening, because opening is what decides whether the caller returns + at all: a FIFO blocks at the open until a peer appears and raises nothing, so no + `except OSError` bounds it — and this runs on the daemon's path, where that block is + the whole daemon and the only recovery is a restart, which belongs to the host alone. + + Absence is not a refusal. `non_regular_kind` **stats**, so a watermark that has not + been written yet arrives as an `OSError`, and that is the ordinary first-run case + this returns None for. + """ + try: + return non_regular_kind(path) + except OSError: + return None + + +def load_state(state_dir: Path, task_name: str) -> dict: + """The persisted watermark, or `{}`. + + An unreadable file is a fresh start with a log line, never a refusal: this state is + an optimisation that prevents a duplicate wake, and losing it costs one redundant + poll rather than a wrong answer. + """ + path = state_path(state_dir, task_name) + kind = _refuse_non_regular(path) + if kind: + logger.warning("wake state %s is a %s, not a regular file — starting over", path, kind) + return {} + try: + raw = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + return {} + try: + data = json.loads(raw) + except json.JSONDecodeError: + logger.warning("wake state %s is not JSON — starting it over", path) + return {} + return data if isinstance(data, dict) else {} + + +def save_state(state_dir: Path, task_name: str, state: dict) -> None: + """Write the watermark atomically; a failure is logged and nothing more.""" + path = state_path(state_dir, task_name) + tmp = path.with_suffix(".json.tmp") + kind = _refuse_non_regular(tmp) + if kind: + # Writing to a FIFO blocks for a reader exactly as reading one blocks for a + # writer; the watermark is an optimisation, so a planted one is skipped, not + # opened at the daemon's expense. + logger.warning("wake state %s is a %s, not a regular file — not written", tmp, kind) + return + try: + path.parent.mkdir(parents=True, exist_ok=True) + tmp.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8") + tmp.replace(path) + except OSError as exc: + logger.warning("wake state %s could not be written: %s", path, exc) + + +def _aware(moment: datetime) -> datetime: + """`moment` as an aware instant in UTC. A naive one is read as local time. + + Every instant this module compares or sends goes through here, for two reasons that + are one reason. **Sending**: the stamp handed to `lark-cli --start` must name its + zone — that CLI's own help offers `2026-01-01T15:04:05+08:00`, and a bare local + instant leaves the zone to the reader, which is the one thing a boundary may not do. + `scheduler.py` already renders the templates' `{{ timestamp }}` as + `datetime.now(timezone.utc).isoformat()`, so an unaware stamp here would be this + module disagreeing with the file that states the convention. **Comparing**: the + watermark outlives the code that wrote it, and `aware - naive` raises `TypeError` + rather than being wrong in a way a log would show; normalising both sides in one + place is what makes that unrepresentable instead of unlikely. + """ + return moment.astimezone(timezone.utc) + + +def _instant(text) -> datetime | None: + """`text` as an aware instant, or None when it is not one.""" + if not isinstance(text, str): + return None + try: + return _aware(datetime.fromisoformat(text)) + except ValueError: + return None + + +# ── the conditions ─────────────────────────────────────────────────────── + + +def _condition_rant(*, agent: str, state: dict, now: datetime) -> Verdict: + """Does this task's agent carry a rant that is not `completed`? + + Read through `rants.list_rants`, the store's own reader, so a legacy array row and + the ledger's kind check are answered in one place. Nothing here writes: a condition + that could mark a rant read would race the agent that is working it. + """ + if not agent: + return Verdict(UNKNOWN, "the task names no agent to look up") + try: + rows = list_rants(default_rants_log(), agent=agent) + except OSError as exc: + return Verdict(UNKNOWN, f"the rant log could not be read: {exc}") + open_rows = [row for row in rows if row.get("status") != "completed"] + if open_rows: + return Verdict( + HOLDS, f"{len(open_rows)} rant(s) not completed for {agent}" + ) + return Verdict(ABSENT, f"no rant is open for {agent}") + + +def _condition_feishu(*, state: dict, now: datetime, runner, cli: str) -> Verdict: + """Does anything on Feishu need the host: a real @, an @all, or a direct message? + + All three count as one question — "is the host being asked for something" — so one + condition answers it rather than three the host would have to remember to list. + """ + try: + identity = _cli_json(["auth", "status", "--json"], runner=runner, cli=cli) + host = _host_open_id(identity) + if host is None: + return Verdict( + UNKNOWN, "auth status carried no identities.user.openId" + ) + chats = _chat_items( + _cli_json( + [ + "im", "+chat-list", "--types=p2p,group", "--sort=active_time", + "--as", "user", "--page-size", str(FEISHU_MAX_CHATS), "--json", + ], + runner=runner, cli=cli, + ) + ) + except ConditionUnavailable as exc: + return Verdict(UNKNOWN, str(exc)) + + if not chats: + # An empty list is not the same statement as "nothing was sent": it is what a + # missing p2p scope, a token that has not refreshed and an account in no chats + # all answer alike, and only one of those means the host has nothing waiting. + return Verdict(UNKNOWN, "the conversation list came back empty") + + seen = {value for value in (state.get("seen") or []) if isinstance(value, str)} + since = _instant(state.get("since")) or (now - timedelta(seconds=FEISHU_LOOKBACK_SECONDS)) + hits = 0 + for chat in chats[:FEISHU_MAX_CHATS]: + chat_id = _chat_id(chat) + if not chat_id: + continue + try: + reply = _cli_json( + [ + "im", "+chat-messages-list", "--chat-id", chat_id, + "--start", since.isoformat(), "--order", "asc", "--as", "user", + "--no-reactions", "--json", + ], + runner=runner, cli=cli, + ) + except ConditionUnavailable as exc: + # One unreadable conversation makes the whole answer unknown: the message + # that matters may be in exactly the one that could not be read. + return Verdict(UNKNOWN, f"conversation {chat_id}: {exc}") + for message in _message_items(reply): + message_id = _message_id(message) + if message_id is None or message_id in seen: + continue + if _needs_the_host(message, host=host, direct=_is_direct(chat)): + hits += 1 + seen.add(message_id) + + state["seen"] = sorted(seen)[-FEISHU_SEEN_CAP:] + state["since"] = ( + now - timedelta(seconds=FEISHU_LOOKBACK_SECONDS) + ).isoformat() + if hits: + return Verdict(HOLDS, f"{hits} message(s) need the host") + return Verdict(ABSENT, "nothing new needs the host") + + +#: The whole vocabulary a `when` may name. The key is what a host writes in `tasks.yml`. +CONDITIONS: dict[str, Callable[..., Verdict]] = { + "rant": _condition_rant, + "feishu": _condition_feishu, +} + +#: What each condition needs handed to it, because they do not share an argument list: +#: `rant` reads a store and `feishu` shells out to a CLI through an injected runner. +_CONDITION_KWARGS = { + "rant": ("agent", "state", "now"), + "feishu": ("state", "now", "runner", "cli"), +} + + +def wakes( + record, + *, + task_name: str, + agent: str, + state_dir: Path, + runner: Callable[[Sequence[str]], object] | None = None, + cli: str = LARK_CLI, + now: datetime | None = None, +) -> tuple[bool, list[str]]: + """Consult this record's `when` conditions; return (does any hold, notes). + + Each condition is consulted only once its own seconds have elapsed since the last + time it was consulted, so a `when: {rant: 600, feishu: 300}` asks the rant store + every ten minutes and Feishu every five, and neither waits on the other. The + watermark is written after the round of checks, so a condition that could not answer + is retried on the next poll rather than counted as read. + + A condition that raises is caught here: no evaluator may end the task's wait, and a + bug in one must not take the scheduler's loop down with it. + """ + when = normalised_when(record) + if not when: + return False, [] + moment = _aware(now) if now is not None else datetime.now(timezone.utc) + state = load_state(state_dir, task_name) + checked = state.get("checked") + if not isinstance(checked, dict): + checked = {} + runner = runner or default_runner + + holds = False + notes: list[str] = [] + consulted = False + for name in sorted(when): + seconds = when[name] + last = _instant(checked.get(name)) + if last is not None and (moment - last).total_seconds() < seconds: + continue + kwargs = {} + for key in _CONDITION_KWARGS[name]: + kwargs[key] = { + "agent": agent, "state": state, "now": moment, + "runner": runner, "cli": cli, + }[key] + try: + verdict = CONDITIONS[name](**kwargs) + except Exception as exc: # noqa: BLE001 — a condition must never end the wait + verdict = Verdict(UNKNOWN, f"the condition raised {type(exc).__name__}: {exc}") + consulted = True + notes.append(f"{name}: {verdict.state}" + (f" — {verdict.reason}" if verdict.reason else "")) + if verdict.holds: + holds = True + if verdict.state != UNKNOWN: + # A condition that could not be read keeps its old stamp, so the next poll + # asks again instead of waiting out a whole interval on a failed read. + checked[name] = moment.isoformat() + + state["checked"] = checked + if consulted: + save_state(state_dir, task_name, state) + return holds, notes + + +# ── the CLI, and the shapes it answers ─────────────────────────────────── + + +def default_runner(argv: Sequence[str]): + """Run one CLI call in the foreground and return the completed process. + + Foreground, one at a time, and never backgrounded: a poll that detached would leave + a condition nobody reads the answer of. + """ + return subprocess.run( + list(argv), capture_output=True, text=True, encoding="utf-8", + errors="replace", timeout=_CLI_TIMEOUT, check=False, + ) + + +def _cli_json(args: Sequence[str], *, runner, cli: str) -> dict: + """One CLI call, parsed. Every failure is `ConditionUnavailable`.""" + argv = [cli, *args] + try: + done = runner(argv) + except OSError as exc: + raise ConditionUnavailable(f"{cli} could not be run: {exc}") from exc + except subprocess.TimeoutExpired as exc: + raise ConditionUnavailable(f"{cli} did not answer in time: {exc}") from exc + if getattr(done, "returncode", 1) != 0: + detail = (done.stderr or done.stdout or "").strip().splitlines() + raise ConditionUnavailable( + f"{cli} exited {done.returncode}: " + f"{detail[-1] if detail else 'no output'}" + ) + try: + data = json.loads(done.stdout or "{}") + except json.JSONDecodeError as exc: + raise ConditionUnavailable(f"{cli} answered something that is not JSON: {exc}") from exc + if not isinstance(data, dict): + raise ConditionUnavailable(f"{cli} answered {type(data).__name__}, not an object") + if data.get("ok") is False: + error = data.get("error") + raise ConditionUnavailable(f"{cli} refused: {error!r}") + return data + + +def _host_open_id(identity: dict) -> str | None: + """The host's own open id, read at run time so no account constant is stored.""" + identities = identity.get("identities") + if not isinstance(identities, dict): + return None + user = identities.get("user") + if not isinstance(user, dict): + return None + for key in ("openId", "open_id"): + value = user.get(key) + if isinstance(value, str) and value.strip(): + return value + return None + + +def _payload(data: dict) -> dict: + """The `data` object of an answer, which is where the lists live.""" + inner = data.get("data") + return inner if isinstance(inner, dict) else data + + +def _chat_items(data: dict) -> list[dict]: + """The conversations an answer lists; a missing list is not an empty one.""" + payload = _payload(data) + for key in ("items", "chats"): + value = payload.get(key) + if isinstance(value, list): + return [item for item in value if isinstance(item, dict)] + raise ConditionUnavailable(f"the conversation list carried no items: {sorted(payload)}") + + +def _message_items(data: dict) -> list[dict]: + """The messages an answer lists. `+chat-messages-list` answers `messages`.""" + payload = _payload(data) + for key in ("messages", "items"): + value = payload.get(key) + if isinstance(value, list): + return [item for item in value if isinstance(item, dict)] + raise ConditionUnavailable(f"the message list carried no messages: {sorted(payload)}") + + +def _chat_id(chat: dict) -> str | None: + for key in ("chat_id", "chatId", "id"): + value = chat.get(key) + if isinstance(value, str) and value: + return value + return None + + +def _is_direct(chat: dict) -> bool: + """Is this a one-to-one conversation? + + Unknown answers **False** on purpose. A group read as direct would wake the task for + every message in it, while a direct read as a group only misses a message with no + mention in it — and the mention form is the one the other two cases cover. A + deployment whose chat list omits the mode therefore loses the quieter half of one + case rather than gaining noise in every conversation. + """ + for key in ("chat_mode", "chatMode", "chat_type", "chatType", "mode", "type"): + value = chat.get(key) + if isinstance(value, str): + return value.strip().lower() in ("p2p", "single", "direct") + return False + + +def _message_id(message: dict) -> str | None: + for key in ("message_id", "messageId", "id"): + value = message.get(key) + if isinstance(value, str) and value: + return value + return None + + +def _sender_of(message: dict) -> str | None: + sender = message.get("sender") + if isinstance(sender, dict): + for inner in ("id", "open_id", "openId"): + value = sender.get(inner) + if isinstance(value, str) and value: + return value + for key in ("sender_id", "senderId"): + value = message.get(key) + if isinstance(value, str) and value: + return value + return None + + +def _sender_is_app(message: dict) -> bool: + """Was this sent by an application — this task's own bot, above all? + + Load-bearing: the bot writes to the host's own conversation, so a direct message + with no sender filter would read as "the host is being asked something" every time + the task itself answered him, and the task would wake itself in a loop. + """ + sender = message.get("sender") + for holder in (sender, message): + if not isinstance(holder, dict): + continue + for key in ("sender_type", "senderType"): + value = holder.get(key) + if isinstance(value, str): + return value.strip().lower() == "app" + return False + + +def _mentions_of(message: dict) -> list[dict]: + mentions = message.get("mentions") + if not isinstance(mentions, list): + return [] + return [item for item in mentions if isinstance(item, dict)] + + +def _mention_open_id(mention: dict) -> str | None: + inner = mention.get("id") + if isinstance(inner, dict): + for key in ("open_id", "openId", "id"): + value = inner.get(key) + if isinstance(value, str) and value: + return value + elif isinstance(inner, str) and inner: + return inner + for key in ("open_id", "openId"): + value = mention.get(key) + if isinstance(value, str) and value: + return value + return None + + +def _is_all_mention(mention: dict) -> bool: + key = mention.get("key") + if isinstance(key, str) and key.strip().lower() in ("@_all", "all"): + return True + if _mention_open_id(mention) in ("all", "@_all"): + return True + name = mention.get("name") + return isinstance(name, str) and name.strip().lower() in ("all", "所有人", "全体成员") + + +def _needs_the_host(message: dict, *, host: str, direct: bool) -> bool: + """Is this message a request the host has to look at?""" + if _sender_is_app(message): + return False + if _sender_of(message) == host: + return False + for mention in _mentions_of(message): + if _is_all_mention(mention) or _mention_open_id(mention) == host: + return True + return direct diff --git a/emrg/tools/submit_rant_tool.py b/emrg/tools/submit_rant_tool.py index c594e723..6f6bfd3c 100644 --- a/emrg/tools/submit_rant_tool.py +++ b/emrg/tools/submit_rant_tool.py @@ -38,6 +38,7 @@ from emrg.server.rants import ( append_rant, cleanup_rants, + default_rants_log, list_rants, update_rant, ) @@ -265,8 +266,10 @@ async def execute(self, arguments: dict) -> ToolResult: return self._execute_cleanup(arguments) def _rants_log(self) -> Path: - from emrg.config import config_dir - return config_dir() / "rants.jsonl" + # The store owns this path (`rants.default_rants_log`): the scheduler's `rant` + # wake condition reads the same file, and two spellings of it would let the + # tool write a ledger nobody watches (issue #2592). + return default_rants_log() def _execute_submit(self, arguments: dict) -> ToolResult: message = str(arguments.get("message", "")).strip() diff --git a/tests/test_wake.py b/tests/test_wake.py new file mode 100644 index 00000000..1688f3c8 --- /dev/null +++ b/tests/test_wake.py @@ -0,0 +1,595 @@ +"""The `when` field: its validation, its two conditions, and the loop that obeys it. + +Three subjects, and each has its own section below. + +* the field itself — what is accepted, and what is refused with which message; +* the conditions — `rant` reads a store, `feishu` reads a CLI, and both answer + *holds* / *absent* / *cannot tell*, where the third is never the second; +* the wait — a poll may shorten a task's own cadence but must never restart it, and a + poll that finds nothing must not run the task (issue #2592). + +The `feishu` half is driven through an injected runner rather than the real `lark-cli`: +a test that authenticated against the host's own account would read his messages, and +one that did not would prove nothing. The argv it is handed is asserted, so the shape +the real CLI receives is still pinned. +""" + +from __future__ import annotations + +import asyncio +import json +from datetime import datetime, timedelta + +import yaml + +from emrg.protocol import InstanceIdentity +from emrg.server import wake as wake_mod +from emrg.server.scheduler import MIN_INTERVAL, TaskScheduler, validate_task_record +from emrg.server.wake import ( + MIN_WHEN_SECONDS, + shortest_when_seconds, + validate_when, + wakes, +) +from tests.task_handler_factory import make_handler + +HOST = "ou_host" + + +# ── harness ────────────────────────────────────────────────────────────── + + +class _Done: + """What `subprocess.run` answers, for a runner that never runs anything.""" + + def __init__(self, payload=None, returncode: int = 0, stderr: str = "", raw=None): + self.returncode = returncode + self.stdout = raw if raw is not None else json.dumps(payload or {}) + self.stderr = stderr + + +def _feishu_runner(*, chats, messages, identity=None, fail=()): + """A `lark-cli` stand-in: route each argv to a payload, and record what it saw. + + `messages` maps a chat id to the list `+chat-messages-list` should answer. `fail` + holds the argv fragments whose call should exit non-zero, so an unreadable answer is + a scenario rather than an accident. + """ + seen: list[list[str]] = [] + + def runner(argv): + seen.append(list(argv)) + joined = " ".join(argv) + for fragment in fail: + if fragment in joined: + return _Done(returncode=1, stderr=f"refused: {fragment}") + if "auth" in argv: + return _Done(identity if identity is not None else { + "identities": {"user": {"openId": HOST}}, + }) + if "+chat-list" in joined: + return _Done({"data": {"chats": chats}}) + if "+chat-messages-list" in joined: + chat_id = argv[argv.index("--chat-id") + 1] + return _Done({"data": {"messages": messages.get(chat_id, [])}}) + return _Done(returncode=1, stderr=f"unexpected call: {joined}") + + runner.seen = seen + return runner + + +def _message(message_id, *, sender="ou_stranger", sender_type="user", mentions=()): + """A message in the shape `+chat-messages-list` answers. + + The sender defaults to a stranger: the host's own messages are excluded by design, so + a host-sender default would make every case below read as noise for the wrong reason. + """ + return { + "message_id": message_id, + "sender": {"id": sender, "sender_type": sender_type}, + "mentions": list(mentions), + } + + +def _mention(open_id=None, *, name=None, key=None): + entry: dict = {} + if open_id is not None: + entry["id"] = {"open_id": open_id} + if name is not None: + entry["name"] = name + if key is not None: + entry["key"] = key + return entry + + +def _record(**when) -> dict: + return {"name": "t", "type": "evolution", "interval": 14400, "when": dict(when)} + + +# ── the field ──────────────────────────────────────────────────────────── + + +def test_a_record_without_when_is_accepted_unchanged(): + """The field is optional: no `when` means no refusal and no behaviour change.""" + assert validate_when(None) is None + assert validate_task_record({"name": "t", "type": "evolution"}) is None + assert shortest_when_seconds({"name": "t"}) is None + assert shortest_when_seconds({"name": "t", "when": None}) is None + + +def test_a_usable_when_is_accepted_on_the_one_validation_path(): + """The loader and the CRUD path share `validate_task_record`, so one test covers both.""" + assert validate_when({"rant": 600}) is None + assert validate_when({"rant": 600, "feishu": 300}) is None + assert validate_when({"rant": MIN_WHEN_SECONDS}) is None + assert validate_task_record(_record(rant=600)) is None + + +def test_every_bad_when_value_is_refused_with_its_own_reason(): + """Four classes of bad input, each refused — never silently ignored. + + The messages are asserted, not just "some error": a refusal a host cannot act on is + the defect `validate_schedule`'s docstring names, and a value that is dropped + without a word turns a mis-set field into a task that never wakes early. + """ + cases = [ + ({"nope": 300}, "unknown when condition 'nope'"), + ({"rant": 0}, "has 0 seconds"), + ({"rant": -60}, "has -60 seconds"), + ({"rant": MIN_WHEN_SECONDS - 1}, f"must be an integer >= {MIN_WHEN_SECONDS}"), + ] + for value, expected in cases: + error = validate_when(value) + assert error is not None, value + assert expected in error, f"{value!r} answered {error!r}" + + # And the record-level path reports it under the task's own name. + error = validate_task_record(_record(nope=300)) + assert error is not None and "unknown when condition 'nope'" in error, error + + +def test_a_when_that_is_not_a_mapping_or_is_empty_is_refused(): + """`when: 'rant'` and `when: {}` are both refused, each with what it expects.""" + not_a_mapping = validate_when("rant") + assert not_a_mapping is not None and "not a mapping" in not_a_mapping + + empty = validate_when({}) + assert empty is not None and "empty when mapping" in empty + + boolean = validate_when({"rant": True}) + assert boolean is not None and "has True seconds" in boolean, ( + "a bool is an int in Python; True must not pass as one second" + ) + + +def test_the_when_floor_is_the_interval_floor(): + """One rule, two carriers: the two floors must not drift apart in silence. + + `MIN_INTERVAL` bounds `interval` and `MIN_WHEN_SECONDS` bounds a `when` value, and + they are the same bound for the same reason — a condition may not poll faster than a + task is allowed to run. Two literals in two modules is exactly the shape that goes + stale, so the agreement is asserted rather than assumed. + """ + assert MIN_WHEN_SECONDS == MIN_INTERVAL + + +# ── the conditions ─────────────────────────────────────────────────────── + + +def test_the_rant_condition_reads_the_store_for_this_tasks_agent(tmp_path, monkeypatch): + """`rant` holds while the task's own agent carries a rant that is not completed.""" + log = tmp_path / "rants.jsonl" + log.write_text( + "\n".join( + json.dumps(row) for row in [ + {"timestamp": "2026-10-26T10:00:00+08:00", "agent": "other", + "status": "pending", "progress": "", "completed": None, "message": "x"}, + ] + ) + "\n", + encoding="utf-8", + ) + monkeypatch.setattr(wake_mod, "default_rants_log", lambda: log) + + state_dir = tmp_path / "wake-state" + held, notes = wakes(_record(rant=600), task_name="t", agent="t", state_dir=state_dir) + assert held is False, notes + assert any("rant: absent" in note for note in notes), notes + + # The same store, now carrying this agent's rant: it holds. + log.write_text( + log.read_text(encoding="utf-8") + + json.dumps({"timestamp": "2026-10-26T11:00:00+08:00", "agent": "t", + "status": "in_progress", "progress": "", "completed": None, + "message": "y"}) + "\n", + encoding="utf-8", + ) + # The condition's own clock has not come round yet — that is the next test's subject, + # so this one starts the watermark over rather than waiting ten minutes for it. + (state_dir / "t.json").unlink() + held, notes = wakes(_record(rant=600), task_name="t", agent="t", state_dir=state_dir) + assert held is True, notes + assert any("rant: holds" in note for note in notes), notes + + +def test_a_condition_that_holds_is_not_marked_read_by_a_wake(tmp_path, monkeypatch): + """A wake does not consume the condition: it still holds at the next consultation. + + Requirement 4 of the issue, at the level the condition can see it. A condition that + marked itself read on a wake would swallow the work the wake was for — the task runs + once for a rant and the rant is then invisible to it for ever. + """ + log = tmp_path / "rants.jsonl" + log.write_text(json.dumps({"timestamp": "2026-10-26T11:00:00+08:00", "agent": "t", + "status": "pending", "progress": "", "completed": None, + "message": "y"}) + "\n", encoding="utf-8") + monkeypatch.setattr(wake_mod, "default_rants_log", lambda: log) + state_dir = tmp_path / "wake-state" + + start = datetime(2026, 10, 26, 12, 0, 0) + first, _ = wakes(_record(rant=60), task_name="t", agent="t", + state_dir=state_dir, now=start) + # An hour later the condition's clock has come round again, and the rant is still open. + second, notes = wakes(_record(rant=60), task_name="t", agent="t", + state_dir=state_dir, now=start + timedelta(hours=1)) + assert first is True + assert second is True, notes + + +def test_each_condition_keeps_its_own_clock(tmp_path, monkeypatch): + """`rant: 600` and `feishu: 300` are consulted on their own cadences. + + Requirement 2's "各自按自己的秒数计时互不影响": the rant store is asked once per ten + minutes even though the wait is sized for the five-minute condition. + """ + log = tmp_path / "rants.jsonl" + log.write_text(json.dumps({"timestamp": "2026-10-26T11:00:00+08:00", "agent": "t", + "status": "pending", "progress": "", "completed": None, + "message": "y"}) + "\n", encoding="utf-8") + monkeypatch.setattr(wake_mod, "default_rants_log", lambda: log) + # A readable answer, so this condition is stamped and its clock can be compared with + # `rant`'s: an unreadable one is deliberately retried and never stamped (next test). + runner = _feishu_runner(chats=[{"chat_id": "oc_g", "chat_mode": "group"}], + messages={"oc_g": []}) + state_dir = tmp_path / "wake-state" + record = _record(rant=600, feishu=300) + + start = datetime(2026, 10, 26, 12, 0, 0) + wakes(record, task_name="t", agent="t", state_dir=state_dir, runner=runner, now=start) + first_calls = len(runner.seen) + assert first_calls > 0, "the first consultation must ask both conditions" + + # Four minutes later: neither condition's clock has come round, so nothing is asked. + wakes(record, task_name="t", agent="t", state_dir=state_dir, runner=runner, + now=start + timedelta(minutes=4)) + assert len(runner.seen) == first_calls, "no condition is due four minutes in" + + # Six minutes in, only `feishu` (300 s) is due — `rant` (600 s) is not. + wakes(record, task_name="t", agent="t", state_dir=state_dir, runner=runner, + now=start + timedelta(minutes=6)) + assert len(runner.seen) > first_calls, "the 300-second condition must be asked again" + + state = json.loads((state_dir / "t.json").read_text(encoding="utf-8")) + assert set(state["checked"]) == {"rant", "feishu"}, state["checked"] + + +def test_a_condition_that_cannot_be_read_is_not_a_quiet_one(tmp_path, monkeypatch): + """`UNKNOWN` never wakes the task, and never counts as `ABSENT` either. + + An unauthenticated CLI says nothing about whether a message is waiting, so the task + stays asleep — but the reason is logged and the condition's stamp is *not* advanced, + so the next poll asks again instead of waiting out a whole interval on a failed read. + """ + monkeypatch.setattr(wake_mod, "default_rants_log", lambda: tmp_path / "missing.jsonl") + runner = _feishu_runner(chats=[], messages={}, identity={"identities": {}}) + state_dir = tmp_path / "wake-state" + start = datetime(2026, 10, 26, 12, 0, 0) + + held, notes = wakes(_record(feishu=300), task_name="t", agent="t", + state_dir=state_dir, runner=runner, now=start) + assert held is False, notes + assert any("feishu: unknown" in note for note in notes), notes + + state = json.loads((state_dir / "t.json").read_text(encoding="utf-8")) + assert "feishu" not in state["checked"], ( + "an unreadable condition must be retried on the next poll, not stamped as read" + ) + + +def test_the_feishu_condition_answers_the_three_cases_that_need_the_host(tmp_path): + """A real @, an @all, and a direct message all hold; none of them is a mention of the host.""" + cases = { + "a real mention of the host": ([_message("m1", mentions=[_mention(HOST)])], []), + "an @all": ([_message("m2", mentions=[_mention(key="@_all")])], []), + "a direct message": ([], [_message("m3", sender="ou_stranger")]), + } + for label, (group_messages, direct_messages) in cases.items(): + state_dir = tmp_path / label.replace(" ", "-") + runner = _feishu_runner( + chats=[{"chat_id": "oc_group", "chat_mode": "group"}, + {"chat_id": "oc_direct", "chat_mode": "p2p"}], + messages={"oc_group": group_messages, "oc_direct": direct_messages}, + ) + held, notes = wakes(_record(feishu=300), task_name="t", agent="t", + state_dir=state_dir, runner=runner) + assert held is True, f"{label} should wake the task: {notes}" + + +def test_the_feishu_condition_ignores_noise_and_the_tasks_own_messages(tmp_path): + """A group message that asks for nothing, and the bot's own reply, both leave it asleep. + + The bot writes into the host's own conversation, so a direct chat with no sender + filter would read every answer this very task sent as a fresh request — the task + waking itself in a loop. `sender_type: app` is what excludes it. + """ + runner = _feishu_runner( + chats=[{"chat_id": "oc_group", "chat_mode": "group"}, + {"chat_id": "oc_direct", "chat_mode": "p2p"}], + messages={ + "oc_group": [_message("m1", sender="ou_stranger")], + "oc_direct": [ + _message("m2", sender="ou_bot", sender_type="app"), + _message("m3", sender=HOST), + ], + }, + ) + held, notes = wakes(_record(feishu=300), task_name="t", agent="t", + state_dir=tmp_path / "wake-state", runner=runner) + assert held is False, notes + assert any("feishu: absent" in note for note in notes), notes + + +def test_the_feishu_condition_de_duplicates_by_message_id(tmp_path): + """One message wakes the task once, however many polls re-read it. + + Feishu reports message times to the minute, so consecutive polls overlap on purpose + and a time can never be the cursor. The id is. + """ + message = _message("om_same", mentions=[_mention(HOST)]) + runner = _feishu_runner( + chats=[{"chat_id": "oc_group", "chat_mode": "group"}], + messages={"oc_group": [message]}, + ) + state_dir = tmp_path / "wake-state" + start = datetime(2026, 10, 26, 12, 0, 0) + + first, _ = wakes(_record(feishu=300), task_name="t", agent="t", + state_dir=state_dir, runner=runner, now=start) + second, notes = wakes(_record(feishu=300), task_name="t", agent="t", + state_dir=state_dir, runner=runner, + now=start + timedelta(minutes=6)) + assert first is True + assert second is False, f"the same message id must not wake the task twice: {notes}" + + # A genuinely new message still does. + runner = _feishu_runner( + chats=[{"chat_id": "oc_group", "chat_mode": "group"}], + messages={"oc_group": [message, _message("om_new", mentions=[_mention(HOST)])]}, + ) + third, notes = wakes(_record(feishu=300), task_name="t", agent="t", + state_dir=state_dir, runner=runner, + now=start + timedelta(minutes=12)) + assert third is True, notes + + +def test_the_feishu_condition_asks_the_cli_the_documented_shape(tmp_path): + """The argv is pinned: a condition that called a different endpoint would be silent. + + The listing must name `user` (a bot cannot see a direct conversation) and must ask + for p2p as well as groups — without `--types`, groups are all that comes back. + """ + runner = _feishu_runner(chats=[{"chat_id": "oc_g", "chat_mode": "group"}], + messages={"oc_g": []}) + wakes(_record(feishu=300), task_name="t", agent="t", + state_dir=tmp_path / "wake-state", runner=runner) + + calls = [" ".join(argv) for argv in runner.seen] + listing = next(call for call in calls if "+chat-list" in call) + assert "--types=p2p,group" in listing, listing + assert "--as user" in listing, listing + reading = next(call for call in calls if "+chat-messages-list" in call) + assert "--as user" in reading, reading + assert "--start" in reading, reading + + +def test_the_stamp_sent_to_the_cli_names_its_zone(tmp_path): + """A boundary without an offset leaves its own meaning to the reader. + + `lark-cli --start` answers this repository's Feishu task in two places: the round, + which the template hands `scheduler.py`'s `datetime.now(timezone.utc).isoformat()`, + and this poll. A bare local instant here would be the second carrier disagreeing + with the first about what a time is, and the CLI's own help offers the offset form + (`2026-01-01T15:04:05+08:00`), so the unaware spelling is not even the documented + one. Measured: the CLI accepts a naive stamp silently, which is exactly why nothing + downstream would notice. + """ + runner = _feishu_runner(chats=[{"chat_id": "oc_g", "chat_mode": "group"}], + messages={"oc_g": []}) + local_clock = datetime(2026, 10, 26, 12, 0, 0) # naive, as a caller may pass it + wakes(_record(feishu=300), task_name="t", agent="t", + state_dir=tmp_path / "wake-state", runner=runner, now=local_clock) + + reading = next(" ".join(argv) for argv in runner.seen if "+chat-messages-list" in " ".join(argv)) + stamp = reading.split("--start ", 1)[1].split(" ", 1)[0] + parsed = datetime.fromisoformat(stamp) + assert parsed.tzinfo is not None, ( + f"the stamp sent to the CLI is {stamp!r}, which names no zone — the round is told " + "an aware UTC instant for the same query" + ) + # …and it is the same instant the caller named, less this module's own lookback — + # read from the module rather than restated, so the window has one producer. + assert parsed == (local_clock - timedelta(seconds=wake_mod.FEISHU_LOOKBACK_SECONDS)).astimezone(), ( + stamp, local_clock + ) + + +def test_a_watermark_written_in_another_zone_does_not_end_the_wait(tmp_path): + """The watermark outlives the code that wrote it, so a mixed pair must not raise. + + `aware - naive` raises `TypeError`, and the subtraction that compares a stamp with + the clock sits outside the `try` that guards a condition. So before the instants + were normalised in one place, a state file carrying the older naive stamp met a + zoned clock and raised out of `wakes()` — and the task's own `run()` catches only + `asyncio.TimeoutError`, so that error ended the task's loop rather than one wait. + Both spellings are one instant here, and the condition is still consulted. + """ + state_dir = tmp_path / "wake-state" + state_dir.mkdir() + (state_dir / "t.json").write_text( + # Naive, because that is what the code before this fix wrote: the watermark + # outlives its writer, so the pairing that has to survive is old-file/now-clock. + json.dumps({"checked": {"feishu": "2026-10-26T10:00:00"}}), encoding="utf-8" + ) + runner = _feishu_runner(chats=[{"chat_id": "oc_g", "chat_mode": "group"}], + messages={"oc_g": []}) + + # 12:00 local is 04:00 UTC, two hours past the stamp — so the condition is due. + held, notes = wakes(_record(feishu=300), task_name="t", agent="t", + state_dir=state_dir, runner=runner, + now=datetime(2026, 10, 26, 12, 0, 0)) + assert held is False, notes + assert runner.seen, "the condition was skipped instead of consulted — its clock never elapsed" + + written = json.loads((state_dir / "t.json").read_text(encoding="utf-8")) + assert datetime.fromisoformat(written["checked"]["feishu"]).tzinfo is not None, written + + +# ── the wait ───────────────────────────────────────────────────────────── + + +def test_the_shortest_when_sizes_the_wait(): + """The scheduler waits for the tightest condition and consults each on its own clock.""" + assert shortest_when_seconds(_record(rant=600, feishu=300)) == 300 + assert shortest_when_seconds(_record(rant=600)) == 600 + assert shortest_when_seconds({"interval": 60}) is None + + +def test_a_poll_that_finds_nothing_does_not_run_the_task_and_does_not_move_the_run( + tmp_path, monkeypatch, +): + """A poll shortens a wait; it must not restart the cadence behind it. + + Two defects are held down here, and both were reachable in the obvious shape: + + * a poll that ran the task on timeout would turn a long interval into the poll's + own cadence — the opposite of what the field is for; + * a poll that recomputed the natural wait from *now* at every iteration would push + the task's own run out for ever, so the task would never run on its own initiative + again. `next_run_at` is read at each poll and must not move. + """ + from emrg.server import scheduler as mod + + # Every runtime file this handler writes — logs, the next-run slot, the saturation + # flag, the wake watermark — derives from `config_dir`, so redirecting that one name + # keeps the loop under a root the test owns. + monkeypatch.setattr(mod, "config_dir", lambda: tmp_path) + handler = make_handler(name="t", config={}, interval=14400) + handler._task["when"] = {"rant": 60} + + async def _ok(): + return mod.TaskHandler.RECORD_OK + + handler._refresh_record = _ok + handler._natural_wait = lambda: 10.0 + handler._wake_poll_seconds = lambda: 0.02 + + runs: list[float] = [] + reported: list[float | None] = [] + polls: list[int] = [] + + async def _cycle(): + runs.append(1.0) + handler._running = False # the run is the end of this test's story + + handler._run_cycle_bounded = _cycle + + async def _condition_wake(): + polls.append(1) + reported.append(handler._next_run_at) + return (len(polls) >= 3, [f"rant: {'holds' if len(polls) >= 3 else 'absent'}"]) + + handler._condition_wake = _condition_wake + + asyncio.run(handler.run()) + + assert len(polls) >= 3, f"the poll never came round: {polls}" + assert len(runs) == 1, "exactly one cycle: the two quiet polls must not have run it" + first_reported = reported[0] + assert first_reported is not None + assert all(value == first_reported for value in reported), ( + "the reported next run moved between polls — the cadence was restarted, " + f"so the task's own run would be pushed out for ever: {reported}" + ) + assert first_reported > __import__("time").time(), ( + "the reported next run must be in the future (the task's own cadence), " + "not the poll's instant" + ) + + +def test_a_task_without_when_never_polls(tmp_path): + """No `when`, no poll: the loop's wait and its run are exactly what they were.""" + handler = make_handler(name="t", config={}, interval=14400) + assert handler._wake_poll_seconds() is None + + +# ── the field survives a client edit ───────────────────────────────────── + + +def _scheduler_fixture(tmp_path, monkeypatch) -> TaskScheduler: + """A scheduler with its own table, and one registered agent to validate against.""" + from emrg.server import scheduler as mod + + monkeypatch.setattr(mod, "config_dir", lambda: tmp_path) + (tmp_path / "agents.yml").write_text( + yaml.safe_dump([{"name": "emrg", "path": str(tmp_path / "emrg")}]) + ) + sched = TaskScheduler(InstanceIdentity()) + sched._tasks_file = tmp_path / "tasks.yml" + sched._tasks_file.write_text(yaml.safe_dump([{ + "name": "emrg-task", "type": "evolution", "config": {"agent": "emrg"}, + "interval": 14400, "enabled": True, "when": {"rant": 600}, + }])) + return sched + + +def test_editing_another_field_keeps_when(tmp_path, monkeypatch): + """Requirement 9: a client that does not know the field must not drop it. + + `task_update` writes only the keys a payload carries, so a form editing `interval` + leaves `when` where it was. This is the guarantee the GUI relies on, and it is why + no GUI change is needed for the field to survive an edit there. + """ + sched = _scheduler_fixture(tmp_path, monkeypatch) + + ok, result = sched.task_update("emrg-task", interval=900) + assert ok, result + + written = yaml.safe_load(sched._tasks_file.read_text(encoding="utf-8")) + record = next(row for row in written if row["name"] == "emrg-task") + assert record["interval"] == 900 + assert record["when"] == {"rant": 600}, record + + +def test_when_can_be_set_and_removed_through_the_update_path(tmp_path, monkeypatch): + """A client that *does* send the field is obeyed, and an explicit null removes it.""" + sched = _scheduler_fixture(tmp_path, monkeypatch) + + ok, result = sched.task_update("emrg-task", when={"feishu": 300}) + assert ok, result + written = yaml.safe_load(sched._tasks_file.read_text(encoding="utf-8")) + assert written[0]["when"] == {"feishu": 300} + + ok, result = sched.task_update("emrg-task", when=None) + assert ok, result + written = yaml.safe_load(sched._tasks_file.read_text(encoding="utf-8")) + assert "when" not in written[0], written[0] + + +def test_the_update_path_refuses_a_bad_when(tmp_path, monkeypatch): + """The CRUD path refuses what the loader refuses, with the same message.""" + sched = _scheduler_fixture(tmp_path, monkeypatch) + + ok, error = sched.task_update("emrg-task", when={"rant": 1}) + assert ok is False and "must be an integer >= " in error, error + + ok, error = sched.task_update("emrg-task", when={"nope": 300}) + assert ok is False and "unknown when condition 'nope'" in error, error diff --git a/tests/test_ws_e2e.py b/tests/test_ws_e2e.py index 034be2fc..c9a8657b 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, agent, interval=None, enabled=True, repo=None, description=None, sandbox=None, schedule=None): + def fake_create(name, task_type, agent, interval=None, enabled=True, repo=None, description=None, sandbox=None, schedule=None, when=None): calls.update(name=name, task_type=task_type, agent=agent) return True, {"name": name, "type": task_type} @@ -3200,7 +3200,7 @@ async def _test(): try: calls = {} - def fake_create(name, task_type, agent, interval=None, enabled=True, repo=None, description=None, sandbox=None, schedule=None): + def fake_create(name, task_type, agent, interval=None, enabled=True, repo=None, description=None, sandbox=None, schedule=None, when=None): calls.update(name=name, task_type=task_type, agent=agent) return True, {"name": name, "type": task_type} @@ -3232,7 +3232,8 @@ async def _test(): calls = {} def fake_create(name, task_type, agent, interval=None, enabled=True, - repo=None, description=None, sandbox=None, schedule=None): + repo=None, description=None, sandbox=None, + schedule=None, when=None): calls.update(interval=interval, schedule=schedule) return True, {"name": name, "type": task_type} @@ -3257,6 +3258,72 @@ def fake_create(name, task_type, agent, interval=None, enabled=True, await cleanup() asyncio.run(_test()) + def test_task_create_passes_the_when_conditions_through(self): + """The conditions arrive at the scheduler in the shape the client sent them. + + Issue #2592: the field would otherwise be one a client can send and the daemon + quietly drops — the exact defect `schedule` had (#2490), which is why the wire's + key list and the scheduler's own argument list have to agree on it. + """ + 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, agent, interval=None, enabled=True, + repo=None, description=None, sandbox=None, + schedule=None, when=None): + calls.update(when=when) + return True, {"name": name, "type": task_type} + + server._scheduler = self._mock_scheduler(create_side_effect=fake_create) + ws = await connect_to_server() + conditions = {"rant": 600, "feishu": 300} + try: + resp = await self._cmd(ws, { + "type": "task_create", "name": "daily", + "task_type": "evolution", "agent": "emrg", + "interval": 14400, "when": conditions, + }) + assert resp["type"] == "task_result" + assert resp.get("ok") is True + assert calls["when"] == conditions, "the wire must not reshape it" + finally: + await ws.close() + finally: + await cleanup() + asyncio.run(_test()) + + def test_task_update_passes_the_when_conditions_through(self): + """The same for the update path, whose key list is a separate filter to keep in step.""" + async def _test(): + with tempfile.TemporaryDirectory() as tmp: + cwd = Path(tmp) + server, _, cleanup = await _boot_server(cwd) + try: + calls = {} + + def fake_update(name, **fields): + calls.update(name=name, fields=fields) + return True, {"name": name, "type": "evolution"} + + server._scheduler = self._mock_scheduler(update_side_effect=fake_update) + ws = await connect_to_server() + try: + resp = await self._cmd(ws, { + "type": "task_update", "name": "daily", + "when": {"rant": 600}, + }) + assert resp.get("ok") is True + assert calls["fields"].get("when") == {"rant": 600}, calls["fields"] + 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():