#[path = "admission_tests.rs"]
mod admission_tests;
#[path = "auth_tests.rs"]
mod auth_tests;
#[path = "handoff_tests.rs"]
mod handoff_tests;
use super::*;
use crate::service::{
protocol::{ServiceEvent, TurnTerminalStatus},
turns::{ServiceTurnSink, TurnWorkerMessage},
};
use crate::{
config::{EffectiveConfig, McPaths, ProviderCredential, Settings},
sessions::{SessionEventKind, SessionManager, TurnStatus, TurnStatusPayload},
skills::SkillDiscovery,
};
use crossbeam_channel::{Receiver, bounded};
use std::sync::atomic::{AtomicUsize, Ordering};
use tempfile::TempDir;
fn runtime(temp: &TempDir) -> Arc<ServiceRuntime> {
runtime_at_path(temp.path())
}
fn runtime_at_path(root: &std::path::Path) -> Arc<ServiceRuntime> {
let paths = McPaths::from_root_and_project_dir(root.join("state"), root.to_owned());
paths.ensure_runtime_dirs().unwrap();
Arc::new(ServiceRuntime {
config: EffectiveConfig {
provider: Some("synthetic".into()),
model: Some("synthetic".into()),
no_color: true,
file_autocomplete_respects_gitignore: true,
custom_providers: BTreeMap::new(),
thinking_level: crate::thinking::ThinkingLevel::Default,
auth: Some(ProviderCredential::NoAuth),
paths: paths.clone(),
},
settings: Settings::default(),
instructions: vec![],
skills: SkillDiscovery::default(),
session_manager: SessionManager::new(paths.sessions),
cwd: root.to_owned(),
mcp: None,
})
}
fn request(
coordinator: &Coordinator,
connection: &str,
method: &str,
session: Option<&str>,
control: Option<Control>,
operation: Option<&str>,
payload: Value,
) -> Request {
Request {
protocol_version: 2,
kind: "request".into(),
request_id: uuid::Uuid::new_v4().to_string(),
instance_id: (method != "initialize").then(|| coordinator.instance.clone()),
connection_id: (method != "initialize").then(|| connection.to_owned()),
session_id: session.map(str::to_owned),
operation_id: operation.map(str::to_owned),
control,
method: method.into(),
payload,
}
}
fn invoke(coordinator: &mut Coordinator, connection: &str, request: Request) -> Value {
coordinator
.submit(
connection,
&serde_json::to_vec(&request).unwrap(),
Instant::now(),
)
.unwrap();
run_until(coordinator, |coordinator| {
!coordinator.connections[connection].queue.is_empty()
});
let client = coordinator.connections.get_mut(connection).unwrap();
let message: Value = serde_json::from_str(&client.queue.pop_front().unwrap()).unwrap();
client.queued_bytes = client.queue.iter().map(String::len).sum();
client.requests.remove(&request.request_id);
message
}
fn connect(coordinator: &mut Coordinator) -> String {
let connection = coordinator.connect(Instant::now()).unwrap();
let init = request(
coordinator,
&connection,
"initialize",
None,
None,
None,
json!({"supported_protocol_versions":[2],"requested_capabilities":[]}),
);
assert!(invoke(coordinator, &connection, init)["error"].is_null());
connection
}
fn create(coordinator: &mut Coordinator, connection: &str) -> String {
let request = request(
coordinator,
connection,
"session.create",
None,
None,
Some(&uuid::Uuid::new_v4().to_string()),
json!({}),
);
invoke(coordinator, connection, request)["payload"]["session_id"]
.as_str()
.unwrap()
.to_owned()
}
fn claim(coordinator: &mut Coordinator, connection: &str, session: &str) -> Control {
let request = request(
coordinator,
connection,
"session.claim",
Some(session),
None,
Some(&uuid::Uuid::new_v4().to_string()),
json!({}),
);
let reply = invoke(coordinator, connection, request);
assert!(reply["error"].is_null(), "{reply}");
Control {
grant_id: reply["payload"]["grant_id"].as_str().unwrap().to_owned(),
generation: reply["payload"]["generation"].as_u64().unwrap(),
}
}
fn run_until(coordinator: &mut Coordinator, ready: impl Fn(&Coordinator) -> bool) {
let deadline = Instant::now() + Duration::from_secs(5);
while !ready(coordinator) {
assert!(Instant::now() < deadline, "coordinator did not settle");
coordinator.tick(Instant::now());
std::thread::sleep(Duration::from_millis(1));
}
}
struct ToolProvider {
calls: Arc<AtomicUsize>,
release: Receiver<()>,
}
impl crate::providers::Provider for ToolProvider {
fn stream_cancellable(
&self,
_: crate::providers::ProviderRequest,
_: &crate::cancellation::AgentCancellation,
on_event: &mut dyn FnMut(crate::providers::ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
use crate::providers::{ProviderEvent, ToolCall};
if self.calls.fetch_add(1, Ordering::SeqCst) == 0 {
self.release.recv().unwrap();
on_event(ProviderEvent::TextDelta("before tool".into()))?;
on_event(ProviderEvent::ToolCall(ToolCall {
id: "synthetic-read".into(),
name: "read".into(),
arguments: json!({"paths":["evidence.txt"],"offset":1,"limit":1}),
}))?;
} else {
on_event(ProviderEvent::TextDelta("after tool".into()))?;
}
on_event(ProviderEvent::Done)
}
}
fn install_tool_worker(
coordinator: &mut Coordinator,
release: Receiver<()>,
calls: Arc<AtomicUsize>,
cleanup: Receiver<()>,
) {
coordinator.execution.set_turn_worker(Arc::new(move |job| {
let provider = ToolProvider {
calls: Arc::clone(&calls),
release: release.clone(),
};
let tools = crate::tools::ToolRuntime::new(&job.runtime.cwd).unwrap();
let agent = crate::agent::AgentSession::new("synthetic", &[], &SkillDiscovery::default());
let mut sink = ServiceTurnSink::new(
job.sender.clone(),
job.request_id.clone(),
job.session.id().into(),
job.turn_id.clone(),
);
agent
.run_print_with_tools_streaming_output(
&provider,
&job.prompt,
Some(&tools),
Some(&job.session),
&job.runtime.cwd,
Some(&mut sink),
)
.unwrap();
let payload = TurnStatusPayload::new(TurnStatus::Complete, Some(sink.assistant_text()))
.into_value()
.unwrap();
crate::sessions::record_session_event(
Some(&job.session),
&job.runtime.cwd,
SessionEventKind::TurnStatus,
payload,
)
.unwrap();
let mut terminal = ServiceEvent::turn_terminal(
job.request_id,
job.session.id().into(),
job.turn_id.clone(),
TurnTerminalStatus::Completed,
sink.assistant_text().into(),
);
sink.finish(&mut terminal);
terminal.payload["persistence"] = json!("committed");
terminal.payload["error"] = Value::Null;
job.sender
.send(TurnWorkerMessage::Terminal {
turn_id: job.turn_id,
event: terminal,
})
.unwrap();
cleanup.recv().unwrap();
}));
}
#[cfg(unix)]
impl PersistentService {
pub(crate) fn synthetic_socket_fixture(
root: &std::path::Path,
release: Receiver<()>,
cleanup_wait: Receiver<()>,
calls: Arc<AtomicUsize>,
) -> Self {
std::fs::write(root.join("evidence.txt"), "synthetic evidence").unwrap();
let mut coordinator = Coordinator::new(runtime_at_path(root)).unwrap();
coordinator.unix_transport = true;
install_tool_worker(&mut coordinator, release, calls, cleanup_wait);
Self::spawn(coordinator).unwrap()
}
}
#[test]
fn zero_clients_execute_tools_once_and_hold_lease_through_cleanup_and_late_claim() {
let temp = TempDir::new().unwrap();
std::fs::write(temp.path().join("evidence.txt"), "synthetic evidence").unwrap();
let runtime = runtime(&temp);
let mut coordinator = Coordinator::new(Arc::clone(&runtime)).unwrap();
let (release, wait) = bounded(1);
let (cleanup, cleanup_wait) = bounded(1);
let calls = Arc::new(AtomicUsize::new(0));
install_tool_worker(&mut coordinator, wait, Arc::clone(&calls), cleanup_wait);
let first = connect(&mut coordinator);
let session = create(&mut coordinator, &first);
let grant = claim(&mut coordinator, &first, &session);
let start = request(
&coordinator,
&first,
"turn.start",
Some(&session),
Some(grant.clone()),
Some("synthetic-operation"),
json!({"prompt":"Read the synthetic fixture"}),
);
let accepted = invoke(&mut coordinator, &first, start.clone());
assert_eq!(accepted["payload"]["status"], "accepted");
coordinator.disconnect(&first);
release.send(()).unwrap();
run_until(&mut coordinator, |state| {
state.actors[&session].phase == "cleaning"
});
let disk = runtime.session_manager.open_existing(&session).unwrap();
assert!(disk.try_frontend_writer().unwrap().is_none());
assert_eq!(calls.load(Ordering::SeqCst), 2);
let second = connect(&mut coordinator);
let claim_request = request(
&coordinator,
&second,
"session.claim",
Some(&session),
None,
Some("late-claim"),
json!({}),
);
let claimed = invoke(&mut coordinator, &second, claim_request);
assert_eq!(claimed["payload"]["snapshot"]["phase"], "cleaning");
assert!(
claimed["payload"]["snapshot"]["turn"]["assistant_text"]
.as_str()
.unwrap()
.contains("after tool")
);
let mut old_intent = start;
old_intent.connection_id = Some(second.clone());
old_intent.request_id = "lost-acceptance".into();
let duplicate = invoke(&mut coordinator, &second, old_intent);
assert_eq!(duplicate["error"]["code"], "operation_already_known");
cleanup.send(()).unwrap();
run_until(&mut coordinator, |state| {
state.actors[&session].phase == "idle"
});
let events: Vec<Value> = coordinator.connections[&second]
.queue
.iter()
.map(|record| serde_json::from_str(record).unwrap())
.collect();
assert_eq!(
events
.iter()
.filter(|event| event["event"] == "turn.terminal")
.count(),
1
);
assert!(
disk.try_frontend_writer().unwrap().is_none(),
"idle controller retains lease"
);
let outcome = coordinator.operations.lookup(
&coordinator.instance,
&coordinator.instance,
"synthetic-operation",
);
assert_eq!(outcome["result"]["persistence"], "committed");
assert_eq!(outcome["state"], "terminal");
coordinator.disconnect(&second);
assert!(coordinator.actors.is_empty());
assert!(disk.try_frontend_writer().unwrap().is_some());
let replay = disk.frontend_replay(None, 32).unwrap();
let replay = serde_json::to_value(replay).unwrap().to_string();
assert!(replay.contains("after tool"));
assert_eq!(calls.load(Ordering::SeqCst), 2);
}
#[test]
fn stale_grants_do_not_start_or_cancel_after_disconnect_and_sessions_are_independent() {
let temp = TempDir::new().unwrap();
let mut coordinator = Coordinator::new(runtime(&temp)).unwrap();
let first = connect(&mut coordinator);
let second = connect(&mut coordinator);
let session = create(&mut coordinator, &first);
let old = claim(&mut coordinator, &first, &session);
let other_session = create(&mut coordinator, &second);
claim(&mut coordinator, &second, &other_session);
coordinator.disconnect(&first);
let current = claim(&mut coordinator, &second, &session);
assert!(current.generation > old.generation);
for (method, payload) in [
("turn.start", json!({"prompt":"never execute"})),
("turn.cancel", json!({"turn_id":"old-turn"})),
("session.detach", json!({})),
] {
let attempt = request(
&coordinator,
&second,
method,
Some(&session),
Some(old.clone()),
Some(&uuid::Uuid::new_v4().to_string()),
payload,
);
assert_eq!(
invoke(&mut coordinator, &second, attempt)["error"]["code"],
"stale_grant"
);
}
assert_eq!(coordinator.actors.len(), 2);
assert!(
coordinator
.actors
.values()
.all(|actor| actor.phase == "idle")
);
}
#[test]
fn bounded_outcomes_expire_without_reexecution_or_secret_retention() {
let temp = TempDir::new().unwrap();
let mut coordinator = Coordinator::new(runtime(&temp)).unwrap();
let connection = connect(&mut coordinator);
let now = Instant::now();
let mut operations = Operations::new();
let intent = request(
&coordinator,
&connection,
"turn.start",
Some("session"),
Some(Control {
grant_id: "grant".into(),
generation: 1,
}),
Some("operation"),
json!({"prompt":"synthetic private intent"}),
);
operations.reserve(&intent, now).unwrap();
assert_eq!(
operations.reserve(&intent, now),
Err(Code::OperationAlreadyKnown)
);
let mut mismatch = intent.clone();
mismatch.payload["prompt"] = json!("different");
assert_eq!(
operations.reserve(&mismatch, now),
Err(Code::OperationIdentityMismatch)
);
operations.update(
"operation",
"terminal",
json!({"status":"completed"}),
Value::Null,
now,
);
assert!(
!operations
.lookup(&coordinator.instance, &coordinator.instance, "operation")
.to_string()
.contains("synthetic private intent")
);
assert_eq!(
operations.reserve(&intent, now + Duration::from_secs(600)),
Err(Code::OperationExpired)
);
operations.expire(now + Duration::from_secs(1200));
assert_eq!(
operations.lookup(&coordinator.instance, &coordinator.instance, "operation")["state"],
"unknown"
);
for index in 0..256 {
let mut intent = intent.clone();
intent.operation_id = Some(format!("operation-{index}"));
operations.reserve(&intent, now).unwrap();
}
assert_eq!(
operations.reserve(&intent, now),
Err(Code::OperationCapacity)
);
operations.expire(now + Duration::from_secs(5000));
assert_eq!(operations.admitted(), 256, "unsettled work never expires");
}
#[test]
fn stalled_controller_is_revoked_without_cancelling_execution() {
let temp = TempDir::new().unwrap();
std::fs::write(temp.path().join("evidence.txt"), "synthetic evidence").unwrap();
let mut coordinator = Coordinator::new(runtime(&temp)).unwrap();
let (release, wait) = bounded(1);
let (cleanup, cleanup_wait) = bounded(1);
let calls = Arc::new(AtomicUsize::new(0));
install_tool_worker(&mut coordinator, wait, Arc::clone(&calls), cleanup_wait);
let connection = connect(&mut coordinator);
let session = create(&mut coordinator, &connection);
let grant = claim(&mut coordinator, &connection, &session);
let start = request(
&coordinator,
&connection,
"turn.start",
Some(&session),
Some(grant),
Some("stalled-turn"),
json!({"prompt":"Read synthetic evidence"}),
);
invoke(&mut coordinator, &connection, start);
for _ in coordinator.connections[&connection].queue.len()..64 {
assert!(coordinator.queue(&connection, json!({"event":"synthetic"}), false));
}
assert!(!coordinator.queue(&connection, json!({"event":"turn.activity"}), true));
assert!(coordinator.connections.contains_key(&connection));
assert!(!coordinator.queue(&connection, json!({"event":"turn.assistant_delta"}), false));
assert!(!coordinator.connections.contains_key(&connection));
assert_eq!(coordinator.actors[&session].phase, "running");
release.send(()).unwrap();
cleanup.send(()).unwrap();
run_until(&mut coordinator, |state| state.actors.is_empty());
assert_eq!(calls.load(Ordering::SeqCst), 2);
assert_eq!(
coordinator.terminals[&session].payload["status"],
"completed"
);
assert!(
coordinator.terminals[&session].payload["activity_dropped"]
.as_u64()
.unwrap()
> 0
);
}
#[test]
fn daemon_entrypoint_keeps_progress_independent_of_connections() {
let temp = TempDir::new().unwrap();
let service = PersistentService::spawn(Coordinator::new(runtime(&temp)).unwrap()).unwrap();
let first = service.connect().unwrap();
service.submit(&first, json!({"protocol_version":2,"kind":"request","request_id":"init","instance_id":null,
"connection_id":null,"session_id":null,"operation_id":null,"control":null,"method":"initialize",
"payload":{"supported_protocol_versions":[2],"requested_capabilities":[]}}).to_string().as_bytes()).unwrap();
let reply: Value =
serde_json::from_str(&service.next_record(&first).unwrap().unwrap()).unwrap();
assert_eq!(reply["payload"]["connection_id"], first);
service.response_written(&first, "init").unwrap();
service.disconnect(&first).unwrap();
let second = service.connect().unwrap();
assert_ne!(first, second);
assert!(service.next_record(&first).is_err());
service.disconnect(&second).unwrap();
}
#[test]
fn decoder_rejects_duplicate_intent_keys_and_missing_nullable_fields() {
assert!(wire::decode(br#"{"payload":{"prompt":"a","prompt":"b"}}"#).is_err());
assert!(wire::decode(br#"{"protocol_version":2,"kind":"request","request_id":"id","method":"initialize","payload":{}}"#).is_err());
}
#[test]
fn running_daemon_releases_unowned_completed_work_with_no_output_consumer() {
let temp = TempDir::new().unwrap();
std::fs::write(temp.path().join("evidence.txt"), "synthetic evidence").unwrap();
let runtime = runtime(&temp);
let mut coordinator = Coordinator::new(Arc::clone(&runtime)).unwrap();
let (release, wait) = bounded(1);
let (cleanup, cleanup_wait) = bounded(1);
let calls = Arc::new(AtomicUsize::new(0));
install_tool_worker(&mut coordinator, wait, Arc::clone(&calls), cleanup_wait);
let connection = connect(&mut coordinator);
let session = create(&mut coordinator, &connection);
let grant = claim(&mut coordinator, &connection, &session);
let start = request(
&coordinator,
&connection,
"turn.start",
Some(&session),
Some(grant),
Some("detached-turn"),
json!({"prompt":"Read synthetic data"}),
);
let service = PersistentService::spawn(coordinator).unwrap();
service
.submit(&connection, &serde_json::to_vec(&start).unwrap())
.unwrap();
let deadline = Instant::now() + Duration::from_secs(5);
while calls.load(Ordering::SeqCst) == 0 {
assert!(Instant::now() < deadline, "turn was not admitted");
std::thread::sleep(Duration::from_millis(1));
}
service.disconnect(&connection).unwrap();
release.send(()).unwrap();
cleanup.send(()).unwrap();
let disk = runtime.session_manager.open_existing(&session).unwrap();
let deadline = Instant::now() + Duration::from_secs(5);
loop {
if disk.try_frontend_writer().unwrap().is_some() {
break;
}
assert!(Instant::now() < deadline);
std::thread::sleep(Duration::from_millis(2));
}
assert_eq!(calls.load(Ordering::SeqCst), 2);
let replay = serde_json::to_value(disk.frontend_replay(None, 32).unwrap())
.unwrap()
.to_string();
assert!(replay.contains("after tool"));
drop(service);
}
#[test]
fn accepted_settings_commit_survives_disconnect_and_next_capture_reads_it() {
let temp = TempDir::new().unwrap();
let runtime = runtime(&temp);
let mut coordinator = Coordinator::new(Arc::clone(&runtime)).unwrap();
let connection = connect(&mut coordinator);
let mutation = request(
&coordinator,
&connection,
"config.set",
None,
None,
Some("settings-write"),
json!({"scope":"global","fast":true}),
);
coordinator
.submit(
&connection,
&serde_json::to_vec(&mutation).unwrap(),
Instant::now(),
)
.unwrap();
assert_eq!(
coordinator.operations.lookup(
&coordinator.instance,
&coordinator.instance,
"settings-write"
)["state"],
"accepted"
);
coordinator.disconnect(&connection);
run_until(&mut coordinator, |state| state.pending.is_empty());
let outcome = coordinator.operations.lookup(
&coordinator.instance,
&coordinator.instance,
"settings-write",
);
assert_eq!(outcome["state"], "terminal", "{outcome}");
assert_eq!(outcome["result"]["status"], "completed", "{outcome}");
assert!(!runtime.settings.fast.enabled);
assert!(
runtime
.capture_turn_settings()
.unwrap()
.settings
.fast
.enabled
);
assert!(
crate::config::read_settings(&runtime.config.paths)
.unwrap()
.fast
.enabled
);
}
#[test]
fn persistence_failure_and_cleanup_panic_release_only_after_worker_exit() {
let temp = TempDir::new().unwrap();
let runtime = runtime(&temp);
let mut coordinator = Coordinator::new(Arc::clone(&runtime)).unwrap();
let (release, wait) = bounded(1);
let (cleanup, cleanup_wait) = bounded(1);
coordinator.execution.set_turn_worker(Arc::new(move |job| {
wait.recv().unwrap();
std::fs::remove_file(job.session.path()).unwrap();
std::fs::create_dir(job.session.path()).unwrap();
let payload = TurnStatusPayload::new(TurnStatus::Complete, Some("unsaved accepted text"))
.into_value()
.unwrap();
assert!(
crate::sessions::record_session_event(
Some(&job.session),
&job.runtime.cwd,
SessionEventKind::TurnStatus,
payload
)
.is_err()
);
let mut event = ServiceEvent::turn_terminal(
job.request_id,
job.session.id().into(),
job.turn_id.clone(),
TurnTerminalStatus::Failed,
"unsaved accepted text".into(),
);
event.payload["persistence"] = json!("failed");
event.payload["error"] = wire::error(Code::PersistenceFailed);
event.payload["sequence"] = json!(1);
job.sender
.send(TurnWorkerMessage::Terminal {
turn_id: job.turn_id,
event,
})
.unwrap();
cleanup_wait.recv().unwrap();
panic!("synthetic cleanup failure after terminal persistence failed");
}));
let connection = connect(&mut coordinator);
let session = create(&mut coordinator, &connection);
let disk = runtime.session_manager.open_existing(&session).unwrap();
let grant = claim(&mut coordinator, &connection, &session);
let start = request(
&coordinator,
&connection,
"turn.start",
Some(&session),
Some(grant),
Some("failed-persistence"),
json!({"prompt":"synthetic"}),
);
invoke(&mut coordinator, &connection, start);
coordinator.disconnect(&connection);
release.send(()).unwrap();
run_until(&mut coordinator, |state| {
state.actors[&session].phase == "cleaning"
});
assert!(disk.try_frontend_writer().unwrap().is_none());
assert_eq!(
coordinator.operations.lookup(
&coordinator.instance,
&coordinator.instance,
"failed-persistence",
)["state"],
"accepted",
"failed persistence must still wait for worker cleanup"
);
cleanup.send(()).unwrap();
run_until(&mut coordinator, |state| state.actors.is_empty());
let terminal = &coordinator.terminals[&session].payload;
assert_eq!(terminal["assistant_text"], "unsaved accepted text");
assert_eq!(terminal["status"], "failed");
assert_eq!(terminal["persistence"], "failed");
assert_eq!(terminal["replay_required"], true);
let outcome = coordinator.operations.lookup(
&coordinator.instance,
&coordinator.instance,
"failed-persistence",
);
assert_eq!(outcome["error"], terminal["error"]);
assert_eq!(outcome["result"]["persistence"], "failed");
assert!(disk.try_frontend_writer().unwrap().is_some());
}
#[test]
fn only_initiating_connection_controls_login_and_disconnect_waits_for_cleanup() {
let temp = TempDir::new().unwrap();
let mut coordinator = Coordinator::new(runtime(&temp)).unwrap();
let (observed, observation) = bounded(1);
let (release, wait) = bounded(1);
coordinator.execution.set_login_worker(Arc::new(move |job| {
observed.send(Arc::clone(&job.cancel)).unwrap();
wait.recv().unwrap();
if job.cancel.load(Ordering::Relaxed) {
"cancelled"
} else {
"exchange_succeeded"
}
}));
let first = connect(&mut coordinator);
let other = connect(&mut coordinator);
let start = request(
&coordinator,
&first,
"auth.login.start",
None,
None,
Some("login-intent"),
json!({"provider_id":"openai-codex"}),
);
let accepted = invoke(&mut coordinator, &first, start);
assert!(accepted["error"].is_null(), "{accepted}");
let cancel_flag = observation.recv_timeout(Duration::from_secs(2)).unwrap();
let wrong_owner = request(
&coordinator,
&other,
"auth.login.cancel",
None,
None,
Some("foreign-stop"),
json!({"login_id":accepted["payload"]["login_id"]}),
);
assert_eq!(
invoke(&mut coordinator, &other, wrong_owner)["error"]["code"],
"unknown_login"
);
coordinator.disconnect(&other);
assert!(!cancel_flag.load(Ordering::Relaxed));
coordinator.disconnect(&first);
run_until(&mut coordinator, |_| cancel_flag.load(Ordering::Relaxed));
assert_eq!(
coordinator
.operations
.lookup(&coordinator.instance, &coordinator.instance, "login-intent")["state"],
"accepted"
);
release.send(()).unwrap();
run_until(&mut coordinator, |state| state.login.is_none());
let outcome =
coordinator
.operations
.lookup(&coordinator.instance, &coordinator.instance, "login-intent");
assert_eq!(outcome["result"]["status"], "cancelled");
assert_eq!(outcome["result"]["cleanup_complete"], true);
assert!(outcome["error"].is_null());
}
#[test]
fn reclaimed_stop_targets_one_turn_while_other_sessions_keep_running() {
let temp = TempDir::new().unwrap();
let mut coordinator = Coordinator::new(runtime(&temp)).unwrap();
let (observed, observation) = bounded(2);
let (release, wait) = bounded(2);
coordinator.execution.set_turn_worker(Arc::new(move |job| {
observed
.send((job.session.id().to_owned(), Arc::clone(&job.cancellation)))
.unwrap();
wait.recv().unwrap();
let status = if job.cancellation.load(Ordering::SeqCst) {
TurnTerminalStatus::Cancelled
} else {
TurnTerminalStatus::Completed
};
let durable = if status == TurnTerminalStatus::Cancelled {
TurnStatus::Cancelled
} else {
TurnStatus::Complete
};
let payload = TurnStatusPayload::new(durable, Some("synthetic terminal"))
.into_value()
.unwrap();
crate::sessions::record_session_event(
Some(&job.session),
&job.runtime.cwd,
SessionEventKind::TurnStatus,
payload,
)
.unwrap();
let mut event = ServiceEvent::turn_terminal(
job.request_id,
job.session.id().into(),
job.turn_id.clone(),
status,
"synthetic terminal".into(),
);
event.payload["sequence"] = json!(1);
event.payload["persistence"] = json!("committed");
event.payload["error"] = Value::Null;
job.sender
.send(TurnWorkerMessage::Terminal {
turn_id: job.turn_id,
event,
})
.unwrap();
}));
let first = connect(&mut coordinator);
let second = connect(&mut coordinator);
let session = create(&mut coordinator, &first);
let other_session = create(&mut coordinator, &second);
let first_grant = claim(&mut coordinator, &first, &session);
let second_grant = claim(&mut coordinator, &second, &other_session);
let mut start = request(
&coordinator,
&first,
"turn.start",
Some(&session),
Some(first_grant),
Some("first-turn"),
json!({"prompt":"first synthetic"}),
);
start.request_id = "same-request-id".into();
let first_reply = invoke(&mut coordinator, &first, start);
let mut other_start = request(
&coordinator,
&second,
"turn.start",
Some(&other_session),
Some(second_grant),
Some("second-turn"),
json!({"prompt":"second synthetic"}),
);
other_start.request_id = "same-request-id".into();
assert!(invoke(&mut coordinator, &second, other_start)["error"].is_null());
let flags: HashMap<_, _> = (0..2)
.map(|_| observation.recv_timeout(Duration::from_secs(2)).unwrap())
.collect();
coordinator.disconnect(&first);
assert!(flags.values().all(|flag| !flag.load(Ordering::SeqCst)));
coordinator
.connections
.get_mut(&second)
.unwrap()
.queue
.clear();
coordinator
.connections
.get_mut(&second)
.unwrap()
.queued_bytes = 0;
let new_grant = claim(&mut coordinator, &second, &session);
let stop = request(
&coordinator,
&second,
"turn.cancel",
Some(&session),
Some(new_grant),
Some("explicit-stop"),
json!({"turn_id":first_reply["payload"]["turn_id"]}),
);
assert_eq!(
invoke(&mut coordinator, &second, stop)["payload"]["status"],
"cancellation_requested"
);
assert!(flags[&session].load(Ordering::SeqCst));
assert!(!flags[&other_session].load(Ordering::SeqCst));
release.send(()).unwrap();
release.send(()).unwrap();
run_until(&mut coordinator, |state| {
state.actors.values().all(|actor| actor.phase == "idle")
});
assert_eq!(
coordinator.terminals[&session].payload["status"],
"cancelled"
);
assert_eq!(
coordinator.terminals[&other_session].payload["status"],
"completed"
);
}
#[test]
fn expired_handshakes_and_exhausted_generations_never_acquire_control() {
let temp = TempDir::new().unwrap();
let mut coordinator = Coordinator::new(runtime(&temp)).unwrap();
let now = Instant::now();
let pending = coordinator.connect(now).unwrap();
coordinator.tick(now + Duration::from_secs(5));
assert!(!coordinator.connections.contains_key(&pending));
let connection = connect(&mut coordinator);
let session = create(&mut coordinator, &connection);
coordinator.generation = wire::MAX_COUNTER - 2;
let control = claim(&mut coordinator, &connection, &session);
assert_eq!(control.generation, wire::MAX_COUNTER - 1);
coordinator.disconnect(&connection);
assert_eq!(coordinator.generation, wire::MAX_COUNTER);
let next = connect(&mut coordinator);
let attempt = request(
&coordinator,
&next,
"session.claim",
Some(&session),
None,
Some("exhausted-claim"),
json!({}),
);
assert_eq!(
invoke(&mut coordinator, &next, attempt)["error"]["code"],
"limit_exceeded"
);
assert!(coordinator.actors.is_empty());
}
#[test]
fn request_reservations_are_connection_scoped_and_malformed_duplicates_keep_original() {
let temp = TempDir::new().unwrap();
let mut coordinator = Coordinator::new(runtime(&temp)).unwrap();
let first = connect(&mut coordinator);
let second = connect(&mut coordinator);
for connection in [&first, &second] {
let mut read = request(
&coordinator,
connection,
"session.active",
None,
None,
None,
json!({}),
);
read.request_id = "shared-request".into();
coordinator
.submit(
connection,
&serde_json::to_vec(&read).unwrap(),
Instant::now(),
)
.unwrap();
let response: Value =
serde_json::from_str(coordinator.connections[connection].queue.front().unwrap())
.unwrap();
assert!(response["error"].is_null());
assert_eq!(
coordinator.connections[connection].requests["shared-request"],
1
);
}
let malformed = json!({"request_id":"shared-request","method":"session.active","payload":{},"unknown":"synthetic"});
coordinator
.submit(
&first,
&serde_json::to_vec(&malformed).unwrap(),
Instant::now(),
)
.unwrap();
let response: Value =
serde_json::from_str(coordinator.connections[&first].queue.back().unwrap()).unwrap();
assert_eq!(response["error"]["code"], "duplicate_request_id");
assert_eq!(
coordinator.connections[&first].requests["shared-request"],
2
);
assert_eq!(
coordinator.connections[&second].requests["shared-request"],
1
);
let service = PersistentService::spawn(coordinator).unwrap();
service.response_written(&first, "shared-request").unwrap();
service
.submit(&first, &serde_json::to_vec(&malformed).unwrap())
.unwrap();
for _ in 0..2 {
service.next_record(&first).unwrap();
}
let response: Value =
serde_json::from_str(&service.next_record(&first).unwrap().unwrap()).unwrap();
assert_eq!(response["error"]["code"], "duplicate_request_id");
}