feat: publish session context cwd

README.md

@@ -59,7 +59,7 @@ Health and debug endpoints:
 - **Opt-in request trace metadata.** With `--meta-propagate`, outbound subscriber → agent requests get mux-owned `params._meta.amux` fields (`peerId`, `peerName`, `role`, `muxId`, and `amuxTurnId` for prompts) for cross-client debugging. Default mode leaves request payload metadata unchanged.
 - **Cold-start session discovery.** `GET /acp/sessions` runs a transient agent-side `session/list` query before any WebSocket attach, useful for dashboards that need to browse persisted sessions before choosing one to resume.
 - **Live `session/list` decoration.** Returned `sessions[]` entries that match a live muxed upstream session get `sessions[i]._meta.amux` fields (`proxySessionId`, `subscriberCount`, optional `drivingSubscriber`), preserving existing `_meta` keys and leaving non-live entries unchanged.
-- **`amux/*` notification namespace.** The mux publishes its own metadata out-of-band: `amux/peer_joined`, `amux/peer_left`, `amux/turn_started`, `amux/turn_complete`, `amux/turn_cancelled`, `amux/session_busy`, `amux/agent_request_opened`, `amux/agent_request_resolved`. ACP frames stay clean; clients see two distinguishable channels and demultiplex by method prefix.
+- **`amux/*` notification namespace.** The mux publishes its own metadata out-of-band: `amux/session_context`, `amux/peer_joined`, `amux/peer_left`, `amux/turn_started`, `amux/turn_complete`, `amux/turn_cancelled`, `amux/session_busy`, `amux/agent_request_opened`, `amux/agent_request_resolved`. ACP frames stay clean; clients see two distinguishable channels and demultiplex by method prefix.
 - **Cancellation.** `$/cancel_request` (request-cancellation RFD) works both directions: subscribers can cancel their own in-flight requests; agents can cancel agent-initiated requests (broadcast to peers + `amux/agent_request_resolved { resolvedBy: "agent:cancelled" }`). The amux extension `amux/cancel_active_turn` lets *any* attached peer cancel the in-flight turn (not just the driver) — internally it sends ACP-native `session/cancel { sessionId }` toward the agent and emits `amux/turn_cancelled` to peers.
 - **Replay log.** Every broadcast-tier frame (`amux/*` + agent notifications) is appended; a late joiner receives the full history before any live event. Raw actionable agent-initiated requests are live-only and are not replayed; late joiners see the inert `amux/agent_request_opened` + `amux/agent_request_resolved` lifecycle pair instead.
 - **TTL grace.** Last subscriber leaving starts a countdown; a reconnect within `--session-ttl-seconds` reuses the same subprocess with all of its caches intact.
@@ -110,6 +110,7 @@ amux parses only JSON-RPC envelopes (`id`, `method`, `params`, `result`, `error`
 
 | Method | Direction | Purpose |
 |---|---|---|
+| `amux/session_context` | proxy → subscriber | Per-attach mux/agent process context, including the cwd inherited by the agent subprocess. |
 | `amux/peer_joined`, `amux/peer_left` | proxy → subscribers | Presence. |
 | `amux/turn_started`, `amux/turn_complete` | proxy → subscribers | Turn bookends with `amuxTurnId`. |
 | `amux/turn_cancelled` | proxy → subscribers | Intent broadcast when any peer triggers cancellation. |

docs/design/amux-namespace.md

@@ -178,6 +178,28 @@ the replay log (see below), which already contains `peer_joined` events for
 every peer still in the session — no per-peer presence replay needed at
 attach time.
 
+### `amux/session_context`
+
+Sent directly to each subscriber on attach with the mux-owned execution
+context for the room. This is not an ACP session metadata claim: it identifies
+the cwd inherited by the agent subprocess, which is the context used for
+tools/terminal work even if a client connected from a different local cwd.
+
+```json
+{
+  "jsonrpc": "2.0",
+  "method": "amux/session_context",
+  "params": {
+    "sessionId": "work",
+    "cwd": "/home/volt/Code/acp-mux"
+  }
+}
+```
+
+- Emitted once per attach to the attaching subscriber.
+- Clients can use it for chrome/status UI that should reflect the agent's
+  actual working directory rather than the local client's launch cwd.
+
 ### `amux/session_busy`
 
 Broadcast when a `session/prompt` is rejected because another turn is

src/protocol/amux.rs

@@ -12,6 +12,7 @@ use serde::Serialize;
 
 const METHOD_PEER_JOINED: &str = "amux/peer_joined";
 const METHOD_PEER_LEFT: &str = "amux/peer_left";
+const METHOD_SESSION_CONTEXT: &str = "amux/session_context";
 const METHOD_TURN_STARTED: &str = "amux/turn_started";
 const METHOD_TURN_COMPLETE: &str = "amux/turn_complete";
 const METHOD_SESSION_BUSY: &str = "amux/session_busy";
@@ -65,6 +66,13 @@ struct PeerLeftParams<'a> {
     peer_id: &'a str,
 }
 
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct SessionContextParams<'a> {
+    session_id: &'a str,
+    cwd: &'a str,
+}
+
 #[derive(Serialize)]
 #[serde(rename_all = "camelCase")]
 struct TurnStartedParams<'a> {
