autumn_plugin_agent/
proactive.rs

1//! Always-on behaviour: heartbeats, follow-ups, and delivery.
2//!
3//! An always-on agent wakes up by itself. This module gives it three ways:
4//!
5//! * **Heartbeat** — a [`Heartbeat`] registers an Autumn scheduled task
6//!   (fixed delay or cron). Each tick runs the agent with a check-in prompt.
7//!   If nothing needs attention, the agent answers [`HEARTBEAT_OK`] and the
8//!   harness stays silent. Heartbeats run with read-only tools by default:
9//!   the agent may look and take notes, not act.
10//! * **Follow-ups** — the `schedule_followup` tool lets the agent book its
11//!   own next wake-up ("check the deploy in 20 minutes") as a delayed
12//!   `agent_run` job.
13//! * **Precheck** — a cheap app callback can skip a heartbeat tick before any
14//!   model call ("no new orders, nothing to look at").
15//!
16//! Whatever a background run has to say goes to a [`Delivery`]: the app's
17//! bridge to mail, chat, push, or an in-app inbox. [`LogDelivery`] is the
18//! default.
19
20use std::future::Future;
21use std::pin::Pin;
22use std::sync::Arc;
23use std::time::Duration;
24
25use autumn_web::task::{Schedule, TaskCoordination, TaskInfo};
26use autumn_web::{AppState, AutumnResult};
27use futures::future::BoxFuture;
28use serde::{Deserialize, Serialize};
29
30use crate::agent::{AgentOutcome, AgentRuntime};
31use crate::error::{AgentError, ErrorKind};
32use crate::ids::{RunId, SessionId};
33use crate::jobs::{AgentRunArgs, RunOrigin, enqueue_agent_run_in, execute_run};
34use crate::memory::MemoryScope;
35use crate::policy::{Strictest, ToolPolicy, ToolRules};
36use crate::tools::{Tool, ToolContext, ToolEffect};
37
38/// The answer that means "nothing needs attention".
39pub const HEARTBEAT_OK: &str = "HEARTBEAT_OK";
40
41/// Longest extra text an answer may carry next to [`HEARTBEAT_OK`] and still
42/// count as silent.
43pub const SILENT_ACK_MAX_CHARS: usize = 300;
44
45/// The default heartbeat prompt.
46pub const DEFAULT_HEARTBEAT_PROMPT: &str = "This is a scheduled heartbeat check-in. Review your memory, your standing instructions, and anything your tools can show you. If something needs the user's attention, report it in a few sentences. If nothing needs attention, reply with exactly HEARTBEAT_OK.";
47
48/// `true` when a background answer should not be delivered: it is
49/// [`HEARTBEAT_OK`], or starts or ends with it and adds at most
50/// [`SILENT_ACK_MAX_CHARS`] characters.
51#[must_use]
52pub fn is_silent(text: &str) -> bool {
53    let text = text.trim();
54    let rest = text
55        .strip_prefix(HEARTBEAT_OK)
56        .or_else(|| text.strip_suffix(HEARTBEAT_OK));
57    rest.is_some_and(|rest| rest.trim().chars().count() <= SILENT_ACK_MAX_CHARS)
58}
59
60/// What started a background run that has something to say.
61#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
62#[serde(rename_all = "snake_case")]
63pub enum ReportSource {
64    /// A heartbeat tick.
65    Heartbeat,
66    /// A follow-up the agent scheduled.
67    Followup,
68    /// An `agent_run` job the app enqueued with `deliver` set.
69    Job,
70    /// A resumed run.
71    Resume,
72}
73
74/// A message from a background run to a person.
75#[derive(Debug, Clone, PartialEq)]
76pub struct Report {
77    /// What started the run.
78    pub source: ReportSource,
79    /// The run's id.
80    pub run_id: RunId,
81    /// The session the run belongs to, if any. Use it to route the report
82    /// back to the right chat thread.
83    pub session_id: Option<SessionId>,
84    /// Text for a person: the answer, or what waits for approval.
85    pub text: String,
86    /// The full outcome. A paused run carries the
87    /// [`RunState`](crate::agent::RunState) to store for the reviewer.
88    pub outcome: AgentOutcome,
89}
90
91/// Where background results go: mail, chat, push, an inbox table.
92pub trait Delivery: Send + Sync + std::fmt::Debug {
93    /// Deliver one report.
94    fn deliver<'a>(&'a self, report: &'a Report) -> BoxFuture<'a, Result<(), AgentError>>;
95}
96
97/// Logs reports with `tracing`. The default [`Delivery`].
98#[derive(Debug, Clone, Copy, Default)]
99pub struct LogDelivery;
100
101impl Delivery for LogDelivery {
102    fn deliver<'a>(&'a self, report: &'a Report) -> BoxFuture<'a, Result<(), AgentError>> {
103        tracing::info!(
104            source = ?report.source,
105            run_id = %report.run_id,
106            session = report.session_id.as_ref().map(SessionId::as_str),
107            text = %report.text,
108            "agent report"
109        );
110        Box::pin(std::future::ready(Ok(())))
111    }
112}
113
114/// The text a person should see for an outcome, or `None` to stay quiet.
115///
116/// Final answers are delivered unless [`is_silent`]; paused runs always are
117/// (someone has to approve). Budget and loop stops are only logged.
118#[must_use]
119pub fn report_text(outcome: &AgentOutcome) -> Option<String> {
120    match outcome {
121        AgentOutcome::Completed { text, .. } if !is_silent(text) => Some(text.clone()),
122        AgentOutcome::AwaitingApproval { state } => {
123            let lines: Vec<String> = state
124                .awaiting()
125                .map(|pending| format!("- {} {}", pending.call.name, pending.call.arguments))
126                .collect();
127            Some(format!(
128                "The agent wants to run these calls and needs approval:\n{}",
129                lines.join("\n")
130            ))
131        }
132        _ => None,
133    }
134}
135
136/// Deliver an outcome when it has something to say. Delivery failures are
137/// logged, not returned: failing the job would re-run the whole agent.
138pub(crate) async fn deliver_outcome(
139    delivery: &dyn Delivery,
140    source: ReportSource,
141    run_id: &RunId,
142    session_id: Option<&SessionId>,
143    outcome: &AgentOutcome,
144) {
145    let Some(text) = report_text(outcome) else {
146        tracing::debug!(run_id = %run_id, ?source, "background run had nothing to report");
147        return;
148    };
149    let report = Report {
150        source,
151        run_id: run_id.clone(),
152        session_id: session_id.cloned(),
153        text,
154        outcome: outcome.clone(),
155    };
156    if let Err(err) = delivery.deliver(&report).await {
157        tracing::error!(run_id = %run_id, error = %err, "cannot deliver the agent report");
158    }
159}
160
161/// A cheap app check that can skip a heartbeat tick before any model call.
162pub type Precheck = Arc<dyn Fn(AppState) -> BoxFuture<'static, bool> + Send + Sync>;
163
164/// When a heartbeat fires.
165#[derive(Debug, Clone, PartialEq, Eq)]
166pub enum HeartbeatSchedule {
167    /// After a fixed delay from the end of the previous tick.
168    Every(Duration),
169    /// On a 6-field cron expression (`sec min hour day month weekday`).
170    Cron {
171        /// The expression, for example `"0 */30 8-18 * * Mon-Fri"`.
172        expression: String,
173        /// IANA time zone, for example `"Europe/Oslo"`. UTC when `None`.
174        timezone: Option<String>,
175    },
176}
177
178/// A periodic check-in for an always-on agent.
179///
180/// ```rust
181/// use std::time::Duration;
182/// use autumn_plugin_agent::proactive::Heartbeat;
183///
184/// // Every 30 minutes during office hours, Oslo time, in the "ops" session.
185/// let heartbeat = Heartbeat::cron("0 */30 8-18 * * Mon-Fri")
186///     .timezone("Europe/Oslo")
187///     .session("ops");
188/// # let _ = Heartbeat::every(Duration::from_secs(1_800));
189/// ```
190#[derive(Clone)]
191pub struct Heartbeat {
192    schedule: HeartbeatSchedule,
193    prompt: String,
194    session: Option<SessionId>,
195    memory_scope: Option<MemoryScope>,
196    read_only: bool,
197    coordination: TaskCoordination,
198    precheck: Option<Precheck>,
199}
200
201impl std::fmt::Debug for Heartbeat {
202    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
203        f.debug_struct("Heartbeat")
204            .field("schedule", &self.schedule)
205            .field("session", &self.session)
206            .field("memory_scope", &self.memory_scope)
207            .field("read_only", &self.read_only)
208            .field("coordination", &self.coordination)
209            .field("precheck", &self.precheck.is_some())
210            .finish_non_exhaustive()
211    }
212}
213
214impl Heartbeat {
215    /// The scheduled-task name the heartbeat registers.
216    pub const TASK_NAME: &'static str = "agent_heartbeat";
217
218    fn with_schedule(schedule: HeartbeatSchedule) -> Self {
219        Self {
220            schedule,
221            prompt: DEFAULT_HEARTBEAT_PROMPT.to_owned(),
222            session: None,
223            memory_scope: None,
224            read_only: true,
225            coordination: TaskCoordination::Fleet,
226            precheck: None,
227        }
228    }
229
230    /// Tick after a fixed delay from the end of the previous tick.
231    #[must_use]
232    pub fn every(interval: Duration) -> Self {
233        Self::with_schedule(HeartbeatSchedule::Every(interval))
234    }
235
236    /// Tick on a 6-field cron expression (`sec min hour day month weekday`).
237    #[must_use]
238    pub fn cron(expression: impl Into<String>) -> Self {
239        Self::with_schedule(HeartbeatSchedule::Cron {
240            expression: expression.into(),
241            timezone: None,
242        })
243    }
244
245    /// The IANA time zone for a cron schedule. Ignored for fixed delays.
246    #[must_use]
247    pub fn timezone(mut self, timezone: impl Into<String>) -> Self {
248        if let HeartbeatSchedule::Cron { timezone: tz, .. } = &mut self.schedule {
249            *tz = Some(timezone.into());
250        }
251        self
252    }
253
254    /// Replace the check-in prompt. Keep the [`HEARTBEAT_OK`] instruction
255    /// in it, or every tick will be delivered.
256    #[must_use]
257    pub fn prompt(mut self, prompt: impl Into<String>) -> Self {
258        self.prompt = prompt.into();
259        self
260    }
261
262    /// Run ticks inside this session, so the agent sees earlier ticks.
263    #[must_use]
264    pub fn session(mut self, session: impl Into<SessionId>) -> Self {
265        self.session = Some(session.into());
266        self
267    }
268
269    /// Bind memory to this scope instead of [`MemoryScope::agent`].
270    #[must_use]
271    pub fn memory_scope(mut self, scope: MemoryScope) -> Self {
272        self.memory_scope = Some(scope);
273        self
274    }
275
276    /// Let ticks use every tool the app policy allows, not only read-only
277    /// and internal ones.
278    #[must_use]
279    pub const fn allow_actions(mut self) -> Self {
280        self.read_only = false;
281        self
282    }
283
284    /// Run on every replica instead of once per tick across the fleet.
285    #[must_use]
286    pub const fn per_replica(mut self) -> Self {
287        self.coordination = TaskCoordination::PerReplica;
288        self
289    }
290
291    /// Skip a tick, before any model call, when `check` answers `false`.
292    #[must_use]
293    pub fn precheck<F, Fut>(mut self, check: F) -> Self
294    where
295        F: Fn(AppState) -> Fut + Send + Sync + 'static,
296        Fut: Future<Output = bool> + Send + 'static,
297    {
298        self.precheck = Some(Arc::new(move |state| Box::pin(check(state))));
299        self
300    }
301
302    /// The configured schedule.
303    #[must_use]
304    pub const fn schedule(&self) -> &HeartbeatSchedule {
305        &self.schedule
306    }
307
308    /// Check the schedule and prompt.
309    ///
310    /// # Errors
311    ///
312    /// Returns [`ErrorKind::Config`] for an interval under one second, a
313    /// blank cron expression, or a blank prompt.
314    pub fn validate(&self) -> Result<(), AgentError> {
315        match &self.schedule {
316            HeartbeatSchedule::Every(interval) if *interval < Duration::from_secs(1) => {
317                return Err(AgentError::new(
318                    ErrorKind::Config,
319                    "heartbeat interval must be at least one second",
320                ));
321            }
322            HeartbeatSchedule::Cron { expression, .. } if expression.trim().is_empty() => {
323                return Err(AgentError::new(
324                    ErrorKind::Config,
325                    "heartbeat cron expression must not be blank",
326                ));
327            }
328            _ => {}
329        }
330        if self.prompt.trim().is_empty() {
331            return Err(AgentError::new(
332                ErrorKind::Config,
333                "heartbeat prompt must not be blank",
334            ));
335        }
336        Ok(())
337    }
338
339    /// The Autumn scheduled task for this heartbeat.
340    pub(crate) fn task_info(&self) -> TaskInfo {
341        let schedule = match &self.schedule {
342            HeartbeatSchedule::Every(interval) => Schedule::FixedDelay(*interval),
343            HeartbeatSchedule::Cron {
344                expression,
345                timezone,
346            } => Schedule::Cron {
347                expression: expression.clone(),
348                timezone: timezone.clone(),
349            },
350        };
351        TaskInfo {
352            name: Self::TASK_NAME.to_owned(),
353            schedule,
354            coordination: self.coordination,
355            handler: heartbeat_tick,
356        }
357    }
358
359    /// Run one tick. Returns `None` when the precheck skipped it.
360    ///
361    /// # Errors
362    ///
363    /// Returns [`AgentError`] when the run fails (see
364    /// [`Agent::run`](crate::agent::Agent::run)).
365    pub async fn tick(
366        &self,
367        runtime: &AgentRuntime,
368        state: &AppState,
369    ) -> Result<Option<AgentOutcome>, AgentError> {
370        if let Some(check) = &self.precheck
371            && !check(state.clone()).await
372        {
373            tracing::debug!("heartbeat precheck found nothing to do; skipping the tick");
374            return Ok(None);
375        }
376        let args = AgentRunArgs::new(self.prompt.clone())
377            .session(self.session.clone())
378            .memory_scope(self.memory_scope.clone())
379            .deliver(true)
380            .origin(RunOrigin::Heartbeat);
381        let policy: Option<Arc<dyn ToolPolicy>> = self.read_only.then(|| {
382            Arc::new(Strictest::new(vec![
383                runtime.policy(),
384                Arc::new(ToolRules::read_only()),
385            ])) as Arc<dyn ToolPolicy>
386        });
387        execute_run(runtime, args, policy).await.map(Some)
388    }
389}
390
391/// The heartbeat as installed on app state by the plugin's startup hook.
392#[derive(Debug, Clone)]
393pub(crate) struct HeartbeatSettings(pub(crate) Heartbeat);
394
395/// The scheduled-task handler behind [`Heartbeat`].
396fn heartbeat_tick(state: AppState) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>> {
397    Box::pin(async move {
398        let (Some(runtime), Some(settings)) = (
399            state.extension::<AgentRuntime>(),
400            state.extension::<HeartbeatSettings>(),
401        ) else {
402            tracing::warn!("agent heartbeat fired before the agent plugin started");
403            return Ok(());
404        };
405        settings
406            .0
407            .tick(&runtime, &state)
408            .await
409            .map(|_| ())
410            .map_err(|err| {
411                tracing::error!(error = %err, "agent heartbeat failed");
412                err.into_autumn_error()
413            })
414    })
415}
416
417/// The `schedule_followup` tool: the agent books its own next wake-up.
418///
419/// The follow-up runs as a delayed `agent_run` job in the same session, with
420/// delivery on. Turn it on with
421/// [`AgentPlugin::followups`](crate::plugin::AgentPlugin::followups).
422#[derive(Debug, Clone)]
423pub struct FollowupTool {
424    max_delay: Duration,
425    memory_scope: Option<MemoryScope>,
426    depth: u32,
427}
428
429impl FollowupTool {
430    /// Longest prompt a follow-up may carry, in characters.
431    pub const MAX_PROMPT_CHARS: usize = 2_000;
432
433    /// Longest chain of follow-ups (a follow-up that schedules a follow-up
434    /// that ...). Stops an agent from waking itself forever.
435    pub const MAX_CHAIN: u32 = 10;
436
437    /// Allow follow-ups up to `max_delay` from now.
438    #[must_use]
439    pub const fn new(max_delay: Duration) -> Self {
440        Self {
441            max_delay,
442            memory_scope: None,
443            depth: 0,
444        }
445    }
446
447    /// The run this tool serves is `depth` follow-ups deep.
448    #[must_use]
449    pub const fn depth(mut self, depth: u32) -> Self {
450        self.depth = depth;
451        self
452    }
453
454    /// Run follow-ups with memory bound to this scope.
455    #[must_use]
456    pub fn memory_scope(mut self, scope: MemoryScope) -> Self {
457        self.memory_scope = Some(scope);
458        self
459    }
460
461    /// Check the input and build the delayed job's arguments.
462    fn plan(
463        &self,
464        input: &serde_json::Value,
465        ctx: &ToolContext,
466    ) -> Result<(AgentRunArgs, Duration), AgentError> {
467        let depth = self.depth.saturating_add(1);
468        if depth > Self::MAX_CHAIN {
469            return Err(AgentError::new(
470                ErrorKind::Tool,
471                format!(
472                    "this run is already {} follow-ups deep; finish now and report instead of scheduling another",
473                    self.depth
474                ),
475            ));
476        }
477        let max_minutes = self.max_delay.as_secs() / 60;
478        let minutes = input
479            .get("delay_minutes")
480            .and_then(serde_json::Value::as_u64)
481            .filter(|minutes| (1..=max_minutes).contains(minutes))
482            .ok_or_else(|| {
483                AgentError::new(
484                    ErrorKind::Tool,
485                    format!("delay_minutes must be a whole number from 1 to {max_minutes}"),
486                )
487            })?;
488        let prompt = input
489            .get("prompt")
490            .and_then(serde_json::Value::as_str)
491            .map(str::trim)
492            .filter(|prompt| !prompt.is_empty())
493            .ok_or_else(|| {
494                AgentError::new(ErrorKind::Tool, "give a `prompt` for the follow-up run")
495            })?;
496        if prompt.chars().count() > Self::MAX_PROMPT_CHARS {
497            return Err(AgentError::new(
498                ErrorKind::Tool,
499                format!(
500                    "the follow-up prompt must be at most {} characters",
501                    Self::MAX_PROMPT_CHARS
502                ),
503            ));
504        }
505        let mut args = AgentRunArgs::new(prompt)
506            .session(ctx.session_id.clone())
507            .memory_scope(self.memory_scope.clone())
508            .deliver(true)
509            .origin(RunOrigin::Followup);
510        args.followup_depth = depth;
511        Ok((args, Duration::from_secs(minutes.saturating_mul(60))))
512    }
513}
514
515impl Tool for FollowupTool {
516    fn name(&self) -> &'static str {
517        "schedule_followup"
518    }
519
520    fn description(&self) -> &'static str {
521        "Wake yourself up later: after delay_minutes, a new run starts with `prompt` in this conversation, and its answer is sent to the user. Use it to check back on something that is not ready yet."
522    }
523
524    fn input_schema(&self) -> serde_json::Value {
525        serde_json::json!({
526            "type": "object",
527            "properties": {
528                "delay_minutes": {
529                    "type": "integer",
530                    "minimum": 1,
531                    "maximum": self.max_delay.as_secs() / 60,
532                },
533                "prompt": {
534                    "type": "string",
535                    "description": "What your future self should do, with every detail it needs.",
536                }
537            },
538            "required": ["delay_minutes", "prompt"]
539        })
540    }
541
542    fn effect(&self) -> ToolEffect {
543        ToolEffect::Internal
544    }
545
546    fn execute<'a>(
547        &'a self,
548        input: serde_json::Value,
549        ctx: &'a ToolContext,
550    ) -> Pin<Box<dyn Future<Output = Result<serde_json::Value, AgentError>> + Send + 'a>> {
551        Box::pin(async move {
552            let (args, delay) = self.plan(&input, ctx)?;
553            let run_id = args.run_id.clone();
554            enqueue_agent_run_in(args, delay).await.map_err(|err| {
555                AgentError::new(
556                    ErrorKind::Tool,
557                    format!("cannot schedule the follow-up: {err}"),
558                )
559            })?;
560            Ok(serde_json::json!({
561                "scheduled": true,
562                "in_minutes": delay.as_secs() / 60,
563                "run_id": run_id,
564            }))
565        })
566    }
567}
568
569#[cfg(test)]
570mod tests;
autumn_plugin_agent/proactive.rs

autumn_plugin_agent/
proactive.rs
