vtcode_core/scheduler/
daemon.rs

1use anyhow::{Context, Result};
2use chrono::{DateTime, Utc};
3use std::path::PathBuf;
4use std::process::Stdio;
5use std::time::Duration;
6use tokio::process::Command;
7
8use super::advance_record_runtime;
9use super::session::resolve_scheduled_workspace_path;
10use super::spec::{ScheduledTaskAction, ScheduledTaskDefinition, ScheduledTaskRecord, TaskRunStatus};
11use super::store::{DurableTaskStore, release_claim_async, try_acquire_claim_async};
12use crate::notifications::{NotificationEvent, send_global_notification};
13use vtcode_commons::VtCodePaths;
14
15#[derive(Debug, Clone)]
16pub struct SchedulerDaemon {
17    store: DurableTaskStore,
18    executable_path: PathBuf,
19}
20
21impl SchedulerDaemon {
22    #[must_use]
23    pub fn new(store: DurableTaskStore, executable_path: PathBuf) -> Self {
24        Self { store, executable_path }
25    }
26
27    pub async fn serve_forever(&self) -> Result<()> {
28        loop {
29            self.run_due_tasks_once().await?;
30            tokio::time::sleep(Duration::from_secs(1)).await;
31        }
32    }
33
34    pub async fn run_due_tasks_once(&self) -> Result<usize> {
35        let now = Utc::now();
36        let mut records = self.store.load_records_async().await?;
37        let mut executed = 0usize;
38
39        records.sort_by_key(|record| record.runtime.next_run_at);
40        for mut record in records {
41            let Some(next_run_at) = record.runtime.next_run_at else {
42                continue;
43            };
44            if now < next_run_at {
45                continue;
46            }
47            if !try_acquire_claim_async(self.store.paths().clone(), record.definition.id.clone()).await? {
48                continue;
49            }
50
51            let result = self.execute_record(&record, now).await;
52            let release_result = release_claim_async(self.store.paths().clone(), record.definition.id.clone()).await;
53            let run_outcome = result?;
54            release_result?;
55
56            apply_run_outcome(&mut record, run_outcome)?;
57            self.store.update_runtime_async(record).await?;
58            executed = executed.saturating_add(1);
59        }
60
61        Ok(executed)
62    }
63
64    async fn execute_record(&self, record: &ScheduledTaskRecord, run_at: DateTime<Utc>) -> Result<RunOutcome> {
65        match &record.definition.action {
66            ScheduledTaskAction::Reminder { message } => {
67                let notification = send_global_notification(NotificationEvent::IdlePrompt {
68                    title: format!("Scheduled reminder: {}", record.definition.name),
69                    message: message.clone(),
70                })
71                .await;
72                let status = match notification {
73                    Ok(()) => TaskRunStatus::ReminderSent,
74                    Err(error) => TaskRunStatus::Failed {
75                        message: format!("failed to send reminder notification: {error:#}"),
76                    },
77                };
78                Ok(RunOutcome {
79                    ran_at: run_at,
80                    status,
81                    artifact_dir: None,
82                    events_file: None,
83                    last_message_file: None,
84                })
85            }
86            ScheduledTaskAction::Prompt { prompt } => {
87                let artifact_dir = self.store.paths().artifact_dir(&record.definition.id, run_at);
88                let events_file = artifact_dir.join("events.jsonl");
89                let last_message_file = artifact_dir.join("last-message.txt");
90                let workspace_label = scheduled_workspace_label(&record.definition);
91                let workspace = resolve_scheduled_task_workspace(&record.definition);
92                let execution = async {
93                    VtCodePaths::ensure_user_dir(&artifact_dir)
94                        .with_context(|| format!("Failed to create run artifact dir {}", artifact_dir.display()))?;
95
96                    let mut command = Command::new(&self.executable_path);
97                    command
98                        .arg("exec")
99                        .arg("--events")
100                        .arg(&events_file)
101                        .arg("--last-message-file")
102                        .arg(&last_message_file)
103                        .arg(prompt)
104                        .stdin(Stdio::null())
105                        .stdout(Stdio::null())
106                        .stderr(Stdio::null());
107
108                    if let Some(workspace) = workspace? {
109                        command.current_dir(&workspace);
110                    }
111
112                    command.status().await.with_context(|| {
113                        format!(
114                            "Failed to spawn scheduled VT Code exec for task {} using {} in {}",
115                            record.definition.id,
116                            self.executable_path.display(),
117                            workspace_label
118                        )
119                    })
120                }
121                .await;
122                let run_status = match execution {
123                    Ok(status) if status.success() => TaskRunStatus::Success,
124                    Ok(status) => TaskRunStatus::Failed {
125                        message: format!("vtcode exec exited with {status}"),
126                    },
127                    Err(error) => TaskRunStatus::Failed { message: format!("{error:#}") },
128                };
129                Ok(RunOutcome {
130                    ran_at: run_at,
131                    status: run_status,
132                    artifact_dir: artifact_dir.is_dir().then_some(artifact_dir),
133                    events_file: events_file.is_file().then_some(events_file),
134                    last_message_file: last_message_file.is_file().then_some(last_message_file),
135                })
136            }
137        }
138    }
139}
140
141fn apply_run_outcome(record: &mut ScheduledTaskRecord, outcome: RunOutcome) -> Result<()> {
142    let RunOutcome {
143        ran_at,
144        status,
145        artifact_dir,
146        events_file,
147        last_message_file,
148    } = outcome;
149
150    record.runtime.last_artifact_dir = artifact_dir;
151    record.runtime.last_events_file = events_file;
152    record.runtime.last_message_file = last_message_file;
153
154    advance_record_runtime(record, ran_at, status)?;
155
156    Ok(())
157}
158
159#[derive(Debug, Clone)]
160struct RunOutcome {
161    ran_at: DateTime<Utc>,
162    status: TaskRunStatus,
163    artifact_dir: Option<PathBuf>,
164    events_file: Option<PathBuf>,
165    last_message_file: Option<PathBuf>,
166}
167
168fn scheduled_workspace_label(definition: &ScheduledTaskDefinition) -> String {
169    definition
170        .workspace
171        .as_ref()
172        .map(|path| path.display().to_string())
173        .unwrap_or_else(|| "<none>".to_string())
174}
175
176fn resolve_scheduled_task_workspace(definition: &ScheduledTaskDefinition) -> Result<Option<PathBuf>> {
177    definition
178        .workspace
179        .as_deref()
180        .map(resolve_scheduled_workspace_path)
181        .transpose()
182        .with_context(|| {
183            format!(
184                "Failed to resolve scheduled task workspace {} for task {}",
185                scheduled_workspace_label(definition),
186                definition.id
187            )
188        })
189}
vtcode_core/scheduler/daemon.rs

vtcode_core/scheduler/
daemon.rs
