From ff45e6c2327ab887695e2f369e8b4d73d4b26432 Mon Sep 17 00:00:00 2001 From: Elliot Slusky Date: Thu, 6 Aug 2026 13:02:37 -0700 Subject: [PATCH] Fix proactive cron reconciliation and notifications --- src/openjarvis/agents/proactive_agent.py | 121 +++++++++++++++------ tests/agents/test_proactive_agent.py | 131 +++++++++++++++++++++++ 2 files changed, 217 insertions(+), 35 deletions(-) create mode 100644 tests/agents/test_proactive_agent.py diff --git a/src/openjarvis/agents/proactive_agent.py b/src/openjarvis/agents/proactive_agent.py index ce75cb81..eda3b590 100644 --- a/src/openjarvis/agents/proactive_agent.py +++ b/src/openjarvis/agents/proactive_agent.py @@ -47,8 +47,6 @@ from openjarvis.core.config import load_config from openjarvis.core.paths import get_config_dir from openjarvis.core.registry import AgentRegistry from openjarvis.core.types import Message, Role, ToolCall - -logger = logging.getLogger(__name__) from openjarvis.tools.approval_store import ( DECISION_ALWAYS_APPROVE, DECISION_ALWAYS_DENY, @@ -59,6 +57,15 @@ from openjarvis.tools.approval_store import ( ) from openjarvis.tools.proactive_tools import get_store +logger = logging.getLogger(__name__) + +_PROACTIVE_CRON_PROMPT = ( + "Run the proactive agent: collect overnight data, execute approved actions, " + "notify pending approvals." +) +_PROACTIVE_TASK_KEY = "proactive-daily" +_PROACTIVE_TASK_KEY_FIELD = "openjarvis_task_key" + _SYSTEM_PROMPT = """You are a proactive personal assistant agent. You have already collected data from the user's connected sources (email, messages, calendar). Your job is to: @@ -266,17 +273,20 @@ def _build_notification_channel(channel_spec: str) -> Optional[Any]: except Exception: _kwargs = {} instance = channel_cls(**_kwargs) - # NOTE: do NOT call instance.connect() here. - # The notification channel only needs to *send* messages — it must - # NOT start a polling/getUpdates loop. Starting a second poll loop - # (with the same bot token) inside the same process would cause a - # Telegram Conflict error that kills the main channel's listener. - # TelegramChannel.send() works fine without connect() because it - # opens a fresh HTTP connection per message rather than relying on - # the long-poll thread. + # Telegram.send() is self-contained, while connect() starts a + # getUpdates loop. A second loop for the same bot token conflicts + # with the server's main listener. Other channel implementations + # may initialize resources required by send() in connect(), so keep + # their established lifecycle intact. + if channel_type != "telegram": + instance.connect() return instance except Exception: - pass + logger.warning( + "Failed to build proactive notification channel %s", + channel_type, + exc_info=True, + ) return None @@ -316,6 +326,7 @@ class ProactiveAgent(ToolUsingAgent): self._notification_channel_id ) self._notification_channel = notification_channel + self._notification_destination = self._notification_channel_id.partition(":")[2] from openjarvis.tools.channel_tools import ChannelSendTool from openjarvis.tools.digest_collect import DigestCollectTool @@ -501,13 +512,13 @@ class ProactiveAgent(ToolUsingAgent): # --- Step 5: Build and send notification --- notification = self._build_notification(executed_results, pending_actions) - if notification and self._notification_channel_id: + if notification and self._notification_destination: send_call = ToolCall( id="proactive-notify-1", name="channel_send", arguments=json.dumps( { - "channel": self._notification_channel_id, + "channel": self._notification_destination, "content": notification, } ), @@ -609,34 +620,74 @@ def register_cron( hours_back = hours_back or 24 timezone = timezone or "America/Los_Angeles" - # Idempotent: jarvis serve calls this on every startup, and create_task - # persists to scheduler.db. Without this check each restart added one - # more copy of the daily cron -- 68 duplicates accumulated, and when due - # they all fired back-to-back, monopolizing the (single-slot) Ollama - # queue so real chat messages stalled behind them. - prompt = "Run the proactive agent: collect overnight data, execute approved actions, notify pending approvals." + metadata = { + "notification_channel_id": notification_channel_id, + "hours_back": hours_back, + "timezone": timezone, + _PROACTIVE_TASK_KEY_FIELD: _PROACTIVE_TASK_KEY, + } + + # Match the stable key for tasks created by this version and the historical + # agent+prompt signature so existing installations are migrated on startup. existing = [ - t for t in scheduler.list_tasks(status="active") - if t.agent == "proactive" and t.prompt == prompt + task + for task in scheduler.list_tasks() + if task.status in {"active", "paused"} + and task.agent == "proactive" + and ( + task.metadata.get(_PROACTIVE_TASK_KEY_FIELD) == _PROACTIVE_TASK_KEY + or (task.prompt == _PROACTIVE_CRON_PROMPT and task.schedule_type == "cron") + ) ] - if existing: - # Keep the first, cancel any extra duplicates from earlier restarts. - for dup in existing[1:]: - try: - scheduler.cancel_task(dup.id) - except Exception: - logger.warning("Failed to cancel duplicate proactive task %s", dup.id) - return existing[0] + + # A scheduler pause is an explicit user choice and must survive restart. + # Keep one deterministically and remove any active or paused duplicates. + paused = [task for task in existing if task.status == "paused"] + if paused: + keep = min(paused, key=lambda task: task.id) + _cancel_proactive_duplicates(scheduler, existing, keep=keep) + return keep + + matching = [ + task + for task in existing + if task.prompt == _PROACTIVE_CRON_PROMPT + and task.schedule_type == "cron" + and task.schedule_value == cron_expr + and task.context_mode == "isolated" + and task.metadata == metadata + ] + if matching: + keep = min(matching, key=lambda task: task.id) + _cancel_proactive_duplicates(scheduler, existing, keep=keep) + return keep + + # Configuration changed. Replace stale active tasks so the schedule and + # notification settings from config.toml take effect on this startup. + _cancel_proactive_duplicates(scheduler, existing) return scheduler.create_task( - prompt=prompt, + prompt=_PROACTIVE_CRON_PROMPT, schedule_type="cron", schedule_value=cron_expr, agent="proactive", context_mode="isolated", - metadata={ - "notification_channel_id": notification_channel_id, - "hours_back": hours_back, - "timezone": timezone, - }, + metadata=metadata, ) + + +def _cancel_proactive_duplicates( + scheduler: Any, tasks: List[Any], *, keep: Optional[Any] = None +) -> None: + """Cancel managed proactive tasks other than *keep*.""" + for task in tasks: + if keep is not None and task.id == keep.id: + continue + try: + scheduler.cancel_task(task.id) + except Exception: + logger.warning( + "Failed to cancel duplicate proactive task %s", + task.id, + exc_info=True, + ) diff --git a/tests/agents/test_proactive_agent.py b/tests/agents/test_proactive_agent.py new file mode 100644 index 00000000..5e656335 --- /dev/null +++ b/tests/agents/test_proactive_agent.py @@ -0,0 +1,131 @@ +"""Regression tests for proactive scheduling and notification setup.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from openjarvis.agents.proactive_agent import ( + _PROACTIVE_CRON_PROMPT, + _build_notification_channel, + register_cron, +) +from openjarvis.core.registry import ChannelRegistry +from openjarvis.scheduler.scheduler import TaskScheduler +from openjarvis.scheduler.store import SchedulerStore + + +@pytest.fixture() +def scheduler(tmp_path): + store = SchedulerStore(tmp_path / "scheduler.db") + scheduler = TaskScheduler(store) + yield scheduler + scheduler.stop() + store.close() + + +def _register(scheduler, *, schedule="0 5 * * *", channel="telegram:123"): + return register_cron( + scheduler, + notification_channel_id=channel, + cron_expr=schedule, + hours_back=24, + timezone="UTC", + ) + + +class TestRegisterCron: + def test_reuses_exact_task_and_cancels_duplicates(self, scheduler): + first = _register(scheduler) + duplicate = scheduler.create_task( + _PROACTIVE_CRON_PROMPT, + "cron", + "0 5 * * *", + agent="proactive", + metadata=first.metadata, + ) + + returned = _register(scheduler) + + assert returned.id in {first.id, duplicate.id} + assert [task.id for task in scheduler.list_tasks(status="active")] == [ + returned.id + ] + cancelled_id = scheduler.list_tasks(status="cancelled")[0].id + assert cancelled_id == ({first.id, duplicate.id} - {returned.id}).pop() + + def test_replaces_task_when_configuration_changes(self, scheduler): + old = _register(scheduler, schedule="0 5 * * *", channel="telegram:old") + + new = _register(scheduler, schedule="0 7 * * *", channel="telegram:new") + + assert new.id != old.id + assert new.schedule_value == "0 7 * * *" + assert new.metadata["notification_channel_id"] == "telegram:new" + assert scheduler.list_tasks(status="cancelled")[0].id == old.id + + def test_preserves_pause_across_restart(self, scheduler): + paused = _register(scheduler) + scheduler.pause_task(paused.id) + + returned = _register(scheduler, schedule="0 7 * * *") + + assert returned.id == paused.id + assert returned.status == "paused" + assert scheduler.list_tasks(status="active") == [] + + def test_migrates_legacy_tasks_without_stable_key(self, scheduler): + legacy = scheduler.create_task( + _PROACTIVE_CRON_PROMPT, + "cron", + "0 5 * * *", + agent="proactive", + metadata={ + "notification_channel_id": "telegram:123", + "hours_back": 24, + "timezone": "UTC", + }, + ) + + current = _register(scheduler) + + assert current.id != legacy.id + assert current.metadata["openjarvis_task_key"] == "proactive-daily" + assert scheduler.list_tasks(status="cancelled")[0].id == legacy.id + + +class TestNotificationChannel: + def test_telegram_is_configured_without_starting_polling(self): + class FakeTelegram: + def __init__(self, **kwargs): + self.kwargs = kwargs + self.connect = MagicMock() + + config = MagicMock() + with ( + patch.object(ChannelRegistry, "contains", return_value=True), + patch.object(ChannelRegistry, "get", return_value=FakeTelegram), + patch("openjarvis.core.config.load_config", return_value=config), + patch( + "openjarvis.system._channel_kwargs.build_channel_kwargs", + return_value={"bot_token": "configured-token"}, + ), + ): + channel = _build_notification_channel("telegram:123") + + assert channel.kwargs == {"bot_token": "configured-token"} + channel.connect.assert_not_called() + + def test_non_telegram_channel_keeps_connect_lifecycle(self): + class FakeChannel: + def __init__(self, **kwargs): + self.connect = MagicMock() + + with ( + patch.object(ChannelRegistry, "contains", return_value=True), + patch.object(ChannelRegistry, "get", return_value=FakeChannel), + ): + channel = _build_notification_channel("twilio:15551234567") + + channel.connect.assert_called_once_with()