diff --git a/.changes/fix-cancel-owned-prompt-admission.md b/.changes/fix-cancel-owned-prompt-admission.md new file mode 100644 index 000000000..09900c3bf --- /dev/null +++ b/.changes/fix-cancel-owned-prompt-admission.md @@ -0,0 +1 @@ +- Cancelling a running prompt by its admission id now reports it as `owned` instead of `cancelled`, and with `cancelOwned` the cancel now stops the turn. diff --git a/crates/pa-daemon/src/prompt_admission.rs b/crates/pa-daemon/src/prompt_admission.rs index f6ebd99af..9a5786bf8 100644 --- a/crates/pa-daemon/src/prompt_admission.rs +++ b/crates/pa-daemon/src/prompt_admission.rs @@ -534,7 +534,7 @@ impl Supervisor { /// The worker's admission registry: admission id -> status (the TS /// daemon-mode `promptAdmissions` map). Shared with the turn runner, -/// which commits a queued admission when its turn starts. +/// which clears an admitted prompt when its turn settles. #[derive(Default, Clone)] pub(crate) struct WorkerAdmissions { admissions: Arc>>, @@ -552,16 +552,20 @@ impl WorkerAdmissions { .insert(admission_id.to_string(), AdmissionStatus::Waiting); } - /// The queued prompt's turn started: a waiting admission commits. - pub(crate) fn commit(&self, admission_id: &str) { + /// Admission at prompt enqueue: a waiting id becomes owned. A cancelled + /// or removed id refuses delivery, including cancellation before enqueue. + pub(crate) fn commit(&self, admission_id: &str) -> bool { let mut admissions = self .admissions .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - if let Some(status) = admissions.get_mut(admission_id) { - if *status == AdmissionStatus::Waiting { + match admissions.get_mut(admission_id) { + Some(status @ AdmissionStatus::Waiting) => { *status = AdmissionStatus::Owned; + true } + Some(AdmissionStatus::Owned) => true, + Some(AdmissionStatus::Cancelled) | None => false, } } @@ -601,27 +605,6 @@ impl Worker { self.prompt_admissions.register(admission_id); } - /// Drop the queued prompt a cancelled admission was holding (the TS - /// controller aborts before the admission commits, so the prompt never - /// runs). - pub(crate) fn drop_queued_admitted_prompt(&self, admission_id: &str) { - { - let mut core = self.core.lock().unwrap(); - core.steering - .retain(|item| item.admission_id.as_deref() != Some(admission_id)); - core.follow_up - .retain(|item| item.admission_id.as_deref() != Some(admission_id)); - } - // The cancelled rows leave the lanes: settle the verdict so the - // drop cannot leave the admission's busy=true (or its snapshot - // rows) promising a revive work that was cancelled. A drop with - // other rows still queued stays busy — that work is real, and a - // mid-turn drop stays busy through the in-flight turn. - self.checkpoint_queue(crate::worker::QueueCheckpoint::Settle { - operation: "queue_dropped", - }); - } - /// `cancel_prompt_admission` (the worker arm the supervisor forwards /// to): the TS status ladder over the worker's registry. pub(crate) fn handle_cancel_prompt_admission(&self, payload: &Value) -> DaemonResponse { @@ -633,41 +616,52 @@ impl Worker { .and_then(Value::as_str) .unwrap_or_default(); let cancel_owned = payload.get("cancelOwned").and_then(Value::as_bool) == Some(true); - match self.prompt_admissions.cancel(admission_id) { - None => response_success( - None, - "cancel_prompt_admission", - Some(json!({ "status": "unknown" })), - ), - Some(AdmissionStatus::Owned) => { - // A committed prompt with `cancelOwned` aborts its running - // turn (the TS controller abort). - if cancel_owned { - let mut core = self.core.lock().unwrap(); - core.abort_requested = true; - drop(core); - // The TS controller abort cancels the committed turn's - // in-flight fetch immediately (`requestAbort` -> - // `agent.abort()`). - self.engine.abort_in_flight_turn(); - } - response_success( - None, - "cancel_prompt_admission", - Some(json!({ "status": "owned" })), - ) + let (status, dropped_queued, abort_running) = { + // The enqueue/commit transition holds this same lock. Decide + // whether the owned admission is still queued atomically with + // removing it; never abort a different in-flight turn. + let mut core = self.core.lock().unwrap(); + let status = self.prompt_admissions.cancel(admission_id); + let queued = core + .steering + .iter() + .chain(&core.follow_up) + .any(|item| item.admission_id.as_deref() == Some(admission_id)); + let dropped_queued = matches!(status, Some(AdmissionStatus::Cancelled)) + || (cancel_owned && status == Some(AdmissionStatus::Owned) && queued); + if dropped_queued { + core.steering + .retain(|item| item.admission_id.as_deref() != Some(admission_id)); + core.follow_up + .retain(|item| item.admission_id.as_deref() != Some(admission_id)); + self.prompt_admissions.clear(admission_id); } - // A waiting or already-cancelled admission: the queued prompt - // (if any) never runs. - Some(_) => { - self.drop_queued_admitted_prompt(admission_id); - response_success( - None, - "cancel_prompt_admission", - Some(json!({ "status": "cancelled" })), - ) + let abort_running = cancel_owned + && status == Some(AdmissionStatus::Owned) + && core.running_admission_ids.contains(admission_id); + if abort_running { + core.abort_requested = true; } + (status, dropped_queued, abort_running) + }; + if dropped_queued { + self.checkpoint_queue(crate::worker::QueueCheckpoint::Settle { + operation: "queue_dropped", + }); + } + if abort_running { + self.engine.abort_in_flight_turn(); } + let status = match status { + None => "unknown", + Some(AdmissionStatus::Owned) => "owned", + Some(AdmissionStatus::Waiting | AdmissionStatus::Cancelled) => "cancelled", + }; + response_success( + None, + "cancel_prompt_admission", + Some(json!({ "status": status })), + ) } } diff --git a/crates/pa-daemon/src/worker.rs b/crates/pa-daemon/src/worker.rs index cb0d7d53b..1c0bec047 100644 --- a/crates/pa-daemon/src/worker.rs +++ b/crates/pa-daemon/src/worker.rs @@ -355,6 +355,7 @@ impl Worker { pending_next_turn: Vec::new(), active_action: None, running_tool_calls: std::collections::HashSet::new(), + running_admission_ids: std::collections::HashSet::new(), }; let active_session_id = config.active_session_id.clone(); let script = config.script.clone(); @@ -740,7 +741,7 @@ impl Worker { recovery: Arc::clone(&recovery), core: Arc::clone(&core), input_pauses: input_pauses.clone(), - prompt_admissions, + prompt_admissions: prompt_admissions.clone(), work_notify: Arc::clone(&work_notify), idle_notify: Arc::clone(&idle_notify), events: events.clone(), @@ -795,7 +796,6 @@ impl Worker { remote_source: None, probe_override: None, }); - let prompt_admissions = crate::prompt_admission::WorkerAdmissions::new(); let navigation = crate::session_navigation::SessionNavigation::new( std::sync::Arc::clone(&engine), Arc::clone(&core), diff --git a/crates/pa-daemon/src/worker/commands.rs b/crates/pa-daemon/src/worker/commands.rs index 38462f1ef..fa5d17e24 100644 --- a/crates/pa-daemon/src/worker/commands.rs +++ b/crates/pa-daemon/src/worker/commands.rs @@ -423,7 +423,8 @@ impl Worker { kept.push_back(item); } else { if let Some(id) = &item.admission_id { - let _ = self.prompt_admissions.cancel(id); + // A withdrawn prompt clears its admission (TS clearAdmission). + self.prompt_admissions.clear(id); } if let Some(done) = item.done { let _ = done.send(TurnSettle::Withdrawn( @@ -565,8 +566,18 @@ impl Worker { return response; } let mut core = self.core.lock().unwrap(); - let steering: Vec = core.steering.drain(..).map(|item| item.message).collect(); - let follow_up: Vec = core.follow_up.drain(..).map(|item| item.message).collect(); + let drain_lane = |lane: &mut VecDeque| -> Vec { + lane.drain(..) + .map(|item| { + if let Some(id) = item.admission_id.as_deref() { + self.prompt_admissions.clear(id); + } + item.message + }) + .collect() + }; + let steering: Vec = drain_lane(&mut core.steering); + let follow_up: Vec = drain_lane(&mut core.follow_up); let snapshot = Self::snapshot_locked(&core); drop(core); // The cleared lanes are idle again: the verdict refresh rides the diff --git a/crates/pa-daemon/src/worker/input.rs b/crates/pa-daemon/src/worker/input.rs index c26c61c32..9661e821e 100644 --- a/crates/pa-daemon/src/worker/input.rs +++ b/crates/pa-daemon/src/worker/input.rs @@ -68,7 +68,7 @@ impl Worker { } // The prompt-admission bookkeeping (wave b9): a prompt carrying an // admission id registers it worker-side; the queued item carries - // it so the turn runner commits the admission when its turn starts. + // it so the turn runner clears the admission at settle. let admission_id = payload .get("admissionId") .and_then(Value::as_str) @@ -81,6 +81,19 @@ impl Worker { let done = if wait { Some(done_tx) } else { None }; let (snapshot, queued_behind_work) = { let mut core = self.core.lock().unwrap(); + // TS commits before accepting the prompt into its action queue. + // Hold the queue lock across this transition and enqueue so a + // cancellation cannot mistake an accepted prompt for waiting. + if let Some(id) = &admission_id { + if !self.prompt_admissions.commit(id) { + return response_failure( + None, + if wait { "prompt_and_wait" } else { "prompt" }, + "Prompt admission was cancelled.", + None, + ); + } + } // An idle session runs the prompt immediately: the lane is the // work hand-off, not a queue, so the projection did not change // (TS prompt admission with queueIfBusy=false never queues). diff --git a/crates/pa-daemon/src/worker/recovery_verdict_tests.rs b/crates/pa-daemon/src/worker/recovery_verdict_tests.rs index a42b20b40..379f7664e 100644 --- a/crates/pa-daemon/src/worker/recovery_verdict_tests.rs +++ b/crates/pa-daemon/src/worker/recovery_verdict_tests.rs @@ -241,8 +241,8 @@ async fn bash_consumed_withdraws_the_undelivered_notice_and_settles() { let _ = std::fs::remove_dir_all(worker.config.socket_path.parent().unwrap()); } -/// Dropping a cancelled admission settles the verdict: the cancelled -/// rows leave no busy evidence and no replayable snapshot. +/// Cancelling an owned queued admission settles the verdict: removed rows +/// leave no busy evidence and no replayable snapshot. #[tokio::test] async fn cancelled_admission_drop_settles_the_verdict() { let worker = created_worker_with_journal().await; @@ -255,7 +255,13 @@ async fn cancelled_admission_drop_settles_the_verdict() { WorkerRecoveryJournal::read_interrupted(&worker.config.recovery_journal_path), "the admitted prompt is live work" ); - worker.drop_queued_admitted_prompt("a1"); + let cancelled = worker + .dispatch( + "cancel_prompt_admission", + &json!({ "admissionId": "a1", "cancelOwned": true }), + ) + .await; + assert_eq!(cancelled.data, Some(json!({ "status": "owned" }))); assert!( !WorkerRecoveryJournal::read_interrupted(&worker.config.recovery_journal_path), "the dropped rows leave no busy evidence" diff --git a/crates/pa-daemon/src/worker/session_core.rs b/crates/pa-daemon/src/worker/session_core.rs index 464ec295a..f41fe2064 100644 --- a/crates/pa-daemon/src/worker/session_core.rs +++ b/crates/pa-daemon/src/worker/session_core.rs @@ -42,6 +42,9 @@ pub(crate) struct SessionCore { /// and remove ids, and the roster summary derives `isRunningTools` /// (`isStreaming && pendingToolCalls.size > 0`) from its size. pub(crate) running_tool_calls: std::collections::HashSet, + /// Admission ids belonging to the current in-flight turn. The queue + /// handoff and owned cancellation both inspect this under the core lock. + pub(crate) running_admission_ids: std::collections::HashSet, /// TS `autoCompactionEnabled` (settings default: on). pub(crate) auto_compaction_enabled: bool, /// The last broadcast queue snapshot (TS `_lastSessionActionSnapshot`): @@ -159,6 +162,7 @@ impl SessionCore { last_activity_ms: 0, compacting: false, running_tool_calls: std::collections::HashSet::new(), + running_admission_ids: std::collections::HashSet::new(), auto_compaction_enabled: true, last_action_snapshot: Some(SessionActionSnapshot::default()), rlm_depth: 0, diff --git a/crates/pa-daemon/src/worker/tests/abort_boundary_tests.rs b/crates/pa-daemon/src/worker/tests/abort_boundary_tests.rs index 1ee4fe38d..2913971de 100644 --- a/crates/pa-daemon/src/worker/tests/abort_boundary_tests.rs +++ b/crates/pa-daemon/src/worker/tests/abort_boundary_tests.rs @@ -537,3 +537,194 @@ async fn abort_and_clear_queue_suspends_plain_prompts() { let idle = worker.dispatch("wait_for_idle", &json!({})).await; assert!(idle.success, "never went idle: {idle:?}"); } + +/// TS commits admission before queue delivery: a queued prompt answers +/// `owned` but does not affect the running turn unless cancelled with +/// `cancelOwned`. The running admission's `cancelOwned` aborts its turn +/// and the settle clears the admission. +#[tokio::test] +async fn cancel_owned_admission_aborts_the_running_prompt() { + let dir = std::env::temp_dir().join(format!("pa-worker-cancel-owned-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&dir).unwrap(); + let config = WorkerConfig { + socket_path: dir.join("worker.sock"), + supervisor_socket_path: PathBuf::new(), + token: "token".to_string(), + worker_instance_id: String::new(), + active_session_id: "cancel-owned-session".to_string(), + agent_dir: dir.join("agent"), + recovery_journal_path: dir.join("recovery.jsonl"), + telemetry_disabled: None, + script: Some(json!({ "responses": [{ "text": "held reply", "delayMs": 60000 }] })), + }; + let worker = Arc::new(Worker::new(config, None)); + let created = worker + .dispatch( + "create", + &json!({ "noSession": true, "cwd": "/tmp", "name": "cancel-owned" }), + ) + .await; + assert!(created.success, "create failed: {created:?}"); + let mut subscription = worker.events.subscribe(); + let wait = tokio::spawn({ + let worker = Arc::clone(&worker); + async move { + worker + .dispatch( + "prompt_and_wait", + &json!({ + "activeSessionId": "cancel-owned-session", + "message": "go", + "admissionId": "adm-1", + }), + ) + .await + } + }); + // The turn commits its admission before it emits `agent_start`. + loop { + let frame = subscription.recv().await.unwrap(); + if frame.outbound_type == "session_event" + && serde_json::from_slice::(&frame.payload).unwrap()["event"]["type"] + == "agent_start" + { + break; + } + } + // The held turn keeps the session busy, so a second prompt queues + // behind it. Admission is already owned before delivery, as in TS. + let queued_wait = tokio::spawn({ + let worker = Arc::clone(&worker); + async move { + worker + .dispatch( + "prompt_and_wait", + &json!({ + "activeSessionId": "cancel-owned-session", + "message": "queued behind the held turn", + "admissionId": "adm-2", + }), + ) + .await + } + }); + // The queued prompt has been accepted while its response waits on the + // held turn; its admission is already owned, not yet settled. + loop { + if worker.prompt_admissions.cancel("adm-2") + == Some(crate::prompt_admission::AdmissionStatus::Owned) + { + break; + } + tokio::task::yield_now().await; + } + let queued_cancel = json!({ + "activeSessionId": "cancel-owned-session", + "admissionId": "adm-2", + }); + let queued_owned = worker + .dispatch("cancel_prompt_admission", &queued_cancel) + .await; + assert_eq!( + queued_owned.data, + Some(json!({ "status": "owned" })), + "{queued_owned:?}" + ); + // Without cancelOwned, the queued prompt remains queued. + let queue = worker.dispatch("get_queue", &json!({})).await; + assert!(serde_json::to_string(&queue.data) + .unwrap() + .contains("queued behind the held turn")); + let queued_removed = worker + .dispatch( + "cancel_prompt_admission", + &json!({"activeSessionId": "cancel-owned-session", "admissionId": "adm-2", "cancelOwned": true}), + ) + .await; + assert_eq!( + queued_removed.data, + Some(json!({ "status": "owned" })), + "{queued_removed:?}" + ); + let cleared_queued = worker + .dispatch("cancel_prompt_admission", &queued_cancel) + .await; + assert_eq!(cleared_queued.data, Some(json!({ "status": "unknown" }))); + // clear_queue withdraws the drained row: its admission clears with it (TS clearAdmission). + let queued_three = worker + .dispatch( + "prompt", + &json!({ + "activeSessionId": "cancel-owned-session", + "message": "cleared by clear_queue", + "admissionId": "adm-3", + }), + ) + .await; + assert!( + queued_three.success, + "queued adm-3 failed: {queued_three:?}" + ); + assert!( + worker.dispatch("clear_queue", &json!({})).await.success, + "clear_queue failed" + ); + let drained_cancel = worker + .dispatch( + "cancel_prompt_admission", + &json!({"activeSessionId": "cancel-owned-session", "admissionId": "adm-3"}), + ) + .await; + assert_eq!( + drained_cancel.data, + Some(json!({ "status": "unknown" })), + "{drained_cancel:?}" + ); + let queued_result = queued_wait.await.unwrap(); + assert!( + !queued_result.success, + "the withdrawn prompt cannot complete" + ); + // Simulate the narrow settle gap: an earlier owned admission still + // exists in the registry, but this different turn is now running. + worker.register_prompt_admission("stale-owned"); + assert!(worker.prompt_admissions.commit("stale-owned")); + let stale = worker + .dispatch( + "cancel_prompt_admission", + &json!({"activeSessionId": "cancel-owned-session", "admissionId": "stale-owned", "cancelOwned": true}), + ) + .await; + assert_eq!(stale.data, Some(json!({ "status": "owned" }))); + assert!( + !worker.core.lock().unwrap().abort_requested, + "stale admission must not abort the unrelated running turn" + ); + let cancel = json!({ + "activeSessionId": "cancel-owned-session", + "admissionId": "adm-1", + "cancelOwned": true, + }); + let owned = worker.dispatch("cancel_prompt_admission", &cancel).await; + assert_eq!(owned.data, Some(json!({ "status": "owned" })), "{owned:?}"); + wait.await.unwrap(); + let idle = worker.dispatch("wait_for_idle", &json!({})).await; + assert!(idle.success, "never went idle: {idle:?}"); + let messages = worker.dispatch("get_messages", &json!({})).await; + let messages = serde_json::to_string(&messages.data).unwrap(); + assert!( + !messages.contains("queued behind the held turn"), + "the cancelled admission's prompt ran: {messages}" + ); + assert!( + !messages.contains("held reply"), + "the turn ran to the end: {messages}" + ); + let cleared = worker.dispatch("cancel_prompt_admission", &cancel).await; + assert_eq!( + cleared.data, + Some(json!({ "status": "unknown" })), + "{cleared:?}" + ); + let _ = std::fs::remove_dir_all(&dir); +} diff --git a/crates/pa-daemon/src/worker/turn.rs b/crates/pa-daemon/src/worker/turn.rs index 554667e99..4549a7b58 100644 --- a/crates/pa-daemon/src/worker/turn.rs +++ b/crates/pa-daemon/src/worker/turn.rs @@ -14,8 +14,8 @@ pub(super) struct TurnRunner { pub(crate) core: Arc>, /// The input-pause table (the admission gate holds queued input). pub(super) input_pauses: crate::session_input_pause::InputPauseTable, - /// The prompt-admission registry: a queued admitted prompt commits - /// when its turn starts and clears when the turn settles. + /// The prompt-admission registry: an admitted prompt clears when its + /// turn settles. pub(super) prompt_admissions: crate::prompt_admission::WorkerAdmissions, pub(super) work_notify: Arc, pub(super) idle_notify: Arc, @@ -91,6 +91,10 @@ impl TurnRunner { None } else if core.steering.front().is_some() { let items = gather_delivery_batch(&mut core, Lane::Steering); + core.running_admission_ids = items + .iter() + .filter_map(|item| item.admission_id.clone()) + .collect(); core.busy = true; core.abort_requested = false; core.retry_abort_requested = false; @@ -100,6 +104,10 @@ impl TurnRunner { Some(items) } else if core.follow_up.front().is_some() { let items = gather_delivery_batch(&mut core, Lane::FollowUp); + core.running_admission_ids = items + .iter() + .filter_map(|item| item.admission_id.clone()) + .collect(); core.busy = true; core.abort_requested = false; core.retry_abort_requested = false; @@ -418,11 +426,6 @@ impl TurnRunner { let Some((first, batched)) = items.split_first() else { return; }; - // An admitted prompt's turn started: its prompt admission commits - // (TS `commitAdmission`) — one per batched item, in delivery order. - for admission_id in items.iter().filter_map(|item| item.admission_id.as_ref()) { - self.prompt_admissions.commit(admission_id); - } self.emit_turn_event(json!({ "type": "agent_start" })); self.emit_turn_event(json!({ "type": "turn_start" })); // The pane reporter's run boundary (TS `agent_start`): working. @@ -1233,6 +1236,7 @@ impl TurnRunner { let mut core = self.core.lock().unwrap(); core.busy = false; core.active_action = None; + core.running_admission_ids.clear(); } self.push_roster_delta(); // The fallback `agent_end` for runs that ended without a model diff --git a/crates/pa-daemon/src/worker/turn_stream_tests.rs b/crates/pa-daemon/src/worker/turn_stream_tests.rs index 397cb264e..99e9b30b5 100644 --- a/crates/pa-daemon/src/worker/turn_stream_tests.rs +++ b/crates/pa-daemon/src/worker/turn_stream_tests.rs @@ -57,6 +57,7 @@ fn burst_runner(engine: Arc) -> TurnRunner { pending_next_turn: Vec::new(), active_action: None, running_tool_calls: std::collections::HashSet::new(), + running_admission_ids: std::collections::HashSet::new(), })); TurnRunner { core, diff --git a/crates/pa-daemon/tests/protocol_breadth_b6_b9.rs b/crates/pa-daemon/tests/protocol_breadth_b6_b9.rs index d1fc413bd..99ffa6718 100644 --- a/crates/pa-daemon/tests/protocol_breadth_b6_b9.rs +++ b/crates/pa-daemon/tests/protocol_breadth_b6_b9.rs @@ -678,7 +678,6 @@ fn wave_b9_prompt_admission_wire_shapes() { &json!({ "type": "cancel_prompt_admission", "activeSessionId": session_id, "admissionId": "never-registered" }), ); let response = client.read_response("c-0"); - assert_eq!(response["success"], true, "{response}"); assert_eq!(response["data"], json!({ "status": "unknown" })); // An empty admission id answers the TS parse error. @@ -710,10 +709,11 @@ fn wave_b9_prompt_admission_wire_shapes() { // not race the registration). std::thread::sleep(Duration::from_millis(100)); - // The cancel answers `cancelled` and the queued prompt never runs. + // The queued prompt committed at admission: cancelOwned withdraws + // that item without aborting the unrelated slow turn. client.send_command( "c-1", - &json!({ "type": "cancel_prompt_admission", "activeSessionId": session_id, "admissionId": "adm-1" }), + &json!({ "type": "cancel_prompt_admission", "activeSessionId": session_id, "admissionId": "adm-1", "cancelOwned": true }), ); // The cancelled wait's failure response may land before the cancel's // own response (both frames traverse the same worker pipe, and the @@ -721,7 +721,7 @@ fn wave_b9_prompt_admission_wire_shapes() { // it), so the read buffers instead of dropping it. let (response, mut lines) = client.read_response_and_lines("c-1"); assert_eq!(response["success"], true, "{response}"); - assert_eq!(response["data"], json!({ "status": "cancelled" })); + assert_eq!(response["data"], json!({ "status": "owned" })); // The cancelled prompt_and_wait fails its wait (the dropped queue item // never completes). let response = match lines @@ -744,6 +744,21 @@ fn wave_b9_prompt_admission_wire_shapes() { assert_eq!(response["data"]["steering"], json!([]), "{response}"); assert_eq!(response["data"]["followUp"], json!([]), "{response}"); + // A non-wait prompt's public admission ends when its success response + // lands, even if its queued turn has not run yet (TS prompt lifetime). + client.send_command( + "p-3", + &json!({ "type": "prompt", "activeSessionId": session_id, "message": "non-wait", "admissionId": "adm-3" }), + ); + let response = client.read_response("p-3"); + assert_eq!(response["success"], true, "{response}"); + client.send_command( + "c-3", + &json!({ "type": "cancel_prompt_admission", "activeSessionId": session_id, "admissionId": "adm-3", "cancelOwned": true }), + ); + let response = client.read_response("c-3"); + assert_eq!(response["data"], json!({ "status": "unknown" })); + // A cancel for a fresh id answers `unknown` (nothing registered). client.send_command( "c-2",