1use std::future::Future;
4use std::sync::Arc;
5use std::time::Duration;
6
7use crate::runtime_storage::{RuntimeStorage, local_runtime_storage, remote_runtime_storage};
8use crate::session_activation::SessionActivator;
9use crate::startup_config::StartupConfig;
10use crate::stream_auth::stream_fn_with_auth_store;
11use crate::turn::daemon::{DaemonConfig, RuntimeCapabilities, TurnHost};
12use crate::{agent_specs, runtime_capabilities, session_ops};
13use anyhow::{Context, Result};
14use theway_core::executor::ExecutorKind;
15use theway_core::multiagent::graph::engine::DagEngine;
16use theway_core::{PermissionPolicy, ThinkingLevel};
17
18use super::session::SessionProjectResources;
19use super::{
20 DaemonServices, SessionExecutionContext, SessionHookResources, SessionMcpResources,
21 SessionRuntimeBuilder,
22};
23
24#[derive(Clone, Copy, Debug, PartialEq, Eq)]
25pub enum DaemonTransport {
26 Grpc,
27 Http,
28 Mcp,
29}
30
31#[derive(Clone, Debug, PartialEq, Eq)]
32pub enum SessionSelection {
33 New,
34 Latest,
35 Id(String),
36}
37
38pub struct DaemonOptions {
39 pub paths: crate::DaemonPaths,
40 pub transport: DaemonTransport,
41 pub host: String,
42 pub port: u16,
43 pub provider: Option<String>,
44 pub model: Option<String>,
45 pub base_url: Option<String>,
46 pub thinking: ThinkingLevel,
47 pub session: SessionSelection,
48 pub approve_control_plane: bool,
49 pub debug: bool,
50 pub trigger_poll_secs: Option<u64>,
51 pub builtin_skills: Vec<String>,
52 pub storage_service_addr: Option<String>,
53 pub executor_kind: Option<String>,
54}
55
56const STORAGE_WATCH_INTERVAL: Duration = Duration::from_secs(1);
57const STORAGE_WATCH_TIMEOUT: Duration = Duration::from_millis(700);
58const STORAGE_WATCH_FAILURES: usize = 3;
59
60async fn monitor_controller_storage(
64 addr: &str,
65 interval: Duration,
66 timeout: Duration,
67 failure_limit: usize,
68) -> Result<()> {
69 debug_assert!(failure_limit > 0);
70 let mut ticker = tokio::time::interval(interval);
71 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
72 let mut failures = 0usize;
73
74 loop {
75 ticker.tick().await;
76 match theway_transport::client::probe_storage_service(addr, timeout).await {
77 Ok(()) => {
78 if failures > 0 {
79 tracing::info!(
80 "controller storage at {addr} recovered after {failures} failed probe(s)"
81 );
82 }
83 failures = 0;
84 }
85 Err(error) => {
86 failures += 1;
87 tracing::warn!(
88 "controller storage probe {failures}/{failure_limit} failed at {addr}: {error}"
89 );
90 if failures >= failure_limit {
91 tracing::warn!(
92 "controller storage at {addr} remained unavailable for {failure_limit} consecutive probes; shutting down daemon"
93 );
94 return Ok(());
95 }
96 }
97 }
98 }
99}
100
101async fn supervise_controller_storage<F>(storage_addr: Option<&str>, server: F) -> Result<()>
102where
103 F: Future<Output = Result<()>>,
104{
105 let Some(addr) = storage_addr else {
106 return server.await;
107 };
108 tokio::pin!(server);
109 tokio::select! {
110 result = &mut server => result,
111 result = monitor_controller_storage(
112 addr,
113 STORAGE_WATCH_INTERVAL,
114 STORAGE_WATCH_TIMEOUT,
115 STORAGE_WATCH_FAILURES,
116 ) => result,
117 }
118}
119
120fn canonical_work_dir(path: &std::path::Path) -> Result<std::path::PathBuf> {
121 let canonical = path
122 .canonicalize()
123 .with_context(|| format!("cd into {}", path.display()))?;
124 if !canonical.is_dir() {
125 anyhow::bail!("work directory is not a directory: {}", canonical.display());
126 }
127 Ok(canonical)
128}
129
130pub async fn run(options: DaemonOptions) -> Result<()> {
131 let mode = options.transport;
132 let paths = options.paths;
133 let cwd = canonical_work_dir(&paths.work_dir)?;
134 let storage: Arc<dyn RuntimeStorage> = match &options.storage_service_addr {
140 Some(addr) => remote_runtime_storage(addr).await?,
141 None => local_runtime_storage(),
142 };
143 let repo = storage.session_repository(&cwd).await?;
144
145 let initial_settings_payload = theway_transport::wire::WireDaemonConfig::default();
153 let mut startup = StartupConfig::from_wire(&initial_settings_payload);
154 if let Some(secs) = options.trigger_poll_secs {
157 startup.trigger_poll_secs = secs;
158 }
159 if let Some(raw) = options.executor_kind.as_deref() {
163 startup.executor_kind =
164 crate::executor::parse_executor_kind(raw).map_err(anyhow::Error::msg)?;
165 }
166 startup.storage_service_addr = options.storage_service_addr.clone();
167 if options.storage_service_addr.is_some() {
174 startup.load_local_sources = false;
175 }
176 if startup.executor_kind == ExecutorKind::Sandbox {
179 startup.load_local_sources = false;
180 }
181
182 let model = resolve_startup_model(
183 &cwd,
184 options.provider.as_deref(),
185 options.model.as_deref(),
186 options.base_url.as_deref(),
187 &startup,
188 )
189 .await?;
190 let thinking = if options.thinking == ThinkingLevel::Off {
191 startup.thinking_level.unwrap_or(ThinkingLevel::Off)
192 } else {
193 options.thinking
194 };
195
196 let (store, resumed) = match &options.session {
197 SessionSelection::New => (repo.create_lazy(&cwd).await?, false),
204 SessionSelection::Latest => (repo.resume(None).await?, true),
205 SessionSelection::Id(id) => (repo.resume(Some(id)).await?, true),
206 };
207 let session_metadata = store.get_metadata_json().await?;
208 let session_id = session_metadata
209 .get("id")
210 .and_then(|v| v.as_str())
211 .unwrap_or("?")
212 .to_string();
213 let _logging = crate::logging::init(&session_id);
214 let telemetry = crate::observability::TelemetryHandle::init().await;
215 let runtime_observer = telemetry.observer();
216 let (feed_tx, feed_rx) =
217 tokio::sync::mpsc::unbounded_channel::<(String, theway_transport::feed::FeedUpdate)>();
218
219 let stream_fn = stream_fn_with_auth_store();
220 let command_output = {
221 let tx = feed_tx.clone();
222 let session_id = session_id.clone();
223 crate::commands::CommandOutput::new(move |line| {
224 let _ = tx.send((
225 session_id.clone(),
226 theway_transport::feed::FeedUpdate::Plain {
227 text: line,
228 level: theway_transport::feed::Level::Output,
229 },
230 ));
231 })
232 };
233 let services = DaemonServices::new().with_command_output(command_output);
234 let dynamic_trigger_registry = services.dynamic_triggers.clone();
235 if let Err(err) = dynamic_trigger_registry
236 .load_from_storage(storage.clone(), cwd.clone(), session_id.clone())
237 .await
238 {
239 tracing::warn!("dynamic triggers: {err}");
240 }
241 let cron_registry = services.cron.clone();
242 if let Err(err) = cron_registry
243 .load_from_storage(storage.clone(), cwd.clone(), session_id.clone())
244 .await
245 {
246 tracing::warn!("cron: {err}");
247 }
248 let dag_engine = Arc::new(DagEngine::with_observer(runtime_observer.clone()));
249 let subagent_registry =
250 theway_core::multiagent::jobs::SubagentJobRegistry::with_observer(runtime_observer.clone());
251 subagent_registry.set_transcript_store(Some(storage.job_transcript_store(&cwd)));
252 let executor: Arc<dyn theway_core::executor::ToolExecutor> =
256 crate::executor::executor_for_kind(startup.executor_kind, cwd.clone());
257 let session_paths = paths.with_work_dir(cwd.clone());
258 let mcp = if startup.load_local_sources {
263 crate::mcp_loader::load_all(&session_paths).await
264 } else {
265 crate::mcp_loader::LoadedMcp::empty()
266 };
267 let mcp_provision = Arc::new(std::sync::RwLock::new(
270 crate::mcp_loader::McpProvisionState::default(),
271 ));
272 let mut mcp_resources = SessionMcpResources::from_loaded(mcp);
273 if !startup.load_local_sources {
274 mcp_resources.provision = Some(mcp_provision.clone());
277 }
278 let project_resources = SessionProjectResources::load(
279 &session_paths,
280 &options.builtin_skills,
281 &startup.builtin_skills,
282 startup.load_local_sources,
283 )
284 .await?;
285 let hook_resources =
286 SessionHookResources::load(&session_paths, startup.load_local_sources).await;
287 let session_context = SessionExecutionContext::new(
288 session_id.clone(),
289 cwd.clone(),
290 repo.clone(),
291 storage.clone(),
292 paths.clone(),
293 executor.clone(),
294 startup.executor_kind,
295 model.clone(),
296 thinking,
297 project_resources,
298 mcp_resources,
299 hook_resources,
300 );
301 dynamic_trigger_registry.set_poll_interval_secs(startup.trigger_poll_secs);
304 let thinking_summary_cfg = startup.thinking_summary.clone();
309 let before_tool_call = PermissionPolicy::default_for_coding_agent().as_before_tool_call();
310 let (control_plane_hook, control_plane_prompt_tx, control_plane_prompt_rx) =
311 if options.approve_control_plane {
312 (Some(crate::control_plane_prompt::allow_hook()), None, None)
313 } else {
314 let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
315 (None, Some(tx), Some(rx))
316 };
317 let lsp_supervisor = Arc::new(if startup.load_local_sources {
321 crate::lsp_supervisor::LspSupervisor::load(&session_context.paths).await
322 } else {
323 crate::lsp_supervisor::LspSupervisor::from_config(&cwd, Default::default())
324 });
325 let lsp_lang_count = lsp_supervisor.language_count();
326 let after_tool_call = if lsp_supervisor.is_empty() {
327 None
328 } else {
329 Some(crate::lsp_supervisor::as_after_tool_call(
330 lsp_supervisor.clone(),
331 ))
332 };
333 let (main_run_tx, main_run_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
334
335 let session_runtime_builder = Arc::new(SessionRuntimeBuilder {
336 thinking,
337 stream_fn: stream_fn.clone(),
338 dag_engine: dag_engine.clone(),
339 subagent_registry: subagent_registry.clone(),
340 services: services.clone(),
341 before_tool_call: Some(before_tool_call.clone()),
342 control_plane_hook,
343 control_plane_prompt_tx,
344 after_tool_call,
345 feed_tx: feed_tx.clone(),
346 main_run_tx: main_run_tx.clone(),
347 debug: options.debug,
348 session_cells: Default::default(),
349 });
350 services
351 .session_activator
352 .set(Arc::new(SessionActivator::new(
353 &session_runtime_builder,
354 storage.clone(),
355 paths.clone(),
356 thinking,
357 options.builtin_skills.clone(),
358 startup.builtin_skills.clone(),
359 startup.load_local_sources,
360 )))
361 .map_err(|_| anyhow::anyhow!("session activator already installed"))?;
362 let initial_runtime = session_runtime_builder
363 .build_opened(&session_context, store, resumed)
364 .await?;
365 let harness = initial_runtime.harness.clone();
366 let trigger_executor = initial_runtime.trigger_executor.clone();
367 let extension_host = initial_runtime.extension_host.clone();
368 let tool_names = initial_runtime.tool_names;
369 let hooks_active = initial_runtime.hooks_active;
370 let _dag_persist = storage.spawn_dag_persist_for_sessions(
371 dag_engine.clone(),
372 cwd.clone(),
373 services.session_execution.clone(),
374 );
375
376 let session_factory: session_ops::SessionFactory = {
377 let plan = session_runtime_builder;
378 let startup_ctx = session_context.clone();
379 Arc::new(move |id: String| {
380 let plan = plan.clone();
381 let startup_ctx = startup_ctx.clone();
382 Box::pin(async move {
383 let ctx = plan
384 .services
385 .session_execution
386 .get_context(&id)
387 .unwrap_or_else(|| Arc::new(startup_ctx.clone()));
388 plan.build(&ctx, &id).await
389 })
390 })
391 };
392
393 let capabilities = RuntimeCapabilities {
394 mcp_servers: session_context.mcp.server_count,
395 mcp_tools: session_context.mcp.tool_names.len(),
396 mcp_server_names: session_context.mcp.server_names.clone(),
397 mcp_tool_names: session_context.mcp.tool_names.clone(),
398 mcp_server_errors: session_context.mcp.server_errors.clone(),
399 tool_names: tool_names.clone(),
400 mcp_notification_hooks: session_context.mcp.notification_hook_count,
401 hook_points: runtime_capabilities::active_hook_registrations(lsp_lang_count, hooks_active),
402 trigger_features: runtime_capabilities::active_trigger_features(),
403 };
404
405 let thinking_summary = thinking_summary_cfg.map(|cfg| {
406 use crate::turn::thinking_summary::{
407 ThinkingSummarizerFn, ThinkingSummarySettings,
408 };
409 let summarizer_model = model.clone();
410 let summarizer_stream = stream_fn.clone();
411 let summarizer_registry = subagent_registry.clone();
412 let summarizer_session = session_id.clone();
413 let summarizer_launch = agent_specs::launch_resolver();
414 let summarizer: ThinkingSummarizerFn = Arc::new(move |text: String| {
415 let summarizer_launch = summarizer_launch.clone();
416 let summarizer_model = summarizer_model.clone();
417 let summarizer_stream = summarizer_stream.clone();
418 let summarizer_registry = summarizer_registry.clone();
419 let summarizer_session = summarizer_session.clone();
420 Box::pin(async move {
421 let Some(launch) = summarizer_launch("general") else {
422 return Err("general subagent spec unavailable".to_string());
423 };
424 let prompt = format!(
425 "Summarize the following reasoning transcript into a STRUCTURED markdown summary. Output ONLY the summary:\n## Goal\n- ...\n## Key steps\n- ...\n## Findings\n- ...\n## Decision\n- ...\n\nThinking transcript:\n\n{}",
426 theway_transport::feed::truncate_chars(&text, 24_000)
427 );
428 let Some(summarizer_model) = summarizer_model else {
431 return Err("no model set for this session; cannot summarize thinking"
432 .to_string());
433 };
434 let result = theway_core::multiagent::runner::run_agent(
435 theway_core::multiagent::runner::AgentRunOptions {
436 launch,
437 tools: Vec::new(),
439 prompt,
440 model: summarizer_model,
441 stream_fn: Some(summarizer_stream),
442 timeout: None,
443 thinking: None,
444 registry: summarizer_registry,
445 source: "thinking-summary".into(),
446 run_id: None,
447 node_id: None,
448 session_id: Some(summarizer_session),
449 observation_parent: None,
450 cancel: tokio_util::sync::CancellationToken::new(),
451 system_prompt_extra: Some(
452 "You are a thinking summarizer: compress verbose step-by-step \
453 reasoning into a concise structured summary. Never run tools. \
454 Never add commentary beyond the summary."
455 .to_string(),
456 ),
457 on_turn_end: None,
458 },
459 )
460 .await;
461 match result.error {
462 Some(error) => Err(error),
463 None => Ok(result.text),
464 }
465 })
466 });
467 ThinkingSummarySettings {
468 min_chars: cfg.min_chars,
469 summarizer,
470 }
471 });
472
473 let mut host = TurnHost::new(DaemonConfig {
474 harness: harness.clone(),
475 extension_host,
476 trigger_executor,
477 retry: crate::agent_session::RetrySettings::default(),
478 registry: crate::commands::Registry::with_daemon_commands()
479 .with_user_home(paths.home.clone())
480 .with_storage(storage.clone())
481 .with_output(services.command_output.clone())
482 .with_automations(dynamic_trigger_registry.clone(), cron_registry.clone()),
483 cwd: cwd.clone(),
484 paths: paths.clone(),
485 provisioned_skills: session_context.resources.provisioned_skills.clone(),
486 provisioned_templates: session_context.resources.provisioned_templates.clone(),
487 mcp_provision: mcp_provision.clone(),
488 session_id,
489 log_path: _logging.as_ref().map(|l| l.log_path.clone()),
490 tool_count: tool_names.len(),
491 feed_rx,
492 feed_tx: feed_tx.clone(),
493 main_run_rx,
494 control_plane_prompt_rx,
495 dag_engine: dag_engine.clone(),
496 subagent_registry: subagent_registry.clone(),
497 session_factory,
498 session_repo: repo.clone(),
499 capabilities,
500 thinking_summary,
501 startup,
502 services,
503 observability: (*telemetry.status()).clone(),
504 });
505
506 let mode_label = match mode {
507 DaemonTransport::Grpc => "grpc",
508 DaemonTransport::Http => "http",
509 DaemonTransport::Mcp => "mcp",
510 };
511 tracing::info!(
512 "thewayd starting in {mode_label} mode on {}:{}",
513 options.host,
514 options.port
515 );
516
517 let port_file = theway_transport::client::port_file_path(&cwd);
521 let daemon_pid = std::process::id();
522 let on_listen: std::sync::Arc<dyn Fn(std::net::SocketAddr) + Send + Sync> = {
523 let port_file = port_file.clone();
524 std::sync::Arc::new(move |addr| {
525 let entry = format!("{} {}", addr.port(), daemon_pid);
526 if let Err(e) = std::fs::write(&port_file, entry) {
527 tracing::warn!("write daemon port file {}: {e}", port_file.display());
528 }
529 })
530 };
531
532 let result = match mode {
533 DaemonTransport::Mcp => {
534 if let Ok(Some(entry)) = theway_transport::client::read_port_file(&cwd) {
537 if entry
538 .pid
539 .map(|p| !theway_transport::client::pid_alive(p))
540 .unwrap_or(true)
541 {
542 let _ = std::fs::remove_file(&port_file);
543 }
544 }
545 let endpoints = host.transport_endpoints();
548 supervise_controller_storage(options.storage_service_addr.as_deref(), async {
549 crate::mcp_server::run_mcp_server(
550 endpoints.external_ops.clone(),
551 endpoints.job_ops.clone(),
552 )
553 .await
554 .map_err(|e| anyhow::anyhow!("mcp server: {e}"))
555 })
556 .await
557 }
558 DaemonTransport::Grpc => {
559 supervise_controller_storage(
560 options.storage_service_addr.as_deref(),
561 theway_transport::grpc::run_grpc(
562 Box::new(host),
563 theway_transport::grpc::GrpcOptions {
564 host: options.host.clone(),
565 port: options.port,
566 on_listen: Some(on_listen.clone()),
567 },
568 ),
569 )
570 .await
571 }
572 DaemonTransport::Http => {
573 supervise_controller_storage(
574 options.storage_service_addr.as_deref(),
575 theway_transport::http::run_web(
576 Box::new(host),
577 theway_transport::wire::WebOptions {
578 host: options.host.clone(),
579 port: options.port,
580 on_listen: Some(on_listen.clone()),
581 },
582 ),
583 )
584 .await
585 }
586 };
587 _dag_persist.flush().await;
588 dag_engine.abort_all_runs("daemon shutdown");
589 telemetry.shutdown().await;
590 drop(_logging);
591 theway_transport::client::remove_port_file_if_owner(&cwd, daemon_pid);
594 result
595}
596
597async fn resolve_startup_model(
603 cwd: &std::path::Path,
604 cli_provider: Option<&str>,
605 cli_model: Option<&str>,
606 cli_base_url: Option<&str>,
607 startup: &StartupConfig,
608) -> Result<Option<theway_llm_provider::Model>> {
609 let local_models = crate::local_models::load_all(cwd, cli_base_url).await?;
615 if !local_models.models.is_empty() {
616 tracing::info!(
617 "loaded {} local model(s): {}",
618 local_models.models.len(),
619 local_models
620 .models
621 .iter()
622 .map(|m| format!("{}:{}", m.provider.0, m.id))
623 .collect::<Vec<_>>()
624 .join(", ")
625 );
626 }
627
628 let cli_overrides_model = cli_provider.is_some() || cli_model.is_some();
633 let (provider_override, model_override) = if cli_overrides_model {
634 (cli_provider, cli_model)
635 } else {
636 match &startup.model_default {
637 Some(default) => (
638 Some(default.provider.as_str()),
639 Some(default.model.as_str()),
640 ),
641 None => (None, None),
642 }
643 };
644 let Some((provider, id)) = provider_override.zip(model_override) else {
645 return Ok(None);
648 };
649 let mut model = crate::model::auto_detect_model(Some(provider), Some(id))?;
650 if let Some(base_url) = cli_base_url.map(str::trim).filter(|url| !url.is_empty()) {
651 model.base_url = base_url.to_string();
652 }
653 Ok(Some(model))
654}
655
656#[cfg(test)]
657tests_bridge_macro::tests_bridge!("orchestration/startup");