feat(medulla): serve the workflow plane to the hosted orchestrator (#5266)

.github/workflows/ci-lite.yml

@@ -464,6 +464,7 @@ jobs:
           openhuman/agent_registry/agents/loader.rs
           openhuman/composio/action_tool.rs
           openhuman/inference/local/mod.rs
+          openhuman/socket/ops.rs
           openhuman/tool_registry/ops_tests.rs
           openhuman/tool_registry/schemas.rs
           openhuman/tools/ops_tests.rs

src/core/jsonrpc.rs

@@ -2574,6 +2574,7 @@ pub async fn bootstrap_core_runtime(
 pub async fn start_core_runtime_services(
     services: crate::core::runtime::ServiceSet,
     config: Option<&crate::openhuman::config::Config>,
+    flows_enabled: bool,
 ) {
     let Some(cfg) = config else {
         log::error!(
@@ -2597,7 +2598,11 @@ pub async fn start_core_runtime_services(
 
     match crate::openhuman::socket::global_socket_manager() {
         Some(socket_mgr) => {
-            crate::core::runtime::services::spawn_socket_auto_connect(services, socket_mgr.clone());
+            crate::core::runtime::services::spawn_socket_auto_connect(
+                services,
+                socket_mgr.clone(),
+                flows_enabled,
+            );
         }
         None => {
             log::warn!(

src/core/runtime/builder.rs

@@ -702,7 +702,12 @@ impl CoreRuntime {
     /// each service keeps its own runtime config gate.
     async fn start_selected_services(&self) {
         use crate::core::runtime::services;
-        jsonrpc::start_core_runtime_services(self.services, self.config.as_ref()).await;
+        jsonrpc::start_core_runtime_services(
+            self.services,
+            self.config.as_ref(),
+            self.ctx.domains().flows,
+        )
+        .await;
 
         if self.services.heartbeat {
             services::spawn_login_gated_services(self.ctx.host_kind().is_desktop_shell());

src/core/runtime/services.rs

@@ -416,19 +416,20 @@ fn spawn_mcp_reconnect_supervisor(config: Config) {
 pub fn spawn_socket_auto_connect(
     services: ServiceSet,
     socket_mgr: std::sync::Arc<crate::openhuman::socket::SocketManager>,
+    _flows_enabled: bool,
 ) {
     if services.socketio {
         tokio::spawn(async move {
             log::info!("[socket] Checking for stored session to auto-connect...");
             let config = match Config::load_or_init().await {
-                Ok(c) => c,
+                Ok(c) => std::sync::Arc::new(c),
                 Err(e) => {
                     log::debug!("[socket] Config not available for auto-connect: {e}");
                     return;
                 }
             };
             let api_url = crate::api::config::effective_backend_api_url(&config.api_url);
-            let token = match crate::api::jwt::get_session_token(&config) {
+            let _initial_token = match crate::api::jwt::get_session_token(&config) {
                 Ok(Some(t)) => t,
                 Ok(None) => {
                     log::info!(
@@ -445,7 +446,22 @@ pub fn spawn_socket_auto_connect(
                 "[socket] Session token found — auto-connecting to {}",
                 api_url
             );
-            if let Err(e) = socket_mgr.connect(&api_url, &token).await {
+            // Keep the authenticated token and user-scoped workflow bridge in
+            // one serialized identity transaction. The active profile may have
+            // changed since CoreRuntime::build(), so the build-time Config is
+            // not authoritative here.
+            let _rebind = socket_mgr.lock_identity_rebind().await;
+            if let Err(e) = socket_mgr.disconnect().await {
+                log::error!("[socket] Auto-connect could not stop the prior connection: {e}");
+                return;
+            }
+            #[cfg(feature = "flows")]
+            if _flows_enabled {
+                crate::openhuman::flows::medulla_bridge::install(std::sync::Arc::clone(&config));
+            }
+            let provider =
+                crate::openhuman::socket::token_provider::token_provider_from_config(config);
+            if let Err(e) = socket_mgr.connect_with_provider(&api_url, provider).await {
                 log::error!("[socket] Auto-connect failed: {e}");
             } else {
                 log::info!("[socket] Auto-connect initiated successfully");

src/openhuman/credentials/ops.rs

@@ -702,6 +702,16 @@ pub async fn clear_session(config: &Config) -> Result<RpcOutcome<serde_json::Val
         .remove_profile(APP_SESSION_PROVIDER, DEFAULT_AUTH_PROFILE_NAME)
         .map_err(|e| e.to_string())?;
 
+    // The core process stays alive on logout. Tear down its authenticated
+    // Socket.IO transport and the user-pinned workflow bridge so neither can
+    // keep serving the signed-out account until a later reconnect.
+    if let Some(manager) = crate::openhuman::socket::global_socket_manager() {
+        if let Err(error) = manager.disconnect().await {
+            tracing::warn!(%error, "failed to disconnect backend socket on logout");
+        }
+    }
+    crate::openhuman::socket::medulla::workflows::clear_workflow_bridge();
+
     // Clear the active user marker so subsequent config loads fall back to the
     // default (unauthenticated) openhuman directory.
     if let Ok(root_dir) = default_root_openhuman_dir() {

src/openhuman/flows/medulla_bridge.rs

@@ -0,0 +1,575 @@
+//! Backs the medulla harness protocol's workflow plane
+//! ([`crate::openhuman::socket::medulla::workflows::WorkflowBridge`]) with this
+//! host's own `flows::` store.
+//!
+//! The transport half is host-agnostic on purpose — it moves adverts, opaque
+//! JSON and one authoring turn, and parses no graph. This module is the
+//! openhuman half: it projects saved [`Flow`]s onto the wire's
+//! [`WorkflowDescriptor`], answers the three reads out of [`super::ops`], and
+//! runs the `workflow_builder` agent for a `copilot` turn.
+//!
+//! **Why the copilot persists here.** In the desktop product a builder turn is
+//! deliberately propose-only: [`super::ops::flows_build`] returns a candidate
+//! graph and the *user* accepts it in the copilot panel, then saves it on the
+//! canvas. On the medulla plane there is no panel and no user — a remote
+//! orchestrator asked for a change and is waiting on the answer — so the accept
+//! is performed here, by the host, from the proposal the turn produced. Two
+//! guards keep that safe rather than merely convenient:
+//!
+//! - a **create** is persisted with `require_approval: true` unconditionally, so
+//!   every outbound action of a remotely-authored workflow still parks for a
+//!   human decision at run time;
+//! - an **update** passes `require_approval: None`, so a remote turn can raise
+//!   a workflow's approval requirement (via `flows_update`'s own side-effect
+//!   enforcement) but can never lower one the user set.
+//!
+//! `flows_create` additionally saves any automatic-trigger graph **disabled**,
+//! so nothing authored over this plane can start firing on its own.
+//!
+//! **Why the outcome is diffed, not reported.** [`WorkflowBridge::copilot`]'s
+//! `changes` are computed by [`diff_workflow`] over the workflow this turn
+//! successfully persisted, using listings taken before and after the turn —
+//! never from what the agent said it did. Scoping the diff matters because a
+//! copilot turn can run for minutes while a user edits unrelated workflows.
+//! That is the whole reason medulla-v1 forwards the outcome verbatim instead of
+//! re-deriving it: the claim that reaches the orchestrator is a statement about
+//! the store, made by the store's owner.
+//!
+//! **Threading.** The three reads are synchronous on the trait but every
+//! `flows::` op is `async`, so they bridge with a `Handle::block_on` — legal
+//! because the transport only ever calls them from `spawn_blocking` (see
+//! [`super::super::socket::medulla::workflows`]), never on a runtime worker.
+//! `copilot` is `async` end to end and needs no hop.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use serde_json::{json, Value};
+use tinyflows::model::WorkflowGraph;
+
+use super::agents::workflow_builder::builder_prompt::{BuildMode, BuilderRequest};
+use super::node_contracts::{all_node_kind_contracts, node_kind_contract, NODE_KINDS};
+use super::ops;
+use super::types::Flow;
+use crate::openhuman::config::Config;
+use crate::openhuman::socket::medulla::payloads::{CopilotOutcome, WorkflowDescriptor};
+use crate::openhuman::socket::medulla::workflows::{set_workflow_bridge, WorkflowBridge};
+
+/// How many runs a `runs` read returns. The orchestrator wants "did this
+/// workflow work lately", not an audit log, and the server caps a result at
+/// 1 MiB — a bounded window answers the question and can never hit that cap.
+const RUN_HISTORY_LIMIT: usize = 20;
+
+/// Cap on a generated advert description. It is prompt surface rendered for
+/// every workflow on every review pass, so a 40-node graph must not crowd out
+/// the rest of the section.
+const MAX_DESCRIPTION_CHARS: usize = 400;
+
+/// Persistence tools hidden from a medulla copilot turn.
+///
+/// The generic builder exposes these for its product surfaces, but this bridge
+/// deliberately owns persistence in [`apply_proposal`] so it can enforce the
+/// observed version and approval floor exactly once.
+const MEDULLA_COPILOT_HIDDEN_TOOLS: &[&str] = &[
+    "save_workflow",
+    "create_workflow",
+    "duplicate_flow",
+    "edit_workflow",
+];
+
+/// Install the flows-backed bridge on the medulla workflow plane.
+///
+/// Called once at boot, gated on the `flows` domain being live. Installing also
+/// advertises immediately, so a core that connected before this ran still
+/// publishes its workflows.
+pub fn install(config: Arc<Config>) {
+    log::debug!("[flows] installing the medulla workflow bridge over the flows store");
+    set_workflow_bridge(Arc::new(FlowsWorkflowBridge::pinned(config)));
+}
+
+/// The `flows::`-backed [`WorkflowBridge`].
+///
+/// Holds no store handle: `flows::` is addressed by [`Config`], and the RPC
+/// surface already re-reads the config per call rather than caching a snapshot
+/// that a workspace switch would invalidate. `config` is `Some` only when a
+/// caller pinned one (tests, and embedders that already hold a config).
+pub struct FlowsWorkflowBridge {
+    config: Option<Arc<Config>>,
+}
+
+impl FlowsWorkflowBridge {
+    /// A bridge that resolves the process's current config on every call.
+    #[must_use]
+    pub fn new() -> Self {
+        Self { config: None }
+    }
+
+    /// A bridge pinned to one config, for callers that already hold one and for
+    /// tests, which must address a temp workspace rather than the real one.
+    #[must_use]
+    pub fn pinned(config: Arc<Config>) -> Self {
+        Self {
+            config: Some(config),
+        }
+    }
+
+    /// The config to address the store with, loaded if not pinned.
+    async fn config(&self) -> Result<Arc<Config>, String> {
+        match &self.config {
+            Some(config) => Ok(Arc::clone(config)),
+            None => crate::openhuman::config::rpc::load_config_with_timeout()
+                .await
+                .map(Arc::new)
+                .map_err(|err| format!("this agent could not read its config: {err}")),
+        }
+    }
+
+    /// Run one `flows::` op to completion from a synchronous trait method.
+    ///
+    /// Only legal off the runtime's async threads. The transport guarantees that
+    /// — every sync bridge call is already inside `spawn_blocking` — and a
+    /// caller that violates it panics there rather than deadlocking, which that
+    /// same `spawn_blocking` converts into a reported error.
+    fn on_store<T, F>(&self, run: impl FnOnce(Arc<Config>) -> F) -> Result<T, String>
+    where
+        F: std::future::Future<Output = Result<T, String>>,
+    {
+        let handle = tokio::runtime::Handle::try_current()
+            .map_err(|_| "this agent's workflow store has no runtime to answer on".to_string())?;
+        handle.block_on(async move {
+            let config = self.config().await?;
+            run(config).await
+        })
+    }
+}
+
+impl Default for FlowsWorkflowBridge {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait]
+impl WorkflowBridge for FlowsWorkflowBridge {
+    fn action_dir(&self) -> Option<String> {
+        self.config
+            .as_ref()
+            .map(|config| config.action_dir.display().to_string())
+    }
+
+    fn list(&self) -> Vec<WorkflowDescriptor> {
+        match self.try_list() {
+            Ok(descriptors) => descriptors,
+            Err(err) => {
+                log::warn!("[flows] medulla bridge could not list workflows: {err}");
+                Vec::new()
+            }
+        }
+    }
+
+    fn try_list(&self) -> Result<Vec<WorkflowDescriptor>, String> {
+        self.on_store(|config| async move {
+            ops::flows_list(&config)
+                .await
+                .map(|outcome| outcome.value)
+                .map(|flows| flows.iter().map(describe_flow).collect::<Vec<_>>())
+        })
+    }
+
+    fn get(&self, id: &str) -> Result<Value, String> {
+        let id = id.to_string();
+        self.on_store(|config| async move {
+            let flow = ops::flows_get(&config, &id).await?.value;
+            serde_json::to_value(&flow)
+                .map_err(|err| format!("failed to render workflow '{id}': {err}"))
+        })
+    }
+
+    fn node_kinds(&self, kind: Option<&str>) -> Result<Value, String> {
+        // Purely static host+engine metadata — no store, no config, no runtime.
+        match kind.map(str::trim).filter(|kind| !kind.is_empty()) {
+            Some(kind) => {
+                let contract = node_kind_contract(kind).ok_or_else(|| {
+                    format!(
+                        "'{kind}' is not a node kind here; the kinds are: {}",
+                        NODE_KINDS.join(", ")
+                    )
+                })?;
+                Ok(json!({ "kinds": [node_kind_json(&contract)] }))
+            }
+            None => Ok(json!({
+                "kinds": all_node_kind_contracts()
+                    .iter()
+                    .map(node_kind_json)
+                    .collect::<Vec<_>>(),
+            })),
+        }
+    }
+
+    fn runs(&self, id: &str) -> Result<Value, String> {
+        let id = id.to_string();
+        self.on_store(|config| async move {
+            let runs = ops::flows_list_runs(&config, &id, RUN_HISTORY_LIMIT)
+                .await?
+                .value;
+            Ok(json!({ "runs": runs.iter().map(run_json).collect::<Vec<_>>() }))
+        })
+    }
+
+    async fn copilot(
+        &self,
+        instruction: &str,
+        workflow_id: Option<&str>,
+    ) -> Result<CopilotOutcome, String> {
+        let config = self.config().await?;
+        let before = ops::flows_list(&config).await?.value;
+        let request = builder_request(instruction, workflow_id, &before)?;
+        let expected_version = workflow_id.and_then(|id| {
+            before
+                .iter()
+                .find(|flow| flow.id == id)
+                .map(|flow| flow.updated_at.as_str())
+        });
+
+        // Headless: no chat thread to stream into, so the turn runs under the
+        // CLI origin with the full run-advancing hide-list — a remote authoring
+        // turn proposes a graph, it never runs one.
+        let built = ops::flows_build_with_extra_hidden_tools(
+            &config,
+            request,
+            None,
+            MEDULLA_COPILOT_HIDDEN_TOOLS,
+        )
+        .await?
+        .value;
+        let mut reply = built
+            .get("assistant_text")
+            .and_then(Value::as_str)
+            .unwrap_or_default()
+            .to_string();
+
+        let mut applied = None;
+        if let Some(proposal) = built.get("proposal").filter(|value| !value.is_null()) {
+            match apply_proposal(&config, workflow_id, proposal, expected_version).await {
+                Ok(saved) => applied = Some(saved),
+                // The turn happened and its reply is worth returning; only the
+                // save failed. Say so in the reply rather than discarding the
+                // whole outcome — and let the diff below report (truthfully)
+                // that nothing changed.
+                Err(err) => {
+                    log::warn!("[flows] medulla copilot could not persist its proposal: {err}");
+                    reply = format!("{reply}\n\nThe proposed graph could not be saved: {err}");
+                }
+            }
+        }
+
+        let changes = applied
+            .as_ref()
+            .map(|saved| diff_workflow(&before, std::slice::from_ref(&saved.flow), &saved.flow.id))
+            .unwrap_or_default();
+        let created = applied
+            .filter(|saved| saved.created)
+            .map(|saved| saved.flow.id);
+        Ok(CopilotOutcome {
+            reply,
+            changes,
+            created,
+        })
+    }
+}
+
+/// Project one saved flow onto the advert the orchestrator reasons over.
+///
+/// `name`/`description` are left as plain `String`s: the wire type skips them
+/// when empty, so a nameless flow omits the key rather than sending `""`.
+fn describe_flow(flow: &Flow) -> WorkflowDescriptor {
+    WorkflowDescriptor {
+        id: flow.id.clone(),
+        name: flow.name.trim().to_string(),
+        description: describe_graph(&flow.graph),
+        node_count: u32::try_from(flow.graph.nodes.len()).unwrap_or(u32::MAX),
+        enabled: Some(flow.enabled),
+        trigger_kind: trigger_kind(&flow.graph),
+        // Provenance is the backend's to stamp: this host runs one identity, so
+        // the batch-level `agentId` on the registration frame is the whole
+        // answer and a per-advert claim would only be redundant.
+        agent_id: None,
+        workspace_id: None,
+    }
+}
+
+/// The open-vocabulary trigger label, when the graph has exactly one trigger
+/// that names its kind. `None` covers "no trigger", "several triggers", and a
+/// trigger with no `trigger_kind` — all cases where a claim would be a guess.
+fn trigger_kind(graph: &WorkflowGraph) -> Option<String> {
+    graph
+        .trigger()?
+        .config
+        .get("trigger_kind")
+        .and_then(Value::as_str)
+        .map(str::to_string)
+}
+
+/// A one-line prose description of what a graph does, for the advert.
+///
+/// A flow has no author-written description field, so it is generated from the
+/// graph's own step names: `"<trigger> → step → step"`. That is the same
+/// information the Workflows list renders, written for a prompt instead of a
+/// row, and truncated so one large graph cannot dominate the section.
+fn describe_graph(graph: &WorkflowGraph) -> String {
+    let mut parts: Vec<String> = Vec::new();
+    parts.push(match trigger_kind(graph) {
+        Some(kind) => format!("on {kind}"),
+        None => "no trigger".to_string(),
+    });
+    parts.extend(
+        graph
+            .nodes
+            .iter()
+            .filter(|node| node.kind != tinyflows::model::NodeKind::Trigger)
+            .map(|node| node.name.trim())
+            .filter(|name| !name.is_empty())
+            .map(str::to_string),
+    );
+
+    let rendered = parts.join(" → ");
+    if rendered.chars().count() <= MAX_DESCRIPTION_CHARS {
+        return rendered;
+    }
+    let truncated: String = rendered.chars().take(MAX_DESCRIPTION_CHARS).collect();
+    format!("{truncated}…")
+}
+
+/// Project a node-kind contract onto the `WorkflowNodeKind` shape medulla-v1's
+/// port declares — camelCase, because the backend spreads this object straight
+/// onto the port type and a snake_case key would land beside the field it was
+/// meant to be rather than in it.
+fn node_kind_json(contract: &crate::openhuman::flows::node_contracts::NodeKindContract) -> Value {
+    json!({
+        "kind": contract.kind,
+        "summary": contract.summary,
+        "description": contract.description,
+        "configFields": contract.config_fields,
+        "ports": contract.ports,
+        "example": contract.example,
+        "notes": contract.notes,
+    })
+}
+
+/// Project a run row onto the `WorkflowRunSummary` shape the port declares.
+///
+/// Timestamps cross as epoch milliseconds because that is what the port's
+/// `startedAt`/`finishedAt` are; the store keeps RFC3339, and an unparseable
+/// stamp is omitted rather than sent as `0` (which reads as 1970, not "unknown").
+/// The per-node `steps` are dropped: they are the run *inspector's* payload, and
+/// a delegated-workflow read only asks whether it worked.
+fn run_json(run: &super::types::FlowRun) -> Value {
+    let mut value = json!({
+        "id": run.id,
+        "workflowId": run.flow_id,
+        "status": run.status,
+    });
+    if let Some(started) = epoch_millis(&run.started_at) {
+        value["startedAt"] = json!(started);
+    }
+    if let Some(finished) = run.finished_at.as_deref().and_then(epoch_millis) {
+        value["finishedAt"] = json!(finished);
+    }
+    if let Some(error) = run.error.as_deref().filter(|error| !error.is_empty()) {
+        value["error"] = json!(error);
+    }
+    value
+}
+
+/// RFC3339 → epoch milliseconds, or `None` for a stamp this build cannot read.
+fn epoch_millis(stamp: &str) -> Option<i64> {
+    chrono::DateTime::parse_from_rfc3339(stamp)
+        .ok()
+        .map(|parsed| parsed.timestamp_millis())
+}
+
+/// Build the authoring-turn request for one copilot call.
+///
+/// Naming a workflow is a **revise** of that exact graph, so the turn is briefed
+/// with the saved graph and its id; naming none is a **create**. An id the store
+/// does not hold fails here rather than silently degrading into a create — the
+/// orchestrator asked to change something specific, and creating a second
+/// workflow instead would be the wrong kind of surprising.
+fn builder_request(
+    instruction: &str,
+    workflow_id: Option<&str>,
+    flows: &[Flow],
+) -> Result<BuilderRequest, String> {
+    let (mode, graph, flow_id) = match workflow_id {
+        Some(id) => {
+            let flow = flows
+                .iter()
+                .find(|flow| flow.id == id)
+                .ok_or_else(|| format!("workflow '{id}' is not in this agent's store"))?;
+            let graph = serde_json::to_value(&flow.graph)
+                .map_err(|err| format!("failed to read workflow '{id}': {err}"))?;
+            (BuildMode::Revise, Some(graph), Some(id.to_string()))
+        }
+        None => (BuildMode::Create, None, None),
+    };
+    Ok(BuilderRequest {
+        mode,
+        instruction: instruction.to_string(),
+        graph,
+        flow_id,
+        run_id: None,
+        error: None,
+        failing_node_ids: Vec::new(),
+    })
+}
+
+/// The exact row committed by [`apply_proposal`].
+#[derive(Debug)]
+struct AppliedProposal {
+    flow: Flow,
+    created: bool,
+}
+
+/// Persist the graph a builder turn proposed, returning the committed row.
+///
+/// See the module docs for why this host performs the accept the copilot panel
+/// would otherwise perform, and for the two approval guards that bound it.
+async fn apply_proposal(
+    config: &Config,
+    workflow_id: Option<&str>,
+    proposal: &Value,
+    expected_version: Option<&str>,
+) -> Result<AppliedProposal, String> {
+    let graph = proposal
+        .get("graph")
+        .filter(|graph| !graph.is_null())
+        .ok_or_else(|| "the copilot proposed no graph".to_string())?
+        .clone();
+
+    match workflow_id {
+        // `require_approval: None` keeps whatever the user set. A remote turn
+        // may still have it forced ON by `flows_update`'s side-effect check; it
+        // can never turn it off.
+        Some(id) => {
+            let name = proposal
+                .get("name")
+                .and_then(Value::as_str)
+                .map(str::trim)
+                .filter(|name| !name.is_empty())
+                .map(str::to_string);
+            let flow = ops::flows_update_disarming_automatic(
+                config,
+                id,
+                name,
+                Some(graph),
+                None,
+                expected_version.map(str::to_string),
+            )
+            .await?
+            .value;
+            Ok(AppliedProposal {
+                flow,
+                created: false,
+            })
+        }
+        None => {
+            let name = proposal
+                .get("name")
+                .and_then(Value::as_str)
+                .map(str::trim)
+                .filter(|name| !name.is_empty())
+                .unwrap_or("Untitled workflow")
+                .to_string();
+            // `true`, not the proposal's own value: nothing authored by a remote
+            // instruction acts outward without a human decision at run time.
+            let flow = ops::flows_create(config, name, graph, true).await?.value;
+            Ok(AppliedProposal {
+                flow,
+                created: true,
+            })
+        }
+    }
+}
+
+/// What actually changed in the store across a copilot turn, as one line per
+/// change.
+///
+/// Derived from two listings, never from the agent's account of its own turn —
+/// see the module docs. A turn that talked and saved nothing yields an empty
+/// list, which is exactly the claim it earns.
+fn diff_flows(before: &[Flow], after: &[Flow]) -> Vec<String> {
+    let mut changes = Vec::new();
+
+    for flow in after {
+        let Some(previous) = before.iter().find(|candidate| candidate.id == flow.id) else {
+            changes.push(format!(
+                "created workflow \"{}\" ({}) with {} node(s)",
+                flow.name,
+                flow.id,
+                flow.graph.nodes.len()
+            ));
+            continue;
+        };
+        if previous.name != flow.name {
+            changes.push(format!(
+                "renamed workflow {} from \"{}\" to \"{}\"",
+                flow.id, previous.name, flow.name
+            ));
+        }
+        if previous.graph != flow.graph {
+            changes.push(format!(
+                "rewrote the graph of workflow \"{}\" ({}): {} node(s) → {} node(s)",
+                flow.name,
+                flow.id,
+                previous.graph.nodes.len(),
+                flow.graph.nodes.len()
+            ));
+        }
+        if previous.enabled != flow.enabled {
+            let state = if flow.enabled { "enabled" } else { "disabled" };
+            changes.push(format!("{state} workflow \"{}\" ({})", flow.name, flow.id));
+        }
+        if previous.require_approval != flow.require_approval {
+            let state = if flow.require_approval {
+                "now requires"
+            } else {
+                "no longer requires"
+            };
+            changes.push(format!(
+                "workflow \"{}\" ({}) {state} approval for outbound actions",
+                flow.name, flow.id
+            ));
+        }
+    }
+
+    for flow in before {
+        if !after.iter().any(|candidate| candidate.id == flow.id) {
+            changes.push(format!("deleted workflow \"{}\" ({})", flow.name, flow.id));
+        }
+    }
+
+    changes
+}
+
+/// Diff only the workflow this copilot turn successfully persisted.
+///
+/// A turn may overlap arbitrary user edits elsewhere in the store. Those edits
+/// are real, but they are not attributable to this turn and must not be claimed
+/// in its result.
+fn diff_workflow(before: &[Flow], after: &[Flow], workflow_id: &str) -> Vec<String> {
+    let before: Vec<Flow> = before
+        .iter()
+        .filter(|flow| flow.id == workflow_id)
+        .cloned()
+        .collect();
+    let after: Vec<Flow> = after
+        .iter()
+        .filter(|flow| flow.id == workflow_id)
+        .cloned()
+        .collect();
+    diff_flows(&before, &after)
+}
+
+#[cfg(test)]
+#[path = "medulla_bridge_tests.rs"]
+mod tests;

src/openhuman/flows/medulla_bridge_tests.rs

@@ -0,0 +1,593 @@
+use super::*;
+use crate::openhuman::config::Config;
+use serde_json::json;
+use tempfile::TempDir;
+use tinyflows::model::{Node, NodeKind};
+
+/// A config pointed at a throwaway workspace, so every test addresses its own
+/// SQLite store instead of the developer's real one.
+fn test_config(tmp: &TempDir) -> Arc<Config> {
+    let config = Config {
+        workspace_dir: tmp.path().join("workspace"),
+        action_dir: tmp.path().join("workspace"),
+        config_path: tmp.path().join("config.toml"),
+        ..Config::default()
+    };
+    std::fs::create_dir_all(&config.workspace_dir).unwrap();
+    Arc::new(config)
+}
+
+fn node(id: &str, kind: NodeKind, name: &str, config: Value) -> Node {
+    Node {
+        id: id.to_string(),
+        kind,
+        type_version: 1,
+        name: name.to_string(),
+        config,
+        ports: Vec::new(),
+        position: None,
+    }
+}
+
+/// A minimal manual-trigger graph with one named step.
+fn graph_with_step(step_name: &str) -> WorkflowGraph {
+    WorkflowGraph {
+        nodes: vec![
+            node(
+                "t",
+                NodeKind::Trigger,
+                "Trigger",
+                json!({ "trigger_kind": "manual" }),
+            ),
+            node(
+                "s",
+                NodeKind::Code,
+                step_name,
+                json!({ "code": "return {}" }),
+            ),
+        ],
+        ..Default::default()
+    }
+}
+
+fn flow(id: &str, name: &str, graph: WorkflowGraph) -> Flow {
+    Flow {
+        id: id.to_string(),
+        name: name.to_string(),
+        enabled: true,
+        graph,
+        created_at: "2026-01-01T00:00:00Z".to_string(),
+        updated_at: "2026-01-01T00:00:00Z".to_string(),
+        last_run_at: None,
+        last_status: None,
+        require_approval: false,
+    }
+}
+
+// ── advert projection ────────────────────────────────────────────────────────
+
+#[test]
+fn describe_flow_counts_every_node_and_names_the_trigger() {
+    let descriptor = describe_flow(&flow("f1", "Deploy", graph_with_step("Ship it")));
+    assert_eq!(descriptor.id, "f1");
+    assert_eq!(descriptor.name, "Deploy");
+    // Trigger + step: the whole graph, not just the executable half.
+    assert_eq!(descriptor.node_count, 2);
+    assert_eq!(descriptor.enabled, Some(true));
+    assert_eq!(descriptor.trigger_kind.as_deref(), Some("manual"));
+    assert_eq!(descriptor.description, "on manual → Ship it");
+}
+
+#[test]
+fn a_blank_name_stays_absent_on_the_wire_rather_than_empty() {
+    let descriptor = describe_flow(&flow("f1", "   ", graph_with_step("Step")));
+    assert!(descriptor.name.is_empty());
+    let wire = serde_json::to_value(&descriptor).unwrap();
+    // The contract is an ABSENT key, not `""` — both sides treat it as optional.
+    assert!(wire.get("name").is_none(), "wire = {wire}");
+    assert_eq!(wire["nodeCount"], 2);
+}
+
+#[test]
+fn a_triggerless_graph_claims_no_trigger_kind() {
+    let graph = WorkflowGraph {
+        nodes: vec![node("s", NodeKind::Code, "Step", json!({}))],
+        ..Default::default()
+    };
+    let descriptor = describe_flow(&flow("f1", "Orphan", graph));
+    assert_eq!(descriptor.trigger_kind, None);
+    assert_eq!(descriptor.description, "no trigger → Step");
+}
+
+#[test]
+fn a_long_graph_description_is_truncated() {
+    let nodes = (0..80)
+        .map(|i| {
+            node(
+                &format!("s{i}"),
+                NodeKind::Code,
+                "A fairly long step name",
+                json!({}),
+            )
+        })
+        .collect();
+    let descriptor = describe_flow(&flow(
+        "f1",
+        "Big",
+        WorkflowGraph {
+            nodes,
+            ..Default::default()
+        },
+    ));
+    assert_eq!(
+        descriptor.description.chars().count(),
+        MAX_DESCRIPTION_CHARS + 1
+    );
+    assert!(descriptor.description.ends_with('…'));
+}
+
+// ── node kinds ───────────────────────────────────────────────────────────────
+
+#[test]
+fn node_kinds_renders_the_catalog_in_the_ports_camel_case_shape() {
+    let bridge = FlowsWorkflowBridge::new();
+    let all = bridge.node_kinds(None).unwrap();
+    let kinds = all["kinds"].as_array().unwrap();
+    assert_eq!(kinds.len(), NODE_KINDS.len());
+    // `configFields`, not `config_fields`: the backend spreads this object onto
+    // the port type, so a snake_case key would miss the field entirely.
+    assert!(kinds[0].get("configFields").is_some());
+    assert!(kinds[0].get("config_fields").is_none());
+
+    let one = bridge.node_kinds(Some("agent")).unwrap();
+    assert_eq!(one["kinds"].as_array().unwrap().len(), 1);
+    assert_eq!(one["kinds"][0]["kind"], "agent");
+}
+
+#[test]
+fn an_unknown_node_kind_reports_the_real_ones() {
+    let err = FlowsWorkflowBridge::new()
+        .node_kinds(Some("teleport"))
+        .unwrap_err();
+    assert!(err.contains("teleport"), "{err}");
+    assert!(err.contains("tool_call"), "{err}");
+}
+
+#[test]
+fn a_blank_kind_filter_means_the_whole_catalog() {
+    let bridge = FlowsWorkflowBridge::new();
+    assert_eq!(
+        bridge.node_kinds(Some("  ")).unwrap(),
+        bridge.node_kinds(None).unwrap()
+    );
+}
+
+// ── run projection ───────────────────────────────────────────────────────────
+
+#[test]
+fn run_json_emits_epoch_millis_and_omits_what_it_cannot_read() {
+    let run = super::super::types::FlowRun {
+        id: "r1".to_string(),
+        flow_id: "f1".to_string(),
+        thread_id: "r1".to_string(),
+        status: "completed".to_string(),
+        started_at: "2026-01-01T00:00:00Z".to_string(),
+        finished_at: Some("not a timestamp".to_string()),
+        steps: Vec::new(),
+        pending_approvals: Vec::new(),
+        error: None,
+    };
+    let value = run_json(&run);
+    assert_eq!(value["id"], "r1");
+    assert_eq!(value["workflowId"], "f1");
+    assert_eq!(value["startedAt"], 1_767_225_600_000_i64);
+    // An unreadable stamp is absent, never `0` — that would read as 1970.
+    assert!(value.get("finishedAt").is_none(), "{value}");
+    assert!(value.get("error").is_none(), "{value}");
+    // The step-by-step payload belongs to the run inspector, not to this read.
+    assert!(value.get("steps").is_none(), "{value}");
+}
+
+// ── the store reads, over a real SQLite store ────────────────────────────────
+
+/// Drive a synchronous bridge read the way the transport does — from a blocking
+/// thread — since that is the only context its `block_on` is legal in.
+async fn off_runtime<T, F>(read: F) -> T
+where
+    T: Send + 'static,
+    F: FnOnce() -> T + Send + 'static,
+{
+    tokio::task::spawn_blocking(read).await.unwrap()
+}
+
+#[tokio::test]
+async fn list_and_get_answer_out_of_the_real_store() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let created = ops::flows_create(
+        &config,
+        "Deploy".to_string(),
+        serde_json::to_value(graph_with_step("Ship it")).unwrap(),
+        false,
+    )
+    .await
+    .unwrap()
+    .value;
+
+    let bridge = Arc::new(FlowsWorkflowBridge::pinned(Arc::clone(&config)));
+
+    let listed = off_runtime({
+        let bridge = Arc::clone(&bridge);
+        move || bridge.list()
+    })
+    .await;
+    assert_eq!(listed.len(), 1);
+    assert_eq!(listed[0].id, created.id);
+    assert_eq!(listed[0].node_count, 2);
+
+    let id = created.id.clone();
+    let detail = off_runtime({
+        let bridge = Arc::clone(&bridge);
+        move || bridge.get(&id)
+    })
+    .await
+    .unwrap();
+    // The graph crosses whole and opaque — nothing above this parses it.
+    assert_eq!(detail["id"], created.id);
+    assert!(detail["graph"]["nodes"].as_array().unwrap().len() == 2);
+}
+
+#[tokio::test]
+async fn an_unknown_id_is_a_readable_error_not_an_empty_answer() {
+    let tmp = TempDir::new().unwrap();
+    let bridge = Arc::new(FlowsWorkflowBridge::pinned(test_config(&tmp)));
+    let err = off_runtime(move || bridge.get("nope")).await.unwrap_err();
+    assert!(err.contains("nope"), "{err}");
+}
+
+#[tokio::test]
+async fn runs_answers_with_an_empty_window_for_a_flow_that_never_ran() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let created = ops::flows_create(
+        &config,
+        "Deploy".to_string(),
+        serde_json::to_value(graph_with_step("Ship it")).unwrap(),
+        false,
+    )
+    .await
+    .unwrap()
+    .value;
+    let bridge = Arc::new(FlowsWorkflowBridge::pinned(config));
+    let runs = off_runtime(move || bridge.runs(&created.id)).await.unwrap();
+    assert_eq!(runs["runs"], json!([]));
+}
+
+// ── persisting a proposal ────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn a_created_workflow_always_requires_approval() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let proposal = json!({
+        "name": "Remote build",
+        "graph": serde_json::to_value(graph_with_step("Ship it")).unwrap(),
+        // The proposal's own answer is deliberately ignored on this path.
+        "require_approval": false,
+    });
+
+    let applied = apply_proposal(&config, None, &proposal, None)
+        .await
+        .unwrap();
+    assert!(applied.created);
+    let saved = ops::flows_get(&config, &applied.flow.id)
+        .await
+        .unwrap()
+        .value;
+    assert_eq!(saved.name, "Remote build");
+    assert!(
+        saved.require_approval,
+        "a remotely authored workflow must still park its outbound actions"
+    );
+}
+
+#[tokio::test]
+async fn an_update_cannot_lower_the_approval_requirement() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let created = ops::flows_create(
+        &config,
+        "Deploy".to_string(),
+        serde_json::to_value(graph_with_step("Ship it")).unwrap(),
+        true,
+    )
+    .await
+    .unwrap()
+    .value;
+
+    let proposal = json!({
+        "name": "Deploy v2",
+        "graph": serde_json::to_value(graph_with_step("Ship it twice")).unwrap(),
+        "require_approval": false,
+    });
+    let applied = apply_proposal(
+        &config,
+        Some(&created.id),
+        &proposal,
+        Some(&created.updated_at),
+    )
+    .await
+    .unwrap();
+    assert!(!applied.created, "an update is not a creation");
+    assert_eq!(applied.flow.id, created.id);
+
+    let saved = ops::flows_get(&config, &created.id).await.unwrap().value;
+    assert_eq!(saved.name, "Deploy v2");
+    assert!(
+        saved.require_approval,
+        "the user's approval requirement survives a remote turn"
+    );
+}
+
+fn schedule_graph(cron: &str) -> Value {
+    json!({
+        "name": "scheduled",
+        "nodes": [{
+            "id": "t",
+            "kind": "trigger",
+            "name": "Trigger",
+            "config": { "trigger_kind": "schedule", "schedule": cron }
+        }],
+        "edges": []
+    })
+}
+
+#[tokio::test]
+async fn a_remote_automatic_revision_requires_explicit_rearming() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let created = ops::flows_create(
+        &config,
+        "Scheduled".to_string(),
+        schedule_graph("0 9 * * *"),
+        true,
+    )
+    .await
+    .unwrap()
+    .value;
+    assert!(!created.enabled, "automatic flows are born disarmed");
+
+    let armed = ops::flows_set_enabled(&config, &created.id, true)
+        .await
+        .unwrap()
+        .value;
+    assert!(armed.enabled, "precondition: the user explicitly armed it");
+
+    let proposal = json!({
+        "name": "Scheduled v2",
+        "graph": schedule_graph("0 10 * * *"),
+    });
+    let applied = apply_proposal(&config, Some(&armed.id), &proposal, Some(&armed.updated_at))
+        .await
+        .unwrap();
+
+    assert!(!applied.flow.enabled);
+    let saved = ops::flows_get(&config, &armed.id).await.unwrap().value;
+    assert!(!saved.enabled, "the revised schedule must stay disarmed");
+}
+
+#[tokio::test]
+async fn a_proposal_without_a_graph_is_refused() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let err = apply_proposal(&config, None, &json!({ "name": "Nothing" }), None)
+        .await
+        .unwrap_err();
+    assert!(err.contains("no graph"), "{err}");
+}
+
+#[tokio::test]
+async fn an_update_refuses_to_overwrite_a_concurrent_edit() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let created = ops::flows_create(
+        &config,
+        "Deploy".to_string(),
+        serde_json::to_value(graph_with_step("Ship it")).unwrap(),
+        true,
+    )
+    .await
+    .unwrap()
+    .value;
+
+    ops::flows_update(
+        &config,
+        &created.id,
+        Some("User edit".to_string()),
+        None,
+        None,
+        Some(created.updated_at.clone()),
+    )
+    .await
+    .unwrap();
+
+    let proposal = json!({
+        "name": "Copilot edit",
+        "graph": serde_json::to_value(graph_with_step("Ship it twice")).unwrap(),
+    });
+    let err = apply_proposal(
+        &config,
+        Some(&created.id),
+        &proposal,
+        Some(&created.updated_at),
+    )
+    .await
+    .unwrap_err();
+    assert!(err.contains("conflict"), "{err}");
+
+    let saved = ops::flows_get(&config, &created.id).await.unwrap().value;
+    assert_eq!(saved.name, "User edit");
+}
+
+// ── the accountability diff ──────────────────────────────────────────────────
+
+#[test]
+fn diff_reports_a_creation_with_its_size() {
+    let after = vec![flow("f1", "Deploy", graph_with_step("Ship it"))];
+    let changes = diff_flows(&[], &after);
+    assert_eq!(changes.len(), 1);
+    assert!(
+        changes[0].contains("created workflow \"Deploy\" (f1)"),
+        "{changes:?}"
+    );
+    assert!(changes[0].contains("2 node(s)"), "{changes:?}");
+}
+
+#[test]
+fn diff_reports_a_rewrite_a_rename_and_a_deletion() {
+    let before = vec![
+        flow("f1", "Deploy", graph_with_step("Ship it")),
+        flow("f2", "Doomed", graph_with_step("Step")),
+    ];
+    let mut renamed = flow("f1", "Deploy v2", graph_with_step("Ship it faster"));
+    renamed.enabled = false;
+    let changes = diff_flows(&before, &[renamed]);
+
+    assert!(
+        changes
+            .iter()
+            .any(|line| line.contains("renamed workflow f1")),
+        "{changes:?}"
+    );
+    assert!(
+        changes
+            .iter()
+            .any(|line| line.contains("rewrote the graph")),
+        "{changes:?}"
+    );
+    assert!(
+        changes
+            .iter()
+            .any(|line| line.starts_with("disabled workflow")),
+        "{changes:?}"
+    );
+    assert!(
+        changes
+            .iter()
+            .any(|line| line.contains("deleted workflow \"Doomed\" (f2)")),
+        "{changes:?}"
+    );
+}
+
+#[test]
+fn a_turn_that_changed_nothing_claims_nothing() {
+    let flows = vec![flow("f1", "Deploy", graph_with_step("Ship it"))];
+    // The whole point: the agent's account of its turn cannot inflate this.
+    assert!(diff_flows(&flows, &flows).is_empty());
+}
+
+#[test]
+fn diff_reports_an_approval_requirement_change() {
+    let before = vec![flow("f1", "Deploy", graph_with_step("Ship it"))];
+    let mut after = before.clone();
+    after[0].require_approval = true;
+    let changes = diff_flows(&before, &after);
+    assert_eq!(changes.len(), 1);
+    assert!(changes[0].contains("now requires approval"), "{changes:?}");
+}
+
+#[test]
+fn scoped_diff_excludes_concurrent_changes_to_other_workflows() {
+    let before = vec![
+        flow("target", "Deploy", graph_with_step("Ship it")),
+        flow("other", "Unrelated", graph_with_step("Before")),
+    ];
+    let after = vec![
+        flow("target", "Deploy v2", graph_with_step("Ship it twice")),
+        flow("other", "User edit", graph_with_step("After")),
+        flow("new", "Also user-created", graph_with_step("Step")),
+    ];
+
+    let changes = diff_workflow(&before, &after, "target");
+    assert!(
+        changes.iter().all(|line| !line.contains("other")
+            && !line.contains("Unrelated")
+            && !line.contains("User edit")
+            && !line.contains("new")
+            && !line.contains("Also user-created")),
+        "{changes:?}"
+    );
+    assert!(
+        changes
+            .iter()
+            .any(|line| line.contains("renamed workflow target")),
+        "{changes:?}"
+    );
+    assert!(
+        changes
+            .iter()
+            .any(|line| line.contains("rewrote the graph")),
+        "{changes:?}"
+    );
+}
+
+#[test]
+fn medulla_copilot_hides_every_persistence_tool() {
+    assert_eq!(
+        MEDULLA_COPILOT_HIDDEN_TOOLS,
+        [
+            "save_workflow",
+            "create_workflow",
+            "duplicate_flow",
+            "edit_workflow"
+        ]
+    );
+}
+
+#[test]
+fn accountability_uses_the_committed_snapshot_not_a_later_user_edit() {
+    let before = vec![flow("target", "Deploy", graph_with_step("Ship it"))];
+    let committed = flow("target", "Copilot edit", graph_with_step("Ship it twice"));
+    let later_user_edit = flow("target", "User edit", graph_with_step("Something else"));
+
+    let changes = diff_workflow(&before, std::slice::from_ref(&committed), "target");
+    assert!(
+        changes.iter().any(|line| line.contains("Copilot edit")),
+        "{changes:?}"
+    );
+    assert!(
+        changes
+            .iter()
+            .all(|line| !line.contains(&later_user_edit.name)),
+        "{changes:?}"
+    );
+}
+
+// ── the copilot turn's request shaping ───────────────────────────────────────
+
+#[test]
+fn naming_no_workflow_briefs_a_create() {
+    let request = builder_request("build a deploy flow", None, &[]).unwrap();
+    assert_eq!(request.mode, BuildMode::Create);
+    assert!(request.graph.is_none());
+    assert!(request.flow_id.is_none());
+    assert_eq!(request.instruction, "build a deploy flow");
+}
+
+#[test]
+fn naming_a_workflow_briefs_a_revise_of_that_exact_graph() {
+    let flows = vec![flow("f1", "Deploy", graph_with_step("Ship it"))];
+    let request = builder_request("add a retry", Some("f1"), &flows).unwrap();
+    assert_eq!(request.mode, BuildMode::Revise);
+    assert_eq!(request.flow_id.as_deref(), Some("f1"));
+    assert_eq!(
+        request.graph.unwrap(),
+        serde_json::to_value(&flows[0].graph).unwrap()
+    );
+}
+
+#[test]
+fn an_unknown_workflow_id_fails_rather_than_silently_creating_a_second_one() {
+    let err = builder_request("add a retry", Some("ghost"), &[]).unwrap_err();
+    assert!(err.contains("ghost"), "{err}");
+}

src/openhuman/flows/mod.rs

@@ -6,6 +6,10 @@
 //! handful of functions re-exported below for the capability seam's
 //! [`crate::openhuman::tinyflows::caps::FlowStateStore`]); the RPC/CLI
 //! controller surface in `schemas` (private, re-exported below).
+//!
+//! [`medulla_bridge`] adapts this store onto the medulla harness protocol's
+//! workflow plane, so a remote orchestrator can read these graphs and brief the
+//! authoring copilot without any of that reaching back into `ops`.
 
 pub mod agents;
 mod build_registry;
@@ -13,6 +17,7 @@ pub mod builder_tools;
 pub mod bus;
 pub mod discovery_tools;
 mod draft_store;
+pub mod medulla_bridge;
 pub mod memory_tools;
 mod n8n_import;
 pub mod node_contracts;

src/openhuman/flows/ops.rs

@@ -3444,6 +3444,7 @@ pub async fn flows_duplicate(config: &Config, id: &str) -> Result<RpcOutcome<Flo
         store::insert_duplicate_flow(config, &source, new_name).map_err(|e| e.to_string())?;
     // Intentionally NO bind_trigger: a duplicate is disabled and must stay
     // inert (no schedule/trigger dispatch) until the user enables it.
+    publish_flow_changed(&flow.id, "created", "system");
     Ok(RpcOutcome::single_log(
         flow,
         format!("flow duplicated from {id}"),
@@ -3737,6 +3738,14 @@ fn publish_flow_changed(flow_id: &str, kind: &str, actor: &str) {
         kind: kind.to_string(),
         actor: actor.to_string(),
     });
+    // Re-advertise the workflow set to the medulla backend. This is the single
+    // funnel every store mutation passes through (create / duplicate / update /
+    // delete / enable), and the backend replaces a socket's whole entry on each
+    // registration — so re-sending here is what keeps a remote orchestrator from
+    // reasoning about a set that no longer exists. A no-op (one debug log, no
+    // task spawned) when no bridge is installed, which is every build that is
+    // not talking to a backend, and every test.
+    crate::openhuman::socket::medulla::workflows::emit_register_workflows();
 }
 
 /// Maps a store-level [`FlowUpdateError`](store::FlowUpdateError) to the RPC
@@ -3809,6 +3818,54 @@ pub async fn flows_update(
     graph_json: Option<Value>,
     require_approval: Option<bool>,
     expected_version: Option<String>,
+) -> Result<RpcOutcome<Flow>, String> {
+    flows_update_inner(
+        config,
+        id,
+        name,
+        graph_json,
+        require_approval,
+        expected_version,
+        false,
+    )
+    .await
+}
+
+/// Update a flow while atomically disarming any automatic-trigger graph.
+///
+/// Remote authoring surfaces use this variant so revising a schedule,
+/// app-event, or webhook flow never preserves a prior local opt-in to run the
+/// old graph. The same guarded store write persists the graph and
+/// `enabled=false`, so no trigger can observe the revised graph armed between
+/// two writes.
+pub(crate) async fn flows_update_disarming_automatic(
+    config: &Config,
+    id: &str,
+    name: Option<String>,
+    graph_json: Option<Value>,
+    require_approval: Option<bool>,
+    expected_version: Option<String>,
+) -> Result<RpcOutcome<Flow>, String> {
+    flows_update_inner(
+        config,
+        id,
+        name,
+        graph_json,
+        require_approval,
+        expected_version,
+        true,
+    )
+    .await
+}
+
+async fn flows_update_inner(
+    config: &Config,
+    id: &str,
+    name: Option<String>,
+    graph_json: Option<Value>,
+    require_approval: Option<bool>,
+    expected_version: Option<String>,
+    disarm_automatic: bool,
 ) -> Result<RpcOutcome<Flow>, String> {
     let existing = store::get_flow(config, id)
         .map_err(|e| e.to_string())?
@@ -3834,13 +3891,15 @@ pub async fn flows_update(
     let was_auto = trigger_is_automatic(&existing.graph);
     let now_auto = trigger_is_automatic(&graph);
     let is_manual_to_auto_transition = now_auto && !was_auto;
-    let enabled_override = is_manual_to_auto_transition.then_some(false);
+    let forced_automatic_disarm = disarm_automatic && now_auto;
+    let enabled_override =
+        (is_manual_to_auto_transition || forced_automatic_disarm).then_some(false);
     // Best-effort flag for the info log / result message below: whether the
     // flow *appeared* live going into this update. Not used for the
     // override decision itself (that's unconditional, see above) — only to
     // avoid telling the user "flow was auto-disabled" when it was already
     // disabled going in.
-    let should_disarm = is_manual_to_auto_transition && existing.enabled;
+    let should_disarm = enabled_override == Some(false) && existing.enabled;
     tracing::debug!(
         target: "flows",
         flow_id = %id,
@@ -3848,6 +3907,7 @@ pub async fn flows_update(
         now_auto,
         currently_enabled = existing.enabled,
         is_manual_to_auto_transition,
+        forced_automatic_disarm,
         should_disarm,
         "[flows] flows_update: auto-trigger disarm decision inputs"
     );
@@ -3896,7 +3956,7 @@ pub async fn flows_update(
         tracing::info!(
             target: "flows",
             flow_id = %id,
-            "[flows] flows_update: auto-disabled — graph changed manual→automatic trigger on an enabled flow"
+            "[flows] flows_update: auto-disabled automatic-trigger graph pending explicit re-arm"
         );
     }
 
@@ -3914,12 +3974,16 @@ pub async fn flows_update(
     publish_flow_changed(id, "updated", "system");
     let mut logs = vec![format!("flow updated: {id}")];
     if should_disarm {
-        logs.push(
+        let reason = if forced_automatic_disarm {
+            "Flow was auto-disabled because this authoring surface revised an automatic trigger \
+             (schedule / app_event / webhook). Enable it explicitly (flows_set_enabled) once \
+             you've reviewed the revision."
+        } else {
             "Flow was auto-disabled because its trigger changed from manual to automatic \
              (schedule / app_event / webhook). Enable it explicitly (flows_set_enabled) once \
              you've reviewed the new trigger."
-                .to_string(),
-        );
+        };
+        logs.push(reason.to_string());
     }
     if side_effect_forced {
         logs.push(
@@ -6165,6 +6229,21 @@ pub async fn flows_build(
     config: &Config,
     req: crate::openhuman::flows::agents::workflow_builder::builder_prompt::BuilderRequest,
     stream: Option<FlowStreamTarget>,
+) -> Result<RpcOutcome<Value>, String> {
+    flows_build_with_extra_hidden_tools(config, req, stream, &[]).await
+}
+
+/// [`flows_build`] with caller-specific tools removed in addition to the
+/// standard streaming/headless safety lists.
+///
+/// This is intentionally crate-private: product surfaces use [`flows_build`]'s
+/// normal builder belt. Host integrations that add their own persistence
+/// boundary can hide tools that would bypass that boundary.
+pub(crate) async fn flows_build_with_extra_hidden_tools(
+    config: &Config,
+    req: crate::openhuman::flows::agents::workflow_builder::builder_prompt::BuilderRequest,
+    stream: Option<FlowStreamTarget>,
+    extra_hidden_tools: &[&str],
 ) -> Result<RpcOutcome<Value>, String> {
     use crate::openhuman::agent::Agent;
     use crate::openhuman::flows::agents::workflow_builder::builder_prompt::render_prompt;
@@ -6225,6 +6304,14 @@ pub async fn flows_build(
         }
         restrict_builder_toolset(&mut agent);
     }
+    if !extra_hidden_tools.is_empty() {
+        tracing::debug!(
+            target: "flows",
+            hidden = ?extra_hidden_tools,
+            "[flows] flows_build: applying caller-specific hidden tools"
+        );
+        agent.hide_tools(extra_hidden_tools);
+    }
 
     // When a chat thread is attached (the copilot pane), stream the builder turn
     // into it exactly like an interactive turn — text/tool deltas and the

src/openhuman/socket/event_handlers.rs

@@ -59,6 +59,7 @@ pub(super) fn handle_sio_event(
     match event_name {
         "ready" => {
             log::info!("[socket] Server ready — auth successful");
+            super::medulla::workflows::begin_connection_generation();
             *shared.status.write() = ConnectionStatus::Connected;
             emit_state_change(shared);
             // Declare the device-tool manifest so the hosted brain knows which
@@ -73,6 +74,12 @@ pub(super) fn handle_sio_event(
             // operator can delegate `medulla:task_run` to a named agent. The
             // backend clears the roster on socket disconnect.
             super::medulla::emit_register_agents();
+            // Advertise the saved workflow graphs this host can be asked to run,
+            // so the orchestrator can name one when delegating. Same
+            // per-connection lifetime as the roster: rebuilt on every reconnect,
+            // dropped server-side on disconnect. A no-op until a host installs a
+            // `WorkflowBridge`.
+            super::medulla::workflows::emit_register_workflows();
         }
         "error" => {
             log::error!("[socket] Server error event: {}", data);
@@ -518,6 +525,52 @@ pub(super) fn handle_sio_event(
                 Err(e) => log::warn!("[socket] failed to parse medulla:task_abort: {e}"),
             }
         }
+        // Capability handshake. The backend waits 10s per probe, so an
+        // unanswered one is not a graceful degradation — it is a stall on the
+        // first delegation to this agent.
+        "medulla:capabilities_request" => {
+            match serde_json::from_value::<super::medulla::payloads::CapabilitiesRequest>(
+                data.clone(),
+            ) {
+                Ok(request) => {
+                    log::info!(
+                        "[socket] medulla:capabilities_request probe_id={} agent_id={}",
+                        request.probe_id,
+                        request.agent_id
+                    );
+                    super::medulla::handle_capabilities_request(request);
+                }
+                // An undecodable probe still has to be answered when it named
+                // itself, for the same reason a decodable one does: silence
+                // spends the backend's whole 10s window.
+                Err(e) => {
+                    log::warn!("[socket] failed to parse medulla:capabilities_request: {e}");
+                    super::medulla::reject_unparsed_capabilities_request(&data, &e.to_string());
+                }
+            }
+        }
+        // Workflow round trip: a read of, or an authoring turn on, this host's
+        // own workflow store.
+        "medulla:workflow_request" => {
+            match serde_json::from_value::<super::medulla::payloads::WorkflowRequest>(data.clone())
+            {
+                Ok(request) => {
+                    log::info!(
+                        "[socket] medulla:workflow_request request_id={} op={:?}",
+                        request.request_id,
+                        request.op
+                    );
+                    super::medulla::workflows::handle_workflow_request(request);
+                }
+                // An undecodable frame still has to be answered when it named
+                // itself: staying silent would cost the backend the op's whole
+                // deadline (up to ten minutes for `copilot`).
+                Err(e) => {
+                    log::warn!("[socket] failed to parse medulla:workflow_request: {e}");
+                    super::medulla::workflows::reject_unparsed_request(&data, &e.to_string());
+                }
+            }
+        }
 
         // Channel inbound message — publish to event bus for ChannelInboundSubscriber
         _ if event_name.ends_with(":message") => {

src/openhuman/socket/manager.rs

@@ -122,6 +122,9 @@ pub struct SocketManager {
     shutdown_tx: tokio::sync::Mutex<Option<watch::Sender<bool>>>,
     /// Join handle for the background connection loop.
     loop_handle: tokio::sync::Mutex<Option<tokio::task::JoinHandle<()>>>,
+    /// Serializes identity-sensitive disconnect → bridge bind → connect
+    /// transactions while still allowing ordinary emits and state reads.
+    identity_rebind: tokio::sync::Mutex<()>,
 }
 
 impl SocketManager {
@@ -139,9 +142,15 @@ impl SocketManager {
             emit_tx: tokio::sync::Mutex::new(None),
             shutdown_tx: tokio::sync::Mutex::new(None),
             loop_handle: tokio::sync::Mutex::new(None),
+            identity_rebind: tokio::sync::Mutex::new(()),
         }
     }
 
+    /// Lock an identity-sensitive socket rebind for its complete transaction.
+    pub(crate) async fn lock_identity_rebind(&self) -> tokio::sync::MutexGuard<'_, ()> {
+        self.identity_rebind.lock().await
+    }
+
     /// Set the webhook router for skill-targeted webhook delivery.
     pub fn set_webhook_router(&self, router: Arc<WebhookRouter>) {
         log::debug!("[socket] WebhookRouter attached");
@@ -272,13 +281,14 @@ impl SocketManager {
 
     /// Disconnect from the server and shut down the background loop.
     pub async fn disconnect(&self) -> Result<(), String> {
+        super::medulla::workflows::end_connection_generation();
         if let Some(tx) = self.shutdown_tx.lock().await.take() {
             let _ = tx.send(true);
         }
         self.shared.ack_registry.cancel_all();
         self.emit_tx.lock().await.take();
         if let Some(handle) = self.loop_handle.lock().await.take() {
-            let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
+            terminate_loop(handle, Duration::from_secs(5)).await;
         }
         *self.shared.status.write() = ConnectionStatus::Disconnected;
         *self.shared.socket_id.write() = None;
@@ -337,6 +347,21 @@ impl SocketManager {
     }
 }
 
+/// Wait for the socket loop to observe shutdown, then abort and join it if a
+/// transport operation outlives the grace period.
+///
+/// Dropping a timed-out `JoinHandle` detaches its task. During an account
+/// switch that would let the old credential finish authenticating after the
+/// new user's workflow bridge is installed, so timeout must mean termination,
+/// not detachment.
+async fn terminate_loop(mut handle: tokio::task::JoinHandle<()>, grace: Duration) {
+    if tokio::time::timeout(grace, &mut handle).await.is_err() {
+        log::warn!("[socket] connection loop did not stop within {grace:?} — aborting");
+        handle.abort();
+        let _ = handle.await;
+    }
+}
+
 fn encode_sio_event(
     event: &str,
     data: serde_json::Value,
@@ -471,6 +496,56 @@ mod tests {
         assert_eq!(mgr.get_state().status, ConnectionStatus::Disconnected);
     }
 
+    #[tokio::test]
+    async fn a_timed_out_socket_loop_is_aborted_and_joined() {
+        use std::sync::atomic::{AtomicBool, Ordering};
+
+        struct MarksDrop(Arc<AtomicBool>);
+        impl Drop for MarksDrop {
+            fn drop(&mut self) {
+                self.0.store(true, Ordering::SeqCst);
+            }
+        }
+
+        let dropped = Arc::new(AtomicBool::new(false));
+        let dropped_in_task = Arc::clone(&dropped);
+        let handle = tokio::spawn(async move {
+            let _guard = MarksDrop(dropped_in_task);
+            std::future::pending::<()>().await;
+        });
+        tokio::task::yield_now().await;
+
+        terminate_loop(handle, Duration::from_millis(1)).await;
+
+        assert!(
+            dropped.load(Ordering::SeqCst),
+            "terminate_loop must join the aborted task before returning"
+        );
+    }
+
+    #[tokio::test]
+    async fn identity_rebind_transactions_are_serialized() {
+        let manager = Arc::new(SocketManager::new());
+        let first = manager.lock_identity_rebind().await;
+
+        let waiting_manager = Arc::clone(&manager);
+        let mut waiter = tokio::spawn(async move {
+            let _second = waiting_manager.lock_identity_rebind().await;
+        });
+        assert!(
+            tokio::time::timeout(Duration::from_millis(10), &mut waiter)
+                .await
+                .is_err(),
+            "a second account rebind must not interleave with the first"
+        );
+
+        drop(first);
+        tokio::time::timeout(Duration::from_secs(1), waiter)
+            .await
+            .expect("the next rebind should proceed after the first commits")
+            .unwrap();
+    }
+
     #[test]
     fn emit_state_change_is_safe_to_call_on_empty_shared() {
         let shared = SharedState {

src/openhuman/socket/medulla/mod.rs

@@ -12,15 +12,28 @@
 //! - `medulla:task_send`  → [`MedullaTaskManager::steer_task`]
 //! - `medulla:task_abort` → [`MedullaTaskManager::abort_task`]
 //!
+//! - `medulla:capabilities_request` → [`handle_capabilities_request`]
+//! - `medulla:workflow_request` → [`workflows::handle_workflow_request`]
+//!
 //! Up (openhuman → backend):
 //! - `medulla:task_envelope` — the live session stream, as
 //!   `tinyplace.harness.session.v2` envelopes (see [`envelope`]).
 //! - `medulla:task_result`   — explicit completion.
 //! - `medulla:register_agents` — roster advertised on connect
 //!   ([`emit_register_agents`]); the backend clears it on disconnect.
+//! - `medulla:register_workflows` — the saved workflow graphs this host can be
+//!   asked to run ([`workflows::emit_register_workflows`]), same lifetime.
+//! - `medulla:capabilities_result` — the answer to a capability probe.
+//! - `medulla:workflow_result` — the answer to a workflow round trip.
+//!
+//! Every *down* event here is request/response with a server-side deadline, so
+//! silence is never free: an unanswered probe costs the backend ten seconds and
+//! an unanswered workflow request up to ten minutes. Both handlers therefore
+//! always reply, even when the answer is an error.
 
 pub mod envelope;
 pub mod payloads;
+pub mod workflows;
 
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicI64, Ordering};
@@ -36,8 +49,8 @@ use crate::openhuman::agent::turn_origin::{with_origin, AgentTurnOrigin};
 use crate::openhuman::agent::Agent;
 
 use payloads::{
-    AgentDescriptor, RegisterAgents, TaskResult, EVENT_REGISTER_AGENTS, EVENT_TASK_ENVELOPE,
-    EVENT_TASK_RESULT,
+    AgentDescriptor, CapabilitiesRequest, CapabilitiesResult, RegisterAgents, TaskResult,
+    EVENT_CAPABILITIES_RESULT, EVENT_REGISTER_AGENTS, EVENT_TASK_ENVELOPE, EVENT_TASK_RESULT,
 };
 
 /// Default agent an unspecified `medulla:task_run` runs as.
@@ -476,7 +489,7 @@ pub fn emit_register_agents() {
     let agents: Vec<AgentDescriptor> = crate::openhuman::agent_registry::default_agents()
         .into_iter()
         .map(|entry| AgentDescriptor {
-            agent_id: entry.id,
+            id: entry.id,
             name: entry.name,
             description: entry.description,
         })
@@ -485,6 +498,130 @@ pub fn emit_register_agents() {
     emit(EVENT_REGISTER_AGENTS, RegisterAgents { agents });
 }
 
+/// Handle `medulla:capabilities_request`: self-report for one probe.
+///
+/// The backend's roster calls this lazily before first delegating to an agent it
+/// has not cached, fans the frame out to every harness socket the user holds,
+/// and waits ten seconds. An agent that never answers does not degrade
+/// gracefully — it spends that whole window on every first delegation. So this
+/// always emits a `medulla:capabilities_result`, even when the only thing it can
+/// truthfully say is `ready`.
+///
+/// Runs on its own task: building the report loads config and reads the workflow
+/// store, neither of which belongs on the socket read loop.
+pub fn handle_capabilities_request(request: CapabilitiesRequest) {
+    let workflows::BridgeGeneration { bridge, cancel } = workflows::bridge_generation();
+    let connection_cancel = workflows::connection_generation();
+    tokio::spawn(async move {
+        tokio::select! {
+            biased;
+            _ = cancel.cancelled() => {
+                log::debug!("[medulla] discarded capability probe from an old bridge");
+            }
+            _ = connection_cancel.cancelled() => {
+                log::debug!("[medulla] discarded capability probe from a closed socket");
+            }
+            _ = async move {
+                let capabilities = describe_self(&request.agent_id, bridge).await;
+                emit_awaited(
+                    EVENT_CAPABILITIES_RESULT,
+                    CapabilitiesResult {
+                        probe_id: request.probe_id,
+                        capabilities,
+                    },
+                )
+                .await;
+            } => {}
+        }
+    });
+}
+
+/// Answer a `medulla:capabilities_request` this build could not decode.
+///
+/// A frame from a newer backend (an added required field, a retyped one) would
+/// otherwise be dropped, and a dropped probe costs the backend its full
+/// ten-second deadline on the *first* delegation to this agent. The `probeId` is
+/// recovered from the raw JSON — without one there is nothing to correlate, so
+/// the frame really is unanswerable and is only logged.
+pub fn reject_unparsed_capabilities_request(raw: &serde_json::Value, reason: &str) {
+    let Some(result) = unparsed_capabilities_result(raw, reason) else {
+        log::warn!("[medulla] undecodable capabilities_request carries no probeId — cannot answer");
+        return;
+    };
+    log::info!(
+        "[medulla] answering undecodable capabilities_request probe_id={} as not ready",
+        result.probe_id
+    );
+    emit(EVENT_CAPABILITIES_RESULT, result);
+}
+
+/// Project an undecodable probe onto its reply frame. Split out so the recovery
+/// (which `probeId` is answerable, and what the answer says) is testable without
+/// a socket.
+///
+/// The report is `ready: false` plus a reason rather than an empty bag: those
+/// two fields are on the backend's `sanitizeCapabilities` allowlist, and a
+/// payload with nothing on that allowlist sanitizes away into the same
+/// non-answer as silence.
+fn unparsed_capabilities_result(
+    raw: &serde_json::Value,
+    reason: &str,
+) -> Option<CapabilitiesResult> {
+    let probe_id = raw.get("probeId").and_then(serde_json::Value::as_str)?;
+    Some(CapabilitiesResult {
+        probe_id: probe_id.to_string(),
+        capabilities: serde_json::json!({
+            "ready": false,
+            "readyReason": format!("this agent could not read the probe: {reason}"),
+        }),
+    })
+}
+
+/// Build this host's capability report for `agent_id`.
+///
+/// Only fields on the backend's allowlist (`sanitizeCapabilities`) are worth
+/// sending — anything else is dropped there — and each is best-effort: a field
+/// this host cannot resolve is omitted rather than guessed, because the
+/// orchestrator reasons about placement from these values.
+async fn describe_self(
+    agent_id: &str,
+    bridge: Option<Arc<dyn workflows::WorkflowBridge>>,
+) -> serde_json::Value {
+    let mut caps = serde_json::Map::new();
+    // Advisory readiness. A connected core that answered the probe at all is
+    // ready by definition; per-agent gating lives in the task path, not here.
+    caps.insert("ready".to_string(), serde_json::Value::Bool(true));
+
+    if let Some(entry) = crate::openhuman::agent_registry::default_agents()
+        .into_iter()
+        .find(|entry| entry.id == agent_id)
+    {
+        if !entry.description.is_empty() {
+            caps.insert("summary".to_string(), entry.description.into());
+        }
+    }
+
+    // `cwd` is the agent's read/write root (`action_dir`), which is what the
+    // orchestrator actually means by "where does this agent work".
+    if let Some(action_dir) = bridge.as_ref().and_then(|bridge| bridge.action_dir()) {
+        caps.insert("cwd".to_string(), action_dir.into());
+    }
+
+    // The workflow adverts ride the probe as well as the push registration, so a
+    // backend that only probes still learns this host's graphs.
+    let workflows = workflows::advertised_workflows_for(bridge).await;
+    if !workflows.is_empty() {
+        match serde_json::to_value(workflows) {
+            Ok(value) => {
+                caps.insert("workflows".to_string(), value);
+            }
+            Err(err) => log::warn!("[medulla] failed to serialize workflow adverts: {err}"),
+        }
+    }
+
+    serde_json::Value::Object(caps)
+}
+
 /// Serialize `payload` and emit it as a Socket.IO event on the global backend
 /// socket. Best-effort: a missing/disconnected socket is logged, not fatal.
 fn emit<T: serde::Serialize>(event: &str, payload: T) {
@@ -508,6 +645,32 @@ fn emit<T: serde::Serialize>(event: &str, payload: T) {
     });
 }
 
+/// Serialize and enqueue one medulla event before returning.
+///
+/// Registration snapshots use this awaited form because their full-replacement
+/// ordering must extend through `SocketManager::emit`'s channel send, not stop
+/// at spawning a task that may be scheduled later.
+async fn emit_awaited<T: serde::Serialize>(event: &str, payload: T) -> bool {
+    let data = match serde_json::to_value(&payload) {
+        Ok(value) => value,
+        Err(err) => {
+            log::warn!("[medulla] failed to serialize payload for {event}: {err}");
+            return false;
+        }
+    };
+    let Some(manager) = crate::openhuman::socket::global_socket_manager() else {
+        log::debug!("[medulla] no socket manager — dropping {event}");
+        return false;
+    };
+    match manager.emit(event, data).await {
+        Ok(()) => true,
+        Err(err) => {
+            log::warn!("[medulla] failed to emit {event}: {err}");
+            false
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -563,6 +726,31 @@ mod tests {
         assert_eq!(key, format!("orchestrator_{}", "x".repeat(32)));
     }
 
+    #[test]
+    fn an_undecodable_probe_answers_not_ready_when_it_names_itself() {
+        let raw = serde_json::json!({ "probeId": "p-1", "agentId": 7 });
+        let result = unparsed_capabilities_result(&raw, "invalid type: integer")
+            .expect("a probe that names itself is answerable");
+        assert_eq!(result.probe_id, "p-1");
+        // `ready` + `readyReason` are the two fields the backend's allowlist
+        // keeps; an answer outside them sanitizes to an empty bag, which the
+        // probe treats as no answer at all.
+        assert_eq!(result.capabilities["ready"], false);
+        assert!(result.capabilities["readyReason"]
+            .as_str()
+            .expect("a readable reason")
+            .contains("invalid type: integer"));
+    }
+
+    #[test]
+    fn an_undecodable_probe_without_a_probe_id_is_unanswerable() {
+        // Nothing to correlate, so nothing can be answered — and the socket read
+        // loop must not panic over it either.
+        let raw = serde_json::json!({ "agentId": "orchestrator" });
+        assert!(unparsed_capabilities_result(&raw, "missing field `probeId`").is_none());
+        reject_unparsed_capabilities_request(&raw, "missing field `probeId`");
+    }
+
     #[test]
     fn remaining_budget_reports_time_left_and_exhaustion() {
         let now = Instant::now();

src/openhuman/socket/medulla/payloads.rs

@@ -80,10 +80,16 @@ pub struct TaskResult {
 }
 
 /// A single agent descriptor advertised in the roster.
+///
+/// The identity key is `id`, NOT `agentId`: the backend's roster registry keys
+/// and validates on `id` (`agentRegistry.ts` `hasValidId`), and medulla-v1's
+/// `AgentDescriptor` port declares the same field. An advert that names its
+/// agent under any other key is silently dropped at the backend boundary, so
+/// this must stay `id` on the wire.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct AgentDescriptor {
-    pub agent_id: String,
+    pub id: String,
     pub name: String,
     #[serde(default, skip_serializing_if = "String::is_empty")]
     pub description: String,
@@ -97,6 +103,167 @@ pub struct RegisterAgents {
     pub agents: Vec<AgentDescriptor>,
 }
 
+// ─────────────────────────────────────────────────────────────────────────────
+// Capability handshake
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// `medulla:capabilities_request` — the backend asks one agent to self-report.
+///
+/// The backend fans this out to every harness socket the user holds and
+/// correlates the answer by `probe_id`; an agent that never answers costs the
+/// probe its full ten-second deadline, so [`super::emit_capabilities_result`]
+/// always replies, even when it has nothing interesting to say.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CapabilitiesRequest {
+    pub probe_id: String,
+    #[serde(default)]
+    pub agent_id: String,
+}
+
+/// `medulla:capabilities_result` — the self-report for one probe.
+///
+/// `capabilities` is an open bag the backend narrows through its own allowlist
+/// (`sanitizeCapabilities`), so it stays raw JSON here rather than a struct this
+/// side would have to keep in lockstep with the server's allowlist.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CapabilitiesResult {
+    pub probe_id: String,
+    pub capabilities: serde_json::Value,
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Workflow plane
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// One saved workflow graph as advertised to the backend — enough to choose it
+/// and explain the choice, never the graph itself.
+///
+/// Field-compatible with the tiny.place `WorkflowAdvert`
+/// (`medulla-public/src/sdk/src/tinyplace/frames/types.rs`) and with the
+/// `WorkflowDescriptor` port in medulla-v1, so an advert crossing this boundary
+/// needs no translation layer on either side. `name`/`description` skip
+/// serialization when empty for exactly that reason: the Rust advert already
+/// omits its empty strings, the port therefore declares them optional, and the
+/// backend passes an absent key through as absent rather than fabricating `""`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WorkflowDescriptor {
+    /// Stable identity — the only field the wire always carries and the only one
+    /// anything routes on (delegation, `get`, `runs`).
+    pub id: String,
+    /// Display name; omitted from the wire when blank.
+    #[serde(default, skip_serializing_if = "String::is_empty")]
+    pub name: String,
+    /// Prompt surface — written like a tool description; omitted when blank.
+    #[serde(default, skip_serializing_if = "String::is_empty")]
+    pub description: String,
+    /// How many steps the graph has: a rough cost signal, rendered verbatim.
+    #[serde(default)]
+    pub node_count: u32,
+    /// Whether this host considers the workflow runnable right now. Advisory —
+    /// the reader applies no policy, this host is the one that refuses.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub enabled: Option<bool>,
+    /// Open vocabulary for what starts the workflow ("manual", "cron", …).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub trigger_kind: Option<String>,
+    /// Provenance: which roster agent owns this workflow. Usually left unset —
+    /// the backend stamps the batch-level `agentId` onto any advert without one.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub agent_id: Option<String>,
+    /// Provenance: the workspace the owning agent is deployed into.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub workspace_id: Option<String>,
+}
+
+/// `medulla:register_workflows` — the workflow advert batch, sent on connect and
+/// re-sent whenever the host's store changes. The backend replaces this socket's
+/// whole entry each time and drops it on disconnect, so a shrinking store is
+/// communicated by re-sending the smaller list (never by a delete event).
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct RegisterWorkflows {
+    pub workflows: Vec<WorkflowDescriptor>,
+    /// Batch-level provenance stamped onto adverts that do not name their own
+    /// owner. Absent means "this host, agent unspecified".
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub agent_id: Option<String>,
+}
+
+/// Which read (or the one authoring turn) a `medulla:workflow_request` asks for.
+///
+/// Snake-case on the wire to match the backend's `WorkflowRequestOp` union.
+/// Unknown ops are a decode failure rather than a silent no-op, so a version
+/// skew shows up as a reported error instead of a ten-second server timeout.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum WorkflowOp {
+    /// Fetch one workflow's detail, including its graph.
+    Get,
+    /// The node-kind catalog, optionally filtered to one `kind`.
+    NodeKinds,
+    /// Recent runs of one workflow.
+    Runs,
+    /// A whole authoring turn on this host's own copilot.
+    Copilot,
+}
+
+/// `medulla:workflow_request` — one round trip addressed to the host that
+/// advertised the workflow. Which optional fields matter depends on `op`:
+/// `get`/`runs` read `workflow_id`, `node_kinds` reads `kind`, `copilot` reads
+/// `instruction` plus an optional `workflow_id` (absent ⇒ create).
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WorkflowRequest {
+    pub request_id: String,
+    pub op: WorkflowOp,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub workflow_id: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub kind: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub instruction: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub agent_id: Option<String>,
+}
+
+/// `medulla:workflow_result` — the answer to one round trip.
+///
+/// `data` is OPAQUE: a graph, a node-kind catalog and a copilot outcome are all
+/// host-shaped and pass through the backend verbatim (only size-bounded, at
+/// 1 MiB). `ok: false` carries a readable message the orchestrator renders as a
+/// tool error — a failed read must still be *answered*, because a dropped
+/// request costs the server its whole deadline instead.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WorkflowResult {
+    pub request_id: String,
+    pub ok: bool,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub data: Option<serde_json::Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub error: Option<String>,
+}
+
+/// What a copilot turn reports back: the assistant's reply, the changes it
+/// actually made (derived by the host from a re-read of its own store, never
+/// from the model's claim), and the id of a workflow it created.
+///
+/// Mirrors the `CopilotOutcome` medulla-public's authoring copilot already
+/// returns and the `WorkflowCopilotOutcome` the library port declares, so it
+/// crosses the seam unchanged.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CopilotOutcome {
+    pub reply: String,
+    #[serde(default)]
+    pub changes: Vec<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub created: Option<String>,
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // Socket.IO event names
 // ─────────────────────────────────────────────────────────────────────────────
@@ -105,11 +272,16 @@ pub struct RegisterAgents {
 pub const EVENT_TASK_RUN: &str = "medulla:task_run";
 pub const EVENT_TASK_SEND: &str = "medulla:task_send";
 pub const EVENT_TASK_ABORT: &str = "medulla:task_abort";
+pub const EVENT_CAPABILITIES_REQUEST: &str = "medulla:capabilities_request";
+pub const EVENT_WORKFLOW_REQUEST: &str = "medulla:workflow_request";
 
 /// Up events emitted by openhuman.
 pub const EVENT_TASK_ENVELOPE: &str = "medulla:task_envelope";
 pub const EVENT_TASK_RESULT: &str = "medulla:task_result";
 pub const EVENT_REGISTER_AGENTS: &str = "medulla:register_agents";
+pub const EVENT_REGISTER_WORKFLOWS: &str = "medulla:register_workflows";
+pub const EVENT_CAPABILITIES_RESULT: &str = "medulla:capabilities_result";
+pub const EVENT_WORKFLOW_RESULT: &str = "medulla:workflow_result";
 
 #[cfg(test)]
 mod tests {
@@ -177,17 +349,147 @@ mod tests {
     }
 
     #[test]
-    fn register_agents_round_trips() {
+    fn register_agents_advertises_the_id_key_the_backend_validates() {
         let roster = RegisterAgents {
             agents: vec![AgentDescriptor {
-                agent_id: "orchestrator".into(),
+                id: "orchestrator".into(),
                 name: "Orchestrator".into(),
                 description: "default".into(),
             }],
         };
         let wire = serde_json::to_value(&roster).unwrap();
-        assert_eq!(wire["agents"][0]["agentId"], "orchestrator");
+        // `agentRegistry.hasValidId` keys on `id`; an `agentId` key here would
+        // make the whole roster vanish server-side.
+        assert_eq!(wire["agents"][0]["id"], "orchestrator");
+        assert!(wire["agents"][0].get("agentId").is_none());
         let back: RegisterAgents = serde_json::from_value(wire).unwrap();
         assert_eq!(roster, back);
     }
+
+    #[test]
+    fn workflow_descriptor_omits_blank_name_and_description() {
+        let advert = WorkflowDescriptor {
+            id: "wf-1".into(),
+            name: String::new(),
+            description: String::new(),
+            node_count: 5,
+            enabled: Some(true),
+            trigger_kind: Some("cron".into()),
+            agent_id: None,
+            workspace_id: None,
+        };
+        let wire = serde_json::to_value(&advert).unwrap();
+        assert_eq!(wire["id"], "wf-1");
+        assert_eq!(wire["nodeCount"], 5);
+        assert_eq!(wire["enabled"], true);
+        assert_eq!(wire["triggerKind"], "cron");
+        // Absent, never `""` — the port declares these optional precisely
+        // because the wire omits them.
+        assert!(wire.get("name").is_none());
+        assert!(wire.get("description").is_none());
+        assert!(wire.get("agentId").is_none());
+        let back: WorkflowDescriptor = serde_json::from_value(wire).unwrap();
+        assert_eq!(advert, back);
+    }
+
+    #[test]
+    fn register_workflows_omits_absent_batch_agent_id() {
+        let batch = RegisterWorkflows {
+            workflows: vec![],
+            agent_id: None,
+        };
+        let wire = serde_json::to_value(&batch).unwrap();
+        assert!(wire["workflows"].as_array().unwrap().is_empty());
+        assert!(wire.get("agentId").is_none());
+    }
+
+    #[test]
+    fn workflow_request_reads_every_op_from_the_wire() {
+        for (wire_op, expected) in [
+            ("get", WorkflowOp::Get),
+            ("node_kinds", WorkflowOp::NodeKinds),
+            ("runs", WorkflowOp::Runs),
+            ("copilot", WorkflowOp::Copilot),
+        ] {
+            let parsed: WorkflowRequest =
+                serde_json::from_value(json!({ "requestId": "r1", "op": wire_op })).unwrap();
+            assert_eq!(parsed.op, expected);
+            assert_eq!(parsed.request_id, "r1");
+            assert!(parsed.workflow_id.is_none());
+        }
+        // An op this build does not know is a decode error, not a silent drop.
+        assert!(serde_json::from_value::<WorkflowRequest>(
+            json!({ "requestId": "r1", "op": "apply_ops" })
+        )
+        .is_err());
+    }
+
+    #[test]
+    fn workflow_request_reads_the_op_specific_fields() {
+        let parsed: WorkflowRequest = serde_json::from_value(json!({
+            "requestId": "r2",
+            "op": "copilot",
+            "instruction": "add a slack step",
+            "workflowId": "wf-1",
+            "agentId": "orchestrator",
+        }))
+        .unwrap();
+        assert_eq!(parsed.op, WorkflowOp::Copilot);
+        assert_eq!(parsed.instruction.as_deref(), Some("add a slack step"));
+        assert_eq!(parsed.workflow_id.as_deref(), Some("wf-1"));
+        assert_eq!(parsed.agent_id.as_deref(), Some("orchestrator"));
+    }
+
+    #[test]
+    fn workflow_result_omits_the_unused_arm() {
+        let ok = WorkflowResult {
+            request_id: "r".into(),
+            ok: true,
+            data: Some(json!({ "graph": [] })),
+            error: None,
+        };
+        let wire = serde_json::to_value(&ok).unwrap();
+        assert_eq!(wire["requestId"], "r");
+        assert_eq!(wire["ok"], true);
+        assert!(wire.get("error").is_none());
+
+        let failed = WorkflowResult {
+            request_id: "r".into(),
+            ok: false,
+            data: None,
+            error: Some("unknown workflow".into()),
+        };
+        let wire = serde_json::to_value(&failed).unwrap();
+        assert_eq!(wire["ok"], false);
+        assert_eq!(wire["error"], "unknown workflow");
+        assert!(wire.get("data").is_none());
+    }
+
+    #[test]
+    fn copilot_outcome_matches_the_library_shape() {
+        let outcome = CopilotOutcome {
+            reply: "added the step".into(),
+            changes: vec!["node:slack added".into()],
+            created: None,
+        };
+        let wire = serde_json::to_value(&outcome).unwrap();
+        assert_eq!(wire["reply"], "added the step");
+        assert_eq!(wire["changes"][0], "node:slack added");
+        assert!(wire.get("created").is_none());
+    }
+
+    #[test]
+    fn capabilities_request_defaults_a_missing_agent_id() {
+        let parsed: CapabilitiesRequest =
+            serde_json::from_value(json!({ "probeId": "p1" })).unwrap();
+        assert_eq!(parsed.probe_id, "p1");
+        assert!(parsed.agent_id.is_empty());
+        let wire = serde_json::to_value(CapabilitiesResult {
+            probe_id: "p1".into(),
+            capabilities: json!({ "ready": true }),
+        })
+        .unwrap();
+        assert_eq!(wire["probeId"], "p1");
+        assert_eq!(wire["capabilities"]["ready"], true);
+    }
 }

src/openhuman/socket/medulla/workflows.rs

@@ -0,0 +1,872 @@
+//! The workflow plane of the medulla harness protocol: advertising this host's
+//! saved workflow graphs, and serving the reads (and the one authoring turn)
+//! the orchestrator round-trips back to whoever owns them.
+//!
+//! **Why a trait and not a direct call into `flows`.** Two different hosts embed
+//! this transport — openhuman over its SQLite `flows` store, and medulla-public
+//! over its own layered JSON workflow store — and they share nothing but the
+//! wire shape. So the store side is a host-supplied [`WorkflowBridge`] the host
+//! installs once at startup ([`set_workflow_bridge`]), and the engine types stay
+//! entirely on the host's side of the seam: a graph, a node-kind catalog and a
+//! run list all cross as opaque [`serde_json::Value`], exactly as they do in
+//! medulla-v1's `WorkflowCatalog` port. Nothing in this module parses a node, an
+//! edge, or a config field.
+//!
+//! **Why every path answers.** The server correlates a request by `requestId`
+//! and waits on a deadline — ten seconds for a read, ten *minutes* for a copilot
+//! turn. A dropped request is therefore not a cheap no-op: it burns that whole
+//! window and, for `copilot`, holds a promise open for the rest of the cycle. So
+//! a missing bridge, an unknown op argument, a bridge error and even a panicking
+//! bridge method all become a `medulla:workflow_result` with `ok: false` and a
+//! readable message. The only way to stay silent is for the socket to be gone,
+//! which the server already handles by retiring the waiter on disconnect.
+
+use std::future::Future;
+use std::panic::AssertUnwindSafe;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, OnceLock};
+
+use async_trait::async_trait;
+use futures::FutureExt;
+use parking_lot::RwLock;
+use serde_json::Value;
+use tokio_util::sync::CancellationToken;
+
+use super::payloads::{
+    CopilotOutcome, RegisterWorkflows, WorkflowDescriptor, WorkflowOp, WorkflowRequest,
+    WorkflowResult, EVENT_REGISTER_WORKFLOWS, EVENT_WORKFLOW_RESULT,
+};
+
+/// The store side of the workflow plane, supplied by the embedding host.
+///
+/// The reads are synchronous because every host backs them with a local store;
+/// they are dispatched on a blocking thread so a slow store cannot stall the
+/// socket runtime. `copilot` is `async` because it is not a read at all — it is
+/// a whole agent turn on the host's own authoring copilot.
+///
+/// Every method returns `Result<_, String>` rather than a host error type: the
+/// error is going straight onto a model's prompt surface as a tool result, so it
+/// must already be a sentence a reader can act on.
+#[async_trait]
+pub trait WorkflowBridge: Send + Sync {
+    /// The adverts this host publishes. Cheap and side-effect free — it is
+    /// called on every (re)connect and on every store change.
+    fn list(&self) -> Vec<WorkflowDescriptor>;
+
+    /// Fallible advert read used by registration.
+    ///
+    /// The default preserves compatibility for hosts whose list cannot fail.
+    /// Store-backed hosts override it so a transient read failure does not look
+    /// like a successful empty replacement batch.
+    fn try_list(&self) -> Result<Vec<WorkflowDescriptor>, String> {
+        Ok(self.list())
+    }
+
+    /// One workflow's detail, graph included, as the host's own JSON shape.
+    fn get(&self, id: &str) -> Result<Value, String>;
+
+    /// The node-kind catalog, optionally narrowed to a single `kind`. This is an
+    /// authoring aid: a manager reads it to brief the copilot accurately.
+    fn node_kinds(&self, kind: Option<&str>) -> Result<Value, String>;
+
+    /// Recent runs of one workflow, for visibility into a delegated graph.
+    fn runs(&self, id: &str) -> Result<Value, String>;
+
+    /// Run one authoring turn on the host's copilot. `workflow_id` absent means
+    /// "create a new workflow". The outcome must be derived from a re-read of
+    /// the host's own store rather than from the model's claim about what it
+    /// did — that is the accountability property the whole delegation rests on.
+    async fn copilot(
+        &self,
+        instruction: &str,
+        workflow_id: Option<&str>,
+    ) -> Result<CopilotOutcome, String>;
+
+    /// Which roster agent owns these workflows, when the host runs more than
+    /// one identity. Stamped onto the advert batch so the backend can route a
+    /// delegated workflow to its owner. Defaults to unset — the backend then
+    /// treats the whole socket as the owner.
+    fn agent_id(&self) -> Option<String> {
+        None
+    }
+
+    /// The action directory associated with this bridge's authenticated
+    /// identity, when the host has one. Capability probes use this instead of
+    /// reloading process-global active-user state.
+    fn action_dir(&self) -> Option<String> {
+        None
+    }
+}
+
+#[derive(Clone)]
+pub(super) struct BridgeGeneration {
+    pub(super) bridge: Option<Arc<dyn WorkflowBridge>>,
+    pub(super) cancel: CancellationToken,
+}
+
+struct BridgeState {
+    bridge: Option<Arc<dyn WorkflowBridge>>,
+    generation: CancellationToken,
+}
+
+struct ConnectionGeneration {
+    current: CancellationToken,
+}
+
+impl ConnectionGeneration {
+    fn disconnected() -> Self {
+        let current = CancellationToken::new();
+        current.cancel();
+        Self { current }
+    }
+
+    fn snapshot(&self) -> CancellationToken {
+        self.current.clone()
+    }
+
+    fn begin(&mut self) {
+        self.current.cancel();
+        self.current = CancellationToken::new();
+    }
+
+    fn end(&mut self) {
+        self.current.cancel();
+        *self = Self::disconnected();
+    }
+}
+
+static BRIDGE_STATE: OnceLock<RwLock<BridgeState>> = OnceLock::new();
+static CONNECTION_GENERATION: OnceLock<RwLock<ConnectionGeneration>> = OnceLock::new();
+
+fn bridge_state() -> &'static RwLock<BridgeState> {
+    BRIDGE_STATE.get_or_init(|| {
+        RwLock::new(BridgeState {
+            bridge: None,
+            generation: CancellationToken::new(),
+        })
+    })
+}
+
+fn connection_generation_state() -> &'static RwLock<ConnectionGeneration> {
+    CONNECTION_GENERATION.get_or_init(|| RwLock::new(ConnectionGeneration::disconnected()))
+}
+
+pub(super) fn bridge_generation() -> BridgeGeneration {
+    let state = bridge_state().read();
+    BridgeGeneration {
+        bridge: state.bridge.clone(),
+        cancel: state.generation.clone(),
+    }
+}
+
+pub(super) fn connection_generation() -> CancellationToken {
+    connection_generation_state().read().snapshot()
+}
+
+/// Start a new authenticated-socket lifetime.
+///
+/// Called by the `ready` handler before it advertises workflows. Work received
+/// before `ready`, while reconnecting, or after a disconnect sees a cancelled
+/// generation and is discarded instead of outliving the backend waiter that
+/// sent it.
+pub(in crate::openhuman::socket) fn begin_connection_generation() {
+    connection_generation_state().write().begin();
+}
+
+/// End the current authenticated-socket lifetime and cancel its work.
+pub(in crate::openhuman::socket) fn end_connection_generation() {
+    connection_generation_state().write().end();
+}
+
+fn replace_bridge(bridge: Option<Arc<dyn WorkflowBridge>>) -> CancellationToken {
+    let mut state = bridge_state().write();
+    state.generation.cancel();
+    let generation = CancellationToken::new();
+    state.bridge = bridge;
+    state.generation = generation.clone();
+    generation
+}
+
+/// Orders overlapping snapshot emissions by successful registration revision.
+///
+/// Reads stay concurrent and off the socket runtime. A newer successful read
+/// supersedes every older one, while a newer failed read does not suppress an
+/// older successful snapshot. The mutex makes the revision check plus the
+/// awaited socket enqueue one ordered operation.
+struct RegistrationSequencer {
+    next_revision: AtomicU64,
+    last_emitted: tokio::sync::Mutex<u64>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum SequencedEmit {
+    Emitted,
+    Superseded,
+    Failed,
+}
+
+impl RegistrationSequencer {
+    const fn new() -> Self {
+        Self {
+            next_revision: AtomicU64::new(0),
+            last_emitted: tokio::sync::Mutex::const_new(0),
+        }
+    }
+
+    fn begin(&self) -> u64 {
+        self.next_revision.fetch_add(1, Ordering::Relaxed) + 1
+    }
+
+    async fn emit_if_newer<F, Fut>(&self, revision: u64, emit: F) -> SequencedEmit
+    where
+        F: FnOnce() -> Fut,
+        Fut: Future<Output = bool>,
+    {
+        let mut last_emitted = self.last_emitted.lock().await;
+        if revision <= *last_emitted {
+            return SequencedEmit::Superseded;
+        }
+        if !emit().await {
+            return SequencedEmit::Failed;
+        }
+        *last_emitted = revision;
+        SequencedEmit::Emitted
+    }
+}
+
+static REGISTRATION_SEQUENCE: RegistrationSequencer = RegistrationSequencer::new();
+
+/// Install (or replace) the host's workflow bridge, and advertise what it holds.
+///
+/// Registering re-advertises immediately so a bridge installed after connect
+/// still reaches a backend that has already seen this socket's `ready`.
+pub fn set_workflow_bridge(bridge: Arc<dyn WorkflowBridge>) {
+    replace_bridge(Some(bridge));
+    emit_register_workflows();
+}
+
+/// Remove the installed bridge and tell the backend this socket now advertises
+/// nothing. Sending an empty batch is the only way to retract: the backend
+/// replaces a socket's whole entry on each registration and has no delete event.
+pub fn clear_workflow_bridge() {
+    let generation = replace_bridge(None);
+    let connection_cancel = connection_generation();
+    let revision = REGISTRATION_SEQUENCE.begin();
+    tokio::spawn(async move {
+        tokio::select! {
+            biased;
+            _ = generation.cancelled() => {}
+            _ = connection_cancel.cancelled() => {}
+            _ = async move {
+                let batch = RegisterWorkflows {
+                    workflows: Vec::new(),
+                    agent_id: None,
+                };
+                REGISTRATION_SEQUENCE
+                    .emit_if_newer(revision, || emit_batch(batch))
+                    .await;
+            } => {}
+        }
+    });
+}
+
+/// Emit `medulla:register_workflows` — the advert batch, sent on connect and
+/// whenever the host signals its workflow set changed.
+///
+/// A host without a bridge stays silent rather than advertising an empty list:
+/// "I have no workflow plane" and "my store is empty" are different claims, and
+/// only the second one should clear a previously registered set.
+pub fn emit_register_workflows() {
+    let generation = bridge_generation();
+    let Some(bridge) = generation.bridge else {
+        log::debug!("[medulla] no workflow bridge installed — not advertising workflows");
+        return;
+    };
+    let revision = REGISTRATION_SEQUENCE.begin();
+    let cancel = generation.cancel;
+    let connection_cancel = connection_generation();
+    // `list()` reads the host's store, so keep it off the socket runtime; a
+    // panicking bridge surfaces as a `JoinError` here instead of unwinding.
+    tokio::spawn(async move {
+        tokio::select! {
+            biased;
+            _ = cancel.cancelled() => {
+                log::debug!("[medulla] discarded workflow registration from an old bridge");
+            }
+            _ = connection_cancel.cancelled() => {
+                log::debug!("[medulla] discarded workflow registration from a closed socket");
+            }
+            _ = async move {
+                match read_batch(bridge).await {
+                    Ok(batch) => {
+                        let count = batch.workflows.len();
+                        match REGISTRATION_SEQUENCE
+                            .emit_if_newer(revision, || emit_batch(batch))
+                            .await
+                        {
+                            SequencedEmit::Emitted => {
+                                log::info!("[medulla] advertised {count} workflows to backend");
+                            }
+                            SequencedEmit::Superseded => log::debug!(
+                                "[medulla] workflow registration snapshot already superseded by a newer \
+                                 successful read — not advertising"
+                            ),
+                            SequencedEmit::Failed => {
+                                log::debug!("[medulla] workflow registration enqueue failed")
+                            }
+                        }
+                    }
+                    Err(err) => {
+                        log::warn!("[medulla] workflow bridge list() failed — not advertising: {err}")
+                    }
+                }
+            } => {}
+        }
+    });
+}
+
+/// The adverts this host currently publishes, for callers that need the list
+/// itself rather than the registration frame — today the capability probe, which
+/// carries the same descriptors in its reply bag. Empty when no bridge is
+/// installed or the bridge failed, so a probe still answers.
+pub async fn advertised_workflows() -> Vec<WorkflowDescriptor> {
+    advertised_workflows_for(bridge_generation().bridge).await
+}
+
+pub(super) async fn advertised_workflows_for(
+    bridge: Option<Arc<dyn WorkflowBridge>>,
+) -> Vec<WorkflowDescriptor> {
+    let Some(bridge) = bridge else {
+        return Vec::new();
+    };
+    read_batch(bridge)
+        .await
+        .map(|batch| batch.workflows)
+        .unwrap_or_default()
+}
+
+/// Read one advert batch off the bridge on a blocking thread. An error or panic
+/// remains a failed read — never an empty replacement or a partial batch.
+async fn read_batch(bridge: Arc<dyn WorkflowBridge>) -> Result<RegisterWorkflows, String> {
+    match tokio::task::spawn_blocking(move || {
+        Ok(RegisterWorkflows {
+            workflows: bridge.try_list()?,
+            agent_id: bridge.agent_id(),
+        })
+    })
+    .await
+    {
+        Ok(batch) => batch,
+        Err(err) => Err(format!("the workflow store failed to list: {err}")),
+    }
+}
+
+async fn emit_batch(batch: RegisterWorkflows) -> bool {
+    super::emit_awaited(EVENT_REGISTER_WORKFLOWS, batch).await
+}
+
+/// Handle an inbound `medulla:workflow_request`.
+///
+/// Returns immediately: the work runs on its own task so a `copilot` turn (up to
+/// ten minutes of agent time) cannot hold the socket read loop, the same shape
+/// `medulla:task_run` and `orch:tool_call` already use.
+pub fn handle_workflow_request(request: WorkflowRequest) {
+    let BridgeGeneration { bridge, cancel } = bridge_generation();
+    let connection_cancel = connection_generation();
+    tokio::spawn(async move {
+        tokio::select! {
+            biased;
+            _ = cancel.cancelled() => {
+                log::debug!("[medulla] discarded workflow request from an old bridge");
+            }
+            _ = connection_cancel.cancelled() => {
+                log::debug!("[medulla] discarded workflow request from a closed socket");
+            }
+            _ = async move {
+                let request_id = request.request_id.clone();
+                let op = request.op;
+                let outcome = match bridge {
+                    Some(bridge) => dispatch(bridge, request).await,
+                    None => Err("this agent has no workflow store installed".to_string()),
+                };
+                match &outcome {
+                    Ok(_) => log::debug!("[medulla] workflow {op:?} request_id={request_id} ok"),
+                    Err(err) => {
+                        log::warn!("[medulla] workflow {op:?} request_id={request_id} failed: {err}")
+                    }
+                }
+                super::emit_awaited(
+                    EVENT_WORKFLOW_RESULT,
+                    result_frame(request_id, outcome),
+                )
+                .await;
+            } => {}
+        }
+    });
+}
+
+/// Answer a `medulla:workflow_request` this build could not decode.
+///
+/// A frame with an unknown `op` (a newer backend) or a missing field would
+/// otherwise be dropped, and a dropped request costs the backend the op's whole
+/// deadline. The `requestId` is recovered from the raw JSON — without one there
+/// is nothing to correlate, so the frame really is unanswerable and is only
+/// logged.
+pub fn reject_unparsed_request(raw: &Value, reason: &str) {
+    let Some(request_id) = raw.get("requestId").and_then(Value::as_str) else {
+        log::warn!("[medulla] undecodable workflow_request carries no requestId — cannot answer");
+        return;
+    };
+    super::emit(
+        EVENT_WORKFLOW_RESULT,
+        result_frame(
+            request_id.to_string(),
+            Err(format!("this agent could not read the request: {reason}")),
+        ),
+    );
+}
+
+/// Project a dispatch outcome onto the wire frame. Split out so the mapping
+/// (which arm is populated, and that exactly one of them is) is directly
+/// testable without a socket or a runtime.
+fn result_frame(request_id: String, outcome: Result<Value, String>) -> WorkflowResult {
+    match outcome {
+        Ok(data) => WorkflowResult {
+            request_id,
+            ok: true,
+            data: Some(data),
+            error: None,
+        },
+        Err(error) => WorkflowResult {
+            request_id,
+            ok: false,
+            data: None,
+            error: Some(error),
+        },
+    }
+}
+
+/// Route one request to `bridge`. Every failure mode — a missing op argument, a
+/// bridge error, a panicking bridge method — comes back as an `Err(String)` the
+/// caller turns into `ok: false`.
+///
+/// The bridge is a parameter rather than read from the registry here so the
+/// whole dispatch table is exercisable against a scripted bridge without
+/// touching process-global state.
+async fn dispatch(
+    bridge: Arc<dyn WorkflowBridge>,
+    request: WorkflowRequest,
+) -> Result<Value, String> {
+    match request.op {
+        WorkflowOp::Get => {
+            let id = require_workflow_id(&request, "get")?;
+            blocking(move || bridge.get(&id)).await
+        }
+        WorkflowOp::NodeKinds => {
+            let kind = request.kind.clone();
+            blocking(move || bridge.node_kinds(kind.as_deref())).await
+        }
+        WorkflowOp::Runs => {
+            let id = require_workflow_id(&request, "runs")?;
+            blocking(move || bridge.runs(&id)).await
+        }
+        WorkflowOp::Copilot => {
+            let instruction = request
+                .instruction
+                .as_deref()
+                .map(str::trim)
+                .filter(|value| !value.is_empty())
+                .ok_or_else(|| "workflow copilot requires a non-empty instruction".to_string())?;
+            let turn = bridge.copilot(instruction, request.workflow_id.as_deref());
+            let outcome = catching_panics(turn).await??;
+            serde_json::to_value(outcome)
+                .map_err(|err| format!("failed to serialize the copilot outcome: {err}"))
+        }
+    }
+}
+
+/// The `workflowId` an op needs, or a message naming what was missing. Blank is
+/// treated as absent: an empty id could only ever miss in the host's store, and
+/// saying so up front beats a "not found" the reader cannot interpret.
+fn require_workflow_id(request: &WorkflowRequest, op: &str) -> Result<String, String> {
+    request
+        .workflow_id
+        .as_deref()
+        .map(str::trim)
+        .filter(|id| !id.is_empty())
+        .map(str::to_string)
+        .ok_or_else(|| format!("workflow {op} requires a workflowId"))
+}
+
+/// Run a synchronous bridge read on a blocking thread, mapping a panic inside it
+/// to an error rather than letting it unwind through the task.
+async fn blocking<F>(read: F) -> Result<Value, String>
+where
+    F: FnOnce() -> Result<Value, String> + Send + 'static,
+{
+    match tokio::task::spawn_blocking(read).await {
+        Ok(result) => result,
+        Err(err) => Err(format!("the workflow store failed to answer: {err}")),
+    }
+}
+
+/// The async twin of [`blocking`]: run a bridge turn that must be awaited on the
+/// socket runtime — today only `copilot` — and map a panic inside it to an error.
+///
+/// `copilot` cannot go on a blocking thread (it is a whole agent turn, not a
+/// store read), so it does not inherit `spawn_blocking`'s `JoinError` net. Left
+/// bare, a panicking `copilot` would abort the task that owes the backend a
+/// `workflow_result` and cost it the op's whole ten-minute deadline. Caught
+/// in-place rather than on a supervised child task so cancelling the answering
+/// task still cancels the turn.
+async fn catching_panics<F, T>(turn: F) -> Result<T, String>
+where
+    F: Future<Output = T>,
+{
+    match AssertUnwindSafe(turn).catch_unwind().await {
+        Ok(value) => Ok(value),
+        Err(panic) => {
+            let reason = panic
+                .downcast_ref::<&str>()
+                .copied()
+                .or_else(|| panic.downcast_ref::<String>().map(String::as_str))
+                .unwrap_or("unknown panic");
+            log::error!("[medulla] workflow copilot panicked: {reason}");
+            Err(format!(
+                "the workflow store failed to answer: the copilot panicked: {reason}"
+            ))
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    /// A bridge whose every read is scripted, so the dispatch table can be
+    /// exercised without a store.
+    #[derive(Default)]
+    struct FakeBridge {
+        fail: bool,
+        fail_list: bool,
+        panic_on_get: bool,
+        panic_on_copilot: bool,
+    }
+
+    #[async_trait]
+    impl WorkflowBridge for FakeBridge {
+        fn list(&self) -> Vec<WorkflowDescriptor> {
+            vec![WorkflowDescriptor {
+                id: "wf-1".into(),
+                name: "Deploy".into(),
+                description: String::new(),
+                node_count: 3,
+                enabled: None,
+                trigger_kind: None,
+                agent_id: None,
+                workspace_id: None,
+            }]
+        }
+
+        fn try_list(&self) -> Result<Vec<WorkflowDescriptor>, String> {
+            if self.fail_list {
+                return Err("store unavailable".into());
+            }
+            Ok(self.list())
+        }
+
+        fn get(&self, id: &str) -> Result<Value, String> {
+            if self.panic_on_get {
+                panic!("store exploded");
+            }
+            if self.fail {
+                return Err("no such workflow".into());
+            }
+            Ok(json!({ "id": id, "graph": [] }))
+        }
+
+        fn node_kinds(&self, kind: Option<&str>) -> Result<Value, String> {
+            Ok(json!({ "kind": kind }))
+        }
+
+        fn runs(&self, id: &str) -> Result<Value, String> {
+            Ok(json!({ "runs": [id] }))
+        }
+
+        async fn copilot(
+            &self,
+            instruction: &str,
+            workflow_id: Option<&str>,
+        ) -> Result<CopilotOutcome, String> {
+            if self.panic_on_copilot {
+                panic!("copilot exploded");
+            }
+            if self.fail {
+                return Err("copilot refused".into());
+            }
+            Ok(CopilotOutcome {
+                reply: instruction.to_string(),
+                changes: vec![],
+                created: workflow_id.is_none().then(|| "wf-new".to_string()),
+            })
+        }
+
+        fn agent_id(&self) -> Option<String> {
+            Some("orchestrator".into())
+        }
+    }
+
+    /// A scripted bridge, handed straight to `dispatch` — the tests never touch
+    /// the process-global registry, so they stay order- and thread-independent.
+    fn bridge_of(fail: bool, panic_on_get: bool) -> Arc<dyn WorkflowBridge> {
+        Arc::new(FakeBridge {
+            fail,
+            panic_on_get,
+            ..FakeBridge::default()
+        })
+    }
+
+    /// A bridge whose authoring turn panics mid-`await` — the async twin of
+    /// `panic_on_get`, and the one arm `blocking()` does not cover.
+    fn panicking_copilot_bridge() -> Arc<dyn WorkflowBridge> {
+        Arc::new(FakeBridge {
+            panic_on_copilot: true,
+            ..FakeBridge::default()
+        })
+    }
+
+    fn request(op: WorkflowOp) -> WorkflowRequest {
+        WorkflowRequest {
+            request_id: "r1".into(),
+            op,
+            workflow_id: None,
+            kind: None,
+            instruction: None,
+            agent_id: None,
+        }
+    }
+
+    #[test]
+    fn result_frame_populates_exactly_one_arm() {
+        let ok = result_frame("r1".into(), Ok(json!({ "a": 1 })));
+        assert!(ok.ok);
+        assert_eq!(ok.data, Some(json!({ "a": 1 })));
+        assert!(ok.error.is_none());
+
+        let failed = result_frame("r1".into(), Err("boom".into()));
+        assert!(!failed.ok);
+        assert!(failed.data.is_none());
+        assert_eq!(failed.error.as_deref(), Some("boom"));
+    }
+
+    #[test]
+    fn connection_generation_cancels_work_at_each_lifetime_boundary() {
+        let mut generation = ConnectionGeneration::disconnected();
+        assert!(generation.snapshot().is_cancelled());
+
+        generation.begin();
+        let connected = generation.snapshot();
+        assert!(!connected.is_cancelled());
+
+        generation.end();
+        assert!(connected.is_cancelled());
+        assert!(generation.snapshot().is_cancelled());
+    }
+
+    #[test]
+    fn missing_workflow_id_is_reported_not_guessed() {
+        let mut req = request(WorkflowOp::Get);
+        assert_eq!(
+            require_workflow_id(&req, "get"),
+            Err("workflow get requires a workflowId".to_string())
+        );
+        // Blank is absent, not an id that will merely miss in the store.
+        req.workflow_id = Some("   ".into());
+        assert!(require_workflow_id(&req, "get").is_err());
+        req.workflow_id = Some(" wf-1 ".into());
+        assert_eq!(require_workflow_id(&req, "get"), Ok("wf-1".to_string()));
+    }
+
+    #[tokio::test]
+    async fn dispatch_routes_each_read_to_the_bridge() {
+        let bridge = bridge_of(false, false);
+        let mut get = request(WorkflowOp::Get);
+        get.workflow_id = Some("wf-1".into());
+        assert_eq!(
+            dispatch(Arc::clone(&bridge), get).await.unwrap(),
+            json!({ "id": "wf-1", "graph": [] })
+        );
+
+        let mut kinds = request(WorkflowOp::NodeKinds);
+        kinds.kind = Some("agent".into());
+        assert_eq!(
+            dispatch(Arc::clone(&bridge), kinds).await.unwrap(),
+            json!({ "kind": "agent" })
+        );
+
+        let mut runs = request(WorkflowOp::Runs);
+        runs.workflow_id = Some("wf-1".into());
+        assert_eq!(
+            dispatch(bridge, runs).await.unwrap(),
+            json!({ "runs": ["wf-1"] })
+        );
+    }
+
+    #[tokio::test]
+    async fn a_read_missing_its_workflow_id_fails_without_reaching_the_store() {
+        let outcome = dispatch(bridge_of(false, true), request(WorkflowOp::Get)).await;
+        // `panic_on_get` proves the store was never consulted.
+        assert_eq!(
+            outcome,
+            Err("workflow get requires a workflowId".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn copilot_requires_an_instruction_and_reports_creation() {
+        let bridge = bridge_of(false, false);
+        // A blank instruction never reaches the host's agent.
+        let mut blank = request(WorkflowOp::Copilot);
+        blank.instruction = Some("  ".into());
+        assert!(dispatch(Arc::clone(&bridge), blank).await.is_err());
+
+        let mut create = request(WorkflowOp::Copilot);
+        create.instruction = Some("build a deploy flow".into());
+        let data = dispatch(bridge, create).await.unwrap();
+        assert_eq!(data["reply"], "build a deploy flow");
+        assert_eq!(data["created"], "wf-new");
+    }
+
+    #[tokio::test]
+    async fn a_bridge_error_becomes_a_readable_failure() {
+        let bridge = bridge_of(true, false);
+        let mut get = request(WorkflowOp::Get);
+        get.workflow_id = Some("wf-1".into());
+        assert_eq!(
+            dispatch(Arc::clone(&bridge), get).await,
+            Err("no such workflow".to_string())
+        );
+
+        let mut copilot = request(WorkflowOp::Copilot);
+        copilot.instruction = Some("change it".into());
+        assert_eq!(
+            dispatch(bridge, copilot).await,
+            Err("copilot refused".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn a_list_error_is_not_converted_to_an_empty_registration() {
+        let bridge: Arc<dyn WorkflowBridge> = Arc::new(FakeBridge {
+            fail_list: true,
+            ..FakeBridge::default()
+        });
+        assert_eq!(
+            read_batch(bridge).await.unwrap_err(),
+            "store unavailable".to_string()
+        );
+    }
+
+    #[tokio::test]
+    async fn a_newer_registration_suppresses_an_older_snapshot() {
+        let sequencer = RegistrationSequencer::new();
+        let older = sequencer.begin();
+        let newer = sequencer.begin();
+        let emitted = std::sync::Mutex::new(Vec::new());
+
+        assert_eq!(
+            sequencer
+                .emit_if_newer(newer, || async {
+                    emitted.lock().unwrap().push("newer");
+                    true
+                })
+                .await,
+            SequencedEmit::Emitted
+        );
+        assert_eq!(
+            sequencer
+                .emit_if_newer(older, || async {
+                    emitted.lock().unwrap().push("older");
+                    true
+                })
+                .await,
+            SequencedEmit::Superseded
+        );
+        assert_eq!(*emitted.lock().unwrap(), vec!["newer"]);
+    }
+
+    #[tokio::test]
+    async fn a_failed_newer_read_does_not_suppress_an_older_success() {
+        let sequencer = RegistrationSequencer::new();
+        let older = sequencer.begin();
+        let _newer = sequencer.begin();
+        let emitted = std::sync::Mutex::new(Vec::new());
+
+        // The newer read failed, so it never calls `emit_if_newer`. The older
+        // successful snapshot must still advance the backend from its
+        // pre-mutation state.
+        assert_eq!(
+            sequencer
+                .emit_if_newer(older, || async {
+                    emitted.lock().unwrap().push("older");
+                    true
+                })
+                .await,
+            SequencedEmit::Emitted
+        );
+        assert_eq!(*emitted.lock().unwrap(), vec!["older"]);
+    }
+
+    #[tokio::test]
+    async fn a_failed_enqueue_does_not_advance_the_success_watermark() {
+        let sequencer = RegistrationSequencer::new();
+        let older = sequencer.begin();
+        let newer = sequencer.begin();
+
+        assert_eq!(
+            sequencer.emit_if_newer(newer, || async { false }).await,
+            SequencedEmit::Failed
+        );
+        assert_eq!(
+            sequencer.emit_if_newer(older, || async { true }).await,
+            SequencedEmit::Emitted
+        );
+    }
+
+    #[tokio::test]
+    async fn a_panicking_bridge_still_answers() {
+        let mut get = request(WorkflowOp::Get);
+        get.workflow_id = Some("wf-1".into());
+        let outcome = dispatch(bridge_of(false, true), get).await;
+        // The request must not be dropped — a dropped one costs the server its
+        // whole deadline.
+        assert!(outcome
+            .unwrap_err()
+            .starts_with("the workflow store failed to answer"));
+    }
+
+    #[tokio::test]
+    async fn a_panicking_copilot_still_answers() {
+        let mut copilot = request(WorkflowOp::Copilot);
+        copilot.instruction = Some("build a deploy flow".into());
+        let outcome = dispatch(panicking_copilot_bridge(), copilot).await;
+        // A panic inside the authoring turn must come back as an answer, not
+        // abort the task that owes the backend a frame: an unanswered copilot
+        // costs the server its whole ten-minute deadline.
+        assert!(outcome
+            .clone()
+            .unwrap_err()
+            .starts_with("the workflow store failed to answer"));
+        let frame = result_frame("r1".into(), outcome);
+        assert!(!frame.ok);
+        assert!(frame.error.is_some());
+    }
+
+    #[test]
+    fn an_undecodable_frame_without_a_request_id_is_only_logged() {
+        // Nothing to correlate, so nothing can be answered — but it must not
+        // panic on the socket read path either.
+        reject_unparsed_request(&json!({ "op": "apply_ops" }), "unknown variant");
+    }
+}

src/openhuman/socket/mod.rs

@@ -7,6 +7,7 @@
 mod event_handlers;
 pub mod manager;
 pub mod medulla;
+mod ops;
 mod schemas;
 pub(crate) mod token_provider;
 pub mod types;

src/openhuman/socket/ops.rs

@@ -0,0 +1,87 @@
+use crate::api::models::socket::SocketState;
+
+use super::SocketManager;
+
+async fn connect_static_using(
+    manager: &SocketManager,
+    url: &str,
+    token: &str,
+    clear_workflows: impl FnOnce(),
+) -> Result<SocketState, String> {
+    let _rebind = manager.lock_identity_rebind().await;
+    manager.disconnect().await?;
+    clear_workflows();
+    manager.connect(url, token).await?;
+    Ok(manager.get_state())
+}
+
+pub async fn connect_static(
+    manager: &SocketManager,
+    url: &str,
+    token: &str,
+) -> Result<SocketState, String> {
+    log::info!("[socket:rpc] connect — disabling identity-bound workflow plane");
+    connect_static_using(
+        manager,
+        url,
+        token,
+        super::medulla::workflows::clear_workflow_bridge,
+    )
+    .await
+}
+
+pub async fn disconnect(manager: &SocketManager) -> Result<SocketState, String> {
+    log::info!("[socket:rpc] disconnect");
+    let _rebind = manager.lock_identity_rebind().await;
+    manager.disconnect().await?;
+    Ok(manager.get_state())
+}
+
+pub async fn connect_with_session(manager: &SocketManager) -> Result<SocketState, String> {
+    log::info!("[socket:rpc] connect_with_session — resolving credentials");
+    let config =
+        std::sync::Arc::new(crate::openhuman::config::rpc::load_config_with_timeout().await?);
+    let api_url = crate::api::config::effective_backend_api_url(&config.api_url);
+    crate::api::jwt::get_session_token(&config)
+        .map_err(|e| format!("failed to read session token: {e}"))?
+        .ok_or("no session token stored — user must log in first")?;
+
+    let _rebind = manager.lock_identity_rebind().await;
+    manager.disconnect().await?;
+    #[cfg(feature = "flows")]
+    if crate::core::runtime::context::CoreContext::current()
+        .is_some_and(|context| context.domains().flows)
+    {
+        crate::openhuman::flows::medulla_bridge::install(std::sync::Arc::clone(&config));
+    }
+    let provider = super::token_provider::token_provider_from_config(config);
+    manager.connect_with_provider(&api_url, provider).await?;
+    Ok(manager.get_state())
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::atomic::{AtomicBool, Ordering};
+
+    use super::*;
+
+    #[tokio::test]
+    async fn static_token_connection_clears_identity_state_after_disconnect() {
+        let manager = SocketManager::new();
+        manager
+            .connect("http://127.0.0.1:1", "opaque-token")
+            .await
+            .unwrap();
+        let cleared = AtomicBool::new(false);
+        connect_static_using(&manager, "http://127.0.0.1:1", "replacement", || {
+            assert_eq!(
+                manager.get_state().status,
+                crate::openhuman::socket::types::ConnectionStatus::Disconnected
+            );
+            cleared.store(true, Ordering::SeqCst);
+        })
+        .await
+        .unwrap();
+        assert!(cleared.load(Ordering::SeqCst));
+    }
+}

src/openhuman/socket/schemas.rs

@@ -171,10 +171,7 @@ fn handle_connect(params: Map<String, Value>) -> ControllerFuture {
             .and_then(|v| v.as_str())
             .ok_or("missing required param 'token'")?;
 
-        log::info!("[socket:rpc] connect url={}", url);
-        mgr.connect(url, token).await?;
-
-        let state = mgr.get_state();
+        let state = super::ops::connect_static(mgr, url, token).await?;
         Ok(json!({ "status": format!("{:?}", state.status) }))
     })
 }
@@ -182,10 +179,7 @@ fn handle_connect(params: Map<String, Value>) -> ControllerFuture {
 fn handle_disconnect(_params: Map<String, Value>) -> ControllerFuture {
     Box::pin(async move {
         let mgr = require_manager()?;
-        log::info!("[socket:rpc] disconnect");
-        mgr.disconnect().await?;
-
-        let state = mgr.get_state();
+        let state = super::ops::disconnect(mgr).await?;
         Ok(json!({ "status": format!("{:?}", state.status) }))
     })
 }
@@ -217,39 +211,7 @@ fn handle_emit(params: Map<String, Value>) -> ControllerFuture {
 fn handle_connect_with_session(_params: Map<String, Value>) -> ControllerFuture {
     Box::pin(async move {
         let mgr = require_manager()?;
-
-        log::info!("[socket:rpc] connect_with_session — resolving credentials");
-
-        // Load config once to derive the API URL and perform the initial
-        // token validation. The config is then moved into a live-refresh
-        // `TokenProvider` so the reconnect loop can re-read the latest
-        // session token from the profile store on every subsequent attempt
-        // (fix for TAURI-RUST-9C / #2892 — stale-token retry storm).
-        let config =
-            std::sync::Arc::new(crate::openhuman::config::rpc::load_config_with_timeout().await?);
-        let api_url = crate::api::config::effective_backend_api_url(&config.api_url);
-
-        // Perform an eager check so the RPC caller gets an immediate error
-        // if no session is stored — the provider will do the same check on
-        // every reconnect attempt inside `ws_loop`.
-        let initial_token = crate::api::jwt::get_session_token(&config)
-            .map_err(|e| format!("failed to read session token: {e}"))?
-            .ok_or("no session token stored — user must log in first")?;
-
-        log::info!(
-            "[socket:rpc] connect_with_session url={} token_len={}",
-            api_url,
-            initial_token.len()
-        );
-
-        // Build a live-refresh provider: on every reconnect attempt the loop
-        // calls this closure to obtain the most recently stored token, picking
-        // up any rotation or re-login that happened while sleeping.
-        let provider = super::token_provider::token_provider_from_config(config);
-
-        mgr.connect_with_provider(&api_url, provider).await?;
-
-        let state = mgr.get_state();
+        let state = super::ops::connect_with_session(mgr).await?;
         Ok(json!({ "status": format!("{:?}", state.status) }))
     })
 }

src/openhuman/socket/token_provider.rs

@@ -73,7 +73,7 @@ pub(super) fn static_token_provider(token: String) -> TokenProvider {
 /// This is the **live-refresh** path used by `handle_connect_with_session`:
 /// when the loop retries after a disconnect it will see any token that was
 /// refreshed or re-stored since the previous attempt.
-pub(super) fn token_provider_from_config(
+pub(crate) fn token_provider_from_config(
     config: Arc<crate::openhuman::config::Config>,
 ) -> TokenProvider {
     Arc::new(move || {

src/openhuman/socket/ws_loop.rs

@@ -163,6 +163,7 @@ pub(super) async fn ws_loop(
         // session-expired escalation below — not just explicit
         // `SocketManager::disconnect()` (CodeRabbit #4355).
         shared.ack_registry.cancel_all();
+        super::medulla::workflows::end_connection_generation();
 
         match outcome {
             ConnectionOutcome::Shutdown => {
@@ -720,6 +721,7 @@ fn handle_sio_packet(
         b'1' => {
             // Socket.IO DISCONNECT
             log::info!("[socket] SIO DISCONNECT from server");
+            super::medulla::workflows::end_connection_generation();
             *shared.status.write() = ConnectionStatus::Disconnected;
             *shared.socket_id.write() = None;
             emit_state_change(shared);