From bef4df648f940307c96370688dd25f6aa2e2e9a8 Mon Sep 17 00:00:00 2001 From: graycyrus Date: Tue, 28 Jul 2026 21:56:30 +0530 Subject: [PATCH 1/2] feat(companion): expose browser_relay/is_extension_connected/shared_tabs for embedding hosts --- src/companion/server.rs | 67 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 66 insertions(+), 1 deletion(-) diff --git a/src/companion/server.rs b/src/companion/server.rs index e8f2d9f..c2dbab9 100644 --- a/src/companion/server.rs +++ b/src/companion/server.rs @@ -31,7 +31,8 @@ use crate::observability::{ExecutionStep, RunObserver, StepStatus}; use super::{ Authenticator, CompanionControlRequest, CompanionControlResponse, PROTOCOL_SUBPROTOCOL, - PairingSecret, RelayPolicy, RelayState, RunEvent, TabId, WebSocketHandshake, WorkflowSummary, + PairingSecret, RelayPolicy, RelayState, RunEvent, SharedTab, TabId, WebSocketHandshake, + WorkflowSummary, }; /// Configuration for the native Chrome companion. @@ -138,6 +139,70 @@ impl CompanionServer { list_workflows(&self.inner.workflows_dir) } + /// Returns a browser relay handle usable by an **external** workflow runner + /// (an embedding host that drives its own engine rather than calling + /// [`start_workflow`](Self::start_workflow)). The returned handle shares this + /// server's live WebSocket session and pending-response map, so wrapping it in + /// a [`RoutingToolInvoker`](crate::browser::RoutingToolInvoker) lets a host + /// route `slug:"browser"` tool calls to the paired extension. + /// + /// The handle is always valid; if no extension is currently connected each + /// `execute` fails closed with `relay_disconnected`. + pub fn browser_relay(&self) -> Arc { + Arc::new(SocketRelay { + inner: self.inner.clone(), + }) + } + + /// Whether a paired extension currently holds an authenticated relay session. + /// External hosts use this to gate author-time / run-time browser readiness. + pub fn is_extension_connected(&self) -> bool { + self.inner + .relay + .lock() + .map(|relay| relay.is_connected()) + .unwrap_or(false) + } + + /// Snapshot of the tabs the user has explicitly shared with the companion. + /// Empty when no extension is connected or nothing is shared. + pub fn shared_tabs(&self) -> Vec { + self.inner + .relay + .lock() + .map(|relay| relay.tabs().list().into_iter().cloned().collect()) + .unwrap_or_default() + } + + /// Binds a workflow run to an explicitly-shared tab so an **external** + /// runner's `slug:"browser"` calls (dispatched through the handle from + /// [`browser_relay`](Self::browser_relay)) are authorized against that tab. + /// This mirrors what [`start_workflow`](Self::start_workflow) does + /// internally for native runs — an embedding host must call this before + /// executing a graph that contains browser nodes, or every browser action + /// fails with `tab_not_shared`. + pub fn bind_run( + &self, + run_id: impl Into, + tab_id: TabId, + ) -> Result<(), CompanionServerError> { + self.inner + .relay + .lock() + .map_err(|_| lock_error())? + .tabs_mut() + .bind_run(run_id.into(), tab_id) + .map_err(super::RelayError::from)?; + Ok(()) + } + + /// Releases a run→tab binding after an external run settles. Idempotent. + pub fn unbind_run(&self, run_id: &str) { + if let Ok(mut relay) = self.inner.relay.lock() { + relay.tabs_mut().unbind_run(run_id); + } + } + /// Starts a native run bound to one explicit shared tab. pub async fn start_workflow( &self, From 71b602bac8691f3f3602edfc44a55c9e100acfd2 Mon Sep 17 00:00:00 2001 From: graycyrus Date: Wed, 29 Jul 2026 15:08:05 +0530 Subject: [PATCH 2/2] feat(companion): CompanionRunHost seam for embedding hosts Optional CompanionServerConfig::run_host lets an embedding host (OpenHuman's flows domain) own workflow listing + execution. When set, both the WS control channel (WorkflowList/Start/Cancel) and the HTTP native endpoints delegate to it via dispatch_{list_workflows,start_run,cancel_run}; None preserves the built-in workflows_dir + start_workflow behaviour for the standalone CLI. --- src/companion/host.rs | 39 +++++++++++++++++++++ src/companion/mod.rs | 2 ++ src/companion/server.rs | 75 ++++++++++++++++++++++++++++++++--------- src/main.rs | 1 + 4 files changed, 102 insertions(+), 15 deletions(-) create mode 100644 src/companion/host.rs diff --git a/src/companion/host.rs b/src/companion/host.rs new file mode 100644 index 0000000..82c99fe --- /dev/null +++ b/src/companion/host.rs @@ -0,0 +1,39 @@ +//! Host seam for embedding companions. +//! +//! By default the companion lists and runs workflows from its `workflows_dir` +//! (JSON files) via [`CompanionServer::start_workflow`](super::CompanionServer::start_workflow). +//! An embedding host (e.g. OpenHuman's flows domain) can instead own workflow +//! listing + execution by supplying a [`CompanionRunHost`] on +//! [`CompanionServerConfig::run_host`](super::CompanionServerConfig): when set, +//! the WS control channel and the HTTP native endpoints delegate to it, so +//! extension-initiated runs flow through the host's own engine (its run +//! registry, gates, and history) instead of the built-in path. + +use async_trait::async_trait; +use serde_json::Value; + +use super::{TabId, WorkflowSummary}; + +/// Lets an embedding host own the workflow-listing and run-execution the +/// companion exposes to the extension. All methods are called only for +/// already-authenticated control/native requests. +#[async_trait] +pub trait CompanionRunHost: Send + Sync { + /// The workflows the host chooses to expose to the extension side-panel + /// (e.g. only flows the user opted into browser-triggering). + async fn list_workflows(&self) -> Result, String>; + + /// Start a run of `workflow_id` bound to the explicitly-shared `tab_id`, + /// returning the host's own run id. The host is responsible for binding the + /// run to the tab and applying browser routing in its own engine. + async fn start_run( + &self, + workflow_id: &str, + tab_id: TabId, + input: Value, + ) -> Result; + + /// Cancel a previously started host run. Returns whether anything was + /// cancelled (mirrors [`CompanionServer::cancel_workflow`](super::CompanionServer::cancel_workflow)). + async fn cancel_run(&self, run_id: &str) -> bool; +} diff --git a/src/companion/mod.rs b/src/companion/mod.rs index e1ca542..dac7a15 100644 --- a/src/companion/mod.rs +++ b/src/companion/mod.rs @@ -7,6 +7,7 @@ mod auth; mod control; +mod host; mod relay; mod server; mod tabs; @@ -16,6 +17,7 @@ pub use auth::{ PairingSecret, SecretStore, WebSocketHandshake, }; pub use control::{CompanionControlRequest, CompanionControlResponse, RunEvent, WorkflowSummary}; +pub use host::CompanionRunHost; pub use relay::{DisconnectOutcome, PendingAction, RelayError, RelayPolicy, RelayState, SessionId}; pub use server::{CompanionServer, CompanionServerConfig, CompanionServerError}; pub use tabs::{RunBinding, RunId, SharedTab, TabId, TabRegistry, TabRegistryError}; diff --git a/src/companion/server.rs b/src/companion/server.rs index c2dbab9..060aac7 100644 --- a/src/companion/server.rs +++ b/src/companion/server.rs @@ -45,9 +45,15 @@ pub struct CompanionServerConfig { /// Host-local pairing secret required in a WebSocket subprotocol. pub pairing_secret: PairingSecret, /// Directory containing workflow JSON files exposed to the side panel. + /// Ignored for listing/running when `run_host` is set. pub workflows_dir: PathBuf, /// Host capabilities used for every non-browser effect. pub capabilities: Capabilities, + /// Optional embedding-host seam. When set, workflow listing + execution + /// (over both the WS control channel and the HTTP native endpoints) delegate + /// to the host instead of `workflows_dir` + [`CompanionServer::start_workflow`]. + /// `None` preserves the built-in standalone behaviour. + pub run_host: Option>, } /// Errors produced by companion configuration, I/O, or workflow startup. @@ -78,6 +84,7 @@ struct ServerInner { pending: tokio::sync::Mutex>, workflows_dir: PathBuf, capabilities: Capabilities, + run_host: Option>, runs: Mutex>, next_session: AtomicU64, next_run: AtomicU64, @@ -106,6 +113,7 @@ impl CompanionServer { pending: tokio::sync::Mutex::new(HashMap::new()), workflows_dir: config.workflows_dir, capabilities: config.capabilities, + run_host: config.run_host, runs: Mutex::new(HashMap::new()), next_session: AtomicU64::new(0), next_run: AtomicU64::new(0), @@ -139,6 +147,41 @@ impl CompanionServer { list_workflows(&self.inner.workflows_dir) } + /// Lists workflows via the configured [`CompanionRunHost`](super::CompanionRunHost) + /// if present, else from `workflows_dir`. Used by both request paths. + async fn dispatch_list_workflows(&self) -> Result, String> { + match &self.inner.run_host { + Some(host) => host.list_workflows().await, + None => self.workflows().map_err(|error| error.to_string()), + } + } + + /// Starts a run via the run host if present, else via the built-in + /// [`start_workflow`](Self::start_workflow). Returns the run id. + async fn dispatch_start_run( + &self, + workflow_id: &str, + tab_id: TabId, + input: Value, + ) -> Result { + match &self.inner.run_host { + Some(host) => host.start_run(workflow_id, tab_id, input).await, + None => self + .start_workflow(workflow_id, tab_id, input) + .await + .map_err(|error| error.to_string()), + } + } + + /// Cancels a run via the run host if present, else via the built-in + /// [`cancel_workflow`](Self::cancel_workflow). + async fn dispatch_cancel_run(&self, run_id: &str) -> bool { + match &self.inner.run_host { + Some(host) => host.cancel_run(run_id).await, + None => self.cancel_workflow(run_id).await, + } + } + /// Returns a browser relay handle usable by an **external** workflow runner /// (an embedding host that drives its own engine rather than calling /// [`start_workflow`](Self::start_workflow)). The returned handle shares this @@ -538,7 +581,7 @@ async fn native_workflows(State(server): State, headers: Header if !native_authorized(&server, &headers) { return StatusCode::UNAUTHORIZED.into_response(); } - match server.workflows() { + match server.dispatch_list_workflows().await { Ok(workflows) => Json(json!({ "protocol_version":BROWSER_PROTOCOL_VERSION, "workflows":workflows @@ -546,7 +589,7 @@ async fn native_workflows(State(server): State, headers: Header .into_response(), Err(error) => ( StatusCode::BAD_REQUEST, - Json(json!({"code":"workflow_list_failed","message":error.to_string()})), + Json(json!({"code":"workflow_list_failed","message":error})), ) .into_response(), } @@ -561,7 +604,7 @@ async fn native_run( return StatusCode::UNAUTHORIZED.into_response(); } match server - .start_workflow(&request.workflow_id, request.tab_id, request.input) + .dispatch_start_run(&request.workflow_id, request.tab_id, request.input) .await { Ok(run_id) => Json(json!({ @@ -571,7 +614,7 @@ async fn native_run( .into_response(), Err(error) => ( StatusCode::BAD_REQUEST, - Json(json!({"code":"workflow_start_failed","message":error.to_string()})), + Json(json!({"code":"workflow_start_failed","message":error})), ) .into_response(), } @@ -801,26 +844,28 @@ async fn handle_control( ); } match request { - CompanionControlRequest::WorkflowList { .. } => match server.workflows() { - Ok(workflows) => CompanionControlResponse::Workflows { - protocol_version: BROWSER_PROTOCOL_VERSION, - request_id, - workflows, - }, - Err(error) => control_error(request_id, "workflow_list_failed", &error.to_string()), - }, + CompanionControlRequest::WorkflowList { .. } => { + match server.dispatch_list_workflows().await { + Ok(workflows) => CompanionControlResponse::Workflows { + protocol_version: BROWSER_PROTOCOL_VERSION, + request_id, + workflows, + }, + Err(error) => control_error(request_id, "workflow_list_failed", &error), + } + } CompanionControlRequest::WorkflowStart { workflow_id, tab_id, input, .. - } => match server.start_workflow(&workflow_id, tab_id, input).await { + } => match server.dispatch_start_run(&workflow_id, tab_id, input).await { Ok(run_id) => control_ok(request_id, json!({"run_id":run_id})), - Err(error) => control_error(request_id, "workflow_start_failed", &error.to_string()), + Err(error) => control_error(request_id, "workflow_start_failed", &error), }, CompanionControlRequest::WorkflowCancel { run_id, .. } => control_ok( request_id, - json!({"cancelled":server.cancel_workflow(&run_id).await}), + json!({"cancelled":server.dispatch_cancel_run(&run_id).await}), ), CompanionControlRequest::RunSubscribe { run_id, .. } => { control_ok(request_id, json!({"subscribed":run_id})) diff --git a/src/main.rs b/src/main.rs index dc17e21..4ca9749 100644 --- a/src/main.rs +++ b/src/main.rs @@ -80,6 +80,7 @@ async fn start_companion(arguments: &[&str]) -> Result<(), String> { pairing_secret: secret, workflows_dir, capabilities: standalone_capabilities(), + run_host: None, }) .map_err(|error| error.to_string())?; eprintln!("TinyFlows companion listening on {}", server.bind_addr());