mod admin;
mod api;
mod backend;
mod config;
pub mod daemon_protocol;
mod hooks;
mod nostr_transport;
mod persistence;
mod project_identity;
mod project_index;
mod protocol;
mod rollover;
mod router;
mod scheduler;
mod server;
mod session_agent;
mod start_outcome;
mod state;
mod tmux;
mod tmux_var;
mod transport;
use anyhow::Context;
use backend::CodingAssistant;
use clap::{Parser, Subcommand, ValueEnum};
use daemon_protocol::IdlePolicy;
use nostr_sdk::ToBech32;
use std::io::Read;
use std::path::PathBuf;
#[derive(Parser)]
#[command(name = "ouija", about = "Cross-machine AI session daemon")]
struct Cli {
#[command(subcommand)]
command: Command,
}
#[derive(Subcommand)]
enum Command {
#[command(name = "start-server")]
StartServer {
#[arg(short, long, default_value = "7880")]
port: u16,
#[arg(short, long)]
name: Option<String>,
#[arg(long)]
data: Option<String>,
#[arg(long)]
ticket: Option<String>,
#[arg(long = "relay")]
relays: Vec<String>,
},
Status,
Nodes,
Ticket {
#[arg(long = "relay")]
relays: Vec<String>,
},
RegenerateTicket {
#[arg(long)]
yes: bool,
},
Connect {
ticket: String,
#[arg(long)]
name: Option<String>,
},
Disconnect {
node: String,
},
Register {
id: String,
pane: Option<String>,
#[arg(long)]
vim_mode: bool,
#[arg(long)]
project_dir: Option<String>,
#[arg(long)]
role: Option<String>,
},
Ask {
to: String,
#[arg(long)]
message_file: PathBuf,
#[arg(long)]
from: Option<String>,
},
Tell {
to: String,
#[arg(long)]
message_file: PathBuf,
#[arg(long)]
reply_to: Option<u64>,
#[arg(long)]
from: Option<String>,
},
Reply {
to: String,
msg_id: u64,
#[arg(long)]
message_file: PathBuf,
#[arg(long)]
no_done: bool,
#[arg(long)]
expect_reply: bool,
#[arg(long)]
from: Option<String>,
},
Ls,
Whoami,
Claim { requested_id: String },
Dormant {
#[command(subcommand)]
command: DormantCommand,
},
#[command(name = "recover-backend-identity")]
RecoverBackendIdentity {
session_id: String,
},
Rollover {
#[command(subcommand)]
action: RolloverAction,
},
Announce {
#[arg(long)]
role: Option<String>,
#[arg(long)]
bulletin: Option<String>,
},
Inject { pane: String, message: String },
Rename {
new_id: String,
#[arg(long)]
from: Option<String>,
},
Unregister { id: String },
#[command(name = "spawn-session")]
SpawnSession {
name: String,
#[arg(long)]
project_dir: Option<String>,
#[arg(long)]
prompt_file: Option<PathBuf>,
#[arg(long, value_parser = parse_manual_reminder)]
reminder: Option<String>,
#[arg(long)]
parent_session: Option<String>,
#[arg(long)]
no_parent_session: bool,
#[arg(long, value_enum, conflicts_with = "idle_policy")]
when_done: Option<WhenDone>,
#[arg(
long,
value_parser = parse_idle_policy,
conflicts_with = "when_done"
)]
idle_policy: Option<IdlePolicy>,
#[arg(long)]
worktree: bool,
#[arg(long)]
branch: Option<String>,
#[arg(long)]
base_branch: Option<String>,
#[arg(long)]
model: Option<String>,
#[arg(long)]
effort: Option<String>,
#[arg(long, value_parser = parse_fresh_context_after_active)]
fresh_context_after_active: Option<u64>,
#[arg(long)]
backend: Option<String>,
#[arg(long)]
from: Option<String>,
#[arg(long)]
wait: bool,
#[arg(long, requires = "wait", value_parser = parse_fresh_context_after_active)]
wait_timeout: Option<u64>,
},
#[command(name = "start-status")]
StartStatus {
name: String,
#[arg(long)]
incarnation: String,
#[arg(long, value_parser = parse_fresh_context_after_active)]
wait: Option<u64>,
},
#[command(name = "kill-session")]
KillSession {
name: String,
#[arg(long, conflicts_with = "keep_worktree")]
delete_worktree: bool,
#[arg(long, hide = true, conflicts_with = "delete_worktree")]
keep_worktree: bool,
},
#[command(name = "recover-lease")]
RecoverLease { name: String },
#[command(name = "prune-stale-sessions")]
PruneStaleSessions {
#[arg(long, short)]
yes: bool,
},
#[command(name = "restart-session")]
RestartSession {
name: String,
#[arg(long)]
fresh: bool,
#[arg(
long,
requires = "fresh",
value_parser = parse_fresh_context_after_active
)]
fresh_context_after_active: Option<u64>,
#[arg(long)]
prompt_file: Option<PathBuf>,
#[arg(long)]
suppress_stored_prompt: bool,
#[arg(long)]
one_shot_file: Option<PathBuf>,
#[arg(long, value_parser = parse_manual_reminder)]
reminder: Option<String>,
#[arg(long)]
backend: Option<String>,
#[arg(long)]
model: Option<String>,
#[arg(long)]
effort: Option<String>,
},
#[command(name = "clear-reminder")]
ClearReminder {
clearing_id: u64,
#[arg(long)]
from: Option<String>,
},
Pending,
#[command(name = "clear-reply")]
ClearReply { sender_id: String },
#[command(name = "stop-server")]
StopServer,
#[command(name = "restart-server")]
RestartServer,
LogPath {
#[arg(long)]
data: Option<String>,
},
#[command(name = "self-update")]
SelfUpdate,
Config {
#[command(subcommand)]
action: Option<ConfigAction>,
},
Task {
#[command(subcommand)]
action: TaskAction,
},
}
#[derive(Subcommand)]
enum DormantCommand {
List,
Show { id: String },
Resume {
id: String,
#[arg(long)]
wait: bool,
#[arg(long, requires = "wait", value_parser = parse_fresh_context_after_active)]
wait_timeout: Option<u64>,
},
}
#[derive(Subcommand)]
enum RolloverAction {
Prepare {
#[arg(long, required = true)]
stdin: bool,
#[arg(long)]
replace_expired: bool,
},
Adopt { token: String },
Cleanup {
#[arg(long)]
force_pending: bool,
},
}
#[derive(Subcommand)]
enum ConfigAction {
Set { key: String, value: String },
SetCodexModelRoute {
alias: String,
#[arg(long)]
model: Option<String>,
#[arg(long)]
codex_home: Option<String>,
},
RemoveCodexModelRoute {
alias: String,
},
AddHuman {
#[arg(long)]
npub: String,
#[arg(long)]
name: String,
#[arg(long)]
default_session: Option<String>,
},
RemoveHuman {
#[arg(long)]
name: String,
},
ListHumans,
SetRouter {
#[arg(long)]
api_key: Option<String>,
#[arg(long)]
model: Option<String>,
#[arg(long)]
base_url: Option<String>,
},
RemoveRouter,
}
#[derive(Subcommand)]
enum TaskAction {
List,
Add {
name: String,
cron: String,
message: String,
#[arg(long)]
target: Option<String>,
#[arg(long, requires = "target")]
inject_only: bool,
#[arg(long)]
project_dir: Option<String>,
#[arg(long)]
backend: Option<String>,
#[arg(long)]
model: Option<String>,
#[arg(long)]
effort: Option<String>,
#[arg(long)]
once: bool,
},
Remove { id: String },
Enable { id: String },
Disable { id: String },
Runs {
#[arg(long)]
task: Option<String>,
},
Trigger { id: String },
}
#[derive(Clone, Debug, PartialEq, Eq, ValueEnum)]
enum WhenDone {
KeepOpen,
AskParent,
Close,
}
impl From<WhenDone> for IdlePolicy {
fn from(value: WhenDone) -> Self {
match value {
WhenDone::KeepOpen => IdlePolicy::KeepOpen,
WhenDone::AskParent => IdlePolicy::AskParentWhenDone,
WhenDone::Close => IdlePolicy::CloseWhenDone,
}
}
}
fn parse_idle_policy(value: &str) -> Result<IdlePolicy, String> {
value.parse()
}
fn parse_manual_reminder(value: &str) -> Result<String, String> {
daemon_protocol::validate_spawn_reminder(Some(value))?;
Ok(value.to_string())
}
fn parse_fresh_context_after_active(value: &str) -> Result<u64, String> {
let Some(unit) = value.chars().last() else {
return Err("duration must be a positive whole number followed by h, m, or s".into());
};
let multiplier = match unit {
'h' => 3_600,
'm' => 60,
's' => 1,
_ => return Err("duration must end with h, m, or s".into()),
};
let amount = &value[..value.len() - unit.len_utf8()];
if amount.is_empty() || !amount.bytes().all(|byte| byte.is_ascii_digit()) {
return Err("duration must be a positive whole number followed by h, m, or s".into());
}
let amount = amount
.parse::<u64>()
.map_err(|_| "duration value is too large".to_string())?;
if amount == 0 {
return Err("duration must be greater than zero".into());
}
amount
.checked_mul(multiplier)
.ok_or_else(|| "duration overflows seconds".to_string())
}
fn validate_spawn_lifecycle(
parent_session: Option<&str>,
no_parent_session: bool,
idle_policy: Option<&IdlePolicy>,
) -> Result<(), String> {
daemon_protocol::validate_spawn_lifecycle(parent_session, no_parent_session, idle_policy)
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
rustls::crypto::ring::default_provider()
.install_default()
.expect("failed to install rustls CryptoProvider");
let cli = Cli::parse();
if !matches!(cli.command, Command::StartServer { .. }) {
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "ouija=warn".parse().expect("valid default filter")),
)
.init();
}
match cli.command {
Command::StartServer {
port,
name,
data,
ticket,
relays,
} => {
let data_dir = match data.as_deref() {
Some(d) => std::path::PathBuf::from(d),
None => config::OuijaConfig::default_data_dir(),
};
std::fs::create_dir_all(&data_dir)?;
let log_file = open_daemon_log(&data_dir)?;
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "ouija=info".parse().expect("valid default filter")),
)
.with_writer(log_file)
.with_ansi(false)
.init();
preflight_checks();
let _ = backend::claude_code::ClaudeCode.install();
let _ = backend::opencode::OpenCode.install();
let _ = backend::codex::Codex.install();
let name = name.unwrap_or_else(|| {
hostname::get()
.ok()
.and_then(|h| h.into_string().ok())
.unwrap_or_else(|| "ouija".to_string())
});
let config_dir = match data.as_deref() {
Some(d) => std::path::PathBuf::from(d),
None => config::OuijaConfig::default_config_dir(),
};
std::fs::create_dir_all(&config_dir)?;
let nostr_keys = nostr_transport::load_or_create_keys(&config_dir)?;
let npub = nostr_keys
.public_key()
.to_bech32()
.unwrap_or_else(|_| "unknown".into());
tracing::info!("daemon identity: {npub}");
{
let registry = backend::BackendRegistry::default_registry();
let available = registry.available();
if available.is_empty() {
eprintln!(
"error: no coding backend found in PATH. Install claude-code, opencode, or codex.\n\
See: https://docs.anthropic.com/en/docs/claude-code\n\
See: https://opencode.ai\n\
See: https://developers.openai.com/codex"
);
std::process::exit(1);
}
tracing::info!("available backends: {}", available.join(", "));
}
let config = config::OuijaConfig::new(name, port, data, npub)?;
let state = state::AppState::new(config);
if let Some(home) = state.settings.read().await.codex_home.clone() {
backend::codex::install_configured_home(Some(&home));
}
{
let route_homes: Vec<String> = state
.settings
.read()
.await
.codex_model_routes
.values()
.filter_map(|route| route.codex_home.clone())
.collect();
for home in route_homes {
backend::codex::install_configured_home(Some(&home));
}
}
let index_state = state.clone();
tokio::spawn(async move {
project_index::refresh_index(&index_state).await;
});
scrub_leaked_tmux_session_environments();
restore_persisted_sessions(&state).await?;
register_human_sessions(&state).await;
let bg_state = state.clone();
tokio::spawn(async move {
setup_nostr_transport(&bg_state, ticket.as_deref(), relays).await;
});
let reaper_state = state.clone();
tokio::spawn(async move {
let mut last_session_hash: u64 = 0;
let mut first_run = true;
let mut heartbeat_counter: u64 = 0;
const HEARTBEAT_CYCLES: u64 = 6;
loop {
let interval = reaper_state.settings.read().await.reaper_interval_secs;
tokio::time::sleep(std::time::Duration::from_secs(interval)).await;
let panes_to_check: Vec<(
crate::daemon_protocol::ResourceOwner,
String,
Vec<String>,
)> = {
let proto = reaper_state.protocol.read().await;
let now = chrono::Utc::now().timestamp();
proto
.sessions
.values()
.filter(|s| {
matches!(s.origin, crate::daemon_protocol::Origin::Local)
&& s.pane.is_some()
&& (s.registered_at == 0 || now - s.registered_at > 60)
})
.filter_map(|s| {
Some((
s.owner(),
s.pane.clone()?,
reaper_process_names(&reaper_state.backends, &s.metadata)?,
))
})
.collect()
};
let dead_sessions: Vec<(crate::daemon_protocol::ResourceOwner, String)> =
if !panes_to_check.is_empty() {
let dead = tokio::task::spawn_blocking(move || {
panes_to_check
.into_iter()
.filter_map(|(owner, pane, names)| {
let name_refs: Vec<&str> =
names.iter().map(String::as_str).collect();
(!crate::tmux::pane_alive(&pane, &name_refs))
.then_some((owner, pane))
})
.collect::<Vec<_>>()
})
.await
.unwrap_or_default();
if !dead.is_empty() {
let observed_at = chrono::Utc::now().timestamp();
reap_dead_sessions(&reaper_state, dead.clone(), observed_at).await;
}
dead
} else {
vec![]
};
let perfire_to_check: Vec<(String, crate::state::PerFireWorktreeClaim)> = {
let pf = reaper_state.perfire_worktree_panes.read().await;
pf.iter().map(|(p, d)| (p.clone(), d.clone())).collect()
};
if !perfire_to_check.is_empty() {
let names: Vec<String> = reaper_state.backends.all_process_names();
let dead_perfire = tokio::task::spawn_blocking(move || {
let name_refs: Vec<&str> = names.iter().map(|s| s.as_str()).collect();
perfire_to_check
.into_iter()
.filter(|(pane, _)| !crate::tmux::pane_alive(pane, &name_refs))
.collect::<Vec<_>>()
})
.await
.unwrap_or_default();
if !dead_perfire.is_empty() {
for (pane_id, claim) in dead_perfire {
tracing::info!(
"per-fire worktree pane {pane_id} died, pruning worktrees in {}",
claim.project_dir
);
reaper_state
.prune_dead_perfire_worktree(&pane_id, &claim)
.await;
}
}
}
let _ = dead_sessions;
for (owner, pane) in reaper_state.collect_excess_idle_sessions().await {
let id = owner.session_id.clone();
tracing::info!(
"auto-closing idle session '{id}' (over max_local_sessions)"
);
crate::nostr_transport::kill_session_owned(&reaper_state, &owner, &pane)
.await;
}
reaper_state.scan_and_autoregister_panes().await;
heartbeat_counter += 1;
let current_hash = reaper_state.local_session_hash().await;
let heartbeat_due = heartbeat_counter >= HEARTBEAT_CYCLES;
if first_run || current_hash != last_session_hash || heartbeat_due {
if first_run || heartbeat_due {
reaper_state.sweep_worktree_presence().await;
}
transport::broadcast_local_sessions(&reaper_state).await;
last_session_hash = current_hash;
first_run = false;
if heartbeat_due {
heartbeat_counter = 0;
}
}
}
});
let scheduler_state = state.clone();
tokio::spawn(crate::scheduler::run_scheduler(scheduler_state));
server::run(state).await?;
}
Command::Status => {
cli_get("/api/status").await?;
}
Command::Nodes => {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}/api/nodes");
let resp: serde_json::Value = reqwest::get(&url).await?.json().await?;
let nodes = resp["nodes"].as_array();
match nodes {
Some(list) if !list.is_empty() => {
println!("{:<16} {:<12} {:<20} SINCE", "NAME", "STATUS", "NPUB");
for p in list {
let name = p["name"].as_str().unwrap_or("-");
let status = p["status"].as_str().unwrap_or("unknown");
let npub = p["npub"].as_str().unwrap_or("-");
let npub_short = if npub.len() > 20 {
format!("{}…{}", &npub[..10], &npub[npub.len() - 6..])
} else {
npub.to_string()
};
let since = p["since"].as_str().unwrap_or("-");
println!("{:<16} {:<12} {:<20} {}", name, status, npub_short, since);
}
}
_ => println!("no nodes"),
}
}
Command::Ticket { relays } => {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}/api/ticket");
let client = reqwest::Client::new();
let mut req = client.get(&url);
for r in &relays {
req = req.query(&[("relay", r.as_str())]);
}
let resp: serde_json::Value = req.send().await?.json().await?;
if let Some(ticket) = resp["ticket"].as_str() {
println!("{ticket}");
} else if let Some(err) = resp["error"].as_str() {
eprintln!("error: {err}");
std::process::exit(1);
} else {
println!("{}", serde_json::to_string_pretty(&resp)?);
}
}
Command::RegenerateTicket { yes } => {
if !yes {
eprintln!(
"WARNING: This will destroy your nostr identity (nsec). All nodes must re-connect."
);
eprintln!("Run with --yes to confirm.");
std::process::exit(1);
}
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}/api/regenerate-ticket?confirm=true");
let client = reqwest::Client::new();
let resp: serde_json::Value = client.post(&url).send().await?.json().await?;
if let Some(ticket) = resp["ticket"].as_str() {
println!("{ticket}");
} else if let Some(err) = resp["error"].as_str() {
eprintln!("Error: {err}");
std::process::exit(1);
}
}
Command::Connect { ticket, name } => {
let body = serde_json::json!({ "ticket": ticket, "name": name });
cli_post("/api/connect", &body).await?;
}
Command::Disconnect { node } => {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}/api/status");
let resp: serde_json::Value = reqwest::get(&url).await?.json().await?;
let daemon_id = resp["nodes"].as_array().and_then(|nodes| {
nodes.iter().find_map(|n| {
let name = n["name"].as_str().unwrap_or("");
let did = n["daemon_id"].as_str().unwrap_or("");
if name == node || did == node {
Some(did.to_string())
} else {
None
}
})
});
match daemon_id {
Some(id) => {
let body = serde_json::json!({ "daemon_id": id });
cli_post("/api/nodes/disconnect", &body).await?;
}
None => {
let body = serde_json::json!({ "daemon_id": node });
cli_post("/api/nodes/disconnect", &body).await?;
}
}
}
Command::Register {
id,
pane,
vim_mode,
project_dir,
role,
} => {
let pane = pane.or_else(|| std::env::var("TMUX_PANE").ok());
let body = serde_json::json!({
"id": id,
"pane": pane,
"vim_mode": vim_mode,
"project_dir": project_dir,
"role": role,
});
cli_post("/api/register", &body).await?;
}
Command::Ask {
to,
message_file,
from,
} => {
let message = read_message_file(&message_file)?;
let sender = resolve_sender(from).await?;
let body = serde_json::json!({
"from": sender.id,
"to": to,
"message": message,
"expects_reply": true,
"sender_ctx": sender.context,
});
cli_post("/api/send", &body).await?;
}
Command::Tell {
to,
message_file,
reply_to,
from,
} => {
let message = read_message_file(&message_file)?;
let sender = resolve_sender(from).await?;
let body = serde_json::json!({
"from": sender.id,
"to": to,
"message": message,
"expects_reply": false,
"responds_to": reply_to,
"sender_ctx": sender.context,
});
cli_post("/api/send", &body).await?;
}
Command::Reply {
to,
msg_id,
message_file,
no_done,
expect_reply,
from,
} => {
let message = read_message_file(&message_file)?;
let sender = resolve_sender(from).await?;
let body = serde_json::json!({
"from": sender.id,
"to": to,
"message": message,
"expects_reply": expect_reply,
"responds_to": msg_id,
"done": !no_done,
"sender_ctx": sender.context,
});
cli_post("/api/send", &body).await?;
}
Command::Ls => {
cli_list_sessions().await?;
}
Command::Whoami => {
cli_whoami().await?;
}
Command::Claim { requested_id } => {
let identity = backend::BackendRegistry::default_registry()
.caller_session_identity()
.context(
"claim requires exactly one complete backend identity from the current adapter",
)?;
let tmux_pane = std::env::var("TMUX_PANE")
.ok()
.filter(|pane| !pane.is_empty());
let pane_var = tmux_pane.as_deref().and_then(tmux_var::get);
let env_var = std::env::var("OUIJA_SESSION_ID")
.ok()
.filter(|id| !id.is_empty());
let request = claim_request(requested_id, tmux_pane, pane_var, env_var, identity);
cli_post(
"/api/session-identities/claim",
&serde_json::to_value(request)?,
)
.await?;
}
Command::Dormant { command } => match command {
DormantCommand::List => {
cli_get("/api/session-identities/dormant").await?;
}
DormantCommand::Show { id } => {
cli_get(&dormant_show_path(&id)).await?;
}
DormantCommand::Resume {
id,
wait,
wait_timeout,
} => {
let path = format!("{}/resume", dormant_show_path(&id));
let response = cli_post_json(&path, &serde_json::json!({})).await?;
println!("{response}");
if wait {
let ticket = start_ticket_from_spawn_response(&id, &response)?;
let timeout = std::time::Duration::from_secs(
wait_timeout.unwrap_or(DEFAULT_START_WAIT_SECS),
);
let report = await_start_outcome(&ticket, timeout).await?;
println!("{}", report.line);
if !report.started {
std::process::exit(1);
}
}
}
},
Command::RecoverBackendIdentity { session_id } => {
let identity = backend::BackendRegistry::default_registry()
.caller_session_identity()
.context(
"recovery requires exactly one complete backend identity from the current adapter",
)?;
let tmux_pane = std::env::var("TMUX_PANE")
.ok()
.filter(|pane| !pane.is_empty());
let pane_var = tmux_pane.as_deref().and_then(tmux_var::get);
let env_var = std::env::var("OUIJA_SESSION_ID")
.ok()
.filter(|id| !id.is_empty());
let caller = backend_recovery_caller_evidence(tmux_pane, pane_var, env_var);
let body = serde_json::json!({
"target_session_id": session_id,
"identity": identity,
"caller": caller,
});
cli_post("/api/backend-identities/recover", &body).await?;
}
Command::Rollover { action } => {
let caller = rollover_live_caller().await?;
let data_dir = config::OuijaConfig::default_data_dir();
let now = chrono::Utc::now().timestamp();
match action {
RolloverAction::Prepare {
stdin: _,
replace_expired,
} => {
let mut input = Vec::new();
std::io::Read::read_to_end(
&mut std::io::stdin().take(16 * 1024 + 1),
&mut input,
)
.context("failed to read continuation JSON from stdin")?;
let payload = rollover::parse_continuation(&input)?;
let token =
rollover::prepare(&data_dir, &caller, payload, replace_expired, now)?;
println!("{token}");
}
RolloverAction::Adopt { token } => {
let payload = rollover::adopt(&data_dir, &caller, &token, now)?;
println!("{}", serde_json::to_string_pretty(&payload)?);
}
RolloverAction::Cleanup { force_pending } => {
if rollover::cleanup(&data_dir, &caller, force_pending, now)? {
println!("continuation removed");
} else {
println!("no continuation for this session");
}
}
}
}
Command::Announce { role, bulletin } => {
if role.is_none() && bulletin.is_none() {
anyhow::bail!("at least one of --role or --bulletin is required");
}
let id = require_my_session_id().await?;
let body = serde_json::json!({
"id": id,
"role": role,
"bulletin": bulletin,
});
cli_post("/api/sessions/update", &body).await?;
}
Command::Inject { pane, message } => {
let body = serde_json::json!({ "pane": pane, "message": message });
cli_post("/api/inject", &body).await?;
}
Command::Rename { new_id, from } => {
let sender = resolve_sender(from).await?;
let body = serde_json::json!({
"old_id": sender.id,
"new_id": new_id,
"sender_ctx": sender.context,
});
cli_post("/api/rename", &body).await?;
}
Command::Unregister { id } => {
let body = serde_json::json!({ "id": id });
cli_post("/api/remove", &body).await?;
}
Command::SpawnSession {
name,
project_dir,
prompt_file,
reminder,
parent_session,
no_parent_session,
when_done,
idle_policy,
worktree,
branch,
base_branch,
model,
effort,
fresh_context_after_active,
backend,
from,
wait,
wait_timeout,
} => {
let prompt = prompt_file.as_deref().map(read_prompt_file).transpose()?;
let idle_policy = when_done.map(IdlePolicy::from).or(idle_policy);
if let Err(err) = validate_spawn_lifecycle(
parent_session.as_deref(),
no_parent_session,
idle_policy.as_ref(),
) {
anyhow::bail!("{err}");
}
let body = serde_json::json!({
"name": name,
"project_dir": project_dir,
"prompt": prompt,
"reminder": reminder,
"parent_session": parent_session,
"no_parent_session": no_parent_session,
"idle_policy": idle_policy,
"worktree": worktree,
"branch": branch,
"base_branch": base_branch,
"model": model,
"effort": effort,
"fresh_context_after_active_secs": fresh_context_after_active,
"backend": backend,
"from": from,
});
let response = cli_post_json("/api/sessions/start", &body).await?;
println!("{response}");
if wait {
let ticket = start_ticket_from_spawn_response(&name, &response)?;
let timeout =
std::time::Duration::from_secs(wait_timeout.unwrap_or(DEFAULT_START_WAIT_SECS));
let report = await_start_outcome(&ticket, timeout).await?;
println!("{}", report.line);
if !report.started {
std::process::exit(1);
}
}
}
Command::StartStatus {
name,
incarnation,
wait,
} => {
let ticket = StartTicket {
session: name,
incarnation,
};
let timeout = std::time::Duration::from_secs(wait.unwrap_or(0));
let report = await_start_outcome(&ticket, timeout).await?;
println!("{}", report.line);
if !report.started {
std::process::exit(1);
}
}
Command::KillSession {
name,
delete_worktree,
keep_worktree: _,
} => {
let body = serde_json::json!({
"name": name,
"delete_worktree": delete_worktree,
});
let response = cli_post_json("/api/sessions/kill", &body).await?;
println!("{}", serde_json::to_string(&response)?);
if !kill_response_succeeded(&response) {
std::process::exit(1);
}
}
Command::RecoverLease { name } => {
let body = serde_json::json!({ "name": name });
cli_post("/api/sessions/recover-lease", &body).await?;
}
Command::PruneStaleSessions { yes } => {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}/api/sessions/prune-stale");
let client = reqwest::Client::new();
let body_json = serde_json::json!({ "confirm": yes });
let mut resp = client.post(&url).json(&body_json).send().await?;
resp = resp.error_for_status()?;
let text = resp.text().await?;
let value: serde_json::Value = serde_json::from_str(&text)?;
let dry_run = value
.get("dry_run")
.and_then(|v| v.as_bool())
.ok_or_else(|| anyhow::anyhow!("server response missing 'dry_run' key: {text}"))?;
if dry_run == yes {
return Err(anyhow::anyhow!(
"server response intent mismatch: requested confirm={} but server returned dry_run={}. Response: {}",
yes,
dry_run,
text
));
} else if dry_run {
if let Some(arr) = value.get("would_prune").and_then(|v| v.as_array()) {
let ids = arr.len();
if ids == 0 {
println!("No stale sessions to prune");
} else {
println!(
"Would prune {} stale session(s): {}",
ids, value["would_prune"]
);
println!("Run with --yes to confirm removal");
}
} else {
return Err(anyhow::anyhow!(
"server response missing 'would_prune' key on dry_run=true: {text}"
));
}
} else {
let arr = value
.get("pruned")
.and_then(|v| v.as_array())
.ok_or_else(|| {
anyhow::anyhow!(
"server response missing 'pruned' key on confirm=true: {text}"
)
})?;
println!("Pruned {} stale session(s)", arr.len());
if value.get("errors").is_some() {
let err_arr =
value
.get("errors")
.and_then(|v| v.as_array())
.ok_or_else(|| {
anyhow::anyhow!(
"server response 'errors' key is not an array: {text}"
)
})?;
eprintln!(
"Failed to prune {} session(s): {}",
err_arr.len(),
value["errors"]
);
if !err_arr.is_empty() {
return Err(anyhow::anyhow!(
"partial failure: {} session(s) failed to prune",
err_arr.len()
));
}
}
if value.get("already_gone").is_some() {
let gone_arr = value
.get("already_gone")
.and_then(|v| v.as_array())
.ok_or_else(|| {
anyhow::anyhow!(
"server response 'already_gone' key is not an array: {text}"
)
})?;
if !gone_arr.is_empty() {
eprintln!(
"Skipped {} session(s) that vanished during prune: {}",
gone_arr.len(),
value["already_gone"]
);
}
}
}
}
Command::RestartSession {
name,
fresh,
fresh_context_after_active,
prompt_file,
suppress_stored_prompt,
one_shot_file,
reminder,
backend,
model,
effort,
} => {
let prompt = prompt_file.as_deref().map(read_prompt_file).transpose()?;
let one_shot_prompt = one_shot_file
.as_deref()
.map(read_one_shot_file)
.transpose()?;
let body = serde_json::json!({
"name": name,
"fresh": fresh,
"fresh_context_after_active_secs": fresh_context_after_active,
"prompt": prompt,
"suppress_stored_prompt": suppress_stored_prompt,
"one_shot_prompt": one_shot_prompt,
"reminder": reminder,
"backend": backend,
"model": model,
"effort": effort,
});
let response = cli_post_json("/api/sessions/restart", &body).await?;
println!("{response}");
if !restart_response_succeeded(&response) {
std::process::exit(1);
}
}
Command::ClearReminder { clearing_id, from } => {
let from = match from {
Some(id) => id,
None => require_my_session_id().await?,
};
let body = serde_json::json!({
"from": from,
"clearing_id": clearing_id,
});
cli_post("/api/clear-reminder", &body).await?;
}
Command::Pending => {
let pane = std::env::var("TMUX_PANE")
.context("TMUX_PANE not set — must be run from a tmux pane")?;
let pane = pane_wire_suffix(&pane);
cli_get(&format!("/api/pane/{pane}/pending-replies")).await?;
}
Command::ClearReply { sender_id } => {
let pane = std::env::var("TMUX_PANE")
.context("TMUX_PANE not set — must be run from a tmux pane")?;
let pane = pane_wire_suffix(&pane);
let sender_id = encode_path_segment(&sender_id);
cli_delete(&format!("/api/pane/{pane}/pending-replies/{sender_id}")).await?;
}
Command::StopServer => {
stop_daemon()?;
}
Command::RestartServer => {
restart_daemon()?;
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let status_url = format!("http://localhost:{port}/api/status");
if wait_for_daemon(&status_url) {
println!("daemon restarted");
} else {
anyhow::bail!("daemon did not come back within 10s");
}
}
Command::LogPath { data } => {
let config = config::OuijaConfig::new("_".into(), 0, data, String::new())?;
println!("{}", config.data_dir.join("messages.jsonl").display());
println!("{}", config.data_dir.join("daemon.log").display());
}
Command::SelfUpdate => {
update_and_restart()?;
}
Command::Config { action } => match action {
None => cli_get("/api/settings").await?,
Some(ConfigAction::Set { key, value }) => {
let parsed: serde_json::Value = match value.as_str() {
"true" => serde_json::Value::Bool(true),
"false" => serde_json::Value::Bool(false),
v => serde_json::Value::String(v.to_string()),
};
let body = serde_json::json!({ key: parsed });
cli_post("/api/settings", &body).await?;
}
Some(ConfigAction::SetCodexModelRoute {
alias,
model,
codex_home,
}) => {
if alias.trim().is_empty() {
anyhow::bail!("alias cannot be empty");
}
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}/api/settings");
let current: serde_json::Value = reqwest::get(&url).await?.json().await?;
let mut routes = current
.get("codex_model_routes")
.cloned()
.unwrap_or_else(|| serde_json::json!({}));
if !routes.is_object() {
routes = serde_json::json!({});
}
routes[alias.trim()] = serde_json::json!({
"model": model,
"codex_home": codex_home,
});
cli_post(
"/api/settings",
&serde_json::json!({
"codex_model_routes": routes,
}),
)
.await?;
}
Some(ConfigAction::RemoveCodexModelRoute { alias }) => {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}/api/settings");
let current: serde_json::Value = reqwest::get(&url).await?.json().await?;
let mut routes = current
.get("codex_model_routes")
.cloned()
.unwrap_or_else(|| serde_json::json!({}));
if let Some(map) = routes.as_object_mut() {
map.remove(alias.trim());
}
cli_post(
"/api/settings",
&serde_json::json!({
"codex_model_routes": routes,
}),
)
.await?;
}
Some(ConfigAction::AddHuman {
npub,
name,
default_session,
}) => {
if !npub.starts_with("npub1") {
anyhow::bail!("npub must start with 'npub1'");
}
let config_dir = config::OuijaConfig::default_config_dir();
std::fs::create_dir_all(&config_dir)?;
let mut settings = persistence::load_settings(&config_dir)?;
if settings.human_sessions.iter().any(|h| h.name == name) {
anyhow::bail!("Nostr DM user '{name}' already exists");
}
settings.human_sessions.push(persistence::HumanSession {
npub,
name: name.clone(),
default_session,
welcomed: false,
});
persistence::save_settings(&config_dir, &settings)?;
println!("added Nostr DM user '{name}'");
}
Some(ConfigAction::RemoveHuman { name }) => {
let config_dir = config::OuijaConfig::default_config_dir();
let mut settings = persistence::load_settings(&config_dir)?;
let before = settings.human_sessions.len();
settings.human_sessions.retain(|h| h.name != name);
if settings.human_sessions.len() == before {
anyhow::bail!("Nostr DM user '{name}' not found");
}
persistence::save_settings(&config_dir, &settings)?;
println!("removed Nostr DM user '{name}'");
}
Some(ConfigAction::ListHumans) => {
let config_dir = config::OuijaConfig::default_config_dir();
let settings = persistence::load_settings(&config_dir)?;
if settings.human_sessions.is_empty() {
println!("no Nostr DM users configured");
} else {
println!("{:<12} {:<20} DEFAULT", "NAME", "NPUB");
for h in &settings.human_sessions {
let npub_short = if h.npub.len() > 16 {
format!("{}...", &h.npub[..16])
} else {
h.npub.clone()
};
let default = h.default_session.as_deref().unwrap_or("--");
println!("{:<12} {:<20} {}", h.name, npub_short, default);
}
}
}
Some(ConfigAction::SetRouter {
api_key,
model,
base_url,
}) => {
let config_dir = config::OuijaConfig::default_config_dir();
std::fs::create_dir_all(&config_dir)?;
let mut settings = persistence::load_settings(&config_dir)?;
settings.router = Some(persistence::RouterConfig {
api_key,
model: model.unwrap_or_else(|| "gemini-2.5-flash".to_string()),
base_url: base_url.unwrap_or_else(|| {
"https://generativelanguage.googleapis.com/v1beta/openai".to_string()
}),
});
persistence::save_settings(&config_dir, &settings)?;
println!("router configured");
}
Some(ConfigAction::RemoveRouter) => {
let config_dir = config::OuijaConfig::default_config_dir();
let mut settings = persistence::load_settings(&config_dir)?;
settings.router = None;
persistence::save_settings(&config_dir, &settings)?;
println!("router removed");
}
},
Command::Task { action } => match action {
TaskAction::List => {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}/api/tasks");
let resp: serde_json::Value = reqwest::get(&url).await?.json().await?;
let tasks = resp["tasks"].as_array();
match tasks {
Some(list) if !list.is_empty() => {
println!(
"{:<10} {:<16} {:<16} {:<12} {:<10} {:<10} {:<12} {:<8} {:<20} RUNS",
"ID",
"NAME",
"CRON",
"MODE",
"TARGET",
"BACKEND",
"MODEL",
"ENABLED",
"NEXT RUN"
);
for t in list {
let id = t["id"].as_str().unwrap_or("-");
let name = t["name"].as_str().unwrap_or("-");
let cron = t["cron"].as_str().unwrap_or("-");
let mode = t["on_fire"]["mode"].as_str().unwrap_or("-");
let target = t["target_session"].as_str().unwrap_or("—");
let backend = t["backend"].as_str().unwrap_or("—");
let model = t["model"].as_str().unwrap_or("—");
let enabled = t["enabled"].as_bool().unwrap_or(false);
let next = t["next_run"].as_str().unwrap_or("-");
let runs = t["run_count"].as_u64().unwrap_or(0);
println!(
"{:<10} {:<16} {:<16} {:<12} {:<10} {:<10} {:<12} {:<8} {:<20} {}",
id, name, cron, mode, target, backend, model, enabled, next, runs
);
}
}
_ => println!("no scheduled tasks"),
}
}
TaskAction::Add {
name,
cron,
target,
message,
inject_only,
project_dir,
backend,
model,
effort,
once,
} => {
let body = serde_json::json!({
"name": name,
"cron": cron,
"target_session": target,
"message": message,
"on_fire": inject_only.then_some(crate::scheduler::OnFire::InjectOnly),
"project_dir": project_dir,
"backend": backend,
"model": model,
"effort": effort,
"once": once,
});
cli_post("/api/tasks", &body).await?;
}
TaskAction::Remove { id } => {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}/api/tasks");
let client = reqwest::Client::new();
let body = serde_json::json!({ "id": id });
let resp = client.delete(&url).json(&body).send().await?;
println!("{}", resp.text().await?);
}
TaskAction::Enable { id } => {
let body = serde_json::json!({ "id": id });
cli_post("/api/tasks/enable", &body).await?;
}
TaskAction::Disable { id } => {
let body = serde_json::json!({ "id": id });
cli_post("/api/tasks/disable", &body).await?;
}
TaskAction::Runs { task } => {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let mut url = format!("http://localhost:{port}/api/task-runs");
if let Some(id) = &task {
url.push_str(&format!("?task={id}"));
}
let resp: serde_json::Value = reqwest::get(&url).await?.json().await?;
let runs = resp["runs"].as_array();
match runs {
Some(list) if !list.is_empty() => {
println!(
"{:<22} {:<12} {:<10} {:<10} ERROR",
"TIME", "TASK", "TARGET", "STATUS"
);
for r in list {
let ts = r["timestamp"].as_str().unwrap_or("-");
let name = r["task_name"].as_str().unwrap_or("-");
let target = r["session_name"].as_str().unwrap_or("-");
let status = r["status"].as_str().unwrap_or("-");
let err = r["error"].as_str().unwrap_or("");
println!(
"{:<22} {:<12} {:<10} {:<10} {}",
ts, name, target, status, err
);
}
}
_ => println!("no task runs"),
}
}
TaskAction::Trigger { id } => {
let body = serde_json::json!({ "id": id });
cli_post("/api/tasks/trigger", &body).await?;
}
},
}
Ok(())
}
async fn setup_nostr_transport(
state: &state::SharedState,
ticket: Option<&str>,
cli_relays: Vec<String>,
) {
let transport = match nostr_transport::ensure_active(state, cli_relays).await {
Ok(t) => t,
Err(e) => {
tracing::error!("nostr transport setup failed: {e}");
return;
}
};
if let Some(ticket) = ticket
&& let Err(e) = transport.connect(ticket, state.clone(), true).await
{
tracing::warn!("failed to connect to ticket node: {e}");
}
reconnect_persisted_nodes(state.clone()).await;
transport::broadcast_local_sessions(state).await;
}
fn abandoned_lease_owns_staged_row(
session: &persistence::PersistedSession,
lease: &crate::daemon_protocol::LifecycleLease,
) -> bool {
if lease.owner.session_id != session.id {
return false;
}
match lease.phase {
crate::daemon_protocol::LifecyclePhase::Starting => {
session.metadata.session_incarnation == lease.owner.incarnation
}
crate::daemon_protocol::LifecyclePhase::Restarting => {
let owns_recorded_inert_pane = lease.inert_pane.is_some()
&& lease.inert_pane.as_ref() == session.pane.as_ref()
&& lease.inert_pane_owner.as_ref().is_some_and(|owner| {
owner.session_id == session.id
&& owner.incarnation == session.metadata.session_incarnation
});
let owns_staged_target = lease.restart_target_owner.as_ref().map_or_else(
|| session.metadata.session_incarnation != lease.owner.incarnation,
|owner| {
owner.session_id == session.id
&& owner.incarnation == session.metadata.session_incarnation
},
);
owns_recorded_inert_pane || owns_staged_target
}
crate::daemon_protocol::LifecyclePhase::Stopping => {
session.metadata.session_incarnation == lease.owner.incarnation
&& lease.inert_pane.as_ref() == session.pane.as_ref()
&& lease.inert_pane_owner.as_ref() == Some(&lease.owner)
}
}
}
fn lifecycle_lease_pane_owners(
lease: &crate::daemon_protocol::LifecycleLease,
) -> Vec<crate::daemon_protocol::ResourceOwner> {
let mut owners = vec![lease.owner.clone()];
if let Some(owner) = &lease.restart_target_owner
&& !owners.contains(owner)
{
owners.push(owner.clone());
}
if let Some(owner) = &lease.inert_pane_owner
&& !owners.contains(owner)
{
owners.push(owner.clone());
}
owners
}
fn reaper_process_names(
backends: &crate::backend::BackendRegistry,
metadata: &crate::daemon_protocol::SessionMeta,
) -> Option<Vec<String>> {
match metadata
.backend
.as_deref()
.and_then(|backend| backends.get(backend))
{
Some(backend) => Some(
backend
.process_names()
.iter()
.map(|name| (*name).to_string())
.collect(),
),
None => Some(backends.all_process_names()),
}
}
async fn reap_dead_sessions(
state: &state::SharedState,
dead_sessions: Vec<(crate::daemon_protocol::ResourceOwner, String)>,
observed_at: i64,
) {
for (owner, pane) in dead_sessions {
let _ = state
.dormant_owned(
owner,
Some(pane),
observed_at,
crate::daemon_protocol::DormancySource::Reaped,
)
.await;
}
}
fn persisted_session_from_entry(
entry: &crate::daemon_protocol::SessionEntry,
) -> Option<persistence::PersistedSession> {
if !matches!(entry.origin, crate::daemon_protocol::Origin::Local) {
return None;
}
let metadata = &entry.metadata;
let timestamp =
chrono::DateTime::from_timestamp(entry.registered_at, 0).unwrap_or_else(chrono::Utc::now);
Some(persistence::PersistedSession {
id: entry.id.clone(),
pane: entry.pane.clone(),
registered_at: timestamp,
last_activity_at: timestamp,
metadata: state::SessionMetadata {
vim_mode: metadata.vim_mode,
project_dir: metadata.project_dir.clone(),
canonical_project_identity: metadata.canonical_project_identity.clone(),
role: metadata.role.clone(),
networked: metadata.networked,
last_metadata_update: metadata
.last_metadata_update
.and_then(|value| chrono::DateTime::from_timestamp(value, 0)),
backend_session_id: metadata.backend_session_id.clone(),
backend: metadata.backend.clone(),
opencode_binding: metadata.opencode_binding.clone(),
restart_generation: metadata.restart_generation,
backend_repair_reservation: metadata.backend_repair_reservation.clone(),
session_incarnation: metadata.session_incarnation,
project_description: metadata.project_description.clone(),
bulletin: metadata.bulletin.clone(),
worktree: metadata.worktree,
model: metadata.model.clone(),
effort: metadata.effort.clone(),
codex_home: metadata.codex_home.clone(),
reminder: metadata.reminder.clone(),
parent_session: metadata.parent_session.clone(),
idle_policy: metadata.idle_policy.clone(),
prompt: metadata.prompt.clone(),
iteration: metadata.iteration,
iteration_log: metadata.iteration_log.clone(),
last_iteration_at: metadata.last_iteration_at,
on_fire: metadata.on_fire.clone(),
worktree_present: metadata.worktree_present,
fresh_context_after_active_secs: metadata.fresh_context_after_active_secs,
active_context_accumulated_secs: metadata.active_context_accumulated_secs,
active_context_segment_started_at: metadata.active_context_segment_started_at,
active_context_restart_due: metadata.active_context_restart_due,
active_context_accounting_provisional: metadata.active_context_accounting_provisional,
},
})
}
const DAEMON_LOG_MAX_BYTES: u64 = 64 * 1024 * 1024;
fn open_daemon_log(data_dir: &std::path::Path) -> anyhow::Result<std::fs::File> {
let path = data_dir.join("daemon.log");
if std::fs::metadata(&path).is_ok_and(|m| m.len() > DAEMON_LOG_MAX_BYTES) {
let _ = std::fs::rename(&path, data_dir.join("daemon.log.1"));
}
Ok(std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&path)?)
}
fn scrub_leaked_tmux_session_environments() {
if cfg!(test) {
return;
}
let Ok(output) = std::process::Command::new("tmux")
.args(["list-sessions", "-F", "#{session_name}"])
.output()
else {
return;
};
if !output.status.success() {
return;
}
let mut scrubbed = 0;
for name in String::from_utf8_lossy(&output.stdout)
.lines()
.map(str::trim)
.filter(|name| !name.is_empty())
{
tmux::scrub_session_env(name);
scrubbed += 1;
}
if scrubbed > 0 {
tracing::info!(scrubbed, "scrubbed leaked tmux session environments");
}
}
async fn restore_persisted_sessions(state: &state::SharedState) -> anyhow::Result<()> {
let (persisted, backfilled) = persistence::load_sessions_with_backfill(&state.config.data_dir)
.context("failed to restore lifecycle authority from sessions.json")?;
if backfilled > 0 {
persistence::save_sessions(&state.config.data_dir, &persisted)
.context("failed to persist canonical project identity backfill")?;
tracing::info!(backfilled, "persisted canonical project identity backfill");
}
let persistence::PersistedLifecycleState {
mut sessions,
mut dormant_sessions,
incarnation_high_water,
lifecycle_leases,
pending_replies,
..
} = persisted;
let abandoned_leases: Vec<_> = lifecycle_leases.values().cloned().collect();
for lease in &abandoned_leases {
if let Some(dormant) = lease.dormant_resume_previous.as_deref() {
dormant_sessions.insert(dormant.id.clone(), dormant.clone());
}
}
{
let mut proto = state.protocol.write().await;
proto.restore_incarnation_high_water(incarnation_high_water);
proto.dormant_sessions = dormant_sessions.clone();
proto.lifecycle_leases = lifecycle_leases;
let live: std::collections::HashSet<&str> =
sessions.iter().map(|s| s.id.as_str()).collect();
proto.pending_replies = pending_replies
.iter()
.filter(|(owed_by, _)| live.contains(owed_by.as_str()))
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
}
if !abandoned_leases.is_empty() {
for lease in &abandoned_leases {
let (Some(backend), Some(backend_session_id), Some(backend_session_owner)) = (
lease.backend.as_deref(),
lease.backend_session_id.as_deref(),
lease.backend_session_owner.as_ref(),
) else {
continue;
};
if lease.phase != crate::daemon_protocol::LifecyclePhase::Restarting
|| !state.backends.uses_http_delivery(backend)
{
continue;
}
if lease.restart_target_owner.as_ref() != Some(backend_session_owner) {
anyhow::bail!(
"abandoned restart backend owner does not match target for '{}'",
lease.owner.session_id
);
}
let persisted_sharer = sessions.iter().any(|session| {
session.metadata.backend.as_deref() == Some(backend)
&& session.metadata.backend_session_id.as_deref() == Some(backend_session_id)
&& session.metadata.session_incarnation != backend_session_owner.incarnation
}) || dormant_sessions.values().any(|dormant| {
dormant.metadata.backend.as_deref() == Some(backend)
&& dormant.metadata.backend_session_id.as_deref() == Some(backend_session_id)
});
if persisted_sharer {
tracing::info!(
backend,
backend_session_id,
"skipping abandoned restart backend cleanup: replacement session owns it"
);
continue;
}
let port = state.opencode_serve_port();
let backend_session_id_segment = encode_path_segment(backend_session_id);
let url = format!("http://127.0.0.1:{port}/session/{backend_session_id_segment}");
let response = state
.http_client
.delete(&url)
.timeout(std::time::Duration::from_secs(5))
.send()
.await
.with_context(|| {
format!(
"failed to delete abandoned restart backend session {backend_session_id} for lifecycle owner {backend_session_owner:?}"
)
})?;
if !response.status().is_success()
&& response.status() != reqwest::StatusCode::NOT_FOUND
{
anyhow::bail!(
"backend cleanup for abandoned restart session '{}' returned {}",
backend_session_id,
response.status()
);
}
}
let mut quarantined: std::collections::BTreeMap<
String,
crate::daemon_protocol::LifecycleLease,
> = std::collections::BTreeMap::new();
for lease in &abandoned_leases {
let (Some(backend), Some(backend_session_id), Some(backend_session_owner)) = (
lease.backend.as_deref(),
lease.backend_session_id.as_deref(),
lease.backend_session_owner.as_ref(),
) else {
continue;
};
if lease.phase != crate::daemon_protocol::LifecyclePhase::Stopping
|| !state.backends.uses_http_delivery(backend)
{
continue;
}
let persisted_sharer = sessions.iter().any(|session| {
session.metadata.backend.as_deref() == Some(backend)
&& session.metadata.backend_session_id.as_deref() == Some(backend_session_id)
&& !abandoned_lease_owns_staged_row(session, lease)
}) || dormant_sessions.values().any(|dormant| {
dormant.metadata.backend.as_deref() == Some(backend)
&& dormant.metadata.backend_session_id.as_deref() == Some(backend_session_id)
});
if persisted_sharer {
tracing::info!(
backend,
backend_session_id,
"skipping abandoned backend abort: persisted session still owns it"
);
continue;
}
if backend_session_owner != &lease.owner {
anyhow::bail!(
"abandoned backend abort owner does not match stopping lease for '{}'",
lease.owner.session_id
);
}
let port = state.opencode_serve_port();
let backend_session_id_segment = encode_path_segment(backend_session_id);
let url = format!("http://127.0.0.1:{port}/session/{backend_session_id_segment}/abort");
let response = state
.http_client
.post(&url)
.timeout(std::time::Duration::from_secs(5))
.send()
.await
.with_context(|| {
format!(
"failed to abort abandoned backend session {backend_session_id} for lifecycle owner {backend_session_owner:?}"
)
})?;
if !response.status().is_success()
&& response.status() != reqwest::StatusCode::NOT_FOUND
{
anyhow::bail!(
"backend abort for abandoned session '{}' returned {}",
backend_session_id,
response.status()
);
}
let protected: std::collections::HashSet<String> = sessions
.iter()
.filter(|session| !abandoned_lease_owns_staged_row(session, lease))
.filter_map(|session| session.metadata.backend_session_id.clone())
.chain(
dormant_sessions
.values()
.filter_map(|dormant| dormant.metadata.backend_session_id.clone()),
)
.collect();
let sweep = crate::nostr_transport::abort_subagent_tree(
state,
&lease.owner,
&lease.owner.session_id,
backend_session_id,
&protected,
)
.await;
let subagents_aborted = match sweep {
Ok(subagents_aborted) => subagents_aborted,
Err(failure) => {
let reason = format!(
"subagent sweep for abandoned session '{backend_session_id}' did not complete: {}",
failure.detail
);
tracing::error!(
session = %lease.owner.session_id,
backend_session_id,
"quarantining lifecycle lease: {reason}"
);
let mut quarantined_lease = lease.clone();
quarantined_lease.sweep_unconfirmed = Some(reason);
quarantined.insert(lease.owner.session_id.clone(), quarantined_lease);
continue;
}
};
tracing::info!(
backend_session_id,
subagents_aborted,
"swept subagents of abandoned backend session"
);
}
let abandoned_leases: Vec<_> = abandoned_leases
.into_iter()
.filter(|lease| !quarantined.contains_key(&lease.owner.session_id))
.collect();
let inert_panes: Vec<_> = abandoned_leases
.iter()
.filter_map(|lease| {
lease
.inert_pane
.clone()
.map(|pane| (pane, lifecycle_lease_pane_owners(lease)))
})
.collect();
if !inert_panes.is_empty() {
tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
for (pane, expected_owners) in inert_panes {
let live_owner = crate::tmux::inspect_pane_owner(&pane)?;
if !live_owner
.as_ref()
.is_some_and(|owner| expected_owners.contains(owner))
{
continue;
}
let status = std::process::Command::new("tmux")
.args(["kill-pane", "-t", &pane])
.status()
.with_context(|| format!("failed to kill abandoned pane {pane}"))?;
let remaining_owner = crate::tmux::inspect_pane_owner(&pane)?;
if !status.success()
&& remaining_owner
.as_ref()
.is_some_and(|owner| expected_owners.contains(owner))
{
anyhow::bail!(
"failed to remove abandoned pane {pane} for lifecycle owner {remaining_owner:?}"
);
}
}
Ok(())
})
.await
.context("abandoned start pane reconciliation task failed")?
.context("failed to reconcile abandoned start panes")?;
}
for lease in &abandoned_leases {
let (Some(project_dir), Some(project_dir_owner)) =
(&lease.project_dir, &lease.project_dir_owner)
else {
continue;
};
if !lease.project_dir_cleanup_on_abandon {
continue;
}
let project_dir_identity = crate::state::project_dir_identity(project_dir);
let persisted_sharer =
sessions.iter().any(|session| {
session.metadata.project_dir.as_deref().is_some_and(|dir| {
crate::state::project_dir_identity(dir) == project_dir_identity
}) && !abandoned_lease_owns_staged_row(session, lease)
}) || lease.restart_previous.as_deref().is_some_and(|previous| {
previous.metadata.project_dir.as_deref().is_some_and(|dir| {
crate::state::project_dir_identity(dir) == project_dir_identity
})
}) || dormant_sessions.values().any(|dormant| {
dormant.metadata.project_dir.as_deref().is_some_and(|dir| {
crate::state::project_dir_identity(dir) == project_dir_identity
}) || crate::state::project_dir_identity(&dormant.canonical_project_identity)
== project_dir_identity
});
if persisted_sharer {
tracing::info!(
"skipping abandoned worktree cleanup for {project_dir}: persisted session still uses it"
);
continue;
}
state
.cleanup_worktree_dir_if_unused(project_dir_owner, project_dir)
.await;
}
for lease in &abandoned_leases {
let (
crate::daemon_protocol::LifecyclePhase::Restarting,
Some(target_owner),
Some(previous),
) = (
lease.phase,
lease.restart_target_owner.as_ref(),
lease.restart_previous.as_deref(),
)
else {
continue;
};
let Some(position) = sessions.iter().position(|session| {
session.id == target_owner.session_id
&& session.metadata.session_incarnation == target_owner.incarnation
}) else {
continue;
};
let previous = persisted_session_from_entry(previous).ok_or_else(|| {
anyhow::anyhow!(
"restart recovery for '{}' recorded a non-local incumbent",
lease.owner.session_id
)
})?;
sessions[position] = previous;
}
sessions.retain(|session| {
!abandoned_leases
.iter()
.find(|lease| lease.owner.session_id == session.id)
.is_some_and(|lease| abandoned_lease_owns_staged_row(session, lease))
});
{
let mut proto = state.protocol.write().await;
for lease in &abandoned_leases {
proto.abort_lifecycle(&lease.owner);
}
for (id, lease) in &quarantined {
proto.lifecycle_leases.insert(id.clone(), lease.clone());
}
}
persistence::save_sessions(
&state.config.data_dir,
&persistence::PersistedLifecycleState::new(
sessions.clone(),
dormant_sessions.clone(),
incarnation_high_water,
quarantined.clone(),
)
.with_pending_replies(state.protocol.read().await.pending_replies.clone()),
)
.context("failed to persist reconciled lifecycle authority")?;
}
if sessions.is_empty() {
return Ok(());
}
let (paneless_http, pane_bound): (Vec<_>, Vec<_>) = sessions.into_iter().partition(|ps| {
ps.pane.is_none()
&& ps
.metadata
.backend
.as_deref()
.is_some_and(|backend| state.backends.uses_http_delivery(backend))
});
let names: Vec<String> = state.backends.all_process_names();
let (mut alive, dead) = tokio::task::spawn_blocking(move || {
let name_refs: Vec<&str> = names.iter().map(|s| s.as_str()).collect();
pane_bound.into_iter().partition::<Vec<_>, _>(|ps| {
ps.pane
.as_ref()
.is_some_and(|p| crate::tmux::pane_alive(p, &name_refs))
})
})
.await
.unwrap_or_default();
alive.extend(paneless_http);
let (marker_effects, dead_owners) = {
let mut proto = state.protocol.write().await;
let mut marker_effects = Vec::new();
let mut dead_owners = Vec::new();
for (ps, is_alive) in alive
.iter()
.map(|session| (session, true))
.chain(dead.iter().map(|session| (session, false)))
{
let entry = crate::daemon_protocol::SessionEntry {
id: ps.id.clone(),
pane: ps.pane.clone(),
origin: crate::daemon_protocol::Origin::Local,
metadata: metadata_for_restored_session(&ps.metadata),
registered_at: ps.registered_at.timestamp(),
..Default::default()
};
let owner = entry.owner();
if is_alive {
if let Some(pane) = &entry.pane {
marker_effects.extend([
crate::daemon_protocol::Effect::SetTmuxVar {
owner: owner.clone(),
pane: pane.clone(),
name: "@ouija_session".into(),
value: entry.id.clone(),
},
crate::daemon_protocol::Effect::SetTmuxVar {
owner: owner.clone(),
pane: pane.clone(),
name: "@ouija_id".into(),
value: entry.id.clone(),
},
crate::daemon_protocol::Effect::SetTmuxVar {
owner: owner.clone(),
pane: pane.clone(),
name: "@ouija_last_session".into(),
value: entry.id.clone(),
},
crate::daemon_protocol::Effect::SetTmuxVar {
owner: owner.clone(),
pane: pane.clone(),
name: "@ouija_incarnation".into(),
value: entry.metadata.session_incarnation.to_string(),
},
]);
}
if let Some(pane) = entry.session_agent_pane() {
marker_effects.push(crate::daemon_protocol::Effect::SpawnAgent {
owner,
pane: pane.map(String::from),
});
}
} else {
dead_owners.push((owner, entry.pane.clone()));
}
proto.sessions.insert(ps.id.clone(), entry);
}
(marker_effects, dead_owners)
};
let observed_at = chrono::Utc::now().timestamp();
for (owner, pane) in dead_owners {
if matches!(
state
.dormant_owned(
owner,
pane,
observed_at,
crate::daemon_protocol::DormancySource::Reaped,
)
.await,
crate::state::DormantOwnedOutcome::PersistenceFailed
) {
anyhow::bail!("failed to persist startup dormancy reconciliation");
}
}
let _ = state.execute_effects(&marker_effects).await;
tracing::info!(
"restored {} live and reconciled {} dead persisted sessions",
alive.len(),
dead.len()
);
Ok(())
}
fn metadata_for_restored_session(
metadata: &state::SessionMetadata,
) -> crate::daemon_protocol::SessionMeta {
crate::daemon_protocol::metadata_to_session_meta(Some(metadata))
}
async fn register_human_sessions(state: &state::AppState) {
let humans = state.settings.read().await.human_sessions.clone();
if humans.is_empty() {
return;
}
let mut proto = state.protocol.write().await;
for h in &humans {
if proto.sessions.contains_key(&h.name) {
tracing::debug!("human session '{}' already registered", h.name);
continue;
}
let entry = crate::daemon_protocol::SessionEntry {
id: h.name.clone(),
pane: None,
origin: crate::daemon_protocol::Origin::Human(h.npub.clone()),
metadata: crate::daemon_protocol::SessionMeta {
role: Some("human".to_string()),
networked: false,
..Default::default()
},
..Default::default()
};
proto.sessions.insert(h.name.clone(), entry);
tracing::info!("registered human session: {}", h.name);
}
}
async fn reconnect_persisted_nodes(state: state::SharedState) {
let conns = match persistence::load_connections(&state.config.data_dir) {
Ok(c) => c,
Err(e) => {
tracing::warn!("failed to load persisted connections: {e}");
return;
}
};
let Some(transport) = state.transport_by_name("nostr").await else {
tracing::warn!("skipping node reconnection: nostr transport not active");
return;
};
let mut reconnected = 0;
let mut connected_npubs = std::collections::HashSet::new();
for conn in &conns {
if !conn.ticket.starts_with("nprofile1") {
tracing::info!("skipping legacy non-nostr connection");
continue;
}
let label = match &conn.node_name {
Some(name) => name.clone(),
None => "unnamed".to_string(),
};
let npub = conn
.daemon_npub
.clone()
.or_else(|| crate::api::extract_npub(&conn.ticket));
if let Some(ref npub) = npub {
connected_npubs.insert(npub.clone());
let node_name = conn
.node_name
.as_deref()
.unwrap_or(&npub[..16.min(npub.len())]);
if let Err(existing) = state.try_add_node(npub, node_name) {
tracing::info!(
"skipping duplicate connection to {label} (already connected as '{existing}')"
);
continue;
}
}
tracing::info!("reconnecting to {label}...");
match transport.connect(&conn.ticket, state.clone(), false).await {
Ok(()) => reconnected += 1,
Err(e) => tracing::warn!("failed to reconnect to {label}: {e}"),
}
}
let peer_pubkeys = nostr_transport::load_peer_pubkeys(&state.config.data_dir);
let relay_urls = nostr_transport::load_relays(&state.config.data_dir);
if !peer_pubkeys.is_empty() && !relay_urls.is_empty() {
use nostr_sdk::prelude::*;
let relay_parsed: Vec<RelayUrl> = relay_urls
.iter()
.filter_map(|u| RelayUrl::parse(u).ok())
.collect();
for pubkey in &peer_pubkeys {
let npub = match pubkey.to_bech32() {
Ok(n) => n,
Err(_) => continue,
};
if connected_npubs.contains(&npub) {
continue;
}
let label = &npub[..16.min(npub.len())];
if let Err(existing) = state.try_add_node(&npub, label) {
tracing::info!(
"skipping duplicate peer_pubkey connection to {label} (already connected as '{existing}')"
);
continue;
}
let profile = Nip19Profile::new(*pubkey, relay_parsed.clone());
let nprofile = match profile.to_bech32() {
Ok(s) => s,
Err(_) => continue,
};
tracing::info!("reconnecting to peer_pubkey {label}...");
match transport.connect(&nprofile, state.clone(), false).await {
Ok(()) => {
if let Err(e) = persistence::add_connection(
&state.config.data_dir,
&nprofile,
None,
Some(&npub),
) {
tracing::warn!("failed to persist fallback connection: {e}");
}
reconnected += 1;
}
Err(e) => tracing::warn!("failed to reconnect to peer_pubkey {label}: {e}"),
}
}
}
if reconnected > 0 {
tracing::info!("reconnected to {reconnected} persisted nodes");
}
}
fn preflight_checks() {
use std::process::Command as Cmd;
if Cmd::new("tmux").arg("-V").output().is_err() {
eprintln!("error: tmux not found");
eprintln!();
eprintln!("ouija requires tmux. Install it:");
eprintln!(" apt install tmux # Debian/Ubuntu");
eprintln!(" brew install tmux # macOS");
eprintln!(" pacman -S tmux # Arch");
std::process::exit(1);
}
let backend = backend::claude_code::ClaudeCode;
if !backend.is_available() {
eprintln!("warning: {} not found on PATH", backend.cli_name());
eprintln!(
" Sessions won't auto-register. Install: https://docs.anthropic.com/en/docs/claude-code"
);
eprintln!();
}
}
const OUIJA_SYSTEMD_UNIT: &str = "ouija.service";
#[derive(Debug, PartialEq, Eq)]
struct DaemonStopPlan {
stop_systemd: bool,
stop_legacy: bool,
}
#[derive(Debug, PartialEq, Eq)]
enum DaemonLifecyclePlan {
LegacyOnly,
SystemdOnly,
SystemdAfterLegacyCleanup,
}
#[derive(Debug, Default)]
struct LegacyStopOutcome {
tmux_killed: bool,
process_killed: bool,
}
impl LegacyStopOutcome {
fn stopped_anything(&self) -> bool {
self.tmux_killed || self.process_killed
}
}
#[derive(Debug, Default)]
struct DaemonStopOutcome {
systemd_stopped: bool,
legacy: LegacyStopOutcome,
}
impl DaemonStopOutcome {
fn stopped_anything(&self) -> bool {
self.systemd_stopped || self.legacy.stopped_anything()
}
}
fn plan_daemon_stop(systemd_unit_available: bool) -> DaemonStopPlan {
DaemonStopPlan {
stop_systemd: systemd_unit_available,
stop_legacy: true,
}
}
fn plan_supervised_lifecycle(
systemd_unit_available: bool,
systemd_unit_active: bool,
) -> DaemonLifecyclePlan {
if !systemd_unit_available {
DaemonLifecyclePlan::LegacyOnly
} else if systemd_unit_active {
DaemonLifecyclePlan::SystemdOnly
} else {
DaemonLifecyclePlan::SystemdAfterLegacyCleanup
}
}
fn legacy_cleanup_settle_delay(plan: &DaemonLifecyclePlan) -> Option<std::time::Duration> {
match plan {
DaemonLifecyclePlan::SystemdAfterLegacyCleanup => Some(std::time::Duration::from_secs(1)),
DaemonLifecyclePlan::LegacyOnly | DaemonLifecyclePlan::SystemdOnly => None,
}
}
fn wait_for_legacy_cleanup_if_needed(plan: &DaemonLifecyclePlan) {
if let Some(delay) = legacy_cleanup_settle_delay(plan) {
std::thread::sleep(delay);
}
}
fn systemd_user_unit_available() -> bool {
use std::process::Command as Cmd;
Cmd::new("systemctl")
.args(["--user", "cat", OUIJA_SYSTEMD_UNIT])
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false)
}
fn systemd_user_unit_active() -> bool {
use std::process::Command as Cmd;
Cmd::new("systemctl")
.args(["--user", "is-active", "--quiet", OUIJA_SYSTEMD_UNIT])
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false)
}
fn systemctl_user(action: &str) -> anyhow::Result<()> {
use std::process::Command as Cmd;
let status = Cmd::new("systemctl")
.args(["--user", action, OUIJA_SYSTEMD_UNIT])
.status()
.with_context(|| format!("failed to run systemctl --user {action} {OUIJA_SYSTEMD_UNIT}"))?;
if !status.success() {
anyhow::bail!("systemctl --user {action} {OUIJA_SYSTEMD_UNIT} failed");
}
Ok(())
}
fn stop_legacy_daemon() -> LegacyStopOutcome {
use std::process::Command as Cmd;
let tmux_killed = Cmd::new("tmux")
.args(["kill-session", "-t", "ouija-daemon"])
.stderr(std::process::Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false);
let pkill_killed = Cmd::new("pkill")
.args(["-f", "ouija start-server"])
.status()
.map(|s| s.success())
.unwrap_or(false);
LegacyStopOutcome {
tmux_killed,
process_killed: pkill_killed,
}
}
fn stop_daemon_processes_with(
systemd_available: bool,
systemd_active: bool,
mut stop_systemd: impl FnMut() -> anyhow::Result<()>,
mut stop_legacy: impl FnMut() -> LegacyStopOutcome,
) -> anyhow::Result<DaemonStopOutcome> {
let plan = plan_daemon_stop(systemd_available);
let mut systemd_stopped = false;
let mut systemd_error = None;
if plan.stop_systemd {
match stop_systemd() {
Ok(()) => systemd_stopped = systemd_active,
Err(err) => systemd_error = Some(err),
}
}
let legacy = if plan.stop_legacy {
stop_legacy()
} else {
LegacyStopOutcome::default()
};
if let Some(err) = systemd_error {
return Err(err);
}
Ok(DaemonStopOutcome {
systemd_stopped,
legacy,
})
}
fn stop_daemon_processes() -> anyhow::Result<DaemonStopOutcome> {
let systemd_available = systemd_user_unit_available();
let systemd_active = systemd_available && systemd_user_unit_active();
stop_daemon_processes_with(
systemd_available,
systemd_active,
|| systemctl_user("stop"),
stop_legacy_daemon,
)
}
fn spawn_legacy_daemon() -> anyhow::Result<()> {
use std::process::Command as Cmd;
Cmd::new("ouija")
.arg("start-server")
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.context("failed to spawn ouija start-server")?;
Ok(())
}
fn start_daemon() -> anyhow::Result<()> {
let systemd_available = systemd_user_unit_available();
let systemd_active = systemd_available && systemd_user_unit_active();
let plan = plan_supervised_lifecycle(systemd_available, systemd_active);
match plan {
DaemonLifecyclePlan::LegacyOnly => spawn_legacy_daemon(),
DaemonLifecyclePlan::SystemdOnly => systemctl_user("start"),
DaemonLifecyclePlan::SystemdAfterLegacyCleanup => {
let _ = stop_legacy_daemon();
wait_for_legacy_cleanup_if_needed(&plan);
systemctl_user("start")
}
}
}
fn restart_daemon() -> anyhow::Result<()> {
let systemd_available = systemd_user_unit_available();
let systemd_active = systemd_available && systemd_user_unit_active();
let plan = plan_supervised_lifecycle(systemd_available, systemd_active);
match plan {
DaemonLifecyclePlan::LegacyOnly => {
let _ = stop_legacy_daemon();
std::thread::sleep(std::time::Duration::from_secs(1));
spawn_legacy_daemon()
}
DaemonLifecyclePlan::SystemdOnly => systemctl_user("restart"),
DaemonLifecyclePlan::SystemdAfterLegacyCleanup => {
let _ = stop_legacy_daemon();
wait_for_legacy_cleanup_if_needed(&plan);
systemctl_user("restart")
}
}
}
fn daemon_http_alive(status_url: &str) -> bool {
use std::process::Command as Cmd;
Cmd::new("curl")
.args(["-sf", status_url])
.stdout(std::process::Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false)
}
fn wait_for_daemon(status_url: &str) -> bool {
for _ in 0..20 {
std::thread::sleep(std::time::Duration::from_millis(500));
if daemon_http_alive(status_url) {
return true;
}
}
false
}
fn sync_current_exe_from_cargo_bin() {
let cargo_bin = std::env::var("HOME")
.map(|h| std::path::PathBuf::from(h).join(".cargo/bin/ouija"))
.unwrap_or_default();
let current_exe = std::env::current_exe().unwrap_or_default();
if cargo_bin.exists() && current_exe != cargo_bin && current_exe.exists() {
let _ = std::fs::remove_file(¤t_exe);
if let Err(e) = std::fs::copy(&cargo_bin, ¤t_exe) {
eprintln!("warning: could not update {}: {e}", current_exe.display());
}
}
}
fn stop_daemon() -> anyhow::Result<()> {
let outcome = stop_daemon_processes()?;
if outcome.stopped_anything() {
println!("ouija daemon stopped");
} else {
println!("no running daemon found");
}
Ok(())
}
fn update_and_restart() -> anyhow::Result<()> {
use std::process::Command as Cmd;
let latest = fetch_latest_crate_version("ouija")?;
let current = env!("CARGO_PKG_VERSION");
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let status_url = format!("http://localhost:{port}/api/status");
let daemon_alive = daemon_http_alive(&status_url);
if latest == current {
println!("already on latest version ({current})");
backend::claude_code::refresh_plugin_cache(&latest);
if !daemon_alive {
println!("daemon is not running — starting it...");
start_daemon()?;
if !wait_for_daemon(&status_url) {
eprintln!("warning: daemon did not start within 10s");
}
}
println!("dashboard: http://localhost:{port}");
return Ok(());
}
println!("updating ouija {current} -> {latest}...");
let status = Cmd::new("cargo")
.args(["install", "ouija", "--version", &latest])
.status()
.context("failed to run cargo install")?;
if !status.success() {
anyhow::bail!("cargo install ouija --version {latest} failed");
}
let opencode_plugin_changed = backend::opencode::installed_plugin_differs();
backend::claude_code::refresh_plugin_cache(&latest);
if opencode_plugin_changed {
let serve_running = Cmd::new("pgrep")
.args(["-f", "opencode serve"])
.stdout(std::process::Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false);
if serve_running {
println!(
"note: the opencode plugin changed and opencode serve is running — restart it to \
pick up the change (this aborts in-flight turns):\n \
pkill -f 'opencode serve' && opencode serve --port 8200 --hostname 127.0.0.1 &"
);
}
}
sync_current_exe_from_cargo_bin();
println!("restarting daemon...");
restart_daemon()?;
if wait_for_daemon(&status_url) {
println!("ouija updated to {latest} and running");
println!("dashboard: http://localhost:{port}");
return Ok(());
}
anyhow::bail!("daemon did not start within 10s")
}
fn fetch_latest_crate_version(name: &str) -> anyhow::Result<String> {
use std::process::Command as Cmd;
let output = Cmd::new("curl")
.args([
"-sf",
"-A",
"ouija-self-update (+https://github.com/dcadenas/ouija)",
&format!("https://crates.io/api/v1/crates/{name}"),
])
.output()
.context("failed to query crates.io")?;
if !output.status.success() {
anyhow::bail!("crates.io query failed");
}
let json: serde_json::Value =
serde_json::from_slice(&output.stdout).context("invalid JSON from crates.io")?;
json["versions"]
.as_array()
.and_then(|versions| {
versions
.iter()
.find(|v| !v["yanked"].as_bool().unwrap_or(true))
.and_then(|v| v["num"].as_str())
.map(String::from)
})
.ok_or_else(|| anyhow::anyhow!("no versions found for {name} on crates.io"))
}
#[derive(Debug, PartialEq, Eq)]
enum SessionIdResolution {
Found(String, IdentitySource),
LookupByPane(String),
None,
}
#[derive(Debug, PartialEq, Eq)]
enum IdentitySource {
PaneVar,
EnvVar,
PaneLookup,
BackendIdentity,
}
#[derive(Debug, PartialEq, Eq)]
struct IdentityConflict {
local_id: String,
local_source: IdentitySource,
canonical_id: String,
}
#[derive(Debug, PartialEq, Eq)]
struct BackendIdentityLookupError {
outcome: Option<String>,
detail: String,
}
impl BackendIdentityLookupError {
fn daemon_rejection(outcome: &str, detail: &str) -> Self {
Self {
outcome: Some(outcome.into()),
detail: detail.into(),
}
}
fn protocol_failure(detail: impl Into<String>) -> Self {
Self {
outcome: None,
detail: detail.into(),
}
}
fn allows_local_fallback(&self) -> bool {
self.outcome.as_deref() == Some("incomplete_legacy")
}
}
impl std::fmt::Display for BackendIdentityLookupError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self.outcome.as_deref() {
Some(outcome) => write!(
f,
"backend identity resolution failed ({outcome}) : {}",
self.detail
),
None => f.write_str(&self.detail),
}
}
}
impl std::error::Error for BackendIdentityLookupError {}
fn arbitrate_backend_identity(
local: Option<(String, IdentitySource)>,
backend_canonical: Option<String>,
) -> Result<Option<(String, IdentitySource)>, IdentityConflict> {
let Some(canonical_id) = backend_canonical else {
return Ok(local);
};
match local {
Some((local_id, local_source)) if local_id != canonical_id => Err(IdentityConflict {
local_id,
local_source,
canonical_id,
}),
Some(_) | None => Ok(Some((canonical_id, IdentitySource::BackendIdentity))),
}
}
fn backend_canonical_for_arbitration(
local: Option<&(String, IdentitySource)>,
backend_lookup: Result<String, BackendIdentityLookupError>,
) -> Result<Option<String>, BackendIdentityLookupError> {
match backend_lookup {
Ok(id) => Ok(Some(id)),
Err(error) if local.is_some() && error.allows_local_fallback() => Ok(None),
Err(error) => Err(error),
}
}
impl std::fmt::Display for IdentitySource {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::PaneVar => write!(f, "the @ouija_session tmux pane var"),
Self::EnvVar => write!(f, "$OUIJA_SESSION_ID"),
Self::PaneLookup => write!(f, "daemon lookup by $TMUX_PANE"),
Self::BackendIdentity => write!(f, "daemon resolution of this backend identity"),
}
}
}
const NO_GUESS_GUIDANCE: &str = "Never guess a sender id — not from the project directory name, \
a branch name, or an `ouija ls` entry. A guessed sender impersonates another session and \
misroutes its replies. Use only an exact id: the one in your injected system prompt \
(\"You are session \\\"<id>\\\" on the ouija mesh\") or a $OUIJA_SESSION_ID provided by your operator.";
#[derive(Debug)]
struct WhoamiFailure {
tmux_pane: Option<String>,
pane_var: Option<String>,
env_var: Option<String>,
lookup: Option<PaneLookupFailure>,
}
#[derive(Debug)]
enum PaneLookupFailure {
DaemonUnreachable(String),
NoSessionForPane,
}
fn format_whoami_failure(failure: &WhoamiFailure) -> String {
let mut lines = vec![
"unable to resolve this session's Ouija identity.".to_string(),
String::new(),
"Signals checked:".to_string(),
];
match &failure.tmux_pane {
Some(pane) => {
lines.push(format!(" - $TMUX_PANE: {pane}"));
match failure.pane_var.as_deref() {
Some("") => lines.push(" - @ouija_session pane var: set but empty".to_string()),
Some(var) => lines.push(format!(" - @ouija_session pane var: {var}")),
None => lines.push(" - @ouija_session pane var: not set".to_string()),
}
}
None => lines.push(
" - $TMUX_PANE: not set (this shell is not attached to a tmux pane)".to_string(),
),
}
match failure.env_var.as_deref() {
Some("") => lines.push(" - $OUIJA_SESSION_ID: set but empty".to_string()),
Some(var) => lines.push(format!(" - $OUIJA_SESSION_ID: {var}")),
None => lines.push(" - $OUIJA_SESSION_ID: not set".to_string()),
}
match &failure.lookup {
Some(PaneLookupFailure::DaemonUnreachable(url)) => {
lines.push(format!(" - daemon lookup: daemon unreachable at {url}"));
}
Some(PaneLookupFailure::NoSessionForPane) => {
lines.push(format!(
" - daemon lookup: no registered session for pane {}",
failure.tmux_pane.as_deref().unwrap_or("?")
));
}
None => {}
}
lines.push(String::new());
lines.push(NO_GUESS_GUIDANCE.to_string());
lines.join("\n")
}
fn format_unregistered_identity(id: &str, source: &IdentitySource) -> String {
format!(
"resolved id '{id}' via {source}, but no local session with that id is registered. \
The session may have been renamed or removed, or the signal is stale. \
Ask the operator for the correct id. {NO_GUESS_GUIDANCE}"
)
}
fn format_identity_conflict(conflict: &IdentityConflict) -> String {
format!(
"backend identity resolves to canonical session '{}', but {} resolves to '{}'. \
Refusing to send with conflicting identity signals. Restart the stale shell or ask \
the operator to repair the session binding. {NO_GUESS_GUIDANCE}",
conflict.canonical_id, conflict.local_source, conflict.local_id
)
}
fn status_lists_local_session(status: &serde_json::Value, id: &str) -> bool {
status["sessions"].as_array().is_some_and(|sessions| {
sessions
.iter()
.any(|s| s["id"].as_str() == Some(id) && s["origin"].as_str() == Some("local"))
})
}
fn unresolved_sender_error() -> String {
format!(
"unable to resolve the current session ID. Run `ouija whoami` for diagnostics. \
If you already know your exact session id (from your injected system prompt or \
$OUIJA_SESSION_ID), pass `--from <id>`. {NO_GUESS_GUIDANCE}"
)
}
fn read_message_file(path: &std::path::Path) -> anyhow::Result<String> {
std::fs::read_to_string(path)
.with_context(|| format!("failed to read message file {}", path.display()))
}
fn pick_session_id(
tmux_pane: Option<&str>,
pane_var: Option<String>,
env_var: Option<String>,
) -> SessionIdResolution {
if tmux_pane.is_some() {
if let Some(id) = pane_var.filter(|s| !s.is_empty()) {
return SessionIdResolution::Found(id, IdentitySource::PaneVar);
}
}
if let Some(id) = env_var.filter(|s| !s.is_empty()) {
return SessionIdResolution::Found(id, IdentitySource::EnvVar);
}
if let Some(pane) = tmux_pane {
return SessionIdResolution::LookupByPane(pane.to_string());
}
SessionIdResolution::None
}
fn sender_context(
self_id: Option<&str>,
tmux_pane: Option<String>,
backend_identity: Option<backend::BackendSessionIdentity>,
trusted_local_claim: bool,
) -> serde_json::Value {
serde_json::json!({
"pane": tmux_pane,
"self_id": self_id,
"backend_identity": backend_identity,
"trusted_local_claim": trusted_local_claim,
})
}
enum WhoamiOutcome {
Resolved {
id: String,
source: IdentitySource,
tmux_pane: Option<String>,
backend_identity: Option<backend::BackendSessionIdentity>,
},
Unresolved(WhoamiFailure),
Conflict(IdentityConflict),
BackendResolutionFailed(String),
}
async fn whoami_outcome() -> WhoamiOutcome {
let tmux_pane = std::env::var("TMUX_PANE").ok();
let pane_var = tmux_pane.as_deref().and_then(tmux_var::get);
let env_var = std::env::var("OUIJA_SESSION_ID").ok();
let backend_identity = backend::BackendRegistry::default_registry().caller_session_identity();
if let Some(pane) = tmux_pane.as_deref() {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let base = format!("http://localhost:{port}");
let status = match reqwest::get(format!("{base}/api/status")).await {
Ok(response) => response.json::<serde_json::Value>().await.ok(),
Err(_) => None,
};
let Some(status) = status else {
return WhoamiOutcome::Unresolved(WhoamiFailure {
tmux_pane,
pane_var,
env_var,
lookup: Some(PaneLookupFailure::DaemonUnreachable(base)),
});
};
let current_id = status["sessions"].as_array().and_then(|sessions| {
sessions
.iter()
.find(|session| {
session["pane"].as_str() == Some(pane)
&& session["origin"].as_str() == Some("local")
})
.and_then(|session| session["id"].as_str().map(String::from))
});
return match current_id {
Some(id) => WhoamiOutcome::Resolved {
id,
source: IdentitySource::PaneLookup,
tmux_pane,
backend_identity,
},
None => WhoamiOutcome::Unresolved(WhoamiFailure {
tmux_pane,
pane_var,
env_var,
lookup: Some(PaneLookupFailure::NoSessionForPane),
}),
};
}
let (local, lookup) =
match pick_session_id(tmux_pane.as_deref(), pane_var.clone(), env_var.clone()) {
SessionIdResolution::Found(id, source) => (Some((id, source)), None),
SessionIdResolution::LookupByPane(pane) => {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let base = format!("http://localhost:{port}");
let status = match reqwest::get(format!("{base}/api/status")).await {
Ok(resp) => resp.json::<serde_json::Value>().await.ok(),
Err(_) => None,
};
match status {
Some(status) => {
let id = status["sessions"].as_array().and_then(|sessions| {
sessions
.iter()
.find(|s| s["pane"].as_str() == Some(&pane))
.and_then(|s| s["id"].as_str().map(String::from))
});
(
id.map(|id| (id, IdentitySource::PaneLookup)),
Some(PaneLookupFailure::NoSessionForPane),
)
}
None => (None, Some(PaneLookupFailure::DaemonUnreachable(base))),
}
}
SessionIdResolution::None => (None, None),
};
let backend_canonical = match backend_identity.as_ref() {
Some(identity) => match backend_canonical_for_arbitration(
local.as_ref(),
resolve_backend_identity_from_daemon(identity).await,
) {
Ok(id) => id,
Err(error) => return WhoamiOutcome::BackendResolutionFailed(error.to_string()),
},
None => None,
};
match arbitrate_backend_identity(local, backend_canonical) {
Ok(Some((id, source))) => WhoamiOutcome::Resolved {
id,
source,
tmux_pane,
backend_identity,
},
Ok(None) => WhoamiOutcome::Unresolved(WhoamiFailure {
tmux_pane,
pane_var,
env_var,
lookup,
}),
Err(conflict) => WhoamiOutcome::Conflict(conflict),
}
}
async fn resolve_backend_identity_from_daemon(
identity: &backend::BackendSessionIdentity,
) -> Result<String, BackendIdentityLookupError> {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}/api/backend-identities/resolve");
let response = reqwest::Client::new()
.post(&url)
.json(&serde_json::json!({
"backend": identity.backend,
"session_id": identity.session_id,
}))
.send()
.await
.map_err(|error| {
BackendIdentityLookupError::protocol_failure(format!(
"could not resolve backend identity via {url}: {error}"
))
})?;
let status = response.status();
let body: serde_json::Value = response.json().await.map_err(|error| {
BackendIdentityLookupError::protocol_failure(format!(
"daemon returned invalid backend identity response: {error}"
))
})?;
if !status.is_success() {
return Err(BackendIdentityLookupError::daemon_rejection(
body["outcome"].as_str().unwrap_or("unknown"),
body["error"].as_str().unwrap_or("no daemon detail"),
));
}
body["session_id"]
.as_str()
.filter(|id| !id.is_empty())
.map(String::from)
.ok_or_else(|| {
BackendIdentityLookupError::protocol_failure(
"daemon resolved backend identity without a session_id",
)
})
}
struct ResolvedSender {
id: String,
context: serde_json::Value,
}
fn resolve_explicit_sender(
explicit: String,
tmux_pane: Option<String>,
pane_var: Option<String>,
env_var: Option<String>,
backend_identity: Option<backend::BackendSessionIdentity>,
) -> ResolvedSender {
let self_id = match pick_session_id(tmux_pane.as_deref(), pane_var, env_var) {
SessionIdResolution::Found(id, _) => Some(id),
SessionIdResolution::LookupByPane(_) | SessionIdResolution::None => None,
};
let context = sender_context(self_id.as_deref(), tmux_pane, backend_identity, true);
ResolvedSender {
id: explicit,
context,
}
}
fn backend_recovery_caller_evidence(
tmux_pane: Option<String>,
pane_var: Option<String>,
env_var: Option<String>,
) -> crate::state::BackendRecoveryCallerEvidence {
crate::state::BackendRecoveryCallerEvidence {
pane: tmux_pane,
pane_var_id: pane_var.filter(|id| !id.is_empty()),
env_id: env_var.filter(|id| !id.is_empty()),
}
}
fn claim_request(
requested_id: String,
tmux_pane: Option<String>,
pane_var: Option<String>,
env_var: Option<String>,
backend_identity: backend::BackendSessionIdentity,
) -> api::LocalClaimRequest {
api::LocalClaimRequest {
requested_id,
caller: crate::state::LocalClaimEvidence {
pane: tmux_pane.filter(|pane| !pane.is_empty()),
pane_var_id: pane_var.filter(|id| !id.is_empty()),
env_id: env_var.filter(|id| !id.is_empty()),
backend_identity,
},
}
}
async fn resolve_sender(explicit: Option<String>) -> anyhow::Result<ResolvedSender> {
if let Some(explicit) = explicit {
let tmux_pane = std::env::var("TMUX_PANE")
.ok()
.filter(|pane| !pane.is_empty());
let pane_var = tmux_pane.as_deref().and_then(tmux_var::get);
let env_var = std::env::var("OUIJA_SESSION_ID")
.ok()
.filter(|id| !id.is_empty());
let backend_identity =
backend::BackendRegistry::default_registry().caller_session_identity();
return Ok(resolve_explicit_sender(
explicit,
tmux_pane,
pane_var,
env_var,
backend_identity,
));
}
match whoami_outcome().await {
WhoamiOutcome::Resolved {
id,
source,
tmux_pane,
backend_identity,
} => {
verify_resolved_id_registered(&id, &source).await?;
let context = sender_context(Some(&id), tmux_pane, backend_identity, false);
Ok(ResolvedSender { id, context })
}
WhoamiOutcome::Conflict(conflict) => anyhow::bail!(format_identity_conflict(&conflict)),
WhoamiOutcome::BackendResolutionFailed(error) => anyhow::bail!(
"backend identity was discovered but could not be resolved safely: {error}"
),
WhoamiOutcome::Unresolved(_) => Err(anyhow::anyhow!(unresolved_sender_error())),
}
}
async fn verify_resolved_id_registered(id: &str, source: &IdentitySource) -> anyhow::Result<()> {
if matches!(source, IdentitySource::PaneLookup) {
return Ok(());
}
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}/api/status");
match reqwest::get(&url).await {
Ok(resp) => match resp.json::<serde_json::Value>().await {
Ok(status) => {
if status_lists_local_session(&status, id) {
Ok(())
} else {
anyhow::bail!(format_unregistered_identity(id, source))
}
}
Err(_) => {
eprintln!(
"warning: could not parse daemon status, could not verify '{id}' is registered"
);
Ok(())
}
},
Err(_) => {
eprintln!("warning: daemon unreachable, could not verify '{id}' is registered");
Ok(())
}
}
}
async fn require_my_session_id() -> anyhow::Result<String> {
match whoami_outcome().await {
WhoamiOutcome::Resolved { id, source, .. } => {
verify_resolved_id_registered(&id, &source).await?;
Ok(id)
}
WhoamiOutcome::Unresolved(_) => Err(anyhow::anyhow!(unresolved_sender_error())),
WhoamiOutcome::Conflict(conflict) => anyhow::bail!(format_identity_conflict(&conflict)),
WhoamiOutcome::BackendResolutionFailed(error) => anyhow::bail!(
"backend identity was discovered but could not be resolved safely: {error}"
),
}
}
async fn cli_whoami() -> anyhow::Result<()> {
match whoami_outcome().await {
WhoamiOutcome::Resolved { id, source, .. } => {
verify_resolved_id_registered(&id, &source).await?;
eprintln!("resolved via {source}");
println!("{id}");
Ok(())
}
WhoamiOutcome::Unresolved(failure) => anyhow::bail!(format_whoami_failure(&failure)),
WhoamiOutcome::Conflict(conflict) => anyhow::bail!(format_identity_conflict(&conflict)),
WhoamiOutcome::BackendResolutionFailed(error) => anyhow::bail!(
"backend identity was discovered but could not be resolved safely: {error}"
),
}
}
async fn cli_get(path: &str) -> anyhow::Result<()> {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}{path}");
let resp = reqwest::get(&url).await?;
let status = resp.status();
let text = resp.text().await?;
let body = classify_http_response(status, &text)?;
println!("{body}");
Ok(())
}
async fn cli_list_sessions() -> anyhow::Result<()> {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}/api/status");
let status: serde_json::Value = reqwest::get(&url).await?.json().await?;
println!("{}", project_session_list(&status));
Ok(())
}
async fn rollover_live_caller() -> anyhow::Result<rollover::LiveCaller> {
let (id, backend_owned) = match whoami_outcome().await {
WhoamiOutcome::Resolved {
id,
source,
backend_identity,
..
} => {
verify_resolved_id_registered(&id, &source).await?;
(
id,
source == IdentitySource::BackendIdentity && backend_identity.is_some(),
)
}
WhoamiOutcome::Unresolved(_) => anyhow::bail!(unresolved_sender_error()),
WhoamiOutcome::Conflict(conflict) => anyhow::bail!(format_identity_conflict(&conflict)),
WhoamiOutcome::BackendResolutionFailed(error) => anyhow::bail!(
"backend identity was discovered but could not be resolved safely: {error}"
),
};
let before = fetch_rollover_incarnation(&id).await?;
let incarnation =
verify_rollover_incarnation_evidence(local_incarnation_hint()?, backend_owned, before)?;
let cwd = std::env::current_dir().context("reading current directory for rollover")?;
let caller = rollover::capture_live_caller(id.clone(), incarnation, &cwd)?;
let after = fetch_rollover_incarnation(&id).await?;
if after != before {
anyhow::bail!(
"session incarnation changed while capturing rollover state ({before} -> {after})"
);
}
Ok(caller)
}
fn verify_rollover_incarnation_evidence(
local_hint: Option<u64>,
backend_owned: bool,
daemon_incarnation: u64,
) -> anyhow::Result<u64> {
match local_hint {
Some(local) if local == daemon_incarnation => Ok(local),
Some(local) => anyhow::bail!(
"local session incarnation {local} does not match daemon incarnation {daemon_incarnation}"
),
None if backend_owned => Ok(daemon_incarnation),
None => anyhow::bail!(
"rollover requires exact caller-owned incarnation evidence from the pane marker, \
OUIJA_SESSION_INCARNATION, or a bound backend identity"
),
}
}
async fn fetch_rollover_incarnation(id: &str) -> anyhow::Result<u64> {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!(
"http://localhost:{port}/api/sessions/{}",
encode_path_segment(id)
);
let response = reqwest::get(&url).await?;
let status = response.status();
let text = response.text().await?;
let body = classify_http_response(status, &text)?;
let session: serde_json::Value =
serde_json::from_str(&body).context("parsing live Ouija session metadata")?;
rollover_incarnation_from_session(&session, id)
}
fn rollover_incarnation_from_session(
session: &serde_json::Value,
expected_id: &str,
) -> anyhow::Result<u64> {
if session["id"].as_str() != Some(expected_id) {
anyhow::bail!("daemon returned a different session while preparing rollover");
}
if session["origin"].as_str() != Some("local") {
anyhow::bail!("rollover is available only to local Ouija sessions");
}
session["session_incarnation"]
.as_str()
.context("daemon status does not expose session_incarnation")
.and_then(|value| {
value
.parse()
.context("daemon returned a non-decimal session_incarnation")
})
}
fn local_incarnation_hint() -> anyhow::Result<Option<u64>> {
let marker = std::env::var("TMUX_PANE").ok().and_then(|pane| {
std::process::Command::new("tmux")
.args([
"display",
"-p",
"-t",
pane.as_str(),
"#{@ouija_incarnation}",
])
.output()
.ok()
.filter(|output| output.status.success())
.and_then(|output| String::from_utf8(output.stdout).ok())
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
});
let hint = marker.or_else(|| std::env::var("OUIJA_SESSION_INCARNATION").ok());
hint.map(|value| {
value
.parse()
.with_context(|| format!("invalid local session incarnation {value:?}"))
})
.transpose()
}
fn project_session_list(status: &serde_json::Value) -> serde_json::Value {
let quarantined: Vec<&serde_json::Value> = status
.get("quarantined_sessions")
.and_then(|value| value.as_array())
.map(|entries| entries.iter().collect())
.unwrap_or_default();
let quarantine_reason = |id: &str| -> Option<String> {
quarantined
.iter()
.find(|entry| entry.get("id").and_then(|v| v.as_str()) == Some(id))
.and_then(|entry| entry.get("reason").and_then(|v| v.as_str()))
.map(str::to_owned)
};
let sessions = status
.get("sessions")
.and_then(|sessions| sessions.as_array())
.map(|sessions| {
sessions
.iter()
.map(|session| {
let mut projected = serde_json::Map::new();
projected.insert(
"id".to_string(),
session
.get("id")
.cloned()
.unwrap_or(serde_json::Value::Null),
);
projected.insert(
"origin".to_string(),
session
.get("origin")
.cloned()
.unwrap_or(serde_json::Value::Null),
);
if let Some(incarnation) = session
.get("session_incarnation")
.and_then(|value| value.as_str())
{
projected.insert(
"session_incarnation".to_string(),
serde_json::Value::String(incarnation.to_string()),
);
}
projected.insert(
"parent_session".to_string(),
session
.get("parent_session")
.cloned()
.unwrap_or(serde_json::Value::Null),
);
if let Some(project) = session
.get("project_dir")
.and_then(|project_dir| project_dir.as_str())
.filter(|project_dir| !project_dir.trim().is_empty())
.and_then(|project_dir| std::path::Path::new(project_dir).file_name())
.and_then(|project| project.to_str())
.filter(|project| !project.trim().is_empty())
{
projected.insert(
"project".to_string(),
serde_json::Value::String(project.to_string()),
);
}
for field in ["role", "bulletin"] {
if let Some(value) = session
.get(field)
.and_then(|value| value.as_str())
.filter(|value| !value.trim().is_empty())
{
projected.insert(
field.to_string(),
serde_json::Value::String(value.to_string()),
);
}
}
if let Some(reason) = session
.get("id")
.and_then(|id| id.as_str())
.and_then(quarantine_reason)
{
projected.insert("quarantined".to_string(), serde_json::Value::Bool(true));
projected
.insert("quarantine_reason".to_string(), serde_json::json!(reason));
}
serde_json::Value::Object(projected)
})
.collect::<Vec<_>>()
})
.unwrap_or_default();
if quarantined.is_empty() {
serde_json::json!({ "sessions": sessions })
} else {
serde_json::json!({
"sessions": sessions,
"quarantined_sessions": quarantined,
})
}
}
async fn cli_post(path: &str, body: &serde_json::Value) -> anyhow::Result<()> {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}{path}");
let client = reqwest::Client::new();
let resp = client.post(&url).json(body).send().await?;
let status = resp.status();
let text = resp.text().await?;
let body = classify_http_response(status, &text)?;
println!("{body}");
Ok(())
}
fn kill_response_succeeded(response: &serde_json::Value) -> bool {
response.get("outcome").and_then(|value| value.as_str()) == Some("removed")
}
fn restart_response_succeeded(response: &serde_json::Value) -> bool {
match response.get("outcome").and_then(|value| value.as_str()) {
Some(outcome) => outcome == "restarted",
None => true,
}
}
const DEFAULT_START_WAIT_SECS: u64 = 300;
const START_WAIT_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(750);
async fn cli_post_json(path: &str, body: &serde_json::Value) -> anyhow::Result<serde_json::Value> {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}{path}");
let client = reqwest::Client::new();
let resp = client.post(&url).json(body).send().await?;
let status = resp.status();
let text = resp.text().await?;
let body = classify_http_response(status, &text)?;
serde_json::from_str(&body)
.map_err(|error| anyhow::anyhow!("server response was not JSON: {error}: {body}"))
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct StartTicket {
session: String,
incarnation: String,
}
fn start_ticket_from_spawn_response(
name: &str,
response: &serde_json::Value,
) -> anyhow::Result<StartTicket> {
let incarnation = response
.get("incarnation")
.and_then(|value| value.as_str())
.ok_or_else(|| {
anyhow::anyhow!(
"server response has no 'incarnation' ticket, so the start outcome cannot be \
confirmed: {response}"
)
})?;
let session = response
.get("session")
.and_then(|value| value.as_str())
.unwrap_or(name);
Ok(StartTicket {
session: session.to_string(),
incarnation: incarnation.to_string(),
})
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct StartPoll {
terminal: bool,
started: bool,
line: String,
}
fn interpret_start_status(ticket: &StartTicket, body: &serde_json::Value) -> StartPoll {
let status = body.get("status").and_then(|value| value.as_str());
let detail = body.get("detail").and_then(|value| value.as_str());
let session = &ticket.session;
let incarnation = &ticket.incarnation;
match status {
Some("started") => StartPoll {
terminal: true,
started: true,
line: format!(
"started: {session} (incarnation {incarnation}){}",
detail.map(|d| format!(" — {d}")).unwrap_or_default()
),
},
Some("failed") => StartPoll {
terminal: true,
started: false,
line: format!(
"failed: {session} (incarnation {incarnation}): {}",
detail.unwrap_or("start failed without a recorded reason")
),
},
Some("superseded") => StartPoll {
terminal: true,
started: false,
line: format!(
"superseded: {session} (incarnation {incarnation}) was replaced by a newer \
incarnation{}",
detail.map(|d| format!(" — {d}")).unwrap_or_default()
),
},
Some("in_progress") => StartPoll {
terminal: false,
started: false,
line: format!("in progress: {session} (incarnation {incarnation})"),
},
Some("unknown") => StartPoll {
terminal: false,
started: false,
line: format!(
"unknown: no outcome retained for {session} (incarnation {incarnation}); it is \
not confirmed started"
),
},
other => StartPoll {
terminal: false,
started: false,
line: format!(
"unrecognized start status {:?} for {session} (incarnation {incarnation})",
other.unwrap_or("<missing>")
),
},
}
}
async fn await_start_outcome(
ticket: &StartTicket,
timeout: std::time::Duration,
) -> anyhow::Result<StartPoll> {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}/api/sessions/start-status");
let query = [
("session", ticket.session.as_str()),
("incarnation", ticket.incarnation.as_str()),
];
let client = reqwest::Client::new();
let deadline = std::time::Instant::now() + timeout;
loop {
let resp = client.get(&url).query(&query).send().await?;
let status = resp.status();
let text = resp.text().await?;
let body = classify_http_response(status, &text)?;
let value: serde_json::Value = serde_json::from_str(&body)
.map_err(|error| anyhow::anyhow!("start-status response was not JSON: {error}"))?;
let mut poll = interpret_start_status(ticket, &value);
if poll.terminal {
return Ok(poll);
}
let remaining = deadline.saturating_duration_since(std::time::Instant::now());
if remaining.is_zero() {
poll.line = format!("not confirmed within the wait budget — {}", poll.line);
return Ok(poll);
}
tokio::time::sleep(START_WAIT_POLL_INTERVAL.min(remaining)).await;
}
}
async fn cli_delete(path: &str) -> anyhow::Result<()> {
let port = std::env::var("OUIJA_PORT").unwrap_or_else(|_| "7880".to_string());
let url = format!("http://localhost:{port}{path}");
let client = reqwest::Client::new();
let resp = client.delete(&url).send().await?;
let status = resp.status();
let text = resp.text().await?;
let body = classify_http_response(status, &text)?;
println!("{body}");
Ok(())
}
fn classify_http_response(status: reqwest::StatusCode, body: &str) -> anyhow::Result<String> {
if status.is_success() {
Ok(body.to_string())
} else if body.is_empty() {
anyhow::bail!("request failed with HTTP {status}")
} else {
anyhow::bail!("request failed with HTTP {status}: {body}")
}
}
fn pane_wire_suffix(pane: &str) -> &str {
pane.strip_prefix('%').unwrap_or(pane)
}
const PATH_SEGMENT: &percent_encoding::AsciiSet = &percent_encoding::CONTROLS
.add(b' ')
.add(b'"')
.add(b'#')
.add(b'%')
.add(b'/')
.add(b'<')
.add(b'>')
.add(b'?')
.add(b'\\');
fn encode_path_segment(segment: &str) -> String {
percent_encoding::utf8_percent_encode(segment, PATH_SEGMENT).to_string()
}
fn dormant_show_path(id: &str) -> String {
format!(
"/api/session-identities/dormant/{}",
encode_path_segment(id)
)
}
fn read_prompt_file(path: &std::path::Path) -> anyhow::Result<String> {
std::fs::read_to_string(path)
.with_context(|| format!("failed to read UTF-8 prompt from {}", path.display()))
}
fn read_one_shot_file(path: &std::path::Path) -> anyhow::Result<String> {
read_prompt_file(path).with_context(|| {
format!(
"failed to read UTF-8 one-shot prompt from {}",
path.display()
)
})
}
#[cfg(test)]
mod tests {
use super::*;
use clap::CommandFactory as _;
fn persisted_dormant(
id: &str,
incarnation: u64,
backend: &str,
backend_session_id: &str,
project_dir: &str,
canonical_project_identity: &str,
) -> crate::daemon_protocol::DormantSession {
crate::daemon_protocol::DormantSession {
id: id.into(),
prior_owner: crate::daemon_protocol::ResourceOwner {
session_id: id.into(),
incarnation: crate::daemon_protocol::SessionIncarnation(incarnation),
},
metadata: crate::daemon_protocol::SessionMeta {
project_dir: Some(project_dir.into()),
canonical_project_identity: Some(canonical_project_identity.into()),
backend: Some(backend.into()),
backend_session_id: Some(backend_session_id.into()),
session_incarnation: crate::daemon_protocol::SessionIncarnation(incarnation),
..Default::default()
},
canonical_project_identity: canonical_project_identity.into(),
dormant_at: 1_753_920_123,
source: crate::daemon_protocol::DormancySource::Reaped,
}
}
fn subcommand_long_help(name: &str) -> String {
let mut command = Cli::command()
.find_subcommand(name)
.unwrap_or_else(|| panic!("missing {name} subcommand"))
.clone();
let mut help = Vec::new();
command
.write_long_help(&mut help)
.expect("render subcommand help");
String::from_utf8(help).expect("clap help is UTF-8")
}
#[test]
fn pane_wire_suffix_strips_leading_percent() {
assert_eq!(pane_wire_suffix("%74"), "74");
}
#[test]
fn a_restart_response_only_counts_an_explicit_restarted_outcome() {
assert!(restart_response_succeeded(
&serde_json::json!({ "result": "restarted 'w'", "outcome": "restarted" })
));
assert!(!restart_response_succeeded(
&serde_json::json!({ "result": "restart failed: ...", "outcome": "failed" })
));
assert!(!restart_response_succeeded(
&serde_json::json!({ "outcome": "superseded" })
));
assert!(restart_response_succeeded(
&serde_json::json!({ "result": "restarted 'w'" })
));
}
#[test]
fn a_spawn_response_yields_the_owner_scoped_ticket() {
let response = serde_json::json!({
"session": "tp-64",
"status": "starting",
"incarnation": "1784923380514537362"
});
let ticket = start_ticket_from_spawn_response("tp-64", &response).expect("ticket");
assert_eq!(ticket.session, "tp-64");
assert_eq!(ticket.incarnation, "1784923380514537362");
}
#[test]
fn a_spawn_response_without_a_ticket_is_an_error_not_a_silent_success() {
let response = serde_json::json!({ "session": "tp-64", "status": "starting" });
let error = start_ticket_from_spawn_response("tp-64", &response)
.expect_err("a missing ticket must not degrade to acceptance");
assert!(error.to_string().contains("incarnation"));
}
#[test]
fn cli_interprets_each_start_status_distinctly() {
let ticket = StartTicket {
session: "tp-64".into(),
incarnation: "7".into(),
};
let started = interpret_start_status(
&ticket,
&serde_json::json!({ "status": "started", "detail": "started 'tp-64' in /tmp (pane %151)" }),
);
assert!(started.terminal && started.started);
assert!(started.line.contains("pane %151"));
let reason = "start failed: OpenCode attach setup failed for 'tp-64' (pane %150)";
let failed = interpret_start_status(
&ticket,
&serde_json::json!({ "status": "failed", "detail": reason }),
);
assert!(failed.terminal);
assert!(!failed.started, "a failed start must not read as staffed");
assert!(
failed.line.contains(reason),
"the logged reason must reach the caller, got: {}",
failed.line
);
let in_progress =
interpret_start_status(&ticket, &serde_json::json!({ "status": "in_progress" }));
assert!(!in_progress.terminal && !in_progress.started);
let superseded =
interpret_start_status(&ticket, &serde_json::json!({ "status": "superseded" }));
assert!(superseded.terminal);
assert!(!superseded.started);
let unknown = interpret_start_status(&ticket, &serde_json::json!({ "status": "unknown" }));
assert!(!unknown.terminal && !unknown.started);
}
#[test]
fn an_unrecognized_start_status_never_reads_as_started() {
let ticket = StartTicket {
session: "tp-64".into(),
incarnation: "7".into(),
};
for body in [
serde_json::json!({}),
serde_json::json!({ "status": "who-knows" }),
serde_json::json!({ "status": 42 }),
] {
let poll = interpret_start_status(&ticket, &body);
assert!(!poll.started, "unrecognized status must not claim success");
assert!(!poll.terminal);
}
}
#[test]
fn dormant_resume_cli_accepts_wait_and_exact_id() {
let cli = Cli::try_parse_from([
"ouija",
"dormant",
"resume",
"feat/exact-id",
"--wait",
"--wait-timeout",
"2m",
])
.expect("dormant resume parses");
match cli.command {
Command::Dormant {
command:
DormantCommand::Resume {
id,
wait,
wait_timeout,
},
} => {
assert_eq!(id, "feat/exact-id");
assert!(wait);
assert_eq!(wait_timeout, Some(120));
}
_ => panic!("expected dormant resume command"),
}
}
#[test]
fn dormant_resume_cli_rejects_wait_timeout_without_wait() {
assert!(
Cli::try_parse_from([
"ouija",
"dormant",
"resume",
"worker",
"--wait-timeout",
"2m",
])
.is_err()
);
}
#[test]
fn spawn_session_cli_accepts_wait_and_a_wait_budget() {
let cli = Cli::try_parse_from([
"ouija",
"spawn-session",
"tp-64",
"--wait",
"--wait-timeout",
"2m",
])
.expect("spawn-session --wait parses");
match cli.command {
Command::SpawnSession {
wait, wait_timeout, ..
} => {
assert!(wait);
assert_eq!(wait_timeout, Some(120));
}
_ => panic!("expected spawn-session command"),
}
}
#[test]
fn spawn_session_cli_rejects_a_wait_budget_without_wait() {
assert!(
Cli::try_parse_from(["ouija", "spawn-session", "tp-64", "--wait-timeout", "2m"])
.is_err()
);
}
#[test]
fn start_status_cli_takes_the_exact_incarnation() {
let cli = Cli::try_parse_from([
"ouija",
"start-status",
"tp-64",
"--incarnation",
"1784923380514537362",
])
.expect("start-status parses");
match cli.command {
Command::StartStatus {
name,
incarnation,
wait,
} => {
assert_eq!(name, "tp-64");
assert_eq!(incarnation, "1784923380514537362");
assert_eq!(wait, None);
}
_ => panic!("expected start-status command"),
}
}
#[test]
fn pane_wire_suffix_leaves_bare_suffix_alone() {
assert_eq!(pane_wire_suffix("74"), "74");
}
#[test]
fn pane_wire_suffix_only_strips_first_percent() {
assert_eq!(pane_wire_suffix("%%74"), "%74");
}
#[test]
fn pane_wire_suffix_handles_empty_string() {
assert_eq!(pane_wire_suffix(""), "");
}
#[test]
fn recover_backend_identity_cli_requires_one_exact_public_id() {
let cli =
Cli::try_parse_from(["ouija", "recover-backend-identity", "divine-invite-darshan"])
.expect("explicit recovery command must parse");
match cli.command {
Command::RecoverBackendIdentity { session_id } => {
assert_eq!(session_id, "divine-invite-darshan");
}
_ => panic!("expected recover-backend-identity command"),
}
assert!(Cli::try_parse_from(["ouija", "recover-backend-identity"]).is_err());
}
#[test]
fn claim_cli_requires_one_exact_requested_id() {
let cli = Cli::try_parse_from(["ouija", "claim", "durable-worker"])
.expect("explicit claim command must parse");
match cli.command {
Command::Claim { requested_id } => {
assert_eq!(requested_id, "durable-worker");
}
_ => panic!("expected claim command"),
}
assert!(Cli::try_parse_from(["ouija", "claim"]).is_err());
assert!(Cli::try_parse_from(["ouija", "claim", "one", "two"]).is_err());
}
#[test]
fn claim_request_preserves_independent_evidence_and_opaque_backend_id() {
let request = claim_request(
"durable-worker".into(),
Some("%712".into()),
Some("pane-owner".into()),
Some("stale-env-owner".into()),
backend::BackendSessionIdentity {
backend: "codex-cli".into(),
session_id: "opaque-thread-id".into(),
},
);
assert_eq!(request.requested_id, "durable-worker");
assert_eq!(request.caller.pane.as_deref(), Some("%712"));
assert_eq!(request.caller.pane_var_id.as_deref(), Some("pane-owner"));
assert_eq!(request.caller.env_id.as_deref(), Some("stale-env-owner"));
assert_eq!(
request.caller.backend_identity.session_id,
"opaque-thread-id"
);
let json = serde_json::to_value(request).unwrap();
assert_eq!(
json["caller"]["backend_identity"]["session_id"],
"opaque-thread-id"
);
assert!(
json.to_string().contains("opaque-thread-id"),
"backend-native identity belongs only in the request body"
);
}
#[test]
fn dormant_cli_parses_list_and_exact_show_id() {
let list = Cli::try_parse_from(["ouija", "dormant", "list"]).unwrap();
assert!(matches!(
list.command,
Command::Dormant {
command: DormantCommand::List
}
));
let show = Cli::try_parse_from(["ouija", "dormant", "show", "feat/identity"]).unwrap();
match show.command {
Command::Dormant {
command: DormantCommand::Show { id },
} => assert_eq!(id, "feat/identity"),
_ => panic!("expected dormant show command"),
}
assert!(Cli::try_parse_from(["ouija", "dormant", "show"]).is_err());
}
#[test]
fn dormant_show_path_encodes_exact_id_as_one_segment() {
assert_eq!(
dormant_show_path("feat/identity %#"),
"/api/session-identities/dormant/feat%2Fidentity%20%25%23"
);
}
#[test]
fn dormant_get_non_success_is_a_cli_error() {
let error = classify_http_response(
reqwest::StatusCode::NOT_FOUND,
r#"{"outcome":"dormant_not_found"}"#,
)
.unwrap_err()
.to_string();
assert!(error.contains("404"));
assert!(error.contains("dormant_not_found"));
}
#[test]
fn backend_recovery_evidence_preserves_every_positive_signal_without_whoami_resolution() {
let evidence = backend_recovery_caller_evidence(
Some("%712".into()),
Some("pane-owner".into()),
Some("stale-env-owner".into()),
);
assert_eq!(evidence.pane.as_deref(), Some("%712"));
assert_eq!(evidence.pane_var_id.as_deref(), Some("pane-owner"));
assert_eq!(evidence.env_id.as_deref(), Some("stale-env-owner"));
}
#[test]
fn spawn_session_cli_preserves_legacy_idle_policy_values() {
let cli = Cli::try_parse_from([
"ouija",
"spawn-session",
"worker",
"--parent-session",
"parent",
"--idle-policy",
"ask-parent-when-done",
])
.expect("spawn-session lifecycle args parse");
match cli.command {
Command::SpawnSession {
parent_session,
no_parent_session,
idle_policy,
..
} => {
assert_eq!(parent_session.as_deref(), Some("parent"));
assert!(!no_parent_session);
assert_eq!(
idle_policy,
Some(crate::daemon_protocol::IdlePolicy::AskParentWhenDone)
);
}
_ => panic!("expected spawn-session command"),
}
}
#[test]
fn spawn_session_cli_accepts_a_positive_active_context_duration() {
let parsed = Cli::try_parse_from([
"ouija",
"spawn-session",
"worker",
"--fresh-context-after-active",
"1h",
"--no-parent-session",
"--when-done",
"keep-open",
]);
match parsed
.expect("a positive active-context duration must parse")
.command
{
Command::SpawnSession {
fresh_context_after_active,
..
} => assert_eq!(fresh_context_after_active, Some(3_600)),
_ => panic!("expected spawn-session command"),
}
}
#[test]
fn spawn_session_help_explains_first_launch_prompt_and_active_context_setup() {
let help = subcommand_long_help("spawn-session");
assert!(help.to_lowercase().contains("complete bounded assignment"));
assert!(help.contains("Bootstrap active-context refresh"));
assert!(help.contains("no `ouija spawn` alias"));
assert!(help.contains("does not support `--one-shot-file`"));
}
#[test]
fn active_context_duration_parser_normalizes_supported_units() {
assert_eq!(parse_fresh_context_after_active("1h"), Ok(3_600));
assert_eq!(parse_fresh_context_after_active("90m"), Ok(5_400));
assert_eq!(parse_fresh_context_after_active("3600s"), Ok(3_600));
}
#[test]
fn active_context_duration_parser_rejects_invalid_or_non_positive_values() {
assert_eq!(
parse_fresh_context_after_active(""),
Err("duration must be a positive whole number followed by h, m, or s".into())
);
assert_eq!(
parse_fresh_context_after_active("0h"),
Err("duration must be greater than zero".into())
);
assert_eq!(
parse_fresh_context_after_active("1"),
Err("duration must end with h, m, or s".into())
);
assert_eq!(
parse_fresh_context_after_active("1.5h"),
Err("duration must be a positive whole number followed by h, m, or s".into())
);
assert_eq!(
parse_fresh_context_after_active("18446744073709551616s"),
Err("duration value is too large".into())
);
assert_eq!(
parse_fresh_context_after_active("5124095576030432h"),
Err("duration overflows seconds".into())
);
}
#[test]
fn restart_session_cli_requires_fresh_for_active_context_duration() {
let parsed = Cli::try_parse_from([
"ouija",
"restart-session",
"worker",
"--fresh-context-after-active",
"1h",
]);
assert!(parsed.is_err());
}
#[test]
fn inject_only_task_cli_requires_and_preserves_exact_target() {
let missing_target = Cli::try_parse_from([
"ouija",
"task",
"add",
"context-audit",
"*/15 * * * *",
"audit",
"--inject-only",
]);
assert!(missing_target.is_err());
let cli = Cli::try_parse_from([
"ouija",
"task",
"add",
"context-audit",
"*/15 * * * *",
"audit",
"--target",
"manual-root",
"--inject-only",
])
.expect("inject-only task parses with exact target");
match cli.command {
Command::Task {
action:
TaskAction::Add {
target,
inject_only,
..
},
} => {
assert_eq!(target.as_deref(), Some("manual-root"));
assert!(inject_only);
}
_ => panic!("expected task add command"),
}
}
#[test]
fn restart_session_cli_accepts_scoped_prompt_controls_and_backend() {
let cli = Cli::try_parse_from([
"ouija",
"restart-session",
"worker",
"--prompt-file",
"/tmp/base.txt",
"--suppress-stored-prompt",
"--one-shot-file",
"/tmp/adopt.txt",
"--backend",
"codex-cli",
])
.expect("restart prompt controls must parse");
match cli.command {
Command::RestartSession {
prompt_file,
suppress_stored_prompt,
one_shot_file,
backend,
..
} => {
assert_eq!(
prompt_file.as_deref(),
Some(std::path::Path::new("/tmp/base.txt"))
);
assert!(suppress_stored_prompt);
assert_eq!(
one_shot_file.as_deref(),
Some(std::path::Path::new("/tmp/adopt.txt"))
);
assert_eq!(backend.as_deref(), Some("codex-cli"));
}
_ => panic!("expected restart-session command"),
}
}
#[test]
fn restart_session_help_separates_durable_prompt_from_launch_only_context() {
let help = subcommand_long_help("restart-session");
let spawn_help = subcommand_long_help("spawn-session");
assert!(help.contains("replacement durable stored base prompt"));
assert!(help.contains("replayed by default after every fresh restart"));
assert!(help.contains("UTF-8 file"));
assert!(help.contains("quoted heredoc"));
assert!(help.contains("verified current-work continuation"));
assert!(help.contains("binding is absent or cannot be trusted"));
assert!(spawn_help.contains("re-entrant, state-checking"));
assert!(spawn_help.contains("UTF-8 file"));
assert!(spawn_help.contains("quoted heredoc"));
}
#[test]
fn one_shot_file_reader_accepts_utf8_and_rejects_invalid_bytes() {
let dir = tempfile::tempdir().unwrap();
let valid = dir.path().join("valid.txt");
std::fs::write(&valid, "verify live state").unwrap();
assert_eq!(read_one_shot_file(&valid).unwrap(), "verify live state");
let invalid = dir.path().join("invalid.txt");
std::fs::write(&invalid, [0xff, 0xfe]).unwrap();
let error = read_one_shot_file(&invalid).unwrap_err().to_string();
assert!(error.contains("UTF-8 one-shot prompt"));
let missing = dir.path().join("missing.txt");
let error = read_one_shot_file(&missing).unwrap_err().to_string();
assert!(error.contains(missing.to_string_lossy().as_ref()));
}
#[test]
fn compact_session_list_preserves_decimal_incarnation() {
let projected = project_session_list(&serde_json::json!({
"sessions": [{
"id": "worker",
"origin": "local",
"session_incarnation": "18446744073709551615"
}]
}));
assert_eq!(
projected["sessions"][0]["session_incarnation"],
"18446744073709551615"
);
}
#[test]
fn rollover_cli_requires_stdin_and_accepts_expired_replacement() {
assert!(Cli::try_parse_from(["ouija", "rollover", "prepare"]).is_err());
let cli = Cli::try_parse_from([
"ouija",
"rollover",
"prepare",
"--stdin",
"--replace-expired",
])
.unwrap();
match cli.command {
Command::Rollover {
action:
RolloverAction::Prepare {
stdin,
replace_expired,
},
} => {
assert!(stdin);
assert!(replace_expired);
}
_ => panic!("expected rollover prepare"),
}
let cli = Cli::try_parse_from(["ouija", "rollover", "adopt", "opaque-token"]).unwrap();
match cli.command {
Command::Rollover {
action: RolloverAction::Adopt { token },
} => assert_eq!(token, "opaque-token"),
_ => panic!("expected rollover adopt"),
}
let cli = Cli::try_parse_from(["ouija", "rollover", "cleanup", "--force-pending"]).unwrap();
match cli.command {
Command::Rollover {
action: RolloverAction::Cleanup { force_pending },
} => assert!(force_pending),
_ => panic!("expected rollover cleanup"),
}
}
#[test]
fn rollover_owner_parser_requires_exact_local_string_incarnation() {
let session = serde_json::json!({
"id": "worker",
"origin": "local",
"session_incarnation": "18446744073709551615",
});
assert_eq!(
rollover_incarnation_from_session(&session, "worker").unwrap(),
u64::MAX
);
assert!(rollover_incarnation_from_session(&session, "other").is_err());
let mut remote = session.clone();
remote["origin"] = serde_json::json!("remote");
assert!(rollover_incarnation_from_session(&remote, "worker").is_err());
let mut numeric = session;
numeric["session_incarnation"] = serde_json::json!(7);
assert!(rollover_incarnation_from_session(&numeric, "worker").is_err());
}
#[test]
fn rollover_incarnation_evidence_rejects_absent_unbound_caller() {
let error = verify_rollover_incarnation_evidence(None, false, 9)
.unwrap_err()
.to_string();
assert!(error.contains("caller-owned incarnation evidence"));
}
#[test]
fn rollover_incarnation_evidence_rejects_stale_hint() {
let error = verify_rollover_incarnation_evidence(Some(8), false, 9)
.unwrap_err()
.to_string();
assert!(error.contains("does not match"));
}
#[test]
fn rollover_incarnation_evidence_accepts_exact_hint_or_bound_backend() {
assert_eq!(
verify_rollover_incarnation_evidence(Some(9), false, 9).unwrap(),
9
);
assert_eq!(
verify_rollover_incarnation_evidence(None, true, 9).unwrap(),
9
);
}
#[test]
fn spawn_session_cli_accepts_primary_when_done_values() {
for (value, expected) in [
("keep-open", IdlePolicy::KeepOpen),
("ask-parent", IdlePolicy::AskParentWhenDone),
("close", IdlePolicy::CloseWhenDone),
] {
let cli = Cli::try_parse_from([
"ouija",
"spawn-session",
"worker",
"--parent-session",
"parent",
"--when-done",
value,
])
.unwrap_or_else(|error| panic!("--when-done {value} must parse: {error}"));
match cli.command {
Command::SpawnSession {
when_done,
idle_policy,
..
} => {
assert_eq!(when_done.map(IdlePolicy::from), Some(expected));
assert_eq!(idle_policy, None);
}
_ => panic!("expected spawn-session command"),
}
}
}
#[test]
fn spawn_session_cli_rejects_both_completion_flags() {
let error = Cli::try_parse_from([
"ouija",
"spawn-session",
"worker",
"--no-parent-session",
"--when-done",
"keep-open",
"--idle-policy",
"keep-open",
])
.err()
.expect("completion flags must conflict")
.to_string();
assert!(error.contains("--when-done"));
assert!(error.contains("--idle-policy"));
assert!(
error.contains("cannot be used with"),
"error must explain the conflict, got: {error}"
);
}
#[test]
fn spawn_session_help_documents_primary_and_deprecated_completion_flags() {
use clap::CommandFactory;
let mut cmd = Cli::command();
let spawn_session = cmd
.find_subcommand_mut("spawn-session")
.expect("spawn-session subcommand exists");
let mut help = Vec::new();
spawn_session.write_long_help(&mut help).unwrap();
let help = String::from_utf8(help).unwrap();
assert!(help.contains("--when-done <WHEN_DONE>"));
for value in ["keep-open", "ask-parent", "close"] {
assert!(
help.contains(value),
"primary completion value {value} missing from help:\n{help}"
);
}
assert!(help.contains("--idle-policy <IDLE_POLICY>"));
assert!(
help.contains("Deprecated"),
"legacy flag must be marked deprecated:\n{help}"
);
for value in ["keep-open", "ask-parent-when-done", "close-when-done"] {
assert!(
help.contains(value),
"legacy completion value {value} missing from help:\n{help}"
);
}
}
#[test]
fn spawn_session_cli_rejects_manual_clear_reminder_commands() {
let error = Cli::try_parse_from([
"ouija",
"spawn-session",
"worker",
"--no-parent-session",
"--idle-policy",
"keep-open",
"--reminder",
"When done, run ouija clear-reminder 7",
])
.err()
.expect("manual clear-reminder instructions must be rejected")
.to_string();
assert!(error.contains("ouija clear-reminder"));
assert!(
error.contains("generated"),
"error must explain that Ouija supplies the command, got: {error}"
);
}
#[test]
fn restart_session_cli_rejects_manual_clear_reminder_commands() {
let error = Cli::try_parse_from([
"ouija",
"restart-session",
"worker",
"--reminder",
"When done, run ouija clear-reminder 7",
])
.err()
.expect("manual clear-reminder instructions must be rejected");
assert!(error.to_string().contains("ouija clear-reminder"));
}
#[test]
fn ask_cli_requires_message_file() {
let cli = Cli::try_parse_from([
"ouija",
"ask",
"parent",
"--message-file",
"/dev/stdin",
"--from",
"worker",
])
.expect("ask --message-file parses");
match cli.command {
Command::Ask {
to,
message_file,
from,
} => {
assert_eq!(to, "parent");
assert_eq!(message_file, PathBuf::from("/dev/stdin"));
assert_eq!(from.as_deref(), Some("worker"));
}
_ => panic!("expected ask command"),
}
assert!(Cli::try_parse_from(["ouija", "ask", "parent", "hello"]).is_err());
assert!(Cli::try_parse_from(["ouija", "ask", "parent", "--stdin"]).is_err());
assert!(Cli::try_parse_from(["ouija", "tell", "parent", "hello"]).is_err());
assert!(Cli::try_parse_from(["ouija", "tell", "parent", "--stdin"]).is_err());
assert!(Cli::try_parse_from(["ouija", "reply", "parent", "47", "hello"]).is_err());
assert!(Cli::try_parse_from(["ouija", "reply", "parent", "47", "--stdin"]).is_err());
}
#[test]
fn spawn_lifecycle_validation_teaches_missing_parent_choice() {
let err = validate_spawn_lifecycle(
None,
false,
Some(&crate::daemon_protocol::IdlePolicy::KeepOpen),
)
.unwrap_err();
assert!(
err.contains("--parent-session <SESSION_ID>"),
"error must teach parent-session choice, got: {err}"
);
assert!(
err.contains("--no-parent-session"),
"error must teach no-parent-session choice, got: {err}"
);
}
#[test]
fn spawn_lifecycle_validation_teaches_missing_idle_policy() {
let err = validate_spawn_lifecycle(None, true, None).unwrap_err();
assert!(
err.contains("--when-done <keep-open|ask-parent|close>"),
"error must teach when-done choices, got: {err}"
);
}
#[test]
fn daemon_log_survives_a_restart_instead_of_being_truncated() {
let dir = tempfile::tempdir().unwrap();
{
use std::io::Write;
let mut first = open_daemon_log(dir.path()).unwrap();
writeln!(first, "before restart").unwrap();
}
{
use std::io::Write;
let mut second = open_daemon_log(dir.path()).unwrap();
writeln!(second, "after restart").unwrap();
}
let contents = std::fs::read_to_string(dir.path().join("daemon.log")).unwrap();
assert!(
contents.contains("before restart") && contents.contains("after restart"),
"restart must append, not truncate; got: {contents:?}"
);
}
#[test]
fn daemon_log_rotates_once_past_the_size_cap() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("daemon.log");
std::fs::write(&path, vec![b'x'; (DAEMON_LOG_MAX_BYTES + 1) as usize]).unwrap();
let _ = open_daemon_log(dir.path()).unwrap();
assert!(
dir.path().join("daemon.log.1").exists(),
"oversized log must be rotated aside, not deleted"
);
assert_eq!(
std::fs::metadata(&path).unwrap().len(),
0,
"the fresh log starts empty after rotation"
);
}
#[test]
fn encode_path_segment_encodes_slashes() {
assert_eq!(
encode_path_segment("feat/646-foo"),
"feat%2F646-foo",
"`/` must be percent-encoded so the URL stays a single path segment"
);
}
#[test]
fn encode_path_segment_passes_through_common_session_chars() {
assert_eq!(encode_path_segment("my-session_42"), "my-session_42");
}
#[test]
fn encode_path_segment_encodes_percent_literal() {
assert_eq!(encode_path_segment("100%"), "100%25");
}
#[test]
fn encode_path_segment_encodes_space_and_hash() {
assert_eq!(encode_path_segment("a b#c?d"), "a%20b%23c%3Fd");
}
#[test]
fn clear_reminder_help_mentions_from_option() {
use clap::CommandFactory;
let mut cmd = Cli::command();
let clear_reminder = cmd
.find_subcommand_mut("clear-reminder")
.expect("clear-reminder subcommand exists");
let mut help = Vec::new();
clear_reminder.write_long_help(&mut help).unwrap();
let help = String::from_utf8(help).unwrap();
assert!(
help.contains("Usage: clear-reminder [OPTIONS] <CLEARING_ID>"),
"clear-reminder usage must advertise options, got:\n{help}"
);
assert!(
help.contains("--from <FROM>"),
"clear-reminder help must advertise explicit sender support, got:\n{help}"
);
}
#[test]
fn clear_reminder_parses_explicit_from_option() {
let cli = Cli::try_parse_from([
"ouija",
"clear-reminder",
"42",
"--from",
"feat/62-add-from-support-to-ouija-clear-reminder",
])
.unwrap();
match cli.command {
Command::ClearReminder { clearing_id, from } => {
assert_eq!(clearing_id, 42);
assert_eq!(
from.as_deref(),
Some("feat/62-add-from-support-to-ouija-clear-reminder")
);
}
_ => panic!("expected clear-reminder command"),
}
}
#[test]
fn rename_parses_explicit_from_option() {
let cli = Cli::try_parse_from(["ouija", "rename", "hub-cx", "--from", "hub"]).unwrap();
match cli.command {
Command::Rename { new_id, from } => {
assert_eq!(new_id, "hub-cx");
assert_eq!(from.as_deref(), Some("hub"));
}
_ => panic!("expected rename command"),
}
}
#[test]
fn classify_http_response_success_returns_body() {
use reqwest::StatusCode;
let out = classify_http_response(StatusCode::OK, "{\"ok\":true}").unwrap();
assert_eq!(out, "{\"ok\":true}");
}
#[test]
fn classify_http_response_2xx_range_all_pass() {
use reqwest::StatusCode;
for code in [
StatusCode::OK,
StatusCode::CREATED,
StatusCode::ACCEPTED,
StatusCode::NO_CONTENT,
] {
assert!(
classify_http_response(code, "").is_ok(),
"{code} must be classified as success"
);
}
}
#[test]
fn pick_session_id_prefers_pane_var_over_env_var_in_tmux() {
let res = pick_session_id(
Some("%74"),
Some("keycast".into()),
Some("feat/95-stale".into()),
);
assert_eq!(
res,
SessionIdResolution::Found("keycast".into(), IdentitySource::PaneVar)
);
}
#[test]
fn pick_session_id_falls_back_to_env_var_when_pane_var_missing() {
let res = pick_session_id(Some("%74"), None, Some("keycast".into()));
assert_eq!(
res,
SessionIdResolution::Found("keycast".into(), IdentitySource::EnvVar)
);
}
#[test]
fn pick_session_id_treats_empty_pane_var_as_absent() {
let res = pick_session_id(Some("%74"), Some("".into()), Some("env-id".into()));
assert_eq!(
res,
SessionIdResolution::Found("env-id".into(), IdentitySource::EnvVar)
);
}
#[test]
fn pick_session_id_falls_through_to_pane_lookup_when_neither_signal_set() {
let res = pick_session_id(Some("%74"), None, None);
assert_eq!(res, SessionIdResolution::LookupByPane("%74".into()));
}
#[test]
fn pick_session_id_outside_tmux_uses_env_var() {
let res = pick_session_id(None, None, Some("opencode-session".into()));
assert_eq!(
res,
SessionIdResolution::Found("opencode-session".into(), IdentitySource::EnvVar)
);
}
#[test]
fn pick_session_id_outside_tmux_with_no_env_var_returns_none() {
let res = pick_session_id(None, None, None);
assert_eq!(res, SessionIdResolution::None);
}
#[test]
fn pick_session_id_outside_tmux_ignores_stray_pane_var_input() {
let res = pick_session_id(None, Some("ghost".into()), Some("real".into()));
assert_eq!(
res,
SessionIdResolution::Found("real".into(), IdentitySource::EnvVar)
);
}
#[test]
fn explicit_sender_remains_authoritative_across_backend_thread_rollover() {
let sender = resolve_explicit_sender(
"hub-4".into(),
Some("%replacement".into()),
None,
None,
Some(backend::BackendSessionIdentity {
backend: "codex-cli".into(),
session_id: "new-thread".into(),
}),
);
assert_eq!(sender.id, "hub-4");
assert_eq!(sender.context["trusted_local_claim"], true);
assert_eq!(sender.context["pane"], "%replacement");
assert!(sender.context["self_id"].is_null());
assert_eq!(
sender.context["backend_identity"],
serde_json::json!({
"backend": "codex-cli",
"session_id": "new-thread",
})
);
}
#[test]
fn explicit_sender_preserves_registered_env_observation_for_daemon_validation() {
let sender =
resolve_explicit_sender("hub-4".into(), None, None, Some("sibling".into()), None);
assert_eq!(sender.id, "hub-4");
assert_eq!(sender.context["self_id"], "sibling");
assert_eq!(sender.context["trusted_local_claim"], true);
}
#[test]
fn backend_canonical_identity_rejects_a_conflicting_local_signal() {
let err = arbitrate_backend_identity(
Some(("stale-pane-id".into(), IdentitySource::PaneVar)),
Some("canonical-backend-id".into()),
)
.unwrap_err();
assert!(err.local_id.contains("stale-pane-id"));
assert!(err.canonical_id.contains("canonical-backend-id"));
}
#[test]
fn backend_canonical_identity_is_reported_when_local_signal_agrees() {
let resolved = arbitrate_backend_identity(
Some(("canonical-backend-id".into(), IdentitySource::EnvVar)),
Some("canonical-backend-id".into()),
)
.unwrap();
assert_eq!(
resolved,
Some((
"canonical-backend-id".into(),
IdentitySource::BackendIdentity
))
);
}
#[test]
fn incomplete_backend_identity_does_not_strand_verified_local_identity() {
let local = ("hub".into(), IdentitySource::PaneVar);
let backend_canonical = backend_canonical_for_arbitration(
Some(&local),
Err(BackendIdentityLookupError::daemon_rejection(
"incomplete_legacy",
"legacy backend metadata is incomplete",
)),
)
.unwrap();
let resolved = arbitrate_backend_identity(Some(local), backend_canonical).unwrap();
assert_eq!(
resolved,
Some(("hub".into(), IdentitySource::PaneVar)),
"a non-canonical incomplete row cannot disprove a verified local identity"
);
}
#[test]
fn incomplete_backend_identity_does_not_strand_registered_env_identity() {
let local = ("hub".into(), IdentitySource::EnvVar);
let backend_canonical = backend_canonical_for_arbitration(
Some(&local),
Err(BackendIdentityLookupError::daemon_rejection(
"incomplete_legacy",
"legacy backend metadata is incomplete",
)),
)
.unwrap();
let resolved = arbitrate_backend_identity(Some(local), backend_canonical).unwrap();
assert_eq!(resolved, Some(("hub".into(), IdentitySource::EnvVar)));
}
#[test]
fn incomplete_backend_identity_without_local_proof_remains_terminal() {
let error = backend_canonical_for_arbitration(
None,
Err(BackendIdentityLookupError::daemon_rejection(
"incomplete_legacy",
"legacy backend metadata is incomplete",
)),
)
.unwrap_err();
assert_eq!(error.outcome.as_deref(), Some("incomplete_legacy"));
}
#[test]
fn non_incomplete_backend_failures_remain_terminal_with_local_hint() {
let local = ("hub".into(), IdentitySource::PaneLookup);
for outcome in ["ambiguous", "not_found"] {
let error = backend_canonical_for_arbitration(
Some(&local),
Err(BackendIdentityLookupError::daemon_rejection(
outcome,
"backend identity has no safe canonical owner",
)),
)
.unwrap_err();
assert_eq!(error.outcome.as_deref(), Some(outcome));
}
let transport_error = backend_canonical_for_arbitration(
Some(&local),
Err(BackendIdentityLookupError::protocol_failure(
"daemon unreachable",
)),
)
.unwrap_err();
assert!(transport_error.outcome.is_none());
}
#[test]
fn read_message_file_preserves_shell_metacharacters() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("message.txt");
std::fs::write(&path, "hello `literal` $(literal)\n").unwrap();
let message = read_message_file(&path).unwrap();
assert_eq!(message, "hello `literal` $(literal)\n");
}
#[test]
fn prompt_file_preserves_shell_metacharacters_and_newlines() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("prompt.txt");
std::fs::write(
&path,
"Review `DANIEL HANDOFF` and $(literal).\nNext line.\n",
)
.unwrap();
let prompt = read_prompt_file(&path).unwrap();
assert_eq!(
prompt,
"Review `DANIEL HANDOFF` and $(literal).\nNext line.\n"
);
}
#[test]
fn spawn_and_restart_reject_legacy_inline_prompt() {
for command in ["spawn-session", "restart-session"] {
let error = Cli::try_parse_from(["ouija", command, "worker", "--prompt", "`DANIEL`"])
.err()
.expect("inline prompt must be rejected")
.to_string();
assert!(error.contains("--prompt"));
}
}
#[test]
fn spawn_session_accepts_prompt_file() {
let cli = Cli::try_parse_from([
"ouija",
"spawn-session",
"worker",
"--prompt-file",
"/dev/stdin",
])
.expect("spawn prompt file must parse");
match cli.command {
Command::SpawnSession { prompt_file, .. } => {
assert_eq!(
prompt_file.as_deref(),
Some(std::path::Path::new("/dev/stdin"))
);
}
_ => panic!("expected spawn-session command"),
}
}
#[test]
fn kill_session_preserves_worktree_by_default_and_accepts_explicit_delete() {
let default = Cli::try_parse_from(["ouija", "kill-session", "worker"]).unwrap();
assert!(matches!(
default.command,
Command::KillSession {
delete_worktree: false,
keep_worktree: false,
..
}
));
let delete =
Cli::try_parse_from(["ouija", "kill-session", "worker", "--delete-worktree"]).unwrap();
assert!(matches!(
delete.command,
Command::KillSession {
delete_worktree: true,
..
}
));
assert!(
Cli::try_parse_from([
"ouija",
"kill-session",
"worker",
"--delete-worktree",
"--keep-worktree",
])
.is_err()
);
}
#[test]
fn kill_response_requires_removed_outcome() {
assert!(kill_response_succeeded(&serde_json::json!({
"outcome": "removed"
})));
assert!(!kill_response_succeeded(&serde_json::json!({
"outcome": "failed"
})));
assert!(!kill_response_succeeded(&serde_json::json!({})));
}
#[test]
fn classify_http_response_404_surfaces_as_error() {
use reqwest::StatusCode;
let err = classify_http_response(
StatusCode::NOT_FOUND,
"{\"error\":\"pane 'x' is not registered\"}",
)
.unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("404"),
"error message must include the status code, got: {msg}"
);
assert!(
msg.contains("pane 'x' is not registered"),
"error message must include the response body, got: {msg}"
);
}
#[test]
fn classify_http_response_500_surfaces_as_error() {
use reqwest::StatusCode;
let err = classify_http_response(StatusCode::INTERNAL_SERVER_ERROR, "boom").unwrap_err();
assert!(err.to_string().contains("500"));
}
#[test]
fn classify_http_response_400_with_empty_body_still_errors() {
use reqwest::StatusCode;
let err = classify_http_response(StatusCode::BAD_REQUEST, "").unwrap_err();
assert!(err.to_string().contains("400"));
}
#[test]
fn session_list_projection_keeps_discovery_fields_only() {
let status = serde_json::json!({
"daemon": "locota",
"transports": [{"name": "nostr", "ready": true}],
"assistant_panes": [{"pane_id": "%1", "session": "ouija"}],
"sessions": [{
"id": "ouija-next-issue",
"origin": "local",
"session_incarnation": "42",
"parent_session": "hub-cx",
"project_dir": "/home/daniel/code/ouija",
"role": "working on ouija",
"bulletin": "ready",
"stale": true,
"worktree_present": true,
"fresh_context_after_active_secs": 3600,
"active_context_accumulated_secs": 1234,
"active_context_segment_open": true,
"active_context_restart_due": true,
"prompt": "internal prompt that should not be listed",
"reminder": "internal reminder that should not be listed",
"backend_session_id": "ses_secret_internal",
"iteration_log": ["noise"]
}]
});
let projected = project_session_list(&status);
assert_eq!(
projected,
serde_json::json!({
"sessions": [{
"id": "ouija-next-issue",
"origin": "local",
"session_incarnation": "42",
"parent_session": "hub-cx",
"project": "ouija",
"role": "working on ouija",
"bulletin": "ready"
}]
})
);
assert!(projected.get("daemon").is_none());
assert!(projected["sessions"][0].get("project_dir").is_none());
assert!(projected["sessions"][0].get("stale").is_none());
assert!(projected["sessions"][0].get("worktree_present").is_none());
assert!(projected["sessions"][0].get("prompt").is_none());
assert!(projected["sessions"][0].get("reminder").is_none());
assert!(projected["sessions"][0].get("backend_session_id").is_none());
assert!(
projected["sessions"][0]
.get("fresh_context_after_active_secs")
.is_none()
);
assert!(
projected["sessions"][0]
.get("active_context_accumulated_secs")
.is_none()
);
assert!(
projected["sessions"][0]
.get("active_context_segment_open")
.is_none()
);
assert!(
projected["sessions"][0]
.get("active_context_restart_due")
.is_none()
);
}
#[test]
fn session_list_projection_omits_empty_optional_discovery_fields() {
let status = serde_json::json!({
"sessions": [{
"id": "quiet-session",
"origin": "remote:locota",
"project_dir": null,
"role": "",
"bulletin": " "
}]
});
let projected = project_session_list(&status);
assert_eq!(
projected,
serde_json::json!({
"sessions": [{
"id": "quiet-session",
"origin": "remote:locota",
"parent_session": null
}]
})
);
}
#[test]
fn metadata_for_restored_session_preserves_persisted_fields() {
let metadata = crate::state::SessionMetadata {
project_dir: Some("/tmp/project".into()),
canonical_project_identity: Some("/tmp/project".into()),
role: Some("worker".into()),
bulletin: Some("busy".into()),
networked: false,
worktree: true,
vim_mode: true,
backend_session_id: Some("ses_old".into()),
backend: Some("opencode".into()),
opencode_binding: Some(crate::daemon_protocol::OpenCodeBinding::StrongManaged),
restart_generation: 9,
backend_repair_reservation: None,
session_incarnation: crate::daemon_protocol::SessionIncarnation(13),
project_description: Some("project".into()),
last_metadata_update: chrono::DateTime::from_timestamp(1_700_000_001, 0),
model: Some("openrouter/sonnet".into()),
effort: Some("high".into()),
codex_home: None,
reminder: Some("keep going".into()),
parent_session: Some("parent".into()),
idle_policy: Some(crate::daemon_protocol::IdlePolicy::CloseWhenDone),
prompt: Some("initial prompt".into()),
iteration: 4,
iteration_log: vec![],
last_iteration_at: Some(1_700_000_002),
on_fire: Some(crate::scheduler::OnFire::ContinueSession),
worktree_present: Some(true),
fresh_context_after_active_secs: Some(3_600),
active_context_accumulated_secs: 1_234,
active_context_segment_started_at: Some(1_700_000_003),
active_context_restart_due: true,
active_context_accounting_provisional: true,
};
let restored = metadata_for_restored_session(&metadata);
assert_eq!(restored.project_dir, metadata.project_dir);
assert_eq!(restored.role, metadata.role);
assert_eq!(restored.bulletin, metadata.bulletin);
assert_eq!(restored.networked, metadata.networked);
assert_eq!(restored.worktree, metadata.worktree);
assert_eq!(restored.vim_mode, metadata.vim_mode);
assert_eq!(restored.backend_session_id, metadata.backend_session_id);
assert_eq!(restored.backend, metadata.backend);
assert_eq!(restored.opencode_binding, metadata.opencode_binding);
assert_eq!(restored.restart_generation, metadata.restart_generation);
assert_eq!(restored.session_incarnation, metadata.session_incarnation);
assert_eq!(restored.project_description, metadata.project_description);
assert_eq!(restored.last_metadata_update, Some(1_700_000_001));
assert_eq!(restored.model, metadata.model);
assert_eq!(restored.effort, metadata.effort);
assert_eq!(restored.reminder, metadata.reminder);
assert_eq!(restored.parent_session, metadata.parent_session);
assert_eq!(restored.idle_policy, metadata.idle_policy);
assert_eq!(restored.prompt, metadata.prompt);
assert_eq!(restored.iteration, metadata.iteration);
assert_eq!(restored.iteration_log, metadata.iteration_log);
assert_eq!(restored.last_iteration_at, metadata.last_iteration_at);
assert_eq!(restored.on_fire, metadata.on_fire);
assert_eq!(restored.worktree_present, metadata.worktree_present);
assert_eq!(
restored.fresh_context_after_active_secs,
metadata.fresh_context_after_active_secs
);
assert_eq!(
restored.active_context_accumulated_secs,
metadata.active_context_accumulated_secs
);
assert_eq!(
restored.active_context_segment_started_at,
metadata.active_context_segment_started_at
);
assert_eq!(
restored.active_context_restart_due,
metadata.active_context_restart_due
);
assert_eq!(
restored.active_context_accounting_provisional,
metadata.active_context_accounting_provisional
);
}
#[tokio::test]
async fn restore_persisted_dormant_sessions_before_live_reconciliation() {
let dir = tempfile::tempdir().unwrap();
let dormant = persisted_dormant(
"rootfix",
42,
"codex-cli",
"thread-rootfix",
"/tmp/worktrees/rootfix",
"/tmp/repository",
);
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![],
std::collections::BTreeMap::from([("rootfix".into(), dormant.clone())]),
crate::daemon_protocol::SessionIncarnation(41),
std::collections::BTreeMap::new(),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "restore-dormant-test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
restore_persisted_sessions(&state).await.unwrap();
let protocol = state.protocol.read().await;
assert_eq!(protocol.dormant_sessions["rootfix"], dormant);
assert!(protocol.sessions.is_empty());
assert_eq!(
protocol.incarnation_high_water,
crate::daemon_protocol::SessionIncarnation(42)
);
}
#[tokio::test]
async fn restore_post_bind_dormant_resume_crash_restores_snapshot_and_releases_lease() {
let dir = tempfile::tempdir().unwrap();
let dormant = persisted_dormant(
"rootfix",
42,
"codex-cli",
"thread-rootfix",
"/tmp/worktrees/rootfix",
"/tmp/repository",
);
let owner = crate::daemon_protocol::ResourceOwner {
session_id: "rootfix".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(43),
};
let registered_at = chrono::Utc::now();
let staged = crate::persistence::PersistedSession {
id: "rootfix".into(),
pane: Some("%post-bind-crash".into()),
registered_at,
last_activity_at: registered_at,
metadata: crate::state::SessionMetadata {
project_dir: dormant.metadata.project_dir.clone(),
canonical_project_identity: dormant.metadata.canonical_project_identity.clone(),
backend: dormant.metadata.backend.clone(),
backend_session_id: dormant.metadata.backend_session_id.clone(),
session_incarnation: owner.incarnation,
..Default::default()
},
};
let lease = crate::daemon_protocol::LifecycleLease {
owner: owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Starting,
backend: Some("codex-cli".into()),
backend_session_id: Some("thread-rootfix".into()),
backend_session_owner: Some(owner.clone()),
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: Some(Box::new(dormant.clone())),
project_dir: Some("/tmp/worktrees/rootfix".into()),
project_dir_owner: Some(owner.clone()),
project_dir_cleanup_on_abandon: false,
inert_pane: Some("%post-bind-crash".into()),
inert_pane_owner: Some(owner),
sweep_unconfirmed: None,
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![staged],
std::collections::BTreeMap::new(),
crate::daemon_protocol::SessionIncarnation(43),
std::collections::BTreeMap::from([("rootfix".into(), lease)]),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "restore-post-bind-dormant-test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
restore_persisted_sessions(&state).await.unwrap();
let protocol = state.protocol.read().await;
assert_eq!(protocol.dormant_sessions["rootfix"], dormant);
assert!(!protocol.sessions.contains_key("rootfix"));
assert!(!protocol.lifecycle_leases.contains_key("rootfix"));
drop(protocol);
let persisted = crate::persistence::load_sessions(dir.path()).unwrap();
assert_eq!(persisted.dormant_sessions["rootfix"], dormant);
assert!(persisted.sessions.is_empty());
assert!(persisted.lifecycle_leases.is_empty());
}
#[tokio::test]
async fn restore_persisted_dormant_backend_suppresses_abandoned_abort() {
use axum::Router;
use axum::http::StatusCode;
use axum::routing::post;
use std::sync::atomic::{AtomicUsize, Ordering};
let dir = tempfile::tempdir().unwrap();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let abort_port = listener.local_addr().unwrap().port();
let daemon_port = abort_port.checked_sub(320).unwrap();
let calls = std::sync::Arc::new(AtomicUsize::new(0));
let calls_for_route = calls.clone();
let app = Router::new().route(
"/session/{session_id}/abort",
post(move || {
let calls = calls_for_route.clone();
async move {
calls.fetch_add(1, Ordering::SeqCst);
StatusCode::NO_CONTENT
}
}),
);
let server = tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let dormant = persisted_dormant(
"rootfix",
42,
"opencode",
"ses_rootfix",
"/tmp/worktrees/rootfix",
"/tmp/repository",
);
let owner = dormant.prior_owner.clone();
let lease = crate::daemon_protocol::LifecycleLease {
owner: owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Stopping,
backend: Some("opencode".into()),
backend_session_id: Some("ses_rootfix".into()),
backend_session_owner: Some(owner.clone()),
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: None,
project_dir_owner: None,
project_dir_cleanup_on_abandon: false,
inert_pane: None,
inert_pane_owner: None,
sweep_unconfirmed: None,
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![],
std::collections::BTreeMap::from([("rootfix".into(), dormant)]),
owner.incarnation,
std::collections::BTreeMap::from([("rootfix".into(), lease)]),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "restore-dormant-backend-test".into(),
npub: "npub1test".into(),
port: daemon_port,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
restore_persisted_sessions(&state).await.unwrap();
assert_eq!(calls.load(Ordering::SeqCst), 0);
assert!(
state
.protocol
.read()
.await
.dormant_sessions
.contains_key("rootfix")
);
assert!(
crate::persistence::load_sessions(dir.path())
.unwrap()
.lifecycle_leases
.is_empty()
);
server.abort();
}
#[tokio::test]
async fn restore_persisted_dormant_worktree_suppresses_abandoned_cleanup() {
let root = tempfile::tempdir().unwrap();
let repo = root.path().join("repo");
let worktree = repo.join(".ouija/worktrees/rootfix");
let data_dir = root.path().join("data");
std::fs::create_dir_all(&data_dir).unwrap();
let run_git = |args: &[&str]| {
let output = std::process::Command::new("git")
.args(args)
.output()
.unwrap();
assert!(
output.status.success(),
"git {:?} failed: {}",
args,
String::from_utf8_lossy(&output.stderr)
);
};
run_git(&["init", "-b", "main", repo.to_str().unwrap()]);
run_git(&[
"-C",
repo.to_str().unwrap(),
"config",
"user.email",
"test@example.com",
]);
run_git(&["-C", repo.to_str().unwrap(), "config", "user.name", "Test"]);
std::fs::write(repo.join("tracked"), "base\n").unwrap();
run_git(&["-C", repo.to_str().unwrap(), "add", "tracked"]);
run_git(&["-C", repo.to_str().unwrap(), "commit", "-m", "initial"]);
run_git(&[
"-C",
repo.to_str().unwrap(),
"worktree",
"add",
"-b",
"rootfix",
worktree.to_str().unwrap(),
]);
let actual = crate::state::project_dir_identity(worktree.to_str().unwrap());
let canonical = crate::state::project_dir_identity(repo.to_str().unwrap());
let dormant = persisted_dormant(
"rootfix",
42,
"codex-cli",
"thread-rootfix",
&actual,
&canonical,
);
let owner = dormant.prior_owner.clone();
let lease = crate::daemon_protocol::LifecycleLease {
owner: owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Stopping,
backend: None,
backend_session_id: None,
backend_session_owner: None,
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: Some(actual),
project_dir_owner: Some(owner.clone()),
project_dir_cleanup_on_abandon: true,
inert_pane: None,
inert_pane_owner: None,
sweep_unconfirmed: None,
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![],
std::collections::BTreeMap::from([("rootfix".into(), dormant)]),
owner.incarnation,
std::collections::BTreeMap::from([("rootfix".into(), lease)]),
);
crate::persistence::save_sessions(&data_dir, &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "restore-dormant-worktree-test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: data_dir.clone(),
config_dir: data_dir.clone(),
});
restore_persisted_sessions(&state).await.unwrap();
assert!(
worktree.join("tracked").is_file(),
"dormant ownership must preserve the parked worktree"
);
assert!(
state
.protocol
.read()
.await
.dormant_sessions
.contains_key("rootfix")
);
assert!(
crate::persistence::load_sessions(&data_dir)
.unwrap()
.lifecycle_leases
.is_empty()
);
}
#[tokio::test]
async fn restore_backfills_before_dead_pane_dormancy() {
let dir = tempfile::tempdir().unwrap();
let repository = tempfile::tempdir().unwrap();
let output = std::process::Command::new("git")
.args(["init", "-q"])
.current_dir(repository.path())
.output()
.unwrap();
assert!(output.status.success());
let registered_at = chrono::Utc::now();
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![crate::persistence::PersistedSession {
id: "dead-worker".into(),
pane: Some("%999999989".into()),
registered_at,
last_activity_at: registered_at,
metadata: crate::state::SessionMetadata {
project_dir: Some(repository.path().to_string_lossy().into_owned()),
canonical_project_identity: None,
backend: Some("codex-cli".into()),
backend_session_id: Some("thread-dead-worker".into()),
session_incarnation: crate::daemon_protocol::SessionIncarnation(42),
..Default::default()
},
}],
std::collections::BTreeMap::new(),
crate::daemon_protocol::SessionIncarnation(42),
std::collections::BTreeMap::new(),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "restore-backfill-order-test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
restore_persisted_sessions(&state).await.unwrap();
let protocol = state.protocol.read().await;
assert!(!protocol.sessions.contains_key("dead-worker"));
let dormant = &protocol.dormant_sessions["dead-worker"];
let expected = repository.path().canonicalize().unwrap();
assert_eq!(
dormant.metadata.canonical_project_identity.as_deref(),
Some(expected.to_string_lossy().as_ref())
);
assert_eq!(
dormant.canonical_project_identity,
expected.to_string_lossy()
);
drop(protocol);
let persisted = crate::persistence::load_sessions(dir.path()).unwrap();
assert!(persisted.sessions.is_empty());
assert!(persisted.dormant_sessions.contains_key("dead-worker"));
}
#[tokio::test]
async fn restore_dead_live_row_uses_atomic_dormancy_coordinator() {
let dir = tempfile::tempdir().unwrap();
let registered_at = chrono::Utc::now();
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![crate::persistence::PersistedSession {
id: "dead-worker".into(),
pane: Some("%999999991".into()),
registered_at,
last_activity_at: registered_at,
metadata: crate::state::SessionMetadata {
project_dir: Some("/tmp/dead-worker".into()),
canonical_project_identity: Some("/tmp/repository".into()),
backend: Some("codex-cli".into()),
backend_session_id: Some("thread-dead-worker".into()),
session_incarnation: crate::daemon_protocol::SessionIncarnation(42),
role: Some("preserve me".into()),
..Default::default()
},
}],
std::collections::BTreeMap::new(),
crate::daemon_protocol::SessionIncarnation(42),
std::collections::BTreeMap::new(),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "restore-dead-live-test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
let before = chrono::Utc::now().timestamp();
restore_persisted_sessions(&state).await.unwrap();
let after = chrono::Utc::now().timestamp();
let protocol = state.protocol.read().await;
assert!(!protocol.sessions.contains_key("dead-worker"));
let dormant = &protocol.dormant_sessions["dead-worker"];
assert!((before..=after).contains(&dormant.dormant_at));
assert_eq!(
dormant.source,
crate::daemon_protocol::DormancySource::Reaped
);
assert_eq!(dormant.metadata.role.as_deref(), Some("preserve me"));
drop(protocol);
let persisted = crate::persistence::load_sessions(dir.path()).unwrap();
assert!(persisted.sessions.is_empty());
assert!(persisted.dormant_sessions.contains_key("dead-worker"));
}
#[tokio::test]
async fn reaper_uses_atomic_dormancy_coordinator() {
let config = crate::state::tests::test_config();
let state = crate::state::AppState::new(config.clone());
state
.apply_and_execute(crate::daemon_protocol::Event::Register {
id: "dead-worker".into(),
pane: Some("%999999992".into()),
metadata: crate::daemon_protocol::SessionMeta {
project_dir: Some("/tmp/dead-worker".into()),
canonical_project_identity: Some("/tmp/repository".into()),
backend: Some("codex-cli".into()),
backend_session_id: Some("thread-dead-worker".into()),
..Default::default()
},
})
.await;
let owner = state.protocol.read().await.sessions["dead-worker"].owner();
reap_dead_sessions(
&state,
vec![(owner.clone(), "%999999992".into())],
1_753_920_321,
)
.await;
let protocol = state.protocol.read().await;
assert!(!protocol.sessions.contains_key("dead-worker"));
assert_eq!(protocol.dormant_sessions["dead-worker"].prior_owner, owner);
assert_eq!(
protocol.dormant_sessions["dead-worker"].dormant_at,
1_753_920_321
);
drop(protocol);
let persisted = crate::persistence::load_sessions(&config.data_dir).unwrap();
assert!(persisted.sessions.is_empty());
assert_eq!(persisted.dormant_sessions["dead-worker"].prior_owner, owner);
}
#[test]
fn reaper_process_names_checks_pane_backed_strong_opencode_bindings() {
let backends = crate::backend::BackendRegistry::default_registry();
let strong = crate::daemon_protocol::SessionMeta {
backend: Some("opencode".into()),
backend_session_id: Some("managed".into()),
opencode_binding: Some(crate::daemon_protocol::OpenCodeBinding::StrongManaged),
..Default::default()
};
let weak = crate::daemon_protocol::SessionMeta {
backend: Some("opencode".into()),
backend_session_id: Some("adopted".into()),
opencode_binding: Some(crate::daemon_protocol::OpenCodeBinding::WeakAdopted),
..Default::default()
};
assert_eq!(
reaper_process_names(&backends, &strong),
Some(vec!["opencode".to_string()])
);
assert_eq!(
reaper_process_names(&backends, &weak),
Some(vec!["opencode".to_string()])
);
}
#[tokio::test]
async fn restore_reaps_pane_backed_strong_opencode_session_with_missing_pane() {
let dir = tempfile::tempdir().unwrap();
let registered_at = chrono::Utc::now();
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![crate::persistence::PersistedSession {
id: "stale-opencode".into(),
pane: Some("%999999990".into()),
registered_at,
last_activity_at: registered_at,
metadata: crate::state::SessionMetadata {
project_dir: Some("/tmp/stale-opencode".into()),
canonical_project_identity: Some("/tmp/stale-opencode".into()),
backend: Some("opencode".into()),
backend_session_id: Some("ses_stale_opencode".into()),
opencode_binding: Some(crate::daemon_protocol::OpenCodeBinding::StrongManaged),
session_incarnation: crate::daemon_protocol::SessionIncarnation(42),
..Default::default()
},
}],
std::collections::BTreeMap::new(),
crate::daemon_protocol::SessionIncarnation(42),
std::collections::BTreeMap::new(),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "restore-stale-opencode-test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
restore_persisted_sessions(&state).await.unwrap();
let protocol = state.protocol.read().await;
assert!(!protocol.sessions.contains_key("stale-opencode"));
assert!(protocol.dormant_sessions.contains_key("stale-opencode"));
}
#[test]
fn reaper_process_names_uses_only_the_stored_backend() {
let backends = crate::backend::BackendRegistry::default_registry();
let claude = crate::daemon_protocol::SessionMeta {
backend: Some("claude-code".into()),
..Default::default()
};
assert_eq!(
reaper_process_names(&backends, &claude),
Some(vec!["claude".to_string()])
);
}
#[tokio::test]
async fn restore_persisted_paneless_strong_opencode_session_spawns_activity_receiver() {
let dir = tempfile::tempdir().unwrap();
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![crate::persistence::PersistedSession {
id: "restored-paneless".into(),
pane: None,
registered_at: chrono::Utc::now(),
last_activity_at: chrono::Utc::now(),
metadata: crate::state::SessionMetadata {
backend: Some("opencode".into()),
backend_session_id: Some("ses_restored_paneless".into()),
opencode_binding: Some(crate::daemon_protocol::OpenCodeBinding::StrongManaged),
session_incarnation: crate::daemon_protocol::SessionIncarnation(42),
fresh_context_after_active_secs: Some(60),
..Default::default()
},
}],
std::collections::BTreeMap::new(),
crate::daemon_protocol::SessionIncarnation(42),
std::collections::BTreeMap::new(),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "restore-paneless-agent-test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
restore_persisted_sessions(&state).await.unwrap();
let owner = state.protocol.read().await.sessions["restored-paneless"].owner();
assert!(
state
.notify_agent_owned(&owner, crate::session_agent::SessionMsg::Active)
.await,
"restored paneless owner must have a live activity receiver"
);
state.query_agent_pending_replies("restored-paneless").await;
assert!(
state.protocol.read().await.sessions["restored-paneless"]
.metadata
.active_context_segment_started_at
.is_some()
);
}
#[tokio::test]
async fn restore_releases_paneless_worktree_lease_without_harming_shared_replacement() {
let dir = tempfile::tempdir().unwrap();
let owner = crate::daemon_protocol::ResourceOwner {
session_id: "pending".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(49),
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![crate::persistence::PersistedSession {
id: "live".into(),
pane: None,
registered_at: chrono::Utc::now(),
last_activity_at: chrono::Utc::now(),
metadata: crate::state::SessionMetadata {
backend: Some("opencode".into()),
backend_session_id: Some("ses_live".into()),
session_incarnation: crate::daemon_protocol::SessionIncarnation(48),
project_dir: Some("/tmp/.ouija/worktrees/project/shared-replacement".into()),
..Default::default()
},
}],
std::collections::BTreeMap::new(),
crate::daemon_protocol::SessionIncarnation(50),
std::collections::BTreeMap::from([(
owner.session_id.clone(),
crate::daemon_protocol::LifecycleLease {
owner: owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Starting,
backend: None,
backend_session_id: None,
backend_session_owner: None,
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: Some("/tmp/.ouija/worktrees/project/shared-replacement".into()),
project_dir_owner: Some(owner.clone()),
project_dir_cleanup_on_abandon: true,
inert_pane: None,
inert_pane_owner: None,
sweep_unconfirmed: None,
},
)]),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
restore_persisted_sessions(&state).await.unwrap();
let restored_snapshot = crate::persistence::load_sessions(dir.path()).unwrap();
assert!(restored_snapshot.lifecycle_leases.is_empty());
assert_eq!(restored_snapshot.sessions.len(), 1);
assert_eq!(restored_snapshot.sessions[0].id, "live");
assert_eq!(
restored_snapshot.incarnation_high_water,
crate::daemon_protocol::SessionIncarnation(50)
);
let mut proto = state.protocol.write().await;
assert!(proto.sessions.contains_key("live"));
assert!(
proto.lifecycle_leases.is_empty(),
"restart must release abandoned pre-launch authority"
);
assert_eq!(
proto.reserve_start("next").unwrap(),
crate::daemon_protocol::StartDisposition::Reserved(
crate::daemon_protocol::ResourceOwner {
session_id: "next".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(51),
}
),
"the first post-restart token must exceed removed persisted owners"
);
}
#[tokio::test]
async fn restore_never_deletes_incumbent_worktree_claimed_by_abandoned_restart() {
let root = tempfile::tempdir().unwrap();
let repo = root.path().join("repo");
let worktree = repo.join(".ouija/worktrees/restart");
let data_dir = root.path().join("data");
std::fs::create_dir_all(&data_dir).unwrap();
let run_git = |args: &[&str]| {
let output = std::process::Command::new("git")
.args(args)
.output()
.unwrap();
assert!(
output.status.success(),
"git {:?} failed: {}",
args,
String::from_utf8_lossy(&output.stderr)
);
};
run_git(&["init", "-b", "main", repo.to_str().unwrap()]);
run_git(&[
"-C",
repo.to_str().unwrap(),
"config",
"user.email",
"test@example.com",
]);
run_git(&["-C", repo.to_str().unwrap(), "config", "user.name", "Test"]);
std::fs::write(repo.join("tracked"), "incumbent\n").unwrap();
run_git(&["-C", repo.to_str().unwrap(), "add", "tracked"]);
run_git(&["-C", repo.to_str().unwrap(), "commit", "-m", "initial"]);
run_git(&[
"-C",
repo.to_str().unwrap(),
"worktree",
"add",
"-b",
"restart",
worktree.to_str().unwrap(),
]);
let incumbent = crate::daemon_protocol::ResourceOwner {
session_id: "worker".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(1),
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![crate::persistence::PersistedSession {
id: incumbent.session_id.clone(),
pane: None,
registered_at: chrono::Utc::now(),
last_activity_at: chrono::Utc::now(),
metadata: crate::state::SessionMetadata {
backend: Some("opencode".into()),
project_dir: Some(worktree.to_string_lossy().into_owned()),
session_incarnation: crate::daemon_protocol::SessionIncarnation(2),
..Default::default()
},
}],
std::collections::BTreeMap::new(),
crate::daemon_protocol::SessionIncarnation(2),
std::collections::BTreeMap::from([(
incumbent.session_id.clone(),
crate::daemon_protocol::LifecycleLease {
owner: incumbent.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Restarting,
backend: None,
backend_session_id: None,
backend_session_owner: None,
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: Some(worktree.to_string_lossy().into_owned()),
project_dir_owner: Some(incumbent),
project_dir_cleanup_on_abandon: false,
inert_pane: None,
inert_pane_owner: None,
sweep_unconfirmed: None,
},
)]),
);
crate::persistence::save_sessions(&data_dir, &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: data_dir.clone(),
config_dir: data_dir.clone(),
});
restore_persisted_sessions(&state).await.unwrap();
assert!(
worktree.join("tracked").is_file(),
"an abandoned restart may terminalize its staged row but must preserve the incumbent worktree"
);
let restored = crate::persistence::load_sessions(&data_dir).unwrap();
assert!(restored.sessions.is_empty());
assert!(restored.lifecycle_leases.is_empty());
}
#[tokio::test]
#[cfg(unix)]
async fn restore_cleans_owned_worktree_below_symlinked_managed_root() {
use std::os::unix::fs::symlink;
let root = tempfile::tempdir().unwrap();
let repo = root.path().join("repo");
let data_dir = root.path().join("data");
let home = root.path().join("home");
let physical_managed_root = root.path().join("managed-storage");
let raw_managed_root = home.join(".ouija");
let raw_worktree = raw_managed_root.join("worktrees/repo/worker");
std::fs::create_dir_all(&data_dir).unwrap();
std::fs::create_dir_all(&home).unwrap();
std::fs::create_dir_all(physical_managed_root.join("worktrees/repo")).unwrap();
symlink(&physical_managed_root, &raw_managed_root).unwrap();
let run_git = |args: &[&str]| {
let output = std::process::Command::new("git")
.args(args)
.output()
.unwrap();
assert!(
output.status.success(),
"git {:?} failed: {}",
args,
String::from_utf8_lossy(&output.stderr)
);
};
run_git(&["init", "-b", "main", repo.to_str().unwrap()]);
run_git(&[
"-C",
repo.to_str().unwrap(),
"config",
"user.email",
"test@example.com",
]);
run_git(&["-C", repo.to_str().unwrap(), "config", "user.name", "Test"]);
std::fs::write(repo.join("tracked"), "base\n").unwrap();
run_git(&["-C", repo.to_str().unwrap(), "add", "tracked"]);
run_git(&["-C", repo.to_str().unwrap(), "commit", "-m", "initial"]);
run_git(&[
"-C",
repo.to_str().unwrap(),
"worktree",
"add",
"-b",
"worker",
raw_worktree.to_str().unwrap(),
]);
let canonical_worktree = crate::state::project_dir_identity(raw_worktree.to_str().unwrap());
assert!(
!canonical_worktree.contains("/.ouija/worktrees/"),
"the physical identity must exercise recovery without a managed-path marker"
);
let owner = crate::daemon_protocol::ResourceOwner {
session_id: "worker".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(1),
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![],
std::collections::BTreeMap::new(),
owner.incarnation,
std::collections::BTreeMap::from([(
owner.session_id.clone(),
crate::daemon_protocol::LifecycleLease {
owner: owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Starting,
backend: None,
backend_session_id: None,
backend_session_owner: None,
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: Some(canonical_worktree),
project_dir_owner: Some(owner),
project_dir_cleanup_on_abandon: true,
inert_pane: None,
inert_pane_owner: None,
sweep_unconfirmed: None,
},
)]),
);
crate::persistence::save_sessions(&data_dir, &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: data_dir.clone(),
config_dir: data_dir.clone(),
});
restore_persisted_sessions(&state).await.unwrap();
assert!(
!raw_worktree.exists(),
"cleanup authority must survive canonicalizing a symlinked managed root"
);
assert!(
crate::persistence::load_sessions(&data_dir)
.unwrap()
.lifecycle_leases
.is_empty()
);
}
#[tokio::test]
async fn restore_finishes_owned_stop_after_registry_removal() {
let root = tempfile::tempdir().unwrap();
let repo = root.path().join("repo");
let worktree = repo.join(".ouija/worktrees/worker");
let data_dir = root.path().join("data");
std::fs::create_dir_all(&data_dir).unwrap();
let run_git = |args: &[&str]| {
let output = std::process::Command::new("git")
.args(args)
.output()
.unwrap();
assert!(
output.status.success(),
"git {:?} failed: {}",
args,
String::from_utf8_lossy(&output.stderr)
);
};
run_git(&["init", "-b", "main", repo.to_str().unwrap()]);
run_git(&[
"-C",
repo.to_str().unwrap(),
"config",
"user.email",
"test@example.com",
]);
run_git(&["-C", repo.to_str().unwrap(), "config", "user.name", "Test"]);
std::fs::write(repo.join("tracked"), "base\n").unwrap();
run_git(&["-C", repo.to_str().unwrap(), "add", "tracked"]);
run_git(&["-C", repo.to_str().unwrap(), "commit", "-m", "initial"]);
run_git(&[
"-C",
repo.to_str().unwrap(),
"worktree",
"add",
"-b",
"worker",
worktree.to_str().unwrap(),
]);
let owner = crate::daemon_protocol::ResourceOwner {
session_id: "worker".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(1),
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![],
std::collections::BTreeMap::new(),
owner.incarnation,
std::collections::BTreeMap::from([(
owner.session_id.clone(),
crate::daemon_protocol::LifecycleLease {
owner: owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Stopping,
backend: None,
backend_session_id: None,
backend_session_owner: None,
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: Some(crate::state::project_dir_identity(
worktree.to_str().unwrap(),
)),
project_dir_owner: Some(owner.clone()),
project_dir_cleanup_on_abandon: true,
inert_pane: Some("%999999999".into()),
inert_pane_owner: Some(owner.clone()),
sweep_unconfirmed: None,
},
)]),
);
crate::persistence::save_sessions(&data_dir, &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: data_dir.clone(),
config_dir: data_dir.clone(),
});
restore_persisted_sessions(&state).await.unwrap();
assert!(
!worktree.exists(),
"startup recovery must finish the explicit kill's durable worktree cleanup"
);
assert!(
crate::persistence::load_sessions(&data_dir)
.unwrap()
.lifecycle_leases
.is_empty()
);
assert!(matches!(
state
.protocol
.write()
.await
.reserve_start("worker")
.unwrap(),
crate::daemon_protocol::StartDisposition::Reserved(_)
));
}
#[tokio::test]
async fn restore_aborts_claimed_http_backend_before_releasing_stop_lease() {
use axum::Router;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::routing::post;
use std::sync::atomic::{AtomicUsize, Ordering};
#[derive(Clone)]
struct AbortProbe {
data_dir: std::path::PathBuf,
calls: std::sync::Arc<AtomicUsize>,
}
async fn abort_backend(
Path(backend_session_id): Path<String>,
State(probe): State<AbortProbe>,
) -> StatusCode {
assert_eq!(backend_session_id, "ses_worker");
assert!(
crate::persistence::load_sessions(&probe.data_dir)
.unwrap()
.lifecycle_leases
.contains_key("worker"),
"the stopping lease must remain durable through HTTP abort"
);
probe.calls.fetch_add(1, Ordering::SeqCst);
StatusCode::NO_CONTENT
}
for pane in [Some("%999999999"), None] {
let dir = tempfile::tempdir().unwrap();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let abort_port = listener.local_addr().unwrap().port();
let daemon_port = abort_port.checked_sub(320).unwrap();
let calls = std::sync::Arc::new(AtomicUsize::new(0));
let app = Router::new()
.route("/session/{session_id}/abort", post(abort_backend))
.with_state(AbortProbe {
data_dir: dir.path().to_path_buf(),
calls: calls.clone(),
});
let server = tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let owner = crate::daemon_protocol::ResourceOwner {
session_id: "worker".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(7),
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![],
std::collections::BTreeMap::new(),
owner.incarnation,
std::collections::BTreeMap::from([(
owner.session_id.clone(),
crate::daemon_protocol::LifecycleLease {
owner: owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Stopping,
backend: Some("opencode".into()),
backend_session_id: Some("ses_worker".into()),
backend_session_owner: Some(owner.clone()),
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: None,
project_dir_owner: None,
project_dir_cleanup_on_abandon: false,
inert_pane: pane.map(str::to_owned),
inert_pane_owner: pane.map(|_| owner.clone()),
sweep_unconfirmed: None,
},
)]),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: daemon_port,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
restore_persisted_sessions(&state).await.unwrap();
assert_eq!(calls.load(Ordering::SeqCst), 1);
assert!(
crate::persistence::load_sessions(dir.path())
.unwrap()
.lifecycle_leases
.is_empty()
);
server.abort();
}
}
#[derive(Clone)]
struct RestoreSubagentProbe {
children: std::collections::HashMap<String, Vec<String>>,
reject_abort: std::collections::HashSet<String>,
aborted: std::sync::Arc<std::sync::Mutex<Vec<String>>>,
}
async fn restore_probe_children(
axum::extract::Path(id): axum::extract::Path<String>,
axum::extract::State(probe): axum::extract::State<RestoreSubagentProbe>,
) -> axum::response::Response {
use axum::response::IntoResponse;
match probe.children.get(&id) {
Some(children) => axum::Json(
children
.iter()
.map(|child| serde_json::json!({ "id": child, "parentID": id }))
.collect::<Vec<_>>(),
)
.into_response(),
None => axum::http::StatusCode::NOT_FOUND.into_response(),
}
}
async fn restore_probe_abort(
axum::extract::Path(id): axum::extract::Path<String>,
axum::extract::State(probe): axum::extract::State<RestoreSubagentProbe>,
) -> axum::http::StatusCode {
probe.aborted.lock().unwrap().push(id.clone());
if probe.reject_abort.contains(&id) {
axum::http::StatusCode::INTERNAL_SERVER_ERROR
} else {
axum::http::StatusCode::NO_CONTENT
}
}
async fn restore_state_with_stopping_lease(
dir: &tempfile::TempDir,
probe: &RestoreSubagentProbe,
) -> (crate::state::SharedState, tokio::task::JoinHandle<()>) {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let serve_port = listener.local_addr().unwrap().port();
let daemon_port = serve_port.checked_sub(320).unwrap();
let app = axum::Router::new()
.route(
"/session/{session_id}/children",
axum::routing::get(restore_probe_children),
)
.route(
"/session/{session_id}/abort",
axum::routing::post(restore_probe_abort),
)
.with_state(probe.clone());
let server = tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let owner = crate::daemon_protocol::ResourceOwner {
session_id: "worker".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(7),
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![],
std::collections::BTreeMap::new(),
owner.incarnation,
std::collections::BTreeMap::from([(
owner.session_id.clone(),
crate::daemon_protocol::LifecycleLease {
owner: owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Stopping,
backend: Some("opencode".into()),
backend_session_id: Some("ses_worker".into()),
backend_session_owner: Some(owner),
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: None,
project_dir_owner: None,
project_dir_cleanup_on_abandon: false,
inert_pane: None,
inert_pane_owner: None,
sweep_unconfirmed: None,
},
)]),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: daemon_port,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
(state, server)
}
#[tokio::test]
async fn restore_sweeps_nested_subagents_of_an_abandoned_stopping_lease() {
let dir = tempfile::tempdir().unwrap();
let probe = RestoreSubagentProbe {
children: std::collections::HashMap::from([
("ses_worker".to_string(), vec!["ses_a".to_string()]),
("ses_a".to_string(), vec!["ses_b".to_string()]),
("ses_b".to_string(), vec![]),
]),
reject_abort: std::collections::HashSet::new(),
aborted: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
};
let (state, server) = restore_state_with_stopping_lease(&dir, &probe).await;
restore_persisted_sessions(&state).await.unwrap();
let mut aborted = probe.aborted.lock().unwrap().clone();
aborted.sort();
assert_eq!(
aborted,
vec![
"ses_a".to_string(),
"ses_b".to_string(),
"ses_worker".to_string()
],
"restore must abort the parent and every nested subagent"
);
assert!(
crate::persistence::load_sessions(dir.path())
.unwrap()
.lifecycle_leases
.is_empty()
);
server.abort();
}
#[tokio::test]
async fn restore_quarantines_an_unconfirmed_sweep_and_still_boots() {
let dir = tempfile::tempdir().unwrap();
let probe = RestoreSubagentProbe {
children: std::collections::HashMap::from([
("ses_worker".to_string(), vec!["ses_child".to_string()]),
("ses_child".to_string(), vec![]),
]),
reject_abort: std::collections::HashSet::from(["ses_child".to_string()]),
aborted: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
};
let (state, server) = restore_state_with_stopping_lease(&dir, &probe).await;
restore_persisted_sessions(&state)
.await
.expect("an unconfirmed sweep must not fail daemon startup");
let persisted = crate::persistence::load_sessions(dir.path()).unwrap();
let lease = persisted
.lifecycle_leases
.get("worker")
.expect("a quarantined lease must be retained on disk");
assert_eq!(
lease.phase,
crate::daemon_protocol::LifecyclePhase::Stopping
);
assert!(
lease.sweep_unconfirmed.as_deref().is_some_and(
|reason| reason.contains("subagent sweep for abandoned session 'ses_worker'")
),
"the quarantine must carry the reason: {:?}",
lease.sweep_unconfirmed
);
let protocol = state.protocol.read().await;
assert_eq!(
protocol.lifecycle_leases["worker"]
.sweep_unconfirmed
.as_deref(),
lease.sweep_unconfirmed.as_deref(),
"the live lease must carry the same reason as the persisted one"
);
let mut probe_protocol = protocol.clone();
drop(protocol);
assert!(matches!(
probe_protocol.clone().reserve_start("worker").unwrap(),
crate::daemon_protocol::StartDisposition::InProgress(_)
));
let effects = probe_protocol.apply(crate::daemon_protocol::Event::Register {
id: "worker".into(),
pane: Some("%42".into()),
metadata: Default::default(),
});
assert!(
effects.iter().any(|effect| matches!(
effect,
crate::daemon_protocol::Effect::RegisterFailed { session_id, .. }
if session_id == "worker"
)),
"a quarantined ID must not be adoptable: {effects:?}"
);
let owner = crate::daemon_protocol::ResourceOwner {
session_id: "worker".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(7),
};
assert_eq!(
probe_protocol.claim_existing_stop(&owner, "%42", true),
crate::daemon_protocol::LifecycleMutationOutcome::Rejected,
"a quarantined ID must not be reapable"
);
server.abort();
}
#[tokio::test]
async fn quarantined_lease_is_reported_by_status_and_ls() {
let dir = tempfile::tempdir().unwrap();
let probe = RestoreSubagentProbe {
children: std::collections::HashMap::from([
("ses_worker".to_string(), vec!["ses_child".to_string()]),
("ses_child".to_string(), vec![]),
]),
reject_abort: std::collections::HashSet::from(["ses_child".to_string()]),
aborted: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
};
let (state, server) = restore_state_with_stopping_lease(&dir, &probe).await;
restore_persisted_sessions(&state).await.unwrap();
let axum::Json(status) = crate::api::status(axum::extract::State(state.clone())).await;
let quarantined = status["quarantined_sessions"].as_array().unwrap();
assert_eq!(quarantined.len(), 1, "got: {status}");
assert_eq!(quarantined[0]["id"], "worker");
assert_eq!(quarantined[0]["backend_session_id"], "ses_worker");
assert_eq!(quarantined[0]["recover_with"], "ouija recover-lease worker");
assert!(
quarantined[0]["reason"]
.as_str()
.unwrap()
.contains("subagent sweep"),
"got: {}",
quarantined[0]["reason"]
);
let listed = project_session_list(&status);
assert_eq!(
listed["quarantined_sessions"], status["quarantined_sessions"],
"ls must carry the quarantine list verbatim"
);
server.abort();
}
#[test]
fn ls_flags_a_surviving_row_of_a_quarantined_session() {
let status = serde_json::json!({
"sessions": [
{ "id": "worker", "origin": "local" },
{ "id": "other", "origin": "local" },
],
"quarantined_sessions": [
{ "id": "worker", "reason": "subagent sweep did not complete" },
],
});
let listed = project_session_list(&status);
let sessions = listed["sessions"].as_array().unwrap();
assert_eq!(sessions[0]["quarantined"], serde_json::json!(true));
assert_eq!(
sessions[0]["quarantine_reason"],
"subagent sweep did not complete"
);
assert!(
sessions[1].get("quarantined").is_none(),
"an unaffected session must not be flagged"
);
}
#[test]
fn ls_omits_the_quarantine_list_when_nothing_is_quarantined() {
let status = serde_json::json!({
"sessions": [{ "id": "worker", "origin": "local" }],
"quarantined_sessions": [],
});
let listed = project_session_list(&status);
assert!(listed.get("quarantined_sessions").is_none());
assert!(listed["sessions"][0].get("quarantined").is_none());
}
#[tokio::test]
async fn restore_never_aborts_a_subagent_owned_by_a_persisted_session() {
let dir = tempfile::tempdir().unwrap();
let probe = RestoreSubagentProbe {
children: std::collections::HashMap::from([
("ses_worker".to_string(), vec!["ses_child".to_string()]),
("ses_child".to_string(), vec![]),
]),
reject_abort: std::collections::HashSet::new(),
aborted: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let serve_port = listener.local_addr().unwrap().port();
let daemon_port = serve_port.checked_sub(320).unwrap();
let app = axum::Router::new()
.route(
"/session/{session_id}/children",
axum::routing::get(restore_probe_children),
)
.route(
"/session/{session_id}/abort",
axum::routing::post(restore_probe_abort),
)
.with_state(probe.clone());
let server = tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let owner = crate::daemon_protocol::ResourceOwner {
session_id: "worker".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(7),
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![crate::persistence::PersistedSession {
id: "neighbour".into(),
pane: None,
registered_at: chrono::Utc::now(),
last_activity_at: chrono::Utc::now(),
metadata: crate::state::SessionMetadata {
backend: Some("opencode".into()),
backend_session_id: Some("ses_child".into()),
session_incarnation: crate::daemon_protocol::SessionIncarnation(6),
..Default::default()
},
}],
std::collections::BTreeMap::new(),
owner.incarnation,
std::collections::BTreeMap::from([(
owner.session_id.clone(),
crate::daemon_protocol::LifecycleLease {
owner: owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Stopping,
backend: Some("opencode".into()),
backend_session_id: Some("ses_worker".into()),
backend_session_owner: Some(owner),
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: None,
project_dir_owner: None,
project_dir_cleanup_on_abandon: false,
inert_pane: None,
inert_pane_owner: None,
sweep_unconfirmed: None,
},
)]),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: daemon_port,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
restore_persisted_sessions(&state).await.unwrap();
assert_eq!(
probe.aborted.lock().unwrap().clone(),
vec!["ses_worker".to_string()],
"a subagent owned by a persisted session must not be aborted"
);
server.abort();
}
#[tokio::test]
async fn restore_rejects_replacement_sharing_a_stopping_lease_backend_pair() {
let dir = tempfile::tempdir().unwrap();
let stale_owner = crate::daemon_protocol::ResourceOwner {
session_id: "worker".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(7),
};
let replacement_owner = crate::daemon_protocol::ResourceOwner {
session_id: "worker".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(8),
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![crate::persistence::PersistedSession {
id: "worker".into(),
pane: None,
registered_at: chrono::Utc::now(),
last_activity_at: chrono::Utc::now(),
metadata: crate::state::SessionMetadata {
backend: Some("opencode".into()),
backend_session_id: Some("ses_worker".into()),
session_incarnation: replacement_owner.incarnation,
..Default::default()
},
}],
std::collections::BTreeMap::new(),
replacement_owner.incarnation,
std::collections::BTreeMap::from([(
stale_owner.session_id.clone(),
crate::daemon_protocol::LifecycleLease {
owner: stale_owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Stopping,
backend: Some("opencode".into()),
backend_session_id: Some("ses_worker".into()),
backend_session_owner: Some(stale_owner),
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: None,
project_dir_owner: None,
project_dir_cleanup_on_abandon: false,
inert_pane: None,
inert_pane_owner: None,
sweep_unconfirmed: None,
},
)]),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
let error = restore_persisted_sessions(&state).await.unwrap_err();
assert!(
error
.to_string()
.contains("failed to restore lifecycle authority")
);
assert!(format!("{error:#}").contains(
"backend pair (opencode, ses_worker) is assigned to distinct owners 'worker' and 'worker'"
));
assert!(state.protocol.read().await.sessions.is_empty());
}
#[tokio::test]
async fn restore_retains_stop_lease_when_http_abort_is_not_confirmed() {
use axum::Router;
use axum::http::StatusCode;
use axum::routing::post;
async fn reject_abort() -> StatusCode {
StatusCode::INTERNAL_SERVER_ERROR
}
let dir = tempfile::tempdir().unwrap();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let abort_port = listener.local_addr().unwrap().port();
let daemon_port = abort_port.checked_sub(320).unwrap();
let server = tokio::spawn(async move {
axum::serve(
listener,
Router::new().route("/session/{session_id}/abort", post(reject_abort)),
)
.await
.unwrap();
});
let owner = crate::daemon_protocol::ResourceOwner {
session_id: "worker".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(9),
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![],
std::collections::BTreeMap::new(),
owner.incarnation,
std::collections::BTreeMap::from([(
owner.session_id.clone(),
crate::daemon_protocol::LifecycleLease {
owner: owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Stopping,
backend: Some("opencode".into()),
backend_session_id: Some("ses_worker".into()),
backend_session_owner: Some(owner),
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: None,
project_dir_owner: None,
project_dir_cleanup_on_abandon: false,
inert_pane: None,
inert_pane_owner: None,
sweep_unconfirmed: None,
},
)]),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: daemon_port,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
let result = restore_persisted_sessions(&state).await;
assert!(result.is_err());
assert!(
crate::persistence::load_sessions(dir.path())
.unwrap()
.lifecycle_leases
.contains_key("worker")
);
server.abort();
}
#[tokio::test]
async fn queued_replacement_reselects_and_recovers_worktree_after_cleanup() {
let root = tempfile::tempdir().unwrap();
let repo = root.path().join("repo");
let legacy = repo.join(".ouija/worktrees/worker");
let replacement = root.path().join("home/.ouija/worktrees/repo/worker");
let data_dir = root.path().join("data");
std::fs::create_dir_all(&data_dir).unwrap();
let run_git = |args: &[&str]| {
let output = std::process::Command::new("git")
.args(args)
.output()
.unwrap();
assert!(
output.status.success(),
"git {:?} failed: {}",
args,
String::from_utf8_lossy(&output.stderr)
);
};
run_git(&["init", "-b", "main", repo.to_str().unwrap()]);
run_git(&[
"-C",
repo.to_str().unwrap(),
"config",
"user.email",
"test@example.com",
]);
run_git(&["-C", repo.to_str().unwrap(), "config", "user.name", "Test"]);
std::fs::write(repo.join("tracked"), "base\n").unwrap();
run_git(&["-C", repo.to_str().unwrap(), "add", "tracked"]);
run_git(&["-C", repo.to_str().unwrap(), "commit", "-m", "initial"]);
run_git(&[
"-C",
repo.to_str().unwrap(),
"worktree",
"add",
"-b",
"worker",
legacy.to_str().unwrap(),
]);
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: data_dir.clone(),
config_dir: data_dir.clone(),
});
let owner = match state.reserve_start("worker").await.unwrap() {
crate::daemon_protocol::StartDisposition::Reserved(owner) => owner,
disposition => panic!("expected reservation, got {disposition:?}"),
};
let stale_owner = crate::daemon_protocol::ResourceOwner {
session_id: "stale".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(0),
};
let cleanup_started = std::sync::Arc::new(tokio::sync::Notify::new());
let cleanup_removed = std::sync::Arc::new(tokio::sync::Notify::new());
let release_cleanup = std::sync::Arc::new(tokio::sync::Notify::new());
let cleanup_state = state.clone();
let cleanup_legacy = legacy.to_string_lossy().into_owned();
let started = cleanup_started.clone();
let removed = cleanup_removed.clone();
let release = release_cleanup.clone();
let cleanup_action_dir = cleanup_legacy.clone();
let cleanup = tokio::spawn(async move {
cleanup_state
.with_owned_worktree_cleanup(&stale_owner, &cleanup_legacy, || async move {
started.notify_one();
crate::state::AppState::cleanup_worktree_dir(&cleanup_action_dir).await;
removed.notify_one();
release.notified().await;
})
.await
});
cleanup_started.notified().await;
let claim_state = state.clone();
let claim_owner = owner.clone();
let candidates = vec![
legacy.to_string_lossy().into_owned(),
replacement.to_string_lossy().into_owned(),
];
let select_legacy = candidates[0].clone();
let select_replacement = candidates[1].clone();
let claim_repo = repo.clone();
let claim = tokio::spawn(async move {
claim_state
.with_reserved_project_dir_choice(
&claim_owner,
claim_owner.clone(),
&candidates,
move || {
if std::path::Path::new(&select_legacy).exists() {
select_legacy
} else {
select_replacement
}
},
move |selected| async move {
let parent = std::path::Path::new(&selected).parent().unwrap();
std::fs::create_dir_all(parent).unwrap();
let output = std::process::Command::new("git")
.args([
"-C",
claim_repo.to_str().unwrap(),
"worktree",
"add",
&selected,
"worker",
])
.output()
.unwrap();
assert!(
output.status.success(),
"replacement worktree creation failed: {}",
String::from_utf8_lossy(&output.stderr)
);
selected
},
)
.await
});
cleanup_removed.notified().await;
assert!(
!claim.is_finished(),
"replacement selection must wait for cleanup's directory gate"
);
release_cleanup.notify_one();
assert_eq!(cleanup.await.unwrap(), Some(()));
let selected = claim.await.unwrap().unwrap().unwrap();
assert_eq!(selected, replacement.to_string_lossy());
let persisted = crate::persistence::load_sessions(&data_dir).unwrap();
assert!(
persisted.lifecycle_leases["worker"].project_dir_cleanup_on_abandon,
"replacement created after queued cleanup must gain recovery cleanup authority"
);
drop(state);
let recovery_state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: data_dir.clone(),
config_dir: data_dir.clone(),
});
restore_persisted_sessions(&recovery_state).await.unwrap();
assert!(
!replacement.exists(),
"startup recovery must remove the abandoned replacement worktree"
);
assert!(
crate::persistence::load_sessions(&data_dir)
.unwrap()
.lifecycle_leases
.is_empty()
);
}
#[tokio::test]
async fn restore_persisted_sessions_removes_unlaunched_http_start_before_releasing_id() {
let dir = tempfile::tempdir().unwrap();
let owner = crate::daemon_protocol::ResourceOwner {
session_id: "pending".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(49),
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![crate::persistence::PersistedSession {
id: owner.session_id.clone(),
pane: None,
registered_at: chrono::Utc::now(),
last_activity_at: chrono::Utc::now(),
metadata: crate::state::SessionMetadata {
backend: Some("opencode".into()),
session_incarnation: owner.incarnation,
..Default::default()
},
}],
std::collections::BTreeMap::new(),
owner.incarnation,
std::collections::BTreeMap::from([(
owner.session_id.clone(),
crate::daemon_protocol::LifecycleLease {
owner: owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Starting,
backend: None,
backend_session_id: None,
backend_session_owner: None,
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: None,
project_dir_owner: None,
project_dir_cleanup_on_abandon: false,
inert_pane: None,
inert_pane_owner: None,
sweep_unconfirmed: None,
},
)]),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
restore_persisted_sessions(&state).await.unwrap();
let restored_snapshot = crate::persistence::load_sessions(dir.path()).unwrap();
assert!(restored_snapshot.sessions.is_empty());
assert!(restored_snapshot.lifecycle_leases.is_empty());
let mut proto = state.protocol.write().await;
assert!(!proto.sessions.contains_key("pending"));
assert!(matches!(
proto.reserve_start("pending").unwrap(),
crate::daemon_protocol::StartDisposition::Reserved(_)
));
}
#[tokio::test]
async fn restore_persisted_sessions_distinguishes_unstaged_and_staged_restarts() {
let dir = tempfile::tempdir().unwrap();
let incumbent_owner = crate::daemon_protocol::ResourceOwner {
session_id: "incumbent".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(40),
};
let staged_lease_owner = crate::daemon_protocol::ResourceOwner {
session_id: "staged".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(41),
};
let snapshot = crate::persistence::PersistedLifecycleState::new(
vec![
crate::persistence::PersistedSession {
id: incumbent_owner.session_id.clone(),
pane: None,
registered_at: chrono::Utc::now(),
last_activity_at: chrono::Utc::now(),
metadata: crate::state::SessionMetadata {
backend: Some("opencode".into()),
session_incarnation: incumbent_owner.incarnation,
..Default::default()
},
},
crate::persistence::PersistedSession {
id: staged_lease_owner.session_id.clone(),
pane: None,
registered_at: chrono::Utc::now(),
last_activity_at: chrono::Utc::now(),
metadata: crate::state::SessionMetadata {
backend: Some("opencode".into()),
session_incarnation: crate::daemon_protocol::SessionIncarnation(42),
..Default::default()
},
},
],
std::collections::BTreeMap::new(),
crate::daemon_protocol::SessionIncarnation(42),
std::collections::BTreeMap::from([
(
incumbent_owner.session_id.clone(),
crate::daemon_protocol::LifecycleLease {
owner: incumbent_owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Restarting,
backend: None,
backend_session_id: None,
backend_session_owner: None,
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: None,
project_dir_owner: None,
project_dir_cleanup_on_abandon: false,
inert_pane: None,
inert_pane_owner: None,
sweep_unconfirmed: None,
},
),
(
staged_lease_owner.session_id.clone(),
crate::daemon_protocol::LifecycleLease {
owner: staged_lease_owner,
phase: crate::daemon_protocol::LifecyclePhase::Restarting,
backend: None,
backend_session_id: None,
backend_session_owner: None,
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: None,
project_dir_owner: None,
project_dir_cleanup_on_abandon: false,
inert_pane: None,
inert_pane_owner: None,
sweep_unconfirmed: None,
},
),
]),
);
crate::persistence::save_sessions(dir.path(), &snapshot).unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
restore_persisted_sessions(&state).await.unwrap();
let restored_snapshot = crate::persistence::load_sessions(dir.path()).unwrap();
assert_eq!(restored_snapshot.sessions.len(), 1);
assert_eq!(restored_snapshot.sessions[0].id, "incumbent");
let proto = state.protocol.read().await;
assert!(proto.sessions.contains_key("incumbent"));
assert!(!proto.sessions.contains_key("staged"));
assert!(proto.lifecycle_leases.is_empty());
}
#[tokio::test]
async fn restore_persisted_sessions_rolls_pending_restart_back_to_literal_incumbent() {
use axum::Router;
use axum::extract::State as AxumState;
use axum::http::StatusCode;
use axum::routing::delete;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
async fn delete_session(AxumState(calls): AxumState<Arc<AtomicUsize>>) -> StatusCode {
calls.fetch_add(1, Ordering::SeqCst);
StatusCode::NO_CONTENT
}
let calls = Arc::new(AtomicUsize::new(0));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let app = Router::new()
.route("/session/{session_id}", delete(delete_session))
.with_state(calls.clone());
let server = tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let dir = tempfile::tempdir().unwrap();
let config = crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: port.checked_sub(320).unwrap(),
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
};
let state = crate::state::AppState::new(config.clone());
let incumbent_owner = crate::daemon_protocol::ResourceOwner {
session_id: "pending-restart".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(40),
};
{
let mut proto = state.protocol.write().await;
proto.restore_incarnation_high_water(incumbent_owner.incarnation);
proto.sessions.insert(
incumbent_owner.session_id.clone(),
crate::daemon_protocol::SessionEntry {
id: incumbent_owner.session_id.clone(),
pane: None,
origin: crate::daemon_protocol::Origin::Local,
metadata: crate::daemon_protocol::SessionMeta {
backend: Some("opencode".into()),
backend_session_id: Some("ses_incumbent".into()),
opencode_binding: Some(
crate::daemon_protocol::OpenCodeBinding::StrongManaged,
),
model: Some("incumbent-model".into()),
fresh_context_after_active_secs: Some(60),
active_context_accumulated_secs: 61,
active_context_segment_started_at: Some(100),
active_context_restart_due: true,
last_metadata_update: Some(777),
session_incarnation: incumbent_owner.incarnation,
..Default::default()
},
registered_at: 123,
active_context_due_boundary: Default::default(),
},
);
state.persist_protocol_state(&proto).unwrap();
}
assert_eq!(
state.claim_existing_start(&incumbent_owner).await.unwrap(),
crate::daemon_protocol::LifecycleMutationOutcome::Applied
);
let target_incarnation = match state
.stage_restart_launch(
&incumbent_owner,
"opencode".into(),
true,
true,
Some(120),
None,
None,
)
.await
{
crate::daemon_protocol::StageFreshLaunchOutcome::Staged { incarnation } => incarnation,
outcome => panic!("expected staged restart target, got {outcome:?}"),
};
assert_ne!(target_incarnation, incumbent_owner.incarnation);
{
let protocol = state.protocol.read().await;
let staged = &protocol.sessions["pending-restart"].metadata;
assert_eq!(staged.fresh_context_after_active_secs, Some(120));
assert_eq!(staged.active_context_accumulated_secs, 0);
assert_eq!(staged.active_context_segment_started_at, None);
assert!(!staged.active_context_restart_due);
assert!(staged.active_context_accounting_provisional);
}
let target_owner = crate::daemon_protocol::ResourceOwner {
session_id: incumbent_owner.session_id.clone(),
incarnation: target_incarnation,
};
assert_eq!(
state
.record_restart_backend_claim(
&incumbent_owner,
&target_owner,
"opencode".into(),
"ses_target".into(),
)
.await
.unwrap(),
crate::daemon_protocol::LifecycleMutationOutcome::Applied
);
let recovery = crate::state::AppState::new(config);
restore_persisted_sessions(&recovery).await.unwrap();
let restored_snapshot = crate::persistence::load_sessions(dir.path()).unwrap();
assert!(restored_snapshot.lifecycle_leases.is_empty());
assert_eq!(restored_snapshot.sessions.len(), 1);
let restored = &restored_snapshot.sessions[0];
assert_eq!(restored.id, incumbent_owner.session_id);
assert_eq!(
restored.metadata.session_incarnation,
incumbent_owner.incarnation
);
assert_eq!(
restored.metadata.backend_session_id.as_deref(),
Some("ses_incumbent")
);
assert_eq!(restored.metadata.model.as_deref(), Some("incumbent-model"));
assert_eq!(restored.metadata.fresh_context_after_active_secs, Some(60));
assert_eq!(restored.metadata.active_context_accumulated_secs, 61);
assert_eq!(
restored.metadata.active_context_segment_started_at,
Some(100)
);
assert!(restored.metadata.active_context_restart_due);
assert!(!restored.metadata.active_context_accounting_provisional);
assert_eq!(
restored
.metadata
.last_metadata_update
.map(|timestamp| timestamp.timestamp()),
Some(777)
);
assert_eq!(calls.load(Ordering::SeqCst), 1);
server.abort();
}
#[test]
fn restart_reconciliation_removes_same_incarnation_inert_fallback_row() {
let owner = crate::daemon_protocol::ResourceOwner {
session_id: "resumed".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(43),
};
let session = crate::persistence::PersistedSession {
id: owner.session_id.clone(),
pane: Some("%fallback".into()),
registered_at: chrono::Utc::now(),
last_activity_at: chrono::Utc::now(),
metadata: crate::state::SessionMetadata {
session_incarnation: owner.incarnation,
..Default::default()
},
};
let lease = crate::daemon_protocol::LifecycleLease {
owner: owner.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Restarting,
backend: None,
backend_session_id: None,
backend_session_owner: None,
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: None,
project_dir_owner: None,
project_dir_cleanup_on_abandon: false,
inert_pane: Some("%fallback".into()),
inert_pane_owner: Some(owner),
sweep_unconfirmed: None,
};
assert!(abandoned_lease_owns_staged_row(&session, &lease));
}
#[test]
fn direct_respawn_recovery_accepts_incumbent_and_staged_pane_owners() {
let incumbent = crate::daemon_protocol::ResourceOwner {
session_id: "resumed".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(43),
};
let staged = crate::daemon_protocol::ResourceOwner {
session_id: "resumed".into(),
incarnation: crate::daemon_protocol::SessionIncarnation(44),
};
let lease = crate::daemon_protocol::LifecycleLease {
owner: incumbent.clone(),
phase: crate::daemon_protocol::LifecyclePhase::Restarting,
backend: None,
backend_session_id: None,
backend_session_owner: None,
restart_target_owner: None,
restart_previous: None,
dormant_resume_previous: None,
project_dir: None,
project_dir_owner: None,
project_dir_cleanup_on_abandon: false,
inert_pane: Some("%existing".into()),
inert_pane_owner: Some(staged.clone()),
sweep_unconfirmed: None,
};
assert_eq!(lifecycle_lease_pane_owners(&lease), vec![incumbent, staged]);
}
#[tokio::test]
async fn restore_persisted_sessions_fails_closed_on_corrupt_authority() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("sessions.json"), "{not-json").unwrap();
let state = crate::state::AppState::new(crate::config::OuijaConfig {
name: "test".into(),
npub: "npub1test".into(),
port: 0,
data_dir: dir.path().to_path_buf(),
config_dir: dir.path().to_path_buf(),
});
let result = restore_persisted_sessions(&state).await;
assert!(
result.is_err(),
"daemon startup must not enable lifecycle allocation after authority load failure"
);
}
#[test]
fn stop_plan_stops_systemd_and_legacy_when_unit_exists() {
let plan = plan_daemon_stop(true);
assert!(plan.stop_systemd);
assert!(
plan.stop_legacy,
"stop-server must still clean up stray legacy daemons when a unit exists"
);
}
#[test]
fn stop_plan_uses_legacy_only_without_systemd_unit() {
let plan = plan_daemon_stop(false);
assert!(!plan.stop_systemd);
assert!(plan.stop_legacy);
}
#[test]
fn supervised_lifecycle_uses_systemd_for_active_unit() {
assert_eq!(
plan_supervised_lifecycle(true, true),
DaemonLifecyclePlan::SystemdOnly
);
}
#[test]
fn supervised_lifecycle_cleans_legacy_before_inactive_unit_start() {
assert_eq!(
plan_supervised_lifecycle(true, false),
DaemonLifecyclePlan::SystemdAfterLegacyCleanup
);
}
#[test]
fn supervised_lifecycle_uses_legacy_without_systemd_unit() {
assert_eq!(
plan_supervised_lifecycle(false, false),
DaemonLifecyclePlan::LegacyOnly
);
}
#[test]
fn stop_daemon_processes_runs_legacy_cleanup_when_systemd_stop_fails() {
use std::cell::Cell;
let legacy_called = Cell::new(false);
let err = stop_daemon_processes_with(
true,
true,
|| -> anyhow::Result<()> { anyhow::bail!("systemd stop failed") },
|| {
legacy_called.set(true);
LegacyStopOutcome {
tmux_killed: true,
process_killed: false,
}
},
)
.unwrap_err();
assert!(legacy_called.get());
assert!(err.to_string().contains("systemd stop failed"));
}
#[test]
fn systemd_after_legacy_cleanup_has_settle_delay() {
assert_eq!(
legacy_cleanup_settle_delay(&DaemonLifecyclePlan::SystemdAfterLegacyCleanup),
Some(std::time::Duration::from_secs(1))
);
assert_eq!(
legacy_cleanup_settle_delay(&DaemonLifecyclePlan::SystemdOnly),
None
);
assert_eq!(
legacy_cleanup_settle_delay(&DaemonLifecyclePlan::LegacyOnly),
None
);
}
#[test]
fn pick_session_id_reports_pane_var_as_source() {
let res = pick_session_id(Some("%3"), Some("keycast".into()), None);
assert_eq!(
res,
SessionIdResolution::Found("keycast".into(), IdentitySource::PaneVar)
);
}
#[test]
fn pick_session_id_reports_env_var_as_source() {
let res = pick_session_id(None, None, Some("hub".into()));
assert_eq!(
res,
SessionIdResolution::Found("hub".into(), IdentitySource::EnvVar)
);
}
#[test]
fn whoami_failure_outside_tmux_lists_missing_signals_and_forbids_guessing() {
let failure = WhoamiFailure {
tmux_pane: None,
pane_var: None,
env_var: None,
lookup: None,
};
let msg = format_whoami_failure(&failure);
assert!(
msg.contains("$TMUX_PANE: not set"),
"must report the missing tmux pane signal, got: {msg}"
);
assert!(
msg.contains("$OUIJA_SESSION_ID: not set"),
"must report the missing env var signal, got: {msg}"
);
assert!(
msg.contains("Never guess"),
"must explicitly forbid guessing a sender id, got: {msg}"
);
assert!(
msg.contains("project directory"),
"must call out the project-basename guess that caused the incident, got: {msg}"
);
assert!(
!msg.contains("ouija register"),
"must never steer an unresolved caller toward `ouija register`, got: {msg}"
);
}
#[test]
fn whoami_failure_in_tmux_reports_pane_lookup_miss() {
let failure = WhoamiFailure {
tmux_pane: Some("%3".into()),
pane_var: None,
env_var: None,
lookup: Some(PaneLookupFailure::NoSessionForPane),
};
let msg = format_whoami_failure(&failure);
assert!(
msg.contains("$TMUX_PANE: %3"),
"must show the pane that was checked, got: {msg}"
);
assert!(
msg.contains("@ouija_session"),
"must report the pane var signal by name, got: {msg}"
);
assert!(
msg.contains("no registered session"),
"must say the daemon lookup found nothing for this pane, got: {msg}"
);
}
#[test]
fn whoami_failure_reports_unreachable_daemon() {
let failure = WhoamiFailure {
tmux_pane: Some("%3".into()),
pane_var: None,
env_var: None,
lookup: Some(PaneLookupFailure::DaemonUnreachable(
"http://localhost:7880".into(),
)),
};
let msg = format_whoami_failure(&failure);
assert!(
msg.contains("daemon unreachable at http://localhost:7880"),
"must distinguish an unreachable daemon from a pane miss, got: {msg}"
);
}
#[test]
fn whoami_unregistered_identity_names_id_and_source_without_guessing() {
let msg = format_unregistered_identity("stale-id", &IdentitySource::EnvVar);
assert!(
msg.contains("stale-id"),
"must name the resolved-but-unregistered id, got: {msg}"
);
assert!(
msg.contains("$OUIJA_SESSION_ID"),
"must say which signal produced the stale id, got: {msg}"
);
assert!(
msg.contains("renamed"),
"must explain the likely cause (rename/removal), got: {msg}"
);
assert!(
msg.contains("Never guess"),
"must forbid guessing a replacement id, got: {msg}"
);
assert!(
!msg.contains("ouija register"),
"must never suggest `ouija register`, got: {msg}"
);
}
#[test]
fn status_lists_local_session_matches_local_origin_only() {
let status = serde_json::json!({
"sessions": [
{"id": "mine", "origin": "local"},
{"id": "peer/mine", "origin": "remote"},
]
});
assert!(status_lists_local_session(&status, "mine"));
assert!(
!status_lists_local_session(&status, "peer/mine"),
"a remote session id is never the local caller's identity"
);
assert!(!status_lists_local_session(&status, "absent"));
}
#[test]
fn unresolved_sender_error_points_at_whoami_not_register() {
let msg = unresolved_sender_error();
assert!(
msg.contains("ouija whoami"),
"unresolved identity must steer callers to whoami diagnostics, got: {msg}"
);
assert!(
msg.contains("Never guess"),
"must forbid guessing a sender id, got: {msg}"
);
assert!(
!msg.contains("ouija register"),
"must never steer callers toward `ouija register`, got: {msg}"
);
}
#[test]
fn config_help_describes_default_backend_setting() {
let mut command = Cli::command();
let config = command
.find_subcommand_mut("config")
.expect("config subcommand");
let mut help = Vec::new();
config.write_long_help(&mut help).unwrap();
let help = String::from_utf8(help).unwrap();
assert!(help.contains("default_backend"));
assert!(help.contains("claude-code, opencode, codex-cli"));
}
}