feat(broker): add POST /api/observer-token to mint scoped read-only tokens (#1223)

* feat(broker): add POST /api/observer-token to mint scoped read-only tokens

Pear's "Join as observer" feature currently builds shareable links by
embedding the full rk_live_... Relaycast workspace API key, granting
whoever holds the link full read/write/spawn access to the whole
workspace. This adds the broker-side capability Pear's fix needs: a
local HTTP endpoint that mints a scoped, read-only ot_live_... observer
token via the Relaycast SDK's create_observer_token, instead of
handing out the raw workspace credential.

- ListenApiRequest::CreateObserverToken (listen_api.rs) resolves the
  target workspace the same way /api/send does (workspaceId /
  workspaceAlias / sole-attached / default), then mints a token scoped
  to stream/messages/threads/dms/channels/activity/agents:read.
- New RelaycastHttpClient::create_observer_token wraps the SDK call
  (relaycast/ws.rs), following the register_action pattern.
- Extracted the shared workspace-resolution logic used by /api/send
  and /api/observer-token into resolve_workspace() (runtime/api.rs),
  with unit tests covering explicit id/alias, case-insensitive alias
  matching, sole-workspace and default-workspace fallback, ambiguous-
  workspace, and not-found paths, plus a test asserting the minted
  scope set.
- Route is protected by the existing listen_api_auth_middleware, same
  as every other /api/* route.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* fix(broker): harden observer-token endpoint per review feedback

Address three review findings on the new POST /api/observer-token
endpoint (PR #1223), which mints scoped read-only Relaycast observer
tokens so callers stop embedding the full rk_live_... workspace key:

- Parse the request body as raw bytes instead of via axum's Json /
  Option<Json<_>> extractor. Axum still attempts to parse the body when
  Content-Type: application/json is present, even if empty, and
  Option<Json<_>> only short-circuits to None when the header itself is
  missing -- so a dashboard client that sets JSON headers unconditionally
  on a request with every field optional would get rejected before the
  handler ran. Empty/whitespace-only bodies are now treated as {}.

- Reject requests with 400 when workspaceId/workspace_id,
  workspaceAlias/workspace_alias, or name are present but not strings,
  instead of silently treating them as absent. For a credential-minting
  endpoint, silently falling back to the default/sole workspace on a
  malformed selector is a real risk.

- Bound the create_observer_token SDK call in the broker runtime task
  with the same timeout() used for Send's Relaycast publish call, so a
  hung SDK call can't block the runtime task indefinitely (the HTTP
  layer's own 504 timeout only frees the caller, not this task).

Adds regression tests for the empty-body case, non-string field
rejection, and malformed JSON bodies.

* fix(broker): exact scope assertion + dedicated observer-token timeout

Address two cubic-dev-ai re-review findings on PR #1223:

- default_observer_token_scopes_are_read_only_and_exclude_unneeded_scopes
  only asserted inclusion of 7 scopes and exclusion of 5 others, so an
  accidentally-added extra scope (including a write scope) would still
  pass. Now asserts the exact set via HashSet equality plus a length
  check, which fails on any scope drift for this credential-minting
  endpoint's minimal grant.

- create_observer_token's runtime timeout reused
  http_api_relaycast_send_timeout() (and its
  AGENT_RELAY_HTTP_API_RELAYCAST_SEND_TIMEOUT_MS env var), which is
  named/scoped for the /api/send publish path. Introduced a dedicated
  http_api_observer_token_timeout() (AGENT_RELAY_HTTP_API_OBSERVER_TOKEN_TIMEOUT_MS,
  same 20s default) so tuning send-path latency can't unintentionally
  affect token minting.

---------

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

Author: willwashburn

crates/broker/src/listen_api.rs

@@ -191,6 +191,17 @@ pub enum ListenApiRequest {
         metadata: Option<Value>,
         reply: tokio::sync::oneshot::Sender<Result<Value, AgentResultRouteError>>,
     },
+    /// `POST /api/observer-token` — mint a scoped, read-only Relaycast
+    /// observer token (`ot_live_...`) for the resolved workspace. Used by
+    /// local dashboard clients (e.g. Pear's "Join as observer" link) so they
+    /// stop embedding the full `rk_live_...` workspace key, which grants
+    /// full read/write/spawn access, in shareable links.
+    CreateObserverToken {
+        workspace_id: Option<WorkspaceId>,
+        workspace_alias: Option<WorkspaceAlias>,
+        name: Option<String>,
+        reply: tokio::sync::oneshot::Sender<Result<Value, String>>,
+    },
     FleetSidecarConnect {
         outbound: mpsc::Sender<ProtocolEnvelope<Value>>,
         reply: tokio::sync::oneshot::Sender<Result<Value, String>>,
@@ -412,6 +423,10 @@ fn listen_api_router_with_auth(
             routing::post(listen_api_interrupt),
         )
         .route("/api/send", routing::post(listen_api_send))
+        .route(
+            "/api/observer-token",
+            routing::post(listen_api_create_observer_token),
+        )
         .route("/api/input/{name}", routing::post(listen_api_send_input))
         .route(
             "/api/input/{name}/stream",
@@ -1285,6 +1300,124 @@ async fn listen_api_send(
     }
 }
 
+/// `POST /api/observer-token` — mint a scoped, read-only observer token for
+/// the resolved workspace so local dashboard clients (e.g. Pear's "Join as
+/// observer" link) never have to embed the full workspace API key.
+async fn listen_api_create_observer_token(
+    axum::extract::State(state): axum::extract::State<ListenApiState>,
+    body: axum::body::Bytes,
+) -> (axum::http::StatusCode, axum::Json<Value>) {
+    // Parse raw bytes instead of using the `Json`/`Option<Json<_>>` extractor:
+    // dashboard clients may send `Content-Type: application/json` with an
+    // empty body (every field here is optional), and axum still attempts to
+    // parse that body and rejects the request before this handler runs.
+    let body: Value = if body.is_empty() {
+        Value::Null
+    } else {
+        match serde_json::from_slice::<Value>(&body) {
+            Ok(value) => value,
+            Err(err) => {
+                return (
+                    axum::http::StatusCode::BAD_REQUEST,
+                    axum::Json(json!({
+                        "success": false,
+                        "error": format!("invalid JSON body: {err}"),
+                    })),
+                );
+            }
+        }
+    };
+
+    // Extract an optional string field, rejecting the request with 400 if a
+    // field is present but not a string, rather than silently treating a
+    // malformed selector as absent (which could mint a token for the wrong
+    // workspace on this credential-minting endpoint).
+    let string_field =
+        |keys: &[&str]| -> Result<Option<String>, (axum::http::StatusCode, axum::Json<Value>)> {
+            for key in keys {
+                let Some(raw) = body.get(*key) else {
+                    continue;
+                };
+                if raw.is_null() {
+                    continue;
+                }
+                let Some(value) = raw.as_str() else {
+                    return Err((
+                        axum::http::StatusCode::BAD_REQUEST,
+                        axum::Json(json!({
+                            "success": false,
+                            "error": format!("field '{key}' must be a string"),
+                        })),
+                    ));
+                };
+                let value = value.trim();
+                return Ok((!value.is_empty()).then(|| value.to_string()));
+            }
+            Ok(None)
+        };
+    let workspace_id = match string_field(&["workspaceId", "workspace_id"]) {
+        Ok(value) => value,
+        Err(response) => return response,
+    };
+    let workspace_alias = match string_field(&["workspaceAlias", "workspace_alias"]) {
+        Ok(value) => value,
+        Err(response) => return response,
+    };
+    let name = match string_field(&["name"]) {
+        Ok(value) => value,
+        Err(response) => return response,
+    };
+
+    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
+    if state
+        .tx
+        .send(ListenApiRequest::CreateObserverToken {
+            workspace_id: workspace_id.map(WorkspaceId::from),
+            workspace_alias: workspace_alias.map(WorkspaceAlias::from),
+            name,
+            reply: reply_tx,
+        })
+        .await
+        .is_err()
+    {
+        return (
+            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
+            axum::Json(json!({ "success": false, "error": "internal channel closed" })),
+        );
+    }
+
+    match timeout(LISTEN_API_SEND_TIMEOUT, reply_rx).await {
+        Ok(Ok(Ok(val))) => (axum::http::StatusCode::OK, axum::Json(val)),
+        Ok(Ok(Err(err))) => {
+            let raw_error = err.to_string();
+            let status = if raw_error.starts_with("ambiguous_workspace:")
+                || raw_error.starts_with("workspace_not_found:")
+            {
+                axum::http::StatusCode::BAD_REQUEST
+            } else {
+                axum::http::StatusCode::BAD_GATEWAY
+            };
+            let error = raw_error
+                .strip_prefix("ambiguous_workspace:")
+                .or_else(|| raw_error.strip_prefix("workspace_not_found:"))
+                .unwrap_or(&raw_error)
+                .to_string();
+            (
+                status,
+                axum::Json(json!({ "success": false, "error": error })),
+            )
+        }
+        Ok(Err(_)) => (
+            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
+            axum::Json(json!({ "success": false, "error": "internal reply dropped" })),
+        ),
+        Err(_) => (
+            axum::http::StatusCode::GATEWAY_TIMEOUT,
+            axum::Json(json!({ "success": false, "error": "broker request timed out" })),
+        ),
+    }
+}
+
 async fn listen_api_history_stats() -> axum::Json<Value> {
     axum::Json(json!({
         "messageCount": 0,
@@ -3071,6 +3204,156 @@ mod auth_tests {
         );
     }
 
+    #[tokio::test]
+    async fn observer_token_route_accepts_empty_body_with_json_content_type() {
+        // Dashboard clients may send `Content-Type: application/json` with an
+        // empty body since every field on this endpoint is optional; it must
+        // not be rejected before the handler runs (regression test for the
+        // `Json`/`Option<Json<_>>` extractor pitfall).
+        let (router, mut rx) = test_router(Some("secret"));
+        let replier = tokio::spawn(async move {
+            match rx.recv().await {
+                Some(ListenApiRequest::CreateObserverToken {
+                    workspace_id,
+                    workspace_alias,
+                    name,
+                    reply,
+                }) => {
+                    assert_eq!(workspace_id, None);
+                    assert_eq!(workspace_alias, None);
+                    assert_eq!(name, None);
+                    let _ = reply.send(Ok(json!({
+                        "success": true,
+                        "id": "ot_1",
+                        "token": "ot_live_abc",
+                        "name": "pear-dashboard-observer",
+                        "scopes": ["stream:read"],
+                        "workspace_id": "ws_1",
+                        "workspace_alias": null,
+                    })));
+                }
+                other => panic!("unexpected request: {:?}", other.map(|_| "other")),
+            }
+        });
+
+        let response = router
+            .oneshot(
+                Request::builder()
+                    .uri("/api/observer-token")
+                    .method("POST")
+                    .header("x-api-key", "secret")
+                    .header("content-type", "application/json")
+                    .body(Body::empty())
+                    .expect("request should build"),
+            )
+            .await
+            .expect("request should succeed");
+
+        assert_eq!(response.status(), StatusCode::OK);
+        replier.await.expect("replier should complete");
+    }
+
+    #[tokio::test]
+    async fn observer_token_route_accepts_missing_body_entirely() {
+        let (router, mut rx) = test_router(Some("secret"));
+        let replier = tokio::spawn(async move {
+            if let Some(ListenApiRequest::CreateObserverToken { reply, .. }) = rx.recv().await {
+                let _ = reply.send(Ok(json!({ "success": true, "id": "ot_1" })));
+            }
+        });
+
+        let response = router
+            .oneshot(
+                Request::builder()
+                    .uri("/api/observer-token")
+                    .method("POST")
+                    .header("x-api-key", "secret")
+                    .body(Body::empty())
+                    .expect("request should build"),
+            )
+            .await
+            .expect("request should succeed");
+
+        assert_eq!(response.status(), StatusCode::OK);
+        replier.await.expect("replier should complete");
+    }
+
+    #[tokio::test]
+    async fn observer_token_route_rejects_non_string_workspace_id() {
+        let (router, mut rx) = test_router(Some("secret"));
+
+        let response = router
+            .oneshot(
+                Request::builder()
+                    .uri("/api/observer-token")
+                    .method("POST")
+                    .header("x-api-key", "secret")
+                    .header("content-type", "application/json")
+                    .body(Body::from(json!({ "workspaceId": 123 }).to_string()))
+                    .expect("request should build"),
+            )
+            .await
+            .expect("request should succeed");
+
+        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+        let body = response_json(response).await;
+        assert_eq!(body["success"], json!(false));
+        assert!(
+            rx.try_recv().is_err(),
+            "malformed workspaceId should not enqueue a request"
+        );
+    }
+
+    #[tokio::test]
+    async fn observer_token_route_rejects_non_string_name() {
+        let (router, mut rx) = test_router(Some("secret"));
+
+        let response = router
+            .oneshot(
+                Request::builder()
+                    .uri("/api/observer-token")
+                    .method("POST")
+                    .header("x-api-key", "secret")
+                    .header("content-type", "application/json")
+                    .body(Body::from(
+                        json!({ "name": ["not", "a", "string"] }).to_string(),
+                    ))
+                    .expect("request should build"),
+            )
+            .await
+            .expect("request should succeed");
+
+        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+        assert!(
+            rx.try_recv().is_err(),
+            "malformed name should not enqueue a request"
+        );
+    }
+
+    #[tokio::test]
+    async fn observer_token_route_rejects_malformed_json_body() {
+        let (router, mut rx) = test_router(Some("secret"));
+
+        let response = router
+            .oneshot(
+                Request::builder()
+                    .uri("/api/observer-token")
+                    .method("POST")
+                    .header("x-api-key", "secret")
+                    .header("content-type", "application/json")
+                    .body(Body::from("{not json"))
+                    .expect("request should build"),
+            )
+            .await
+            .expect("request should succeed");
+
+        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+        assert!(
+            rx.try_recv().is_err(),
+            "unparseable body should not enqueue a request"
+        );
+    }
+
     #[tokio::test]
     async fn ws_route_rejects_missing_api_key_when_auth_enabled() {
         let (router, _rx) = test_router(Some("secret"));

crates/broker/src/relaycast/ws.rs

@@ -5,8 +5,8 @@ use relaycast::{
     agent::DmOptions, format_registration_error,
     retry_agent_registration as sdk_retry_agent_registration, ActionDefinition, ActionInvocation,
     AgentClient, AgentRegistrationClient, AgentRegistrationError, AgentRegistrationRetryOutcome,
-    CompleteInvocationRequest, MessageListQuery, RegisterActionRequest, RelayCast,
-    RelayCastOptions, ReleaseAgentRequest,
+    CompleteInvocationRequest, CreateObserverTokenRequest, MessageListQuery, ObserverToken,
+    RegisterActionRequest, RelayCast, RelayCastOptions, ReleaseAgentRequest,
 };
 use serde_json::Value;
 
@@ -191,6 +191,25 @@ impl RelaycastHttpClient {
             .map_err(|error| anyhow::anyhow!("{error}"))
     }
 
+    /// Mint a scoped, read-only observer token for this workspace. Used by
+    /// the local HTTP API so callers (e.g. Pear's "Join as observer" link)
+    /// can hand out a narrow `ot_live_...` credential instead of the raw
+    /// `rk_live_...` workspace key, which grants full read/write/spawn
+    /// access. The raw token material is only present on this response (and
+    /// on `rotate_observer_token`'s), never on subsequent reads.
+    pub async fn create_observer_token(
+        &self,
+        request: CreateObserverTokenRequest,
+    ) -> Result<ObserverToken> {
+        let relay = self
+            .relay_client()
+            .context("SDK relay client not initialized")?;
+        relay
+            .create_observer_token(request)
+            .await
+            .map_err(|error| anyhow::anyhow!("{error}"))
+    }
+
     /// Fetch a single action invocation, including its `input`. The
     /// `action.invoked` WebSocket event omits the input payload, so the handler
     /// must read it back here before executing.