diff --git a/README.md b/README.md index 4bee949..c2d3747 100644 --- a/README.md +++ b/README.md @@ -57,7 +57,11 @@ Configure: `set-mode` (`build|plan|review`), `set-model`, `set-effort` (`auto|lo Work: `prompt`, `follow`, `watch`, `steer`, `queue`, `discard-queued`, `exec`, `cancel`, `runtime` -`prompt --new TEXT` creates a session. `prompt SESSION_ID TEXT` continues one. `-` reads the prompt from stdin. `--delivery auto` prompts when idle, steers when the turn is steerable, otherwise queues. It waits until idle, including queued turns. +`prompt --new TEXT` creates a session. `prompt SESSION_ID TEXT` continues one. `-` reads the prompt from stdin. `--delivery auto` prompts when idle, steers when the turn is steerable, otherwise queues. Auto delivery and idle waits require the updated backend observation metadata; unknown/unavailable status is not usable. + +Prompt/follow/watch wait for self-session idle, including queued turns, not for +child workflows or detached processes. Idle is not task success. Idle results +retain runtime observation health and separate direct delegation activity. `queue SESSION_ID -` and `steer SESSION_ID -` read stdin and return a submission receipt without loading history. `prompt --delivery queue|steer --until-idle false` @@ -65,6 +69,10 @@ also returns after submission; idle/direct prompt delivery still waits for its t `discard-queued` takes the `inputId` from an `input_state` event. That is the same value as `clientInputId` from the submit response. +`cancel` sends a cancellation notification for the current turn. Its +`cancellationState: "notification_sent"` confirms the local write/flush, not run +termination or task success. It does not discard queued inputs or cancel children. + `find` reports `pagesScanned`, `truncated`, `pageCapReached`, and `stopReason` so callers can distinguish a complete search from a result/page limit. @@ -102,7 +110,11 @@ a final `type: done` result. `watch` also emits `runtime` events. Exit codes: 0 ok, 1 usage, 2 connection, 3 RPC, 4 timeout or cancel. -Event types: `session`, `submitted`, `text`, `tool`, `mode`, `plan`, `input_state`, `permission`, `elicitation`, `runtime`, `update`, `done`. +Event types: `session`, `submitted`, `text`, `tool`, `delegation`, `mode`, `plan`, `input_state`, `permission`, `elicitation`, `runtime`, `update`, `done`. + +`delegation` records describe workflow lifecycle separately from dispatch `tool` +status. Compact `inspect` includes snapshot `delegations`; successful dispatch +alone does not establish delegation success. Submission receipts retain `clientInputId`; partial-failure errors include known IDs and delivery state in `recovery`. Acceptance is not task completion. See diff --git a/skills/qmacp/SKILL.md b/skills/qmacp/SKILL.md index 1da138b..e9c8d2f 100644 --- a/skills/qmacp/SKILL.md +++ b/skills/qmacp/SKILL.md @@ -28,7 +28,9 @@ keep the returned `sessionId`. Default endpoint: `ws://127.0.0.1:3030/acp/ws`. `--permission cancel` to fail closed. Elicitation is always cancelled. - JSON/NDJSON goes to stdout; connection logs go to stderr. Require exit 0 and JSON `ok` not false; `prompt`/`follow` also need a terminal `"type":"done"`. - Do not use `--pretty` on streams. Acceptance is not proof of task completion. + Do not use `--pretty` on streams. Acceptance and self idle are not task success. + `cancel` with `cancellationState: "notification_sent"` confirms only sending the + notification, not run termination. - After a timeout or uncertain submission, inspect `runtime` and history before retrying, steering, queuing, discarding, or cancelling. Do not claim changes succeeded without exit 0 and JSON `ok: true`. Retain receipts and error @@ -52,7 +54,9 @@ qmacp --quiet runtime "$SESSION_ID" Long prompts: use stdin (`-`), not nested quoting. Make them self-contained. Existing-session `prompt` defaults to `--delivery auto`: idle -> prompt, -busy and steerable -> steer, otherwise queue. It waits until idle by default. +busy and steerable -> steer, otherwise queue. Auto delivery and idle waits require +the updated backend observation contract; unknown/unavailable is not a usable +phase. Waiting means self-session idle, not descendant/process completion. Read [live control](control.md) for delivery, waiting, cancellation, and how to label agent-to-agent messages. diff --git a/skills/qmacp/callbacks.md b/skills/qmacp/callbacks.md index 5eb2bbc..89207a8 100644 --- a/skills/qmacp/callbacks.md +++ b/skills/qmacp/callbacks.md @@ -108,9 +108,10 @@ notify_callback() { steering; `steer --run-id` can target an exact active run. Do not assume queue-only delivery wakes an idle session. See [live control](control.md) for direct commands, including stdin delivery without history loading. -- `--until-idle false` avoids waiting for all subsequent work after - steer/queue. It is **not** fire-and-forget for an idle session: the prompt - RPC still waits for that turn, bounded here by `--timeout 900`. Keep the +- `--until-idle false` skips the self-session idle wait after steer/queue. + That wait never implies descendant workflow or detached process completion. + It is **not** fire-and-forget for an idle session: the prompt RPC still waits + for that turn, bounded here by `--timeout 900`. Keep the delivery process supervised. - Keep stdout NDJSON and stderr in the watcher logs; check exit status and the final `done` event. Submission/`done` alone does not prove the agent handled diff --git a/skills/qmacp/cli.md b/skills/qmacp/cli.md index 1c17bf5..6d1f013 100644 --- a/skills/qmacp/cli.md +++ b/skills/qmacp/cli.md @@ -39,29 +39,50 @@ qmacp [--url URL] [--allow-insecure] [--pretty] [--permission allow-once|reject- - `prompt`, `follow`, `exec`, and opt-in `new --stream`: NDJSON. `watch` emits `runtime` events, then a final JSON object. Connection logs go to stderr. -- Success requires exit 0 and JSON `ok` not false. `prompt`, `follow`, and +- Command success requires exit 0 and JSON `ok` not false. `prompt`, `follow`, and `new --stream` also require a terminal `"type":"done"`. `exec` emits individual command results, not a global `done`. - Exit codes: 0 success, 1 usage, 2 connection, 3 RPC, 4 timeout/cancel. -Stream event types: `session`, `submitted`, `text`, `tool`, `mode`, `plan`, -`input_state`, `permission`, `elicitation`, `runtime`, `update`, `done`. -Do not claim a cancel/discard/close/config change succeeded without exit 0 and -JSON `ok: true`. Inspect current state after uncertain delivery rather than retrying -blindly; see [live control](control.md). +Stream event types: `session`, `submitted`, `text`, `tool`, `delegation`, `mode`, +`plan`, `input_state`, `permission`, `elicitation`, `runtime`, `update`, `done`. +`delegation` is a separate lifecycle record from `querymt/session/delegationUpdate`, +also delivered by the exported `AcpClient::connect_with_events` channel. It retains +backend identities, state, child/tool links, stage/activity, outcome details and +`sourceEpoch`/`sessionSeq` when supplied, without inferred defaults. A completed +`tool` can mean only successful dispatch while its delegation is still verifying +or later fails. Delegation ordering is separate from self-runtime revisions; +records are forwarded, not merged into a client-side state cache. + +Require exit 0 and JSON `ok: true` for command success. For `cancel`, +`cancellationState: "notification_sent"` confirms only the local notification +write/flush, not backend acceptance or execution termination. Inspect current +state after uncertain delivery rather than retrying blindly; see [live control](control.md). `new --stream`/`prompt --new` emit `sessionId` before configuration. Default `new` still returns one object. Queue/steer prompts emit `submitted` with `clientInputId` and `receipt` before waiting; `done` retains them. On failure, `recovery` preserves known IDs, the failing `stage` -(`create|configure|submit|observe`), and creation/delivery state. `not_submitted` +(`create|configure|submit|open|observe`), and creation/delivery state. `not_submitted` or `rejected` differs from `unknown`; `accepted` means a receipt was received, not that the task completed. Inspect state before retrying an uncertain or accepted submission. A lost creation response cannot provide a session ID. -`observationState` is `not_waited`, `idle`, or `prompt_returned`, not proof of -task success. `text` contains only assistant text observed on this connection. +Successful prompt `observationState` is `not_waited`, `idle`, or `prompt_returned`; +follow reports `idle`. None proves task success. Idle results retain `runtime`, +including backend observation health, run identity and direct delegation activity. +Self idle does not mean descendants finished or detached OS processes exited. +The prompt RPC `stopReason` is separate from idle evidence; it is also retained in +recovery when observation fails after the RPC returned. It does not describe a +later queued run. No terminal outcome is inferred from idle. + +Open/observation failures emit no successful `done`; recovery reports +`observationState: "unavailable"` with the session ID and runtime evidence when +available. A retained runtime may be the last known observation, not current +liveness. Updated backend observation metadata is required for auto delivery and +waits; absent/unknown/unavailable health is not a usable busy or idle state. +`text` contains only assistant text observed on this connection. ## Permissions diff --git a/skills/qmacp/control.md b/skills/qmacp/control.md index 529ec21..1a9173c 100644 --- a/skills/qmacp/control.md +++ b/skills/qmacp/control.md @@ -26,10 +26,12 @@ self-contained; elicitation is always cancelled. Existing sessions accept - `queue`: submit input for after the current turn. Whether it starts an idle session depends on server support; use `auto` when that support is unknown. -`--until-idle` defaults true and waits through queued turns. Pass `--timeout` -explicitly for predictable bounds. `--until-idle false` skips waiting for idle -following steer/queue; it is not fire-and-forget when idle delivery uses the -prompt RPC, which still waits for that turn. +`--until-idle` defaults true and waits for this session's own idle state, including +queued turns. It does not wait for descendant workflows or detached OS processes, +and idle is not task success. Pass `--timeout` explicitly for predictable bounds. +`--until-idle false` skips waiting for idle following steer/queue; it is not +fire-and-forget when idle delivery uses the prompt RPC, which still waits for +that turn. Queue/steer delivery skips history loading. Direct commands return a receipt; `prompt --until-idle false` returns after submission when either is selected. @@ -62,21 +64,33 @@ qmacp --quiet cancel "$SESSION_ID" qmacp --quiet close "$SESSION_ID" ``` -- `follow` streams session events until idle and loads cwd from the current - process; run it from the session workspace. -- `watch` polls `runtime` (`"type":"runtime"`) until idle unless `--idle false`. - Default `follow`/`watch` timeout is 120s. +- `follow` requires successful session open, then streams events until self idle; + run it from the session workspace (cwd comes from the current process). +- `watch` polls `runtime` (`"type":"runtime"`) until self idle unless `--idle false`. + Default `follow`/`watch` timeout is 120s. Unknown/unavailable observation fails + rather than completing either command, including nonwaiting watch. +- Auto delivery and waits require the updated backend `observation.availability` + contract (`live` or known `inactive`); a bare phase is not usable evidence. + `runtime.delegation_activity` is separate direct-workflow activity and may remain + live after self idle. Prompt/follow idle results retain the full `runtime`. + Live `delegation` events report workflow lifecycle separately from dispatch + `tool` events; neither changes follow's self-idle completion scope. - `steer --run-id` is optional; if omitted, qmacp pins `runtime.active_run_id`. Missing/empty run IDs fail before submission rather than steering unguarded. - Queued submit returns `clientInputId`. `input_state` events expose the same ID as `inputId`; pass that exact value to `discard-queued`. +- `cancel` returns `cancellationState: "notification_sent"` after the local + WebSocket write/flush. It does not confirm backend acceptance or run termination, + discard queued inputs, or cancel child workflows. Inspect `runtime` separately; + a queued next run can keep the session busy. - Do not close a session unless the user asked to end it or a qmtcode test explicitly requires `close`. After a timeout, inspect `runtime` and session history before resending or changing work. Do not steer, queue, discard, or cancel just because time ran out. Require exit 0 and JSON `ok` not false; prompt/follow also need a terminal -`"type":"done"`. Require JSON `ok: true` for cancel/discard/close. A submission +`"type":"done"`. For cancel, exit 0 and JSON `ok: true` prove only notification +sending; for discard/close they confirm the respective operation. A submission receipt is not proof the agent completed the task. For event-driven waiting outside the agent turn, see [callbacks](callbacks.md). diff --git a/skills/qmacp/sessions.md b/skills/qmacp/sessions.md index 9a11546..50a2a54 100644 --- a/skills/qmacp/sessions.md +++ b/skills/qmacp/sessions.md @@ -36,6 +36,10 @@ creation receipts, opt in to `new --stream`; see [CLI output](cli.md#output). `find` reports `pagesScanned`, `truncated`, `pageCapReached`, and `stopReason` (`exhausted|limit|page_cap`). A truncated empty result is not exhaustive. +Summaries retain full backend `runtime` observation/activity metadata. Missing +phase stays null; unknown/unavailable or unobserved status never becomes idle. +`find --phase idle` matches only usable self-idle observations, even when managed +dependency activity remains live. It does not test task or descendant completion. ## Inspect and configure @@ -52,7 +56,12 @@ qmacp --quiet set-effort "$SESSION_ID" high # auto|low|medium|high|max ``` `status` is mode/model/profile/effort; `runtime` is phase/queue/steerability. -`inspect --tools` defaults true; use `--tools false` for compact history. +Compact `inspect` includes `delegations` from snapshot `delegationUpdates`, separate +from dispatch `tools`, using the same records as live `delegation` events. +`inspect --tools` defaults true; `--tools false` hides only tools, not delegations. +Snapshot records are point observations; a later live record with the same +`sourceEpoch` and higher `sessionSeq` is newer. Historical nonterminal rows do not +prove live activity, and a terminal workflow record does not prove OS process exit. `--full` dumps the raw snapshot (large). Do not treat earlier output as current state; query again before acting. Require exit 0 and JSON `ok: true` for changes. diff --git a/skills/qmacp/testing.md b/skills/qmacp/testing.md index 5db2ba7..80d8867 100644 --- a/skills/qmacp/testing.md +++ b/skills/qmacp/testing.md @@ -27,7 +27,11 @@ requirements; do not alter unrelated sessions or user-run services. - Load/list features: if `caps.acp.loadSession`/`list`/`close` are false, do not keep calling `inspect`/`sessions`/`close`. -Query `runtime` or `status` again rather than treating earlier output as current. -Do not claim cancel/discard/close/config success without exit 0 and JSON `ok: true`. +Query `runtime` again for execution state rather than treating earlier output as +current; `status` is configuration only. Auto delivery and waits require updated +backend observation metadata. Unknown/unavailable is not idle or task success. +Require exit 0 and JSON `ok: true` for command success. Cancel's +`cancellationState: "notification_sent"` proves only notification write/flush; +check runtime/outcome evidence separately for termination. Prompt/follow checks also require a terminal `"type":"done"`, but a receipt or idle state alone is not semantic proof of the task result. Inspect the relevant result. diff --git a/src/client.rs b/src/client.rs index baed070..f5543b0 100644 --- a/src/client.rs +++ b/src/client.rs @@ -585,6 +585,14 @@ async fn handle_inbound( event_sink.emit(event).await; } + if matches!( + method.as_str(), + "querymt/session/delegationUpdate" | "_querymt/session/delegationUpdate" + ) && let Some(event) = events::compact_delegation_update(¶ms) + { + event_sink.emit(event).await; + } + if let Some(id) = value.get("id").cloned() { let (reply, event) = inbound_request_reply(id, &method, ¶ms, permission)?; if let Some(event) = event { diff --git a/src/commands.rs b/src/commands.rs index 3e0d31f..867e063 100644 --- a/src/commands.rs +++ b/src/commands.rs @@ -186,7 +186,7 @@ pub enum Command { /// How to send text to an existing session. #[arg(long, value_enum, default_value_t = Delivery::Auto)] delivery: Delivery, - /// Wait until runtime is idle, including queued turns. Default true. + /// Wait for self-session idle, including queued turns, not child/process completion. #[arg(long, default_value_t = true, action = clap::ArgAction::Set)] until_idle: bool, /// `SESSION_ID TEXT` or, with --new, just `TEXT`. Use `-` to read stdin. @@ -195,7 +195,7 @@ pub enum Command { }, /// Current QueryMT runtime state for a session. Runtime { session_id: String }, - /// Stream compact session events until idle. + /// Stream compact session events until self-session idle, not task success. Follow { session_id: String, #[arg(long, default_value_t = 1)] @@ -203,7 +203,7 @@ pub enum Command { #[arg(long, default_value_t = 120)] timeout: u64, }, - /// Poll runtime state until idle or timeout. + /// Poll runtime state until self-session idle or timeout, not child/process completion. Watch { session_id: String, #[arg(long, default_value_t = 1)] @@ -239,7 +239,7 @@ pub enum Command { #[arg(trailing_var_arg = true, allow_hyphen_values = true)] commands: Vec, }, - /// Cancel the current turn for a session. + /// Send a cancellation request for the current turn; does not confirm termination. Cancel { session_id: String }, /// Close a materialized session. Close { session_id: String }, @@ -1086,9 +1086,12 @@ async fn prompt(client: &AcpClient, args: PromptArgs) -> Result Result Result Result<&str, CliError> { + session::observed_phase(&state["runtime"]).ok_or_else(|| { + CliError::rpc( + "session runtime observation is unavailable or invalid; execution state is unknown", + ) + .with_recovery(json!({ + "sessionId": state["sessionId"], "stage": "observe", + "observationState": "unavailable", "runtime": state["runtime"], + })) + }) +} + fn delivery_from_runtime(phase: &str, steerable: bool) -> Delivery { if phase == "idle" { Delivery::Prompt @@ -1166,7 +1181,7 @@ async fn wait_until_idle( client: &AcpClient, session_id: &str, timeout: Duration, -) -> Result<(), CliError> { +) -> Result { wait_until_idle_every(client, session_id, timeout, Duration::from_millis(400)).await } @@ -1175,22 +1190,25 @@ async fn wait_until_idle_every( session_id: &str, timeout: Duration, interval: Duration, -) -> Result<(), CliError> { - let mut phase = "unknown".to_string(); +) -> Result { + let mut last_runtime = Value::Null; tokio::time::timeout(timeout.max(Duration::from_secs(1)), async { loop { - let value = runtime(client, session_id.to_string()).await?; - phase = value.pointer("/runtime/phase") - .and_then(Value::as_str).unwrap_or("unknown").to_string(); - if phase == "idle" { - return Ok(()); + let value = runtime(client, session_id.to_string()).await + .map_err(|error| error.with_recovery(json!({"runtime": last_runtime})))?; + last_runtime = value["runtime"].clone(); + if require_observed_phase(&value)? == "idle" { + return Ok(value); } tokio::time::sleep(interval.max(Duration::from_millis(200))).await; } }).await.map_err(|_| CliError::Timeout(format!( - "timed out after {}s waiting for session {session_id} to become idle (last phase: {phase}); remote work was not cancelled", - timeout.as_secs(), - )))? + "timed out after {}s waiting for session {session_id} to become idle (last phase: {}); remote work was not cancelled", + timeout.as_secs(), last_runtime["phase"].as_str().unwrap_or("unknown"), + )).with_recovery(json!({ + "sessionId": session_id, "stage": "observe", "observationState": "unavailable", + "runtime": last_runtime, + })))? } async fn follow( @@ -1202,21 +1220,34 @@ async fn follow( ) -> Result { client.initialize().await.map_err(CliError::from_request)?; let cwd = session::resolve_cwd(None).map_err(CliError::rpc)?; - let _ = open_session(client, session_id.clone(), cwd).await; + open_session(client, session_id.clone(), cwd) + .await + .map_err(|error| { + error.with_recovery(json!({ + "sessionId": session_id, "stage": "open", "observationState": "unavailable", + })) + })?; let _ = client.take_assistant_text().await; - wait_until_idle_every( + let observed = wait_until_idle_every( client, &session_id, Duration::from_secs(timeout.max(1)), Duration::from_secs(interval.max(1)), ) - .await?; + .await + .map_err(|error| { + error.with_recovery(json!({ + "sessionId": session_id, "stage": "observe", "observationState": "unavailable", + })) + })?; let text = client.take_assistant_text().await; let done = json!({ "type": "done", "ok": true, "sessionId": session_id, "stopReason": "idle", + "observationState": "idle", + "runtime": observed["runtime"], "text": text, "interval": interval, }); @@ -1229,14 +1260,20 @@ async fn follow( } async fn runtime(client: &AcpClient, session_id: String) -> Result { - client.initialize().await.map_err(CliError::from_request)?; + let recovery = json!({ + "sessionId": session_id, "stage": "observe", "observationState": "unavailable", + }); + client + .initialize() + .await + .map_err(|error| CliError::from_request(error).with_recovery(recovery.clone()))?; let response = client .extension( "querymt/session/runtimeState", json!({ "session_id": session_id }), ) .await - .map_err(CliError::from_request)?; + .map_err(|error| CliError::from_request(error).with_recovery(recovery))?; Ok(json!({ "sessionId": session_id, "runtime": extension_payload(&response), @@ -1251,32 +1288,32 @@ async fn watch( wait_idle: bool, pretty: bool, ) -> Result { - let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout.max(1)); let interval = Duration::from_secs(interval.max(1)); - loop { - let value = runtime(client, session_id.clone()).await?; - let phase = value - .pointer("/runtime/phase") - .and_then(Value::as_str) - .unwrap_or("unknown") - .to_string(); - output::write_event(&json!({ - "type": "runtime", - "sessionId": session_id, - "phase": phase, - "runtime": value.get("runtime"), - })) - .map_err(CliError::rpc)?; - if !wait_idle || phase == "idle" { - return write_ok(pretty, value); - } - if tokio::time::Instant::now() >= deadline { - return Err(CliError::Timeout(format!( - "watch timed out after {timeout}s while session {session_id} was {phase}" - ))); + let mut last_runtime = Value::Null; + tokio::time::timeout(Duration::from_secs(timeout.max(1)), async { + loop { + let value = runtime(client, session_id.clone()).await + .map_err(|error| error.with_recovery(json!({"runtime": last_runtime})))?; + last_runtime = value["runtime"].clone(); + output::write_event(&json!({ + "type": "runtime", + "sessionId": session_id, + "phase": session::observed_phase(&last_runtime).unwrap_or("unknown"), + "runtime": last_runtime, + })) + .map_err(CliError::rpc)?; + let phase = require_observed_phase(&value)?; + if !wait_idle || phase == "idle" { + return write_ok(pretty, value); + } + tokio::time::sleep(interval).await; } - tokio::time::sleep(interval).await; - } + }).await.map_err(|_| CliError::Timeout(format!( + "watch timed out after {timeout}s waiting for session {session_id}; remote work was not cancelled" + )).with_recovery(json!({ + "sessionId": session_id, "stage": "observe", "observationState": "unavailable", + "runtime": last_runtime, + })))? } async fn submit_input( @@ -1499,7 +1536,7 @@ async fn cancel(client: &AcpClient, session_id: String) -> Result Option { Some(Value::Object(event)) } +/// Delegation workflow state is independent of its dispatch tool's status. +pub fn compact_delegation_update(params: &Value) -> Option { + let fields = params.as_object()?; + for key in ["sessionId", "delegationId", "state"] { + fields.get(key)?.as_str()?; + } + let mut event = fields.clone(); + event.insert("type".to_string(), json!("delegation")); + Some(Value::Object(event)) +} + pub fn compact_session_update(session_id: &Value, update: &Value) -> Option { let kind = update .get("sessionUpdate") diff --git a/src/session.rs b/src/session.rs index bd92b90..4fdc5a8 100644 --- a/src/session.rs +++ b/src/session.rs @@ -4,6 +4,8 @@ use agent_client_protocol::schema::v1 as acp; use anyhow::{Context, Result, bail}; use serde_json::{Map, Value, json}; +use crate::events; + const SESSION_LOAD_SNAPSHOT_META_KEY: &str = "querymt/sessionLoadSnapshot.v1"; const INSPECT_TEXT_LIMIT: usize = 4_000; const FIND_PAGE_CAP: usize = 20; @@ -33,6 +35,29 @@ pub fn profile_meta(profile_id: &str) -> Map { meta } +/// A usable self-execution phase requires an available backend observation. +pub fn observed_phase(runtime: &Value) -> Option<&str> { + let phase = runtime.get("phase")?.as_str()?; + match runtime.pointer("/observation/availability")?.as_str()? { + "live" + if matches!( + phase, + "idle" + | "starting" + | "model" + | "tools" + | "waiting" + | "closing" + | "cancel_requested" + ) => + { + Some(phase) + } + "inactive" if phase == "idle" => Some(phase), + _ => None, + } +} + pub fn session_summary(session: &acp::SessionInfo) -> Value { compact_session_summary(session) } @@ -44,12 +69,16 @@ pub fn compact_session_summary(session: &acp::SessionInfo) -> Value { .and_then(|meta| serde_json::to_value(meta).ok()) .unwrap_or(Value::Null); let runtime = meta.get("runtimeStatus").cloned().unwrap_or(Value::Null); + let phase = runtime + .get("phase") + .filter(|phase| !phase.is_null()) + .map(|_| observed_phase(&runtime).unwrap_or("unknown")); let mut summary = json!({ "sessionId": session.session_id.to_string(), "cwd": path_string(&session.cwd), "title": session.title, "updatedAt": session.updated_at, - "phase": runtime.get("phase").cloned(), + "phase": phase, "messageCount": meta.get("messageCount").cloned(), "runtime": runtime, }); @@ -172,10 +201,21 @@ pub fn compact_inspect( } else { Value::Array(Vec::new()) }, + "delegations": compact_delegations(snapshot), "errors": compact_errors(snapshot), }) } +fn compact_delegations(snapshot: Option<&Value>) -> Vec { + snapshot + .and_then(|snapshot| snapshot.get("delegationUpdates")) + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(events::compact_delegation_update) + .collect() +} + fn session_load_snapshot(response: &Value) -> Option<&Value> { response .get("_meta") @@ -287,12 +327,7 @@ pub fn matches_query_and_phase(session: &Value, query: &str, phase: Option<&str> let current = session .get("phase") .and_then(Value::as_str) - .or_else(|| { - session - .get("runtime") - .and_then(|runtime| runtime.get("phase")) - .and_then(Value::as_str) - }) + .or_else(|| session.get("runtime").and_then(observed_phase)) .unwrap_or(""); if !current.eq_ignore_ascii_case(phase) { return false; diff --git a/tests/agent_workflows.rs b/tests/agent_workflows.rs index b85e2ae..409c211 100644 --- a/tests/agent_workflows.rs +++ b/tests/agent_workflows.rs @@ -13,6 +13,7 @@ use tokio_tungstenite::{accept_async, tungstenite::Message}; enum Reply { Result(Value), + WithNotifications(Vec, Value), Error(i64, &'static str), Disconnect, Pending, @@ -49,6 +50,15 @@ impl MockServer { Reply::Result(result) => json!({ "jsonrpc": "2.0", "id": request["id"], "result": result, }), + Reply::WithNotifications(notifications, result) => { + for notification in notifications { + socket + .send(Message::Text(notification.to_string().into())) + .await + .unwrap(); + } + json!({"jsonrpc": "2.0", "id": request["id"], "result": result}) + } Reply::Error(code, message) => json!({ "jsonrpc": "2.0", "id": request["id"], "error": {"code": code, "message": message}, @@ -115,7 +125,10 @@ fn events(output: &Output) -> Vec { } fn runtime(phase: &str, steerable: bool) -> Reply { - Reply::Result(json!({"phase": phase, "steerable": steerable, "active_run_id": "run-1"})) + Reply::Result(json!({ + "phase": phase, "steerable": steerable, "active_run_id": "run-1", + "observation": {"availability": "live", "epoch": "actor-1", "revision": 1}, + })) } fn accepted(request: &Value) -> Reply { @@ -199,6 +212,14 @@ async fn nonwaiting_prompts_emit_receipts_without_resume_load_or_prompt_rpcs() { let done = events.last().unwrap(); assert_eq!(done["type"], "done"); assert_eq!(done["observationState"], "not_waited"); + assert_eq!( + done["stopReason"], + if method == "steer" { + "steered" + } else { + "queued" + } + ); assert_eq!(receipt["clientInputId"], done["clientInputId"]); let requests = server.finish().await; assert_eq!(requests.last().unwrap()["params"]["prompt"][0]["text"], "-"); @@ -723,6 +744,639 @@ async fn idle_observation_does_not_replay_history_or_claim_task_completion() { server.finish().await; } +fn delegation(state: &str, seq: u64) -> Value { + json!({ + "version": 1, "sessionId": "parent", "delegationId": "delegation-1", + "toolCallId": "dispatch-1", "childSessionId": "child-1", "targetAgentId": "coder", + "objective": "fixture task", "state": state, "activity": "historical", + "requestedAt": 1, "forkedAt": 2, "updatedAt": seq, + "sourceEpoch": "workflow-1", "sessionSeq": seq, + }) +} + +fn delegation_notification(method: &str, update: &Value) -> Value { + json!({"jsonrpc": "2.0", "method": method, "params": update}) +} + +fn assert_delegation_event(event: &Value, update: &Value) { + assert_eq!(event["type"], "delegation"); + for (key, value) in update.as_object().unwrap() { + assert_eq!(&event[key], value, "lost delegation field {key}"); + } +} + +#[tokio::test] +async fn follow_streams_delegation_lifecycle_separately_from_tool_dispatch_during_load() { + let mut verifying = delegation("forked", 3); + verifying["activity"] = json!("live"); + verifying["stage"] = json!("verifying"); + let mut terminal = delegation("completed", 4); + terminal["finishedAt"] = json!(4); + terminal["resultSummary"] = json!("verified"); + let expected = [verifying.clone(), terminal.clone()]; + let server = MockServer::start(move |request| match request["method"].as_str().unwrap() { + "session/resume" => Reply::Error(-32601, "resume unsupported"), + "session/load" => Reply::WithNotifications(vec![ + json!({"jsonrpc": "2.0", "method": "session/update", "params": { + "sessionId": "parent", "update": {"sessionUpdate": "tool_call_update", "toolCallId": "dispatch-1", "status": "completed"}, + }}), + delegation_notification("querymt/session/delegationUpdate", &verifying), + delegation_notification("querymt/session/delegationUpdate", &terminal), + ], json!({"_meta": {"querymt/sessionLoadSnapshot.v1": {"delegationUpdates": [verifying]}}})), + "_querymt/session/runtimeState" => runtime("idle", false), + _ => unexpected(request), + }).await; + let output = invoke( + cli().args(["--url", &server.url, "--quiet", "follow", "parent"]), + None, + ) + .await; + assert!(output.status.success(), "{output:?}"); + let events = events(&output); + let tool = events.iter().find(|event| event["type"] == "tool").unwrap(); + assert_eq!(tool["id"], "dispatch-1"); + assert_eq!(tool["status"], "completed"); + let updates: Vec<_> = events + .iter() + .filter(|event| event["type"] == "delegation") + .collect(); + assert_eq!(updates.len(), 2, "{events:?}"); + for (event, update) in updates.iter().zip(&expected) { + assert_delegation_event(event, update); + } + assert_eq!(events.last().unwrap()["observationState"], "idle"); + server.finish().await; +} + +#[tokio::test] +async fn exported_event_channel_delivers_terminal_delegation_update_buffered_during_load() { + let historical = delegation("forked", 2); + let mut terminal = delegation("cancelled", 4); + terminal["finishedAt"] = json!(4); + terminal["error"] = json!("fixture cancellation"); + let expected = terminal.clone(); + let server = MockServer::start(move |request| { + assert_eq!(request["method"], "session/load"); + Reply::WithNotifications( + vec![delegation_notification("_querymt/session/delegationUpdate", &terminal)], + json!({"_meta": {"querymt/sessionLoadSnapshot.v1": {"delegationUpdates": [historical]}}}), + ) + }).await; + let (tx, mut rx) = tokio::sync::mpsc::channel(8); + let client = qmacp::client::AcpClient::connect_with_events( + &server.url, + "fixture", + qmacp::policy::PermissionPolicy::Cancel, + tx, + ) + .await + .unwrap(); + client.initialize().await.unwrap(); + let (_, raw) = client + .load_session("parent".into(), "/tmp".into()) + .await + .unwrap(); + assert_eq!( + raw["_meta"]["querymt/sessionLoadSnapshot.v1"]["delegationUpdates"][0]["state"], + "forked" + ); + let event = tokio::time::timeout(Duration::from_secs(1), rx.recv()) + .await + .unwrap() + .unwrap(); + assert_delegation_event(&event, &expected); + drop(client); + server.finish().await; +} + +fn enriched_remote_load(update: &Value) -> Value { + let activity = if update["activity"] == "live" { + vec![update.clone()] + } else { + Vec::new() + }; + json!({"_meta": { + "location": "remote", "nodeId": "fixture-peer", + "querymt/sessionLoadSnapshot.v1": { + "runtimeStatus": { + "phase": "idle", "steerable": false, "active_run_id": null, + "observation": {"availability": "live", "epoch": "remote-actor", "revision": 9}, + "delegation_activity": activity, + }, + "delegationUpdates": [update], + "audit": {"events": [{"kind": {"type": "tool_call_end", "data": { + "tool_call_id": "dispatch-1", "tool_name": "delegate", "is_error": false, "result": "dispatched", + }}}]}, + }, + }}) +} + +#[tokio::test] +async fn exported_load_preserves_enriched_initial_snapshot_without_notifications() { + let mut verifying = delegation("forked", 3); + verifying["activity"] = json!("live"); + verifying["stage"] = json!("verifying"); + let expected = enriched_remote_load(&verifying); + let response = expected.clone(); + let server = MockServer::start(move |request| { + assert_eq!(request["method"], "session/load"); + Reply::Result(response.clone()) + }) + .await; + let (tx, mut rx) = tokio::sync::mpsc::channel(8); + let client = qmacp::client::AcpClient::connect_with_events( + &server.url, + "fixture", + qmacp::policy::PermissionPolicy::Cancel, + tx, + ) + .await + .unwrap(); + client.initialize().await.unwrap(); + let (loaded, raw) = tokio::time::timeout( + Duration::from_secs(5), + client.load_session("parent".into(), "/tmp".into()), + ) + .await + .unwrap() + .unwrap(); + assert!(loaded.meta.is_some()); + assert_eq!(raw, expected); + assert!(matches!( + rx.try_recv(), + Err(tokio::sync::mpsc::error::TryRecvError::Empty) + )); + drop(client); + let requests = server.finish().await; + assert_eq!( + requests.len(), + 2, + "only initialize and load, no runtime query" + ); +} + +#[tokio::test] +async fn inspect_keeps_initial_remote_delegation_separate_from_dispatch_on_reconnect() { + for state in ["forked", "failed"] { + let mut update = delegation(state, if state == "forked" { 3 } else { 4 }); + if state == "forked" { + update["activity"] = json!("live"); + update["stage"] = json!("verifying"); + } else { + update["finishedAt"] = json!(4); + update["error"] = json!("verification failed"); + } + let expected = enriched_remote_load(&update); + for full in [false, true] { + let response = expected.clone(); + let server = MockServer::start(move |request| { + assert_eq!(request["method"], "session/load"); + Reply::Result(response.clone()) + }) + .await; + let mut command = cli(); + command.args(["--url", &server.url, "--quiet", "inspect", "parent"]); + if full { + command.arg("--full"); + } + let output = invoke(&mut command, None).await; + assert!(output.status.success(), "{output:?}"); + let result_events = events(&output); + assert_eq!( + result_events.len(), + 1, + "snapshot only, no live notifications" + ); + let result = &result_events[0]; + if full { + assert_eq!(result["raw"], expected); + } else { + assert_eq!(result["tools"][0]["id"], "dispatch-1"); + assert_eq!(result["tools"][0]["status"], "ok"); + let updates = result["delegations"] + .as_array() + .expect("compact delegation snapshot"); + assert_eq!(updates.len(), 1); + assert_delegation_event(&updates[0], &update); + } + assert_eq!(server.finish().await.len(), 2, "no catchup runtime query"); + } + } +} + +#[tokio::test] +async fn unavailable_runtime_prevents_auto_delivery() { + for state in [ + json!({"phase": "unknown", "steerable": false, "observation": {"availability": "unavailable", "reason": "owner_unavailable"}}), + json!({"phase": "idle", "steerable": false, "observation": {"availability": "unavailable"}}), + json!({"phase": "idle", "steerable": false}), + ] { + let server = MockServer::start(move |request| match request["method"].as_str().unwrap() { + "_querymt/session/runtimeState" => Reply::Result(state.clone()), + "session/resume" | "session/load" => Reply::Result(json!({})), + "session/prompt" => Reply::Result(json!({"stopReason": "end_turn"})), + "_querymt/session/queue" => accepted(request), + _ => unexpected(request), + }) + .await; + let output = invoke( + cli().args([ + "--url", + &server.url, + "--quiet", + "prompt", + "--until-idle", + "false", + "target", + "work", + ]), + None, + ) + .await; + assert_eq!(output.status.code(), Some(3), "{output:?}"); + let error = events(&output).pop().unwrap(); + assert_eq!(error["recovery"]["deliveryState"], "not_submitted"); + assert_eq!(error["recovery"]["observationState"], "unavailable"); + assert_eq!(server.finish().await.len(), 2, "must not open or submit"); + } +} + +#[tokio::test] +async fn unavailable_observation_fails_follow_and_watch_instead_of_reporting_completion() { + for command in ["follow", "watch"] { + let state = json!({"phase": "unknown", "steerable": false, "observation": { + "availability": "unavailable", "reason": "owner_unavailable", + }}); + let expected = state.clone(); + let server = MockServer::start(move |request| match request["method"].as_str().unwrap() { + "session/resume" | "session/load" => Reply::Result(json!({})), + "_querymt/session/runtimeState" => Reply::Result(state.clone()), + _ => unexpected(request), + }) + .await; + let mut cli = cli(); + cli.args([ + "--url", + &server.url, + "--quiet", + command, + "target", + "--timeout", + "1", + ]); + if command == "watch" { + cli.args(["--idle", "false"]); + } + let output = invoke(&mut cli, None).await; + assert_eq!(output.status.code(), Some(3), "{output:?}"); + let events = events(&output); + assert!(events.iter().all(|event| event["type"] != "done")); + let recovery = &events.last().unwrap()["recovery"]; + assert_eq!(recovery["sessionId"], "target"); + assert_eq!(recovery["stage"], "observe"); + assert_eq!(recovery["observationState"], "unavailable"); + assert_eq!(recovery["runtime"], expected); + server.finish().await; + } +} + +#[tokio::test] +async fn self_idle_waits_preserve_dependency_activity_without_waiting_for_children() { + for command in ["prompt", "follow", "watch"] { + let state = json!({ + "phase": "idle", "steerable": false, "active_run_id": null, + "observation": {"availability": "inactive"}, + "delegation_activity": [{"delegationId": "delegate-1", "stage": "verifying", "activity": "live", "sourceEpoch": "workflow-epoch", "sessionSeq": 4}], + }); + let expected = state.clone(); + let server = MockServer::start(move |request| match request["method"].as_str().unwrap() { + "session/resume" | "session/load" => Reply::Result(json!({})), + "_querymt/session/queue" => accepted(request), + "_querymt/session/runtimeState" => Reply::Result(state.clone()), + _ => unexpected(request), + }) + .await; + let mut cli = cli(); + cli.args(["--url", &server.url, "--quiet", command]); + if command == "prompt" { + cli.args(["--delivery", "queue", "--timeout", "1", "target", "work"]); + } else { + cli.args(["target", "--timeout", "1"]); + } + let output = invoke(&mut cli, None).await; + assert!(output.status.success(), "{output:?}"); + let done = events(&output).pop().unwrap(); + assert_eq!(done["runtime"], expected); + if command != "watch" { + assert_eq!(done["observationState"], "idle"); + } + server.finish().await; + } +} + +#[tokio::test] +async fn watch_timeout_bounds_a_stalled_runtime_request() { + let server = MockServer::start(|request| { + assert_eq!(request["method"], "_querymt/session/runtimeState"); + Reply::Pending + }) + .await; + let output = invoke( + cli().args([ + "--url", + &server.url, + "--quiet", + "watch", + "target", + "--timeout", + "1", + ]), + None, + ) + .await; + assert_eq!(output.status.code(), Some(4), "{output:?}"); + assert_eq!( + events(&output).last().unwrap()["recovery"]["sessionId"], + "target" + ); + server.finish().await; +} + +#[tokio::test] +async fn accepted_prompt_observation_disconnect_preserves_receipt_without_replaying() { + let mut observed = false; + let server = MockServer::start(move |request| match request["method"].as_str().unwrap() { + "_querymt/session/queue" => accepted(request), + "_querymt/session/runtimeState" if !observed => { + observed = true; + runtime("tools", false) + } + "_querymt/session/runtimeState" => Reply::Disconnect, + _ => unexpected(request), + }) + .await; + let output = invoke( + cli().args([ + "--url", + &server.url, + "--quiet", + "prompt", + "--delivery", + "queue", + "target", + "work", + ]), + None, + ) + .await; + assert_eq!(output.status.code(), Some(2), "{output:?}"); + let events = events(&output); + let submitted = events + .iter() + .find(|event| event["type"] == "submitted") + .unwrap(); + let recovery = &events.last().unwrap()["recovery"]; + assert_eq!(recovery["deliveryState"], "accepted"); + assert_eq!(recovery["clientInputId"], submitted["clientInputId"]); + assert_eq!(recovery["receipt"], submitted["receipt"]); + assert_eq!(recovery["observationState"], "unavailable"); + assert_eq!(recovery["runtime"]["active_run_id"], "run-1"); + assert_eq!(recovery["runtime"]["phase"], "tools"); + assert!(events.iter().all(|event| event["type"] != "done")); + assert_eq!(server.finish().await.len(), 4); +} + +#[tokio::test] +async fn accepted_queue_and_steer_recovery_does_not_claim_prompt_outcome() { + for delivery in ["queue", "steer"] { + let mut submitted = false; + let server = MockServer::start(move |request| match request["method"].as_str().unwrap() { + "_querymt/session/queue" | "_querymt/session/steer" => { + submitted = true; + accepted(request) + } + "_querymt/session/runtimeState" if !submitted => runtime("tools", true), + "_querymt/session/runtimeState" => Reply::Result(json!({ + "phase": "unknown", "observation": {"availability": "unavailable", "reason": "owner_unavailable"}, + })), + _ => unexpected(request), + }).await; + let output = invoke( + cli().args([ + "--url", + &server.url, + "--quiet", + "prompt", + "--delivery", + delivery, + "--timeout", + "1", + "target", + "work", + ]), + None, + ) + .await; + assert_eq!(output.status.code(), Some(3), "{output:?}"); + let events = events(&output); + let submitted = events + .iter() + .find(|event| event["type"] == "submitted") + .unwrap(); + let recovery = &events.last().unwrap()["recovery"]; + assert_eq!(recovery["sessionId"], "target"); + assert_eq!(recovery["stage"], "observe"); + assert_eq!(recovery["delivery"], delivery); + assert_eq!(recovery["deliveryState"], "accepted"); + assert_eq!(recovery["clientInputId"], submitted["clientInputId"]); + assert_eq!(recovery["receipt"], submitted["receipt"]); + assert_eq!(recovery["observationState"], "unavailable"); + assert_eq!( + recovery["runtime"]["observation"]["reason"], + "owner_unavailable" + ); + assert!(events.iter().all(|event| event["type"] != "done")); + assert!( + recovery.get("stopReason").is_none(), + "{delivery}: {recovery}" + ); + let requests = server.finish().await; + assert!( + requests + .iter() + .all(|request| request["method"] != "session/prompt") + ); + assert_eq!(requests.len(), if delivery == "steer" { 4 } else { 3 }); + } +} + +#[tokio::test] +async fn returned_prompt_keeps_its_stop_reason_when_later_observation_fails() { + let server = MockServer::start(|request| match request["method"].as_str().unwrap() { + "session/new" => Reply::Result(json!({"sessionId": "created-session"})), + "session/prompt" => Reply::Result(json!({"stopReason": "cancelled"})), + "_querymt/session/runtimeState" => Reply::Result(json!({ + "phase": "unknown", "observation": {"availability": "unavailable"}, + })), + _ => unexpected(request), + }) + .await; + let output = invoke( + cli().args(["--url", &server.url, "--quiet", "prompt", "--new", "work"]), + None, + ) + .await; + assert_eq!(output.status.code(), Some(3), "{output:?}"); + let events = events(&output); + let recovery = &events.last().unwrap()["recovery"]; + assert_eq!(recovery["sessionId"], "created-session"); + assert_eq!(recovery["creationState"], "created"); + assert_eq!(recovery["deliveryState"], "accepted"); + assert_eq!(recovery["stopReason"], "cancelled"); + assert_eq!(recovery["observationState"], "unavailable"); + assert!(events.iter().all(|event| event["type"] != "done")); + server.finish().await; +} + +#[tokio::test] +async fn discovery_preserves_unknown_observations_and_filters_only_known_phases() { + let remote_runtime = json!({ + "phase": "tools", "active_run_id": "remote-run", + "observation": {"availability": "live", "epoch": "remote-actor", "revision": 9}, + }); + let listed = vec![ + json!({"sessionId": "missing", "cwd": "/tmp"}), + json!({"sessionId": "offline", "cwd": "/tmp", "_meta": {"runtimeStatus": { + "phase": "unknown", "observation": {"availability": "unavailable", "reason": "owner_unavailable"}, + }}}), + json!({"sessionId": "unobserved", "cwd": "/tmp", "_meta": {"runtimeStatus": {"phase": "idle"}}}), + json!({"sessionId": "parent", "cwd": "/tmp", "_meta": {"runtimeStatus": { + "phase": "idle", "observation": {"availability": "inactive"}, + "delegation_activity": [{"stage": "verifying", "activity": "live"}], + }}}), + json!({"sessionId": "busy", "cwd": "/tmp", "_meta": {"runtimeStatus": { + "phase": "tools", "active_run_id": "run-B", "observation": {"availability": "live", "epoch": "actor-1", "revision": 8}, + }}}), + json!({"sessionId": "null-phase", "cwd": "/tmp", "_meta": {"runtimeStatus": {"phase": null}}}), + json!({"sessionId": "remote-busy", "cwd": "/tmp", "_meta": {"runtimeStatus": remote_runtime}}), + ]; + for phase in [None, Some("idle"), Some("unknown"), Some("tools")] { + let listed = listed.clone(); + let server = MockServer::start(move |request| { + assert_eq!(request["method"], "session/list"); + Reply::Result(json!({"sessions": listed})) + }) + .await; + let mut cli = cli(); + cli.args(["--url", &server.url, "--quiet"]); + if let Some(phase) = phase { + cli.args(["find", "--phase", phase]); + } else { + cli.arg("sessions"); + } + let output = invoke(&mut cli, None).await; + assert!(output.status.success(), "{output:?}"); + let result = events(&output).pop().unwrap(); + let sessions = result["sessions"].as_array().unwrap(); + let ids: Vec<_> = sessions + .iter() + .map(|session| session["sessionId"].as_str().unwrap()) + .collect(); + match phase { + Some("idle") => assert_eq!(ids, ["parent"]), + Some("unknown") => assert_eq!(ids, ["offline", "unobserved"]), + Some("tools") => assert_eq!(ids, ["busy", "remote-busy"]), + None => { + assert!(sessions[0]["phase"].is_null()); + assert_eq!( + sessions[1]["runtime"]["observation"]["availability"], + "unavailable" + ); + assert_eq!( + sessions[3]["runtime"]["delegation_activity"][0]["stage"], + "verifying" + ); + assert_eq!(sessions[4]["runtime"]["active_run_id"], "run-B"); + assert_eq!(sessions[4]["runtime"]["observation"]["revision"], 8); + assert!(sessions[5]["phase"].is_null()); + assert_eq!(sessions[5]["runtime"], json!({"phase": null})); + assert_eq!(sessions[6]["phase"], "tools"); + assert!(sessions[6]["messageCount"].is_null()); + assert_eq!(sessions[6]["runtime"], remote_runtime); + } + _ => unreachable!(), + } + assert_eq!( + server.finish().await.len(), + 2, + "listing must not issue per-session runtime queries" + ); + } +} + +#[tokio::test] +async fn cancel_reports_only_a_sent_notification_without_waiting_for_termination() { + let server = MockServer::start(|request| { + assert_eq!(request["method"], "session/cancel"); + assert!( + request.get("id").is_none(), + "cancel must remain a notification" + ); + assert_eq!(request["params"], json!({"sessionId": "target"})); + Reply::Pending + }) + .await; + let output = invoke( + cli().args(["--url", &server.url, "--quiet", "cancel", "target"]), + None, + ) + .await; + assert!(output.status.success(), "{output:?}"); + let result = events(&output).pop().unwrap(); + assert_eq!(result["cancellationState"], "notification_sent"); + assert_eq!(result["sessionId"], "target"); + assert_eq!(result["ok"], true); + assert!(result.get("cancelled").is_none()); + assert_eq!(server.finish().await.len(), 2); +} + +#[tokio::test] +async fn follow_does_not_report_done_when_opening_the_session_fails() { + let server = MockServer::start(|request| match request["method"].as_str().unwrap() { + "session/resume" => Reply::Error(-32601, "resume unsupported"), + "session/load" => Reply::Error(-32603, "session observation failed"), + "_querymt/session/runtimeState" => runtime("idle", false), + _ => unexpected(request), + }) + .await; + let output = invoke( + cli().args([ + "--url", + &server.url, + "--quiet", + "follow", + "target", + "--timeout", + "1", + ]), + None, + ) + .await; + assert_eq!(output.status.code(), Some(3), "{output:?}"); + let events = events(&output); + assert!(events.iter().all(|event| event["type"] != "done")); + let recovery = &events.last().unwrap()["recovery"]; + assert_eq!(recovery["sessionId"], "target"); + assert_eq!(recovery["stage"], "open"); + assert_eq!(recovery["observationState"], "unavailable"); + let requests = server.finish().await; + assert!( + requests + .iter() + .all(|request| request["method"] != "_querymt/session/runtimeState") + ); +} + #[tokio::test] async fn find_distinguishes_the_result_limit_from_the_page_cap() { let cap = qmacp::session::find_page_cap();