diff --git a/Agent.md b/Agent.md index f4827a3a..0653d270 100644 --- a/Agent.md +++ b/Agent.md @@ -118,7 +118,7 @@ Community needs voiced in HN agent-UI discussions map directly to EMRG's design: pkill -f "emrg.server"; rm -f ~/.emrg/emrgd.token; python -m emrg ``` -Python: `uv run pytest tests/ -v` (1058) — import check: `uv run python -c "from emrg.client.app import run_client"` +Python: `uv run pytest tests/ -v` (1061) — import check: `uv run python -c "from emrg.client.app import run_client"` GUI: `cd emrg/gui && npm test` (259: 45 daemon_client + 20 conn-manager + 22 app-commands + 129 renderer smoke + 15 i18n + 8 integration + 3 commands + 8 build-config + 7 gui-state + 2 tool-group) — syntax: `node --check main.js preload.js daemon_client.js renderer/js/*.js` CI: `uv run pytest` (ubuntu + **windows-2025 matrix** — Windows pytest 回归在 PR CI 即失败，v0.2.29 教训 #725) + GUI tests + **actionlint workflow lint** (`rhysd/actionlint@v1.7.12` gate, #444 — workflow 解析错误在 PR CI 即失败，如 `if:` secrets 上下文) Re-trigger: `scripts/re-trigger-ci.sh [branch]` (workflow_dispatch, #527 — 替代空 commit 重触发：Actions outage 会整段丢弃 push 事件，dispatch 走 API 路径不受影响) diff --git a/emrg/server/scheduler.py b/emrg/server/scheduler.py index 9ee71bd6..1bce8b21 100644 --- a/emrg/server/scheduler.py +++ b/emrg/server/scheduler.py @@ -234,6 +234,21 @@ def __init__( if restored: self.evolutions = restored + # ── Cycle progress heartbeat (rant 2026-08-25T09:25:32 ③) ── + # While a cycle is running, periodic + event-driven writes to + # .heartbeat.json persist how far the cycle got (round / + # tool_count / timestamps). A daemon killed mid-cycle leaves the file + # with status=running — the next start reports exactly where the + # previous cycle was interrupted, so silent-death incidents become + # diagnosable (8-24 20:40 R46 / 8-25 02:00 emrg-task 01:26 lessons). + self._heartbeat_file = self._task_runs_dir / f"{self.name}.heartbeat.json" + self._cycle_progress: dict = { + "cycle_started_at": None, + "round": 0, + "tool_count": 0, + } + self._report_interrupted_cycle() + # ── Saturation — slow down, never stop (rant 2026-08-09T09:35:55) ── # Rant 2026-08-20T10:58:55 (host design-finalized): the empty-cycle # counter / slowdown-vote / threshold machinery is DELETED. Slowdown @@ -257,6 +272,16 @@ def __init__( self._slowdown_active = False self._slowdown_active = self._load_saturation_state() + # ── Next-run persistence (rant 2026-08-25T09:25:32 ④) ── + # The scheduled next-run time is persisted to + # ~/.emrg/next-run/.json so a daemon restart does not reset the + # schedule: the handler resumes at the original slot (only ever + # shortens the wait, never extends; a slot that passed during downtime + # runs immediately instead of waiting a fresh full interval). + self._next_run_dir = config_dir() / "next-run" + self._next_run_file = self._next_run_dir / f"{self.name}.json" + self._resume_next_run_at: float | None = self._load_next_run_state() + # Resolve project path from config (new schema) or fall back to # config.path for backward-compat with old tasks.yml entries. project_name = config.get("project", "") @@ -347,6 +372,127 @@ def _save_saturation_state(self) -> None: except Exception: pass + # ── Cycle progress heartbeat (rant 2026-08-25T09:25:32 ③) ── + # The heartbeat file (.heartbeat.json, sibling of the task-runs + # JSONL) exists ONLY while a cycle is running. It is written periodically + # by _heartbeat_loop and on every tool frame; removed on clean cycle end. + # A daemon killed mid-cycle leaves it behind with status=running, and the + # next handler start logs exactly where the cycle was interrupted. + _HEARTBEAT_PERIOD = 60 # seconds between periodic heartbeat writes + + def _write_heartbeat(self, status: str) -> None: + """Persist current cycle progress (best-effort, never raises).""" + try: + self._task_runs_dir.mkdir(parents=True, exist_ok=True) + self._heartbeat_file.write_text( + json.dumps({ + "task": self.name, + "status": status, + "cycle_started_at": self._cycle_progress.get("cycle_started_at"), + "last_heartbeat_at": datetime.now().isoformat(timespec="seconds"), + "round": self._cycle_progress.get("round", 0), + "tool_count": self._cycle_progress.get("tool_count", 0), + }, ensure_ascii=False), + encoding="utf-8", + ) + except Exception: + pass # heartbeat is best-effort; never affects the running cycle + + def _clear_heartbeat(self) -> None: + """Remove the running-cycle heartbeat marker (cycle ended cleanly).""" + try: + self._heartbeat_file.unlink(missing_ok=True) + except Exception: + pass + + async def _heartbeat_loop(self) -> None: + """Periodically persist cycle progress while a cycle is running.""" + try: + while True: + await asyncio.sleep(self._HEARTBEAT_PERIOD) + self._write_heartbeat("running") + except asyncio.CancelledError: + pass + + def _report_interrupted_cycle(self) -> None: + """Restart-time detection of a daemon killed mid-cycle. + + Rant 2026-08-25T09:25:32 ③: a heartbeat file left with + status=running means the previous cycle never finished — restore its + progress into _cycle_progress and log "上次中断于 round X" so the + silent-death incident becomes diagnosable from emrgd.log alone. + """ + try: + if not self._heartbeat_file.exists(): + return + data = json.loads(self._heartbeat_file.read_text(encoding="utf-8")) + if data.get("status") != "running": + return + round_n = int(data.get("round") or 0) + tool_n = int(data.get("tool_count") or 0) + self._cycle_progress["round"] = round_n + self._cycle_progress["tool_count"] = tool_n + self._cycle_progress["cycle_started_at"] = data.get("cycle_started_at") + self._logger.warning( + "TaskHandler[%s]: previous cycle interrupted at round %s " + "(tool_count=%s, started %s, last heartbeat %s) — see %s", + self.name, round_n, tool_n, + data.get("cycle_started_at"), data.get("last_heartbeat_at"), + self._heartbeat_file, + ) + except Exception: + pass + + # ── Next-run persistence (rant 2026-08-25T09:25:32 ④) ── + + def _load_next_run_state(self) -> float | None: + """Restore the persisted next-run time (~/.emrg/next-run/.json). + + Returns the epoch timestamp only when it is still in the future (a + daemon restart must not reset the schedule); None when absent, stale, + or unreadable. + """ + try: + if self._next_run_file.exists(): + data = json.loads(self._next_run_file.read_text(encoding="utf-8")) + ts = data.get("next_run_at") + if isinstance(ts, (int, float)) and ts > time.time(): + return float(ts) + except Exception: + pass + return None + + def _save_next_run_state(self) -> None: + """Persist the current _next_run_at; delete the file when cleared.""" + try: + if self._next_run_at is None: + self._next_run_file.unlink(missing_ok=True) + return + self._next_run_dir.mkdir(parents=True, exist_ok=True) + self._next_run_file.write_text( + json.dumps({"next_run_at": self._next_run_at}, ensure_ascii=False), + encoding="utf-8", + ) + except Exception: + pass + + def _resume_wait_timeout(self, wait_timeout: float) -> float: + """Fold the restored next-run time into the wait (one-shot). + + Rant 2026-08-25T09:25:32 ④: a daemon restart must not reset the next + run — if the persisted slot is still ahead, wait only the remainder + (never longer than the normal interval); if it already passed during + downtime, run immediately instead of waiting a fresh full interval. + """ + resume = self._resume_next_run_at + if resume is None: + return wait_timeout + self._resume_next_run_at = None + remaining = resume - time.time() + if remaining <= 0: + return 0.0 + return min(wait_timeout, remaining) + # ── Task-run persistence (rant 2026-08-19T20:50:36, host Plan B) ── # Execution records persist to ~/.emrg/logs/task-runs/.jsonl # (append-only JSONL, one line per cycle) so the GUI task recent-runs @@ -456,6 +602,11 @@ async def run(self) -> None: # 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) # 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. @@ -471,6 +622,7 @@ async def run(self) -> None: ) # Wait for interval or manual trigger (interruptible) self._next_run_at = time.time() + wait_timeout + self._save_next_run_state() # survives daemon restarts (rant 2026-08-25T09:25:32 ④) manual_trigger = False try: await asyncio.wait_for( @@ -504,6 +656,17 @@ async def run(self) -> None: self._cycle_running = True self._cycle_start_time = time.time() # per-cycle elapsed base (rant 2026-08-22T07:18:35) self._next_run_at = None # running — no next time yet + self._save_next_run_state() # clear the persisted slot (cycle starting now) + # Cycle progress heartbeat (rant 2026-08-25T09:25:32 ③): a + # periodic writer keeps .heartbeat.json fresh while the + # cycle runs; if the daemon dies mid-cycle the file survives and + # the next start reports where the cycle was interrupted. + self._cycle_progress = { + "cycle_started_at": datetime.now().isoformat(timespec="seconds"), + "round": 0, + "tool_count": 0, + } + heartbeat_task = asyncio.create_task(self._heartbeat_loop()) try: await self._run_evolution_cycle() except Exception: @@ -513,6 +676,8 @@ async def run(self) -> None: finally: self._cycle_running = False self._cycle_start_time = None + heartbeat_task.cancel() + self._clear_heartbeat() # cycle over — remove the running marker self._trigger_event.clear() # clear any spurious set during cycle await self._write_final_summary() @@ -774,6 +939,25 @@ async def _run_evolution_cycle(self) -> None: if "tool_name" in resp: tool_count += 1 + # Cycle progress heartbeat (rant 2026-08-25T09:25:32 ③): + # mirror the cumulative count into the heartbeat state — + # the write happens AFTER the round field is applied below + # so the persisted file carries the last frame's round. + self._cycle_progress["tool_count"] = tool_count + + # Round number (daemon exposes it on tool_start/done frames, + # rant 2026-08-25T09:25:32 ③): the heartbeat records the last + # round the cycle reached, so a restart can report "上次中断于 + # round X" precisely. + rnd = resp.get("round") + if isinstance(rnd, int): + self._cycle_progress["round"] = rnd + + if "tool_name" in resp: + # Write only after both fields are updated — the on-disk + # heartbeat is then as fresh as the last tool activity + # when the daemon dies. + self._write_heartbeat("running") resp_error = resp.get("error") if isinstance(resp_error, str): diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 51df6ef6..b9a2a68c 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -1744,6 +1744,133 @@ def test_slowdown_state_persisted_across_restart(tmp_path): mod.config_dir = orig +# ── Cycle progress heartbeat (rant 2026-08-25T09:25:32 ③) ───── +# While a cycle runs, tool frames mirror round/tool_count into _cycle_progress +# and write .heartbeat.json (status=running) — a daemon killed mid-cycle +# leaves the marker behind, and the next handler start reports exactly where +# the cycle was interrupted (silent-death incidents diagnosable from +# emrgd.log alone). + +def test_cycle_heartbeat_tracks_progress_and_clears(tmp_path): + """Tool frames update _cycle_progress + write the running heartbeat; a + clean cycle end removes the marker (run() finally path).""" + import json as _json + + handler, captured = _make_cycle_handler(tmp_path, frames=[ + {"tool_name": "bash", "round": 1}, + {"tool_name": "read", "round": 2}, + {"request_id": "r1", "content": "Done", "done": True, + "delta": False, "session_id": "s"}, + {"type": "vibe_check_result", "ok": True, + "result": {"work": "", "recommend_slowdown": False, + "slowdown_reason": ""}}, + ]) + asyncio.run(handler._run_evolution_cycle()) + # in-memory progress tracked from the streamed frames + assert handler._cycle_progress["tool_count"] == 2 + assert handler._cycle_progress["round"] == 2 + assert handler._cycle_progress["cycle_started_at"] is None or \ + handler._cycle_progress["cycle_started_at"] + # heartbeat file was written on the last tool frame (status=running) + hb = tmp_path / "logs" / "task-runs" / "emrg-task.heartbeat.json" + assert hb.exists(), "heartbeat file must be written during the cycle" + data = _json.loads(hb.read_text(encoding="utf-8")) + assert data["status"] == "running" + assert data["task"] == "emrg-task" + assert data["tool_count"] == 2 + assert data["round"] == 2 + assert data["last_heartbeat_at"], "heartbeat must carry a timestamp" + # clean cycle end removes the running marker (run() finally → _clear_heartbeat) + handler._clear_heartbeat() + assert not hb.exists(), "clean cycle end must remove the heartbeat marker" + + +def test_cycle_heartbeat_interrupted_reported_on_restart(tmp_path, caplog): + """A heartbeat left with status=running (daemon killed mid-cycle) is + reported by the next handler start: progress restored into + _cycle_progress + a warning naming the interrupting round.""" + import json as _json + + from emrg.server import scheduler as mod + + hb_dir = tmp_path / "logs" / "task-runs" + hb_dir.mkdir(parents=True) + (hb_dir / "emrg-task.heartbeat.json").write_text( + _json.dumps({ + "task": "emrg-task", "status": "running", + "cycle_started_at": "2026-08-25T00:00:00", + "last_heartbeat_at": "2026-08-25T00:05:00", + "round": 3, "tool_count": 7, + }), encoding="utf-8") + orig = mod.config_dir + try: + mod.config_dir = lambda: tmp_path + with caplog.at_level(logging.WARNING, logger="emrg.server.scheduler"): + h = TaskHandler(name="emrg-task", config={}, interval=60, + identity=InstanceIdentity()) + finally: + mod.config_dir = orig + # progress restored (informational; the next cycle resets it at start) + assert h._cycle_progress["round"] == 3 + assert h._cycle_progress["tool_count"] == 7 + warnings = [r.message for r in caplog.records if r.levelno >= logging.WARNING] + assert any("interrupted at round 3" in m for m in warnings), warnings + + +# ── Next-run persistence (rant 2026-08-25T09:25:32 ④) ───────── +# The scheduled next-run time persists to ~/.emrg/next-run/.json so a +# daemon restart does not reset the schedule (only shortens the wait; a slot +# that passed during downtime runs immediately instead of a fresh interval). + +def test_next_run_state_persisted_across_restart(tmp_path): + """_next_run_at survives a daemon restart via the next-run file: a fresh + handler over the same config_dir resumes the original slot one-shot, + never extending the wait; stale/cleared states fall back to normal.""" + import json as _json + import time as _time + + from emrg.server import scheduler as mod + + orig = mod.config_dir + try: + mod.config_dir = lambda: tmp_path + h1 = TaskHandler(name="emrg-task", config={}, interval=60, + identity=InstanceIdentity()) + h1._next_run_at = _time.time() + 3600 + h1._save_next_run_state() + nrf = tmp_path / "next-run" / "emrg-task.json" + assert nrf.exists(), "next-run state must persist to disk" + # "daemon restart": a fresh handler over the same config_dir restores + # the persisted slot (one-shot, never extends the normal wait) + h2 = TaskHandler(name="emrg-task", config={}, interval=60, + identity=InstanceIdentity()) + assert h2._resume_next_run_at is not None, \ + "future next-run slot restored from disk" + resumed = h2._resume_wait_timeout(600) + assert 0 < resumed <= 600, resumed + assert h2._resume_wait_timeout(600) == 600, \ + "one-shot: the second wait falls back to the normal interval" + # a slot that expired while the daemon was down → run immediately + h3 = TaskHandler(name="emrg-task", config={}, interval=60, + identity=InstanceIdentity()) + h3._resume_next_run_at = _time.time() - 5 # passed during downtime + assert h3._resume_wait_timeout(600) == 0.0, \ + "an already-passed slot runs immediately" + # a stale file (slot in the past) is ignored entirely + (tmp_path / "next-run" / "emrg-task.json").write_text( + _json.dumps({"next_run_at": _time.time() - 60}), encoding="utf-8") + h4 = TaskHandler(name="emrg-task", config={}, interval=60, + identity=InstanceIdentity()) + assert h4._resume_next_run_at is None, \ + "a past slot must not be resumed" + # clearing the state deletes the persisted file + h4._next_run_at = None + h4._save_next_run_state() + assert not nrf.exists(), "cleared state must delete the next-run file" + finally: + mod.config_dir = orig + + def test_evolution_cycle_aborted_error_not_counted(tmp_path): """Server error frame (e.g. 'session busy') → no evolution log, no count.""" handler, captured = _make_cycle_handler(tmp_path, frames=[