diff --git a/.changeset/sep-ws-rewire.md b/.changeset/sep-ws-rewire.md new file mode 100644 index 00000000..7c5724d0 --- /dev/null +++ b/.changeset/sep-ws-rewire.md @@ -0,0 +1,22 @@ +--- +'@smooai/smooth': minor +--- + +Talk to your org's Smooth Operator again — rewired over the SEP WebSocket. + +The buffered REST route `POST /organizations/{org}/smooth-operator/chat` was +deleted upstream (SMOODEV-2673), so both `ask_business` (the MCP tool Claude +Desktop / Cursor use) and `th api smooth-operator chat` were pointed at a dead +endpoint and 404'd. New `smooai::smooth_operator_ws::operator_turn` mints a +short-lived socket token from api-prime, connects +`wss://smooth-operator.smoo.ai/ws`, creates/resumes a conversation session, +sends the message, and buffers the streamed turn into a final reply — hand-rolled +because the `smooth-operator` crates are server-side and ship no Rust client. + +Destructive tools now confirm **inline**: the socket parks the turn +(`write_confirmation_required`) and the decision rides the same connection, so +approval is a flag (`approve: true` / `--confirm`) instead of a second call. +Without it the action is declined and reported, never silently run. The old +`th api smooth-operator confirm` subcommand is retired with an explanation. + +Verified live against production on both surfaces. diff --git a/Cargo.lock b/Cargo.lock index 5759833e..8527cb68 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2061,7 +2061,7 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", - "webpki-roots", + "webpki-roots 1.0.6", ] [[package]] @@ -4140,7 +4140,7 @@ dependencies = [ "wasm-bindgen-futures", "wasm-streams", "web-sys", - "webpki-roots", + "webpki-roots 1.0.6", ] [[package]] @@ -4918,6 +4918,7 @@ dependencies = [ "tempfile", "tiny_http", "tokio", + "tokio-tungstenite 0.26.2", "toml", "tracing", "tracing-subscriber", @@ -5744,8 +5745,12 @@ checksum = "7a9daff607c6d2bf6c16fd681ccb7eecc83e4e2cdc1ca067ffaadfca5de7f084" dependencies = [ "futures-util", "log", + "rustls", + "rustls-pki-types", "tokio", + "tokio-rustls", "tungstenite 0.26.2", + "webpki-roots 0.26.11", ] [[package]] @@ -6018,6 +6023,8 @@ dependencies = [ "httparse", "log", "rand 0.9.2", + "rustls", + "rustls-pki-types", "sha1", "thiserror 2.0.18", "utf-8", @@ -6468,6 +6475,15 @@ dependencies = [ "web-sys", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.6", +] + [[package]] name = "webpki-roots" version = "1.0.6" diff --git a/crates/smooth-cli/Cargo.toml b/crates/smooth-cli/Cargo.toml index 7f10b151..0e4514b8 100644 --- a/crates/smooth-cli/Cargo.toml +++ b/crates/smooth-cli/Cargo.toml @@ -87,6 +87,7 @@ base64 = "0.22" tempfile = "3" rmcp = { workspace = true, features = ["server", "macros", "transport-io"] } schemars = "1" +tokio-tungstenite = { version = "0.26", features = ["rustls-tls-webpki-roots"] } [dev-dependencies] # Pearl th-abc4e2: admin::tests verify the baked-in Supabase anon diff --git a/crates/smooth-cli/src/mcp_serve.rs b/crates/smooth-cli/src/mcp_serve.rs index 6905b8f7..e05abb79 100644 --- a/crates/smooth-cli/src/mcp_serve.rs +++ b/crates/smooth-cli/src/mcp_serve.rs @@ -16,11 +16,10 @@ //! - **Local, free, no sign-in**: `pearls_ready` / `pearls_create` (the //! workspace pearl store) and `remember` / `recall` (local memory). //! - **Your business, behind Sign in with Smoo (`th auth login`)**: -//! `ask_business` — one turn of the Smooth Operator org agent (the same -//! user-only `POST /organizations/{org}/smooth-operator/chat` the -//! `th api smooth-operator` CLI drives), which never sends or takes a -//! destructive action without explicit approval; and `knowledge_search` — -//! a fast read of the org knowledge base. +//! `ask_business` — one turn of the Smooth Operator org agent over its SEP +//! WebSocket transport (see `smooai::smooth_operator_ws`), which never sends +//! or takes a destructive action unless the caller passes `approve: true`; +//! and `knowledge_search` — a fast read of the org knowledge base. use anyhow::Result; use rmcp::{ @@ -38,8 +37,6 @@ use serde_json::json; use smooth_pearls::{MemoryStore, NewPearl, PearlType, Priority}; -use crate::smooai::user_client::UserClient; - /// The Smooth MCP server. Stateless beyond the tool router — each tool opens /// the pearl store fresh (matching `th pearls` CLI semantics), so the server /// tracks the workspace it's launched in with no long-lived DB handle. @@ -103,15 +100,14 @@ pub struct KnowledgeSearchArgs { /// Arguments for `ask_business` — one turn of the Smooth Operator org agent. #[derive(Debug, Deserialize, JsonSchema)] pub struct AskBusinessArgs { - /// What to ask or tell your business, in plain language. Omit only when - /// approving a pending action (set `approve` + `conversation_id` instead). - #[serde(default)] + /// What to ask or tell your business, in plain language. pub message: Option, - /// Continue an existing conversation (returned by a previous call). + /// Continue an existing conversation (pass back the id a previous call returned). #[serde(default)] pub conversation_id: Option, - /// Approve (true) or decline (false) an action the operator paused on. - /// Requires `conversation_id`. The operator never sends/acts without this. + /// Allow the operator to take destructive actions this turn — send email, + /// write to the CRM/knowledge base. Default false: it declines them and + /// tells you what it would have done, so you can re-run with approve=true. #[serde(default)] pub approve: Option, /// Act on a specific org id. Defaults to your active org. @@ -254,63 +250,37 @@ impl SmoothMcp { /// Talk to your business — one turn of the Smooth Operator org agent. /// /// The agent runs on your live Smoo org (CRM, inbox, knowledge, analytics) - /// and answers in plain language. It never sends email or takes a - /// destructive action without explicit approval: when it pauses on one, - /// this returns the pending action and a `conversation_id`; approve it by - /// calling again with `approve: true` and that `conversation_id`. + /// and answers in plain language, over its SEP WebSocket transport. It never + /// sends email or takes a destructive action unless you pass `approve: true` + /// — otherwise it declines the action, tells you what it would have done, and + /// returns a `conversation_id` so you can re-run with `approve: true` to + /// allow it (and continue the same thread). /// /// # Errors /// MCP error if not signed in to Smoo (user session), no active org, no - /// `message`/approval given, or the request fails. + /// `message`, or the operator call fails. #[tool( name = "ask_business", - description = "Talk to your business in plain language. Smooth Operator (the agent on your Smoo org) answers questions about revenue/CRM/knowledge and can draft — and, with your approval, send — email. Needs Sign in with Smoo (`th auth login`). To approve a paused action, call again with approve=true and the conversation_id it returned." + description = "Talk to your business in plain language. Smooth Operator (the agent on your Smoo org) answers questions about revenue/CRM/knowledge and can draft — and, only when you pass approve=true, send — email or other writes. Needs Sign in with Smoo (`th auth login`). Continue a thread by passing back the conversation_id it returns." )] pub async fn ask_business(&self, params: Parameters) -> Result { let a = params.0; - // `from_user_session` IS the gate: the org agent is a user-only route and - // 401s under M2M, so this errors (with a login hint) when there's no - // Smoo user session. - let client = UserClient::from_user_session().await.map_err(|e| sign_in_err(&e))?; + let message = a + .message + .as_deref() + .filter(|m| !m.trim().is_empty()) + .ok_or_else(|| ErrorData::invalid_params("Provide `message` to ask your business.".to_string(), None))?; let org = crate::active_org::resolve(a.org).map_err(|e| ErrorData::invalid_request(format!("No active Smoo org. {e}"), None))?; - let turn = if let (Some(approve), Some(cid)) = (a.approve, a.conversation_id.as_deref()) { - // Resolve a paused action the operator asked us to confirm. - client - .post( - &format!("/organizations/{org}/smooth-operator/confirm"), - &json!({ "conversationId": cid, "approve": approve }), - ) - .await - } else { - let message = a.message.as_deref().filter(|m| !m.trim().is_empty()).ok_or_else(|| { - ErrorData::invalid_params( - "Provide `message` to ask, or `approve` + `conversation_id` to resolve a pending action.".to_string(), - None, - ) - })?; - let mut body = json!({ "message": message }); - if let Some(cid) = &a.conversation_id { - body["conversationId"] = json!(cid); - } - client.post(&format!("/organizations/{org}/smooth-operator/chat"), &body).await - } - .map_err(|e| ErrorData::internal_error(e.to_string(), None))?; - - // Defense-in-depth on the approval path: a paused action with no - // conversation id can't be approved, so fail loudly rather than emit a - // send-prompt the user can never act on. (The API always returns one; - // this only fires on a contract violation.) - let paused = turn.get("pendingAction").is_some_and(|v| !v.is_null()); - let has_cid = turn.get("conversationId").and_then(|v| v.as_str()).is_some_and(|s| !s.is_empty()); - if paused && !has_cid { - return Err(ErrorData::internal_error( - "The operator paused on an action but returned no conversation id, so it can't be approved. Try again.".to_string(), - None, - )); - } + // The user session is minted into the SEP token inside `operator_turn`; + // it errors (with a `th auth login` hint) when there's no Smoo session. + // `approve` gates destructive tools inline: false (default) declines + + // surfaces; true allows them this turn. + let turn = crate::smooai::smooth_operator_ws::operator_turn(&org, message, a.conversation_id.as_deref(), a.approve.unwrap_or(false)) + .await + .map_err(|e| ErrorData::internal_error(format!("{e:#}"), None))?; - Ok(render_turn(&turn)) + Ok(crate::smooai::smooth_operator_ws::render_operator_turn(&turn)) } } @@ -343,37 +313,6 @@ fn sign_in_err(e: &anyhow::Error) -> ErrorData { ) } -/// Render one Smooth Operator turn for the MCP client: the reply, any tools it -/// used, an approval prompt for a paused action, and the conversation id so the -/// caller can continue or approve. -fn render_turn(turn: &serde_json::Value) -> String { - let reply = turn.get("reply").and_then(|v| v.as_str()).unwrap_or_default(); - let cid = turn.get("conversationId").and_then(|v| v.as_str()).unwrap_or_default(); - - let mut out = String::new(); - if !reply.is_empty() { - out.push_str(reply); - out.push('\n'); - } - if let Some(tools) = turn.get("toolCalls").and_then(|v| v.as_array()).filter(|t| !t.is_empty()) { - let names: Vec<&str> = tools.iter().filter_map(|t| t.get("name").and_then(|v| v.as_str())).collect(); - if !names.is_empty() { - let _ = writeln!(out, "\n_(used: {})_", names.join(", ")); - } - } - if let Some(pa) = turn.get("pendingAction").filter(|v| !v.is_null()) { - let summary = pa.get("summary").and_then(|v| v.as_str()).unwrap_or("an action"); - let _ = writeln!( - out, - "\n⏸ Needs your approval before it happens: {summary}\n To approve, call ask_business again with approve=true and conversation_id=\"{cid}\" (or approve=false to decline)." - ); - } - if !cid.is_empty() { - let _ = writeln!(out, "\n[conversation_id: {cid}]"); - } - out.trim().to_string() -} - #[tool_handler(router = self.tool_router)] impl ServerHandler for SmoothMcp { fn get_info(&self) -> ServerInfo { @@ -473,32 +412,4 @@ mod tests { client.cancel().await.expect("client shutdown"); server.abort(); } - - /// The turn renderer surfaces reply, tool calls, a pending-action approval - /// prompt, and the conversation id — the safety-critical "never send without - /// approval" path is a pure function, so test it directly. - #[test] - fn render_turn_surfaces_pending_action_and_conversation() { - let turn = serde_json::json!({ - "conversationId": "c-42", - "reply": "Drafted the renewal.", - "toolCalls": [{ "name": "knowledge.search" }, { "name": "email.draft" }], - "pendingAction": { "name": "email.send", "summary": "send the renewal to Acme" } - }); - let out = render_turn(&turn); - assert!(out.contains("Drafted the renewal.")); - assert!(out.contains("knowledge.search") && out.contains("email.draft")); - assert!(out.contains("Needs your approval") && out.contains("send the renewal to Acme")); - assert!(out.contains("approve=true") && out.contains("c-42")); - assert!(out.contains("[conversation_id: c-42]")); - } - - #[test] - fn render_turn_plain_reply_has_no_approval_prompt() { - let turn = serde_json::json!({ "conversationId": "c-1", "reply": "Revenue is up 12%." }); - let out = render_turn(&turn); - assert!(out.contains("Revenue is up 12%.")); - assert!(!out.contains("Needs your approval")); - assert!(out.contains("[conversation_id: c-1]")); - } } diff --git a/crates/smooth-cli/src/smooai/mod.rs b/crates/smooth-cli/src/smooai/mod.rs index 0cb2ad80..0752ecf5 100644 --- a/crates/smooth-cli/src/smooai/mod.rs +++ b/crates/smooth-cli/src/smooai/mod.rs @@ -26,6 +26,7 @@ pub mod observability; pub mod products; pub mod profile; pub mod smooth_operator; +pub mod smooth_operator_ws; pub mod teams; pub mod testing; pub mod user_client; diff --git a/crates/smooth-cli/src/smooai/smooth_operator.rs b/crates/smooth-cli/src/smooai/smooth_operator.rs index 1bbc96a7..ae767a23 100644 --- a/crates/smooth-cli/src/smooai/smooth_operator.rs +++ b/crates/smooth-cli/src/smooai/smooth_operator.rs @@ -4,34 +4,32 @@ //! 401 under an M2M client) — every tool run is audit-logged against the //! real person. Pearl th-f15107; smooai PR #2383. //! -//! Three subcommands mirror the three routes: -//! chat POST /organizations/{org}/smooth-operator/chat {message, conversationId?} -//! confirm POST /organizations/{org}/smooth-operator/confirm {conversationId, approve} -//! history GET /organizations/{org}/smooth-operator/conversations/{id} +//! **Transport (SMOODEV-2673):** the buffered REST chat/confirm routes were +//! deleted; the operator is now driven over its **SEP WebSocket**. `chat` mints +//! a short-lived socket token and runs one turn via +//! [`crate::smooai::smooth_operator_ws::operator_turn`]. Destructive tools park +//! the turn mid-flight and are confirmed **inline** on the same socket, so +//! approval is a flag on `chat` (`--confirm`) rather than a follow-up command — +//! the old `confirm` subcommand is retired and now explains the change. +//! Without `--confirm`, destructive actions are declined and reported, never +//! silently run. //! -//! Responses are buffered JSON (`SmoothOperatorTurnResult`) — token streaming is -//! phase 2 on the smooai side. A turn may return a `pendingAction`: a -//! destructive tool (e.g. `email.send`) the loop paused on. `chat` resolves -//! it inline — a y/N prompt on a TTY, or the up-front `--confirm`/ -//! `--no-confirm` flag for non-interactive/agent use. `--no-confirm` is -//! never a default: without a flag on a non-TTY we print the pending action -//! and stop rather than silently approving or declining. - -use std::io::IsTerminal; +//! chat SEP WS wss://smooth-operator.smoo.ai/ws (token from api-prime) +//! history GET /organizations/{org}/smooth-operator/conversations/{id} use anyhow::{Context, Result}; use clap::Subcommand; use owo_colors::OwoColorize; use serde::Deserialize; -use serde_json::{json, Value}; +use serde_json::json; use super::print_json; use crate::smooai::user_client::UserClient; #[derive(Subcommand)] pub enum Cmd { - /// Send a message to the org smooth-operator and print its reply. Resolves any - /// destructive-action confirmation inline (TTY prompt or `--confirm`/`--no-confirm`). + /// Send a message to the org smooth-operator and print its reply. Runs one + /// turn over the SEP WebSocket; destructive actions run only with `--confirm`. Chat { /// The message to send to the smooth-operator. message: String, @@ -51,8 +49,8 @@ pub enum Cmd { #[arg(long)] json: bool, }, - /// Approve or decline the destructive action a prior `chat` turn paused on, - /// without resending the message. Use this to inspect first, then decide. + /// RETIRED — the operator now confirms destructive actions inline during the + /// turn. Use `chat --confirm` instead; this command explains the change. Confirm { /// The conversation id from the `chat` turn that returned a pendingAction. conversation_id: String, @@ -82,36 +80,7 @@ pub enum Cmd { }, } -// ---- API response shapes (subset of smooai SmoothOperatorTurnResult / SmoothOperatorHistory) ---- - -#[derive(Deserialize)] -#[serde(rename_all = "camelCase")] -struct TurnResult { - conversation_id: String, - #[serde(default)] - reply: String, - #[serde(default)] - tool_calls: Vec, - #[serde(default)] - pending_action: Option, -} - -#[derive(Deserialize)] -#[serde(rename_all = "camelCase")] -struct ToolCall { - name: String, - /// `ran` | `pending` | `error`. - #[serde(default)] - status: String, -} - -#[derive(Deserialize)] -#[serde(rename_all = "camelCase")] -struct PendingAction { - name: String, - #[serde(default)] - summary: String, -} +// ---- API response shapes (subset of smooai SmoothOperatorHistory) ---- #[derive(Deserialize)] #[serde(rename_all = "camelCase")] @@ -131,35 +100,6 @@ struct HistoryMessage { tool_name: Option, } -/// What to do with a `pendingAction`, given the flags and whether we're on a TTY. -#[derive(Debug, PartialEq, Eq)] -enum Decision { - Approve, - Decline, - /// Ask the user interactively (TTY only). - Prompt, - /// Non-TTY with no flag — refuse to guess; print the action and stop. - NeedFlag, -} - -/// Pure confirm-flag resolution. clap already rejects `--confirm --no-confirm` -/// together, so the `(true, true)` arm can't be reached in practice. -fn decide(confirm: bool, no_confirm: bool, is_tty: bool) -> Decision { - match (confirm, no_confirm) { - (true, _) => Decision::Approve, - (_, true) => Decision::Decline, - _ if is_tty => Decision::Prompt, - _ => Decision::NeedFlag, - } -} - -/// One compact line per tool call, e.g. `ran crm.search_contacts`. The status -/// (`ran`/`pending`/`error`) reads as the verb. Glyph-free so it's cheap to -/// test; the print path colors the leading verb. -fn tool_call_line(tc: &ToolCall) -> String { - format!("{} {}", tc.status, tc.name) -} - fn resolve_org(override_org: Option) -> Result { if let Some(o) = override_org.filter(|s| !s.trim().is_empty()) { return Ok(o); @@ -182,123 +122,42 @@ pub async fn cmd(cmd: Cmd) -> Result<()> { confirm, no_confirm, json, - } => chat(&client, resolve_org(org)?, message, conversation, confirm, no_confirm, json).await, - Cmd::Confirm { - conversation_id, - approve, - decline, - org, - json, - } => { - if !approve && !decline { - anyhow::bail!("pass `--approve` or `--decline`"); - } - confirm(&client, &resolve_org(org)?, &conversation_id, approve, json).await - } + } => chat(resolve_org(org)?, message, conversation, confirm, no_confirm, json).await, + // The separate confirm round-trip is obsolete: the SEP WebSocket parks + // the turn and takes the approval inline, so approval is a flag on + // `chat` now rather than a follow-up command. + Cmd::Confirm { .. } => anyhow::bail!( + "`confirm` is obsolete — the operator now confirms destructive actions inline over its WebSocket transport. \ + Re-run the ask with `th api smooth-operator chat \"…\" --confirm` to approve actions during the turn." + ), Cmd::History { conversation_id, org, json } => history(&client, &resolve_org(org)?, &conversation_id, json).await, } } -async fn chat(client: &UserClient, org: String, message: String, conversation: Option, confirm_flag: bool, no_confirm: bool, json: bool) -> Result<()> { - let mut body = json!({ "message": message }); - if let Some(cid) = conversation.filter(|s| !s.trim().is_empty()) { - body["conversationId"] = Value::String(cid); - } - let raw = client - .post(&format!("/organizations/{org}/smooth-operator/chat"), &body) +/// One operator turn over the SEP WebSocket. The buffered REST chat route was +/// deleted in SMOODEV-2673; `operator_turn` mints the socket token and drives +/// the session. `--confirm` approves destructive tools inline; otherwise they +/// are declined and reported. +async fn chat(org: String, message: String, conversation: Option, confirm_flag: bool, no_confirm: bool, json: bool) -> Result<()> { + let approve = confirm_flag && !no_confirm; + let turn = crate::smooai::smooth_operator_ws::operator_turn(&org, &message, conversation.as_deref().filter(|s| !s.trim().is_empty()), approve) .await - .context("POST smooth-operator chat")?; - resolve_turn(client, &org, raw, confirm_flag, no_confirm, json).await -} - -async fn confirm(client: &UserClient, org: &str, conversation_id: &str, approve: bool, json: bool) -> Result<()> { - let raw = post_confirm(client, org, conversation_id, approve).await?; + .context("smooth-operator turn")?; if json { - print_json(&raw); - return Ok(()); + println!( + "{}", + serde_json::to_string_pretty(&json!({ + "reply": turn.reply, + "conversationId": turn.conversation_id, + "declined": turn.declined, + }))? + ); + } else { + println!("{}", crate::smooai::smooth_operator_ws::render_operator_turn(&turn)); } - render_turn(&parse_turn(&raw)?); Ok(()) } -async fn post_confirm(client: &UserClient, org: &str, conversation_id: &str, approve: bool) -> Result { - let body = json!({ "conversationId": conversation_id, "approve": approve }); - client - .post(&format!("/organizations/{org}/smooth-operator/confirm"), &body) - .await - .context("POST smooth-operator confirm") -} - -/// Render a turn, then keep resolving any `pendingAction` (a turn can pause on -/// several destructive tools) until the loop settles or we hit a non-TTY with -/// no decision flag. -async fn resolve_turn(client: &UserClient, org: &str, raw: Value, confirm_flag: bool, no_confirm: bool, json: bool) -> Result<()> { - if json { - print_json(&raw); - // In JSON mode we don't auto-confirm — the caller inspects the - // pendingAction and drives `confirm` itself. - return Ok(()); - } - let mut turn = parse_turn(&raw)?; - loop { - render_turn(&turn); - let Some(pending) = turn.pending_action.as_ref() else { return Ok(()) }; - let approve = match decide(confirm_flag, no_confirm, std::io::stdin().is_terminal()) { - Decision::Approve => true, - Decision::Decline => false, - Decision::Prompt => prompt_confirm(pending)?, - Decision::NeedFlag => { - println!( - " {} pending action needs a decision — re-run with {} or {}, or `th api smooth-operator confirm {} --approve|--decline`", - "!".yellow().bold(), - "--confirm".bold(), - "--no-confirm".bold(), - turn.conversation_id.cyan(), - ); - return Ok(()); - } - }; - let raw = post_confirm(client, org, &turn.conversation_id, approve).await?; - turn = parse_turn(&raw)?; - } -} - -fn prompt_confirm(pending: &PendingAction) -> Result { - dialoguer::Confirm::new() - .with_prompt(format!("Approve {} — {}?", pending.name, pending.summary)) - .default(false) - .interact() - .context("read confirmation") -} - -fn parse_turn(raw: &Value) -> Result { - serde_json::from_value(raw.clone()).context("parse smooth-operator turn result") -} - -fn render_turn(turn: &TurnResult) { - println!(); - for tc in &turn.tool_calls { - let line = tool_call_line(tc); - let glyph = match tc.status.as_str() { - "error" => "✗".red().to_string(), - "pending" => "…".yellow().to_string(), - _ => "✓".green().to_string(), - }; - println!(" {} {}", glyph, line.dimmed()); - } - if !turn.reply.trim().is_empty() { - if !turn.tool_calls.is_empty() { - println!(); - } - println!("{}", turn.reply); - } - if let Some(p) = &turn.pending_action { - println!(); - println!(" {} {} — {}", "⚠".yellow().bold(), p.name.bold(), p.summary); - } - println!(); -} - async fn history(client: &UserClient, org: &str, conversation_id: &str, json: bool) -> Result<()> { let raw = client .get(&format!("/organizations/{org}/smooth-operator/conversations/{conversation_id}")) @@ -330,73 +189,11 @@ async fn history(client: &UserClient, org: &str, conversation_id: &str, json: bo #[cfg(test)] mod tests { - use super::{decide, resolve_org, tool_call_line, Decision, ToolCall}; - - fn tc(name: &str, status: &str) -> ToolCall { - ToolCall { - name: name.into(), - status: status.into(), - } - } - - #[test] - fn confirm_flag_wins_over_tty() { - assert_eq!(decide(true, false, true), Decision::Approve); - assert_eq!(decide(true, false, false), Decision::Approve); - } - - #[test] - fn no_confirm_flag_declines() { - assert_eq!(decide(false, true, true), Decision::Decline); - assert_eq!(decide(false, true, false), Decision::Decline); - } - - #[test] - fn tty_with_no_flag_prompts() { - assert_eq!(decide(false, false, true), Decision::Prompt); - } - - #[test] - fn non_tty_with_no_flag_refuses_to_guess() { - // The safety guarantee: never auto-approve or auto-decline for an agent. - assert_eq!(decide(false, false, false), Decision::NeedFlag); - } - - #[test] - fn tool_call_line_is_compact_verb_plus_name() { - assert_eq!(tool_call_line(&tc("crm.search_contacts", "ran")), "ran crm.search_contacts"); - assert_eq!(tool_call_line(&tc("email.send", "pending")), "pending email.send"); - assert_eq!(tool_call_line(&tc("crm.create_contact", "error")), "error crm.create_contact"); - // Unknown status is passed through, not dropped. - assert_eq!(tool_call_line(&tc("x.y", "weird")), "weird x.y"); - } + use super::resolve_org; #[test] fn resolve_org_prefers_flag_then_env() { assert_eq!(resolve_org(Some("org-flag".into())).unwrap(), "org-flag"); assert!(resolve_org(Some(" ".into())).is_err() || resolve_org(None).is_err()); } - - #[test] - fn parse_turn_reads_camelcase_and_defaults() { - // Full shape. - let full = serde_json::json!({ - "conversationId": "c1", - "reply": "found 3", - "toolCalls": [{ "name": "crm.search_contacts", "input": {}, "status": "ran" }], - "pendingAction": { "toolCallId": "t1", "name": "email.send", "input": {}, "summary": "send to jane" } - }); - let t = super::parse_turn(&full).unwrap(); - assert_eq!(t.conversation_id, "c1"); - assert_eq!(t.tool_calls.len(), 1); - assert_eq!(t.pending_action.unwrap().name, "email.send"); - - // Minimal shape — only conversationId; the rest default. - let min = serde_json::json!({ "conversationId": "c2" }); - let t = super::parse_turn(&min).unwrap(); - assert_eq!(t.conversation_id, "c2"); - assert!(t.reply.is_empty()); - assert!(t.tool_calls.is_empty()); - assert!(t.pending_action.is_none()); - } } diff --git a/crates/smooth-cli/src/smooai/smooth_operator_ws.rs b/crates/smooth-cli/src/smooai/smooth_operator_ws.rs new file mode 100644 index 00000000..06214a7a --- /dev/null +++ b/crates/smooth-cli/src/smooai/smooth_operator_ws.rs @@ -0,0 +1,280 @@ +//! SEP WebSocket turn helper — talk to the org Smooth Operator over its +//! WebSocket transport. +//! +//! The buffered REST path (`POST /organizations/{org}/smooth-operator/chat`) was +//! deleted in SMOODEV-2673; the supported transport is now the SEP WebSocket. +//! There is no importable Rust client in the `smooth-operator` crates (they are +//! server-side only), so this hand-rolls one buffered turn against the wire +//! shapes in `smooth-operator-server`'s `protocol.rs` / `handler.rs`: +//! +//! 1. mint a short-lived ES256 token — `POST /organizations/{org}/smooth-operator/token` +//! (authed as the user session) → `{ token, expiresAt }`; +//! 2. connect `wss://smooth-operator.smoo.ai/ws?token=…`; +//! 3. `create_conversation_session` (pass `conversationId` to resume) → the +//! `immediate_response` carries `data.sessionId`; +//! 4. `send_message { sessionId, message, stream: true }`; +//! 5. read frames, ignoring `stream_*`; on `write_confirmation_required` reply +//! `confirm_tool_action { approved }` (never send/act without an explicit +//! `approve = true`); stop on `eventual_response` and return its reply. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use anyhow::{anyhow, Context, Result}; +use futures_util::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use tokio::net::TcpStream; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; + +use crate::smooai::user_client::UserClient; + +const DEFAULT_WS_URL: &str = "wss://smooth-operator.smoo.ai/ws"; +/// Per-frame read timeout — a healthy turn streams frequently; this bounds a +/// hung server without cutting off a long-running tool. +const FRAME_TIMEOUT: Duration = Duration::from_secs(90); + +type Ws = WebSocketStream>; + +/// The buffered result of one operator turn. +pub struct OperatorTurn { + /// The operator's final reply text. + pub reply: String, + /// The conversation id — pass it back to continue the thread. + pub conversation_id: String, + /// Human-readable descriptions of destructive actions the operator paused on + /// that we DECLINED (because `approve` was false). Surfaced so the caller can + /// re-run with `approve = true` to allow them. + pub declined: Vec, +} + +fn ws_url() -> String { + std::env::var("SMOOTH_OPERATOR_WS_URL").unwrap_or_else(|_| DEFAULT_WS_URL.to_string()) +} + +fn request_id() -> String { + static RID: AtomicU64 = AtomicU64::new(1); + format!("th-{}-{}", std::process::id(), RID.fetch_add(1, Ordering::Relaxed)) +} + +/// Run one buffered turn of the org Smooth Operator over the SEP WebSocket. +/// +/// `approve` gates destructive tools: when the operator pauses on one +/// (`write_confirmation_required`), it runs only if `approve` is true; otherwise +/// it is declined and recorded in [`OperatorTurn::declined`]. Declining is the +/// safe default for a headless client. +/// +/// # Errors +/// Returns an error if there is no Smoo user session, the token mint fails, the +/// WebSocket can't connect, the turn times out, or the operator emits an error / +/// an interactive step this client can't satisfy (OTP, rich interaction). +pub async fn operator_turn(org: &str, message: &str, conversation_id: Option<&str>, approve: bool) -> Result { + // 1. Mint the short-lived SEP token from api-prime (user session only). + let http = UserClient::from_user_session().await.context("sign in to Smoo (run `th auth login`)")?; + let minted = http + .post(&format!("/organizations/{org}/smooth-operator/token"), &json!({})) + .await + .context("mint smooth-operator token")?; + let token = minted + .get("token") + .and_then(Value::as_str) + .ok_or_else(|| anyhow!("token endpoint returned no token"))?; + + // 2. Connect. + let (mut ws, _resp) = connect_async(format!("{}?token={token}", ws_url())) + .await + .context("connect smooth-operator websocket")?; + + // 3. Create or resume the conversation session. + let mut create = json!({ "action": "create_conversation_session", "requestId": request_id() }); + if let Some(cid) = conversation_id { + create["conversationId"] = json!(cid); + } + send(&mut ws, &create).await?; + let created = recv_until(&mut ws, "immediate_response").await?; + let session_id = created + .pointer("/data/sessionId") + .and_then(Value::as_str) + .ok_or_else(|| anyhow!("session create returned no sessionId"))? + .to_string(); + let conv_id = created.pointer("/data/conversationId").and_then(Value::as_str).unwrap_or_default().to_string(); + + // 4. Send the message. + let send_rid = request_id(); + send( + &mut ws, + &json!({ "action": "send_message", "requestId": send_rid, "sessionId": session_id, "message": message, "stream": true }), + ) + .await?; + + // 5. Buffer the streamed turn. + let mut declined = Vec::new(); + loop { + let ev = recv(&mut ws).await?; + match ev.get("type").and_then(Value::as_str) { + Some("write_confirmation_required") => { + let rid = ev.get("requestId").and_then(Value::as_str).unwrap_or(&send_rid).to_string(); + let desc = ev + .pointer("/data/data/actionDescription") + .and_then(Value::as_str) + .unwrap_or("an action") + .to_string(); + if !approve { + declined.push(desc); + } + send( + &mut ws, + &json!({ "action": "confirm_tool_action", "requestId": rid, "sessionId": session_id, "approved": approve }), + ) + .await?; + } + Some("eventual_response") => { + let reply = extract_reply(ev.pointer("/data/data/response")); + let _ = ws.close(None).await; + return Ok(OperatorTurn { + reply, + conversation_id: conv_id, + declined, + }); + } + Some("otp_verification_required" | "interaction_required") => { + let _ = ws.close(None).await; + return Err(anyhow!( + "the operator needs an interactive step (verification / rich interaction) that isn't supported from this client" + )); + } + Some("error") => { + let msg = ev.get("message").and_then(Value::as_str).unwrap_or("unknown operator error"); + return Err(anyhow!("smooth-operator error: {msg}")); + } + // stream_preamble / stream_chunk / the send ack — ignore. + _ => {} + } + } +} + +async fn send(ws: &mut Ws, value: &Value) -> Result<()> { + ws.send(Message::Text(value.to_string().into())).await.context("send websocket frame") +} + +async fn recv(ws: &mut Ws) -> Result { + let frame = tokio::time::timeout(FRAME_TIMEOUT, ws.next()) + .await + .context("timed out waiting for the operator")? + .ok_or_else(|| anyhow!("operator websocket closed unexpectedly"))? + .context("operator websocket error")?; + match frame { + Message::Text(t) => serde_json::from_str(&t).context("parse operator event"), + Message::Close(_) => Err(anyhow!("operator websocket closed")), + // ping/pong/binary — not an event; caller loops. + _ => Ok(json!({ "type": "_ignore" })), + } +} + +async fn recv_until(ws: &mut Ws, ty: &str) -> Result { + loop { + let ev = recv(ws).await?; + match ev.get("type").and_then(Value::as_str) { + Some(t) if t == ty => return Ok(ev), + Some("error") => { + let msg = ev.get("message").and_then(Value::as_str).unwrap_or("unknown"); + return Err(anyhow!("smooth-operator error: {msg}")); + } + _ => {} + } + } +} + +/// Extract the reply text from an `eventual_response`'s `response` value. The +/// engine returns `{ responseParts: [...] }`; parts may be strings or +/// `{ text | content }` objects. Falls back to the raw value. +fn extract_reply(response: Option<&Value>) -> String { + let Some(resp) = response else { + return String::new(); + }; + if let Some(parts) = resp.get("responseParts").and_then(Value::as_array) { + let mut out = String::new(); + for p in parts { + if let Some(s) = p.as_str() { + out.push_str(s); + } else if let Some(s) = p.get("text").and_then(Value::as_str) { + out.push_str(s); + } else if let Some(s) = p.get("content").and_then(Value::as_str) { + out.push_str(s); + } + } + if !out.trim().is_empty() { + return out.trim().to_string(); + } + } + if let Some(s) = resp.as_str() { + return s.to_string(); + } + if let Some(s) = resp.get("text").and_then(Value::as_str) { + return s.to_string(); + } + resp.to_string() +} + +/// Render a buffered turn for a text client (the MCP tool / CLI): reply, any +/// declined-pending-approval actions, and the conversation id to continue. +#[must_use] +pub fn render_operator_turn(turn: &OperatorTurn) -> String { + use std::fmt::Write as _; + let mut out = turn.reply.clone(); + for d in &turn.declined { + let _ = write!( + out, + "\n\n⏸ I did NOT do this without your approval: {d}\n To allow it, re-run with approve=true (conversation_id=\"{}\").", + turn.conversation_id + ); + } + if !turn.conversation_id.is_empty() { + let _ = write!(out, "\n\n[conversation_id: {}]", turn.conversation_id); + } + out.trim().to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn extract_reply_from_response_parts() { + let resp = json!({ "responseParts": [{ "text": "Revenue is " }, { "text": "up 12%." }] }); + assert_eq!(extract_reply(Some(&resp)), "Revenue is up 12%."); + } + + #[test] + fn extract_reply_falls_back_to_string() { + assert_eq!(extract_reply(Some(&json!("plain text"))), "plain text"); + assert_eq!(extract_reply(None), ""); + } + + #[test] + fn render_surfaces_declined_action_and_conversation() { + let turn = OperatorTurn { + reply: "Drafted the renewal.".to_string(), + conversation_id: "c-9".to_string(), + declined: vec!["send the renewal email to Acme".to_string()], + }; + let out = render_operator_turn(&turn); + assert!(out.contains("Drafted the renewal.")); + assert!(out.contains("did NOT do this without your approval: send the renewal email to Acme")); + assert!(out.contains("approve=true")); + assert!(out.contains("[conversation_id: c-9]")); + } + + #[test] + fn render_plain_reply_has_no_approval_note() { + let turn = OperatorTurn { + reply: "All good.".to_string(), + conversation_id: "c-1".to_string(), + declined: vec![], + }; + let out = render_operator_turn(&turn); + assert!(out.contains("All good.")); + assert!(!out.contains("approval")); + assert!(out.contains("[conversation_id: c-1]")); + } +} diff --git a/docs/Engineering/Using-th-CLI.md b/docs/Engineering/Using-th-CLI.md index fdc1c2c1..fd540524 100644 --- a/docs/Engineering/Using-th-CLI.md +++ b/docs/Engineering/Using-th-CLI.md @@ -402,18 +402,25 @@ th api smooth-operator chat "..." --json # th api smooth-operator history # message history ``` -Destructive tools (e.g. `email.send`) **never auto-run** — a turn that triggers -one returns a `pendingAction` and pauses. `chat` resolves it with a y/N prompt -on a TTY, or the up-front `--confirm` / `--no-confirm` flag for -non-interactive/agent use. With **no flag on a non-TTY** it prints the pending -action and stops rather than guessing — `--no-confirm` is never a silent -default. To inspect first, then approve without resending the message: +> **Transport (SMOODEV-2673).** The buffered REST `chat`/`confirm` routes were +> **deleted**. `chat` now mints a short-lived socket token +> (`POST /organizations/{org}/smooth-operator/token`) and runs one turn over the +> **SEP WebSocket** (`wss://smooth-operator.smoo.ai/ws`) — +> `create_conversation_session` (resume by id) → `send_message` → await +> `eventual_response`. Implementation: `crates/smooth-cli/src/smooai/smooth_operator_ws.rs` +> (hand-rolled: the `smooth-operator` crates are server-side and ship no Rust client). + +Destructive tools (e.g. `email.send`) **never auto-run**. The socket *parks the +turn mid-flight* (`write_confirmation_required`) and takes the decision +**inline**, so approval is a flag on `chat` rather than a second command: +without `--confirm` the action is declined and reported ("I did NOT do this +without your approval"), and the turn still completes. The old +`th api smooth-operator confirm` subcommand is **retired** — it now prints an +explanation pointing at `--confirm`. ```bash -th api smooth-operator chat "Send jane@acme.com the follow-up" # pauses, prints the pending email.send -th api smooth-operator confirm --approve # run it -th api smooth-operator confirm --decline # or drop it -th api smooth-operator chat "Send jane@acme.com the follow-up" --confirm # one-shot, when already authorized +th api smooth-operator chat "Send jane@acme.com the follow-up" # drafts; declines the send, tells you +th api smooth-operator chat "Send jane@acme.com the follow-up" --confirm # allows the send this turn ``` Responses are buffered JSON (token streaming is phase 2). Every tool run is @@ -567,7 +574,7 @@ For agents collaborating across **different clones/machines** of the same repo, `th mcp serve` speaks JSON-RPC on stdout (built on the `rmcp` SDK) — **do not mix other output onto stdout**; the tools log only to stderr. It exposes two tiers: - **Local — free, no sign-in.** `pearls_ready` / `pearls_create` act on the pearl store of the workspace the host launched the server in; `remember` / `recall` keep local notes. -- **Your business — behind Sign in with Smoo (`th auth login`).** `ask_business` is the star: one turn of **Smooth Operator**, the org agent (the same user-only `POST /organizations/{org}/smooth-operator/chat` the `th api smooth-operator` CLI drives) — ask about revenue/CRM/knowledge and draft, or with **explicit approval**, send email. It resolves your active org automatically, and never sends or takes a destructive action without approval: when it pauses on one, it returns the pending action + a `conversation_id`; approve by calling `ask_business` again with `approve=true` and that id. `knowledge_search` is a fast read of the org knowledge base. Both gate on the user session (they 401 under M2M), so unauthenticated calls return a clear "run `th auth login`" message rather than failing opaquely. +- **Your business — behind Sign in with Smoo (`th auth login`).** `ask_business` is the star: one turn of **Smooth Operator**, the org agent, over the SEP WebSocket (the same transport the `th api smooth-operator` CLI now drives) — ask about revenue/CRM/knowledge and draft, or with **explicit approval**, send email. It resolves your active org automatically, and never sends or takes a destructive action without approval: when it pauses on one, it returns the pending action + a `conversation_id`; approve by calling `ask_business` again with `approve=true` and that id. `knowledge_search` is a fast read of the org knowledge base. Both gate on the user session (they 401 under M2M), so unauthenticated calls return a clear "run `th auth login`" message rather than failing opaquely. The `.mcpb` **Desktop Extension** for one-click install lives in `packaging/mcpb/` (`build-mcpb.sh` stages the `th` binary + manifest and runs `npx @anthropic-ai/mcpb pack`). The same tool layer is what a hosted Streamable-HTTP server at `mcp.smoo.ai` will reuse for the zero-install Claude Desktop connector (pearl th-794b1e).