1use super::*;
3use crate::{
4 agent_host::{AgentHostBuilder, AgentWorkerOptions},
5 agent_model::ModelProviderCatalog,
6 agent_resource::ResourceDirectory,
7 agent_task_store::{AgentTaskStore, AgentTaskSubmissionResult},
8};
9use serde_json::Value;
10
11pub async fn run(
12 config: Config,
13 directories: &[ResourceDirectory],
14 tool_directories: &[ResourceDirectory],
15 name: &str,
16 input: &str,
17 json_output: bool,
18 non_interactive: bool,
19) -> Result<()> {
20 let root = std::env::current_dir()?;
21 let (catalog, authority) = cli_tool_bindings(&config, &root)?;
22 let mut builder = AgentHostBuilder::new(config, &root, ModelProviderCatalog::with_builtins()?)?
23 .with_agent_resources(directories.to_vec())
24 .with_tool_bindings(tool_directories.to_vec(), catalog, authority);
25 let approvals: Arc<dyn ApprovalHandler> = if non_interactive {
26 Arc::new(DenyApprovals)
27 } else {
28 Arc::new(super::terminal::TerminalApprovals)
29 };
30 builder = builder.with_policy(RuntimePolicy::default(), approvals);
31 let host = builder.build(name).await?;
32 let (sender, receiver) = watch::channel(false);
33 let signal = tokio::spawn(async move {
34 let _ = tokio::signal::ctrl_c().await;
35 let _ = sender.send(true);
36 });
37 let result = host.run(input, receiver).await;
38 signal.abort();
39 print_run(result?, json_output)
40}
41
42fn print_run(run: AgentRun, json_output: bool) -> Result<()> {
43 if json_output {
44 println!("{}", serde_json::to_string_pretty(&run)?);
45 } else {
46 println!(
47 "Run: {}\nStatus: {}\nLifecycle: {:?}\nOutcome: {:?}\nReason: {}\nElapsed: {}ms\nSuspended: {}ms\nExecution elapsed: {}ms\nBudgets: {}\nUsage: {}",
48 run.id,
49 run.status,
50 run.lifecycle,
51 run.outcome,
52 run.reason_code.as_deref().unwrap_or("-"),
53 run.elapsed_ms,
54 run.suspended_ms,
55 run.execution_elapsed_ms,
56 serde_json::to_string(&run.budgets.0)?,
57 serde_json::to_string(&run.usage.0)?,
58 );
59 if let Some(output) = &run.output {
60 println!("\n{output}");
61 }
62 if let Some(error) = &run.error {
63 println!("Error: {error}");
64 }
65 }
66 ensure!(
67 run.outcome == Some(crate::agent_store::RunOutcomeKind::Completed),
68 "agent run {} ended with status {}",
69 run.id,
70 run.status
71 );
72 Ok(())
73}
74
75pub async fn resume(
76 config: Config,
77 directories: &[ResourceDirectory],
78 tool_directories: &[ResourceDirectory],
79 id: &str,
80 json_output: bool,
81 non_interactive: bool,
82) -> Result<()> {
83 let store = read_store(config.clone())
84 .await?
85 .context("no runtime history in this workspace")?;
86 let previous = store
87 .get_run(id)
88 .await?
89 .context("run not found in workspace")?;
90 let root = std::env::current_dir()?;
91 let (catalog, authority) = cli_tool_bindings(&config, &root)?;
92 let mut builder = AgentHostBuilder::new(config, &root, ModelProviderCatalog::with_builtins()?)?
93 .with_agent_resources(directories.to_vec())
94 .with_tool_bindings(tool_directories.to_vec(), catalog, authority);
95 let approvals: Arc<dyn ApprovalHandler> = if non_interactive {
96 Arc::new(DenyApprovals)
97 } else {
98 Arc::new(super::terminal::TerminalApprovals)
99 };
100 builder = builder.with_policy(RuntimePolicy::default(), approvals);
101 let host = builder.build(&previous.agent_name).await?;
102 let (sender, receiver) = watch::channel(false);
103 let signal = tokio::spawn(async move {
104 let _ = tokio::signal::ctrl_c().await;
105 let _ = sender.send(true);
106 });
107 let result = host.resume(id, receiver).await;
108 signal.abort();
109 print_run(result?, json_output)
110}
111
112pub async fn continue_run(
113 config: Config,
114 directories: &[ResourceDirectory],
115 tool_directories: &[ResourceDirectory],
116 id: &str,
117 input: &str,
118 json_output: bool,
119 non_interactive: bool,
120) -> Result<()> {
121 let store = read_store(config.clone())
122 .await?
123 .context("no runtime history in this workspace")?;
124 let previous = store
125 .get_run(id)
126 .await?
127 .context("run not found in workspace")?;
128 let root = std::env::current_dir()?;
129 let (catalog, authority) = cli_tool_bindings(&config, &root)?;
130 let mut builder = AgentHostBuilder::new(config, &root, ModelProviderCatalog::with_builtins()?)?
131 .with_agent_resources(directories.to_vec())
132 .with_tool_bindings(tool_directories.to_vec(), catalog, authority);
133 let approvals: Arc<dyn ApprovalHandler> = if non_interactive {
134 Arc::new(DenyApprovals)
135 } else {
136 Arc::new(super::terminal::TerminalApprovals)
137 };
138 builder = builder.with_policy(RuntimePolicy::default(), approvals);
139 let host = builder.build(&previous.agent_name).await?;
140 let (sender, receiver) = watch::channel(false);
141 let signal = tokio::spawn(async move {
142 let _ = tokio::signal::ctrl_c().await;
143 let _ = sender.send(true);
144 });
145 let result = host.continue_run(id, input, receiver).await;
146 signal.abort();
147 print_run(result?, json_output)
148}
149
150fn cli_tool_bindings(
151 config: &Config,
152 root: &Path,
153) -> Result<(
154 tool_binding::ToolImplementationCatalog,
155 Arc<HostToolAuthority>,
156)> {
157 let mut authority = HostToolAuthority::new(root, vec![Capability::ReadOnly])?;
158 authority.enroll_path(root)?;
159 let sources = config
160 .connectors
161 .filesystem
162 .keys()
163 .map(|name| format!("filesystem:{name}"))
164 .chain(
165 config
166 .connectors
167 .git
168 .keys()
169 .map(|name| format!("git:{name}")),
170 )
171 .chain(config.connectors.s3.keys().map(|name| format!("s3:{name}")))
172 .chain(
173 config
174 .connectors
175 .script
176 .keys()
177 .map(|name| format!("script:{name}")),
178 );
179 for source in sources {
180 authority.enroll_source(source)?;
181 }
182 Ok((tool_binding::core_catalog()?, Arc::new(authority)))
183}
184
185fn cli_host_builder(
186 config: Config,
187 root: &Path,
188 directories: &[ResourceDirectory],
189 tool_directories: &[ResourceDirectory],
190 allow_workspace_write: bool,
191) -> Result<AgentHostBuilder> {
192 let (catalog, authority) = cli_tool_bindings(&config, root)?;
193 let mut policy = RuntimePolicy::default();
194 if allow_workspace_write {
195 policy.allow.push(Capability::WorkspaceWrite);
196 policy
197 .require_approval
198 .retain(|capability| *capability != Capability::WorkspaceWrite);
199 }
200 Ok(
201 AgentHostBuilder::new(config, root, ModelProviderCatalog::with_builtins()?)?
202 .with_agent_resources(directories.to_vec())
203 .with_tool_bindings(tool_directories.to_vec(), catalog, authority)
204 .with_policy(policy, Arc::new(super::terminal::TerminalApprovals)),
205 )
206}
207
208#[allow(clippy::too_many_arguments)]
209pub async fn enqueue(
210 config: Config,
211 directories: &[ResourceDirectory],
212 tool_directories: &[ResourceDirectory],
213 name: &str,
214 input: &str,
215 request_key: &str,
216 queue_limit: u32,
217 json_output: bool,
218 allow_workspace_write: bool,
219) -> Result<()> {
220 let root = std::env::current_dir()?;
221 let submitter = cli_host_builder(
222 config,
223 &root,
224 directories,
225 tool_directories,
226 allow_workspace_write,
227 )?
228 .build_task_submitter(name)
229 .await?;
230 let result = submitter
231 .submit_with_queue_limit(request_key, input, Vec::new(), queue_limit)
232 .await?;
233 let value = sanitized_json(&result)?;
234 if json_output {
235 println!("{}", serde_json::to_string_pretty(&value)?);
236 } else {
237 match &result {
238 AgentTaskSubmissionResult::Created(task) => {
239 println!(
240 "Job: {}\nDisposition: created\nStatus: {:?}",
241 task.id, task.status
242 );
243 }
244 AgentTaskSubmissionResult::ExistingIdentical(task) => {
245 println!(
246 "Job: {}\nDisposition: existing_identical\nStatus: {:?}",
247 task.id, task.status
248 );
249 }
250 AgentTaskSubmissionResult::RequestConflict => println!("Disposition: request_conflict"),
251 AgentTaskSubmissionResult::QueueFull => println!("Disposition: queue_full"),
252 }
253 }
254 match result {
255 AgentTaskSubmissionResult::Created(_) | AgentTaskSubmissionResult::ExistingIdentical(_) => {
256 Ok(())
257 }
258 AgentTaskSubmissionResult::RequestConflict => {
259 anyhow::bail!("agent job request key conflicts with an existing submission")
260 }
261 AgentTaskSubmissionResult::QueueFull => anyhow::bail!("agent job queue is full"),
262 }
263}
264
265#[allow(clippy::too_many_arguments)]
266pub async fn worker(
267 config: Config,
268 directories: &[ResourceDirectory],
269 tool_directories: &[ResourceDirectory],
270 worker_id: Option<String>,
271 max_concurrency: u32,
272 lease_seconds: u64,
273 heartbeat_seconds: u64,
274 poll_ms: u64,
275 max_attempts: u32,
276 shutdown_seconds: u64,
277 allow_workspace_write: bool,
278) -> Result<()> {
279 let root = std::env::current_dir()?;
280 let options = AgentWorkerOptions {
281 worker_id,
282 concurrency: max_concurrency as usize,
283 lease_duration: Duration::from_secs(lease_seconds),
284 heartbeat_interval: Duration::from_secs(heartbeat_seconds),
285 polling_interval: Duration::from_millis(poll_ms),
286 max_claim_attempts: max_attempts,
287 graceful_shutdown_timeout: Duration::from_secs(shutdown_seconds),
288 };
289 let worker = cli_host_builder(
290 config,
291 &root,
292 directories,
293 tool_directories,
294 allow_workspace_write,
295 )?
296 .build_worker(options)
297 .await?;
298 let (sender, receiver) = watch::channel(false);
299 let signal = tokio::spawn(async move {
300 if tokio::signal::ctrl_c().await.is_ok() {
301 let _ = sender.send(true);
302 }
303 });
304 let result = worker.run(receiver).await;
305 signal.abort();
306 result
307}
308
309fn sanitized_json<T: serde::Serialize>(value: &T) -> Result<Value> {
310 fn remove_claim_tokens(value: &mut Value) {
311 match value {
312 Value::Object(object) => {
313 object.remove("claim_token");
314 for value in object.values_mut() {
315 remove_claim_tokens(value);
316 }
317 }
318 Value::Array(values) => values.iter_mut().for_each(remove_claim_tokens),
319 _ => {}
320 }
321 }
322 let mut value = serde_json::to_value(value)?;
323 remove_claim_tokens(&mut value);
324 Ok(value)
325}
326
327async fn read_task_store(config: Config) -> Result<Option<AgentTaskStore>> {
328 task_store(config, true).await
329}
330
331async fn write_task_store(config: Config) -> Result<Option<AgentTaskStore>> {
332 task_store(config, false).await
333}
334
335async fn task_store(mut config: Config, read_only: bool) -> Result<Option<AgentTaskStore>> {
336 let root = std::env::current_dir()?.canonicalize()?;
337 if config.db.path.is_relative() {
338 config.db.path = root.join(&config.db.path);
339 }
340 if !config.db.path.try_exists()? {
341 return Ok(None);
342 }
343 let pool = if read_only {
344 tools::read_pool(&config).await?
345 } else {
346 crate::db::connect(&config).await?
347 };
348 let exists: bool = sqlx::query_scalar(
349 "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'agent_tasks')",
350 )
351 .fetch_one(&pool)
352 .await?;
353 if !exists {
354 return Ok(None);
355 }
356 Ok(Some(AgentTaskStore::new(pool, &workspace_id(&root)?)?))
357}
358
359pub async fn jobs(config: Config, limit: u32, json_output: bool) -> Result<()> {
360 let inspections = match read_task_store(config).await? {
361 Some(store) => store.list_inspections(limit).await?,
362 None => Vec::new(),
363 };
364 if json_output {
365 println!(
366 "{}",
367 serde_json::to_string_pretty(&sanitized_json(&inspections)?)?
368 );
369 } else {
370 println!("{:<36} {:<18} {:<24} AGENT", "JOB", "STATUS", "REASON");
371 for inspection in inspections {
372 println!(
373 "{:<36} {:<18?} {:<24} {}",
374 inspection.task.id,
375 inspection.task.status,
376 inspection.task.scheduling_reason.as_deref().unwrap_or("-"),
377 inspection.task.agent_name
378 );
379 }
380 }
381 Ok(())
382}
383
384pub async fn job_inspect(
385 config: Config,
386 id: &str,
387 after_sequence: i64,
388 limit: u32,
389 json_output: bool,
390) -> Result<()> {
391 let store = read_task_store(config)
392 .await?
393 .context("no durable agent jobs in this workspace")?;
394 let inspection = store
395 .inspect(id)
396 .await?
397 .context("agent job not found in this workspace")?;
398 let events = store.events(id, after_sequence, limit).await?;
399 let cursor = (events.len() == limit as usize)
400 .then(|| events.last().map(|event| event.sequence))
401 .flatten();
402 if json_output {
403 let value = json!({
404 "task": inspection.task,
405 "run": inspection.run,
406 "events": events,
407 "next_after_sequence": cursor,
408 });
409 println!(
410 "{}",
411 serde_json::to_string_pretty(&sanitized_json(&value)?)?
412 );
413 } else {
414 let task = inspection.task;
415 println!(
416 "Job: {}\nAgent: {}\nStatus: {:?}\nReason: {}\nRun: {}",
417 task.id,
418 task.agent_name,
419 task.status,
420 task.scheduling_reason.as_deref().unwrap_or("-"),
421 task.run_id.as_deref().unwrap_or("-")
422 );
423 for event in events {
424 println!(
425 "{:>5} {} {}",
426 event.sequence, event.event_type, event.payload.0
427 );
428 }
429 if let Some(cursor) = cursor {
430 println!("Next page: --after-sequence {cursor}");
431 }
432 }
433 Ok(())
434}
435
436pub async fn job_cancel(config: Config, id: &str, json_output: bool) -> Result<()> {
437 let store = write_task_store(config)
438 .await?
439 .context("no durable agent jobs in this workspace")?;
440 let task = store.cancel(id).await?;
441 if json_output {
442 println!("{}", serde_json::to_string_pretty(&sanitized_json(&task)?)?);
443 } else {
444 println!(
445 "Job: {}\nStatus: {:?}\nReason: {}",
446 task.id,
447 task.status,
448 task.scheduling_reason.as_deref().unwrap_or("-")
449 );
450 }
451 Ok(())
452}
453
454async fn read_store(mut config: Config) -> Result<Option<AgentRunStore>> {
455 let root = std::env::current_dir()?.canonicalize()?;
456 if config.db.path.is_relative() {
457 config.db.path = root.join(&config.db.path);
458 }
459 if !config.db.path.try_exists()? {
460 return Ok(None);
461 }
462 let pool = tools::read_pool(&config).await?;
463 let exists: bool = sqlx::query_scalar(
464 "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'agent_runs')",
465 )
466 .fetch_one(&pool)
467 .await?;
468 if !exists {
469 return Ok(None);
470 }
471 Ok(Some(AgentRunStore::new(pool, &workspace_id(&root)?)?))
472}
473
474pub async fn history(config: Config, limit: u32, json_output: bool) -> Result<()> {
475 let runs = match read_store(config).await? {
476 Some(store) => store.history(limit).await?,
477 None => vec![],
478 };
479 if json_output {
480 println!("{}", serde_json::to_string_pretty(&runs)?);
481 } else {
482 println!("{:<36} {:<12} {:<24} AGENT", "RUN", "STATUS", "RECOVERY");
483 for run in runs {
484 println!(
485 "{:<36} {:<12} {:<24} {}",
486 run.id,
487 run.status,
488 format!("{:?}", run.recovery_disposition),
489 run.agent_name
490 );
491 }
492 }
493 Ok(())
494}
495
496pub async fn inspect(
497 config: Config,
498 id: &str,
499 after_sequence: i64,
500 limit: u32,
501 json_output: bool,
502) -> Result<()> {
503 let store = read_store(config)
504 .await?
505 .context("no runtime history in this workspace")?;
506 let run = store
507 .get_run(id)
508 .await?
509 .context("run not found in this workspace")?;
510 let events = store.events(id, after_sequence, limit).await?;
511 let artifacts = store.artifacts(id).await?;
512 let lineage = store.lineage(id).await?;
513 let children = store.children(id).await?;
514 let working_state = store.working_state(id).await?;
515 let invocations: Vec<Value> = store
516 .tool_invocations(id)
517 .await?
518 .into_iter()
519 .map(|call| {
520 json!({
521 "call_id":call.call_id, "tool":call.tool_name, "status":call.status,
522 "error":call.error, "started_at":call.started_at, "completed_at":call.completed_at
523 })
524 })
525 .collect();
526 let cursor = if events.len() == limit as usize {
527 events.last().map(|e| e.sequence)
528 } else {
529 None
530 };
531 if json_output {
532 println!(
533 "{}",
534 serde_json::to_string_pretty(
535 &json!({"run":run, "working_state":working_state, "events":events, "tool_invocations":invocations, "next_after_sequence":cursor, "artifacts":artifacts, "lineage":lineage, "children":children})
536 )?
537 );
538 } else {
539 println!(
540 "Run: {}\nAgent: {}\nModel: {}\nWorkspace: {}\nStatus: {}\nLifecycle: {:?}\nOutcome: {:?}\nRecovery: {:?}\nReason: {}\nElapsed: {}ms\nSuspended: {}ms\nExecution elapsed: {}ms\nBudgets: {}\nUsage: {}",
541 run.id, run.agent_name, run.model, run.workspace_id, run.status,
542 run.lifecycle, run.outcome, run.recovery_disposition, run.reason_code.as_deref().unwrap_or("-"),
543 run.elapsed_ms, run.suspended_ms, run.execution_elapsed_ms, serde_json::to_string(&run.budgets.0)?,
544 serde_json::to_string(&run.usage.0)?
545 );
546 println!(
547 "Root: {}\nParent: {:?}\nDepth: {}",
548 lineage.root_run_id, lineage.parent_run_id, lineage.depth
549 );
550 println!(
551 "Working state: {:?}\nProjection: {}\nRevision: {}\nEvent cursor: {}",
552 working_state.status,
553 working_state
554 .projection_version
555 .map_or("-".into(), |v| v.to_string()),
556 working_state.revision.map_or("-".into(), |v| v.to_string()),
557 working_state
558 .event_cursor
559 .map_or("-".into(), |v| v.to_string()),
560 );
561 if let Some(snapshot) = &working_state.snapshot {
562 println!(
563 "Latest projected event: {} {}\nArtifacts: {} retained, {} total, {} omitted",
564 snapshot.latest_event.sequence,
565 snapshot.latest_event.event_type,
566 snapshot.artifacts.items.len(),
567 snapshot.artifacts.total,
568 snapshot.artifacts.omitted,
569 );
570 for artifact in &snapshot.artifacts.items {
571 println!(
572 "Artifact reference: {} {} {} {} bytes",
573 artifact.sequence, artifact.relative_path, artifact.sha256, artifact.size
574 );
575 }
576 }
577 for child in children {
578 println!("Child: {}", child.run_id);
579 }
580 for event in events {
581 println!(
582 "{:>5} {:>8}ms {} {}",
583 event.sequence,
584 event.timestamp - run.created_at,
585 event.event_type,
586 event.payload.0
587 );
588 }
589 if let Some(cursor) = cursor {
590 println!("Next page: --after-sequence {cursor}");
591 }
592 for artifact in artifacts {
593 println!(
594 "Artifact: {} ({} bytes; sha256 {})",
595 artifact.relative_path, artifact.size, artifact.sha256
596 );
597 }
598 if let Some(output) = run.output {
599 println!("\nOutput:\n{output}");
600 }
601 if let Some(error) = run.error {
602 println!("\nError: {error}");
603 }
604 }
605 Ok(())
606}