From 6239114c461a13a39ae1a5c6c3ada23acd2a9b62 Mon Sep 17 00:00:00 2001 From: chengyongru Date: Tue, 16 Jun 2026 14:07:08 +0800 Subject: [PATCH] fix(cron): prevent unbound automation execution --- nanobot/cli/commands.py | 1 + nanobot/cron/service.py | 61 ++++++++++++++++++- nanobot/webui/ws_http.py | 3 + tests/channels/test_websocket_http_routes.py | 14 +++++ tests/cron/test_cron_service.py | 62 ++++++++++++++++++++ 5 files changed, 140 insertions(+), 1 deletion(-) diff --git a/nanobot/cli/commands.py b/nanobot/cli/commands.py index 93f5dc30..818e353e 100644 --- a/nanobot/cli/commands.py +++ b/nanobot/cli/commands.py @@ -785,6 +785,7 @@ 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 16fab16b..ff74d425 100644 --- a/nanobot/cron/service.py +++ b/nanobot/cron/service.py @@ -136,24 +136,70 @@ class CronService: """Service for managing and executing scheduled jobs.""" _MAX_RUN_HISTORY = 20 + _UNBOUND_AGENT_JOB_REASON = ( + "agent cron payload is missing bound session delivery context; " + "recreate it from a chat session" + ) def __init__( self, 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 self._timer_active = False 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) + ) + + def _enforce_agent_binding(self, job: CronJob) -> bool: + """Disable user cron jobs that cannot be routed to a concrete session.""" + if not self._is_unbound_agent_job(job): + return False + if ( + 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 + ): + return False + + job.enabled = False + job.state.next_run_at_ms = None + job.state.last_status = "error" + job.state.last_error = self._UNBOUND_AGENT_JOB_REASON + job.updated_at_ms = max(job.updated_at_ms, _now_ms()) + logger.warning( + "Cron: disabled unbound agent job '{}' ({}): {}", + job.name, + job.id, + self._UNBOUND_AGENT_JOB_REASON, + ) + return True + + def _enforce_store_agent_bindings(self) -> bool: + if not self._store: + return False + changed = False + for job in self._store.jobs: + changed = self._enforce_agent_binding(job) or changed + return changed + def _load_jobs(self) -> tuple[list[CronJob], int] | None: """Load jobs from disk. @@ -312,6 +358,8 @@ class CronService: jobs, version = loaded self._store = CronStore(version=version, jobs=jobs) self._merge_action() + if self._enforce_store_agent_bindings() and self._running: + self._save_store() return self._store @@ -456,6 +504,8 @@ class CronService: return now = _now_ms() for job in self._store.jobs: + if self._enforce_agent_binding(job): + continue if job.enabled: job.state.next_run_at_ms = _compute_next_run(job.schedule, now) @@ -638,6 +688,7 @@ class CronService: delete_after_run=delete_after_run, ) _normalize_agent_turn_job(job) + self._enforce_agent_binding(job) if self._running: store = self._load_store() store.jobs.append(job) @@ -695,7 +746,8 @@ class CronService: if job.id == job_id: job.enabled = enabled job.updated_at_ms = _now_ms() - if enabled: + self._enforce_agent_binding(job) + if job.enabled: job.state.next_run_at_ms = _compute_next_run(job.schedule, _now_ms()) else: job.state.next_run_at_ms = None @@ -747,10 +799,13 @@ class CronService: if delete_after_run is not None: job.delete_after_run = delete_after_run _normalize_agent_turn_job(job) + self._enforce_agent_binding(job) job.updated_at_ms = _now_ms() if job.enabled: job.state.next_run_at_ms = _compute_next_run(job.schedule, _now_ms()) + else: + job.state.next_run_at_ms = None if self._running: self._save_store() @@ -769,6 +824,10 @@ class CronService: store = self._load_store() for job in store.jobs: if job.id == job_id: + if self._is_unbound_agent_job(job): + self._enforce_agent_binding(job) + self._save_store() + return False if not force and not job.enabled: return False await self._execute_job(job) diff --git a/nanobot/webui/ws_http.py b/nanobot/webui/ws_http.py index 34ac0151..0cda45a1 100644 --- a/nanobot/webui/ws_http.py +++ b/nanobot/webui/ws_http.py @@ -24,6 +24,7 @@ from websockets.http11 import Request as WsRequest from websockets.http11 import Response from nanobot.command.builtin import builtin_command_palette +from nanobot.cron.session_turns import is_bound_cron_job from nanobot.cron.types import CronJob, CronSchedule from nanobot.utils.subagent_channel_display import scrub_subagent_messages_for_channel from nanobot.webui.file_preview import WebUIFilePreviewError, file_preview_payload @@ -579,6 +580,8 @@ class GatewayHTTPHandler: return _http_error(404, "automation not found") if job.payload.kind == "system_event": return _http_error(403, "system automation is protected") + if action in {"enable", "run"} and not is_bound_cron_job(job): + return _http_error(409, "automation has no linked chat") if action == "enable": if self.cron_service.enable_job(job_id, enabled=True) is None: diff --git a/tests/channels/test_websocket_http_routes.py b/tests/channels/test_websocket_http_routes.py index 4451f486..b23d1097 100644 --- a/tests/channels/test_websocket_http_routes.py +++ b/tests/channels/test_websocket_http_routes.py @@ -1026,6 +1026,20 @@ async def test_webui_automations_route_lists_all_jobs_and_allows_user_actions( ) assert disabled_run.status_code == 409 + unbound_run = await _http_get( + f"{base_url}/api/webui/automations/run?id={incomplete_job.id}", + headers=auth, + ) + assert unbound_run.status_code == 409 + assert "no linked chat" in unbound_run.text + + unbound_enable = await _http_get( + f"{base_url}/api/webui/automations/enable?id={incomplete_job.id}", + headers=auth, + ) + assert unbound_enable.status_code == 409 + assert "no linked chat" in unbound_enable.text + protected_delete = await _http_get( f"{base_url}/api/webui/automations/delete?id=heartbeat", headers=auth, diff --git a/tests/cron/test_cron_service.py b/tests/cron/test_cron_service.py index d81d4121..a52029e5 100644 --- a/tests/cron/test_cron_service.py +++ b/tests/cron/test_cron_service.py @@ -43,6 +43,68 @@ def test_add_job_accepts_valid_timezone(tmp_path) -> None: assert job.state.next_run_at_ms is not None +@pytest.mark.asyncio +async def test_require_bound_agent_jobs_disables_unbound_adds(tmp_path) -> None: + called: list[str] = [] + + async def on_job(job): + called.append(job.id) + + service = CronService( + tmp_path / "cron" / "jobs.json", + on_job=on_job, + require_bound_agent_jobs=True, + ) + job = service.add_job( + name="unbound", + schedule=CronSchedule(kind="every", every_ms=60_000), + message="hello", + ) + + assert job.enabled is False + assert job.state.next_run_at_ms is None + assert job.state.last_status == "error" + assert "missing bound session delivery context" in (job.state.last_error or "") + assert await service.run_job(job.id, force=True) is False + assert called == [] + + +def test_require_bound_agent_jobs_disables_loaded_unbound_jobs(tmp_path) -> None: + store_path = tmp_path / "cron" / "jobs.json" + store_path.parent.mkdir(parents=True) + store_path.write_text( + json.dumps( + { + "version": 1, + "jobs": [ + { + "id": "unbound-1", + "name": "Unbound reminder", + "enabled": True, + "schedule": {"kind": "every", "everyMs": 60_000}, + "payload": { + "kind": "agent_turn", + "message": "check status", + }, + "state": {"nextRunAtMs": 1}, + "createdAtMs": 1, + "updatedAtMs": 1, + } + ], + } + ), + encoding="utf-8", + ) + + job = CronService(store_path, require_bound_agent_jobs=True).get_job("unbound-1") + + assert job is not None + assert job.enabled is False + assert job.state.next_run_at_ms is None + assert job.state.last_status == "error" + assert "missing bound session delivery context" in (job.state.last_error or "") + + def test_add_job_migrates_legacy_delivery_context(tmp_path) -> None: service = CronService(tmp_path / "cron" / "jobs.json") meta = {"slack": {"thread_ts": "1234567890.123456", "channel_type": "channel"}}