From e08edee4260111b1358f384ccf767df8a2d6eb56 Mon Sep 17 00:00:00 2001 From: theBlanketBaron Date: Fri, 6 Nov 2026 22:28:33 -0500 Subject: [PATCH] fix(shim): a messages read right after connecting waits for registration (MUS-114) The restart-file e2e failed intermittently in CI (#185, #194) and in 2 of 15 local runs: a fresh shim's muster://messages read was refused as unregistered before its Session SET landed. Tool calls already held for registration and retried once on that refusal (MUS-72, MUS-79); the read did neither. Both now go through as_registered. 0 of 30 local runs failed with the fix. Co-Authored-By: Claude Opus 5.5 --- crates/shim/src/tools.rs | 74 +++++++++++++++++++++++++--------------- docs/roadmap.md | 8 +++++ 2 files changed, 55 insertions(+), 27 deletions(-) diff --git a/crates/shim/src/tools.rs b/crates/shim/src/tools.rs index 22532b35..318e5050 100644 --- a/crates/shim/src/tools.rs +++ b/crates/shim/src/tools.rs @@ -114,32 +114,8 @@ impl Handler for CoordHandler { let name = name.to_string(); Box::pin(async move { // Every tool here writes as this session, which the daemon refuses - // until the session's row is on the roster. An agent's first call can - // come before the shim's registration lands (MUS-72), so hold it - // briefly rather than fail it. - wait_until_registered(&host).await; - let first = dispatch_tool(&host, &name, &args).await; - // The hold can pass on a row still bound to an earlier connection - // for this session (one that died and the sweeper hasn't cleared), - // and the daemon then refuses the write, because it finds a caller - // by this connection's binding (MUS-79). A refused write stored - // nothing, so send the session again and retry it once. - match first { - Err(e) if !host.is_codex && is_unbound_refusal(&e.message) => { - log::warn!( - "[muster-shim] {name} was refused as unregistered; sending the session again and retrying once" - ); - let snapshot = host.session.lock().unwrap().clone(); - if let Err(err) = crate::emit_session_set(&host.client, &snapshot) { - log::warn!("[muster-shim] re-SET before a retry failed: {err}"); - } - // The roster can't tell this connection's binding from an - // older one, so give the SET a moment to land. - tokio::time::sleep(UNBOUND_RETRY_AFTER).await; - dispatch_tool(&host, &name, &args).await - } - other => other, - } + // until the session's row is on the roster (MUS-72, MUS-79). + as_registered(&host, &name, || dispatch_tool(&host, &name, &args)).await }) } @@ -152,7 +128,15 @@ impl Handler for CoordHandler { "whoami" => Ok(read_whoami(&host, &uri, &parsed.query)), "roster" => Ok(read_roster(&host, &uri)), "rooms" => Ok(read_rooms(&host, &uri)), - "messages" => read_messages(&host, &uri, &parsed.query).await, + // The daemon finds whose mail to read by this connection's + // binding, so a read right after connecting is refused the + // way an early write is (MUS-114): hold it the same way. + "messages" => { + as_registered(&host, "muster://messages", || { + read_messages(&host, &uri, &parsed.query) + }) + .await + } other => Err(ResourceError { code: METHOD_NOT_FOUND, message: format!("no resource at 'muster://{other}'"), @@ -437,6 +421,42 @@ async fn dispatch_tool( } } +/// Run `run`, a command sent as this session, once the session is on the +/// roster. An agent's first call can come before the shim's registration lands +/// (MUS-72), so it's held briefly rather than failed. The hold can also pass on +/// a row still bound to an earlier connection for this session (one that died +/// and the sweeper hasn't cleared), and the daemon then refuses the command, +/// because it finds a caller by this connection's binding (MUS-79). A refused +/// command did nothing, so the session is sent again and the command retried +/// once. Reads that the daemon resolves by connection need this as much as +/// writes do (MUS-114). +async fn as_registered( + host: &ToolHost, + what: &str, + run: impl Fn() -> Fut, +) -> Result +where + Fut: std::future::Future>, +{ + wait_until_registered(host).await; + match run().await { + Err(e) if !host.is_codex && is_unbound_refusal(&e.message) => { + log::warn!( + "[muster-shim] {what} was refused as unregistered; sending the session again and retrying once" + ); + let snapshot = host.session.lock().unwrap().clone(); + if let Err(err) = crate::emit_session_set(&host.client, &snapshot) { + log::warn!("[muster-shim] re-SET before a retry failed: {err}"); + } + // The roster can't tell this connection's binding from an older + // one, so give the SET a moment to land. + tokio::time::sleep(UNBOUND_RETRY_AFTER).await; + run().await + } + other => other, + } +} + /// How long a retried write waits for its re-SET to land (MUS-79). const UNBOUND_RETRY_AFTER: std::time::Duration = std::time::Duration::from_millis(750); diff --git a/docs/roadmap.md b/docs/roadmap.md index 18f9ff8e..02e0f9bc 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -616,6 +616,14 @@ Merged and needing a live run: MUS-36 and MUS-37. another key; release.yml says so; threat-model.md has the note; tests cover each. Built with a placeholder key that fails closed. Left: Josh adds his public key (to release/allowed_signers and both scripts) before signing the first release (MUS-14). +- [x] **MUS-114: A resource read right after connecting waits for registration, as a write does.** + From CI on 2026-11-06: `scripts/test-restart-file-e2e.sh` failed on #185 and #194, whose diffs couldn't + touch it, and in 2 of 15 local runs. The read of `muster://messages?inbox=true&unread=true` was refused + with `[unregistered] caller (client …) has no session on the roster`: a fresh shim read before its + Session SET landed, and its row from the run before (same session id, a dead connection) let the + registration wait pass. Tool calls already wait and retry once (MUS-72, MUS-79); the messages read did + neither. Done: both go through one `as_registered`, which holds until the session is on the roster and, + on that refusal, sends the session again and retries once. - [ ] **MUS-30: Decide the future of the TypeScript plugins.** The operator decides whether to keep, freeze or remove them. Done when: the decision is recorded here, and if they're removed, the repository's `package.json` workspace goes with them.