feat(companion): expose relay/tab handles for embedding hosts (OpenHuman wiring)

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<Vec<WorkflowSummary>, 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<String, String>;
+
+    /// 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;
+}

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};

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.
@@ -44,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<Arc<dyn super::CompanionRunHost>>,
 }
 
 /// Errors produced by companion configuration, I/O, or workflow startup.
@@ -77,6 +84,7 @@ struct ServerInner {
     pending: tokio::sync::Mutex<HashMap<String, PendingSender>>,
     workflows_dir: PathBuf,
     capabilities: Capabilities,
+    run_host: Option<Arc<dyn super::CompanionRunHost>>,
     runs: Mutex<HashMap<String, CancellationToken>>,
     next_session: AtomicU64,
     next_run: AtomicU64,
@@ -105,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),
@@ -138,6 +147,105 @@ 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<Vec<WorkflowSummary>, 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<String, String> {
+        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
+    /// 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<dyn BrowserRelay> {
+        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<SharedTab> {
+        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<String>,
+        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,
@@ -473,15 +581,15 @@ async fn native_workflows(State(server): State<CompanionServer>, 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
         }))
         .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(),
     }
@@ -496,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!({
@@ -506,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(),
     }
@@ -736,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}))

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());