use super::*;
use std::sync::atomic::{AtomicBool, Ordering};
fn setup() -> (SharedApplicationState, InlineTurn, Value) {
let mut state = ApplicationState::default();
let issued = state
.route("first", "executor.create", None, &json!({}), Instant::now())
.unwrap();
let proof =
json!({"executor_id":issued["executor_id"],"resume_secret":issued["resume_secret"]});
state
.route("first", "executor.confirm", None, &proof, Instant::now())
.unwrap();
let schema = json!({"type":"object","properties":{"text":{"type":"string","maxLength":128}},"required":["text"],"additionalProperties":false});
let turn = InlineTurn { executor_id: issued["executor_id"].as_str().unwrap().into(), executor_generation: 1,
tool: Arc::new(decode(&json!({"name":"client_echo","description":"Echo","input_schema":schema,"output_schema":schema})).unwrap()),
skill: None, resource_identity: None };
(Arc::new(Mutex::new(state)), turn, proof)
}
fn start(
state: SharedApplicationState,
turn: &InlineTurn,
cancellation: Arc<AtomicBool>,
) -> std::thread::JoinHandle<ToolResult> {
let tools = turn.tools(state, "session".into(), "turn".into());
std::thread::spawn(move || {
let mut context = ToolDispatchContext::new(None, None);
context.provider_call_id = Some("provider-id".into());
context.application_call_id = Some("persisted-call-id".into());
context.cancellation = crate::cancellation::AgentCancellation::new(cancellation);
(tools.callback)("client_echo", json!({"text":"input"}), &context)
})
}
fn wait_for_call(state: &SharedApplicationState, dispatched: bool) {
let deadline = Instant::now() + Duration::from_secs(3);
loop {
if state
.lock()
.unwrap()
.calls
.get("persisted-call-id")
.is_some_and(|call| call.dispatched == dispatched)
{
return;
}
assert!(
Instant::now() < deadline,
"call did not reach expected state"
);
std::thread::sleep(Duration::from_millis(1));
}
}
fn submit(
state: &mut ApplicationState,
connection: &str,
turn: &InlineTurn,
generation: u64,
) -> Result<Value, Code> {
state.route(connection, "turn.application_tool_result", Some("session"), &json!({
"executor_id":turn.executor_id,"executor_generation":generation,"turn_id":"turn","call_id":"persisted-call-id",
"revision_sha256":turn.manifest()["revision_sha256"],"outcome":{"status":"success","value":{"text":"retained result"}}
}), Instant::now())
}
fn reserve_test_evidence(state: &SharedApplicationState, turn: &InlineTurn) {
state
.lock()
.unwrap()
.reserve_evidence(
"operation",
"session",
&json!({"profile":PROFILE,"tools":[turn.manifest()]}),
)
.unwrap();
}
#[test]
fn resume_fences_old_socket_and_preserves_deadline_without_redispatch() {
let (state, turn, proof) = setup();
reserve_test_evidence(&state, &turn);
let worker = start(Arc::clone(&state), &turn, Arc::new(AtomicBool::new(false)));
wait_for_call(&state, true);
{
let mut state = state.lock().unwrap();
assert_eq!(state.drain_events().len(), 1);
let deadline = state.calls["persisted-call-id"].deadline;
let resumed = state
.route("second", "executor.resume", None, &proof, Instant::now())
.unwrap();
assert_eq!(resumed["executor_generation"], 2);
assert_eq!(resumed["pending_calls"][0]["wait_kind"], "execution");
assert_eq!(resumed["pending_calls"][0]["external_effects"], "uncertain");
assert!(
resumed["pending_calls"][0]["remaining_ms"]
.as_u64()
.unwrap()
<= 120_000
);
assert!(submit(&mut state, "first", &turn, 1).is_err());
assert!(submit(&mut state, "second", &turn, 1).is_err());
assert_eq!(state.calls["persisted-call-id"].deadline, deadline);
submit(&mut state, "second", &turn, 2).unwrap();
assert!(submit(&mut state, "second", &turn, 2).is_err());
}
assert!(worker.join().unwrap().success);
assert!(state.lock().unwrap().drain_events().is_empty());
let mut state = state.lock().unwrap();
state.finish_turn("turn");
let evidence = state
.read_evidence("session", &json!({"turn_id":"turn"}))
.unwrap();
assert_eq!(evidence["calls"][0]["executor_generation"], 1);
assert_eq!(evidence["calls"][0]["status"], "succeeded");
assert_eq!(evidence["calls"][0]["external_effects"], "reported");
assert!(evidence["calls"][0]["wait_kind"].is_null());
assert!(evidence["calls"][0]["remaining_ms"].is_null());
assert!(!evidence.to_string().contains("retained result"));
}
#[test]
fn callback_need_starts_availability_wait_and_resume_allows_first_dispatch() {
let (state, turn, proof) = setup();
reserve_test_evidence(&state, &turn);
state
.lock()
.unwrap()
.disconnect("first", Instant::now() - Duration::from_secs(80));
let worker = start(Arc::clone(&state), &turn, Arc::new(AtomicBool::new(false)));
wait_for_call(&state, false);
{
let mut state = state.lock().unwrap();
assert!(
state.calls["persisted-call-id"].deadline > Instant::now() + Duration::from_secs(58)
);
assert!(state.drain_events().is_empty());
let evidence = state
.read_evidence("session", &json!({"turn_id":"turn"}))
.unwrap();
let waiting = &evidence["calls"][0];
assert_eq!(waiting["status"], "waiting_for_executor");
assert_eq!(waiting["wait_kind"], "availability");
assert_eq!(waiting["external_effects"], "not_dispatched");
assert!(waiting["remaining_ms"].as_u64().unwrap() <= 60_000);
state
.route("second", "executor.resume", None, &proof, Instant::now())
.unwrap();
}
wait_for_call(&state, true);
{
let mut state = state.lock().unwrap();
let events = state.drain_events();
assert_eq!(events.len(), 1);
assert_eq!(events[0].connection, "second");
assert_eq!(events[0].payload["timeout_ms"], 120_000);
submit(&mut state, "second", &turn, 2).unwrap();
}
assert!(worker.join().unwrap().success);
}
#[test]
fn escaped_call_identities_cannot_overflow_a_resumed_executor_response() {
let (state, turn, proof) = setup();
let mut state = state.lock().unwrap();
let mut admitted = 0;
for index in 0..MAX_CALLS {
let id = format!("{index}{}", "\u{0001}".repeat(120));
let call = Call {
executor: turn.executor_id.clone(),
session: "\u{0001}".repeat(128),
turn: "\u{0001}".repeat(128),
provider_call_id: "\u{0001}".repeat(128),
revision: "f".repeat(64),
resource_identity: None,
deadline: Instant::now() + Duration::from_secs(600),
dispatched: false,
outcome: None,
output_schema: turn.tool.output_schema.clone(),
cancellation: crate::cancellation::AgentCancellation::new(Arc::new(AtomicBool::new(
false,
))),
};
if state.reserve_resume_capacity(&id, &call).is_err() {
break;
}
state.calls.insert(id, call);
admitted += 1;
}
assert!(admitted > 0 && admitted < MAX_CALLS);
let resumed = state
.route("second", "executor.resume", None, &proof, Instant::now())
.unwrap();
crate::service::protocol::payload_is_bounded(&resumed).unwrap();
assert_eq!(resumed["pending_calls"].as_array().unwrap().len(), admitted);
assert_eq!(resumed["executor_generation"], 2);
assert!(state.drain_events().is_empty());
}
#[test]
fn availability_and_execution_expiry_close_calls_and_reject_late_results() {
for dispatched in [false, true] {
let (state, turn, _) = setup();
reserve_test_evidence(&state, &turn);
if !dispatched {
state.lock().unwrap().disconnect("first", Instant::now());
}
let worker = start(Arc::clone(&state), &turn, Arc::new(AtomicBool::new(false)));
wait_for_call(&state, dispatched);
{
let mut state = state.lock().unwrap();
state.drain_events();
state.calls.get_mut("persisted-call-id").unwrap().deadline = Instant::now();
assert!(submit(&mut state, "first", &turn, 1).is_err());
}
let result = worker.join().unwrap();
assert!(!result.success);
assert_eq!(
result.content,
if dispatched {
"tool_timeout"
} else {
"executor_unavailable"
}
);
assert_eq!(
result.metadata["external_effects"],
if dispatched {
"uncertain"
} else {
"not_dispatched"
}
);
let evidence = state
.lock()
.unwrap()
.read_evidence("session", &json!({"turn_id":"turn"}))
.unwrap();
assert_eq!(evidence["calls"][0]["status"], "failed");
assert_eq!(
evidence["calls"][0]["external_effects"],
result.metadata["external_effects"]
);
let events = state.lock().unwrap().drain_events();
if dispatched {
assert_eq!(events[0].name, "turn.application_tool_cancel");
} else {
assert!(events.is_empty());
}
}
}
#[test]
fn accepted_result_wins_over_later_stop_but_stop_rejects_new_results() {
for accepted_first in [false, true] {
let (state, turn, _) = setup();
let cancellation = Arc::new(AtomicBool::new(false));
let worker = start(Arc::clone(&state), &turn, Arc::clone(&cancellation));
wait_for_call(&state, true);
{
let mut state = state.lock().unwrap();
if accepted_first {
submit(&mut state, "first", &turn, 1).unwrap();
}
cancellation.store(true, Ordering::SeqCst);
assert!(submit(&mut state, "first", &turn, 1).is_err());
}
assert_eq!(worker.join().unwrap().success, accepted_first);
}
}
#[test]
fn strict_contracts_reject_invalid_timeouts_schemas_and_collisions() {
let (_, turn, _) = setup();
let value = serde_json::to_value(&turn).unwrap();
let capture = |value: &Value| -> Result<Contract, Code> {
let contract = decode(&value["tool"])?;
validate_contract(&contract)?;
Ok(contract)
};
for invalid in [
json!(null),
json!(999),
json!(600_001),
json!(1.5),
json!("120000"),
] {
let mut value = value.clone();
value["tool"]["timeout_ms"] = invalid;
assert!(capture(&value).is_err());
}
for timeout in [1000, 600000] {
let mut value = value.clone();
value["tool"]["timeout_ms"] = json!(timeout);
assert_eq!(capture(&value).unwrap().timeout_ms, timeout);
}
let mut omitted = value.clone();
omitted["tool"]
.as_object_mut()
.unwrap()
.remove("timeout_ms");
assert_eq!(capture(&omitted).unwrap().timeout_ms, 120_000);
for name in ["read", "mcp__echo", "Uppercase", "with-dash"] {
let mut value = value.clone();
value["tool"]["name"] = json!(name);
assert!(capture(&value).is_err());
}
let mut unknown = value.clone();
unknown["tool"]["input_schema"]["properties"]["text"]["pattern"] = json!(".*");
assert!(capture(&unknown).is_err());
let mut non_required = value;
non_required["tool"]["output_schema"]["required"] = json!([]);
assert!(capture(&non_required).is_err());
}
#[test]
fn unconfirmed_expired_and_released_secrets_cannot_authorize() {
let mut state = ApplicationState::default();
let now = Instant::now();
let issued = state
.route("first", "executor.create", None, &json!({}), now)
.unwrap();
let id = issued["executor_id"].as_str().unwrap();
assert!(state.authorize("first", id, 1).is_err());
let proof = json!({"executor_id":id,"resume_secret":issued["resume_secret"]});
assert!(
state
.route(
"second",
"executor.resume",
None,
&proof,
now + Duration::from_secs(60)
)
.is_err()
);
let (state, turn, proof) = setup();
let mut state = state.lock().unwrap();
state
.route(
"first",
"executor.release",
None,
&json!({"executor_id":turn.executor_id,"executor_generation":1}),
Instant::now(),
)
.unwrap();
assert!(
state
.route("second", "executor.resume", None, &proof, Instant::now())
.is_err()
);
}
#[test]
fn rotation_lost_reply_expiry_and_generation_fencing() {
let (shared, _, old_proof) = setup();
let mut state = shared.lock().unwrap();
let now = Instant::now();
let attachment = json!({"executor_id":old_proof["executor_id"],"executor_generation":1});
assert_eq!(
state.route("foreign", "executor.rotate", None, &attachment, now),
Err(Code::StaleConnection)
);
let mut invalid = attachment.clone();
invalid["timeout_ms"] = json!(1);
assert_eq!(
state.route("first", "executor.rotate", None, &invalid, now),
Err(Code::InvalidPayload)
);
let candidate = state
.route("first", "executor.rotate", None, &attachment, now)
.unwrap();
let mut confirmation = attachment.clone();
confirmation["resume_secret"] = candidate["resume_secret"].clone();
let expired = now + Duration::from_secs(60);
assert_eq!(
state.route(
"first",
"executor.rotate_confirm",
None,
&confirmation,
expired
),
Err(Code::StaleConnection)
);
let candidate_proof =
json!({"executor_id":old_proof["executor_id"],"resume_secret":candidate["resume_secret"]});
assert_eq!(
state.route("second", "executor.resume", None, &candidate_proof, expired),
Err(Code::StaleConnection)
);
state
.route("second", "executor.resume", None, &old_proof, expired)
.unwrap();
assert_eq!(
state.route("first", "executor.rotate", None, &attachment, expired),
Err(Code::StaleConnection)
);
let next_attachment = json!({"executor_id":old_proof["executor_id"],"executor_generation":2});
let next = state
.route("second", "executor.rotate", None, &next_attachment, expired)
.unwrap();
let next_proof =
json!({"executor_id":old_proof["executor_id"],"resume_secret":next["resume_secret"]});
assert_eq!(
state
.route("third", "executor.resume", None, &next_proof, expired)
.unwrap()["executor_generation"],
3
);
assert_eq!(
state.route("fourth", "executor.resume", None, &old_proof, expired),
Err(Code::StaleConnection)
);
}
#[test]
fn rotation_supersedes_candidate_and_old_resume_cancels_pending() {
let (shared, _, proof) = setup();
let mut state = shared.lock().unwrap();
let now = Instant::now();
let attachment = json!({"executor_id":proof["executor_id"],"executor_generation":1});
let first = state
.route("first", "executor.rotate", None, &attachment, now)
.unwrap();
let second = state
.route("first", "executor.rotate", None, &attachment, now)
.unwrap();
let mut confirmation = attachment.clone();
confirmation["resume_secret"] = first["resume_secret"].clone();
assert_eq!(
state.route("first", "executor.rotate_confirm", None, &confirmation, now),
Err(Code::StaleConnection)
);
state
.route("second", "executor.resume", None, &proof, now)
.unwrap();
confirmation["resume_secret"] = second["resume_secret"].clone();
assert_eq!(
state.route("first", "executor.rotate_confirm", None, &confirmation, now),
Err(Code::StaleConnection)
);
let candidate_proof =
json!({"executor_id":proof["executor_id"],"resume_secret":second["resume_secret"]});
assert_eq!(
state.route("third", "executor.resume", None, &candidate_proof, now),
Err(Code::StaleConnection)
);
}
#[test]
fn rotation_requires_confirmed_authority_and_strict_confirmation_payload() {
let now = Instant::now();
let mut state = ApplicationState::default();
let issued = state
.route("first", "executor.create", None, &json!({}), now)
.unwrap();
let attachment = json!({"executor_id":issued["executor_id"],"executor_generation":1});
assert_eq!(
state.route("first", "executor.rotate", None, &attachment, now),
Err(Code::StaleConnection)
);
let proof =
json!({"executor_id":issued["executor_id"],"resume_secret":issued["resume_secret"]});
state
.route("first", "executor.confirm", None, &proof, now)
.unwrap();
let candidate = state
.route("first", "executor.rotate", None, &attachment, now)
.unwrap();
let mut confirmation = attachment.clone();
confirmation["resume_secret"] = candidate["resume_secret"].clone();
for (field, invalid_value) in [
("resume_secret", Value::Null),
("executor_generation", json!(1.5)),
("extra", json!(true)),
] {
let mut invalid = confirmation.clone();
invalid[field] = invalid_value;
assert_eq!(
state.route("first", "executor.rotate_confirm", None, &invalid, now),
Err(Code::InvalidPayload)
);
}
state
.route("first", "executor.confirm", None, &proof, now)
.unwrap();
state
.route("first", "executor.rotate_confirm", None, &confirmation, now)
.unwrap();
assert_eq!(
state.route("second", "executor.resume", None, &proof, now),
Err(Code::StaleConnection)
);
let new_proof =
json!({"executor_id":issued["executor_id"],"resume_secret":candidate["resume_secret"]});
state
.route(
"second",
"executor.resume",
None,
&new_proof,
now + Duration::from_secs(60),
)
.unwrap();
let attachment = json!({"executor_id":issued["executor_id"],"executor_generation":2});
let candidate = state
.route(
"second",
"executor.rotate",
None,
&attachment,
now + Duration::from_secs(60),
)
.unwrap();
state
.route(
"second",
"executor.release",
None,
&attachment,
now + Duration::from_secs(60),
)
.unwrap();
let candidate_proof =
json!({"executor_id":issued["executor_id"],"resume_secret":candidate["resume_secret"]});
for proof in [new_proof, candidate_proof] {
assert_eq!(
state.route(
"third",
"executor.resume",
None,
&proof,
now + Duration::from_secs(60)
),
Err(Code::StaleConnection)
);
}
}