Skip to content

emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32)#973

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
argszero merged 1 commit into from
Aug 25, 2026
Merged

emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32) #973

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Agent.md
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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 路径不受影响)
Expand Down
184 changes: 184 additions & 0 deletions emrg/server/scheduler.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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
# <task>.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
Expand All@@ -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/<task>.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", "")
Expand DownExpand Up@@ -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 (<task>.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/<task>.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/<task>.jsonl
# (append-only JSONL, one line per cycle) so the GUI task recent-runs
Expand DownExpand Up@@ -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.
Expand All@@ -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(
Expand DownExpand Up@@ -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 <task>.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:
Expand All@@ -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()
Expand DownExpand Up@@ -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):
Expand Down
127 changes: 127 additions & 0 deletions tests/test_scheduler.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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 <task>.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/<task>.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=[
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32) by pm25coder · Pull Request #973 · argszero/emrg · GitHub
Skip to content
emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32)#973

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

argszero
merged 1 commit into

from

Aug 25, 2026
 Merged


emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32)
#973
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


2 changes: 1 addition & 1 deletion
Agent.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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 路径不受影响)
Expand Down

184 changes: 184 additions & 0 deletions
emrg/server/scheduler.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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
 # <task>.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
Expand All@@ -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/<task>.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", "")
Expand DownExpand Up@@ -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 (<task>.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/<task>.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/<task>.jsonl
 # (append-only JSONL, one line per cycle) so the GUI task recent-runs
Expand DownExpand Up@@ -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.
Expand All@@ -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(
Expand DownExpand Up@@ -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 <task>.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:
Expand All@@ -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()
Expand DownExpand Up@@ -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):
Expand Down

127 changes: 127 additions & 0 deletions
tests/test_scheduler.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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 <task>.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/<task>.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=[
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32) by pm25coder · Pull Request #973 · argszero/emrg · GitHub
Skip to content
emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32)#973

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

argszero
merged 1 commit into

from

Aug 25, 2026
 Merged


emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32)
#973
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


2 changes: 1 addition & 1 deletion
Agent.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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 路径不受影响)
Expand Down

184 changes: 184 additions & 0 deletions
emrg/server/scheduler.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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
 # <task>.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
Expand All@@ -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/<task>.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", "")
Expand DownExpand Up@@ -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 (<task>.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/<task>.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/<task>.jsonl
 # (append-only JSONL, one line per cycle) so the GUI task recent-runs
Expand DownExpand Up@@ -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.
Expand All@@ -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(
Expand DownExpand Up@@ -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 <task>.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:
Expand All@@ -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()
Expand DownExpand Up@@ -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):
Expand Down

127 changes: 127 additions & 0 deletions
tests/test_scheduler.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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 <task>.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/<task>.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=[
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32) by pm25coder · Pull Request #973 · argszero/emrg · GitHub
Skip to content
emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32)#973

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

argszero
merged 1 commit into

from

Aug 25, 2026
 Merged


emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32)
#973
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


2 changes: 1 addition & 1 deletion
Agent.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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 路径不受影响)
Expand Down

184 changes: 184 additions & 0 deletions
emrg/server/scheduler.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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
 # <task>.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
Expand All@@ -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/<task>.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", "")
Expand DownExpand Up@@ -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 (<task>.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/<task>.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/<task>.jsonl
 # (append-only JSONL, one line per cycle) so the GUI task recent-runs
Expand DownExpand Up@@ -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.
Expand All@@ -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(
Expand DownExpand Up@@ -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 <task>.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:
Expand All@@ -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()
Expand DownExpand Up@@ -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):
Expand Down

127 changes: 127 additions & 0 deletions
tests/test_scheduler.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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 <task>.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/<task>.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=[
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '
emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32) by pm25coder · Pull Request #973 · argszero/emrg · GitHub
Skip to content
emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32)#973

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

argszero
merged 1 commit into

from

Aug 25, 2026
 Merged


emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32)
#973
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


2 changes: 1 addition & 1 deletion
Agent.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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 路径不受影响)
Expand Down

184 changes: 184 additions & 0 deletions
emrg/server/scheduler.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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
 # <task>.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
Expand All@@ -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/<task>.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", "")
Expand DownExpand Up@@ -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 (<task>.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/<task>.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/<task>.jsonl
 # (append-only JSONL, one line per cycle) so the GUI task recent-runs
Expand DownExpand Up@@ -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.
Expand All@@ -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(
Expand DownExpand Up@@ -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 <task>.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:
Expand All@@ -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()
Expand DownExpand Up@@ -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):
Expand Down

127 changes: 127 additions & 0 deletions
tests/test_scheduler.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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 <task>.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/<task>.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=[
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32) by pm25coder · Pull Request #973 · argszero/emrg · GitHub
Skip to content
emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32)#973

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

argszero
merged 1 commit into

from

Aug 25, 2026
 Merged


emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32)
#973
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


2 changes: 1 addition & 1 deletion
Agent.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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 路径不受影响)
Expand Down

184 changes: 184 additions & 0 deletions
emrg/server/scheduler.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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
 # <task>.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
Expand All@@ -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/<task>.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", "")
Expand DownExpand Up@@ -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 (<task>.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/<task>.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/<task>.jsonl
 # (append-only JSONL, one line per cycle) so the GUI task recent-runs
Expand DownExpand Up@@ -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.
Expand All@@ -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(
Expand DownExpand Up@@ -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 <task>.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:
Expand All@@ -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()
Expand DownExpand Up@@ -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):
Expand Down

127 changes: 127 additions & 0 deletions
tests/test_scheduler.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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 <task>.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/<task>.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=[
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
})();
emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32) by pm25coder · Pull Request #973 · argszero/emrg · GitHub
Skip to content
emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32)#973

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

argszero
merged 1 commit into

from

Aug 25, 2026
 Merged


emrg: scheduler cycle heartbeat + next-run persistence (rant 2026-08-25T09:25:32)
#973
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


2 changes: 1 addition & 1 deletion
Agent.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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 路径不受影响)
Expand Down

184 changes: 184 additions & 0 deletions
emrg/server/scheduler.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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
 # <task>.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
Expand All@@ -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/<task>.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", "")
Expand DownExpand Up@@ -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 (<task>.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/<task>.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/<task>.jsonl
 # (append-only JSONL, one line per cycle) so the GUI task recent-runs
Expand DownExpand Up@@ -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.
Expand All@@ -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(
Expand DownExpand Up@@ -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 <task>.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:
Expand All@@ -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()
Expand DownExpand Up@@ -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):
Expand Down

127 changes: 127 additions & 0 deletions
tests/test_scheduler.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -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 <task>.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/<task>.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=[
Expand Down
Loading