mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-06 17:38:35 +00:00
fix(cron): prevent unbound automation execution
This commit is contained in:
parent
04545b95d9
commit
6239114c46
@ -785,6 +785,7 @@ def _run_gateway(
|
|||||||
# Create cron service with workspace-scoped store
|
# Create cron service with workspace-scoped store
|
||||||
cron_store_path = config.workspace_path / "cron" / "jobs.json"
|
cron_store_path = config.workspace_path / "cron" / "jobs.json"
|
||||||
cron = CronService(cron_store_path)
|
cron = CronService(cron_store_path)
|
||||||
|
cron.require_bound_agent_jobs = True
|
||||||
|
|
||||||
# Create agent with cron service
|
# Create agent with cron service
|
||||||
agent = AgentLoop.from_config(
|
agent = AgentLoop.from_config(
|
||||||
|
|||||||
@ -136,24 +136,70 @@ class CronService:
|
|||||||
"""Service for managing and executing scheduled jobs."""
|
"""Service for managing and executing scheduled jobs."""
|
||||||
|
|
||||||
_MAX_RUN_HISTORY = 20
|
_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__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
store_path: Path,
|
store_path: Path,
|
||||||
on_job: Callable[[CronJob], Coroutine[Any, Any, str | None]] | None = None,
|
on_job: Callable[[CronJob], Coroutine[Any, Any, str | None]] | None = None,
|
||||||
max_sleep_ms: int = 300_000, # 5 minutes
|
max_sleep_ms: int = 300_000, # 5 minutes
|
||||||
|
require_bound_agent_jobs: bool = False,
|
||||||
):
|
):
|
||||||
self.store_path = store_path
|
self.store_path = store_path
|
||||||
self._action_path = store_path.parent / "action.jsonl"
|
self._action_path = store_path.parent / "action.jsonl"
|
||||||
self._run_records_dir = store_path.parent / "runs"
|
self._run_records_dir = store_path.parent / "runs"
|
||||||
self._lock = FileLock(str(self._action_path.parent) + ".lock")
|
self._lock = FileLock(str(self._action_path.parent) + ".lock")
|
||||||
self.on_job = on_job
|
self.on_job = on_job
|
||||||
|
self.require_bound_agent_jobs = require_bound_agent_jobs
|
||||||
self._store: CronStore | None = None
|
self._store: CronStore | None = None
|
||||||
self._timer_task: asyncio.Task | None = None
|
self._timer_task: asyncio.Task | None = None
|
||||||
self._running = False
|
self._running = False
|
||||||
self._timer_active = False
|
self._timer_active = False
|
||||||
self.max_sleep_ms = max_sleep_ms
|
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:
|
def _load_jobs(self) -> tuple[list[CronJob], int] | None:
|
||||||
"""Load jobs from disk.
|
"""Load jobs from disk.
|
||||||
|
|
||||||
@ -312,6 +358,8 @@ class CronService:
|
|||||||
jobs, version = loaded
|
jobs, version = loaded
|
||||||
self._store = CronStore(version=version, jobs=jobs)
|
self._store = CronStore(version=version, jobs=jobs)
|
||||||
self._merge_action()
|
self._merge_action()
|
||||||
|
if self._enforce_store_agent_bindings() and self._running:
|
||||||
|
self._save_store()
|
||||||
|
|
||||||
return self._store
|
return self._store
|
||||||
|
|
||||||
@ -456,6 +504,8 @@ class CronService:
|
|||||||
return
|
return
|
||||||
now = _now_ms()
|
now = _now_ms()
|
||||||
for job in self._store.jobs:
|
for job in self._store.jobs:
|
||||||
|
if self._enforce_agent_binding(job):
|
||||||
|
continue
|
||||||
if job.enabled:
|
if job.enabled:
|
||||||
job.state.next_run_at_ms = _compute_next_run(job.schedule, now)
|
job.state.next_run_at_ms = _compute_next_run(job.schedule, now)
|
||||||
|
|
||||||
@ -638,6 +688,7 @@ class CronService:
|
|||||||
delete_after_run=delete_after_run,
|
delete_after_run=delete_after_run,
|
||||||
)
|
)
|
||||||
_normalize_agent_turn_job(job)
|
_normalize_agent_turn_job(job)
|
||||||
|
self._enforce_agent_binding(job)
|
||||||
if self._running:
|
if self._running:
|
||||||
store = self._load_store()
|
store = self._load_store()
|
||||||
store.jobs.append(job)
|
store.jobs.append(job)
|
||||||
@ -695,7 +746,8 @@ class CronService:
|
|||||||
if job.id == job_id:
|
if job.id == job_id:
|
||||||
job.enabled = enabled
|
job.enabled = enabled
|
||||||
job.updated_at_ms = _now_ms()
|
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())
|
job.state.next_run_at_ms = _compute_next_run(job.schedule, _now_ms())
|
||||||
else:
|
else:
|
||||||
job.state.next_run_at_ms = None
|
job.state.next_run_at_ms = None
|
||||||
@ -747,10 +799,13 @@ class CronService:
|
|||||||
if delete_after_run is not None:
|
if delete_after_run is not None:
|
||||||
job.delete_after_run = delete_after_run
|
job.delete_after_run = delete_after_run
|
||||||
_normalize_agent_turn_job(job)
|
_normalize_agent_turn_job(job)
|
||||||
|
self._enforce_agent_binding(job)
|
||||||
|
|
||||||
job.updated_at_ms = _now_ms()
|
job.updated_at_ms = _now_ms()
|
||||||
if job.enabled:
|
if job.enabled:
|
||||||
job.state.next_run_at_ms = _compute_next_run(job.schedule, _now_ms())
|
job.state.next_run_at_ms = _compute_next_run(job.schedule, _now_ms())
|
||||||
|
else:
|
||||||
|
job.state.next_run_at_ms = None
|
||||||
|
|
||||||
if self._running:
|
if self._running:
|
||||||
self._save_store()
|
self._save_store()
|
||||||
@ -769,6 +824,10 @@ class CronService:
|
|||||||
store = self._load_store()
|
store = self._load_store()
|
||||||
for job in store.jobs:
|
for job in store.jobs:
|
||||||
if job.id == job_id:
|
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:
|
if not force and not job.enabled:
|
||||||
return False
|
return False
|
||||||
await self._execute_job(job)
|
await self._execute_job(job)
|
||||||
|
|||||||
@ -24,6 +24,7 @@ from websockets.http11 import Request as WsRequest
|
|||||||
from websockets.http11 import Response
|
from websockets.http11 import Response
|
||||||
|
|
||||||
from nanobot.command.builtin import builtin_command_palette
|
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.cron.types import CronJob, CronSchedule
|
||||||
from nanobot.utils.subagent_channel_display import scrub_subagent_messages_for_channel
|
from nanobot.utils.subagent_channel_display import scrub_subagent_messages_for_channel
|
||||||
from nanobot.webui.file_preview import WebUIFilePreviewError, file_preview_payload
|
from nanobot.webui.file_preview import WebUIFilePreviewError, file_preview_payload
|
||||||
@ -579,6 +580,8 @@ class GatewayHTTPHandler:
|
|||||||
return _http_error(404, "automation not found")
|
return _http_error(404, "automation not found")
|
||||||
if job.payload.kind == "system_event":
|
if job.payload.kind == "system_event":
|
||||||
return _http_error(403, "system automation is protected")
|
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 action == "enable":
|
||||||
if self.cron_service.enable_job(job_id, enabled=True) is None:
|
if self.cron_service.enable_job(job_id, enabled=True) is None:
|
||||||
|
|||||||
@ -1026,6 +1026,20 @@ async def test_webui_automations_route_lists_all_jobs_and_allows_user_actions(
|
|||||||
)
|
)
|
||||||
assert disabled_run.status_code == 409
|
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(
|
protected_delete = await _http_get(
|
||||||
f"{base_url}/api/webui/automations/delete?id=heartbeat",
|
f"{base_url}/api/webui/automations/delete?id=heartbeat",
|
||||||
headers=auth,
|
headers=auth,
|
||||||
|
|||||||
@ -43,6 +43,68 @@ def test_add_job_accepts_valid_timezone(tmp_path) -> None:
|
|||||||
assert job.state.next_run_at_ms is not 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:
|
def test_add_job_migrates_legacy_delivery_context(tmp_path) -> None:
|
||||||
service = CronService(tmp_path / "cron" / "jobs.json")
|
service = CronService(tmp_path / "cron" / "jobs.json")
|
||||||
meta = {"slack": {"thread_ts": "1234567890.123456", "channel_type": "channel"}}
|
meta = {"slack": {"thread_ts": "1234567890.123456", "channel_type": "channel"}}
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user