#![allow(
deprecated,
reason = "the retained CLI composition still seeds and bridges legacy atm-core boundary stores during the Phase AC transition"
)]
use std::fmt;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Once};
use atm_core::ack::{AckOutcome, AckRequest};
use atm_core::boundary;
use atm_core::boundary::ClientTransport;
use atm_core::clear::{ClearOutcome, ClearQuery};
use atm_core::doctor::{BootstrapTraceReport, DoctorQuery, DoctorReport};
use atm_core::error::AtmError;
use atm_core::graft::AtmGraftClient;
use atm_core::home;
use atm_core::list::{ListOutcome, ListQuery};
use atm_core::observability::{CommandEvent, ObservabilityPort, action_name, outcome_label};
use atm_core::protocol::{
self, RequestEnvelope, ResponseEnvelope, SendRequestEnvelope, SendResponseEnvelope,
};
use atm_core::read::{PeekQuery, ReadOutcome, ReadQuery};
use atm_core::send::{SendOutcome, SendRequest};
#[cfg(not(test))]
use atm_daemon_bootstrap::install_sqlite_retained_runtime_factory;
use atm_daemon_client::{
BootstrapCommandEvent, BootstrapTraceability, DaemonLocalIpcEndpoint, DaemonSupervisor,
FramePayload, MessageKind, RequestId as DaemonRequestId, RpcEnvelope,
exchange_envelope as daemon_exchange_envelope, parse_bootstrap_agent, parse_bootstrap_team,
resolve_daemon_bin, resolve_daemon_local_ipc_endpoint,
resolve_daemon_local_ipc_endpoint_from_home, try_connect as daemon_try_connect,
unexpected_response,
};
#[cfg(test)]
use atm_daemon_client::{HOST_RUNTIME_LAUNCH_LOCK_FILE, LaunchGateGuard};
#[cfg(test)]
use atm_runtime_test_support::{
SQLITE_RUNTIME_PATH_ENV,
install_sqlite_retained_runtime_factory as install_test_runtime_factory, open_sqlite_boundary,
};
use crate::observability::CliObservability;
const SAME_HOST_REQUEST_DEADLINE: std::time::Duration = std::time::Duration::from_secs(3);
static INSTALL_RETAINED_RUNTIME_FACTORY: Once = Once::new();
#[cfg(not(test))]
fn install_retained_runtime_factory() {
INSTALL_RETAINED_RUNTIME_FACTORY.call_once(|| {
install_sqlite_retained_runtime_factory();
});
}
#[cfg(test)]
fn install_retained_runtime_factory() {
INSTALL_RETAINED_RUNTIME_FACTORY.call_once(|| {
install_test_runtime_factory();
});
}
#[derive(Debug, Default)]
pub(crate) struct SendCommandEntryPoint;
impl SendCommandEntryPoint {
pub(crate) const fn new() -> Self {
Self
}
}
#[derive(Debug, Default)]
pub(crate) struct ReceiveCommandEntryPoint;
impl ReceiveCommandEntryPoint {
pub(crate) const fn new() -> Self {
Self
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum CliBootstrapError {
AtmHomeUnresolved {
command: &'static str,
},
RuntimeRootInvalid {
command: &'static str,
atm_home: PathBuf,
invocation_dir: PathBuf,
},
RuntimeBootstrapRefused {
command: &'static str,
atm_home: PathBuf,
daemon_endpoint: PathBuf,
},
}
impl CliBootstrapError {
fn into_atm_error(self) -> AtmError {
match self {
Self::AtmHomeUnresolved { command } => AtmError::atm_home_unresolved(format!(
"failed to resolve ATM_HOME before bootstrapping `atm {command}`"
)),
Self::RuntimeRootInvalid {
command,
atm_home,
invocation_dir,
} => AtmError::runtime_root_invalid(format!(
"invalid runtime root for `atm {command}`: atm_home={} invocation_dir={}",
atm_home.display(),
invocation_dir.display()
)),
Self::RuntimeBootstrapRefused {
command,
atm_home,
daemon_endpoint,
} => AtmError::runtime_bootstrap_refused(format!(
"refused bootstrap for `atm {command}` because daemon endpoint {} does not match the canonical ATM_HOME root {}",
daemon_endpoint.display(),
atm_home.display()
)),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct InvocationDir<'a>(&'a Path);
impl<'a> InvocationDir<'a> {
pub(crate) fn new(path: &'a Path) -> Self {
Self(path)
}
fn as_path(self) -> &'a Path {
self.0
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct AtmHomePath<'a>(&'a Path);
impl<'a> AtmHomePath<'a> {
pub(crate) fn new(path: &'a Path) -> Self {
Self(path)
}
fn as_path(self) -> &'a Path {
self.0
}
}
pub(crate) fn resolve_command_runtime_context(
command: &'static str,
) -> Result<(PathBuf, PathBuf), AtmError> {
let invocation_dir = home::command_invocation_dir().inspect_err(|error| {
log_runtime_root_failure(command, error);
})?;
let atm_home = home::atm_home().map_err(|source| {
let error = CliBootstrapError::AtmHomeUnresolved { command }
.into_atm_error()
.with_source(source);
log_runtime_root_failure(command, &error);
error
})?;
Ok((atm_home, invocation_dir))
}
fn log_runtime_root_failure(command: &'static str, error: &AtmError) {
tracing::error!(
command,
error_code = %error.code.as_str(),
error = %error,
"raw cli runtime-root failure"
);
}
#[derive(Debug)]
struct LocalIpcClientTransportAdapter {
endpoint: DaemonLocalIpcEndpoint,
}
impl LocalIpcClientTransportAdapter {
fn new(endpoint: DaemonLocalIpcEndpoint) -> Self {
Self { endpoint }
}
fn probe_connection(&self) -> Result<interprocess::local_socket::Stream, AtmError> {
daemon_try_connect(&self.endpoint)
}
fn round_trip(&self, request: RequestEnvelope) -> Result<ResponseEnvelope, AtmError> {
let envelope = encode_request_envelope(request)?;
let response =
daemon_exchange_envelope(&self.endpoint, envelope, SAME_HOST_REQUEST_DEADLINE)?;
decode_response_envelope(response)
}
}
impl boundary::sealed::Sealed for LocalIpcClientTransportAdapter {}
impl ClientTransport for LocalIpcClientTransportAdapter {
fn send(&self, request: RequestEnvelope) -> Result<ResponseEnvelope, AtmError> {
self.round_trip(request)
}
}
pub(crate) struct CliComposition<'a> {
transport: Arc<dyn ClientTransport + Send + Sync + 'a>,
observability_port: &'a CliObservability,
bootstrap_trace: Option<BootstrapTraceReport>,
send_command: SendCommandEntryPoint,
receive_command: ReceiveCommandEntryPoint,
}
impl fmt::Debug for CliComposition<'_> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("CliComposition")
.field("transport", &"dyn ClientTransport")
.field("observability_port", &"dyn ObservabilityPort")
.field("bootstrap_trace", &self.bootstrap_trace)
.field("send_command", &self.send_command)
.field("receive_command", &self.receive_command)
.finish()
}
}
impl<'a> CliComposition<'a> {
pub(crate) fn from_transport(
transport: Arc<dyn ClientTransport + Send + Sync + 'a>,
observability_port: &'a CliObservability,
) -> Self {
install_retained_runtime_factory();
Self {
transport,
observability_port,
bootstrap_trace: None,
send_command: SendCommandEntryPoint::new(),
receive_command: ReceiveCommandEntryPoint::new(),
}
}
#[cfg(test)]
pub(crate) fn from_transport_with_bootstrap_trace(
transport: Arc<dyn ClientTransport + Send + Sync + 'a>,
observability_port: &'a CliObservability,
bootstrap_trace: BootstrapTraceReport,
) -> Self {
install_retained_runtime_factory();
Self {
transport,
observability_port,
bootstrap_trace: Some(bootstrap_trace),
send_command: SendCommandEntryPoint::new(),
receive_command: ReceiveCommandEntryPoint::new(),
}
}
#[expect(
dead_code,
reason = "reserved for future phase that inspects the active transport variant"
)]
pub(crate) fn transport(&self) -> &(dyn ClientTransport + Send + Sync + 'a) {
self.transport.as_ref()
}
pub(crate) fn send_request(
&self,
request: RequestEnvelope,
) -> Result<ResponseEnvelope, AtmError> {
match self.transport.send(request)? {
ResponseEnvelope::Error(error) => Err(error.into_atm_error()),
response => Ok(response),
}
}
#[expect(
dead_code,
reason = "reserved for future phase that threads observability port into command helpers"
)]
pub(crate) fn observability_port(&self) -> &(dyn ObservabilityPort + Send + Sync) {
self.observability_port
}
#[expect(
dead_code,
reason = "reserved for future command-routing phase — exposes send entry-point to callers"
)]
pub(crate) fn send_command(&self) -> &SendCommandEntryPoint {
&self.send_command
}
#[expect(
dead_code,
reason = "reserved for future command-routing phase — exposes receive entry-point to callers"
)]
pub(crate) fn receive_command(&self) -> &ReceiveCommandEntryPoint {
&self.receive_command
}
pub(crate) fn send(&self, request: SendRequest) -> Result<SendOutcome, AtmError> {
match self.send_request(RequestEnvelope::Send(SendRequestEnvelope::Compose(request)))? {
ResponseEnvelope::Send(SendResponseEnvelope::Sent(outcome)) => {
self.observability_port.emit_command_event(CommandEvent {
command: "send",
action: action_name("send"),
outcome: outcome_label(if outcome.dry_run { "dry_run" } else { "sent" }),
team: outcome.team.clone(),
agent: outcome.agent.clone(),
sender: outcome.sender.clone(),
message_id: Some(outcome.message_id),
requires_ack: outcome.requires_ack,
dry_run: outcome.dry_run,
task_id: outcome.task_id.clone(),
error_code: None,
error_message: None,
});
Ok(outcome)
}
other => Err(unexpected_response("send", other)),
}
}
pub(crate) fn ack(&self, request: AckRequest) -> Result<AckOutcome, AtmError> {
match self.send_request(RequestEnvelope::Send(SendRequestEnvelope::Acknowledge(
request,
)))? {
ResponseEnvelope::Send(SendResponseEnvelope::Acknowledged(outcome)) => {
self.observability_port.emit_command_event(CommandEvent {
command: "ack",
action: action_name("ack"),
outcome: outcome_label("ok"),
team: outcome.team.clone(),
agent: outcome.agent.clone(),
sender: outcome.agent.clone(),
message_id: Some(outcome.message_id),
requires_ack: false,
dry_run: false,
task_id: outcome.task_id.clone(),
error_code: None,
error_message: None,
});
Ok(outcome)
}
other => Err(unexpected_response("ack", other)),
}
}
pub(crate) fn receive(&self, query: ReadQuery) -> Result<ReadOutcome, AtmError> {
match self.send_request(RequestEnvelope::Receive(query))? {
ResponseEnvelope::Receive(outcome) => {
self.observability_port.emit_command_event(CommandEvent {
command: "read",
action: action_name("read"),
outcome: outcome_label("ok"),
team: outcome.team.clone(),
agent: outcome.agent.clone(),
sender: outcome.agent.clone(),
message_id: None,
requires_ack: false,
dry_run: false,
task_id: None,
error_code: None,
error_message: None,
});
Ok(*outcome)
}
other => Err(unexpected_response("receive", other)),
}
}
pub(crate) fn peek(&self, query: PeekQuery) -> Result<ReadOutcome, AtmError> {
match self.send_request(RequestEnvelope::Peek(query))? {
ResponseEnvelope::Peek(outcome) => {
self.observability_port.emit_command_event(CommandEvent {
command: "peek",
action: action_name("peek"),
outcome: outcome_label("ok"),
team: outcome.team.clone(),
agent: outcome.agent.clone(),
sender: outcome.agent.clone(),
message_id: None,
requires_ack: false,
dry_run: false,
task_id: None,
error_code: None,
error_message: None,
});
Ok(*outcome)
}
other => Err(unexpected_response("peek", other)),
}
}
pub(crate) fn list(&self, query: ListQuery) -> Result<ListOutcome, AtmError> {
match self.send_request(RequestEnvelope::List(query))? {
ResponseEnvelope::List(outcome) => {
self.observability_port.emit_command_event(CommandEvent {
command: "list",
action: action_name("list"),
outcome: outcome_label("ok"),
team: outcome.team.clone(),
agent: outcome.agent.clone(),
sender: outcome.agent.clone(),
message_id: None,
requires_ack: false,
dry_run: false,
task_id: None,
error_code: None,
error_message: None,
});
Ok(outcome)
}
other => Err(unexpected_response("list", other)),
}
}
pub(crate) fn clear(&self, query: ClearQuery) -> Result<ClearOutcome, AtmError> {
match self.send_request(RequestEnvelope::Clear(query))? {
ResponseEnvelope::Clear(outcome) => {
self.observability_port.emit_command_event(CommandEvent {
command: "clear",
action: action_name("clear"),
outcome: outcome_label("ok"),
team: outcome.team.clone(),
agent: outcome.agent.clone(),
sender: outcome.agent.clone(),
message_id: None,
requires_ack: false,
dry_run: false,
task_id: None,
error_code: None,
error_message: None,
});
Ok(outcome)
}
other => Err(unexpected_response("clear", other)),
}
}
#[allow(
dead_code,
reason = "AA.3 restores the direct local doctor path in DoctorCommand, but the daemon-routed doctor request seam remains covered by transport tests."
)]
pub(crate) fn doctor(&self, query: DoctorQuery) -> Result<DoctorReport, AtmError> {
match self.send_request(RequestEnvelope::Doctor(query))? {
ResponseEnvelope::Doctor(mut report) => {
report.bootstrap_trace = self.bootstrap_trace.clone();
Ok(*report)
}
other => Err(unexpected_response("doctor", other)),
}
}
pub(crate) fn bootstrap(
command: &'static str,
observability: &'a CliObservability,
invocation_dir: InvocationDir<'_>,
atm_home: AtmHomePath<'_>,
) -> Result<Self, AtmError> {
let invocation_dir = invocation_dir.as_path();
let atm_home = atm_home.as_path();
let canonical_endpoint =
resolve_daemon_local_ipc_endpoint_from_home(atm_home).map_err(|source| {
let error = CliBootstrapError::RuntimeRootInvalid {
command,
atm_home: atm_home.to_path_buf(),
invocation_dir: invocation_dir.to_path_buf(),
}
.into_atm_error()
.with_source(source);
log_runtime_root_failure(command, &error);
error
})?;
let endpoint =
if std::env::var_os("ATM_DAEMON_SOCKET").is_some_and(|value| !value.is_empty()) {
let overridden = resolve_daemon_local_ipc_endpoint().map_err(|source| {
let error = CliBootstrapError::RuntimeRootInvalid {
command,
atm_home: atm_home.to_path_buf(),
invocation_dir: invocation_dir.to_path_buf(),
}
.into_atm_error()
.with_source(source);
log_runtime_root_failure(command, &error);
error
})?;
if overridden.as_ref() != canonical_endpoint.as_ref() {
let error = CliBootstrapError::RuntimeBootstrapRefused {
command,
atm_home: atm_home.to_path_buf(),
daemon_endpoint: overridden.as_ref().to_path_buf(),
}
.into_atm_error();
log_runtime_root_failure(command, &error);
return Err(error);
}
overridden
} else {
canonical_endpoint
};
let daemon_bin = resolve_daemon_bin("atm")?;
let transport = Arc::new(LocalIpcClientTransportAdapter::new(endpoint.clone()));
let supervisor = DaemonSupervisor::new(endpoint, daemon_bin);
let emit_bootstrap_event = |event: BootstrapCommandEvent| {
observability.emit(CommandEvent {
command: event.command,
action: action_name(event.action),
outcome: outcome_label(event.outcome),
team: event.team,
agent: event.agent.clone(),
sender: event.agent,
message_id: None,
requires_ack: false,
dry_run: false,
task_id: None,
error_code: event.error_code,
error_message: event.error_message,
})
};
let traceability = BootstrapTraceability::new(
command,
&emit_bootstrap_event,
parse_bootstrap_team()?,
parse_bootstrap_agent()?,
);
supervisor.ensure_daemon_available_with_traceability(&traceability, || {
transport.probe_connection().map(|_| ())
})?;
let mut composition = Self::from_transport(transport, observability);
composition.bootstrap_trace = Some(bootstrap_trace_to_core(traceability.snapshot()));
Ok(composition)
}
}
fn encode_request_envelope(request: RequestEnvelope) -> Result<RpcEnvelope, AtmError> {
let request_id = protocol::next_request_id();
let frame = protocol::request_to_frame_payload(request_id, request)?;
Ok(RpcEnvelope::from_frame_payload(encode_daemon_frame(frame)?))
}
fn decode_response_envelope(envelope: RpcEnvelope) -> Result<ResponseEnvelope, AtmError> {
let frame = decode_daemon_frame(envelope.into_frame_payload())?;
let (_, response) = protocol::response_from_frame_payload(frame)?;
Ok(response)
}
fn encode_daemon_frame(frame: protocol::FramePayload) -> Result<FramePayload, AtmError> {
Ok(FramePayload {
request_id: DaemonRequestId::new(frame.request_id.into_inner())?,
message_kind: MessageKind::try_from(frame.message_kind.code())?,
flags: frame.flags,
bytes: frame.bytes,
})
}
fn decode_daemon_frame(frame: FramePayload) -> Result<protocol::FramePayload, AtmError> {
Ok(protocol::FramePayload {
request_id: protocol::RequestId::new(frame.request_id.into_inner())?,
message_kind: protocol::MessageKind::try_from(frame.message_kind.code())?,
flags: frame.flags,
bytes: frame.bytes,
})
}
fn bootstrap_trace_to_core(
report: atm_daemon_client::BootstrapTraceReport,
) -> BootstrapTraceReport {
use atm_core::doctor::{
BootstrapAutoStartOutcome as CoreAutoStart, BootstrapConnectOutcome as CoreConnect,
BootstrapLaunchGateOutcome as CoreLaunch,
};
BootstrapTraceReport {
daemon_connect: match report.daemon_connect {
atm_daemon_client::BootstrapConnectOutcome::Connected => CoreConnect::Connected,
atm_daemon_client::BootstrapConnectOutcome::NotFound => CoreConnect::NotFound,
atm_daemon_client::BootstrapConnectOutcome::Timeout => CoreConnect::Timeout,
atm_daemon_client::BootstrapConnectOutcome::Failed => CoreConnect::Failed,
},
daemon_launch_gate: match report.daemon_launch_gate {
atm_daemon_client::BootstrapLaunchGateOutcome::Launched => CoreLaunch::Launched,
atm_daemon_client::BootstrapLaunchGateOutcome::Failed => CoreLaunch::Failed,
atm_daemon_client::BootstrapLaunchGateOutcome::Skipped => CoreLaunch::Skipped,
},
daemon_auto_start: match report.daemon_auto_start {
atm_daemon_client::BootstrapAutoStartOutcome::AutoStarted => CoreAutoStart::AutoStarted,
atm_daemon_client::BootstrapAutoStartOutcome::Failed => CoreAutoStart::Failed,
atm_daemon_client::BootstrapAutoStartOutcome::Skipped => CoreAutoStart::Skipped,
},
connect_detail: report.connect_detail,
launch_gate_detail: report.launch_gate_detail,
auto_start_detail: report.auto_start_detail,
}
}
impl AtmGraftClient for CliComposition<'_> {
fn send_message(&self, request: SendRequest) -> Result<SendOutcome, AtmError> {
CliComposition::send(self, request)
}
fn read_message(&self, query: ReadQuery) -> Result<ReadOutcome, AtmError> {
CliComposition::receive(self, query)
}
fn acknowledge_message(&self, request: AckRequest) -> Result<AckOutcome, AtmError> {
CliComposition::ack(self, request)
}
}
#[cfg(test)]
mod tests {
#[cfg(unix)]
use std::ffi::OsString;
use std::fs;
use std::io;
use std::io::Write;
#[cfg(unix)]
use std::os::unix::ffi::OsStringExt;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use atm_core::ack::AckRequest;
use atm_core::boundary;
use atm_core::boundary::ClientTransport;
use atm_core::clear::ClearQuery;
use atm_core::doctor::{
BootstrapAutoStartOutcome, BootstrapConnectOutcome, BootstrapLaunchGateOutcome,
BootstrapTraceReport, DoctorQuery, DoctorStatus,
};
use atm_core::error::AtmError;
use atm_core::graft::AtmGraftClient;
use atm_core::protocol::{
ProtocolErrorEnvelope, RequestEnvelope, ResponseEnvelope, SendRequestEnvelope,
};
use atm_core::read::{PeekQuery, ReadQuery};
use atm_core::schema::{AgentMember, AtmMessageId, InboxMessage, TeamConfig};
use atm_core::send::{SendMessageSource, SendRequest};
use atm_core::test_support::{
EnvGuard, ROLE_TEAM_LEAD, TEST_LEAD, TEST_RECIPIENT, TEST_RECIPIENT_ADDRESS, TEST_SENDER,
TEST_TEAM,
};
use atm_core::transport::testing::{
FakeClientTransport, HealthyObservability, LoopbackClientTransport,
};
use atm_core::types::ReadSelection;
use atm_core::types::{AgentName, TeamName};
use atm_daemon_client::DaemonBinaryPath;
use chrono::Utc;
use serde_json::{Map, Value};
use serial_test::serial;
use tempfile::TempDir;
use tracing_subscriber::fmt::MakeWriter;
use super::{
AtmHomePath, CliComposition, DaemonLocalIpcEndpoint, DaemonSupervisor,
HOST_RUNTIME_LAUNCH_LOCK_FILE, InvocationDir, LaunchGateGuard,
LocalIpcClientTransportAdapter, SQLITE_RUNTIME_PATH_ENV, open_sqlite_boundary,
resolve_command_runtime_context,
};
use crate::observability::CliObservability;
struct CwdGuard {
original: std::path::PathBuf,
}
impl CwdGuard {
fn change_to(path: &std::path::Path) -> Self {
let original = atm_core::home::command_invocation_dir().expect("current dir");
std::env::set_current_dir(path).expect("set current dir");
Self { original }
}
}
impl Drop for CwdGuard {
fn drop(&mut self) {
std::env::set_current_dir(&self.original).expect("restore current dir");
}
}
#[derive(Clone, Default)]
struct SharedLogBuffer(Arc<Mutex<Vec<u8>>>);
struct SharedLogWriter(SharedLogBuffer);
impl<'a> MakeWriter<'a> for SharedLogBuffer {
type Writer = SharedLogWriter;
fn make_writer(&'a self) -> Self::Writer {
SharedLogWriter(self.clone())
}
}
impl Write for SharedLogWriter {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
self.0
.0
.lock()
.expect("log buffer lock")
.extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
impl SharedLogBuffer {
fn contents(&self) -> String {
String::from_utf8(self.0.lock().expect("log buffer lock").clone()).expect("utf8 logs")
}
}
fn capture_runtime_root_logs<T>(f: impl FnOnce() -> T) -> (T, String) {
let buffer = SharedLogBuffer::default();
let subscriber = tracing_subscriber::fmt()
.with_ansi(false)
.without_time()
.with_writer(buffer.clone())
.finish();
let result = tracing::subscriber::with_default(subscriber, f);
(result, buffer.contents())
}
struct LoopbackFixture {
_env_guard: EnvGuard,
_tempdir: TempDir,
home_dir: std::path::PathBuf,
current_dir: std::path::PathBuf,
}
impl LoopbackFixture {
fn new(recipient: &str) -> Self {
super::install_retained_runtime_factory();
let tempdir = tempfile::tempdir().expect("tempdir");
let home_dir = tempdir.path().to_path_buf();
let sqlite_db_path = home_dir.join("runtime").join("mail.sqlite3");
let current_dir = tempdir.path().join("cwd");
fs::create_dir_all(¤t_dir).expect("cwd");
fs::write(current_dir.join(".atm.toml"), "[atm]\n").expect("fixture atm config");
let env_guard = EnvGuard::set_many([
(
"ATM_HOME",
Some(home_dir.to_str().expect("utf-8 tempdir path")),
),
(
SQLITE_RUNTIME_PATH_ENV,
Some(sqlite_db_path.to_str().expect("utf-8 sqlite db path")),
),
]);
let fixture = Self {
_env_guard: env_guard,
_tempdir: tempdir,
home_dir,
current_dir,
};
fixture.write_team_config(recipient);
fixture
}
fn team_dir(&self) -> std::path::PathBuf {
self.home_dir.join(".claude").join("teams").join(TEST_TEAM)
}
fn inbox_path(&self, agent: &str) -> std::path::PathBuf {
self.team_dir()
.join("inboxes")
.join(format!("{agent}.json"))
}
fn sqlite_db_path(&self) -> std::path::PathBuf {
self.home_dir.join("runtime").join("mail.sqlite3")
}
fn write_team_config(&self, recipient: &str) {
let team_dir = self.team_dir();
fs::create_dir_all(&team_dir).expect("team dir");
fs::create_dir_all(team_dir.join("inboxes")).expect("inboxes dir");
fs::create_dir_all(team_dir.join(".atm-state").join("workflow")).expect("workflow dir");
let config = TeamConfig {
members: vec![
AgentMember::with_name(TEST_SENDER.parse().expect("sender")),
AgentMember::with_name(recipient.parse().expect("recipient")),
AgentMember::with_name(TEST_LEAD.parse().expect("lead")),
],
..Default::default()
};
fs::write(
team_dir.join("config.json"),
serde_json::to_vec(&config).expect("team config"),
)
.expect("write team config");
self.seed_sqlite_roster(recipient);
}
fn seed_sqlite_roster(&self, recipient: &str) {
let assembly = open_sqlite_boundary(self.sqlite_db_path()).expect("sqlite db");
let roster_store = assembly.roster_store_arc();
let team = TEST_TEAM.parse::<TeamName>().expect("team");
let members = [TEST_SENDER, recipient, TEST_LEAD]
.into_iter()
.map(|agent| boundary::RosterEntry {
team_name: team.clone(),
agent_name: agent.parse().expect("agent"),
member_kind: boundary::RosterMemberKind::Permanent,
harness: boundary::RosterHarness::ClaudeCode,
agent_type: atm_core::schema::AgentType::default(),
model: atm_core::types::ModelName::default(),
recipient_pane_id: None,
metadata_json: Map::new(),
})
.collect::<Vec<_>>();
roster_store
.replace_roster(
&team,
&members,
Some(&boundary::ReplaySource::new("config.json").expect("source")),
)
.expect("seed sqlite roster");
}
fn write_inbox_values(&self, agent: &str, values: &[Value]) {
let inbox_path = self.inbox_path(agent);
if let Some(parent) = inbox_path.parent() {
fs::create_dir_all(parent).expect("inbox dir");
}
fs::write(
inbox_path,
serde_json::to_string_pretty(values).expect("json array"),
)
.expect("write inbox");
let messages = values
.iter()
.cloned()
.map(serde_json::from_value::<InboxMessage>)
.collect::<Result<Vec<_>, _>>()
.expect("message envelopes");
self.seed_sqlite_mailbox(agent, &messages);
}
fn inbox_contents(&self, agent: &str) -> Vec<InboxMessage> {
if self.sqlite_db_path().exists() {
let assembly = open_sqlite_boundary(self.sqlite_db_path()).expect("sqlite db");
let mail_store = assembly.mail_store_arc();
let team = TEST_TEAM.parse::<TeamName>().expect("team");
let agent_name = agent.parse::<AgentName>().expect("agent");
let metadata_rows = mail_store
.query_mailbox_metadata(&team, &agent_name, None)
.expect("mailbox rows");
return metadata_rows
.into_iter()
.map(|row| {
mail_store
.load_message(&team, &agent_name, &row.message_key)
.expect("message record")
.expect("stored message")
.envelope
})
.collect();
}
let inbox_path = self.inbox_path(agent);
if let Ok(raw) = fs::read_to_string(&inbox_path) {
if raw.trim_start().starts_with('[') {
let values: Vec<Value> = serde_json::from_str(&raw).expect("json array");
return values
.into_iter()
.map(|value| serde_json::from_value(value).expect("message envelope"))
.collect();
}
return raw
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| serde_json::from_str::<InboxMessage>(line).expect("json line"))
.collect();
}
Vec::new()
}
fn write_inbox_messages(&self, agent: &str, messages: &[InboxMessage]) {
let values = messages
.iter()
.map(|message| serde_json::to_value(message).expect("message value"))
.collect::<Vec<_>>();
self.write_inbox_values(agent, &values);
self.seed_sqlite_mailbox(agent, messages);
}
fn seed_sqlite_mailbox(&self, agent: &str, messages: &[InboxMessage]) {
let assembly = open_sqlite_boundary(self.sqlite_db_path()).expect("sqlite db");
let mail_store = assembly.mail_store_arc();
let team = TEST_TEAM.parse::<TeamName>().expect("team");
let agent_name = agent.parse::<AgentName>().expect("agent");
for (index, message) in messages.iter().enumerate() {
let message_key = if let Some(message_id) = message.message_id {
boundary::MessageKey::from(message_id)
} else {
boundary::MessageKey::new(format!("ext:{agent}:{index}")).expect("message key")
};
mail_store
.upsert_message(boundary::Message {
team: team.clone(),
agent: agent_name.clone(),
message_key: message_key.clone(),
envelope: message.clone(),
})
.expect("seed sqlite message");
mail_store
.upsert_message_state(boundary::UpsertMailMessageStateRequest {
team: team.clone(),
agent: agent_name.clone(),
actor: agent_name.clone(),
state: boundary::MailMessageState {
team: team.clone(),
agent: agent_name.clone(),
actor: agent_name.clone(),
message_key,
read: message.read,
pending_ack_at: message.pending_ack_at,
acknowledged_at: message.acknowledged_at,
expires_at: message.expires_at,
deleted_at: None,
updated_at: Some(message.timestamp),
},
})
.expect("seed sqlite message state");
}
}
fn send_request(&self, body: &str) -> SendRequest {
SendRequest::new(
self.home_dir.clone(),
self.current_dir.clone(),
TEST_SENDER.parse().expect("caller"),
TEST_RECIPIENT_ADDRESS,
TEST_TEAM.parse().expect("team"),
SendMessageSource::Inline(body.to_string()),
None,
false,
None,
false,
)
.expect("send request")
}
fn send_request_to(&self, recipient: &str, body: &str) -> SendRequest {
SendRequest::new(
self.home_dir.clone(),
self.current_dir.clone(),
TEST_SENDER.parse().expect("caller"),
recipient,
TEST_TEAM.parse().expect("team"),
SendMessageSource::Inline(body.to_string()),
None,
false,
None,
false,
)
.expect("send request")
}
fn send_request_with_flags(
&self,
body: &str,
requires_ack: bool,
task_id: Option<atm_core::types::TaskId>,
) -> SendRequest {
SendRequest::new(
self.home_dir.clone(),
self.current_dir.clone(),
TEST_SENDER.parse().expect("caller"),
TEST_RECIPIENT_ADDRESS,
TEST_TEAM.parse().expect("team"),
SendMessageSource::Inline(body.to_string()),
None,
requires_ack,
task_id,
false,
)
.expect("send request")
}
fn ack_request(&self, message_id: AtmMessageId, reply_body: &str) -> AckRequest {
AckRequest {
home_dir: self.home_dir.clone(),
current_dir: self.current_dir.clone(),
caller_identity: TEST_SENDER.parse().expect("caller"),
caller_team: TEST_TEAM.parse().expect("team"),
message_id,
reply_body: reply_body.to_string(),
}
}
fn read_query(&self) -> ReadQuery {
ReadQuery::new(
self.home_dir.clone(),
self.current_dir.clone(),
TEST_RECIPIENT.parse().expect("caller"),
None,
TEST_TEAM.parse().expect("team"),
ReadSelection::All,
false,
false,
None,
None,
None,
None,
None,
None,
)
.expect("read query")
}
fn read_query_for(&self, caller: &str, message_id: AtmMessageId) -> ReadQuery {
ReadQuery::new(
self.home_dir.clone(),
self.current_dir.clone(),
caller.parse().expect("caller"),
None,
TEST_TEAM.parse().expect("team"),
ReadSelection::All,
false,
false,
Some(&message_id.to_string()),
None,
None,
None,
None,
None,
)
.expect("read query")
}
fn peek_query_for(
&self,
caller: &str,
target: Option<&str>,
message_id: AtmMessageId,
) -> PeekQuery {
PeekQuery::new(
self.home_dir.clone(),
self.current_dir.clone(),
caller.parse().expect("caller"),
target,
TEST_TEAM.parse().expect("team"),
ReadSelection::All,
false,
Some(&message_id.to_string()),
None,
None,
None,
None,
None,
)
.expect("peek query")
}
fn clear_query(&self) -> ClearQuery {
ClearQuery {
home_dir: self.home_dir.clone(),
current_dir: self.current_dir.clone(),
caller_identity: TEST_RECIPIENT.parse().expect("caller"),
caller_team: TEST_TEAM.parse().expect("team"),
older_than: None,
idle_only: false,
dry_run: false,
}
}
fn doctor_query(&self) -> DoctorQuery {
DoctorQuery {
home_dir: self.home_dir.clone(),
current_dir: self.current_dir.clone(),
team_override: Some(TEST_TEAM.parse().expect("team")),
}
}
fn message(&self, text: &str, read: bool) -> InboxMessage {
InboxMessage {
from: TEST_LEAD.parse().expect("lead"),
text: text.to_string(),
timestamp: Utc::now().into(),
read,
source_team: Some(TEST_TEAM.parse().expect("team")),
summary: None,
message_id: Some(AtmMessageId::new()),
requires_ack: false,
pending_ack_at: None,
acknowledged_at: None,
acknowledges_message_id: None,
parent_message_id: None,
thread_mode: None,
expires_at: None,
task_id: None,
extra: serde_json::Map::new(),
}
}
fn pending_ack_message(&self, text: &str) -> (AtmMessageId, InboxMessage) {
let message_id = AtmMessageId::new();
let mut message = self.message(text, true);
message.message_id = Some(message_id);
message.requires_ack = true;
message.pending_ack_at = Some(Utc::now().into());
(message_id, message)
}
}
#[test]
fn fake_transport_maps_protocol_error_envelope_to_atm_error() {
let tempdir = TempDir::new().expect("tempdir");
let observability = CliObservability::fallback();
let transport = Arc::new(FakeClientTransport::new(|_| {
Ok(ResponseEnvelope::Error(ProtocolErrorEnvelope::from_error(
&AtmError::daemon_unavailable("synthetic daemon failure")
.with_recovery("retry after the daemon is reachable"),
)))
}));
let composition = CliComposition::from_transport(transport, &observability);
let error = composition
.send_request(RequestEnvelope::Doctor(atm_core::doctor::DoctorQuery {
home_dir: tempdir.path().join("home"),
current_dir: tempdir.path().join("cwd"),
team_override: None,
}))
.expect_err("protocol error");
assert_eq!(
error.code,
atm_core::error_codes::AtmErrorCode::DaemonUnavailable
);
assert!(error.to_string().contains("synthetic daemon failure"));
let recovery = error.primary_recovery().expect("daemon recovery");
assert!(recovery.contains("atm-daemon binary is installed"));
assert!(recovery.contains("daemon socket path is reachable"));
assert!(recovery.contains("ATM_HOME are set correctly"));
}
#[test]
#[serial(env)]
fn loopback_transport_send_persists_inbox_without_daemon() {
let fixture = LoopbackFixture::new(TEST_RECIPIENT);
let transport_observability = Arc::new(atm_core::observability::NullObservability);
let composition_observability = CliObservability::fallback();
let composition = CliComposition::from_transport(
Arc::new(LoopbackClientTransport::new(transport_observability)),
&composition_observability,
);
let outcome = composition
.send(fixture.send_request("hello from loopback"))
.expect("send outcome");
assert_eq!(outcome.agent.as_str(), TEST_RECIPIENT);
assert_eq!(outcome.sender.as_str(), TEST_SENDER);
let inbox = fixture.inbox_contents(TEST_RECIPIENT);
assert_eq!(inbox.len(), 1);
assert_eq!(inbox[0].text, "hello from loopback");
assert_eq!(inbox[0].from.as_str(), TEST_SENDER);
}
#[test]
#[serial(env)]
fn loopback_transport_rejects_self_addressed_send_without_persisting_inbox() {
let fixture = LoopbackFixture::new(TEST_RECIPIENT);
let transport_observability = Arc::new(atm_core::observability::NullObservability);
let composition_observability = CliObservability::fallback();
let composition = CliComposition::from_transport(
Arc::new(LoopbackClientTransport::new(transport_observability)),
&composition_observability,
);
let self_address = format!("{TEST_SENDER}@{TEST_TEAM}");
let error = composition
.send(fixture.send_request_to(&self_address, "hello self"))
.expect_err("self-addressed send must fail");
assert_eq!(
error.code,
atm_core::error_codes::AtmErrorCode::SelfAddressedSendInvalid
);
assert!(fixture.inbox_contents(TEST_SENDER).is_empty());
}
#[test]
#[serial(env)]
fn loopback_transport_missing_config_notice_retains_at_most_one_team_lead_message_under_concurrency()
{
let fixture = LoopbackFixture::new(TEST_RECIPIENT);
fs::remove_file(fixture.team_dir().join("config.json")).expect("remove config");
fixture.write_inbox_values(TEST_RECIPIENT, &[]);
fixture.write_inbox_values(ROLE_TEAM_LEAD, &[]);
let transport = Arc::new(LoopbackClientTransport::new(Arc::new(
atm_core::observability::NullObservability,
)));
let (first, second) = std::thread::scope(|scope| {
let first_request = fixture.send_request("first");
let second_request = fixture.send_request("second");
let first_transport = transport.clone();
let second_transport = transport.clone();
let first = scope.spawn(move || {
first_transport.send(RequestEnvelope::Send(SendRequestEnvelope::Compose(
first_request,
)))
});
let second = scope.spawn(move || {
second_transport.send(RequestEnvelope::Send(SendRequestEnvelope::Compose(
second_request,
)))
});
(
first.join().expect("first transport result"),
second.join().expect("second transport result"),
)
});
for (label, result) in [("first", &first), ("second", &second)] {
if let Err(error) = result {
assert_eq!(
error.code,
atm_core::error_codes::AtmErrorCode::MailboxLockTimeout,
"{label} response: {result:?}"
);
}
}
let notices = fixture.inbox_contents(ROLE_TEAM_LEAD);
assert!(
notices.len() <= 1,
"loopback missing-config fallback should retain at most one notice; got {}",
notices.len()
);
}
#[test]
#[serial(env)]
fn loopback_transport_send_preserves_ack_and_task_metadata_without_daemon() {
let fixture = LoopbackFixture::new(TEST_RECIPIENT);
let composition_observability = CliObservability::fallback();
let composition = CliComposition::from_transport(
Arc::new(LoopbackClientTransport::new(Arc::new(
atm_core::observability::NullObservability,
))),
&composition_observability,
);
let outcome = composition
.send(fixture.send_request_with_flags(
"needs acknowledgement",
false,
Some("TASK-314".parse().expect("task id")),
))
.expect("send outcome");
assert!(outcome.requires_ack);
assert_eq!(
outcome.task_id.as_ref().map(|value| value.as_str()),
Some("TASK-314")
);
let inbox = fixture.inbox_contents(TEST_RECIPIENT);
assert_eq!(inbox.len(), 1);
assert_eq!(
inbox[0].task_id.as_ref().map(|value| value.as_str()),
Some("TASK-314")
);
assert!(inbox[0].pending_ack_at.is_some());
}
#[test]
#[serial(env)]
fn loopback_transport_phase_ad_messaging_regression_matrix_without_daemon() {
let fixture = LoopbackFixture::new(TEST_RECIPIENT);
let composition_observability = CliObservability::fallback();
let composition = CliComposition::from_transport(
Arc::new(LoopbackClientTransport::new(Arc::new(
atm_core::observability::NullObservability,
))),
&composition_observability,
);
let plain_outcome = composition
.send(fixture.send_request("plain informational"))
.expect("plain send outcome");
assert!(!plain_outcome.requires_ack);
let plain_message_id = plain_outcome.message_id;
let ack_required_outcome = composition
.send(fixture.send_request_with_flags("needs acknowledgement", true, None))
.expect("ack-required send outcome");
assert!(ack_required_outcome.requires_ack);
let ack_required_message_id = ack_required_outcome.message_id;
let task_outcome = composition
.send(fixture.send_request_with_flags(
"task payload",
false,
Some("TASK-314".parse().expect("task id")),
))
.expect("task send outcome");
assert!(task_outcome.requires_ack);
let task_message_id = task_outcome.message_id;
let peek_outcome = composition
.peek(fixture.peek_query_for(TEST_RECIPIENT, None, plain_message_id))
.expect("peek outcome");
assert!(!peek_outcome.mutation_applied);
assert_eq!(peek_outcome.selected_message_id, Some(plain_message_id));
assert_eq!(
peek_outcome
.message
.as_ref()
.map(|message| message.envelope.read),
Some(false)
);
let inbox_after_peek = fixture.inbox_contents(TEST_RECIPIENT);
let plain_after_peek = inbox_after_peek
.iter()
.find(|message| message.message_id == Some(plain_message_id))
.expect("plain inbox message after peek");
assert!(!plain_after_peek.read);
assert!(plain_after_peek.pending_ack_at.is_none());
assert!(plain_after_peek.acknowledged_at.is_none());
let cross_agent_peek = composition
.peek(fixture.peek_query_for(
TEST_SENDER,
Some(TEST_RECIPIENT_ADDRESS),
plain_message_id,
))
.expect("cross-agent peek outcome");
assert!(!cross_agent_peek.mutation_applied);
assert_eq!(cross_agent_peek.selected_message_id, Some(plain_message_id));
assert_eq!(
cross_agent_peek
.message
.as_ref()
.map(|message| message.envelope.read),
Some(false)
);
let inbox_after_cross_agent_peek = fixture.inbox_contents(TEST_RECIPIENT);
let plain_after_cross_agent_peek = inbox_after_cross_agent_peek
.iter()
.find(|message| message.message_id == Some(plain_message_id))
.expect("plain inbox message after cross-agent peek");
assert!(!plain_after_cross_agent_peek.read);
assert!(plain_after_cross_agent_peek.pending_ack_at.is_none());
assert!(plain_after_cross_agent_peek.acknowledged_at.is_none());
let read_outcome = composition
.receive(fixture.read_query_for(TEST_RECIPIENT, plain_message_id))
.expect("read outcome");
assert!(read_outcome.mutation_applied);
assert_eq!(read_outcome.selected_message_id, Some(plain_message_id));
assert_eq!(
read_outcome
.message
.as_ref()
.map(|message| message.envelope.read),
Some(true)
);
assert_eq!(
read_outcome
.message
.as_ref()
.and_then(|message| message.envelope.pending_ack_at),
None
);
let inbox_after_read = fixture.inbox_contents(TEST_RECIPIENT);
let plain_after_read = inbox_after_read
.iter()
.find(|message| message.message_id == Some(plain_message_id))
.expect("plain inbox message after read");
assert!(plain_after_read.read);
assert!(plain_after_read.pending_ack_at.is_none());
let ack_required_after_send = inbox_after_read
.iter()
.find(|message| message.message_id == Some(ack_required_message_id))
.expect("ack-required inbox message");
assert!(ack_required_after_send.pending_ack_at.is_some());
let task_after_send = inbox_after_read
.iter()
.find(|message| message.message_id == Some(task_message_id))
.expect("task inbox message");
assert!(task_after_send.pending_ack_at.is_some());
assert_eq!(
task_after_send.task_id.as_ref().map(|value| value.as_str()),
Some("TASK-314")
);
let self_address = format!("{TEST_SENDER}@{TEST_TEAM}");
let error = composition
.send(fixture.send_request_to(&self_address, "hello self"))
.expect_err("self-addressed send must fail");
assert_eq!(
error.code,
atm_core::error_codes::AtmErrorCode::SelfAddressedSendInvalid
);
}
#[test]
#[serial(env)]
fn loopback_transport_read_surfaces_messages_without_daemon() {
let fixture = LoopbackFixture::new(TEST_RECIPIENT);
fixture.write_inbox_messages(TEST_RECIPIENT, &[fixture.message("read me", false)]);
let composition_observability = CliObservability::fallback();
let composition = CliComposition::from_transport(
Arc::new(LoopbackClientTransport::new(Arc::new(
atm_core::observability::NullObservability,
))),
&composition_observability,
);
let outcome = composition
.receive(fixture.read_query())
.expect("read outcome");
assert_eq!(outcome.agent.as_str(), TEST_RECIPIENT);
assert_eq!(outcome.count, 1);
assert_eq!(
outcome.message.expect("selected message").envelope.text,
"read me"
);
}
#[test]
#[serial(env)]
fn loopback_transport_read_rejects_cross_agent_target_without_daemon() {
let fixture = LoopbackFixture::new(TEST_RECIPIENT);
let composition_observability = CliObservability::fallback();
let composition = CliComposition::from_transport(
Arc::new(LoopbackClientTransport::new(Arc::new(
atm_core::observability::NullObservability,
))),
&composition_observability,
);
let error = composition
.receive(
ReadQuery::new(
fixture.home_dir.clone(),
fixture.current_dir.clone(),
TEST_SENDER.parse().expect("caller"),
Some(TEST_RECIPIENT_ADDRESS),
TEST_TEAM.parse().expect("team"),
ReadSelection::All,
false,
false,
None,
None,
None,
None,
None,
None,
)
.expect("query"),
)
.expect_err("cross-agent loopback read must fail");
assert!(error.is_validation(), "{error:?}");
assert!(error.message.contains("owner-only `atm read`"), "{error:?}");
}
#[test]
#[serial(env)]
fn loopback_transport_clear_removes_read_messages_without_daemon() {
let fixture = LoopbackFixture::new(TEST_RECIPIENT);
fixture.write_inbox_messages(TEST_RECIPIENT, &[fixture.message("done", true)]);
let composition_observability = CliObservability::fallback();
let composition = CliComposition::from_transport(
Arc::new(LoopbackClientTransport::new(Arc::new(
atm_core::observability::NullObservability,
))),
&composition_observability,
);
let outcome = composition
.clear(fixture.clear_query())
.expect("clear outcome");
assert_eq!(outcome.removed_total, 1);
assert_eq!(outcome.remaining_total, 0);
let read_outcome = composition
.receive(fixture.read_query())
.expect("read outcome after clear");
assert_eq!(read_outcome.count, 0);
}
#[test]
#[serial(env)]
fn loopback_transport_doctor_reports_health_without_daemon() {
let fixture = LoopbackFixture::new(TEST_RECIPIENT);
let observability = Arc::new(HealthyObservability);
let composition_observability = CliObservability::fallback();
let composition = CliComposition::from_transport(
Arc::new(LoopbackClientTransport::new(observability)),
&composition_observability,
);
let report = composition
.doctor(fixture.doctor_query())
.expect("doctor report");
assert_eq!(report.summary.status, DoctorStatus::Healthy);
assert_eq!(report.summary.error_count, 0);
}
#[test]
#[serial(env)]
fn doctor_projects_bootstrap_trace_into_report() {
let fixture = LoopbackFixture::new(TEST_RECIPIENT);
let observability = Arc::new(HealthyObservability);
let composition_observability = CliObservability::fallback();
let composition = CliComposition::from_transport_with_bootstrap_trace(
Arc::new(LoopbackClientTransport::new(observability)),
&composition_observability,
BootstrapTraceReport {
daemon_connect: BootstrapConnectOutcome::Connected,
daemon_launch_gate: BootstrapLaunchGateOutcome::Skipped,
daemon_auto_start: BootstrapAutoStartOutcome::Skipped,
connect_detail: None,
launch_gate_detail: None,
auto_start_detail: None,
},
);
let report = composition
.doctor(fixture.doctor_query())
.expect("doctor report");
assert_eq!(
report.bootstrap_trace.as_ref().expect("bootstrap trace"),
&BootstrapTraceReport {
daemon_connect: BootstrapConnectOutcome::Connected,
daemon_launch_gate: BootstrapLaunchGateOutcome::Skipped,
daemon_auto_start: BootstrapAutoStartOutcome::Skipped,
connect_detail: None,
launch_gate_detail: None,
auto_start_detail: None,
}
);
}
#[test]
fn bootstrap_propagates_daemon_availability_failure() {
let tempdir = TempDir::new().expect("tempdir");
let supervisor = DaemonSupervisor::new(
DaemonLocalIpcEndpoint::new(tempdir.path().join("daemon.sock"))
.expect("daemon endpoint"),
DaemonBinaryPath::new(tempdir.path().join("missing-atm-daemon"))
.expect("daemon binary path"),
);
let error = supervisor
.ensure_daemon_available_with_lock_path(
|| Err(AtmError::daemon_unavailable("daemon not running for test")),
Duration::from_millis(10),
Duration::from_millis(1),
tempdir.path().join(HOST_RUNTIME_LAUNCH_LOCK_FILE),
)
.expect_err("bootstrap should fail when daemon auto-start cannot launch");
assert_eq!(
error.code,
atm_core::error_codes::AtmErrorCode::DaemonUnavailable
);
assert!(error.to_string().contains("daemon binary is missing"));
}
#[test]
#[serial(env)]
fn loopback_transport_ack_appends_reply_without_daemon() {
let fixture = LoopbackFixture::new(TEST_RECIPIENT);
let (message_id, pending_ack) = fixture.pending_ack_message("please ack");
fixture.write_inbox_messages(TEST_SENDER, &[pending_ack]);
let composition_observability = CliObservability::fallback();
let composition = CliComposition::from_transport(
Arc::new(LoopbackClientTransport::new(Arc::new(
atm_core::observability::NullObservability,
))),
&composition_observability,
);
let outcome = composition
.ack(fixture.ack_request(message_id, "received and starting"))
.expect("ack outcome");
assert_eq!(outcome.team.as_str(), TEST_TEAM);
assert_eq!(outcome.agent.as_str(), TEST_SENDER);
assert_eq!(outcome.message_id, message_id);
assert!(matches!(
&outcome.reply_disposition,
atm_core::ack::AckReplyDisposition::Sent { reply_target, .. }
if reply_target.to_string() == format!("{TEST_LEAD}@{TEST_TEAM}")
));
let sender_inbox = fixture.inbox_contents(TEST_SENDER);
assert_eq!(sender_inbox.len(), 1);
assert!(sender_inbox[0].pending_ack_at.is_none());
assert!(sender_inbox[0].acknowledged_at.is_some());
let replies = fixture.inbox_contents(TEST_LEAD);
assert_eq!(replies.len(), 1);
assert_eq!(replies[0].text, "received and starting");
assert_eq!(replies[0].acknowledges_message_id, Some(message_id));
assert!(replies[0].pending_ack_at.is_none());
}
#[test]
#[serial(env)]
fn loopback_transport_ack_historical_self_poison_suppresses_replacement_reply() {
let fixture = LoopbackFixture::new(TEST_RECIPIENT);
let (message_id, mut pending_ack) = fixture.pending_ack_message("historical self poison");
pending_ack.from = TEST_SENDER.parse().expect("self sender");
fixture.write_inbox_messages(TEST_SENDER, &[pending_ack]);
let composition_observability = CliObservability::fallback();
let composition = CliComposition::from_transport(
Arc::new(LoopbackClientTransport::new(Arc::new(
atm_core::observability::NullObservability,
))),
&composition_observability,
);
let outcome = composition
.ack(fixture.ack_request(message_id, "resolved"))
.expect("self poison ack outcome");
assert!(matches!(
outcome.reply_disposition,
atm_core::ack::AckReplyDisposition::SuppressedSelfAck
));
let sender_inbox = fixture.inbox_contents(TEST_SENDER);
assert_eq!(sender_inbox.len(), 1);
assert_eq!(sender_inbox[0].message_id, Some(message_id));
assert!(sender_inbox[0].pending_ack_at.is_none());
assert!(sender_inbox[0].acknowledged_at.is_some());
assert!(fixture.inbox_contents(TEST_LEAD).is_empty());
}
#[test]
#[serial(env)]
fn cli_composition_supports_graft_client_surface_without_daemon() {
let fixture = LoopbackFixture::new(TEST_RECIPIENT);
let composition_observability = CliObservability::fallback();
let composition = CliComposition::from_transport(
Arc::new(LoopbackClientTransport::new(Arc::new(
atm_core::observability::NullObservability,
))),
&composition_observability,
);
let client: &dyn AtmGraftClient = &composition;
let send_outcome = client
.send_message(fixture.send_request("graft send"))
.expect("send through graft client surface");
assert_eq!(send_outcome.sender.as_str(), TEST_SENDER);
let read_outcome = client
.read_message(fixture.read_query())
.expect("read through graft client surface");
assert_eq!(read_outcome.count, 1);
let (message_id, pending_ack) = fixture.pending_ack_message("please ack");
fixture.write_inbox_messages(TEST_SENDER, &[pending_ack]);
let ack_outcome = client
.acknowledge_message(fixture.ack_request(message_id, "received and starting"))
.expect("ack through graft client surface");
assert_eq!(ack_outcome.message_id, message_id);
}
#[test]
fn daemon_path_newtypes_reject_empty_paths_and_preserve_path_access() {
let tempdir = TempDir::new().expect("tempdir");
let socket_path = tempdir.path().join("atm.sock");
let daemon_path = tempdir.path().join("atm-daemon");
let socket = DaemonLocalIpcEndpoint::new(socket_path.clone()).expect("socket path");
let daemon = DaemonBinaryPath::new(daemon_path.clone()).expect("daemon path");
assert_eq!(socket.as_ref(), socket_path.as_path());
assert_eq!(daemon.as_ref(), daemon_path.as_path());
let socket_error =
DaemonLocalIpcEndpoint::new(std::path::PathBuf::new()).expect_err("empty");
assert!(
socket_error
.to_string()
.contains("daemon local IPC endpoint")
);
let daemon_error = DaemonBinaryPath::new(std::path::PathBuf::new()).expect_err("empty");
assert!(daemon_error.to_string().contains("daemon binary path"));
}
#[test]
fn launch_gate_is_host_wide_across_different_socket_paths() {
let tempdir = TempDir::new().expect("tempdir");
let runtime_dir = atm_core::home::host_runtime_dir_from_home(tempdir.path());
let first =
LaunchGateGuard::try_acquire_at(runtime_dir.join(HOST_RUNTIME_LAUNCH_LOCK_FILE))
.expect("first acquire");
assert!(first.is_some());
let second =
LaunchGateGuard::try_acquire_at(runtime_dir.join(HOST_RUNTIME_LAUNCH_LOCK_FILE))
.expect("second acquire");
assert!(second.is_none());
drop(first);
}
#[test]
fn launch_gate_isolation_tracks_the_explicit_atm_home_root() {
let tempdir = TempDir::new().expect("tempdir");
let first_atm_home = tempdir.path().join("workspace-a");
let second_atm_home = tempdir.path().join("workspace-b");
let first_runtime_dir = atm_core::home::host_runtime_dir_from_home(&first_atm_home);
let second_runtime_dir = atm_core::home::host_runtime_dir_from_home(&second_atm_home);
let first =
LaunchGateGuard::try_acquire_at(first_runtime_dir.join(HOST_RUNTIME_LAUNCH_LOCK_FILE))
.expect("first acquire");
assert!(first.is_some());
let second =
LaunchGateGuard::try_acquire_at(second_runtime_dir.join(HOST_RUNTIME_LAUNCH_LOCK_FILE))
.expect("second acquire");
assert!(
second.is_some(),
"different ATM_HOME roots must not share one launch gate"
);
drop(first);
drop(second);
}
#[test]
fn host_runtime_lock_path_follows_the_explicit_home_root() {
let tempdir = TempDir::new().expect("tempdir");
let path = atm_core::home::host_runtime_lock_path_from_home(
tempdir.path(),
HOST_RUNTIME_LAUNCH_LOCK_FILE,
);
assert_eq!(
path,
tempdir
.path()
.join(".atm")
.join("daemon")
.join("launch.lock")
);
}
#[test]
#[serial(env)]
fn resolve_command_runtime_context_reuses_atm_home_across_sibling_worktrees() {
let tempdir = TempDir::new().expect("tempdir");
let atm_home = tempdir.path().join("atm-home");
let workspace_a = tempdir.path().join("workspace-a");
let workspace_b = tempdir.path().join("workspace-b");
std::fs::create_dir_all(&atm_home).expect("atm home");
std::fs::create_dir_all(&workspace_a).expect("workspace a");
std::fs::create_dir_all(&workspace_b).expect("workspace b");
let _env = EnvGuard::set_many([
("ATM_HOME", Some(atm_home.to_str().expect("utf8 atm home"))),
("HOME", Some(tempdir.path().to_str().expect("utf8 home"))),
("USERPROFILE", None),
]);
let (home_a, invocation_a) = {
let _cwd = CwdGuard::change_to(&workspace_a);
resolve_command_runtime_context("send").expect("workspace a context")
};
let (home_b, invocation_b) = {
let _cwd = CwdGuard::change_to(&workspace_b);
resolve_command_runtime_context("send").expect("workspace b context")
};
let canonical_atm_home = std::fs::canonicalize(&atm_home).expect("canonical atm home");
let canonical_workspace_a =
std::fs::canonicalize(&workspace_a).expect("canonical workspace a");
let canonical_workspace_b =
std::fs::canonicalize(&workspace_b).expect("canonical workspace b");
assert_eq!(
std::fs::canonicalize(&home_a).expect("canonical home a"),
canonical_atm_home
);
assert_eq!(
std::fs::canonicalize(&home_b).expect("canonical home b"),
canonical_atm_home
);
assert_eq!(
std::fs::canonicalize(&invocation_a).expect("canonical invocation a"),
canonical_workspace_a
);
assert_eq!(
std::fs::canonicalize(&invocation_b).expect("canonical invocation b"),
canonical_workspace_b
);
assert_eq!(
atm_daemon_client::resolve_daemon_local_ipc_endpoint_from_home(&home_a)
.expect("workspace a daemon endpoint")
.as_ref(),
atm_daemon_client::resolve_daemon_local_ipc_endpoint_from_home(&home_b)
.expect("workspace b daemon endpoint")
.as_ref()
);
}
#[test]
#[serial(env)]
fn resolve_command_runtime_context_reports_atm_home_unresolved_and_logs() {
let tempdir = TempDir::new().expect("tempdir");
let workspace = tempdir.path().join("workspace");
std::fs::create_dir_all(&workspace).expect("workspace");
let _env = EnvGuard::set_many([("ATM_HOME", None), ("HOME", None), ("USERPROFILE", None)]);
let (result, logs) = {
let _cwd = CwdGuard::change_to(&workspace);
capture_runtime_root_logs(|| resolve_command_runtime_context("send"))
};
let error = result.expect_err("missing ATM_HOME/home should fail");
assert_eq!(
error.code,
atm_core::error_codes::AtmErrorCode::AtmHomeUnresolved
);
assert!(logs.contains("raw cli runtime-root failure"));
assert!(logs.contains("ATM_HOME_UNRESOLVED"));
}
#[cfg(unix)]
#[test]
#[serial(env)]
fn bootstrap_reports_runtime_root_invalid_for_non_utf8_socket_override_and_logs() {
let tempdir = TempDir::new().expect("tempdir");
let invocation_dir = tempdir.path().join("workspace");
let invalid_atm_home = std::path::PathBuf::from(OsString::from_vec(vec![0xff, 0xfe, b'a']));
std::fs::create_dir_all(&invocation_dir).expect("invocation dir");
let observability = CliObservability::fallback();
let (result, logs) = capture_runtime_root_logs(|| {
CliComposition::bootstrap(
"send",
&observability,
InvocationDir::new(&invocation_dir),
AtmHomePath::new(&invalid_atm_home),
)
});
let error = result.expect_err("non-utf8 ATM_HOME should fail bootstrap");
assert_eq!(
error.code,
atm_core::error_codes::AtmErrorCode::RuntimeRootInvalid
);
assert!(logs.contains("raw cli runtime-root failure"));
assert!(logs.contains("ATM_RUNTIME_ROOT_INVALID"));
}
#[test]
#[serial(env)]
fn bootstrap_refuses_conflicting_daemon_socket_override() {
let tempdir = TempDir::new().expect("tempdir");
let atm_home = tempdir.path().join("atm-home");
let invocation_dir = tempdir.path().join("workspace");
std::fs::create_dir_all(&atm_home).expect("atm home");
std::fs::create_dir_all(&invocation_dir).expect("invocation dir");
let _env = EnvGuard::set_many([
("ATM_HOME", Some(atm_home.to_str().expect("utf8 atm home"))),
(
"ATM_DAEMON_SOCKET",
Some(
tempdir
.path()
.join("other.sock")
.to_str()
.expect("utf8 socket"),
),
),
("ATM_IDENTITY", Some(TEST_SENDER)),
("ATM_TEAM", Some(TEST_TEAM)),
]);
let observability = CliObservability::fallback();
let (result, logs) = capture_runtime_root_logs(|| {
CliComposition::bootstrap(
"send",
&observability,
InvocationDir::new(&invocation_dir),
AtmHomePath::new(&atm_home),
)
});
let error = result.expect_err("conflicting daemon socket override should fail");
assert_eq!(
error.code,
atm_core::error_codes::AtmErrorCode::RuntimeBootstrapRefused
);
assert!(logs.contains("raw cli runtime-root failure"));
assert!(logs.contains("ATM_RUNTIME_BOOTSTRAP_REFUSED"));
}
#[test]
#[serial(env)]
fn resolve_command_runtime_context_reports_atm_home_unresolved() {
let _env = EnvGuard::set_many([("ATM_HOME", None), ("HOME", None), ("USERPROFILE", None)]);
let error =
resolve_command_runtime_context("send").expect_err("missing ATM_HOME should fail");
assert_eq!(
error.code,
atm_core::error_codes::AtmErrorCode::AtmHomeUnresolved
);
}
#[test]
#[cfg(unix)]
#[serial(env)]
fn bootstrap_reports_runtime_root_invalid_for_invalid_socket_override() {
use std::ffi::OsString;
use std::os::unix::ffi::OsStringExt;
struct SocketEnvRestore(Option<OsString>);
impl Drop for SocketEnvRestore {
fn drop(&mut self) {
match self.0.take() {
Some(value) => {
unsafe { std::env::set_var("ATM_DAEMON_SOCKET", value) };
}
None => {
unsafe { std::env::remove_var("ATM_DAEMON_SOCKET") };
}
}
}
}
let tempdir = TempDir::new().expect("tempdir");
let atm_home = tempdir.path().join("atm-home");
let invocation_dir = tempdir.path().join("workspace");
std::fs::create_dir_all(&atm_home).expect("atm home");
std::fs::create_dir_all(&invocation_dir).expect("invocation dir");
let _env = EnvGuard::set_many([
("ATM_HOME", Some(atm_home.to_str().expect("utf8 atm home"))),
("ATM_IDENTITY", Some(TEST_SENDER)),
("ATM_TEAM", Some(TEST_TEAM)),
]);
let _socket_restore = SocketEnvRestore(std::env::var_os("ATM_DAEMON_SOCKET"));
unsafe {
std::env::set_var(
"ATM_DAEMON_SOCKET",
OsString::from_vec(vec![0x66, 0x6f, 0x80]),
);
}
let error = CliComposition::bootstrap(
"send",
&CliObservability::fallback(),
InvocationDir::new(&invocation_dir),
AtmHomePath::new(&atm_home),
)
.expect_err("invalid daemon socket override should fail");
assert_eq!(
error.code,
atm_core::error_codes::AtmErrorCode::RuntimeRootInvalid
);
}
#[test]
fn launch_gate_busy_maps_to_typed_rejection() {
let tempdir = TempDir::new().expect("tempdir");
let runtime_dir = atm_core::home::host_runtime_dir_from_home(tempdir.path());
let _gate =
LaunchGateGuard::try_acquire_at(runtime_dir.join(HOST_RUNTIME_LAUNCH_LOCK_FILE))
.expect("acquire")
.expect("gate");
let socket_path =
DaemonLocalIpcEndpoint::new(tempdir.path().join("one.sock")).expect("socket");
let second =
LaunchGateGuard::try_acquire_at(runtime_dir.join(HOST_RUNTIME_LAUNCH_LOCK_FILE))
.expect("second acquire");
assert!(second.is_none());
let error = LaunchGateGuard::rejected_error(&socket_path);
assert_eq!(
error.code,
atm_core::error_codes::AtmErrorCode::DaemonLaunchGateRejected
);
}
#[test]
fn gate_timeout_maps_to_launch_gate_rejected() {
let tempdir = TempDir::new().expect("tempdir");
let runtime_dir = atm_core::home::host_runtime_dir_from_home(tempdir.path());
let launch_lock_path = runtime_dir.join(HOST_RUNTIME_LAUNCH_LOCK_FILE);
let _gate = LaunchGateGuard::try_acquire_at(launch_lock_path.clone())
.expect("acquire")
.expect("gate");
let socket_path =
DaemonLocalIpcEndpoint::new(tempdir.path().join("missing.sock")).expect("socket");
let daemon_bin = DaemonBinaryPath::new(tempdir.path().join("atm-daemon")).expect("daemon");
let supervisor = DaemonSupervisor::new(socket_path.clone(), daemon_bin);
let transport = LocalIpcClientTransportAdapter::new(socket_path);
let error = supervisor
.ensure_daemon_available_with_lock_path(
|| transport.probe_connection().map(|_| ()),
Duration::from_millis(0),
Duration::from_millis(0),
launch_lock_path,
)
.expect_err("timeout should fail");
assert_eq!(
error.code,
atm_core::error_codes::AtmErrorCode::DaemonLaunchGateRejected
);
}
#[cfg(windows)]
#[test]
fn launch_gate_treats_windows_lock_and_sharing_violations_as_contention() {
assert!(atm_daemon_client::is_launch_gate_contention_error(
&std::io::Error::from_raw_os_error(32)
));
assert!(atm_daemon_client::is_launch_gate_contention_error(
&std::io::Error::from_raw_os_error(33)
));
}
#[cfg(unix)]
#[test]
fn spawn_failure_maps_to_auto_start_failed() {
use std::fs;
let tempdir = TempDir::new().expect("tempdir");
let runtime_dir = atm_core::home::host_runtime_dir_from_home(tempdir.path());
let launch_lock_path = runtime_dir.join(HOST_RUNTIME_LAUNCH_LOCK_FILE);
let socket_path =
DaemonLocalIpcEndpoint::new(tempdir.path().join("missing.sock")).expect("socket");
let daemon_path = tempdir.path().join(if cfg!(windows) {
"invalid-atm-daemon.exe"
} else {
"invalid-atm-daemon"
});
fs::write(&daemon_path, b"not an executable daemon binary").expect("write daemon");
let daemon_bin = DaemonBinaryPath::new(daemon_path).expect("daemon");
let supervisor = DaemonSupervisor::new(socket_path.clone(), daemon_bin);
let transport = LocalIpcClientTransportAdapter::new(socket_path);
let error = supervisor
.ensure_daemon_available_with_lock_path(
|| transport.probe_connection().map(|_| ()),
Duration::from_millis(10),
Duration::from_millis(0),
launch_lock_path,
)
.expect_err("spawn should fail");
assert_eq!(
error.code,
atm_core::error_codes::AtmErrorCode::DaemonAutoStartFailed
);
}
#[cfg(unix)]
#[test]
fn daemon_path_newtypes_reject_non_utf8_paths_at_boundary() {
use std::os::unix::ffi::OsStringExt;
let invalid = std::ffi::OsString::from_vec(vec![0x66, 0x6f, 0x80]);
let socket_error = DaemonLocalIpcEndpoint::new(std::path::PathBuf::from(invalid.clone()))
.expect_err("utf8");
assert!(socket_error.to_string().contains("valid UTF-8"));
let daemon_error =
DaemonBinaryPath::new(std::path::PathBuf::from(invalid)).expect_err("utf8");
assert!(daemon_error.to_string().contains("valid UTF-8"));
}
}