Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 47 additions & 27 deletions crates/shim/src/tools.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
})
}

Expand All @@ -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}'"),
Expand Down Expand Up @@ -441,6 +425,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<T, Fut>(
host: &ToolHost,
what: &str,
run: impl Fn() -> Fut,
) -> Result<T, ToolError>
where
Fut: std::future::Future<Output = Result<T, ToolError>>,
{
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);

Expand Down
8 changes: 8 additions & 0 deletions docs/roadmap.md
Original file line number Diff line number Diff line change
Expand Up @@ -644,6 +644,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.
Expand Down
Loading