Vakyartha documentation

1//! Proactive heartbeat (docs/design/29-personal-os.md P7): one bounded LLM
2//! review turn per interval in a dedicated persistent session. Anti-nag by
3//! architecture — a "nothing" reply costs tokens but produces zero inbox
4//! noise and zero deliveries; findings park as `Kind::Heartbeat` inbox
5//! entries and reach chat surfaces only when a line carries an URGENT
6//! marker.
7
8use std::sync::Arc;
9use std::sync::atomic::{AtomicBool, Ordering};
10use std::time::Duration;
11
12use chrono::{DateTime, Timelike, Utc};
13use tokio_util::sync::CancellationToken;
14use vak_core::Core;
15use vak_session::{SessionHeader, SessionLog};
16
17use crate::AppState;
18
19/// Fixed ledger id of the dedicated heartbeat session.
20pub(crate) const HEARTBEAT_SESSION_ID: &str = "heartbeat";
21
22/// Upper bound on one heartbeat turn so a stuck stream cannot wedge the
23/// scheduler (mirrors REFLECTION_CALL_TIMEOUT).
24const HEARTBEAT_TURN_TIMEOUT: Duration = Duration::from_secs(120);
25
26/// Grace period for the agent loop to observe cancellation and hand the
27/// ledger back before the cycle gives up on it.
28const CANCEL_GRACE: Duration = Duration::from_secs(10);
29
30/// Cadence of the cheap due-check pass; the real gate is
31/// `[heartbeat].interval_secs` (minimum 300).
32pub(crate) const TICK_SECS: u64 = 5;
33
34pub(crate) struct HeartbeatRuntime {
35 /// Per-process fire timer; a restart re-fires immediately, matching
36 /// interval tasks' self-healing behavior.
37 last_fire: std::sync::Mutex<Option<DateTime<Utc>>>,
38 inflight: AtomicBool,
39}
40
41impl HeartbeatRuntime {
42 pub(crate) fn new() -> Self {
43 HeartbeatRuntime {
44 last_fire: std::sync::Mutex::new(None),
45 inflight: AtomicBool::new(false),
46 }
47 }
48}
49
50fn lock<T>(m: &std::sync::Mutex<T>) -> std::sync::MutexGuard<'_, T> {
51 m.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
52}
53
54/// Pure due-matrix: never-run fires immediately; otherwise only once a
55/// full interval has elapsed.
56fn heartbeat_due(last_fire: Option<DateTime<Utc>>, now: DateTime<Utc>, interval_secs: u64) -> bool {
57 match last_fire {
58 None => true,
59 Some(t) => (now - t).num_seconds() >= interval_secs as i64,
60 }
61}
62
63#[derive(Debug, PartialEq, Eq)]
64enum HeartbeatReply {
65 Nothing,
66 Findings {
67 /// Original trimmed finding lines (URGENT markers intact).
68 lines: Vec<String>,
69 /// Lines with any leading "URGENT:" marker removed.
70 stripped: Vec<String>,
71 urgent: bool,
72 },
73}
74
75fn classify_reply(text: &str, max_findings: usize) -> HeartbeatReply {
76 let trimmed = text.trim();
77 if trimmed.is_empty() || trimmed.eq_ignore_ascii_case("nothing") {
78 return HeartbeatReply::Nothing;
79 }
80 let lines: Vec<String> = trimmed
81 .lines()
82 .map(str::trim)
83 .filter(|l| !l.is_empty())
84 .take(max_findings.max(1))
85 .map(String::from)
86 .collect();
87 let urgent = lines.iter().any(|l| l.starts_with("URGENT:"));
88 let stripped = lines
89 .iter()
90 .map(|l| {
91 l.strip_prefix("URGENT:")
92 .map(str::trim_start)
93 .unwrap_or(l)
94 .to_string()
95 })
96 .collect();
97 HeartbeatReply::Findings {
98 lines,
99 stripped,
100 urgent,
101 }
102}
103
104/// One cheap scheduler pass. Errors are values: failures log and leave the
105/// next cycle to retry — this function never panics.
106pub(crate) async fn heartbeat_tick(state: &AppState) {
107 if !state.core.config().heartbeat.enabled {
108 return;
109 }
110 if state.heartbeat.inflight.swap(true, Ordering::SeqCst) {
111 return;
112 }
113 let result = heartbeat_cycle(state).await;
114 state.heartbeat.inflight.store(false, Ordering::SeqCst);
115 if let Err(e) = result {
116 eprintln!("[heartbeat] cycle failed: {e}");
117 }
118}
119
120async fn heartbeat_cycle(state: &AppState) -> Result<(), String> {
121 let cfg = state.core.config().heartbeat.clone();
122 let now = Utc::now();
123 let due = {
124 let last = lock(&state.heartbeat.last_fire);
125 heartbeat_due(*last, now, cfg.interval_secs)
126 };
127 if !due {
128 return Ok(());
129 }
130 if let Some(window) = cfg.quiet_hours {
131 let local = chrono::Local::now();
132 let minutes = local.hour() * 60 + local.minute();
133 if window.contains(minutes) {
134 return Ok(());
135 }
136 }
137 // Same day-spend read the budget alerts use; denial skips this cycle
138 // silently — an unattended prober must never be what bursts a cap.
139 let mut day_total = vak_core::finops::FinOpsLedger::new(&state.core.shared_data_home())
140 .day_total_usd(Utc::now());
141 if state.core.sessions_home() != state.core.shared_data_home() {
142 day_total += vak_core::finops::FinOpsLedger::new(&state.core.sessions_home())
143 .day_total_usd(Utc::now());
144 }
145 if let Some(cap) = state.core.config().finops.max_day_usd
146 && day_total >= cap
147 {
148 return Ok(());
149 }
150 // Claim the slot BEFORE dispatching so overlapping ticks cannot
151 // double-fire while a turn outlives one tick period.
152 *lock(&state.heartbeat.last_fire) = Some(now);
153 run_heartbeat_turn(state, &cfg).await
154}
155
156async fn run_heartbeat_turn(
157 state: &AppState,
158 cfg: &vak_config::HeartbeatResolved,
159) -> Result<(), String> {
160 let provider = state
161 .core
162 .provider()
163 .map_err(|e| format!("no provider credential: {e}"))?;
164 // Dedicated child core over the SERVER cwd: the model pin stays scoped
165 // here instead of mutating shared runtime overrides.
166 let core = vak_core::Core::new_with_trust(state.core.cwd().clone(), true)
167 // Unattended by construction: the turn's approver is `AutoDeny` and
168 // no one is reading the reply as it streams. Both facts are stamped
169 // before `take_persistent_session` composes and freezes the prompt,
170 // so it never advertises a capability this turn cannot use.
171 .map(|c| {
172 c.with_surface(vak_core::Surface::Background)
173 .with_approver_answerable(false)
174 })
175 .map_err(|e| format!("heartbeat core failed: {e}"))?;
176 core.set_provider_instance(provider);
177 core.set_sessions_home(state.core.shared_data_home());
178 if let Some(pin) = cfg
179 .model
180 .as_deref()
181 .map(str::trim)
182 .filter(|p| !p.is_empty())
183 {
184 let (pin_provider, pin_model) = crate::split_model_pin(pin, &core.effective_provider());
185 core.set_route(pin_provider, pin_model);
186 }
187
188 let ledger = take_persistent_session(&core).await?;
189 let prompt = format!(
190 "Heartbeat review pass. You are an unattended watchdog: use your \
191 tools (session_search, read, glob, grep) to review recent activity \
192 in this workspace — recent sessions, tasks, failing checks, stale \
193 work. Reply either exactly \"nothing\" (when nothing needs \
194 attention) or up to {max} short actionable findings, one per line. \
195 Prefix any finding that cannot wait until the next check-in with \
196 \"URGENT:\".",
197 max = cfg.max_findings
198 );
199
200 let approver: Arc<dyn vak_agent::Approver> = Arc::new(vak_agent::AutoDeny);
201 let cancel = CancellationToken::new();
202 let (events_tx, events_rx) = tokio::sync::mpsc::channel::<vak_agent::AgentEvent>(512);
203 let mut fut = std::pin::pin!(core.run_turn_with(
204 ledger,
205 &prompt,
206 cancel.clone(),
207 Some(approver),
208 None,
209 None,
210 events_tx,
211 ));
212 let outcome = match tokio::time::timeout(HEARTBEAT_TURN_TIMEOUT, fut.as_mut()).await {
213 Ok(result) => Some(result),
214 Err(_) => {
215 eprintln!("[heartbeat] turn exceeded its bound; cancelling");
216 cancel.cancel();
217 // Abort preserves partial output: give the loop room to land
218 // whatever it has before the cycle moves on.
219 tokio::time::timeout(CANCEL_GRACE, fut.as_mut()).await.ok()
220 }
221 };
222 drop(events_rx);
223
224 match outcome {
225 Some(Ok((outcome, log))) => {
226 let text = crate::projection::text_with_run_cards(&log, outcome_text(&outcome));
227 record_reply(state, cfg, &text).await;
228 Ok(())
229 }
230 Some(Err(e)) => Err(format!("heartbeat turn failed: {e}")),
231 None => Err("heartbeat turn did not finish within its bound".to_string()),
232 }
233}
234
235/// Open-or-create the fixed "heartbeat" ledger. The filesystem lock makes
236/// concurrent takers (another process, a wedged prior cycle) fail closed
237/// for this cycle instead of corrupting appends.
238async fn take_persistent_session(core: &Core) -> Result<SessionLog, String> {
239 match core.open_session(HEARTBEAT_SESSION_ID).await {
240 Ok(log) => Ok(log),
241 Err(_) => create_persistent_session(core).await,
242 }
243}
244
245async fn create_persistent_session(core: &Core) -> Result<SessionLog, String> {
246 let prepared = core.prepare_turn().await;
247 let path = vak_session::SessionPath::new_session_file(
248 &core.sessions_home(),
249 core.cwd(),
250 HEARTBEAT_SESSION_ID,
251 );
252 let header = SessionHeader {
253 agent: Some(vak_core::vak_agent_identity()),
254 session_id: HEARTBEAT_SESSION_ID.to_string(),
255 created_at: chrono::Utc::now(),
256 cwd: core.cwd().clone(),
257 parent_session_id: None,
258 contract_id: None,
259 work_item_id: None,
260 conversation: Some(vak_session::ConversationContext::local(
261 HEARTBEAT_SESSION_ID,
262 "background",
263 )),
264 contract: vak_session::FrozenContract {
265 app_version: vak_core::APP_VERSION.to_string(),
266 provider: core.effective_provider(),
267 model: core.effective_model(),
268 // Legacy single-model admission (empty ladder), same as
269 // worker children; dispatch falls back to the primary leg.
270 route_ladder: Vec::new(),
271 route_objective: String::new(),
272 route_annotations: Vec::new(),
273 system_prompt: prepared.system_prompt,
274 permission_mode: permission_mode_tag(core.effective_permission_mode()).to_string(),
275 capabilities: core.capability_descriptors(),
276 prompt_layers: Vec::new(),
277 },
278 };
279 SessionLog::create(path, header).map_err(|e| format!("heartbeat session create: {e}"))
280}
281
282fn permission_mode_tag(mode: vak_config::PermissionMode) -> &'static str {
283 match mode {
284 vak_config::PermissionMode::ReadOnly => "read-only",
285 vak_config::PermissionMode::WorkspaceWrite => "workspace-write",
286 vak_config::PermissionMode::FullAccess => "full-access",
287 }
288}
289
290fn outcome_text(o: &vak_agent::TurnOutcome) -> String {
291 use vak_agent::TurnOutcome;
292 match o {
293 TurnOutcome::Completed { response } => response.text_content(),
294 TurnOutcome::Aborted { partial } => partial
295 .as_ref()
296 .map(|m| m.text_content())
297 .unwrap_or_default(),
298 // Failures carry no review reply; classify treats them as
299 // nothing-to-report rather than inventing findings from an error.
300 TurnOutcome::Failed { .. } | TurnOutcome::MaxTurnsReached => String::new(),
301 }
302}
303
304/// Anti-nag routing: "nothing"/"" logs a debug line and records nothing
305/// anywhere; findings always park one inbox entry, and additionally ride
306/// the delivery chokepoint only when some line is marked URGENT.
307async fn record_reply(state: &AppState, cfg: &vak_config::HeartbeatResolved, text: &str) {
308 let reply = classify_reply(text, cfg.max_findings);
309 let HeartbeatReply::Findings {
310 stripped, urgent, ..
311 } = reply
312 else {
313 eprintln!("[heartbeat] nothing to report");
314 return;
315 };
316 let n = stripped.len();
317 let title = format!("heartbeat: {n} finding{}", if n == 1 { "" } else { "s" });
318 let body = stripped.join("\n");
319 let home = state.core.shared_data_home();
320 let _ = vak_core::inbox::record(
321 &home,
322 vak_core::inbox::Kind::Heartbeat,
323 &title,
324 &body,
325 Some(HEARTBEAT_SESSION_ID),
326 None,
327 );
328 if !urgent {
329 return;
330 }
331 let mut targets = crate::configured_delivery_targets(state);
332 if targets.is_empty() {
333 targets.push(crate::FALLBACK_ALERT_TARGET.to_string());
334 }
335 let text = format!("{title}\n{body}");
336 for target in targets {
337 let _ = crate::gateway::deliver_and_record(
338 &state.core,
339 &target,
340 &text,
341 vak_core::inbox::Kind::Heartbeat,
342 title.clone(),
343 Some(HEARTBEAT_SESSION_ID),
344 None,
345 )
346 .await;
347 }
348}
349
350#[cfg(test)]
351#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
352mod tests {
353 use super::{HeartbeatReply, classify_reply, heartbeat_due};
354 use chrono::{TimeZone, Utc};
355
356 #[test]
357 fn due_matrix_never_run_interval_boundaries_and_future() {
358 let now = Utc
359 .with_ymd_and_hms(2026, 8, 25, 12, 0, 0)
360 .single()
361 .unwrap();
362 assert!(heartbeat_due(None, now, 1800));
363 let ago_100 = now - chrono::Duration::seconds(100);
364 assert!(!heartbeat_due(Some(ago_100), now, 1800));
365 // Exactly one interval elapsed counts as due (>= semantics).
366 let ago_exact = now - chrono::Duration::seconds(1800);
367 assert!(heartbeat_due(Some(ago_exact), now, 1800));
368 let ago_more = now - chrono::Duration::seconds(1801);
369 assert!(heartbeat_due(Some(ago_more), now, 1800));
370 // A marker in the future (clock rewind) waits it out.
371 let ahead = now + chrono::Duration::seconds(60);
372 assert!(!heartbeat_due(Some(ahead), now, 1800));
373 }
374
375 #[test]
376 fn nothing_variants_classify_as_nothing() {
377 for s in ["", " ", "nothing", " Nothing\n", "NOTHING"] {
378 assert_eq!(
379 classify_reply(s, 3),
380 HeartbeatReply::Nothing,
381 "'{s}' must be silent"
382 );
383 }
384 }
385
386 #[test]
387 fn plain_findings_are_not_urgent_and_pass_through_verbatim() {
388 let reply = classify_reply("fix the flaky cron\nclose stale PR #4", 3);
389 match reply {
390 HeartbeatReply::Findings {
391 lines,
392 stripped,
393 urgent,
394 } => {
395 assert_eq!(lines, vec!["fix the flaky cron", "close stale PR #4"]);
396 assert_eq!(stripped, lines);
397 assert!(!urgent);
398 }
399 other => panic!("expected findings, got {other:?}"),
400 }
401 }
402
403 #[test]
404 fn urgent_marker_is_detected_and_stripped_from_body() {
405 let reply = classify_reply("URGENT: disk almost full\nminor lint debt", 3);
406 match reply {
407 HeartbeatReply::Findings {
408 lines,
409 stripped,
410 urgent,
411 } => {
412 assert_eq!(lines, vec!["URGENT: disk almost full", "minor lint debt"]);
413 assert_eq!(stripped, vec!["disk almost full", "minor lint debt"]);
414 assert!(urgent);
415 }
416 other => panic!("expected findings, got {other:?}"),
417 }
418 }
419
420 #[test]
421 fn findings_cap_at_max_findings_and_drop_blank_lines() {
422 let reply = classify_reply("a\n\n \nb\nc\nd\ne", 3);
423 match reply {
424 HeartbeatReply::Findings { lines, .. } => {
425 assert_eq!(lines, vec!["a", "b", "c"]);
426 }
427 other => panic!("expected findings, got {other:?}"),
428 }
429 }
430}
431

