use std::sync::Arc;
use super::actor::{Live, SessionActor};
use super::types::*;
fn compact_request(cmd: &str, arg: &str) -> Option<crate::engine::commands::CommandResult> {
match crate::engine::commands::evaluate_engine_command(cmd, arg) {
Some(r @ crate::engine::commands::CommandResult::Compact { .. }) => Some(r),
_ => None,
}
}
pub fn engine_command_reply(
cmd: &str,
arg: &str,
runtime: &mut crate::Runtime,
session: &mut crate::Session,
) -> (serde_json::Value, bool) {
use crate::engine::commands::{handle_engine_command, CommandResult};
match handle_engine_command(cmd, arg, runtime) {
None => (serde_json::json!({ "kind": "unhandled" }), false),
Some(CommandResult::Quit) => (serde_json::json!({ "kind": "quit" }), false),
Some(CommandResult::ModelChanged {
model,
reasoning_clamped,
}) => {
session.model = runtime.model().to_string();
let mut text = format!("model → {}", model);
let mut clamp = serde_json::Value::Null;
if let Some(c) = reasoning_clamped {
session.thinking_level = runtime.thinking_level().to_string();
text.push_str(&format!(
"\nthinking → {} (clamped from {}: not supported by {})",
c.to.as_str(),
c.from.as_str(),
runtime.model()
));
clamp = serde_json::json!({ "from": c.from.as_str(), "to": c.to.as_str() });
}
(
serde_json::json!({
"kind": "notice",
"event": "model_changed",
"text": text,
"model": runtime.model(),
"clamp": clamp,
}),
true,
)
}
Some(CommandResult::ThinkingChanged { spec }) => {
session.thinking_level = spec.config_value();
(
serde_json::json!({
"kind": "notice",
"event": "thinking_changed",
"text": format!("thinking → {}", spec.level()),
"level": spec.level().to_string(),
"config_value": spec.config_value(),
}),
true,
)
}
Some(CommandResult::Compact { .. }) => (serde_json::json!({ "kind": "none" }), false),
Some(CommandResult::Error(e)) => (serde_json::json!({ "kind": "error", "text": e }), false),
Some(CommandResult::Output(text)) => {
(serde_json::json!({ "kind": "output", "text": text }), false)
}
Some(_) => (serde_json::json!({ "kind": "none" }), false),
}
}
pub async fn auth_command_reply(arg: &str, runtime: &crate::Runtime) -> serde_json::Value {
match arg.trim() {
"reload" => match runtime.reload_auth().await {
Ok(text) => serde_json::json!({ "kind": "output", "text": text }),
Err(e) => serde_json::json!({
"kind": "error",
"text": format!("auth reload failed: {e}"),
}),
},
_ => serde_json::json!({ "kind": "error", "text": "usage: /auth reload" }),
}
}
impl SessionActor {
pub(crate) async fn engine_command(&mut self, id: u64, cmd: String, arg: String) {
use crate::engine::commands::CommandResult;
if cmd == "saveas" {
let value = self.session_name(&arg).await;
self.emit(SessionEventWire::QueryResult { id, value });
return;
}
if cmd == "keepwarm" {
let value = self.cache_keepwarm_command(&arg).await;
self.emit(SessionEventWire::QueryResult { id, value });
return;
}
if cmd == "auth" {
let value = auth_command_reply(&arg, &self.runtime).await;
self.emit(SessionEventWire::QueryResult { id, value });
return;
}
crate::engine::commands::prepare_engine_command(&cmd, &arg, &self.runtime).await;
let cancelling = cmd == "workers" && arg.trim_start().starts_with("cancel");
let uncancelled = if cancelling { self.uncancelled_workers() } else { Vec::new() };
let (value, view_changed) =
engine_command_reply(&cmd, &arg, &mut self.runtime, &mut self.conv.session);
if cancelling {
let after = self
.runtime
.subagent_registry()
.lock()
.unwrap_or_else(|p| p.into_inner())
.display_rows();
for (id, agent) in &cancelled_by_command(uncancelled, &after) {
self.note_in_main_conversation(super::actor::subagent_user_cancel_event(id, agent));
}
}
if cmd == "memory" {
self.note_memory_choice(&arg);
self.observe_memory();
}
if view_changed {
self.publish_view().await;
let view = (**self.view.load()).clone();
self.emit(SessionEventWire::SettingChanged(SettingApplied {
id: super::wire::VIEW_CHANGED_SETTING_ID,
setting: cmd.clone(),
ok: true,
message: None,
view,
clamp: None,
}));
}
if let Some(CommandResult::Compact {
custom_instructions,
}) = compact_request(&cmd, &arg)
{
self.compact(custom_instructions, "manual").await;
}
self.emit(SessionEventWire::QueryResult { id, value });
}
pub(crate) fn effective_memory_mode(&self) -> Option<super::SessionMemoryMode> {
use crate::runtime::memory_context::DurableStatus;
let recorded = self.config.memory_mode;
if !self.runtime.is_live() {
return recorded;
}
let off = matches!(self.runtime.memory_context_status().durable, DurableStatus::Off);
let would_grant = match recorded {
Some(mode) => mode.lease_mode().is_some(),
None => self.memory_default_is_durable(),
};
if off && would_grant && self.memory_seen_active {
Some(super::SessionMemoryMode::Off)
} else {
recorded
}
}
fn memory_default_is_durable(&self) -> bool {
let memory = &self.host.config().memory;
memory.default_mode_confirmed
&& matches!(
memory.default_mode.as_str(),
"recall_each_prompt" | "capture_only" | "capture_and_recall"
)
}
pub(crate) fn observe_memory(&mut self) {
use crate::runtime::memory_context::DurableStatus;
if self.runtime.is_live()
&& matches!(
self.runtime.memory_context_status().durable,
DurableStatus::Active { .. }
)
{
self.memory_seen_active = true;
}
}
pub(crate) fn sync_memory_choice(&mut self) {
self.observe_memory();
let mode = self.effective_memory_mode();
if mode != self.config.memory_mode {
self.config.memory_mode = mode;
*self
.memory_mode_slot
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = mode;
if mode == Some(super::SessionMemoryMode::Off) {
self.emit(SessionEventWire::SystemNotice(
"memory stays off for this session: its consent was revoked, and a park or daemon reload keeps it off (/memory on turns it back on)"
.into(),
));
}
}
}
pub(crate) fn close_command(&mut self, id: u64, arg: &str) -> Option<EndReason> {
let reply = |text: String, kind: &str| SessionEventWire::QueryResult {
id,
value: serde_json::json!({ "kind": kind, "text": text }),
};
let force = match arg.trim() {
"" => false,
"force" => true,
_ => {
self.emit(reply("usage: /close [force]".into(), "error"));
return None;
}
};
let live = self.runtime.is_live();
let workers: Vec<(String, String)> = if live {
let mut running: Vec<(String, String)> = self
.runtime
.subagent_registry()
.lock()
.unwrap_or_else(|p| p.into_inner())
.display_rows()
.into_iter()
.filter(|r| {
r.status == crate::runtime::subagent::SubagentStatus::Running
&& !r.cancel_requested
})
.map(|r| (r.handle_id, r.agent_name))
.collect();
running.sort();
running
} else {
Vec::new()
};
let mut blockers = Vec::new();
if self.streaming {
blockers.push("a turn is running".to_string());
}
if self.compact.is_some() {
blockers.push("a compaction is running".to_string());
}
if !workers.is_empty() {
let names: Vec<String> = workers.iter().map(|(id, agent)| format!("{id} {agent}")).collect();
blockers.push(format!(
"{} worker{} running ({})",
workers.len(),
if workers.len() == 1 { " is" } else { "s are" },
names.join(", ")
));
}
if self.driver.is_some() {
blockers.push("an /auto run is armed".to_string());
}
if !blockers.is_empty() && !force {
self.emit(reply(
format!(
"not closed: {}. Closing the session would stop {}. Wait, or use /close force.",
blockers.join("; "),
if blockers.len() == 1 { "it" } else { "them" },
),
"error",
));
return None;
}
if !workers.is_empty() {
let cancelled = self
.runtime
.subagent_registry()
.lock()
.unwrap_or_else(|p| p.into_inner())
.cancel_running(crate::runtime::subagent::WorkerCancelScope::All);
for (handle_id, agent) in workers.iter().filter(|(h, _)| cancelled.contains(h)) {
self.note_in_main_conversation(super::actor::subagent_close_cancel_event(handle_id, agent));
}
}
let session = (**self.journal_id.load()).clone();
let text = if self.conv.is_live() && self.conv.api_messages.is_empty() {
format!("closing session {session} (it never ran a turn, so there is nothing to resume)")
} else {
format!("closing session {session}; resume it with: synaps --attach --continue {session}")
};
self.emit(reply(text, "output"));
Some(EndReason::ClientQuit)
}
fn uncancelled_workers(&self) -> Vec<(String, String)> {
let mut workers: Vec<(String, String)> = self
.runtime
.subagent_registry()
.lock()
.unwrap_or_else(|p| p.into_inner())
.display_rows()
.into_iter()
.filter(|r| !r.cancel_requested)
.map(|r| (r.handle_id, r.agent_name))
.collect();
workers.sort();
workers
}
fn note_memory_choice(&mut self, arg: &str) {
use crate::runtime::memory_context::DurableStatus;
let Some(choice) = super::SessionMemoryMode::from_command_arg(arg) else {
return;
};
let took = match (self.runtime.memory_context_status().durable, choice.lease_mode()) {
(DurableStatus::Off, None) => true,
(DurableStatus::Active { mode, .. }, Some(want)) => mode == want,
_ => false,
};
if took {
self.config.memory_mode = Some(choice);
*self
.memory_mode_slot
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some(choice);
}
}
async fn session_name(&mut self, arg: &str) -> serde_json::Value {
let trimmed = arg.trim();
let Some(_guard) = name_lock().await else {
return serde_json::json!({ "kind": "error", "text": NAME_LOCK_BUSY });
};
self.persister.flush().await;
if trimmed.is_empty() {
self.conv.session.clear_name();
let _ = self.conv.session.save().await;
self.sync_name();
self.emit_conversation();
return serde_json::json!({
"kind": "output", "text": "session name cleared", "name": serde_json::Value::Null,
});
}
match self.conv.session.set_name(trimmed) {
Ok(()) => {
let _ = self.conv.session.save().await;
self.sync_name();
self.emit_conversation();
serde_json::json!({
"kind": "output",
"text": format!("session named '{}'", trimmed),
"name": trimmed,
})
}
Err(e) => serde_json::json!({ "kind": "error", "text": e.to_string() }),
}
}
pub(crate) async fn submit_prepared(
&mut self,
messages: Vec<crate::SharedMessage>,
user_text: Option<String>,
) {
if self.streaming {
self.emit(SessionEventWire::SystemNotice(
"cannot load a skill while streaming".into(),
));
return;
}
self.conv.api_messages.extend(messages);
if let Some(text) = user_text {
self.conv
.api_messages
.push(Arc::new(serde_json::json!({"role": "user", "content": text})));
}
self.start_turn(TurnTrigger::PluginCommand, None).await;
}
pub(crate) async fn plugin_command(&mut self, id: u64, plugin: String, name: String, arg: String) {
let value = match Self::find_plugin_command(&plugin, &name) {
None => serde_json::json!({
"kind": "error",
"text": format!("unknown plugin command /{plugin}:{name}"),
}),
Some(cmd) => match crate::skills::commands::execute_plugin_command_with_tools(
&cmd,
&arg,
self.runtime.tools_shared(),
)
.await
{
Ok(output) => serde_json::json!({
"kind": "plugin_output",
"status": output.status,
"stdout": output.stdout,
"stderr": output.stderr,
}),
Err(e) => serde_json::json!({ "kind": "error", "text": e.to_string() }),
},
};
self.emit(SessionEventWire::QueryResult { id, value });
}
fn find_plugin_command(
plugin: &str,
name: &str,
) -> Option<Arc<crate::skills::registry::RegisteredPluginCommand>> {
let host = crate::host::EngineHost::current()?;
match host.command_registry().resolve(&format!("{plugin}:{name}")) {
crate::skills::registry::Resolution::PluginCommand(c) => Some(c),
_ => None,
}
}
pub(crate) async fn resume(&mut self, id: u64, query: String) {
if self.streaming {
self.emit(SessionEventWire::QueryResult {
id,
value: serde_json::json!({ "kind": "error", "text": "cannot resume while streaming" }),
});
return;
}
self.persister.flush().await;
let session = match crate::resolve_session(&query) {
Ok(s) => s,
Err(e) => {
self.emit(SessionEventWire::QueryResult {
id,
value: serde_json::json!({ "kind": "error", "text": e.to_string() }),
});
return;
}
};
let old_id = self.conv.session.id.clone();
let new_id = session.id.clone();
let new_lock = if new_id == old_id {
None } else {
match Self::try_lock_session(&new_id) {
Ok(lock) => Some(lock),
Err(e @ agent_core::session_lock::SessionLockError::Held { .. })
| Err(e @ agent_core::session_lock::SessionLockError::CompactedInto { .. }) => {
self.emit(SessionEventWire::QueryResult {
id,
value: serde_json::json!({
"kind": "error",
"text": format!("cannot resume: {e}"),
}),
});
return;
}
Err(e) => {
tracing::warn!(session = %new_id, "resume: session lock: {e}");
None
}
}
};
self.cache_keepwarm_conversation_replaced();
self.runtime.set_model(session.model.clone());
let clamp_notice = self
.runtime
.restore_session_reasoning(&session.thinking_level)
.map(|clamp| {
format!(
"thinking → {} (clamped from {}: not supported by {})",
clamp.to.as_str(),
clamp.from.as_str(),
self.runtime.model()
)
});
if let Some(ref sp) = session.system_prompt {
self.runtime.set_system_prompt(sp.clone());
}
self.save().await;
let via = if crate::chain::load_chain(&query).is_ok() {
Some(format!("chain '{}'", query))
} else if crate::find_session_by_name(&query).is_ok() {
Some(format!("name '{}'", query))
} else {
None
};
if new_id != old_id {
self.session_lock = new_lock;
}
self.conv = Live::new(crate::engine::session::ConversationState::from_resumed(session));
if self.config.persist && self.session_lock.is_some() {
crate::engine::setup::recover_turn_draft(&mut self.conv).await;
}
if clamp_notice.is_some() {
self.conv.session.thinking_level = self.runtime.thinking_level().to_string();
}
self.consecutive_auto_turns = 0;
self.runtime.begin_conversation(Some(new_id.clone()));
self.set_journal_id(new_id.clone());
self.publish_view().await;
self.emit(SessionEventWire::Resumed {
id,
old_id,
new_id,
via,
clamp_notice,
});
self.emit_conversation();
}
#[allow(dead_code)]
pub(crate) fn emit_subagent_rows(&mut self) {
let rows = self
.runtime
.subagent_registry()
.lock()
.unwrap_or_else(|p| p.into_inner())
.display_rows();
self.emit(SessionEventWire::SubagentRows(rows));
}
#[allow(dead_code)]
pub(crate) fn has_live_subagents(&self) -> bool {
self.runtime
.subagent_registry()
.lock()
.unwrap_or_else(|p| p.into_inner())
.display_rows()
.iter()
.any(|r| matches!(r.status, crate::runtime::subagent::SubagentStatus::Running))
}
}
fn cancelled_by_command(
before: Vec<(String, String)>,
after: &[crate::runtime::subagent::SubagentDisplayRow],
) -> Vec<(String, String)> {
before
.into_iter()
.filter(|(id, _)| after.iter().any(|r| &r.handle_id == id && r.cancel_requested))
.collect()
}
#[cfg(test)]
mod tests {
use crate::Runtime;
fn row(id: &str, cancel_requested: bool) -> crate::runtime::subagent::SubagentDisplayRow {
serde_json::from_value(serde_json::json!({
"subagent_id": 0, "agent_name": "a", "status": "running",
"cancel_requested": cancel_requested, "elapsed_secs": 0.0,
"finished_elapsed": null, "handle_id": id,
}))
.unwrap()
}
#[test]
fn the_cancel_note_names_only_workers_the_command_flagged() {
let before: Vec<(String, String)> = ["sa_1", "sa_2", "sa_3"]
.iter()
.map(|id| (id.to_string(), "a".to_string()))
.collect();
let after = [row("sa_1", true), row("sa_3", false)];
let named: Vec<String> =
super::cancelled_by_command(before, &after).into_iter().map(|(id, _)| id).collect();
assert_eq!(named, ["sa_1"]);
}
#[tokio::test]
async fn restore_session_reasoning_keeps_explicit_provenance() {
let mut runtime = Runtime::new().await.unwrap();
runtime.set_model("openai-codex/gpt-5.6-sol".to_string());
let clamp = runtime.restore_session_reasoning("ultra");
assert!(clamp.is_none());
assert_eq!(
runtime.reasoning_level(),
agent_core::reasoning::ReasoningLevel::Ultra
);
assert!(runtime.is_reasoning_explicit());
runtime.set_model("openai-codex/gpt-5.6-terra".to_string());
assert_eq!(
runtime.reasoning_level(),
agent_core::reasoning::ReasoningLevel::Ultra
);
}
#[tokio::test]
async fn restore_session_reasoning_clamps_unsupported_level() {
let mut runtime = Runtime::new().await.unwrap();
runtime.set_model("xai-auth/grok-4.6".to_string());
let clamp = runtime.restore_session_reasoning("xhigh");
assert!(clamp.is_some());
assert_eq!(
runtime.reasoning_level(),
agent_core::reasoning::ReasoningLevel::High
);
assert!(runtime.is_reasoning_explicit());
}
}
#[cfg(test)]
mod auth_command_tests {
use super::auth_command_reply;
#[tokio::test]
async fn auth_command_usage_and_token_free_error() {
let rt = crate::Runtime::new_headless(); let v = auth_command_reply("bogus", &rt).await;
assert_eq!(v["kind"], "error");
assert!(v["text"].as_str().unwrap().contains("usage: /auth reload"));
let v = auth_command_reply(" reload ", &rt).await;
let t = v["text"].as_str().unwrap();
assert_eq!(v["kind"], "error");
assert!(t.contains("nothing to reload"), "{t}");
assert!(!t.contains("test-token"), "{t}");
}
#[test]
fn auth_is_a_builtin_command() {
assert!(crate::skills::BUILTIN_COMMANDS.contains(&"auth"));
}
}
static NAME_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
pub const NAME_LOCK_WAIT: std::time::Duration = std::time::Duration::from_secs(3);
async fn name_lock() -> Option<tokio::sync::MutexGuard<'static, ()>> {
tokio::time::timeout(NAME_LOCK_WAIT, NAME_LOCK.lock()).await.ok()
}
const NAME_LOCK_BUSY: &str = "another session is saving a rename; retry in a moment";
#[doc(hidden)]
pub async fn hold_name_lock_for_tests() -> tokio::sync::MutexGuard<'static, ()> {
NAME_LOCK.lock().await
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub(crate) struct RenameArgs {
pub name: Option<String>,
#[serde(
default,
deserialize_with = "crate::session::wire::present_option",
skip_serializing_if = "Option::is_none"
)]
pub expect: Option<Option<String>>,
}
fn rename_error(code: &str, text: impl Into<String>) -> serde_json::Value {
serde_json::json!({ "kind": "error", "code": code, "text": text.into() })
}
fn rename_ok(name: &Option<String>, in_memory_only: bool) -> serde_json::Value {
let text = match (name, in_memory_only) {
(Some(n), false) => format!("session named '{n}'"),
(None, false) => "session name cleared".to_string(),
(Some(n), true) => format!("session named '{n}' (in memory only: this session is not saved)"),
(None, true) => "session name cleared (in memory only: this session is not saved)".to_string(),
};
serde_json::json!({ "kind": "output", "text": text, "name": name, "in_memory_only": in_memory_only })
}
fn name_error_reply(e: agent_core::session::NameError) -> serde_json::Value {
let mut v = rename_error(e.code(), e.to_string());
if let agent_core::session::NameError::Taken { by, .. } = &e {
v["taken_by"] = serde_json::json!(by);
}
if matches!(e, agent_core::session::NameError::Io(_)) {
v["code"] = serde_json::json!("failed");
}
v
}
impl SessionActor {
pub(crate) async fn host_rename(&mut self, arg: &str) -> serde_json::Value {
let Ok(args) = serde_json::from_str::<RenameArgs>(arg) else {
return rename_error(
"failed",
"rename is the daemon's control command (`synaps daemon rename`); use /saveas",
);
};
let want = args.name.as_deref().map(str::trim).filter(|n| !n.is_empty()).map(str::to_string);
if let Some(n) = &want {
if let Err(e) = agent_core::session::validate_name(n) {
return rename_error("invalid", e);
}
}
let Some(_guard) = name_lock().await else {
return rename_error("busy", NAME_LOCK_BUSY);
};
let current: Option<String> =
if self.is_parked() { (**self.name.load()).clone() } else { self.conv.session.name.clone() };
if let Some(expected) = &args.expect {
if *expected != current {
let text = match (¤t, expected) {
(Some(c), None) => format!("session is named '{c}'; rename never replaces a name without --force"),
(Some(c), Some(e)) => format!("session is named '{c}', not '{e}'; not renamed"),
(None, _) => "session has no name any more; not renamed".to_string(),
};
let mut v = rename_error("named", text);
v["current_name"] = serde_json::json!(current);
return v;
}
}
if self.is_parked() {
self.rename_parked(want, args.expect).await
} else if want == current {
rename_ok(&want, !self.config.persist)
} else {
self.rename_live(want).await
}
}
async fn rename_live(&mut self, want: Option<String>) -> serde_json::Value {
if let Some(n) = &want {
if let Err(e) = self.conv.session.check_name(n) {
return name_error_reply(e);
}
}
if !self.config.persist {
self.conv.session.name = want.clone();
self.sync_name();
self.emit_conversation();
return rename_ok(&want, true);
}
if self.session_lock.is_none() {
return rename_error(
"locked",
"this daemon does not hold the session's journal lock (it may be open in another process); not renamed",
);
}
self.persister.flush().await;
let dir = agent_core::session_lock::sessions_dir();
let id = self.conv.session.id.clone();
let mtimes = agent_core::session_catalog::SessionMtimes::capture(&dir, &id);
let old = std::mem::replace(&mut self.conv.session.name, want.clone());
if let Err(e) = self.conv.session.save_durable().await {
self.conv.session.name = old;
return rename_error("failed", format!("rename not saved (rolled back): {e}"));
}
mtimes.restore(&dir, &id);
self.sync_name();
self.emit_conversation();
rename_ok(&want, false)
}
async fn rename_parked(&mut self, want: Option<String>, expect: Option<Option<String>>) -> serde_json::Value {
use agent_core::session_catalog::RenameError;
let dir = agent_core::session_lock::sessions_dir();
let id = (**self.journal_id.load()).clone();
let name = want.clone();
let res = tokio::task::spawn_blocking(move || {
let expect = expect.as_ref().map(|e| e.as_deref());
agent_core::session_catalog::rename_past_session_if(&dir, &id, name.as_deref(), expect)
})
.await;
match res {
Ok(Ok(())) => {}
Ok(Err(e)) => {
let code = match &e {
RenameError::InvalidName(_) => "invalid",
RenameError::NameTaken { .. } => "taken",
RenameError::ChainConflict => "chain_conflict",
RenameError::Live { .. } => "locked",
RenameError::NameMismatch { .. } => "named",
_ => "failed",
};
let mut v = rename_error(code, e.to_string());
if let RenameError::NameTaken { other_id } = &e {
v["taken_by"] = serde_json::json!(other_id);
}
if let RenameError::NameMismatch { current } = e {
v["current_name"] = serde_json::json!(current);
self.mirror_name(current);
}
return v;
}
Err(e) => return rename_error("failed", format!("rename failed: {e}")),
}
self.mirror_name(want.clone());
rename_ok(&want, false)
}
}