1use crate::agent::Agent;
32use crate::attachment::{self, Attachment};
33use crate::config::Config;
34use crate::factory::AgentFactory;
35use crate::json_validation;
36use crate::listen::{self, ListenFormat};
37use crate::output::AgentOutput;
38use crate::process_registration::{self, ProcessRegistration, RegisterOptionsOwned};
39use crate::progress::{ProgressHandler, SilentProgress};
40use crate::providers::claude::Claude;
41use crate::providers::ollama::Ollama;
42use crate::sandbox::SandboxConfig;
43use crate::session::{SessionEntry, SessionStore};
44use crate::session_log::{
45 AgentLogEvent, LiveLogContext, LogEventCallback, SessionLogCoordinator, SessionLogMetadata,
46 live_adapter_for_provider, logs_dir,
47};
48use crate::streaming::StreamingSession;
49use crate::worktree;
50use anyhow::{Result, bail};
51use log::{debug, warn};
52use std::sync::Arc;
53use std::time::Duration;
54
55fn format_duration(d: Duration) -> String {
57 let total_secs = d.as_secs();
58 let h = total_secs / 3600;
59 let m = (total_secs % 3600) / 60;
60 let s = total_secs % 60;
61 let mut parts = Vec::new();
62 if h > 0 {
63 parts.push(format!("{h}h"));
64 }
65 if m > 0 {
66 parts.push(format!("{m}m"));
67 }
68 if s > 0 || parts.is_empty() {
69 parts.push(format!("{s}s"));
70 }
71 parts.join("")
72}
73
74#[derive(Debug, Clone, Default)]
79pub struct SessionMetadata {
80 pub name: Option<String>,
81 pub description: Option<String>,
82 pub tags: Vec<String>,
83}
84
85struct SessionLogGuard {
90 coordinator: Option<SessionLogCoordinator>,
93 wrapper_session_id: String,
94 log_path: Option<std::path::PathBuf>,
95 external_writer: Option<crate::session_log::SessionLogWriter>,
98 _owned_external: Option<SessionLogCoordinator>,
102}
103
104impl SessionLogGuard {
105 fn log_path_string(&self) -> Option<String> {
106 self.log_path
107 .as_ref()
108 .map(|p| p.to_string_lossy().to_string())
109 }
110
111 async fn finish(mut self, success: bool, error: Option<String>) {
116 if let Some(coord) = self.coordinator.take() {
119 if let Err(e) = coord.finish(success, error).await {
120 warn!("Failed to finalize session log: {e}");
121 }
122 }
123 if let Some(w) = self.external_writer.take() {
124 let _ = w.clear_event_callback();
125 }
126 }
127}
128
129impl Drop for SessionLogGuard {
130 fn drop(&mut self) {
131 if let Some(ref w) = self.external_writer {
136 let _ = w.clear_event_callback();
137 }
138 if let Some(ref c) = self.coordinator {
139 let _ = c.writer().clear_event_callback();
140 }
141 }
142}
143
144#[derive(Default)]
156pub enum SessionLogMode {
157 #[default]
160 Disabled,
161 Auto,
165 External(SessionLogCoordinator),
168}
169
170pub struct AgentBuilder {
175 provider: Option<String>,
176 provider_explicit: bool,
180 model: Option<String>,
181 system_prompt: Option<String>,
182 root: Option<String>,
183 auto_approve: bool,
184 add_dirs: Vec<String>,
185 files: Vec<String>,
186 env_vars: Vec<(String, String)>,
187 worktree: Option<Option<String>>,
188 sandbox: Option<Option<String>>,
189 size: Option<String>,
190 json_mode: bool,
191 json_schema: Option<serde_json::Value>,
192 session_id: Option<String>,
193 metadata: SessionMetadata,
194 output_format: Option<String>,
195 input_format: Option<String>,
196 replay_user_messages: bool,
197 include_partial_messages: bool,
198 verbose: bool,
199 quiet: bool,
200 show_usage: bool,
201 max_turns: Option<u32>,
202 timeout: Option<std::time::Duration>,
203 mcp_config: Option<String>,
204 progress: Box<dyn ProgressHandler>,
205 session_log_mode: SessionLogMode,
206 log_event_callback: Option<LogEventCallback>,
210 stream_events_format: Option<ListenFormat>,
213 stream_show_thinking: bool,
216 on_spawn_hook: Option<crate::agent::OnSpawnHook>,
221 exit_hint: Option<crate::exit_mode::ExitHint>,
227 register_process_opts: Option<RegisterOptionsOwned>,
234}
235
236impl Default for AgentBuilder {
237 fn default() -> Self {
238 Self::new()
239 }
240}
241
242impl AgentBuilder {
243 pub fn new() -> Self {
245 Self {
246 provider: None,
247 provider_explicit: false,
248 model: None,
249 system_prompt: None,
250 root: None,
251 auto_approve: false,
252 add_dirs: Vec::new(),
253 files: Vec::new(),
254 env_vars: Vec::new(),
255 worktree: None,
256 sandbox: None,
257 size: None,
258 json_mode: false,
259 json_schema: None,
260 session_id: None,
261 metadata: SessionMetadata::default(),
262 output_format: None,
263 input_format: None,
264 replay_user_messages: false,
265 include_partial_messages: false,
266 verbose: false,
267 quiet: false,
268 show_usage: false,
269 max_turns: None,
270 timeout: None,
271 mcp_config: None,
272 progress: Box::new(SilentProgress),
273 session_log_mode: SessionLogMode::Disabled,
274 log_event_callback: None,
275 stream_events_format: None,
276 stream_show_thinking: false,
277 on_spawn_hook: None,
278 register_process_opts: None,
279 exit_hint: None,
280 }
281 }
282
283 pub fn provider(mut self, provider: &str) -> Self {
290 self.provider = Some(provider.to_string());
291 self.provider_explicit = true;
292 self
293 }
294
295 pub fn model(mut self, model: &str) -> Self {
297 self.model = Some(model.to_string());
298 self
299 }
300
301 pub fn system_prompt(mut self, prompt: &str) -> Self {
303 self.system_prompt = Some(prompt.to_string());
304 self
305 }
306
307 pub fn root(mut self, root: &str) -> Self {
309 self.root = Some(root.to_string());
310 self
311 }
312
313 pub fn auto_approve(mut self, approve: bool) -> Self {
315 self.auto_approve = approve;
316 self
317 }
318
319 pub fn add_dir(mut self, dir: &str) -> Self {
321 self.add_dirs.push(dir.to_string());
322 self
323 }
324
325 pub fn file(mut self, path: &str) -> Self {
327 self.files.push(path.to_string());
328 self
329 }
330
331 pub fn env(mut self, key: &str, value: &str) -> Self {
333 self.env_vars.push((key.to_string(), value.to_string()));
334 self
335 }
336
337 pub fn worktree(mut self, name: Option<&str>) -> Self {
339 self.worktree = Some(name.map(String::from));
340 self
341 }
342
343 pub fn sandbox(mut self, name: Option<&str>) -> Self {
345 self.sandbox = Some(name.map(String::from));
346 self
347 }
348
349 pub fn size(mut self, size: &str) -> Self {
351 self.size = Some(size.to_string());
352 self
353 }
354
355 pub fn json(mut self) -> Self {
357 self.json_mode = true;
358 self
359 }
360
361 pub fn json_schema(mut self, schema: serde_json::Value) -> Self {
364 self.json_schema = Some(schema);
365 self.json_mode = true;
366 self
367 }
368
369 pub fn exit(mut self, hint: Option<&str>) -> Self {
382 self.exit_hint = Some(crate::exit_mode::ExitHint::from_optional(
383 hint.map(str::to_string),
384 ));
385 self
386 }
387
388 pub fn session_id(mut self, id: &str) -> Self {
390 self.session_id = Some(id.to_string());
391 self
392 }
393
394 pub fn name(mut self, name: &str) -> Self {
401 self.metadata.name = Some(name.to_string());
402 self
403 }
404
405 pub fn description(mut self, description: &str) -> Self {
408 self.metadata.description = Some(description.to_string());
409 self
410 }
411
412 pub fn tag(mut self, tag: &str) -> Self {
415 self.metadata.tags.push(tag.to_string());
416 self
417 }
418
419 pub fn metadata(mut self, metadata: SessionMetadata) -> Self {
421 self.metadata = metadata;
422 self
423 }
424
425 pub fn output_format(mut self, format: &str) -> Self {
427 self.output_format = Some(format.to_string());
428 self
429 }
430
431 pub fn input_format(mut self, format: &str) -> Self {
436 self.input_format = Some(format.to_string());
437 self
438 }
439
440 pub fn replay_user_messages(mut self, replay: bool) -> Self {
446 self.replay_user_messages = replay;
447 self
448 }
449
450 pub fn include_partial_messages(mut self, include: bool) -> Self {
460 self.include_partial_messages = include;
461 self
462 }
463
464 pub fn verbose(mut self, v: bool) -> Self {
466 self.verbose = v;
467 self
468 }
469
470 pub fn quiet(mut self, q: bool) -> Self {
472 self.quiet = q;
473 self
474 }
475
476 pub fn show_usage(mut self, show: bool) -> Self {
478 self.show_usage = show;
479 self
480 }
481
482 pub fn max_turns(mut self, turns: u32) -> Self {
484 self.max_turns = Some(turns);
485 self
486 }
487
488 pub fn timeout(mut self, duration: std::time::Duration) -> Self {
491 self.timeout = Some(duration);
492 self
493 }
494
495 pub fn mcp_config(mut self, config: &str) -> Self {
502 self.mcp_config = Some(config.to_string());
503 self
504 }
505
506 pub fn on_progress(mut self, handler: Box<dyn ProgressHandler>) -> Self {
508 self.progress = handler;
509 self
510 }
511
512 pub fn session_log(mut self, mode: SessionLogMode) -> Self {
514 self.session_log_mode = mode;
515 self
516 }
517
518 pub fn enable_session_log(mut self, enable: bool) -> Self {
521 self.session_log_mode = if enable {
522 SessionLogMode::Auto
523 } else {
524 SessionLogMode::Disabled
525 };
526 self
527 }
528
529 pub fn on_log_event<F>(mut self, f: F) -> Self
534 where
535 F: Fn(&AgentLogEvent) + Send + Sync + 'static,
536 {
537 self.log_event_callback = Some(Arc::new(f));
538 if matches!(self.session_log_mode, SessionLogMode::Disabled) {
539 self.session_log_mode = SessionLogMode::Auto;
540 }
541 self
542 }
543
544 pub fn stream_events_to_stderr(mut self, format: ListenFormat) -> Self {
551 self.stream_events_format = Some(format);
552 if matches!(self.session_log_mode, SessionLogMode::Disabled) {
553 self.session_log_mode = SessionLogMode::Auto;
554 }
555 self
556 }
557
558 pub fn stream_show_thinking(mut self, show: bool) -> Self {
561 self.stream_show_thinking = show;
562 self
563 }
564
565 pub fn on_spawn<F>(mut self, f: F) -> Self
575 where
576 F: Fn(u32) + Send + Sync + 'static,
577 {
578 self.on_spawn_hook = Some(Arc::new(f));
579 self
580 }
581
582 pub fn register_process(mut self, opts: RegisterOptionsOwned) -> Self {
601 self.register_process_opts = Some(opts);
602 self
603 }
604}
605
606fn apply_registration(builder: &mut AgentBuilder, reg: &ProcessRegistration) {
611 for (k, v) in reg.env_vars() {
612 builder.env_vars.push((k.clone(), v.clone()));
613 }
614 let reg_hook = reg.on_spawn_hook();
615 let prev_hook = builder.on_spawn_hook.take();
616 builder.on_spawn_hook = Some(Arc::new(move |pid: u32| {
617 reg_hook(pid);
618 if let Some(ref h) = prev_hook {
619 h(pid);
620 }
621 }));
622}
623
624fn status_for_result<T>(result: &Result<T>) -> (&'static str, Option<i32>) {
629 match result {
630 Ok(_) => ("exited", Some(0)),
631 Err(err) => {
632 let exit_code = err
633 .downcast_ref::<crate::process::ProcessError>()
634 .and_then(|pe| pe.exit_code)
635 .unwrap_or(1);
636 ("killed", Some(exit_code))
637 }
638 }
639}
640
641impl AgentBuilder {
642 fn persist_session_metadata_with_id(
652 &self,
653 provider: &str,
654 model: &str,
655 effective_root: Option<&str>,
656 explicit_session_id: Option<&str>,
657 ) -> Option<String> {
658 let has_metadata = self.metadata.name.is_some()
659 || self.metadata.description.is_some()
660 || !self.metadata.tags.is_empty()
661 || self.exit_hint.is_some();
665 if !has_metadata {
666 return None;
667 }
668
669 let session_id = explicit_session_id
670 .map(String::from)
671 .or_else(|| self.session_id.clone())
672 .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
673 let workspace_path = effective_root
674 .map(String::from)
675 .or_else(|| self.root.clone())
676 .unwrap_or_else(|| {
677 std::env::current_dir()
678 .map(|p| p.to_string_lossy().to_string())
679 .unwrap_or_default()
680 });
681
682 let entry = SessionEntry {
683 session_id: session_id.clone(),
684 provider: provider.to_string(),
685 model: model.to_string(),
686 worktree_path: workspace_path,
687 worktree_name: String::new(),
688 created_at: chrono::Utc::now().to_rfc3339(),
689 provider_session_id: None,
690 sandbox_name: None,
691 is_worktree: self.worktree.is_some(),
692 discovered: false,
693 discovery_source: None,
694 log_path: None,
695 log_completeness: "partial".to_string(),
696 name: self.metadata.name.clone(),
697 description: self.metadata.description.clone(),
698 tags: self.metadata.tags.clone(),
699 dependencies: Vec::new(),
700 retried_from: None,
701 interactive: false,
702 exit: self
703 .exit_hint
704 .as_ref()
705 .map(|h| crate::exit_mode::ExitConstraints {
706 hint: Some(h.clone()),
707 json_mode: self.json_mode,
708 schema: self.json_schema.clone(),
709 }),
710 };
711
712 let mut store = SessionStore::load(self.root.as_deref()).unwrap_or_default();
713 store.add(entry);
714 if let Err(e) = store.save(self.root.as_deref()) {
715 warn!("Failed to persist session metadata: {e}");
716 }
717
718 Some(session_id)
719 }
720
721 fn prepend_files(&self, prompt: &str) -> Result<String> {
723 if self.files.is_empty() {
724 return Ok(prompt.to_string());
725 }
726 let attachments: Vec<Attachment> = self
727 .files
728 .iter()
729 .map(|f| Attachment::from_path(std::path::Path::new(f)))
730 .collect::<Result<Vec<_>>>()?;
731 let prefix = attachment::format_attachments_prefix(&attachments);
732 Ok(format!("{prefix}{prompt}"))
733 }
734
735 fn resolve_provider(&self) -> Result<String> {
737 if let Some(ref p) = self.provider {
738 let p = p.to_lowercase();
739 if !Config::VALID_PROVIDERS.contains(&p.as_str()) {
740 bail!(
741 "Invalid provider '{}'. Available: {}",
742 p,
743 Config::VALID_PROVIDERS.join(", ")
744 );
745 }
746 return Ok(p);
747 }
748 let config = Config::load(self.root.as_deref()).unwrap_or_default();
749 if let Some(p) = config.provider() {
750 return Ok(p.to_string());
751 }
752 Ok("claude".to_string())
753 }
754
755 async fn create_agent(&self, provider: &str) -> Result<(Box<dyn Agent + Send + Sync>, String)> {
762 let base_system_prompt = self.system_prompt.clone().or_else(|| {
764 Config::load(self.root.as_deref())
765 .unwrap_or_default()
766 .system_prompt()
767 .map(String::from)
768 });
769
770 let system_prompt = if self.json_mode && provider != "claude" {
772 let mut prompt = base_system_prompt.unwrap_or_default();
773 if let Some(ref schema) = self.json_schema {
774 let schema_str = serde_json::to_string_pretty(schema).unwrap_or_default();
775 prompt.push_str(&format!(
776 "\n\nYou MUST respond with valid JSON only. No markdown fences, no explanations. \
777 Your response must conform to this JSON schema:\n{schema_str}"
778 ));
779 } else {
780 prompt.push_str(
781 "\n\nYou MUST respond with valid JSON only. No markdown fences, no explanations.",
782 );
783 }
784 Some(prompt)
785 } else {
786 base_system_prompt
787 };
788
789 self.progress
790 .on_spinner_start(&format!("Initializing {provider} agent"));
791
792 let progress = &*self.progress;
793 let mut on_downgrade = |from: &str, to: &str, reason: &str| {
794 progress.on_warning(&format!("Downgrading provider: {from} → {to} ({reason})"));
795 };
796 let (mut agent, effective_provider) = AgentFactory::create_with_fallback(
797 provider,
798 self.provider_explicit,
799 system_prompt,
800 self.model.clone(),
801 self.root.clone(),
802 self.auto_approve,
803 self.add_dirs.clone(),
804 &mut on_downgrade,
805 )
806 .await?;
807 let provider = effective_provider.as_str();
808
809 let effective_max_turns = self.max_turns.or_else(|| {
811 Config::load(self.root.as_deref())
812 .unwrap_or_default()
813 .max_turns()
814 });
815 if let Some(turns) = effective_max_turns {
816 agent.set_max_turns(turns);
817 }
818
819 let mut output_format = self.output_format.clone();
821 if self.json_mode && output_format.is_none() {
822 output_format = Some("json".to_string());
823 if provider != "claude" {
824 agent.set_capture_output(true);
825 }
826 }
827 agent.set_output_format(output_format);
828
829 if provider == "claude"
831 && let Some(claude_agent) = agent.as_any_mut().downcast_mut::<Claude>()
832 {
833 claude_agent.set_verbose(self.verbose);
834 if let Some(ref session_id) = self.session_id {
835 claude_agent.set_session_id(session_id.clone());
836 }
837 if let Some(ref input_fmt) = self.input_format {
838 claude_agent.set_input_format(Some(input_fmt.clone()));
839 }
840 if self.replay_user_messages {
841 claude_agent.set_replay_user_messages(true);
842 }
843 if self.include_partial_messages {
844 claude_agent.set_include_partial_messages(true);
845 }
846 if self.json_mode
847 && let Some(ref schema) = self.json_schema
848 {
849 let schema_str = serde_json::to_string(schema).unwrap_or_default();
850 claude_agent.set_json_schema(Some(schema_str));
851 }
852 if self.mcp_config.is_some() {
853 claude_agent.set_mcp_config(self.mcp_config.clone());
854 }
855 }
856
857 if provider == "ollama"
859 && let Some(ollama_agent) = agent.as_any_mut().downcast_mut::<Ollama>()
860 {
861 let config = Config::load(self.root.as_deref()).unwrap_or_default();
862 if let Some(ref size) = self.size {
863 let resolved = config.ollama_size_for(size);
864 ollama_agent.set_size(resolved.to_string());
865 }
866 }
867
868 if let Some(ref sandbox_opt) = self.sandbox {
870 let sandbox_name = sandbox_opt
871 .as_deref()
872 .map(String::from)
873 .unwrap_or_else(crate::sandbox::generate_name);
874 let template = crate::sandbox::template_for_provider(provider);
875 let workspace = self.root.clone().unwrap_or_else(|| ".".to_string());
876 agent.set_sandbox(SandboxConfig {
877 name: sandbox_name,
878 template: template.to_string(),
879 workspace,
880 });
881 }
882
883 if !self.env_vars.is_empty() {
884 agent.set_env_vars(self.env_vars.clone());
885 }
886
887 if let Some(ref hook) = self.on_spawn_hook {
888 agent.set_on_spawn_hook(hook.clone());
889 }
890
891 self.progress.on_spinner_finish();
892 self.progress.on_success(&format!(
893 "{} initialized with model {}",
894 provider,
895 agent.get_model()
896 ));
897
898 Ok((agent, effective_provider))
899 }
900
901 fn start_session_log(
909 &mut self,
910 command: &str,
911 resumed: bool,
912 provider: &str,
913 model: &str,
914 provider_session_id: Option<&str>,
915 ) -> Option<SessionLogGuard> {
916 let mode = std::mem::replace(&mut self.session_log_mode, SessionLogMode::Disabled);
917 match mode {
918 SessionLogMode::Disabled => None,
919 SessionLogMode::External(c) => {
920 let wrapper_session_id = c
921 .writer()
922 .log_path()
923 .ok()
924 .and_then(|p| p.file_stem().map(|s| s.to_string_lossy().to_string()))
925 .unwrap_or_default();
926 let log_path = c.writer().log_path().ok();
927 self.apply_event_callback(c.writer());
928 Some(SessionLogGuard {
929 coordinator: None, wrapper_session_id,
931 log_path,
932 external_writer: Some(c.writer().clone()),
933 _owned_external: Some(c),
934 })
935 }
936 SessionLogMode::Auto => {
937 let wrapper_session_id = self
938 .session_id
939 .clone()
940 .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
941 let metadata = SessionLogMetadata {
942 provider: provider.to_string(),
943 wrapper_session_id: wrapper_session_id.clone(),
944 provider_session_id: provider_session_id.map(str::to_string),
945 workspace_path: self.root.clone().or_else(|| {
946 std::env::current_dir()
947 .ok()
948 .map(|p| p.to_string_lossy().to_string())
949 }),
950 command: command.to_string(),
951 model: Some(model.to_string()),
952 resumed,
953 backfilled: false,
954 };
955 let live_ctx = LiveLogContext {
956 root: self.root.clone(),
957 provider_session_id: metadata.provider_session_id.clone(),
958 workspace_path: metadata.workspace_path.clone(),
959 started_at: chrono::Utc::now(),
960 is_worktree: self.worktree.is_some(),
961 };
962 let adapter = live_adapter_for_provider(provider, live_ctx, true);
963 let callback = self.build_event_callback();
964 match SessionLogCoordinator::start_with_callback(
965 &logs_dir(self.root.as_deref()),
966 metadata,
967 adapter,
968 callback,
969 ) {
970 Ok(c) => {
971 let _ = c.writer().set_global_index_dir(Config::global_base_dir());
972 let log_path = c.writer().log_path().ok();
973 Some(SessionLogGuard {
974 coordinator: Some(c),
975 wrapper_session_id,
976 log_path,
977 external_writer: None,
978 _owned_external: None,
979 })
980 }
981 Err(e) => {
982 warn!("Failed to start session log coordinator: {e}");
983 None
984 }
985 }
986 }
987 }
988 }
989
990 fn build_event_callback(&self) -> Option<LogEventCallback> {
994 let user_cb = self.log_event_callback.clone();
995 let stream_fmt = self.stream_events_format;
996 let show_thinking = self.stream_show_thinking;
997
998 if user_cb.is_none() && stream_fmt.is_none() {
999 return None;
1000 }
1001
1002 Some(Arc::new(move |event: &AgentLogEvent| {
1003 if let Some(ref user) = user_cb {
1004 user(event);
1005 }
1006 if let Some(fmt) = stream_fmt
1007 && let Some(text) = listen::format_event(event, fmt, show_thinking)
1008 {
1009 eprintln!("{text}");
1010 }
1011 }))
1012 }
1013
1014 fn apply_event_callback(&self, writer: &crate::session_log::SessionLogWriter) {
1019 if let Some(cb) = self.build_event_callback() {
1020 if let Err(e) = writer.set_event_callback(cb) {
1021 warn!("Failed to register session log event callback: {e}");
1022 }
1023 }
1024 }
1025
1026 pub async fn exec(mut self, prompt: &str) -> Result<AgentOutput> {
1030 let registration = self
1031 .register_process_opts
1032 .as_ref()
1033 .map(|opts| process_registration::register(opts.as_borrowed()));
1034 if let Some(ref reg) = registration {
1035 apply_registration(&mut self, reg);
1036 }
1037 let result = self.exec_inner(prompt).await;
1038 if let Some(reg) = registration {
1039 let (status, code) = status_for_result(&result);
1040 reg.update_status(status, code);
1041 }
1042 result
1043 }
1044
1045 async fn exec_inner(self, prompt: &str) -> Result<AgentOutput> {
1046 if self.exit_hint.is_some() {
1047 bail!(
1048 "`exit()` is only valid with `run()`. `exec()` already produces a structured \
1049 output natively, so combining the two would be ambiguous."
1050 );
1051 }
1052 let provider = self.resolve_provider()?;
1053 debug!("exec: provider={provider}");
1054
1055 let effective_root = if let Some(ref wt_opt) = self.worktree {
1057 let wt_name = wt_opt
1058 .as_deref()
1059 .map(String::from)
1060 .unwrap_or_else(worktree::generate_name);
1061 let repo_root = worktree::git_repo_root(self.root.as_deref())?;
1062 let wt_path = worktree::create_worktree(&repo_root, &wt_name)?;
1063 self.progress
1064 .on_success(&format!("Worktree created at {}", wt_path.display()));
1065 Some(wt_path.to_string_lossy().to_string())
1066 } else {
1067 self.root.clone()
1068 };
1069
1070 let mut builder = self;
1071 if effective_root.is_some() {
1072 builder.root = effective_root;
1073 }
1074
1075 let (agent, provider) = builder.create_agent(&provider).await?;
1076
1077 let log_guard =
1081 builder.start_session_log("exec", false, &provider, agent.get_model(), None);
1082
1083 let _ = builder.persist_session_metadata_with_id(
1088 &provider,
1089 agent.get_model(),
1090 builder.root.as_deref(),
1091 log_guard.as_ref().map(|g| g.wrapper_session_id.as_str()),
1092 );
1093
1094 let prompt_with_files = builder.prepend_files(prompt)?;
1096
1097 let effective_prompt = if builder.json_mode && provider != "claude" {
1099 format!(
1100 "IMPORTANT: You MUST respond with valid JSON only. No markdown, no explanation.\n\n{prompt_with_files}"
1101 )
1102 } else {
1103 prompt_with_files
1104 };
1105
1106 let result = if let Some(timeout_dur) = builder.timeout {
1107 match tokio::time::timeout(timeout_dur, agent.run(Some(&effective_prompt))).await {
1108 Ok(r) => r?,
1109 Err(_) => {
1110 agent.cleanup().await.ok();
1111 bail!("Agent timed out after {}", format_duration(timeout_dur));
1112 }
1113 }
1114 } else {
1115 agent.run(Some(&effective_prompt)).await?
1116 };
1117
1118 agent.cleanup().await?;
1120
1121 let log_path_string = log_guard.as_ref().and_then(|g| g.log_path_string());
1122
1123 if let Some(mut output) = result {
1124 if let Some(ref schema) = builder.json_schema {
1126 if !builder.json_mode {
1127 warn!(
1128 "json_schema is set but json_mode is false — \
1129 schema will not be sent to the agent, only used for output validation"
1130 );
1131 }
1132 if let Some(ref result_text) = output.result {
1133 debug!(
1134 "exec: validating result ({} bytes): {:.300}",
1135 result_text.len(),
1136 result_text
1137 );
1138 if let Err(errors) = json_validation::validate_json_schema(result_text, schema)
1139 {
1140 let preview = if result_text.len() > 500 {
1141 &result_text[..500]
1142 } else {
1143 result_text.as_str()
1144 };
1145 bail!(
1146 "JSON schema validation failed: {}\nRaw agent output ({} bytes):\n{}",
1147 errors.join("; "),
1148 result_text.len(),
1149 preview
1150 );
1151 }
1152 }
1153 }
1154 output.log_path = log_path_string;
1155 let success = !output.is_error;
1156 let err_msg = output.error_message.clone();
1157 if let Some(g) = log_guard {
1158 g.finish(success, err_msg).await;
1159 }
1160 Ok(output)
1161 } else {
1162 let mut output = AgentOutput::from_text(&provider, "");
1164 output.log_path = log_path_string;
1165 if let Some(g) = log_guard {
1166 g.finish(true, None).await;
1167 }
1168 Ok(output)
1169 }
1170 }
1171
1172 pub async fn exec_streaming(self, prompt: &str) -> Result<StreamingSession> {
1241 let provider = self.resolve_provider()?;
1242 debug!("exec_streaming: provider={provider}");
1243
1244 if provider != "claude" {
1245 bail!("Streaming input is only supported by the Claude provider");
1246 }
1247
1248 let prompt_with_files = self.prepend_files(prompt)?;
1250
1251 let mut builder = self;
1254 builder.provider_explicit = true;
1255 let (agent, _provider) = builder.create_agent(&provider).await?;
1256
1257 let claude_agent = agent
1259 .as_any_ref()
1260 .downcast_ref::<Claude>()
1261 .ok_or_else(|| anyhow::anyhow!("Failed to downcast agent to Claude"))?;
1262
1263 claude_agent.execute_streaming(Some(&prompt_with_files))
1264 }
1265
1266 pub async fn run(mut self, prompt: Option<&str>) -> Result<()> {
1270 let registration = self
1271 .register_process_opts
1272 .as_ref()
1273 .map(|opts| process_registration::register(opts.as_borrowed()));
1274 if let Some(ref reg) = registration {
1275 apply_registration(&mut self, reg);
1276 }
1277 let result = self.run_inner(prompt).await;
1278 if let Some(reg) = registration {
1279 let (status, code) = status_for_result(&result);
1280 reg.update_status(status, code);
1281 }
1282 result
1283 }
1284
1285 async fn run_inner(self, prompt: Option<&str>) -> Result<()> {
1286 let provider = self.resolve_provider()?;
1287 debug!("run: provider={provider}");
1288
1289 let prompt_with_files = match prompt {
1291 Some(p) => Some(self.prepend_files(p)?),
1292 None if !self.files.is_empty() => {
1293 let attachments: Vec<Attachment> = self
1294 .files
1295 .iter()
1296 .map(|f| Attachment::from_path(std::path::Path::new(f)))
1297 .collect::<Result<Vec<_>>>()?;
1298 Some(attachment::format_attachments_prefix(&attachments))
1299 }
1300 None => None,
1301 };
1302
1303 let prompt_with_exit = match (self.exit_hint.as_ref(), prompt_with_files) {
1307 (Some(hint), Some(p)) => {
1308 let suffix = crate::exit_mode::build_exit_suffix(
1309 hint.as_str(),
1310 self.json_mode,
1311 self.json_schema.as_ref(),
1312 );
1313 Some(format!("{p}\n\n{suffix}"))
1314 }
1315 (Some(hint), None) => Some(crate::exit_mode::build_exit_suffix(
1316 hint.as_str(),
1317 self.json_mode,
1318 self.json_schema.as_ref(),
1319 )),
1320 (None, p) => p,
1321 };
1322
1323 let mut builder = self;
1324 let (agent, effective_provider) = builder.create_agent(&provider).await?;
1325 let log_guard =
1326 builder.start_session_log("run", false, &effective_provider, agent.get_model(), None);
1327 let _ = builder.persist_session_metadata_with_id(
1328 &effective_provider,
1329 agent.get_model(),
1330 builder.root.as_deref(),
1331 log_guard.as_ref().map(|g| g.wrapper_session_id.as_str()),
1332 );
1333 agent.run_interactive(prompt_with_exit.as_deref()).await?;
1334 agent.cleanup().await?;
1335 if let Some(g) = log_guard {
1336 g.finish(true, None).await;
1337 }
1338 Ok(())
1339 }
1340
1341 pub async fn resume(mut self, session_id: &str) -> Result<()> {
1343 let registration = self
1344 .register_process_opts
1345 .as_ref()
1346 .map(|opts| process_registration::register(opts.as_borrowed()));
1347 if let Some(ref reg) = registration {
1348 apply_registration(&mut self, reg);
1349 }
1350 let result = self.resume_inner(session_id).await;
1351 if let Some(reg) = registration {
1352 let (status, code) = status_for_result(&result);
1353 reg.update_status(status, code);
1354 }
1355 result
1356 }
1357
1358 async fn resume_inner(self, session_id: &str) -> Result<()> {
1359 let provider = self.resolve_provider()?;
1360 debug!("resume: provider={provider}, session={session_id}");
1361
1362 let mut builder = self;
1364 builder.provider_explicit = true;
1365 let (agent, effective_provider) = builder.create_agent(&provider).await?;
1366 let log_guard = builder.start_session_log(
1367 "resume",
1368 true,
1369 &effective_provider,
1370 agent.get_model(),
1371 Some(session_id),
1372 );
1373 agent.run_resume(Some(session_id), false).await?;
1374 agent.cleanup().await?;
1375 if let Some(g) = log_guard {
1376 g.finish(true, None).await;
1377 }
1378 Ok(())
1379 }
1380
1381 pub async fn resume_with_prompt(
1394 mut self,
1395 session_id: &str,
1396 prompt: &str,
1397 ) -> Result<Option<AgentOutput>> {
1398 let registration = self
1399 .register_process_opts
1400 .as_ref()
1401 .map(|opts| process_registration::register(opts.as_borrowed()));
1402 if let Some(ref reg) = registration {
1403 apply_registration(&mut self, reg);
1404 }
1405 let result = self.resume_with_prompt_inner(session_id, prompt).await;
1406 if let Some(reg) = registration {
1407 let (status, code) = status_for_result(&result);
1408 reg.update_status(status, code);
1409 }
1410 result
1411 }
1412
1413 async fn resume_with_prompt_inner(
1414 self,
1415 session_id: &str,
1416 prompt: &str,
1417 ) -> Result<Option<AgentOutput>> {
1418 let provider = self.resolve_provider()?;
1419 debug!("resume_with_prompt: provider={provider}, session={session_id}");
1420
1421 let mut builder = self;
1423 builder.provider_explicit = true;
1424 let (agent, effective_provider) = builder.create_agent(&provider).await?;
1425 let log_guard = builder.start_session_log(
1431 "resume",
1432 true,
1433 &effective_provider,
1434 agent.get_model(),
1435 Some(session_id),
1436 );
1437 let output = agent.run_resume_with_prompt(session_id, prompt).await?;
1438 agent.cleanup().await?;
1439 if let Some(g) = log_guard {
1440 g.finish(true, None).await;
1441 }
1442 Ok(output)
1443 }
1444
1445 pub async fn continue_last(mut self) -> Result<()> {
1447 let registration = self
1448 .register_process_opts
1449 .as_ref()
1450 .map(|opts| process_registration::register(opts.as_borrowed()));
1451 if let Some(ref reg) = registration {
1452 apply_registration(&mut self, reg);
1453 }
1454 let result = self.continue_last_inner().await;
1455 if let Some(reg) = registration {
1456 let (status, code) = status_for_result(&result);
1457 reg.update_status(status, code);
1458 }
1459 result
1460 }
1461
1462 async fn continue_last_inner(self) -> Result<()> {
1463 let provider = self.resolve_provider()?;
1464 debug!("continue_last: provider={provider}");
1465
1466 let mut builder = self;
1468 builder.provider_explicit = true;
1469 let (agent, effective_provider) = builder.create_agent(&provider).await?;
1470 let log_guard =
1471 builder.start_session_log("resume", true, &effective_provider, agent.get_model(), None);
1472 agent.run_resume(None, true).await?;
1473 agent.cleanup().await?;
1474 if let Some(g) = log_guard {
1475 g.finish(true, None).await;
1476 }
1477 Ok(())
1478 }
1479}
1480
1481#[cfg(test)]
1482#[path = "builder_tests.rs"]
1483mod tests;