feat(broker): publish fleet workers' provider session id on their Relaycast agent

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - `teams.json` agents accept a per-agent `model` field when `up --spawn` starts them; an explicit `--model` or `-m` inside `cli` still wins.
+- The cloud dashboard can link a recorded Claude Code or Codex session to the fleet worker running it, so you can message that worker from the session page. After each spawn and supervised restart, the broker records the worker's provider session as `session_id` and `session_kind` in its Relaycast agent metadata.
 
 ### Changed
 

crates/broker/src/relaycast/mod.rs

@@ -27,6 +27,6 @@ pub(crate) use workspace::{
 pub(crate) use ws::{
     format_worker_preregistration_error, register_new_spawn_identity,
     registration_retry_after_secs, retry_agent_registration, retry_agent_registration_create_only,
-    RegRetryOutcome, RelaycastHttpClient, RelaycastRegistrationError, WsControl,
-    MAX_AGENT_REGISTRATION_ELAPSED, MAX_AGENT_REGISTRATION_OUTER_TIMEOUT,
+    RegRetryOutcome, RelaycastHttpClient, RelaycastRegistrationError, SessionMetadataPublish,
+    WsControl, MAX_AGENT_REGISTRATION_ELAPSED, MAX_AGENT_REGISTRATION_OUTER_TIMEOUT,
 };

crates/broker/src/runtime/api.rs

@@ -804,6 +804,11 @@ impl BrokerRuntime {
                                 name.as_str(),
                                 registration_metadata,
                             );
+                            super::fleet::spawn_session_metadata_publish(
+                                relaycast_http,
+                                name.as_str(),
+                                &effective_spec,
+                            );
                         }
                         if owns_identity {
                             if let Some(worker) = workers.workers.get(&name) {

crates/broker/src/runtime/fleet.rs

@@ -8,6 +8,7 @@ use crate::{
     listen_api::{DeliveryRouteError, ListenApiRequest, SetInboundDeliveryModeOk},
     node_control::{delivery_ack, handler_unavailable_result, DeliveryDecision, ReceiptAckability},
     node_delivery_probe::DeliverDisposition,
+    relaycast::SessionMetadataPublish,
     terminal_control::{
         request_terminal_reconnect, TerminalControlCommand, TerminalControlEvent,
         TerminalDeliveryDiagnostics, TerminalFromCloud, TerminalMode, TerminalToCloud,
@@ -2294,6 +2295,98 @@ pub(super) fn spawn_declared_metadata_publish(
     });
 }
 
+/// The provider session a spawned worker runs, as published on its Relaycast
+/// agent: `(session_id, session_kind)`.
+///
+/// `session_id` is the id the worker's CLI records its session under — the
+/// Claude Code session UUID or the Codex thread id — resolved by
+/// `WorkerRegistry::spawn` before the process starts (and reported as
+/// `sessionId` by `GET /api/spawned`). `None` when the spawn has no session id,
+/// e.g. a CLI the broker cannot pre-assign one for.
+///
+/// `session_kind` names the CLI family, using the same values desktop session
+/// agents publish where they overlap: `claude-terminal` for an interactive
+/// (PTY) Claude Code session, `codex` for Codex. Other CLIs report their
+/// normalized CLI name (headless Claude reports `claude`).
+pub(super) fn worker_session_metadata(spec: &AgentSpec) -> Option<(String, Option<String>)> {
+    let session_id = spec
+        .session_id
+        .as_deref()
+        .map(str::trim)
+        .filter(|id| !id.is_empty())?
+        .to_string();
+    let cli = spec
+        .cli
+        .as_deref()
+        // Parse the executable the way worker startup does, so a quoted path
+        // with spaces (`"/opt/AI Tools/codex" --flag`) names `codex`.
+        .and_then(|cli| crate::cli::command_parse::parse_cli_command(cli).ok())
+        .map(|(command, _)| crate::cli::command_parse::normalize_cli_name(&command).to_lowercase())
+        .or_else(|| {
+            spec.provider
+                .as_ref()
+                .map(|provider| super::headless::headless_provider_cli_name(provider).to_string())
+        });
+    let kind = cli.map(|cli| {
+        let family = cli.split(':').next().unwrap_or(&cli).to_string();
+        match family.as_str() {
+            "claude" if spec.runtime == AgentRuntime::Pty => "claude-terminal".to_string(),
+            _ => family,
+        }
+    });
+    Some((session_id, kind))
+}
+
+/// Publish a freshly spawned (or respawned) worker's provider session id onto
+/// its Relaycast agent, on its own task.
+///
+/// Called after every successful spawn of a worker that holds a hosted
+/// identity, so a respawn that resumes or starts a different session updates
+/// the published id. The session claim is taken here, synchronously and even
+/// when the spawn has no session id, so any publish still pending from an
+/// earlier worker of the same name is superseded and cannot label this one
+/// with the old session. Detached and best-effort for the same reasons as
+/// [`spawn_declared_metadata_publish`]: a failure is logged, never fatal.
+pub(super) fn spawn_session_metadata_publish(
+    relaycast_http: &RelaycastHttpClient,
+    name: &str,
+    spec: &AgentSpec,
+) {
+    let Ok(claim) = relaycast_http.claim_session_metadata(name) else {
+        return;
+    };
+    let Some((session_id, session_kind)) = worker_session_metadata(spec) else {
+        return;
+    };
+    let http = relaycast_http.clone();
+    tokio::spawn(async move {
+        let agent = claim.agent_name();
+        match http
+            .publish_session_metadata(&claim, &session_id, session_kind.as_deref())
+            .await
+        {
+            Ok(SessionMetadataPublish::Published) => tracing::debug!(
+                worker = %agent,
+                session_id = %session_id,
+                "published provider session id for spawned agent"
+            ),
+            Ok(SessionMetadataPublish::Superseded) => tracing::debug!(
+                worker = %agent,
+                session_id = %session_id,
+                "skipped provider session id publish; a later spawn of this name owns it"
+            ),
+            Ok(SessionMetadataPublish::NothingToPublish) => {}
+            Err(error) => tracing::error!(
+                worker = %agent,
+                session_id = %session_id,
+                error = %error,
+                "failed to publish provider session id; the agent is registered and running \
+                 but its session cannot be linked to it"
+            ),
+        }
+    });
+}
+
 /// Bind an agent to this node by sending node-control `agent.register` and
 /// awaiting the engine reply with the minted agent token. This is the single
 /// "register agent via node" step both the `/api/spawn` path and the node
@@ -4652,6 +4745,61 @@ mod tests {
         assert_eq!(delivery_book.active_agent_id("worker-a"), None);
     }
 
+    #[test]
+    fn worker_session_metadata_reports_provider_session_and_kind() {
+        let mut spec = test_agent_spec(Some(" thread-123 "), None);
+        assert_eq!(
+            worker_session_metadata(&spec),
+            Some(("thread-123".to_string(), Some("codex".to_string())))
+        );
+
+        spec.cli = Some("/usr/local/bin/claude --model opus".to_string());
+        assert_eq!(
+            worker_session_metadata(&spec),
+            Some((
+                "thread-123".to_string(),
+                Some("claude-terminal".to_string())
+            ))
+        );
+
+        // A quoted executable path with spaces names the binary, not the
+        // first whitespace-separated fragment of its directory.
+        spec.cli = Some("\"/opt/AI Tools/codex\" --full-auto".to_string());
+        assert_eq!(
+            worker_session_metadata(&spec),
+            Some(("thread-123".to_string(), Some("codex".to_string())))
+        );
+
+        spec.cli = Some("claude:opus".to_string());
+        spec.runtime = AgentRuntime::Headless;
+        assert_eq!(
+            worker_session_metadata(&spec),
+            Some(("thread-123".to_string(), Some("claude".to_string())))
+        );
+
+        spec.cli = None;
+        spec.provider = Some(ProtocolHeadlessProvider::Opencode);
+        assert_eq!(
+            worker_session_metadata(&spec),
+            Some(("thread-123".to_string(), Some("opencode".to_string())))
+        );
+
+        spec.provider = None;
+        assert_eq!(
+            worker_session_metadata(&spec),
+            Some(("thread-123".to_string(), None))
+        );
+    }
+
+    #[test]
+    fn worker_session_metadata_is_none_without_a_session_id() {
+        assert_eq!(worker_session_metadata(&test_agent_spec(None, None)), None);
+        assert_eq!(
+            worker_session_metadata(&test_agent_spec(Some("   "), None)),
+            None
+        );
+    }
+
     #[tokio::test]
     async fn refresh_fleet_inventory_session_ref_publishes_immediate_sync() {
         let (tx, mut rx) = mpsc::channel(4);

crates/broker/src/runtime/maintenance.rs

@@ -657,6 +657,7 @@ impl BrokerRuntime {
                     }
                 }
 
+                let has_hosted_identity = worker_relay_key.is_some();
                 match workers
                     .spawn(
                         rst.payload.spec.clone(),
@@ -673,6 +674,15 @@ impl BrokerRuntime {
                 {
                     Ok(effective_spec) => {
                         fleet_load_changed = true;
+                        // A respawn may resume the same session or start a new
+                        // one; either way the published id must follow it.
+                        if has_hosted_identity {
+                            super::fleet::spawn_session_metadata_publish(
+                                relaycast_http,
+                                name.as_str(),
+                                &effective_spec,
+                            );
+                        }
                         // A supervised tokenless worker owns the identity
                         // across respawns, but each process is a new exact
                         // generation. Refresh custody before any later

crates/broker/src/runtime/relaycast_events.rs

@@ -873,6 +873,16 @@ pub(super) async fn spawn_worker_from_request(
         .await
     {
         Ok(effective_spec) => {
+            if worker_relay_key.is_some() {
+                // The provider session id is only final once the spawn has
+                // resolved it, so it is published here rather than alongside
+                // the declared metadata at registration.
+                super::fleet::spawn_session_metadata_publish(
+                    workspace_http,
+                    name.as_str(),
+                    &effective_spec,
+                );
+            }
             if owns_identity {
                 if let Some(worker) = workers.workers.get(&name) {
                     workers

crates/broker/src/runtime/tests.rs

@@ -7946,6 +7946,18 @@ async fn assert_http_spawn_metadata_publication(supplied_token: bool, valid_cwd:
         then.status(200)
             .json_body(json!({"ok":true,"data":identity}));
     });
+    // The provider session id rides its own merge-only PATCH once the spawn
+    // has resolved it, so the dashboard can link the recorded session to this
+    // worker without disturbing the declared keys or the `fleet` record.
+    let session_metadata = server.mock(|when, then| {
+        when.method(PATCH)
+            .path("/v1/agents/metadata-worker")
+            .json_body(json!({"metadata":{
+                "session_id":"metadata-session", "session_kind":"cat"
+            }}));
+        then.status(200)
+            .json_body(json!({"ok":true,"data":identity}));
+    });
     let unexpected_cleanup = server.mock(|when, then| {
         when.method(POST).path("/v1/agents/release");
         then.status(500);
@@ -8015,20 +8027,22 @@ async fn assert_http_spawn_metadata_publication(supplied_token: bool, valid_cwd:
     if valid_cwd {
         assert_eq!(response.unwrap()["success"], true);
         let published = tokio::time::timeout(Duration::from_secs(2), async {
-            while metadata.hits() == 0 {
+            while metadata.hits() == 0 || session_metadata.hits() == 0 {
                 tokio::task::yield_now().await;
             }
         })
         .await;
         assert!(
             published.is_ok(),
-            "successful spawn did not publish declared metadata"
+            "successful spawn did not publish declared and session metadata"
         );
         metadata.assert_hits(1);
+        session_metadata.assert_hits(1);
     } else {
         assert!(response.unwrap_err().contains("cwd"));
         tokio::time::sleep(Duration::from_millis(100)).await;
         metadata.assert_hits(0);
+        session_metadata.assert_hits(0);
     }
     create.assert_hits(usize::from(!supplied_token));
     bind.assert_hits(usize::from(!supplied_token));