use std::collections::BTreeMap;
use std::future::Future;
use std::path::PathBuf;
use std::sync::Arc;
use crate::trigger_engine::execution::TriggerExecutor;
use crate::trigger_engine::execution::{NotificationStatusSnapshot, RunningTriggerState};
use crate::trigger_engine::notification_hook::{HookState, NotificationHookStatus};
use async_trait::async_trait;
use serde_json::json;
use theway_core::{AgentHarness, AgentTool, SessionTreeEntry, Skill, SkillSource, ThinkingLevel};
use theway_daemon::runtime_storage::{RuntimeStorage, local_runtime_storage};
use theway_llm_provider::{Model, Provider, UserContentBlock, get_model};
use tokio_util::sync::CancellationToken;
#[allow(unused_imports)]
pub use theway_transport::auth::{model_credential_hint, save_api_key};
#[cfg(test)]
pub use theway_transport::commands::console;
#[allow(unused_imports)]
pub use theway_transport::commands::{
CommandOutcome, SlashCommand, WebRelayAction, attach_skill_prompt, cli_model_help_text, parse,
parse_model_spec,
};
#[derive(Clone)]
pub struct CommandOutput {
emit: Arc<dyn Fn(String) + Send + Sync>,
}
impl CommandOutput {
pub fn new(emit: impl Fn(String) + Send + Sync + 'static) -> Self {
Self {
emit: Arc::new(emit),
}
}
pub fn stdout() -> Self {
Self::new(|line| println!("{line}"))
}
fn emit_line(&self, line: String) {
(self.emit)(line);
}
async fn scope<F>(&self, future: F) -> F::Output
where
F: Future,
{
ACTIVE_COMMAND_OUTPUT.scope(self.clone(), future).await
}
}
impl Default for CommandOutput {
fn default() -> Self {
#[cfg(test)]
{
Self::new(console::emit_line)
}
#[cfg(not(test))]
Self::stdout()
}
}
tokio::task_local! {
static ACTIVE_COMMAND_OUTPUT: CommandOutput;
}
fn emit_command_line(line: String) {
let fallback = line.clone();
if ACTIVE_COMMAND_OUTPUT
.try_with(|output| output.emit_line(line))
.is_err()
{
CommandOutput::default().emit_line(fallback);
}
}
macro_rules! cprintln {
() => { $crate::commands::emit_command_line(String::new()) };
($($arg:tt)*) => { $crate::commands::emit_command_line(std::format!($($arg)*)) };
}
pub mod auth;
pub mod goal;
pub mod misc;
pub mod model;
pub mod session;
pub mod skill_cmd;
pub mod skills;
pub mod triggers;
pub use misc::print_help_with_skills;
#[allow(unused_imports)]
pub use misc::print_help;
#[allow(unused_imports)]
pub(crate) use triggers::{render_cron_jobs, render_dynamic_trigger_rules, render_triggers_status};
use goal::{GoalCommand, GoalStartCommand};
use misc::{
BugReportCommand, CompactCommand, DiagCommand, FindCommand, HistoryCommand, TemplateCommand,
WebConnectCommand, WebDisconnectCommand,
};
use model::{CostCommand, ModelCommand, ThinkingCommand};
use session::{
CollapseCommand, ForkCommand, NameCommand, SaveCommand, SessionCommand, ShareCommand,
UndoCommand,
};
use skill_cmd::SkillCommand;
use skills::SkillsCommand;
use triggers::{CronCommand, InboxCommand, NewTriggerCommand, TriggersCommand};
#[cfg(test)]
#[allow(unused_imports)]
use misc::help_text;
#[cfg(test)]
#[allow(unused_imports)]
use model::{model_catalog_text, model_groups, unknown_model_error, unknown_provider_error};
#[cfg(test)]
#[allow(unused_imports)]
use skills::parse_skill_source;
#[cfg(test)]
#[allow(unused_imports)]
use triggers::{
collect_trigger_audit_rows, render_running_triggers, render_trigger_audit,
render_trigger_sources, trigger_decision_details,
};
pub const THINKING_LEVEL_USAGE: &str = "[off|minimal|low|medium|high|xhigh]";
pub struct CommandCtx<'a> {
pub harness: &'a Arc<AgentHarness>,
pub trigger_executor: &'a Arc<TriggerExecutor>,
pub session_id: &'a str,
pub log_path: Option<&'a PathBuf>,
pub tool_count: usize,
pub cwd: &'a std::path::Path,
pub inherit_slot: &'a Arc<std::sync::Mutex<Option<InheritedSessionSettings>>>,
pub mcp_provision: Option<&'a Arc<std::sync::RwLock<crate::mcp_loader::McpProvisionState>>>,
pub auth_base: Option<&'a std::path::PathBuf>,
pub collapse_unload_slot:
&'a Arc<std::sync::Mutex<Option<crate::commands::CollapseUnloadRequest>>>,
}
pub struct DaemonCtx {
pub harness: Arc<AgentHarness>,
pub trigger_executor: Arc<TriggerExecutor>,
pub storage: Arc<dyn RuntimeStorage>,
pub dynamic_triggers: crate::triggers::dynamic::DynamicTriggerRegistry,
pub cron: crate::triggers::cron::CronRegistry,
pub inherit_slot: Arc<std::sync::Mutex<Option<crate::commands::InheritedSessionSettings>>>,
pub collapse_unload_slot: Arc<std::sync::Mutex<Option<crate::commands::CollapseUnloadRequest>>>,
}
#[derive(Clone, Debug)]
pub struct CollapseUnloadRequest {
pub source_id: String,
pub child_id: String,
pub note: String,
}
#[derive(Clone, Debug)]
pub struct InheritedSessionSettings {
pub session_id: String,
pub model_spec: String,
pub thinking_level: Option<String>,
}
pub struct Registry {
inner: theway_transport::commands::Registry<DaemonCtx>,
file_commands: std::sync::RwLock<Vec<crate::file_commands::FileCommand>>,
user_home: std::path::PathBuf,
storage: Arc<dyn RuntimeStorage>,
output: CommandOutput,
dynamic_triggers: crate::triggers::dynamic::DynamicTriggerRegistry,
cron: crate::triggers::cron::CronRegistry,
}
fn default_dynamic_trigger_registry() -> crate::triggers::dynamic::DynamicTriggerRegistry {
#[cfg(test)]
{
crate::triggers::global_registry().clone()
}
#[cfg(not(test))]
crate::triggers::dynamic::DynamicTriggerRegistry::new()
}
fn default_cron_registry() -> crate::triggers::cron::CronRegistry {
#[cfg(test)]
{
crate::triggers::global_cron_registry().clone()
}
#[cfg(not(test))]
crate::triggers::cron::CronRegistry::new()
}
impl Registry {
#[allow(dead_code)]
pub fn new() -> Self {
Self {
inner: theway_transport::commands::Registry::new(),
file_commands: std::sync::RwLock::new(Vec::new()),
user_home: std::path::PathBuf::new(),
storage: local_runtime_storage(),
output: CommandOutput::default(),
dynamic_triggers: default_dynamic_trigger_registry(),
cron: default_cron_registry(),
}
}
pub fn with_daemon_commands() -> Self {
let mut r = Self {
inner: theway_transport::commands::Registry::new(),
file_commands: std::sync::RwLock::new(Vec::new()),
user_home: std::path::PathBuf::new(),
storage: local_runtime_storage(),
output: CommandOutput::default(),
dynamic_triggers: default_dynamic_trigger_registry(),
cron: default_cron_registry(),
};
r.register(Arc::new(auth::LoginCommand));
r.register(Arc::new(auth::LogoutCommand));
r.register(Arc::new(auth::SessionsCommand));
r.register(Arc::new(SkillsCommand));
r.register(Arc::new(SkillCommand));
r.register(Arc::new(ModelCommand));
r.register(Arc::new(ThinkingCommand));
r.register(Arc::new(CostCommand));
r.register(Arc::new(DiagCommand));
r.register(Arc::new(TemplateCommand));
r.register(Arc::new(SaveCommand));
r.register(Arc::new(CompactCommand));
r.register(Arc::new(UndoCommand));
r.register(Arc::new(BugReportCommand));
r.register(Arc::new(NameCommand));
r.register(Arc::new(ForkCommand));
r.register(Arc::new(CollapseCommand));
r.register(Arc::new(SessionCommand));
r.register(Arc::new(WebConnectCommand));
r.register(Arc::new(WebDisconnectCommand));
r.register(Arc::new(ShareCommand));
r.register(Arc::new(FindCommand));
r.register(Arc::new(HistoryCommand));
r.register(Arc::new(GoalCommand));
r.register(Arc::new(GoalStartCommand));
r.register(Arc::new(TriggersCommand));
r.register(Arc::new(NewTriggerCommand));
r.register(Arc::new(CronCommand));
r.register(Arc::new(InboxCommand));
r
}
pub fn with_builtins() -> Self {
Self::with_daemon_commands()
}
#[must_use]
pub fn with_user_home(mut self, home: std::path::PathBuf) -> Self {
self.user_home = home;
self
}
pub fn user_home(&self) -> &std::path::Path {
&self.user_home
}
#[must_use]
pub fn with_storage(mut self, storage: Arc<dyn RuntimeStorage>) -> Self {
self.storage = storage;
self
}
#[must_use]
pub fn with_output(mut self, output: CommandOutput) -> Self {
self.output = output;
self
}
pub fn with_automations(
mut self,
dynamic_triggers: crate::triggers::dynamic::DynamicTriggerRegistry,
cron: crate::triggers::cron::CronRegistry,
) -> Self {
self.dynamic_triggers = dynamic_triggers;
self.cron = cron;
self
}
pub fn register(&mut self, command: Arc<dyn SlashCommand<DaemonCtx>>) {
self.inner.register(command);
}
pub fn set_file_commands(&self, commands: Vec<crate::file_commands::FileCommand>) {
*self
.file_commands
.write()
.unwrap_or_else(|e| e.into_inner()) = commands;
}
pub fn file_commands(&self) -> Vec<crate::file_commands::FileCommand> {
self.file_commands
.read()
.unwrap_or_else(|e| e.into_inner())
.clone()
}
pub fn file_command_names(&self) -> Vec<String> {
self.file_commands()
.into_iter()
.map(|c| format!("/{}", c.name))
.collect()
}
}
impl std::ops::Deref for Registry {
type Target = theway_transport::commands::Registry<DaemonCtx>;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
impl Default for Registry {
fn default() -> Self {
Self::with_daemon_commands()
}
}
fn preview_text(text: &str, max_chars: usize) -> String {
let mut preview = text.chars().take(max_chars).collect::<String>();
if preview.chars().count() < text.chars().count() {
preview.push('…');
}
preview.replace('\n', " ")
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SkillShortcut {
pub command: String,
pub source: SkillSource,
pub description: String,
}
pub fn skill_shortcuts(skills: &[Skill], registry: &Registry) -> Vec<SkillShortcut> {
let mut counts: BTreeMap<&str, usize> = BTreeMap::new();
for skill in skills
.iter()
.filter(|skill| !skill.disable_model_invocation)
{
*counts.entry(skill.name.as_str()).or_default() += 1;
}
let mut shortcuts = skills
.iter()
.filter(|skill| !skill.disable_model_invocation)
.filter(|skill| counts.get(skill.name.as_str()) == Some(&1))
.filter(|skill| registry.find(&skill.name).is_none())
.map(|skill| SkillShortcut {
command: format!("/{}", skill.name),
source: skill.source,
description: preview_text(&skill.description, 72),
})
.collect::<Vec<_>>();
shortcuts.sort_by(|a, b| a.command.cmp(&b.command));
shortcuts
}
fn resolve_skill_shortcut<'a>(
skills: &'a [Skill],
registry: &Registry,
name: &str,
) -> Result<Option<&'a Skill>, String> {
if registry.find(name).is_some() {
return Ok(None);
}
let matching = skills
.iter()
.filter(|skill| skill.name == name)
.collect::<Vec<_>>();
if matching.is_empty() {
return Ok(None);
}
let enabled = matching
.iter()
.copied()
.filter(|skill| !skill.disable_model_invocation)
.collect::<Vec<_>>();
match enabled.as_slice() {
[skill] => Ok(Some(*skill)),
[] => Err(format!(
"skill '{name}' is disabled; run /skills enable {name} [source] or /skills to list loaded skills"
)),
_ => Err(format!(
"multiple enabled skills named '{name}'; use /skill {name} after resolving the source with /skills show {name} [source]"
)),
}
}
fn run_skill_shortcut(
name: &str,
argv: &[String],
registry: &Registry,
ctx: &CommandCtx<'_>,
) -> Option<CommandOutcome> {
match resolve_skill_shortcut(&ctx.harness.skills(), registry, name) {
Ok(Some(skill)) => {
if argv.is_empty() {
Some(CommandOutcome::AttachSkill {
name: skill.name.clone(),
})
} else {
Some(CommandOutcome::RunAgentPrompt {
prompt: attach_skill_prompt(argv.join(" "), Some(&skill.name)),
error_context: "skill command failed: ",
})
}
}
Ok(None) => None,
Err(e) => Some(CommandOutcome::Error(e)),
}
}
pub async fn dispatch(input: &str, registry: &Registry, ctx: &CommandCtx<'_>) -> CommandOutcome {
registry
.output
.scope(dispatch_impl(input, registry, ctx))
.await
}
pub async fn dispatch_with_output(
input: &str,
registry: &Registry,
ctx: &CommandCtx<'_>,
output: CommandOutput,
) -> CommandOutcome {
output.scope(dispatch_impl(input, registry, ctx)).await
}
async fn dispatch_impl(input: &str, registry: &Registry, ctx: &CommandCtx<'_>) -> CommandOutcome {
let (name, argv) = match parse(input) {
Some(parts) => parts,
None => return CommandOutcome::Error("not a slash command".into()),
};
if name == "help" {
print_help_with_skills(
registry,
argv.first().map(String::as_str),
&ctx.harness.skills(),
);
return CommandOutcome::Handled;
}
if name == "reload" {
return reload_everything(registry, ctx).await;
}
let Some(cmd) = registry.find(&name) else {
if let Some(file_cmd) = registry
.file_commands()
.into_iter()
.find(|fc| fc.name == name)
{
let args_tail = args_tail_of(input, &name);
return CommandOutcome::RunAgentPrompt {
prompt: crate::file_commands::expand_file_command(&file_cmd, &args_tail),
error_context: "",
};
}
return run_skill_shortcut(&name, &argv, registry, ctx).unwrap_or_else(|| {
CommandOutcome::RunAgentPrompt {
prompt: input.to_string(),
error_context: "",
}
});
};
let extra = DaemonCtx {
harness: ctx.harness.clone(),
trigger_executor: ctx.trigger_executor.clone(),
storage: registry.storage.clone(),
dynamic_triggers: registry.dynamic_triggers.clone(),
cron: registry.cron.clone(),
inherit_slot: ctx.inherit_slot.clone(),
collapse_unload_slot: ctx.collapse_unload_slot.clone(),
};
let sdk_ctx = theway_transport::commands::CommandCtx {
session_id: ctx.session_id,
log_path: ctx.log_path,
tool_count: ctx.tool_count,
cwd: ctx.cwd,
extra: &extra,
};
cmd.run(&argv, &sdk_ctx).await
}
fn args_tail_of(input: &str, name: &str) -> String {
let trimmed = input.trim();
let skip = 1 + name.len();
if trimmed.len() <= skip {
return String::new();
}
trimmed[skip..].trim_start().to_string()
}
async fn reload_everything(registry: &Registry, ctx: &CommandCtx<'_>) -> CommandOutcome {
let scanned = crate::file_commands::scan_file_commands(ctx.cwd, registry.user_home());
let count = scanned.len();
registry.set_file_commands(scanned.clone());
cprintln!("reloaded commands: {count} file command(s)");
for fc in &scanned {
if fc.description.is_empty() {
cprintln!(" - /{}", fc.name);
} else {
cprintln!(" - /{} — {}", fc.name, fc.description);
}
}
match ctx.harness.reload_skills_from_disk().await {
Ok(out) => {
cprintln!(
"reloaded skills: {} loaded, {} diagnostics",
out.skills.len(),
out.diagnostics.len()
);
}
Err(e) => {
return CommandOutcome::Error(format!("reload skills failed: {e}"));
}
}
if let (Some(slot), Some(auth_base)) = (ctx.mcp_provision, ctx.auth_base) {
let old_tools = slot.read().unwrap().tools.clone();
let configs = slot.read().unwrap().configs.clone();
let result =
crate::mcp_loader::connect_servers(&configs, ctx.cwd, &auth_base.join("auth.json"))
.await;
let (new_tools, new_hooks) = {
let mut slot_state = slot.write().unwrap();
slot_state.replace_connection_result(configs, result);
(slot_state.tools.clone(), slot_state.hooks.clone())
};
ctx.harness.replace_mcp_tools(&old_tools, new_tools.clone());
let registered = new_hooks.len();
{
use crate::trigger_engine::notification_hook::NotificationHook;
use crate::trigger_engine::notification_hook::NotificationHookSink;
let mut slot_state = slot.write().unwrap();
for hook in &new_hooks {
let label = hook.label().to_string();
if slot_state.registered_labels.insert(label) {
ctx.trigger_executor.register(hook.clone());
}
}
}
cprintln!(
"reconnected mcp servers: {} connected, {} hook(s) registered",
slot.read().unwrap().server_names.len(),
registered
);
}
CommandOutcome::Handled
}
#[cfg(test)]
tests_bridge_macro::tests_bridge!("commands");