1//! Proactive heartbeat (docs/design/29-personal-os.md P7): one bounded LLM
2//! review turn per interval in a dedicated persistent session. Anti-nag by
3//! architecture — a "nothing" reply costs tokens but produces zero inbox
4//! noise and zero deliveries; findings park as `Kind::Heartbeat` inbox
5//! entries and reach chat surfaces only when a line carries an URGENT
6//! marker.
7
8use std::sync::Arc;
9use std::sync::atomic::{AtomicBool, Ordering};
10use std::time::Duration;
11
12use chrono::{DateTime, Timelike, Utc};
13use tokio_util::sync::CancellationToken;
14use vak_core::Core;
15use vak_session::{SessionHeader, SessionLog};
16
17use crate::AppState;
18
19/// Fixed ledger id of the dedicated heartbeat session.
20pub(crate) const HEARTBEAT_SESSION_ID: &str = "heartbeat";
21
22/// Upper bound on one heartbeat turn so a stuck stream cannot wedge the
23/// scheduler (mirrors REFLECTION_CALL_TIMEOUT).
24const HEARTBEAT_TURN_TIMEOUT: Duration = Duration::from_secs(120);
25
26/// Grace period for the agent loop to observe cancellation and hand the
27/// ledger back before the cycle gives up on it.
28const CANCEL_GRACE: Duration = Duration::from_secs(10);
29
30/// Cadence of the cheap due-check pass; the real gate is
31/// `[heartbeat].interval_secs` (minimum 300).
32pub(crate) const TICK_SECS: u64 = 5;
33
34pub(crate) struct HeartbeatRuntime {
35 /// Per-process fire timer; a restart re-fires immediately, matching
36 /// interval tasks' self-healing behavior.
37 last_fire: std::sync::Mutex<Option<DateTime<Utc>>>,
38 inflight: AtomicBool,
39}
40
41impl HeartbeatRuntime {
42 pub(crate) fn new() -> Self {
43 HeartbeatRuntime {
44 last_fire: std::sync::Mutex::new(None),
45 inflight: AtomicBool::new(false),
46 }
47 }
48}
49
50fn lock<T>(m: &std::sync::Mutex<T>) -> std::sync::MutexGuard<'_, T> {
51 m.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
52}
53
54/// Pure due-matrix: never-run fires immediately; otherwise only once a
55/// full interval has elapsed.
56fn heartbeat_due(last_fire: Option<DateTime<Utc>>, now: DateTime<Utc>, interval_secs: u64) -> bool {
57 match last_fire {
58 None => true,
59 Some(t) => (now - t).num_seconds() >= interval_secs as i64,
60 }
61}
62
63#[derive(Debug, PartialEq, Eq)]
64enum HeartbeatReply {
65 Nothing,
66 Findings {
67 /// Original trimmed finding lines (URGENT markers intact).
68 lines: Vec<String>,
69 /// Lines with any leading "URGENT:" marker removed.
70 stripped: Vec<String>,
71 urgent: bool,
72 },
73}
74
75fn classify_reply(text: &str, max_findings: usize) -> HeartbeatReply {
76 let trimmed = text.trim();
77 if trimmed.is_empty() || trimmed.eq_ignore_ascii_case("nothing") {
78 return HeartbeatReply::Nothing;
79 }
80 let lines: Vec<String> = trimmed
81 .lines()
82 .map(str::trim)
83 .filter(|l| !l.is_empty())
84 .take(max_findings.max(1))
85 .map(String::from)
86 .collect();
87 let urgent = lines.iter().any(|l| l.starts_with("URGENT:"));
88 let stripped = lines
89 .iter()
90 .map(|l| {
91 l.strip_prefix("URGENT:")
92 .map(str::trim_start)
93 .unwrap_or(l)
94 .to_string()
95 })
96 .collect();
97 HeartbeatReply::Findings {
98 lines,
99 stripped,
100 urgent,
101 }
102}
103
104/// One cheap scheduler pass. Errors are values: failures log and leave the
105/// next cycle to retry — this function never panics.
106pub(crate) async fn heartbeat_tick(state: &AppState) {
107 if !state.core.config().heartbeat.enabled {
108 return;
109 }
110 if state.heartbeat.inflight.swap(true, Ordering::SeqCst) {
111 return;
112 }
113 let result = heartbeat_cycle(state).await;
114 state.heartbeat.inflight.store(false, Ordering::SeqCst);
115 if let Err(e) = result {
116 eprintln!("[heartbeat] cycle failed: {e}");
117 }
118}
119
120async fn heartbeat_cycle(state: &AppState) -> Result<(), String> {
121 let cfg = state.core.config().heartbeat.clone();
122 let now = Utc::now();
123 let due = {
124 let last = lock(&state.heartbeat.last_fire);
125 heartbeat_due(*last, now, cfg.interval_secs)
126 };
127 if !due {
128 return Ok(());
129 }
130 if let Some(window) = cfg.quiet_hours {
131 let local = chrono::Local::now();
132 let minutes = local.hour() * 60 + local.minute();
133 if window.contains(minutes) {
134 return Ok(());
135 }
136 }
137 // Same day-spend read the budget alerts use; denial skips this cycle
138 // silently — an unattended prober must never be what bursts a cap.
139 let mut day_total = vak_core::finops::FinOpsLedger::new(&state.core.shared_data_home())
140 .day_total_usd(Utc::now());
141 if state.core.sessions_home() != state.core.shared_data_home() {
142 day_total += vak_core::finops::FinOpsLedger::new(&state.core.sessions_home())
143 .day_total_usd(Utc::now());
144 }
145 if let Some(cap) = state.core.config().finops.max_day_usd
146 && day_total >= cap
147 {
148 return Ok(());
149 }
150 // Claim the slot BEFORE dispatching so overlapping ticks cannot
151 // double-fire while a turn outlives one tick period.
152 *lock(&state.heartbeat.last_fire) = Some(now);
153 run_heartbeat_turn(state, &cfg).await
154}
155
156async fn run_heartbeat_turn(
157 state: &AppState,
158 cfg: &vak_config::HeartbeatResolved,
159) -> Result<(), String> {
160 let provider = state
161 .core
162 .provider()
163 .map_err(|e| format!("no provider credential: {e}"))?;
164 // Dedicated child core over the SERVER cwd: the model pin stays scoped
165 // here instead of mutating shared runtime overrides.
166 let core = vak_core::Core::new_with_trust(state.core.cwd().clone(), true)
167 // Unattended by construction: the turn's approver is `AutoDeny` and
168 // no one is reading the reply as it streams. Both facts are stamped
169 // before `take_persistent_session` composes and freezes the prompt,
170 // so it never advertises a capability this turn cannot use.
171 .map(|c| {
172 c.with_surface(vak_core::Surface::Background)
173 .with_approver_answerable(false)
174 })
175 .map_err(|e| format!("heartbeat core failed: {e}"))?;
176 core.set_provider_instance(provider);
177 core.set_sessions_home(state.core.shared_data_home());
178 if let Some(pin) = cfg
179 .model
180 .as_deref()
181 .map(str::trim)
182 .filter(|p| !p.is_empty())
183 {
184 let (pin_provider, pin_model) = crate::split_model_pin(pin, &core.effective_provider());
185 core.set_route(pin_provider, pin_model);
186 }
187
188 let ledger = take_persistent_session(&core).await?;
189 let prompt = format!(
190 "Heartbeat review pass. You are an unattended watchdog: use your \
191 tools (session_search, read, glob, grep) to review recent activity \
192 in this workspace — recent sessions, tasks, failing checks, stale \
193 work. Reply either exactly \"nothing\" (when nothing needs \
194 attention) or up to {max} short actionable findings, one per line. \
195 Prefix any finding that cannot wait until the next check-in with \
196 \"URGENT:\".",
197 max = cfg.max_findings
198 );
199
200 let approver: Arc<dyn vak_agent::Approver> = Arc::new(vak_agent::AutoDeny);
201 let cancel = CancellationToken::new();
202 let (events_tx, events_rx) = tokio::sync::mpsc::channel::<vak_agent::AgentEvent>(512);
203 let mut fut = std::pin::pin!(core.run_turn_with(
204 ledger,
205 &prompt,
206 cancel.clone(),
207 Some(approver),
208 None,
209 None,
210 events_tx,
211 ));
212 let outcome = match tokio::time::timeout(HEARTBEAT_TURN_TIMEOUT, fut.as_mut()).await {
213 Ok(result) => Some(result),
214 Err(_) => {
215 eprintln!("[heartbeat] turn exceeded its bound; cancelling");
216 cancel.cancel();
217 // Abort preserves partial output: give the loop room to land
218 // whatever it has before the cycle moves on.
219 tokio::time::timeout(CANCEL_GRACE, fut.as_mut()).await.ok()
220 }
221 };
222 drop(events_rx);
223
224 match outcome {
225 Some(Ok((outcome, log))) => {
226 let text = crate::projection::text_with_run_cards(&log, outcome_text(&outcome));
227 record_reply(state, cfg, &text).await;
228 Ok(())
229 }
230 Some(Err(e)) => Err(format!("heartbeat turn failed: {e}")),
231 None => Err("heartbeat turn did not finish within its bound".to_string()),
232 }
233}
234
235/// Open-or-create the fixed "heartbeat" ledger. The filesystem lock makes
236/// concurrent takers (another process, a wedged prior cycle) fail closed
237/// for this cycle instead of corrupting appends.
238async fn take_persistent_session(core: &Core) -> Result<SessionLog, String> {
239 match core.open_session(HEARTBEAT_SESSION_ID).await {
240 Ok(log) => Ok(log),
241 Err(_) => create_persistent_session(core).await,
242 }
243}
244
245async fn create_persistent_session(core: &Core) -> Result<SessionLog, String> {
246 let prepared = core.prepare_turn().await;
247 let path = vak_session::SessionPath::new_session_file(
248 &core.sessions_home(),
249 core.cwd(),
250 HEARTBEAT_SESSION_ID,
251 );
252 let header = SessionHeader {
253 agent: Some(vak_core::vak_agent_identity()),
254 session_id: HEARTBEAT_SESSION_ID.to_string(),
255 created_at: chrono::Utc::now(),
256 cwd: core.cwd().clone(),
257 parent_session_id: None,
258 contract_id: None,
259 work_item_id: None,
260 conversation: Some(vak_session::ConversationContext::local(
261 HEARTBEAT_SESSION_ID,
262 "background",
263 )),
264 contract: vak_session::FrozenContract {
265 app_version: vak_core::APP_VERSION.to_string(),
266 provider: core.effective_provider(),
267 model: core.effective_model(),
268 // Legacy single-model admission (empty ladder), same as
269 // worker children; dispatch falls back to the primary leg.
270 route_ladder: Vec::new(),
271 route_objective: String::new(),
272 route_annotations: Vec::new(),
273 system_prompt: prepared.system_prompt,
274 permission_mode: permission_mode_tag(core.effective_permission_mode()).to_string(),
275 capabilities: core.capability_descriptors(),
276 prompt_layers: Vec::new(),
277 },
278 };
279 SessionLog::create(path, header).map_err(|e| format!("heartbeat session create: {e}"))
280}
281
282fn permission_mode_tag(mode: vak_config::PermissionMode) -> &'static str {
283 match mode {
284 vak_config::PermissionMode::ReadOnly => "read-only",
285 vak_config::PermissionMode::WorkspaceWrite => "workspace-write",
286 vak_config::PermissionMode::FullAccess => "full-access",
287 }
288}
289
290fn outcome_text(o: &vak_agent::TurnOutcome) -> String {
291 use vak_agent::TurnOutcome;
292 match o {
293 TurnOutcome::Completed { response } => response.text_content(),
294 TurnOutcome::Aborted { partial } => partial
295 .as_ref()
296 .map(|m| m.text_content())
297 .unwrap_or_default(),
298 // Failures carry no review reply; classify treats them as
299 // nothing-to-report rather than inventing findings from an error.
300 TurnOutcome::Failed { .. } | TurnOutcome::MaxTurnsReached => String::new(),
301 }
302}
303
304/// Anti-nag routing: "nothing"/"" logs a debug line and records nothing
305/// anywhere; findings always park one inbox entry, and additionally ride
306/// the delivery chokepoint only when some line is marked URGENT.
307async fn record_reply(state: &AppState, cfg: &vak_config::HeartbeatResolved, text: &str) {
308 let reply = classify_reply(text, cfg.max_findings);
309 let HeartbeatReply::Findings {
310 stripped, urgent, ..
311 } = reply
312 else {
313 eprintln!("[heartbeat] nothing to report");
314 return;
315 };
316 let n = stripped.len();
317 let title = format!("heartbeat: {n} finding{}", if n == 1 { "" } else { "s" });
318 let body = stripped.join("\n");
319 let home = state.core.shared_data_home();
320 let _ = vak_core::inbox::record(
321 &home,
322 vak_core::inbox::Kind::Heartbeat,
323 &title,
324 &body,
325 Some(HEARTBEAT_SESSION_ID),
326 None,
327 );
328 if !urgent {
329 return;
330 }
331 let mut targets = crate::configured_delivery_targets(state);
332 if targets.is_empty() {
333 targets.push(crate::FALLBACK_ALERT_TARGET.to_string());
334 }
335 let text = format!("{title}\n{body}");
336 for target in targets {
337 let _ = crate::gateway::deliver_and_record(
338 &state.core,
339 &target,
340 &text,
341 vak_core::inbox::Kind::Heartbeat,
342 title.clone(),
343 Some(HEARTBEAT_SESSION_ID),
344 None,
345 )
346 .await;
347 }
348}
349
350#[cfg(test)]
351#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
352mod tests {
353 use super::{HeartbeatReply, classify_reply, heartbeat_due};
354 use chrono::{TimeZone, Utc};
355
356 #[test]
357 fn due_matrix_never_run_interval_boundaries_and_future() {
358 let now = Utc
359 .with_ymd_and_hms(2026, 8, 25, 12, 0, 0)
360 .single()
361 .unwrap();
362 assert!(heartbeat_due(None, now, 1800));
363 let ago_100 = now - chrono::Duration::seconds(100);
364 assert!(!heartbeat_due(Some(ago_100), now, 1800));
365 // Exactly one interval elapsed counts as due (>= semantics).
366 let ago_exact = now - chrono::Duration::seconds(1800);
367 assert!(heartbeat_due(Some(ago_exact), now, 1800));
368 let ago_more = now - chrono::Duration::seconds(1801);
369 assert!(heartbeat_due(Some(ago_more), now, 1800));
370 // A marker in the future (clock rewind) waits it out.
371 let ahead = now + chrono::Duration::seconds(60);
372 assert!(!heartbeat_due(Some(ahead), now, 1800));
373 }
374
375 #[test]
376 fn nothing_variants_classify_as_nothing() {
377 for s in ["", " ", "nothing", " Nothing\n", "NOTHING"] {
378 assert_eq!(
379 classify_reply(s, 3),
380 HeartbeatReply::Nothing,
381 "'{s}' must be silent"
382 );
383 }
384 }
385
386 #[test]
387 fn plain_findings_are_not_urgent_and_pass_through_verbatim() {
388 let reply = classify_reply("fix the flaky cron\nclose stale PR #4", 3);
389 match reply {
390 HeartbeatReply::Findings {
391 lines,
392 stripped,
393 urgent,
394 } => {
395 assert_eq!(lines, vec!["fix the flaky cron", "close stale PR #4"]);
396 assert_eq!(stripped, lines);
397 assert!(!urgent);
398 }
399 other => panic!("expected findings, got {other:?}"),
400 }
401 }
402
403 #[test]
404 fn urgent_marker_is_detected_and_stripped_from_body() {
405 let reply = classify_reply("URGENT: disk almost full\nminor lint debt", 3);
406 match reply {
407 HeartbeatReply::Findings {
408 lines,
409 stripped,
410 urgent,
411 } => {
412 assert_eq!(lines, vec!["URGENT: disk almost full", "minor lint debt"]);
413 assert_eq!(stripped, vec!["disk almost full", "minor lint debt"]);
414 assert!(urgent);
415 }
416 other => panic!("expected findings, got {other:?}"),
417 }
418 }
419
420 #[test]
421 fn findings_cap_at_max_findings_and_drop_blank_lines() {
422 let reply = classify_reply("a\n\n \nb\nc\nd\ne", 3);
423 match reply {
424 HeartbeatReply::Findings { lines, .. } => {
425 assert_eq!(lines, vec!["a", "b", "c"]);
426 }
427 other => panic!("expected findings, got {other:?}"),
428 }
429 }
430}
431

Indexing the workspace…