@@ -184,6 +192,13 @@ pub fn peer_left(session_id: &str, peer_id: &str) -> Vec<u8> {
     )
 }
 
+pub fn session_context(session_id: &str, cwd: &str) -> Vec<u8> {
+    encode(
+        METHOD_SESSION_CONTEXT,
+        SessionContextParams { session_id, cwd },
+    )
+}
+
 pub fn turn_started(
     session_id: &str,
     amux_turn_id: AmuxTurnId,

src/session/registry.rs

@@ -250,6 +250,12 @@ impl SessionRegistry {
             .agent_cmd
             .as_ref()
             .ok_or(RegistryError::AgentCmdMissing)?;
+        let agent_cwd = std::env::current_dir()
+            .map(|path| path.to_string_lossy().to_string())
+            .unwrap_or_else(|err| {
+                tracing::warn!(error = %err, "failed to read current dir for session context");
+                String::new()
+            });
         let agent = AgentProcess::spawn(&cmd.program, &cmd.args)
             .await
             .map_err(RegistryError::AgentSpawn)?;
@@ -262,6 +268,7 @@ impl SessionRegistry {
                 session_ttl: self.session_ttl,
                 meta_propagate: self.meta_propagate,
                 session_list_index: self.session_list_index.clone(),
+                agent_cwd,
             },
         );
         sessions.insert(session_id.to_string(), handle.clone());

src/session/state.rs

@@ -450,6 +450,7 @@ pub struct ReplayResetSnapshot {
 #[serde(rename_all = "camelCase")]
 pub struct SessionSnapshot {
     pub session_id: String,
+    pub agent_cwd: String,
     pub subscribers: Vec<SubscriberSnapshot>,
     pub pending_request_count: usize,
     pub initialize_cached: bool,
@@ -529,6 +530,7 @@ enum AgentReqState {
 
 struct SessionInner {
     session_id: String,
+    agent_cwd: String,
     session_list_index: Arc<SessionListMetadataIndex>,
     canonical_session_id: Option<String>,
     subscribers: HashMap<String, Subscriber>,
@@ -579,6 +581,7 @@ struct SessionInner {
 impl SessionInner {
     fn new(
         session_id: String,
+        agent_cwd: String,
         replay_policy: ReplayTurns,
         meta_propagate: bool,
         session_list_index: Arc<SessionListMetadataIndex>,
@@ -596,6 +599,7 @@ impl SessionInner {
         };
         Self {
             session_id,
+            agent_cwd,
             session_list_index,
             canonical_session_id: None,
             subscribers: HashMap::new(),
@@ -719,6 +723,7 @@ impl SessionInner {
         );
         self.subscribers.insert(peer_id.clone(), subscriber);
         self.publish_session_list_metadata();
+        self.send_session_context_to(&peer_id);
 
         if let Some(sub) = self.subscribers.get(&peer_id) {
             for entry in snapshot {
@@ -732,6 +737,16 @@ impl SessionInner {
         Ok(())
     }
 
+    fn send_session_context_to(&self, peer_id: &str) {
+        let Some(sub) = self.subscribers.get(peer_id) else {
+            return;
+        };
+        let frame = Bytes::from(amux::session_context(&self.session_id, &self.agent_cwd));
+        if sub.outbound.send(OutMsg::Frame(frame)).is_err() {
+            tracing::debug!(%peer_id, "subscriber dropped before session context delivered");
+        }
+    }
+
     /// Build a serializable snapshot of session state for /debug/sessions.
     fn build_snapshot(&self, ttl_pending: bool) -> SessionSnapshot {
         let subs: Vec<SubscriberSnapshot> = self
@@ -751,6 +766,7 @@ impl SessionInner {
         });
         SessionSnapshot {
             session_id: self.session_id.clone(),
+            agent_cwd: self.agent_cwd.clone(),
             subscribers: subs,
             pending_request_count: self.pending.len(),
             initialize_cached: self.initialize_cache.is_some(),
@@ -1776,6 +1792,7 @@ pub struct SessionOptions {
     pub session_ttl: Duration,
     pub meta_propagate: bool,
     pub session_list_index: Arc<SessionListMetadataIndex>,
+    pub agent_cwd: String,
 }
 
 pub fn spawn_session(
@@ -1837,6 +1854,7 @@ async fn run_session(
 ) {
     let mut inner = SessionInner::new(
         session_id.clone(),
+        options.agent_cwd.clone(),
         options.replay_policy,
         options.meta_propagate,
         options.session_list_index.clone(),

tests/server.rs

@@ -148,6 +148,9 @@ async fn ws_loopback_roundtrip_via_cat() {
             break;
         }
         let v: serde_json::Value = serde_json::from_str(t.as_str()).expect("frame is JSON");
+        if v.get("method") == Some(&serde_json::json!("amux/session_context")) {
+            continue;
+        }
         if v.get("method") == Some(&serde_json::json!("amux/agent_request_opened")) {
             saw_opened = true;
             continue;
@@ -176,6 +179,26 @@ async fn ws_loopback_roundtrip_via_cat() {
     panic!("session did not tear down after last subscriber");
 }
 
+#[tokio::test]
+async fn subscriber_receives_agent_context_cwd_on_attach() {
+    let (addr, _) = spawn_server_with_cat().await;
+    let expected_cwd = std::env::current_dir()
+        .unwrap()
+        .to_string_lossy()
+        .to_string();
+    let url = format!("ws://{addr}/acp?session=ctx&peer_id=p1");
+    let (mut ws, _) = tokio_tungstenite::connect_async(url)
+        .await
+        .expect("ws connect");
+
+    let context = ws_next_method(&mut ws, "amux/session_context").await;
+
+    assert_eq!(context["params"]["sessionId"], serde_json::json!("ctx"));
+    assert_eq!(context["params"]["cwd"], serde_json::json!(expected_cwd));
+
+    let _ = ws.send(ClientMsg::Close(None)).await;
+}
+
 #[tokio::test]
 async fn ws_two_subscribers_see_naive_fanout() {
     let (addr, _) = spawn_server_with_cat().await;
@@ -650,11 +673,19 @@ async fn amux_peer_joined_and_peer_left() {
 
     let (mut ws_a, _) = tokio_tungstenite::connect_async(url_a).await.unwrap();
     // A is the initial sub — peer_joined for A is emitted to an empty
-    // map, so A sees nothing yet.
+    // map, so A sees only its direct session_context before B joins.
     let a_early = drain_for(&mut ws_a, Duration::from_millis(100)).await;
     assert!(
-        a_early.is_empty(),
-        "A should see no events before B joins, got {a_early:?}"
+        a_early
+            .iter()
+            .any(|v| v.get("method") == Some(&serde_json::json!("amux/session_context"))),
+        "A should receive direct session_context on attach, got {a_early:?}"
+    );
+    assert!(
+        a_early
+            .iter()
+            .all(|v| v.get("method") == Some(&serde_json::json!("amux/session_context"))),
+        "A should see no peer/presence events before B joins, got {a_early:?}"
     );
 
     let (mut ws_b, _) = tokio_tungstenite::connect_async(url_b).await.unwrap();
@@ -1114,10 +1145,19 @@ async fn replay_turns_disabled_emits_no_history() {
     let (mut ws_b, _) = tokio_tungstenite::connect_async(url_b).await.unwrap();
     let early = drain_for(&mut ws_b, Duration::from_millis(150)).await;
     // peer_joined for B's own join doesn't broadcast to B; without a
-    // replay log, B sees nothing until the next live event.
+    // replay log, B sees only the per-attach session_context until the
+    // next live event.
     assert!(
-        early.is_empty(),
-        "B should see no replay frames, got {early:?}"
+        early
+            .iter()
+            .any(|v| v.get("method") == Some(&serde_json::json!("amux/session_context"))),
+        "B should receive direct session_context on attach, got {early:?}"
+    );
+    assert!(
+        early
+            .iter()
+            .all(|v| v.get("method") == Some(&serde_json::json!("amux/session_context"))),
+        "B should see no replay frames beyond session_context, got {early:?}"
     );
 
     let _ = ws_a.send(ClientMsg::Close(None)).await;