Proactive Agent — runs on a cron (default 5am local) to autonomously handle
routine tasks based on learned user behavior.
Lifecycle per run
- Load USER.md + MEMORY.md for behavioral context.
- Collect overnight data from connected sources via
digest_collect.
- Use the LLM to classify each item and propose actions with a tier + permission key.
- For each proposed action:
- TRIVIAL tier → queue + immediately approve
- Known always_approve → queue + immediately approve
- Known always_deny → skip silently
- Everything else → queue as pending, notify user
- Execute all approved actions via
execute_pending_actions.
- Send the user a concise summary: what was done + numbered list of what needs approval.
Approval reply format (user replies to the notification message):
{action_id} yes approve one action
{action_id} no deny one action
always yes {action_id} approve + remember for this pattern
always no {action_id} deny + remember for this pattern
yes all / no all bulk decision
Wire up parse_approval_response from proactive_tools in your channel
message handler to process replies without running the full agent.
Scheduling
The agent self-registers a 5am daily cron task when register_cron is
called from your app startup:
from openjarvis.agents.proactive_agent import register_cron
register_cron(scheduler, notification_channel_id="your-channel-id")
Classes
ProactiveAgent(*args: Any, **kwargs: Any)
Bases: ToolUsingAgent
Autonomous agent that handles routine tasks based on learned user behavior.
Source code in src/openjarvis/agents/proactive_agent.py
| def __init__(self, *args: Any, **kwargs: Any) -> None:
self._notification_channel_id: str = kwargs.pop("notification_channel_id", "")
self._hours_back: int = kwargs.pop("hours_back", 24)
self._approval_store: Optional[ApprovalStore] = kwargs.pop(
"approval_store", None
)
self._timezone: str = kwargs.pop("timezone", "America/Los_Angeles")
# Read config defaults before super().__init__ so we can inject tools
try:
cfg = load_config()
p = cfg.proactive
if not self._notification_channel_id:
self._notification_channel_id = p.notification_channel
self._hours_back = p.hours_back
self._timezone = p.timezone
except Exception:
pass
# Build the required tools and inject them into the executor.
# This must happen before super().__init__ is called because
# ToolUsingAgent builds the ToolExecutor from kwargs["tools"].
store = self._approval_store or get_store()
self._approval_store = store
notification_channel = _build_notification_channel(
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
from openjarvis.tools.proactive_tools import (
CheckPermissionTool,
ExecutePendingActionsTool,
GetPendingActionsTool,
QueueActionTool,
RecordDecisionTool,
)
proactive_tools = [
DigestCollectTool(),
ExecutePendingActionsTool(store=store),
ChannelSendTool(channel=notification_channel),
CheckPermissionTool(store=store),
QueueActionTool(store=store),
GetPendingActionsTool(store=store),
RecordDecisionTool(store=store),
]
# Merge with any tools passed by the caller
caller_tools: List[Any] = kwargs.pop("tools", None) or []
kwargs["tools"] = proactive_tools + caller_tools
# The agent emits a JSON array of proposals — one entry per actionable
# item — and a typical morning digest produces dozens. The default
# max_tokens (often ~1024) truncates the array mid-element, which the
# parser then rejects. Give it real room unless the caller overrode.
kwargs.setdefault("max_tokens", 8192)
# Deterministic-ish output makes the JSON shape more reliable.
kwargs.setdefault("temperature", 0.2)
super().__init__(*args, **kwargs)
|
Functions
register_cron(scheduler: Any, *, notification_channel_id: str = '', cron_expr: str = '', hours_back: int = 0, timezone: str = '') -> Any
Register the proactive agent as a daily cron task.
All defaults are read from config.toml [proactive] when not explicitly
passed. Call this once from app startup after the scheduler is started.
| PARAMETER |
DESCRIPTION |
scheduler
|
A TaskScheduler instance.
TYPE:
Any
|
notification_channel_id
|
Override the channel ID from config. If empty, uses notification_channel
from [proactive] in config.toml.
TYPE:
str
DEFAULT:
''
|
cron_expr
|
Override the cron schedule. Defaults to config value ("0 5 * * *").
TYPE:
str
DEFAULT:
''
|
hours_back
|
Override hours of data to scan. Defaults to config value (24).
TYPE:
int
DEFAULT:
0
|
timezone
|
Override timezone string. Defaults to config value.
TYPE:
str
DEFAULT:
''
|
Source code in src/openjarvis/agents/proactive_agent.py
| def register_cron(
scheduler: Any,
*,
notification_channel_id: str = "",
cron_expr: str = "",
hours_back: int = 0,
timezone: str = "",
) -> Any:
"""Register the proactive agent as a daily cron task.
All defaults are read from ``config.toml [proactive]`` when not explicitly
passed. Call this once from app startup after the scheduler is started.
Parameters
----------
scheduler:
A ``TaskScheduler`` instance.
notification_channel_id:
Override the channel ID from config. If empty, uses ``notification_channel``
from ``[proactive]`` in config.toml.
cron_expr:
Override the cron schedule. Defaults to config value (``"0 5 * * *"``).
hours_back:
Override hours of data to scan. Defaults to config value (24).
timezone:
Override timezone string. Defaults to config value.
"""
try:
cfg = load_config()
p = cfg.proactive
notification_channel_id = notification_channel_id or p.notification_channel
cron_expr = cron_expr or p.schedule
hours_back = hours_back or p.hours_back
timezone = timezone or p.timezone
except Exception:
cron_expr = cron_expr or "0 5 * * *"
hours_back = hours_back or 24
timezone = timezone or "America/Los_Angeles"
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 = [
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")
)
]
# 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=_PROACTIVE_CRON_PROMPT,
schedule_type="cron",
schedule_value=cron_expr,
agent="proactive",
context_mode="isolated",
metadata=metadata,
)
|