acpmux vault: refuse clearly on an older daemon, bound output reads by the deadline, status reports the daemon's store

cmux-tui/crates/acpmux/spec/acpmux-protocol.json

@@ -1699,6 +1699,12 @@
       "params": true,
       "result": true
     },
+    {
+      "kind": "request",
+      "name": "_acpmux/vault_status",
+      "params": true,
+      "result": true
+    },
     {
       "kind": "request",
       "name": "_acpmux/turn_attestation",

cmux-tui/crates/acpmux/src/capabilities.rs

@@ -95,7 +95,7 @@ pub const CAPABILITIES: &[Capability] = &[
     // asks the daemon to pull or forget when the values live only in its memory.
     Capability {
         id: crate::vault::CAPABILITY,
-        wire: &[method::MUX_VAULT_PULL, method::MUX_VAULT_FORGET],
+        wire: &[method::MUX_VAULT_PULL, method::MUX_VAULT_FORGET, method::MUX_VAULT_STATUS],
     },
     Capability {
         id: "machine-consent-v1",

cmux-tui/crates/acpmux/src/cli/host_vault.rs

@@ -25,6 +25,10 @@ use crate::cli::output::print_json;
 use crate::vault::pull::{self, Failure, Outcome};
 use crate::vault::store::{OsStore, SecretStore};
 
+/// The running daemon predates `_acpmux/vault_pull` and the OS store is not
+/// usable here, so nothing can keep the values.
+pub const DAEMON_OUTDATED: &str = "vault.daemon_outdated";
+
 fn failure(failure: Failure) -> anyhow::Error {
     let message = match failure.code.as_str() {
         "vault.not_joined" => "this machine is not joined; run `acpmux machine join` first",
@@ -33,6 +37,9 @@ fn failure(failure: Failure) -> anyhow::Error {
             "no OS secret store (Keychain or Secret Service) here; vault env is not stored in files"
         }
         "vault.busy" => "another vault pull is running",
+        DAEMON_OUTDATED => {
+            "the running acpmux daemon is older than this CLI and cannot keep vault env (this machine has no usable OS secret store); run `acpmux daemon restart`, then pull again"
+        }
         _ => "vault pull failed",
     };
     let message = if failure.code == pull::PARTIAL {
@@ -52,6 +59,11 @@ pub(crate) fn failure_for_test(f: Failure) -> anyhow::Error {
     failure(f)
 }
 
+/// The string at `key` of `v`, or "".
+fn text_at<'a>(v: &'a Value, key: &str) -> &'a str {
+    v.get(key).and_then(Value::as_str).unwrap_or_default()
+}
+
 /// The outcome JSON with `store`.
 fn with_store(mut out: Value, store: &str) -> Value {
     if let Some(map) = out.as_object_mut() {
@@ -78,15 +90,16 @@ fn print_outcome(outcome: &Value, store: &str, json_out: bool) {
         return;
     }
     let names = |key: &str| -> Vec<String> {
-        outcome[key]
-            .as_array()
+        outcome
+            .get(key)
+            .and_then(Value::as_array)
             .into_iter()
             .flatten()
             .filter_map(|v| v.as_str().map(str::to_owned))
             .collect()
     };
     let (env, stored, deleted) = (names("env"), names("stored"), names("deleted"));
-    match outcome["state"].as_str().unwrap_or_default() {
+    match outcome.get("state").and_then(Value::as_str).unwrap_or_default() {
         "not_granted" => {
             println!(
                 "the vault is not shared with this machine (grant it in cmux, then pull again)"
@@ -108,12 +121,8 @@ fn print_outcome(outcome: &Value, store: &str, json_out: bool) {
             if store == "memory" { " [kept in the daemon's memory]" } else { "" }
         ),
     }
-    for failed in outcome["failed"].as_array().into_iter().flatten() {
-        eprintln!(
-            "{}: {}",
-            failed["name"].as_str().unwrap_or_default(),
-            failed["code"].as_str().unwrap_or_default()
-        );
+    for failed in outcome.get("failed").and_then(Value::as_array).into_iter().flatten() {
+        eprintln!("{}: {}", text_at(failed, "name"), text_at(failed, "code"));
     }
 }
 
@@ -142,39 +151,55 @@ pub async fn forget_cmd(json_out: bool) -> Result<()> {
 /// Ask the running daemon to do `method`. With none running (or one too old
 /// to know it), `here` does it in this process when the OS store is usable;
 /// otherwise the daemon is started and asked, since only it can keep the
-/// values (in memory).
+/// values (in memory). A running daemon too old to know `method` is not
+/// asked again: that is [`DAEMON_OUTDATED`], which says to restart it.
 async fn by_daemon<F, Fut>(method: &str, here: F) -> Result<Value>
 where
     F: FnOnce() -> Fut,
     Fut: std::future::Future<Output = Result<Outcome, Failure>>,
 {
     let ask =
         |client: Arc<crate::client::Client>| async move { client.request(method, json!({})).await };
-    if let Ok(client) = crate::daemon::connect(false).await
-        && let Ok(reply) = ask(client).await
-    {
-        return Ok(reply);
+    let mut outdated = false;
+    if let Ok(client) = crate::daemon::connect(false).await {
+        match ask(client).await {
+            Ok(reply) => return Ok(reply),
+            Err(e) => outdated = method_not_found(&e),
+        }
     }
     if tokio::task::spawn_blocking(crate::vault::store::os_usable).await.unwrap_or(false) {
         return Ok(crate::server::vault_reply(OsStore.kind(), &here().await));
     }
+    if outdated {
+        return Err(failure(Failure {
+            code: DAEMON_OUTDATED.into(),
+            deleted: Vec::new(),
+            outcome: None,
+        }));
+    }
     let client = crate::daemon::connect(true).await?;
     ask(client).await
 }
 
+/// Whether the daemon answered that it has no such method (an older daemon).
+pub(crate) fn method_not_found(e: &anyhow::Error) -> bool {
+    e.downcast_ref::<crate::client::DaemonError>().is_some_and(|d| d.0.code == -32601)
+}
+
 /// Print a pull's or forget's reply (`server/vault.rs`). A failed or partial
 /// one prints its outcome (with `failed`) on stdout when it has one and
 /// exits 1 with the error envelope.
 fn finish(reply: Value, json_out: bool) -> Result<()> {
-    let store = reply["store"].as_str().unwrap_or("unknown").to_owned();
+    let store = reply.get("store").and_then(Value::as_str).unwrap_or("unknown").to_owned();
     if let Some(outcome) = reply.get("outcome").filter(|o| !o.is_null()) {
         print_outcome(outcome, &store, json_out);
     }
     let Some(f) = reply.get("failure").filter(|f| !f.is_null()) else { return Ok(()) };
     Err(failure(Failure {
-        code: f["code"].as_str().unwrap_or("vault.internal").to_owned(),
-        deleted: f["deleted"]
-            .as_array()
+        code: f.get("code").and_then(Value::as_str).unwrap_or("vault.internal").to_owned(),
+        deleted: f
+            .get("deleted")
+            .and_then(Value::as_array)
             .into_iter()
             .flatten()
             .filter_map(|v| v.as_str().map(str::to_owned))
@@ -183,10 +208,19 @@ fn finish(reply: Value, json_out: bool) -> Result<()> {
     }))
 }
 
-/// Where this machine keeps vault values: the OS store's kind, or `memory`
-/// when it cannot take a write here.
-fn store_kind() -> &'static str {
-    if crate::vault::store::os_usable() { OsStore.kind() } else { "memory" }
+/// Where this machine keeps vault values: the running daemon's store
+/// (`_acpmux/vault_status`), since it is the one that keeps them. With no
+/// daemon, or one too old to say, what this process finds: the OS store's
+/// kind, or `memory` when it cannot take a write here.
+async fn store_kind() -> String {
+    if let Ok(client) = crate::daemon::connect(false).await
+        && let Ok(reply) = client.request(crate::rpc::method::MUX_VAULT_STATUS, json!({})).await
+        && let Some(store) = reply.get("store").and_then(Value::as_str)
+    {
+        return store.to_owned();
+    }
+    let here = || if crate::vault::store::os_usable() { OsStore.kind() } else { "memory" };
+    tokio::task::spawn_blocking(here).await.unwrap_or("unknown").to_owned()
 }
 
 /// Whether a harness is signed in on this machine: a presence check of its own
@@ -314,8 +348,8 @@ pub async fn status_cmd(json_out: bool) -> Result<()> {
         tokio::task::spawn_blocking(|| (claude_signed_in(claude_keychain_item), codex_signed_in()))
             .await
             .unwrap_or((SignedIn::Unknown, SignedIn::Unknown));
-    let kind = tokio::task::spawn_blocking(store_kind).await.unwrap_or("unknown");
-    let out = status_json(joined, &state, kind, claude, codex);
+    let kind = store_kind().await;
+    let out = status_json(joined, &state, &kind, claude, codex);
     if json_out {
         print_json(&out);
         return Ok(());
@@ -335,11 +369,9 @@ pub async fn status_cmd(json_out: bool) -> Result<()> {
             state.env.keys().cloned().collect::<Vec<_>>().join(", ")
         }
     );
-    for h in out["harnesses"].as_array().into_iter().flatten() {
-        let signed = h["signedIn"].as_str().unwrap_or("unknown");
-        let id = h["harness"].as_str().unwrap_or("");
-        let step = h["signIn"].as_str().unwrap_or("");
-        match signed {
+    for h in out.get("harnesses").and_then(Value::as_array).into_iter().flatten() {
+        let (id, step) = (text_at(h, "harness"), text_at(h, "signIn"));
+        match h.get("signedIn").and_then(Value::as_str).unwrap_or("unknown") {
             "yes" => println!("{id}: signed in"),
             "no" => println!("{id}: not signed in here; {step}"),
             _ => println!("{id}: sign-in not visible here; if it fails, {step}"),

cmux-tui/crates/acpmux/src/protocol/mod.rs

@@ -439,6 +439,7 @@ pub fn registry() -> Vec<MethodDecl> {
         untyped(method::MUX_PAIR_REDEEM, Request),
         untyped(method::MUX_VAULT_PULL, Request),
         untyped(method::MUX_VAULT_FORGET, Request),
+        untyped(method::MUX_VAULT_STATUS, Request),
         untyped(method::MUX_TURN_ATTESTATION, Request),
         // Notifications.
         notification::<events::SessionEvent>(method::MUX_EVENT),

cmux-tui/crates/acpmux/src/rpc.rs

@@ -704,6 +704,7 @@ pub mod method {
     // The person's vault on this machine (`crate::vault`, `server/vault.rs`).
     pub const MUX_VAULT_PULL: &str = "_acpmux/vault_pull";
     pub const MUX_VAULT_FORGET: &str = "_acpmux/vault_forget";
+    pub const MUX_VAULT_STATUS: &str = "_acpmux/vault_status";
     // Notifications from acpmux to attached clients.
     pub const MUX_EVENT: &str = "_acpmux/event";
     /// A persisted session-to-session message, projected separately so chat clients can render

cmux-tui/crates/acpmux/src/server/vault.rs

@@ -1,5 +1,6 @@
 //! `_acpmux/vault_pull` and `_acpmux/vault_forget`: `machine vault pull` and
-//! `forget` done by this daemon, local socket only. A machine with no usable
+//! `forget` done by this daemon, local socket only. `_acpmux/vault_status`
+//! answers `{store}`, the store this daemon keeps the values in. A machine with no usable
 //! OS store keeps the vault env in the daemon's memory (`vault/store.rs`), so
 //! only the daemon can pull or drop it. The reply holds names and codes only,
 //! never a value.
@@ -10,12 +11,15 @@ use super::{Conn, Origin};
 use crate::rpc::{RpcError, method};
 
 pub(super) async fn dispatch(conn: &Conn, m: &str) -> Option<Result<Value, RpcError>> {
-    if !matches!(m, method::MUX_VAULT_PULL | method::MUX_VAULT_FORGET) {
+    if !matches!(m, method::MUX_VAULT_PULL | method::MUX_VAULT_FORGET | method::MUX_VAULT_STATUS) {
         return None;
     }
     if conn.origin != Origin::Local {
         return Some(Err(RpcError::method_not_found(&format!("{m} (local socket only)"))));
     }
+    if m == method::MUX_VAULT_STATUS {
+        return Some(Ok(json!({"store": crate::vault::store_now().await})));
+    }
     let (store, result) = if m == method::MUX_VAULT_PULL {
         crate::vault::pull_now().await
     } else {

cmux-tui/crates/acpmux/src/vault.rs

@@ -112,7 +112,7 @@ pub fn load_values(
     state: &State,
     store: &dyn SecretStore,
 ) -> (Vec<(String, Zeroizing<String>)>, Vec<(String, &'static str)>) {
-    load_values_by(state, store, Instant::now() + LOAD_BUDGET)
+    load_values_by(state, store, store::after(LOAD_BUDGET))
 }
 
 /// The longest one load reads the store for: entries not read by then are
@@ -470,6 +470,12 @@ pub async fn pull_now() -> (&'static str, Result<pull::Outcome, pull::Failure>)
     (kind, result)
 }
 
+/// The store this daemon keeps vault values in (`_acpmux/vault_status`):
+/// the OS store's kind, or `memory`.
+pub async fn store_now() -> &'static str {
+    tokio::task::spawn_blocking(|| store::for_daemon().kind()).await.unwrap_or("unknown")
+}
+
 /// `machine vault forget` asked of this daemon (`_acpmux/vault_forget`).
 pub async fn forget_now() -> (&'static str, Result<pull::Outcome, pull::Failure>) {
     let Ok(store) = tokio::task::spawn_blocking(store::for_daemon).await else {

cmux-tui/crates/acpmux/src/vault/store.rs

@@ -210,7 +210,7 @@ impl SecretStore for OsStore {
     }
 
     fn get(&self, name: &str) -> Result<Option<Zeroizing<String>>, &'static str> {
-        self.get_by(name, std::time::Instant::now() + TIMEOUT)
+        self.get_by(name, after(TIMEOUT))
     }
 
     fn get_by(
@@ -296,38 +296,55 @@ pub fn capture(
     cmd: Command,
     from_stderr: bool,
 ) -> Result<(std::process::ExitStatus, Zeroizing<Vec<u8>>), &'static str> {
-    capture_until(cmd, from_stderr, std::time::Instant::now() + TIMEOUT)
+    capture_until(cmd, from_stderr, after(TIMEOUT))
+}
+
+/// `now + wait`, or now if that would not fit an `Instant`.
+pub fn after(wait: std::time::Duration) -> std::time::Instant {
+    let now = std::time::Instant::now();
+    now.checked_add(wait).unwrap_or(now)
 }
 
 /// [`capture`], killing the command at `deadline` (at most the store timeout).
+/// The output is read by the deadline too: a process the command started may
+/// hold the pipe open after the command itself exits (`sh -c 'sleep 20 &
+/// exit 0'`), and then the read ends at the deadline as a timeout.
 pub fn capture_until(
     mut cmd: Command,
     from_stderr: bool,
     deadline: std::time::Instant,
 ) -> Result<(std::process::ExitStatus, Zeroizing<Vec<u8>>), &'static str> {
     let wait = deadline.saturating_duration_since(std::time::Instant::now()).min(TIMEOUT);
+    let end = after(wait);
     cmd.stdin(Stdio::null());
     let mut child = cmd.spawn().map_err(|_| UNAVAILABLE)?;
     let pipe: Option<Box<dyn std::io::Read + Send>> = if from_stderr {
         child.stderr.take().map(|p| Box::new(p) as Box<dyn std::io::Read + Send>)
     } else {
         child.stdout.take().map(|p| Box::new(p) as Box<dyn std::io::Read + Send>)
     };
-    let reader = std::thread::spawn(move || drain(pipe));
+    // The reader is never joined: a process the command started may still
+    // hold the pipe open; the reader ends when it closes, and the caller has
+    // returned at the deadline by then.
+    let (sent, read) = std::sync::mpsc::sync_channel(1);
+    std::thread::spawn(move || {
+        let _ = sent.send(drain(pipe));
+    });
     let status = match child.wait_timeout(wait) {
         Ok(Some(status)) => status,
         _ => {
             let _ = child.kill();
             let _ = child.wait();
-            // Not joined: a process the command started may still hold the
-            // pipe open; the reader ends when it closes, and the caller
-            // returns at the deadline.
-            drop(reader);
             return Err("vault.secret_store_timeout");
         }
     };
-    let raw = reader.join().map_err(|_| "vault.secret_store_read_failed")?;
-    Ok((status, raw))
+    match read.recv_timeout(end.saturating_duration_since(std::time::Instant::now())) {
+        Ok(raw) => Ok((status, raw)),
+        Err(std::sync::mpsc::RecvTimeoutError::Timeout) => Err("vault.secret_store_timeout"),
+        Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
+            Err("vault.secret_store_read_failed")
+        }
+    }
 }
 
 /// All of a child's pipe, into a zeroed buffer (at most 1 MiB).
@@ -346,8 +363,9 @@ const ITEM_PREFIX: &str = "v1:";
 
 /// The item text of `value`: the marker, then base64, one printable word.
 pub fn encode(value: &str) -> Zeroizing<String> {
-    let mut out =
-        Zeroizing::new(String::with_capacity(ITEM_PREFIX.len() + value.len() * 4 / 3 + 4));
+    let mut out = Zeroizing::new(String::with_capacity(
+        value.len().div_ceil(3).saturating_mul(4).saturating_add(ITEM_PREFIX.len()),
+    ));
     out.push_str(ITEM_PREFIX);
     base64::engine::general_purpose::STANDARD.encode_string(value.as_bytes(), &mut out);
     out