fix(cron): report reconciled startup job count

This commit is contained in:
chengyongru
2026-08-22 02:38:47 +08:00
committed by chengyongru
parent 20488a585d
commit dbc1801d3c
4 changed files with 10 additions and 27 deletions
+6 -9
View File
@@ -706,11 +706,6 @@ def _run_gateway(
else: else:
console.print("[yellow]Warning: No channels enabled[/yellow]") console.print("[yellow]Warning: No channels enabled[/yellow]")
cron_status = cron.status()
cron_job_count = cast(int, cron_status["jobs"])
if cron_job_count > 0:
console.print(f"[green]✓[/green] Cron: {cron_job_count} scheduled jobs")
hb_cfg = config.gateway.heartbeat hb_cfg = config.gateway.heartbeat
if hb_cfg.enabled: if hb_cfg.enabled:
console.print(f"[green]✓[/green] Heartbeat: every {hb_cfg.interval_s}s") console.print(f"[green]✓[/green] Heartbeat: every {hb_cfg.interval_s}s")
@@ -785,8 +780,7 @@ def _run_gateway(
console.print(f"[green]✓[/green] Dream: {dream_cfg.describe_schedule()}") console.print(f"[green]✓[/green] Dream: {dream_cfg.describe_schedule()}")
else: else:
console.print("[yellow]○[/yellow] Dream: disabled") console.print("[yellow]○[/yellow] Dream: disabled")
# Advance the cursor first: it must happen unconditionally (issue # Cursor repair must not depend on a healthy cron store.
# #4242), even if the cron store turns out to be corrupt below.
_advance_dream_cursor_if_behind(agent.context.memory) _advance_dream_cursor_if_behind(agent.context.memory)
cron.remove_system_job("dream") cron.remove_system_job("dream")
@@ -803,10 +797,13 @@ def _run_gateway(
payload=CronPayload(kind="system_event"), payload=CronPayload(kind="system_event"),
)) ))
else: else:
# Retire any previously persisted heartbeat job so that disabling
# gateway.heartbeat in config takes effect after restart.
cron.remove_system_job("heartbeat") cron.remove_system_job("heartbeat")
cron_status = cron.status()
cron_job_count = cast(int, cron_status["jobs"])
if cron_job_count > 0:
console.print(f"[green]✓[/green] Cron: {cron_job_count} scheduled jobs")
async def _open_browser_when_ready() -> None: async def _open_browser_when_ready() -> None:
"""Wait for the gateway to bind, then point the user's browser at the webui.""" """Wait for the gateway to bind, then point the user's browser at the webui."""
if not open_browser_url: if not open_browser_url:
+1 -8
View File
@@ -719,14 +719,7 @@ class CronService:
return job return job
def remove_system_job(self, job_id: str) -> bool: def remove_system_job(self, job_id: str) -> bool:
"""Remove an internal system job by id (startup reconciliation). """Remove a protected system job during startup reconciliation."""
Unlike ``remove_job``, this bypasses the protected-system-job guard:
the gateway retires a persisted system job whose config has been
disabled, so e.g. ``gateway.heartbeat.enabled=false`` actually takes
effect after restart instead of the leftover job firing forever.
Returns True when a job was removed.
"""
store = self._require_store() store = self._require_store()
before = len(store.jobs) before = len(store.jobs)
store.jobs = [j for j in store.jobs if j.id != job_id] store.jobs = [j for j in store.jobs if j.id != job_id]
+3 -3
View File
@@ -3216,13 +3216,14 @@ def test_gateway_local_trigger_queue_submits_agent_turns(
return None return None
def status(self) -> dict[str, int]: def status(self) -> dict[str, int]:
seen.setdefault("cron_reconciliation", []).append("status")
return {"jobs": 0} return {"jobs": 0}
def register_system_job(self, _job) -> None: def register_system_job(self, _job) -> None:
return None return None
def remove_system_job(self, job_id: str) -> bool: def remove_system_job(self, job_id: str) -> bool:
seen.setdefault("removed_system_jobs", []).append(job_id) seen.setdefault("cron_reconciliation", []).append(f"remove:{job_id}")
return False return False
class _FakeAgentLoop(_GatewayAgentContractStub): class _FakeAgentLoop(_GatewayAgentContractStub):
@@ -3301,8 +3302,7 @@ def test_gateway_local_trigger_queue_submits_agent_turns(
turn_delivery_factory = agent_kwargs["turn_delivery_factory"] turn_delivery_factory = agent_kwargs["turn_delivery_factory"]
assert isinstance(turn_delivery_factory, TurnDeliveryFactory) assert isinstance(turn_delivery_factory, TurnDeliveryFactory)
assert turn_delivery_factory.bus is bus assert turn_delivery_factory.bus is bus
# Disabled system jobs must be retired on startup, not just unregistered. assert seen["cron_reconciliation"] == ["remove:dream", "remove:heartbeat", "status"]
assert seen["removed_system_jobs"] == ["dream", "heartbeat"]
assert isinstance(turn_delivery_factory.route_policy, WebuiTurnRoutePolicy) assert isinstance(turn_delivery_factory.route_policy, WebuiTurnRoutePolicy)
assert turn_delivery_factory.route_policy.sessions is agent.sessions assert turn_delivery_factory.route_policy.sessions is agent.sessions
-7
View File
@@ -786,8 +786,6 @@ def test_remove_job_refuses_system_jobs(tmp_path) -> None:
def test_remove_system_job_retires_persisted_system_job(tmp_path) -> None: def test_remove_system_job_retires_persisted_system_job(tmp_path) -> None:
"""Disabling a system job (e.g. gateway.heartbeat.enabled=false) must
actually retire the previously persisted job, not just skip registration."""
store_path = tmp_path / "cron" / "jobs.json" store_path = tmp_path / "cron" / "jobs.json"
service = CronService(store_path) service = CronService(store_path)
service.register_system_job(CronJob( service.register_system_job(CronJob(
@@ -802,12 +800,8 @@ def test_remove_system_job_retires_persisted_system_job(tmp_path) -> None:
assert removed is True assert removed is True
assert service.get_job("heartbeat") is None assert service.get_job("heartbeat") is None
# Removal must persist: a fresh instance (next gateway start) must not
# resurrect the job from the on-disk store.
assert CronService(store_path).get_job("heartbeat") is None assert CronService(store_path).get_job("heartbeat") is None
# Idempotent: removing a missing system job reports False without raising.
assert service.remove_system_job("heartbeat") is False assert service.remove_system_job("heartbeat") is False
# User-facing removal still protects remaining system jobs.
other = CronService(store_path) other = CronService(store_path)
other.register_system_job(CronJob( other.register_system_job(CronJob(
id="dream", id="dream",
@@ -819,7 +813,6 @@ def test_remove_system_job_retires_persisted_system_job(tmp_path) -> None:
def test_remove_system_job_without_store_file(tmp_path) -> None: def test_remove_system_job_without_store_file(tmp_path) -> None:
"""Fresh install with the system job disabled: no jobs.json exists yet."""
store_path = tmp_path / "cron" / "jobs.json" store_path = tmp_path / "cron" / "jobs.json"
service = CronService(store_path) service = CronService(store_path)