From b24b5f19fc04b56945cf0da9a3fd117678de8ad6 Mon Sep 17 00:00:00 2001 From: chengyongru Date: Tue, 16 Jun 2026 14:16:12 +0800 Subject: [PATCH] fix(cron): always require bound automation sessions --- nanobot/cli/commands.py | 1 - nanobot/cron/service.py | 10 ++------ tests/cron/test_cron_service.py | 42 +++++++++++++++++++++++++++---- tests/cron/test_cron_tool_list.py | 18 +++++++++++++ 4 files changed, 57 insertions(+), 14 deletions(-) diff --git a/nanobot/cli/commands.py b/nanobot/cli/commands.py index 818e353e..93f5dc30 100644 --- a/nanobot/cli/commands.py +++ b/nanobot/cli/commands.py @@ -785,7 +785,6 @@ def _run_gateway( # Create cron service with workspace-scoped store cron_store_path = config.workspace_path / "cron" / "jobs.json" cron = CronService(cron_store_path) - cron.require_bound_agent_jobs = True # Create agent with cron service agent = AgentLoop.from_config( diff --git a/nanobot/cron/service.py b/nanobot/cron/service.py index ff74d425..23426bec 100644 --- a/nanobot/cron/service.py +++ b/nanobot/cron/service.py @@ -146,14 +146,12 @@ class CronService: store_path: Path, on_job: Callable[[CronJob], Coroutine[Any, Any, str | None]] | None = None, max_sleep_ms: int = 300_000, # 5 minutes - require_bound_agent_jobs: bool = False, ): self.store_path = store_path self._action_path = store_path.parent / "action.jsonl" self._run_records_dir = store_path.parent / "runs" self._lock = FileLock(str(self._action_path.parent) + ".lock") self.on_job = on_job - self.require_bound_agent_jobs = require_bound_agent_jobs self._store: CronStore | None = None self._timer_task: asyncio.Task | None = None self._running = False @@ -161,11 +159,7 @@ class CronService: self.max_sleep_ms = max_sleep_ms def _is_unbound_agent_job(self, job: CronJob) -> bool: - return ( - self.require_bound_agent_jobs - and job.payload.kind == "agent_turn" - and not is_bound_cron_job(job) - ) + return job.payload.kind == "agent_turn" and not is_bound_cron_job(job) def _enforce_agent_binding(self, job: CronJob) -> bool: """Disable user cron jobs that cannot be routed to a concrete session.""" @@ -175,7 +169,7 @@ class CronService: not job.enabled and job.state.next_run_at_ms is None and job.state.last_status == "error" - and job.state.last_error == self._UNBOUND_AGENT_JOB_REASON + and job.state.last_error ): return False diff --git a/tests/cron/test_cron_service.py b/tests/cron/test_cron_service.py index a52029e5..2072e063 100644 --- a/tests/cron/test_cron_service.py +++ b/tests/cron/test_cron_service.py @@ -17,6 +17,14 @@ async def _wait_until(predicate, *, timeout: float = 1.0, interval: float = 0.01 assert predicate() +def _bound_chat(chat_id: str = "chat-1") -> dict[str, str]: + return { + "session_key": f"websocket:{chat_id}", + "origin_channel": "websocket", + "origin_chat_id": chat_id, + } + + def test_add_job_rejects_unknown_timezone(tmp_path) -> None: service = CronService(tmp_path / "cron" / "jobs.json") @@ -37,6 +45,7 @@ def test_add_job_accepts_valid_timezone(tmp_path) -> None: name="tz ok", schedule=CronSchedule(kind="cron", expr="0 9 * * *", tz="America/Vancouver"), message="hello", + **_bound_chat(), ) assert job.schedule.tz == "America/Vancouver" @@ -44,7 +53,7 @@ def test_add_job_accepts_valid_timezone(tmp_path) -> None: @pytest.mark.asyncio -async def test_require_bound_agent_jobs_disables_unbound_adds(tmp_path) -> None: +async def test_unbound_agent_jobs_are_disabled_on_add(tmp_path) -> None: called: list[str] = [] async def on_job(job): @@ -53,7 +62,6 @@ async def test_require_bound_agent_jobs_disables_unbound_adds(tmp_path) -> None: service = CronService( tmp_path / "cron" / "jobs.json", on_job=on_job, - require_bound_agent_jobs=True, ) job = service.add_job( name="unbound", @@ -69,7 +77,7 @@ async def test_require_bound_agent_jobs_disables_unbound_adds(tmp_path) -> None: assert called == [] -def test_require_bound_agent_jobs_disables_loaded_unbound_jobs(tmp_path) -> None: +def test_unbound_agent_jobs_are_disabled_on_load(tmp_path) -> None: store_path = tmp_path / "cron" / "jobs.json" store_path.parent.mkdir(parents=True) store_path.write_text( @@ -96,7 +104,7 @@ def test_require_bound_agent_jobs_disables_loaded_unbound_jobs(tmp_path) -> None encoding="utf-8", ) - job = CronService(store_path, require_bound_agent_jobs=True).get_job("unbound-1") + job = CronService(store_path).get_job("unbound-1") assert job is not None assert job.enabled is False @@ -325,6 +333,7 @@ async def test_execute_job_records_run_history(tmp_path) -> None: name="hist", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) await service.run_job(job.id) @@ -349,6 +358,7 @@ async def test_run_history_records_errors(tmp_path) -> None: name="fail", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) await service.run_job(job.id) @@ -370,6 +380,7 @@ async def test_run_history_records_skipped_jobs(tmp_path) -> None: name="skip", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) await service.run_job(job.id) @@ -394,7 +405,7 @@ async def test_run_history_records_job_cancellation(tmp_path) -> None: name="cancel", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", - session_key="websocket:chat-1", + **_bound_chat(), ) assert await service.run_job(job.id) is True @@ -417,6 +428,7 @@ async def test_run_history_trimmed_to_max(tmp_path) -> None: name="trim", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) for _ in range(25): await service.run_job(job.id) @@ -433,6 +445,7 @@ async def test_run_history_persisted_to_disk(tmp_path) -> None: name="persist", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) await service.run_job(job.id) @@ -457,6 +470,7 @@ async def test_run_job_disabled_does_not_flip_running_state(tmp_path) -> None: name="disabled", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) service.enable_job(job.id, enabled=False) @@ -475,6 +489,7 @@ async def test_run_job_preserves_running_service_state(tmp_path) -> None: name="manual", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) result = await service.run_job(job.id, force=True) @@ -497,6 +512,7 @@ async def test_running_service_honors_external_disable(tmp_path) -> None: name="external-disable", schedule=CronSchedule(kind="every", every_ms=200), message="hello", + **_bound_chat(), ) await service.start() try: @@ -545,6 +561,7 @@ async def test_start_server_not_jobs(tmp_path): name="hist", schedule=CronSchedule(kind="every", every_ms=100), message="hello", + **_bound_chat(), ) assert len(service.list_jobs()) == 1 await _wait_until(lambda: bool(called), timeout=0.8) @@ -565,6 +582,7 @@ async def test_subsecond_job_not_delayed_to_one_second(tmp_path): name="fast", schedule=CronSchedule(kind="every", every_ms=100), message="hello", + **_bound_chat(), ) await service.start() try: @@ -588,6 +606,7 @@ async def test_running_service_picks_up_external_add(tmp_path): name="heartbeat", schedule=CronSchedule(kind="every", every_ms=100), message="tick", + **_bound_chat("heartbeat"), ) await service.start() try: @@ -598,6 +617,7 @@ async def test_running_service_picks_up_external_add(tmp_path): name="external", schedule=CronSchedule(kind="every", every_ms=100), message="ping", + **_bound_chat("external"), ) await _wait_until(lambda: "external" in called, timeout=0.8) @@ -619,6 +639,7 @@ async def test_add_job_during_jobs_exec(tmp_path): name="test", schedule=CronSchedule(kind="every", every_ms=150), message="tick", + **_bound_chat("test"), ) run_once = False @@ -627,6 +648,7 @@ async def test_add_job_during_jobs_exec(tmp_path): name="heartbeat", schedule=CronSchedule(kind="every", every_ms=100), message="tick", + **_bound_chat("heartbeat"), ) assert len(service.list_jobs()) == 1 await service.start() @@ -647,6 +669,7 @@ async def test_external_update_preserves_run_history_records(tmp_path): name="history", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) await service.run_job(job.id, force=True) @@ -688,6 +711,7 @@ async def test_timer_execution_is_not_rolled_back_by_list_jobs_reload(tmp_path): name="race", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) job.state.next_run_at_ms = max(1, int(time.time() * 1000) - 1_000) service._save_store() @@ -712,6 +736,7 @@ def test_update_job_changes_name(tmp_path) -> None: name="old name", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) result = service.update_job(job.id, name="new name") assert isinstance(result, CronJob) @@ -725,6 +750,7 @@ def test_update_job_changes_schedule(tmp_path) -> None: name="sched", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) old_next = job.state.next_run_at_ms @@ -741,6 +767,7 @@ def test_update_job_changes_message(tmp_path) -> None: name="msg", schedule=CronSchedule(kind="every", every_ms=60_000), message="old message", + **_bound_chat(), ) result = service.update_job(job.id, message="new message") assert isinstance(result, CronJob) @@ -753,6 +780,7 @@ def test_update_job_changes_cron_expression(tmp_path) -> None: name="cron-job", schedule=CronSchedule(kind="cron", expr="0 9 * * *", tz="UTC"), message="hello", + **_bound_chat(), ) result = service.update_job( job.id, @@ -788,6 +816,7 @@ def test_update_job_validates_schedule(tmp_path) -> None: name="validate", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) with pytest.raises(ValueError, match="unknown timezone"): service.update_job( @@ -805,6 +834,7 @@ async def test_update_job_preserves_run_history(tmp_path) -> None: name="hist", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) await service.run_job(job.id) @@ -820,6 +850,7 @@ def test_update_job_offline_writes_action(tmp_path) -> None: name="offline", schedule=CronSchedule(kind="every", every_ms=60_000), message="hello", + **_bound_chat(), ) service.update_job(job.id, name="updated-offline") @@ -873,6 +904,7 @@ async def test_list_jobs_during_on_job_does_not_cause_stale_reload(tmp_path) -> name=name, schedule=CronSchedule(kind="every", every_ms=3_600_000), message="test", + **_bound_chat(name), ) # Force next_run to the past so _on_timer picks them up for job in service._store.jobs: diff --git a/tests/cron/test_cron_tool_list.py b/tests/cron/test_cron_tool_list.py index bcf518ed..d2099d45 100644 --- a/tests/cron/test_cron_tool_list.py +++ b/tests/cron/test_cron_tool_list.py @@ -20,6 +20,14 @@ def _make_tool_with_tz(tmp_path, tz: str) -> CronTool: return CronTool(service, default_timezone=tz) +def _bound_chat(chat_id: str = "chat-1") -> dict[str, str]: + return { + "session_key": f"websocket:{chat_id}", + "origin_channel": "websocket", + "origin_chat_id": chat_id, + } + + # -- _format_timing tests -- @@ -146,6 +154,7 @@ def test_list_cron_job_shows_expression_and_timezone(tmp_path) -> None: name="Morning scan", schedule=CronSchedule(kind="cron", expr="0 9 * * 1-5", tz="America/Denver"), message="scan", + **_bound_chat(), ) result = tool._list_jobs() assert "cron: 0 9 * * 1-5 (America/Denver)" in result @@ -157,6 +166,7 @@ def test_list_every_job_shows_human_interval(tmp_path) -> None: name="Frequent check", schedule=CronSchedule(kind="every", every_ms=1_800_000), message="check", + **_bound_chat(), ) result = tool._list_jobs() assert "every 30m" in result @@ -168,6 +178,7 @@ def test_list_every_job_hours(tmp_path) -> None: name="Hourly check", schedule=CronSchedule(kind="every", every_ms=7_200_000), message="check", + **_bound_chat(), ) result = tool._list_jobs() assert "every 2h" in result @@ -179,6 +190,7 @@ def test_list_every_job_seconds(tmp_path) -> None: name="Fast check", schedule=CronSchedule(kind="every", every_ms=30_000), message="check", + **_bound_chat(), ) result = tool._list_jobs() assert "every 30s" in result @@ -190,6 +202,7 @@ def test_list_every_job_non_minute_seconds(tmp_path) -> None: name="Ninety-second check", schedule=CronSchedule(kind="every", every_ms=90_000), message="check", + **_bound_chat(), ) result = tool._list_jobs() assert "every 90s" in result @@ -201,6 +214,7 @@ def test_list_every_job_milliseconds(tmp_path) -> None: name="Sub-second check", schedule=CronSchedule(kind="every", every_ms=200), message="check", + **_bound_chat(), ) result = tool._list_jobs() assert "every 200ms" in result @@ -212,6 +226,7 @@ def test_list_at_job_shows_iso_timestamp(tmp_path) -> None: name="One-shot", schedule=CronSchedule(kind="at", at_ms=1773684000000), message="fire", + **_bound_chat(), ) result = tool._list_jobs() assert "at 2026-" in result @@ -226,6 +241,7 @@ async def test_list_shows_last_run_state(tmp_path) -> None: name="Stateful job", schedule=CronSchedule(kind="cron", expr="0 9 * * *", tz="UTC"), message="test", + **_bound_chat(), ) # Simulate a completed run by updating state in the store job.state.last_run_at_ms = 1773673200000 @@ -245,6 +261,7 @@ async def test_list_shows_error_message(tmp_path) -> None: name="Failed job", schedule=CronSchedule(kind="cron", expr="0 9 * * *", tz="UTC"), message="test", + **_bound_chat(), ) job.state.last_run_at_ms = 1773673200000 job.state.last_status = "error" @@ -262,6 +279,7 @@ def test_list_shows_next_run(tmp_path) -> None: name="Upcoming job", schedule=CronSchedule(kind="cron", expr="0 9 * * *", tz="UTC"), message="test", + **_bound_chat(), ) result = tool._list_jobs() assert "Next run:" in result