Grayscale/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge PR #4299: feat(cron): bind scheduled automations to sessions

Browse Source 
feat(cron): bind scheduled automations to sessions
This commit is contained in:
Xubin Ren
2026-06-13 00:07:55 +08:00
committed by GitHub
parent 1b3b322674 32d8a1dd7b
commit dac4e39bcf
58 changed files with 2451 additions and 412 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/cli/test_commands.py
+158 -121
View File
@@ -8,13 +8,20 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from typer.testing import CliRunner
from nanobot.bus.events import OutboundMessage
from nanobot.cli.commands import _proactive_delivery_metadata, app
from nanobot.bus.events import InboundMessage, OutboundMessage
from nanobot.cli.commands import app
from nanobot.config.schema import Config
from nanobot.cron.service import CronJobSkippedError
from nanobot.cron.session_turns import CRON_DEFER_UNTIL_IDLE_META, CRON_TRIGGER_META
from nanobot.cron.types import CronJob, CronPayload
from nanobot.cron.webui_metadata import cron_proactive_delivery_metadata
from nanobot.providers.factory import ProviderSnapshot, make_provider
from nanobot.providers.openai_codex_provider import _strip_model_prefix
from nanobot.providers.registry import find_by_name
from nanobot.webui.metadata import (
WEBUI_MESSAGE_SOURCE_METADATA_KEY,
WEBUI_TURN_METADATA_KEY,
)
runner = CliRunner()
@@ -22,11 +29,11 @@ runner = CliRunner()
def test_proactive_websocket_delivery_gets_fresh_turn_id() -> None:
metadata = {
"webui": True,
"webui_turn_id": "turn-that-created-the-reminder",
WEBUI_TURN_METADATA_KEY: "turn-that-created-the-reminder",
"workspace_scope": {"mode": "default"},
}
out = _proactive_delivery_metadata(
out = cron_proactive_delivery_metadata(
"websocket",
metadata,
turn_seed="cron:drink-water",
@@ -35,9 +42,9 @@ def test_proactive_websocket_delivery_gets_fresh_turn_id() -> None:
assert out["webui"] is True
assert out["workspace_scope"] == {"mode": "default"}
assert out["webui_turn_id"].startswith("cron:drink-water:")
assert out["webui_turn_id"] != metadata["webui_turn_id"]
assert out["_webui_message_source"] == {"kind": "cron", "label": "drink water"}
assert out[WEBUI_TURN_METADATA_KEY].startswith("cron:drink-water:")
assert out[WEBUI_TURN_METADATA_KEY] != metadata[WEBUI_TURN_METADATA_KEY]
assert out[WEBUI_MESSAGE_SOURCE_METADATA_KEY] == {"kind": "cron", "label": "drink water"}
def _fake_provider():
@@ -1338,7 +1345,7 @@ def test_gateway_uses_workspace_directory_for_cron_store(monkeypatch, tmp_path:
assert seen["cron_store"] == config.workspace_path / "cron" / "jobs.json"
def test_gateway_cron_evaluator_receives_scheduled_reminder_context(
def test_gateway_unbound_agent_cron_is_skipped(
monkeypatch, tmp_path: Path
) -> None:
config_file = tmp_path / "instance" / "config.json"
@@ -1403,11 +1410,10 @@ def test_gateway_cron_evaluator_receives_scheduled_reminder_context(
seen["agent"] = self
async def process_direct(self, *_args, **_kwargs):
return OutboundMessage(
channel="telegram",
chat_id="user-1",
content="Time to stretch.",
)
raise AssertionError("unbound cron job must not use process_direct")
async def submit_cron_turn(self, _msg: InboundMessage):
raise AssertionError("unbound cron job must not run as a bound cron turn")
async def close_mcp(self) -> None:
return None
@@ -1423,16 +1429,10 @@ def test_gateway_cron_evaluator_receives_scheduled_reminder_context(
raise _StopGatewayError("stop")
async def _capture_evaluate_response(
response: str,
task_context: str,
provider_arg: object,
model: str,
*_args,
**_kwargs,
) -> bool:
seen["response"] = response
seen["task_context"] = task_context
seen["provider"] = provider_arg
seen["model"] = model
return True
raise AssertionError("unbound cron job must not be evaluated for delivery")
monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
@@ -1465,124 +1465,71 @@ def test_gateway_cron_evaluator_receives_scheduled_reminder_context(
),
)
response = asyncio.run(cron.on_job(job))
with pytest.raises(CronJobSkippedError, match="unbound agent cron job"):
asyncio.run(cron.on_job(job))
assert response == "Time to stretch."
assert seen["response"] == "Time to stretch."
assert seen["provider"] is runtime_provider
assert seen["model"] == "runtime-model"
assert seen["task_context"] == (
"The scheduled time has arrived. Deliver this reminder to the user now, "
"as a brief and natural message in their language. Speak directly to them — "
"do not narrate progress, summarize, include user IDs, or add status reports "
"like 'Done' or 'Reminded'.\n\n"
"Reminder: Remind me to stretch."
)
bus.publish_outbound.assert_awaited_once_with(
OutboundMessage(
channel="telegram",
chat_id="user-1",
content="Time to stretch.",
)
)
assert seen["session_key"] == "telegram:user-1"
saved_session = seen["saved_session"]
assert isinstance(saved_session, _FakeSession)
assert saved_session.messages == [
{
"role": "assistant",
"content": "Time to stretch.",
"_channel_delivery": True,
}
]
bus.publish_outbound.reset_mock()
old_turn_id = "turn-that-created-the-reminder"
websocket_job = CronJob(
id="drink-water",
name="drink water",
payload=CronPayload(
message="Remind me to drink water.",
deliver=True,
channel="websocket",
to="chat-1",
channel_meta={
"webui": True,
"webui_turn_id": old_turn_id,
"workspace_scope": {"mode": "default"},
},
session_key="websocket:chat-1",
),
)
response = asyncio.run(cron.on_job(websocket_job))
assert response == "Time to stretch."
bus.publish_outbound.assert_awaited_once()
delivered = bus.publish_outbound.await_args.args[0]
assert delivered.channel == "websocket"
assert delivered.chat_id == "chat-1"
assert delivered.metadata["webui"] is True
assert delivered.metadata["workspace_scope"] == {"mode": "default"}
assert delivered.metadata["webui_turn_id"].startswith("cron:drink-water:")
assert delivered.metadata["webui_turn_id"] != old_turn_id
assert delivered.metadata["_webui_message_source"] == {
"kind": "cron",
"label": "drink water",
}
bus.publish_outbound.assert_not_awaited()
def test_gateway_cron_job_suppresses_intermediate_progress(
def test_gateway_bound_cron_runs_as_session_turn(
monkeypatch, tmp_path: Path
) -> None:
"""Cron jobs must pass on_progress=_silent to process_direct so that
tool hints and streaming deltas are never leaked to the user channel
before evaluate_response decides whether to deliver."""
config_file = tmp_path / "instance" / "config.json"
config_file.parent.mkdir(parents=True)
config_file.write_text("{}")
config = Config()
config.agents.defaults.workspace = str(tmp_path / "config-workspace")
provider = _fake_provider()
bus = MagicMock()
bus.publish_outbound = AsyncMock()
seen: dict[str, object] = {}
seen: dict[str, object] = {"run_records": []}
monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: provider)
monkeypatch.setattr(
"nanobot.providers.factory.build_provider_snapshot",
lambda _config: _test_provider_snapshot(object(), _config),
lambda _config: _test_provider_snapshot(provider, _config),
)
monkeypatch.setattr(
"nanobot.providers.factory.load_provider_snapshot",
lambda _config_path=None: _test_provider_snapshot(object(), config),
lambda _config_path=None: _test_provider_snapshot(provider, config),
)
monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: bus)
monkeypatch.setattr("nanobot.session.manager.SessionManager", lambda _workspace: object())
class _FakeSessionManager:
def __init__(self, _workspace: Path) -> None:
pass
monkeypatch.setattr("nanobot.session.manager.SessionManager", _FakeSessionManager)
class _FakeCron:
def __init__(self, _store_path: Path) -> None:
self.on_job = None
seen["cron"] = self
def write_run_record(self, run_id: str, record: dict[str, object]) -> None:
seen["run_records"].append((run_id, record))
class _FakeAgentLoop:
@classmethod
def from_config(cls, config, bus=None, **extra):
return cls(**extra)
def __init__(self, *args, **kwargs) -> None:
self.model = "test-model"
self.provider = object()
self.provider = kwargs.get("provider", object())
self.tools = {}
seen["agent"] = self
async def process_direct(self, *_args, on_progress=None, **_kwargs):
seen["on_progress"] = on_progress
async def submit_cron_turn(self, msg: InboundMessage):
seen["cron_msg"] = msg
return OutboundMessage(
channel="telegram",
chat_id="user-1",
content="Done.",
channel=msg.channel,
chat_id=msg.chat_id,
content="Checked the repo.",
)
async def close_mcp(self) -> None:
@@ -1598,41 +1545,131 @@ def test_gateway_cron_job_suppresses_intermediate_progress(
def __init__(self, *_args, **_kwargs) -> None:
raise _StopGatewayError("stop")
async def _always_reject(*_args, **_kwargs) -> bool:
return False
async def _unexpected_evaluator(*_args, **_kwargs) -> bool:
raise AssertionError("bound cron must not use legacy response evaluator")
monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _StopAfterCronSetup)
monkeypatch.setattr(
"nanobot.cli.commands.evaluate_response",
_always_reject,
)
monkeypatch.setattr("nanobot.cli.commands.evaluate_response", _unexpected_evaluator)
result = runner.invoke(app, ["gateway", "--config", str(config_file)])
assert isinstance(result.exception, _StopGatewayError)
cron = seen["cron"]
job = CronJob(
id="cron-silent-test",
name="test-silent",
id="repo-check",
name="Repo check",
payload=CronPayload(
message="Run something.",
deliver=True,
channel="telegram",
to="user-1",
message="Check repository health.",
session_key="websocket:chat-1",
origin_channel="websocket",
origin_chat_id="chat-1",
),
)
response = asyncio.run(cron.on_job(job))
assert response == "Done."
# on_progress must be a callable (the _silent noop), not None and not bus_progress
assert seen["on_progress"] is not None
assert callable(seen["on_progress"])
# Verify it actually swallows calls (no side effects)
asyncio.run(seen["on_progress"]("tool_hint", "🔧 $ echo test"))
# Nothing published to bus since evaluator rejected
bus.publish_outbound.assert_not_awaited()
assert response == "Checked the repo."
msg = seen["cron_msg"]
assert isinstance(msg, InboundMessage)
assert msg.channel == "websocket"
assert msg.chat_id == "chat-1"
assert msg.sender_id == "cron"
assert msg.session_key_override == "websocket:chat-1"
assert "Cron job: Check repository health." in msg.content
assert msg.metadata["webui"] is True
assert msg.metadata[WEBUI_MESSAGE_SOURCE_METADATA_KEY] == {
"kind": "cron",
"label": "Repo check",
}
trigger = msg.metadata[CRON_TRIGGER_META]
assert trigger["job_id"] == "repo-check"
assert trigger["job_name"] == "Repo check"
assert trigger["persist_content"] == (
"Scheduled cron job triggered: Repo check\n\nCheck repository health."
)
assert msg.metadata[CRON_DEFER_UNTIL_IDLE_META] is True
statuses = [record["status"] for _run_id, record in seen["run_records"]]
assert statuses == ["queued", "ok"]
assert seen["run_records"][0][0] == seen["run_records"][1][0]
discord_job = CronJob(
id="thread-check",
name="Thread check",
payload=CronPayload(
message="Check the Discord thread.",
session_key="discord:456:thread:777",
origin_channel="discord",
origin_chat_id="777",
origin_metadata={
"context_chat_id": "456",
"parent_channel_id": "456",
"thread_id": "777",
},
),
)
response = asyncio.run(cron.on_job(discord_job))
assert response == "Checked the repo."
msg = seen["cron_msg"]
assert isinstance(msg, InboundMessage)
assert msg.channel == "discord"
assert msg.chat_id == "777"
assert msg.session_key_override == "discord:456:thread:777"
assert msg.metadata["context_chat_id"] == "456"
assert msg.metadata["parent_channel_id"] == "456"
assert msg.metadata["thread_id"] == "777"
telegram_job = CronJob(
id="telegram-topic",
name="Telegram topic",
payload=CronPayload(
message="Check the Telegram topic.",
session_key="telegram:-100123:topic:42",
origin_channel="telegram",
origin_chat_id="-100123",
origin_metadata={"message_thread_id": 42},
),
)
response = asyncio.run(cron.on_job(telegram_job))
assert response == "Checked the repo."
msg = seen["cron_msg"]
assert isinstance(msg, InboundMessage)
assert msg.channel == "telegram"
assert msg.chat_id == "-100123"
assert msg.session_key_override == "telegram:-100123:topic:42"
assert msg.metadata["message_thread_id"] == 42
feishu_job = CronJob(
id="feishu-topic",
name="Feishu topic",
payload=CronPayload(
message="Check the Feishu topic.",
session_key="feishu:oc_abc:om_root123",
origin_channel="feishu",
origin_chat_id="oc_abc",
origin_metadata={
"chat_type": "group",
"message_id": "om_root123",
"thread_id": "om_root123",
},
),
)
response = asyncio.run(cron.on_job(feishu_job))
assert response == "Checked the repo."
msg = seen["cron_msg"]
assert isinstance(msg, InboundMessage)
assert msg.channel == "feishu"
assert msg.chat_id == "oc_abc"
assert msg.session_key_override == "feishu:oc_abc:om_root123"
assert msg.metadata["message_id"] == "om_root123"
assert msg.metadata["thread_id"] == "om_root123"
def test_gateway_workspace_override_does_not_migrate_legacy_cron(