Add info on how to run demo script

Signed-off-by: Lee Smet <lee.smet@hotmail.com>

README.md

@@ -1,2 +1,28 @@
 # herocoordinator
 
+## Demo setup
+
+A python script is provided in the [scripts directory](./scripts/supervisor_flow_demo.py). This script
+generates some demo jobs to be run by [a supervisor](https://git.ourworld.tf/herocode/supervisor).
+Communication happens over [mycelium](https://github.com/threefoldtech/mycelium). To run the demo a
+supervisor must be running, which uses a mycelium instance to read and write messages. A __different__
+mycelium instance needs to run for the coordinator (the supervisor can run on a different host than
+the coordinator, so long as the 2 mycelium instances used can reach eachother).
+
+An example of a local setup:
+
+```bash
+# Run a redis docker
+docker run -it -d -p 6379:6379 --name redis redis
+# Spawn mycelium node 1 with default settings. This also creates a TUN interface though that is not
+# necessary for the messages
+mycelium
+# Spawn mycelium node 2, connect to the first node
+mycelium --key-file key.bin --peers tcp://127.0.0.1:9651 --disable-quic --disable-peer-discovery --api-addr 127.0.0.1:9989 --jsonrpc-addr 127.0.0.1:9990 --no-tun -t 8651
+# Start the supervisor
+supervisor --admin-secret admin123 --user-secret user123 --register-secret register123 --mycelium-url http://127.0.0.1:9990 --topic supervisor.rpc
+# Start the coordinator
+cargo run # (alternatively if a compiled binary is present that can be run)
+# Finally, invoke the demo script
+python3 scripts/supervisor_flow_demo.py
+```

src/clients/mod.rs

@@ -1,7 +1,9 @@
 pub mod mycelium_client;
 pub mod supervisor_client;
+pub mod supervisor_hub;
 pub mod types;
 
 pub use mycelium_client::{MyceliumClient, MyceliumClientError};
 pub use supervisor_client::{SupervisorClient, SupervisorClientError};
+pub use supervisor_hub::SupervisorHub;
 pub use types::Destination;

src/clients/supervisor_client.rs

@@ -1,20 +1,20 @@
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
 
 use base64::Engine;
 use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
 use serde_json::{Value, json};
 use thiserror::Error;
+use tokio::time::timeout;
 
-use crate::clients::{Destination, MyceliumClient, MyceliumClientError};
+use crate::clients::{Destination, MyceliumClient, MyceliumClientError, SupervisorHub};
 
 #[derive(Clone)]
 pub struct SupervisorClient {
-    mycelium: Arc<MyceliumClient>, // Delegated Mycelium transport
-    destination: Destination,      // ip or pk
-    topic: String,                 // e.g. "supervisor.rpc"
-    secret: Option<String>,        // optional, required by several supervisor methods
-    id_counter: Arc<AtomicU64>,    // JSON-RPC id generator (for inner supervisor requests)
+    hub: Arc<SupervisorHub>, // Global hub with background pop loop and shared id generator
+    destination: Destination, // ip or pk
+    secret: Option<String>,  // optional, required by several supervisor methods
 }
 
 #[derive(Debug, Error)]
@@ -46,24 +46,22 @@ impl From<MyceliumClientError> for SupervisorClientError {
 }
 
 impl SupervisorClient {
-    /// Preferred constructor: provide a shared Mycelium client.
-    pub fn new_with_client(
-        mycelium: Arc<MyceliumClient>,
+    /// Preferred constructor using a shared SupervisorHub (single global listener).
+    pub fn new_with_hub(
+        hub: Arc<SupervisorHub>,
         destination: Destination,
-        topic: impl Into<String>,
         secret: Option<String>,
     ) -> Self {
         Self {
-            mycelium,
+            hub,
             destination,
-            topic: topic.into(),
             secret,
-            id_counter: Arc::new(AtomicU64::new(1)),
         }
     }
 
-    /// Backward-compatible constructor that builds a Mycelium client from base_url.
-    /// base_url defaults to Mycelium spec "http://127.0.0.1:8990" if empty.
+    /// Backward-compatible constructor that builds a new Hub from base_url/topic.
+    /// NOTE: This spawns a background popMessage listener for the given topic.
+    /// Prefer `new_with_hub` so the process has a single global hub.
     pub fn new(
         base_url: impl Into<String>,
         destination: Destination,
@@ -78,8 +76,16 @@ impl SupervisorClient {
         Ok(Self::new_with_client(mycelium, destination, topic, secret))
     }
 
-    fn next_id(&self) -> u64 {
-        self.id_counter.fetch_add(1, Ordering::Relaxed)
+    /// Backward-compatible constructor that reuses an existing Mycelium client.
+    /// NOTE: This creates a new hub and its own background listener. Prefer `new_with_hub`.
+    pub fn new_with_client(
+        mycelium: Arc<MyceliumClient>,
+        destination: Destination,
+        topic: impl Into<String>,
+        secret: Option<String>,
+    ) -> Self {
+        let hub = SupervisorHub::new_with_client(mycelium, topic);
+        Self::new_with_hub(hub, destination, secret)
     }
 
     /// Internal helper used by tests to inspect dst JSON shape.
@@ -93,7 +99,7 @@ impl SupervisorClient {
     fn build_supervisor_payload(&self, method: &str, params: Value) -> Value {
         json!({
             "jsonrpc": "2.0",
-            "id": self.next_id(),
+            "id": self.hub.next_id(),
             "method": method,
             "params": params,
         })
@@ -128,50 +134,37 @@ impl SupervisorClient {
             .map(|s| s.to_string())
     }
 
-    /// Generic call: build supervisor JSON-RPC message, send via Mycelium pushMessage, return outbound message id (hex).
-    pub async fn call(&self, method: &str, params: Value) -> Result<String, SupervisorClientError> {
-        let inner = self.build_supervisor_payload(method, params);
-        let payload_b64 = Self::encode_payload(&inner)?;
-        let result = self
-            .mycelium
-            .push_message(
-                &self.destination,
-                &Self::encode_topic(self.topic.as_bytes()),
-                &payload_b64,
-                None,
-            )
-            .await?;
-
-        if let Some(id) = MyceliumClient::extract_message_id_from_result(&result) {
-            return Ok(id);
-        }
-        // Some servers might return the oneOf wrapped, handle len==1 array defensively (not in spec but resilient)
-        if let Some(arr) = result.as_array()
-            && arr.len() == 1
-            && let Some(id) = MyceliumClient::extract_message_id_from_result(&arr[0])
-        {
-            return Ok(id);
-        }
-        Err(SupervisorClientError::InvalidResponse(format!(
-            "result did not contain message id: {result}"
-        )))
+    fn need_secret(&self) -> Result<&str, SupervisorClientError> {
+        self.secret
+            .as_deref()
+            .ok_or(SupervisorClientError::MissingSecret)
     }
 
-    /// Variant of call that also returns the inner supervisor JSON-RPC id used in the payload.
-    /// This id is required to correlate asynchronous popMessage replies coming from Mycelium.
-    pub async fn call_with_ids(
+    // -----------------------------
+    // Core: request-reply call via Hub with default 10s timeout
+    // -----------------------------
+
+    /// Send a supervisor JSON-RPC request and await its reply via the Hub.
+    /// Returns (outbound_message_id, reply_envelope_json).
+    pub async fn call_with_reply_timeout(
         &self,
         method: &str,
         params: Value,
-    ) -> Result<(String, u64), SupervisorClientError> {
-        let inner_id = self.next_id();
+        timeout_secs: u64,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        let inner_id = self.hub.next_id();
+        // Register waiter before sending to avoid race
+        let rx = self.hub.register_waiter(inner_id).await;
+
         let inner = self.build_supervisor_payload_with_id(method, params, inner_id);
         let payload_b64 = Self::encode_payload(&inner)?;
+
         let result = self
-            .mycelium
+            .hub
+            .mycelium()
             .push_message(
                 &self.destination,
-                &Self::encode_topic(self.topic.as_bytes()),
+                &Self::encode_topic(self.hub.topic().as_bytes()),
                 &payload_b64,
                 None,
             )
@@ -185,29 +178,210 @@ impl SupervisorClient {
         {
             id
         } else {
+            // Clean pending entry to avoid leak
+            let _ = self.hub.remove_waiter(inner_id).await;
             return Err(SupervisorClientError::InvalidResponse(format!(
                 "result did not contain message id: {result}"
             )));
         };
 
-        Ok((out_id, inner_id))
+        let d = Duration::from_secs(timeout_secs);
+        match timeout(d, rx).await {
+            Ok(Ok(reply)) => Ok((out_id, reply)),
+            Ok(Err(_canceled)) => Err(SupervisorClientError::InvalidResponse(
+                "oneshot canceled before receiving reply".into(),
+            )),
+            Err(_elapsed) => {
+                // Cleanup on timeout
+                let _ = self.hub.remove_waiter(inner_id).await;
+                Err(SupervisorClientError::TransportTimeout)
+            }
+        }
     }
 
+    /// Send and await with default 10s timeout.
+    pub async fn call_with_reply(
+        &self,
+        method: &str,
+        params: Value,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply_timeout(method, params, 60).await
+    }
 
-    fn need_secret(&self) -> Result<&str, SupervisorClientError> {
-        self.secret
-            .as_deref()
-            .ok_or(SupervisorClientError::MissingSecret)
+    /// Back-compat: Send and await a reply but return only the outbound id (discard reply).
+    /// This keeps existing call sites working while the system migrates to reply-aware paths.
+    pub async fn call(&self, method: &str, params: Value) -> Result<String, SupervisorClientError> {
+        let (out_id, _reply) = self.call_with_reply(method, params).await?;
+        Ok(out_id)
     }
 
     // -----------------------------
-    // Typed wrappers for Supervisor API
-    // Asynchronous-only: returns outbound message id
+    // Typed wrappers for Supervisor API (await replies)
     // -----------------------------
 
     // Runners
+    pub async fn list_runners_wait(&self) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply("list_runners", json!([])).await
+    }
+
+    pub async fn register_runner_wait(
+        &self,
+        name: impl Into<String>,
+        queue: impl Into<String>,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        let secret = self.need_secret()?;
+        let params = json!([{
+            "secret": secret,
+            "name": name.into(),
+            "queue": queue.into()
+        }]);
+        self.call_with_reply("register_runner", params).await
+    }
+
+    pub async fn remove_runner_wait(
+        &self,
+        actor_id: impl Into<String>,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply("remove_runner", json!([actor_id.into()]))
+            .await
+    }
+
+    pub async fn start_runner_wait(
+        &self,
+        actor_id: impl Into<String>,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply("start_runner", json!([actor_id.into()]))
+            .await
+    }
+
+    pub async fn stop_runner_wait(
+        &self,
+        actor_id: impl Into<String>,
+        force: bool,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply("stop_runner", json!([actor_id.into(), force]))
+            .await
+    }
+
+    pub async fn get_runner_status_wait(
+        &self,
+        actor_id: impl Into<String>,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply("get_runner_status", json!([actor_id.into()]))
+            .await
+    }
+
+    pub async fn get_all_runner_status_wait(
+        &self,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply("get_all_runner_status", json!([]))
+            .await
+    }
+
+    pub async fn start_all_wait(&self) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply("start_all", json!([])).await
+    }
+
+    pub async fn stop_all_wait(
+        &self,
+        force: bool,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply("stop_all", json!([force])).await
+    }
+
+    pub async fn get_all_status_wait(&self) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply("get_all_status", json!([])).await
+    }
+
+    // Jobs (await)
+    pub async fn jobs_create_wait(
+        &self,
+        job: Value,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        let secret = self.need_secret()?;
+        let params = json!([{
+            "secret": secret,
+            "job": job
+        }]);
+        self.call_with_reply("jobs.create", params).await
+    }
+
+    pub async fn jobs_list_wait(&self) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply("jobs.list", json!([])).await
+    }
+
+    pub async fn job_run_wait(&self, job: Value) -> Result<(String, Value), SupervisorClientError> {
+        let secret = self.need_secret()?;
+        let params = json!([{
+            "secret": secret,
+            "job": job
+        }]);
+        self.call_with_reply("job.run", params).await
+    }
+
+    pub async fn job_start_wait(
+        &self,
+        job_id: impl Into<String>,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        let secret = self.need_secret()?;
+        let params = json!([{
+            "secret": secret,
+            "job_id": job_id.into()
+        }]);
+        self.call_with_reply("job.start", params).await
+    }
+
+    pub async fn job_status_wait(
+        &self,
+        job_id: impl Into<String>,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply("job.status", json!([job_id.into()]))
+            .await
+    }
+
+    pub async fn job_result_wait(
+        &self,
+        job_id: impl Into<String>,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply("job.result", json!([job_id.into()]))
+            .await
+    }
+
+    pub async fn job_stop_wait(
+        &self,
+        job_id: impl Into<String>,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        let secret = self.need_secret()?;
+        let params = json!([{
+            "secret": secret,
+            "job_id": job_id.into()
+        }]);
+        self.call_with_reply("job.stop", params).await
+    }
+
+    pub async fn job_delete_wait(
+        &self,
+        job_id: impl Into<String>,
+    ) -> Result<(String, Value), SupervisorClientError> {
+        let secret = self.need_secret()?;
+        let params = json!([{
+            "secret": secret,
+            "job_id": job_id.into()
+        }]);
+        self.call_with_reply("job.delete", params).await
+    }
+
+    pub async fn rpc_discover_wait(&self) -> Result<(String, Value), SupervisorClientError> {
+        self.call_with_reply("rpc.discover", json!([])).await
+    }
+
+    // -----------------------------
+    // Backward-compatible variants returning only outbound id (discarding reply)
+    // -----------------------------
+
     pub async fn list_runners(&self) -> Result<String, SupervisorClientError> {
-        self.call("list_runners", json!([])).await
+        let (id, _) = self.list_runners_wait().await?;
+        Ok(id)
     }
 
     pub async fn register_runner(
@@ -215,27 +389,24 @@ impl SupervisorClient {
         name: impl Into<String>,
         queue: impl Into<String>,
     ) -> Result<String, SupervisorClientError> {
-        let secret = self.need_secret()?;
-        let params = json!([{
-            "secret": secret,
-            "name": name.into(),
-            "queue": queue.into()
-        }]);
-        self.call("register_runner", params).await
+        let (id, _) = self.register_runner_wait(name, queue).await?;
+        Ok(id)
     }
 
     pub async fn remove_runner(
         &self,
         actor_id: impl Into<String>,
     ) -> Result<String, SupervisorClientError> {
-        self.call("remove_runner", json!([actor_id.into()])).await
+        let (id, _) = self.remove_runner_wait(actor_id).await?;
+        Ok(id)
     }
 
     pub async fn start_runner(
         &self,
         actor_id: impl Into<String>,
     ) -> Result<String, SupervisorClientError> {
-        self.call("start_runner", json!([actor_id.into()])).await
+        let (id, _) = self.start_runner_wait(actor_id).await?;
+        Ok(id)
     }
 
     pub async fn stop_runner(
@@ -243,141 +414,96 @@ impl SupervisorClient {
         actor_id: impl Into<String>,
         force: bool,
     ) -> Result<String, SupervisorClientError> {
-        self.call("stop_runner", json!([actor_id.into(), force]))
-            .await
+        let (id, _) = self.stop_runner_wait(actor_id, force).await?;
+        Ok(id)
     }
 
     pub async fn get_runner_status(
         &self,
         actor_id: impl Into<String>,
     ) -> Result<String, SupervisorClientError> {
-        self.call("get_runner_status", json!([actor_id.into()]))
-            .await
+        let (id, _) = self.get_runner_status_wait(actor_id).await?;
+        Ok(id)
     }
 
     pub async fn get_all_runner_status(&self) -> Result<String, SupervisorClientError> {
-        self.call("get_all_runner_status", json!([])).await
+        let (id, _) = self.get_all_runner_status_wait().await?;
+        Ok(id)
     }
 
     pub async fn start_all(&self) -> Result<String, SupervisorClientError> {
-        self.call("start_all", json!([])).await
+        let (id, _) = self.start_all_wait().await?;
+        Ok(id)
     }
 
     pub async fn stop_all(&self, force: bool) -> Result<String, SupervisorClientError> {
-        self.call("stop_all", json!([force])).await
+        let (id, _) = self.stop_all_wait(force).await?;
+        Ok(id)
     }
 
     pub async fn get_all_status(&self) -> Result<String, SupervisorClientError> {
-        self.call("get_all_status", json!([])).await
+        let (id, _) = self.get_all_status_wait().await?;
+        Ok(id)
     }
 
-    // Jobs
     pub async fn jobs_create(&self, job: Value) -> Result<String, SupervisorClientError> {
-        let secret = self.need_secret()?;
-        let params = json!([{
-            "secret": secret,
-            "job": job
-        }]);
-        self.call("jobs.create", params).await
+        let (id, _) = self.jobs_create_wait(job).await?;
+        Ok(id)
     }
 
     pub async fn jobs_list(&self) -> Result<String, SupervisorClientError> {
-        self.call("jobs.list", json!([])).await
+        let (id, _) = self.jobs_list_wait().await?;
+        Ok(id)
     }
 
     pub async fn job_run(&self, job: Value) -> Result<String, SupervisorClientError> {
-        let secret = self.need_secret()?;
-        let params = json!([{
-            "secret": secret,
-            "job": job
-        }]);
-        self.call("job.run", params).await
-    }
-
-    /// Typed wrapper returning both outbound Mycelium id and inner supervisor JSON-RPC id.
-    pub async fn job_run_with_ids(
-        &self,
-        job: Value,
-    ) -> Result<(String, u64), SupervisorClientError> {
-        let secret = self.need_secret()?;
-        let params = json!([{
-            "secret": secret,
-            "job": job
-        }]);
-        self.call_with_ids("job.run", params).await
+        let (id, _) = self.job_run_wait(job).await?;
+        Ok(id)
     }
 
     pub async fn job_start(
         &self,
         job_id: impl Into<String>,
     ) -> Result<String, SupervisorClientError> {
-        let secret = self.need_secret()?;
-        let params = json!([{
-            "secret": secret,
-            "job_id": job_id.into()
-        }]);
-        self.call("job.start", params).await
+        let (id, _) = self.job_start_wait(job_id).await?;
+        Ok(id)
     }
 
     pub async fn job_status(
         &self,
         job_id: impl Into<String>,
     ) -> Result<String, SupervisorClientError> {
-        self.call("job.status", json!([job_id.into()])).await
+        let (id, _) = self.job_status_wait(job_id).await?;
+        Ok(id)
     }
 
-    /// Asynchronous job.status returning outbound and inner IDs for correlation
-    pub async fn job_status_with_ids(
-        &self,
-        job_id: impl Into<String>,
-    ) -> Result<(String, u64), SupervisorClientError> {
-        self.call_with_ids("job.status", json!([job_id.into()])).await
-    }
-
-
     pub async fn job_result(
         &self,
         job_id: impl Into<String>,
     ) -> Result<String, SupervisorClientError> {
-        self.call("job.result", json!([job_id.into()])).await
+        let (id, _) = self.job_result_wait(job_id).await?;
+        Ok(id)
     }
 
-    /// Asynchronous job.result returning outbound and inner IDs for correlation
-    pub async fn job_result_with_ids(
-        &self,
-        job_id: impl Into<String>,
-    ) -> Result<(String, u64), SupervisorClientError> {
-        self.call_with_ids("job.result", json!([job_id.into()])).await
-    }
-
-
     pub async fn job_stop(
         &self,
         job_id: impl Into<String>,
     ) -> Result<String, SupervisorClientError> {
-        let secret = self.need_secret()?;
-        let params = json!([{
-            "secret": secret,
-            "job_id": job_id.into()
-        }]);
-        self.call("job.stop", params).await
+        let (id, _) = self.job_stop_wait(job_id).await?;
+        Ok(id)
     }
 
     pub async fn job_delete(
         &self,
         job_id: impl Into<String>,
     ) -> Result<String, SupervisorClientError> {
-        let secret = self.need_secret()?;
-        let params = json!([{
-            "secret": secret,
-            "job_id": job_id.into()
-        }]);
-        self.call("job.delete", params).await
+        let (id, _) = self.job_delete_wait(job_id).await?;
+        Ok(id)
     }
 
-    // Discovery
     pub async fn rpc_discover(&self) -> Result<String, SupervisorClientError> {
-        self.call("rpc.discover", json!([])).await
+        let (id, _) = self.rpc_discover_wait().await?;
+        Ok(id)
     }
 }
 
@@ -390,27 +516,27 @@ mod tests {
     use std::net::IpAddr;
 
     fn mk_client() -> SupervisorClient {
-        // Uses the legacy constructor but will not issue real network calls in these tests.
-        SupervisorClient::new(
-            "http://127.0.0.1:8990",
+        // Build a hub but it won't issue real network calls in these serializer-only tests.
+        let mycelium = Arc::new(MyceliumClient::new("http://127.0.0.1:8990").unwrap());
+        let hub = SupervisorHub::new_with_client(mycelium, "supervisor.rpc");
+        SupervisorClient::new_with_hub(
+            hub,
             Destination::Pk(
                 "bb39b4a3a4efd70f3e05e37887677e02efbda14681d0acd3882bc0f754792c32".to_string(),
             ),
-            "supervisor.rpc",
             Some("secret".to_string()),
         )
-        .unwrap()
     }
 
     #[test]
     fn builds_dst_ip_and_pk() {
-        let c_ip = SupervisorClient::new(
-            "http://127.0.0.1:8990",
+        let mycelium = Arc::new(MyceliumClient::new("http://127.0.0.1:8990").unwrap());
+        let hub_ip = SupervisorHub::new_with_client(mycelium.clone(), "supervisor.rpc");
+        let c_ip = SupervisorClient::new_with_hub(
+            hub_ip,
             Destination::Ip("2001:db8::1".parse().unwrap()),
-            "supervisor.rpc",
             None,
-        )
-        .unwrap();
+        );
         let v_ip = c_ip.build_dst();
         assert_eq!(v_ip.get("ip").unwrap().as_str().unwrap(), "2001:db8::1");
 

src/clients/supervisor_hub.rs

@@ -0,0 +1,143 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
+use serde_json::Value;
+use tokio::sync::{Mutex, oneshot};
+
+use crate::clients::mycelium_client::MyceliumClient;
+
+/// Global hub that:
+/// - Owns a single MyceliumClient
+/// - Spawns a background popMessage loop filtered by topic
+/// - Correlates supervisor JSON-RPC replies by inner id to waiting callers via oneshot channels
+#[derive(Clone)]
+pub struct SupervisorHub {
+    mycelium: Arc<MyceliumClient>,
+    topic: String,
+    pending: Arc<Mutex<HashMap<u64, oneshot::Sender<Value>>>>,
+    id_counter: Arc<AtomicU64>,
+}
+
+impl SupervisorHub {
+    /// Create a new hub and start the background popMessage task.
+    /// - base_url: Mycelium JSON-RPC endpoint, e.g. "http://127.0.0.1:8990"
+    /// - topic: plain-text topic (e.g., "supervisor.rpc")
+    pub fn new(
+        base_url: impl Into<String>,
+        topic: impl Into<String>,
+    ) -> Result<Arc<Self>, crate::clients::MyceliumClientError> {
+        let myc = Arc::new(MyceliumClient::new(base_url)?);
+        Ok(Self::new_with_client(myc, topic))
+    }
+
+    /// Variant that reuses an existing Mycelium client.
+    pub fn new_with_client(mycelium: Arc<MyceliumClient>, topic: impl Into<String>) -> Arc<Self> {
+        let hub = Arc::new(Self {
+            mycelium,
+            topic: topic.into(),
+            pending: Arc::new(Mutex::new(HashMap::new())),
+            id_counter: Arc::new(AtomicU64::new(1)),
+        });
+        Self::spawn_pop_loop(hub.clone());
+        hub
+    }
+
+    fn spawn_pop_loop(hub: Arc<Self>) {
+        tokio::spawn(async move {
+            loop {
+                match hub.mycelium.pop_message(Some(false), Some(20), None).await {
+                    Ok(Some(inb)) => {
+                        // Extract and decode payload
+                        let Some(payload_b64) = inb.get("payload").and_then(|v| v.as_str()) else {
+                            // Not a payload-bearing message; ignore
+                            continue;
+                        };
+                        let Ok(raw) = BASE64_STANDARD.decode(payload_b64.as_bytes()) else {
+                            tracing::warn!(target: "supervisor_hub", "Failed to decode inbound payload base64");
+                            continue;
+                        };
+                        let Ok(rpc): Result<Value, _> = serde_json::from_slice(&raw) else {
+                            tracing::warn!(target: "supervisor_hub", "Failed to parse inbound payload JSON");
+                            continue;
+                        };
+
+                        // Extract inner JSON-RPC id
+                        let inner_id_u64 = match rpc.get("id") {
+                            Some(Value::Number(n)) => n.as_u64(),
+                            Some(Value::String(s)) => s.parse::<u64>().ok(),
+                            _ => None,
+                        };
+
+                        if let Some(inner_id) = inner_id_u64 {
+                            // Try to deliver to a pending waiter
+                            let sender_opt = {
+                                let mut guard = hub.pending.lock().await;
+                                guard.remove(&inner_id)
+                            };
+                            if let Some(tx) = sender_opt {
+                                let _ = tx.send(rpc);
+                            } else {
+                                tracing::warn!(
+                                    target: "supervisor_hub",
+                                    inner_id,
+                                    payload = %String::from_utf8_lossy(&raw),
+                                    "Unmatched supervisor reply; no waiter registered"
+                                );
+                            }
+                        } else {
+                            tracing::warn!(target: "supervisor_hub", "Inbound supervisor reply missing id; dropping");
+                        }
+                    }
+                    Ok(None) => {
+                        // No message; continue polling
+                        continue;
+                    }
+                    Err(e) => {
+                        tracing::warn!(target: "supervisor_hub", error = %e, "popMessage error; backing off");
+                        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+                    }
+                }
+            }
+        });
+    }
+
+    /// Allocate a new inner supervisor JSON-RPC id.
+    pub fn next_id(&self) -> u64 {
+        self.id_counter.fetch_add(1, Ordering::Relaxed)
+    }
+
+    /// Register a oneshot sender for the given inner id and return the receiver side.
+    pub async fn register_waiter(&self, inner_id: u64) -> oneshot::Receiver<Value> {
+        let (tx, rx) = oneshot::channel();
+        let mut guard = self.pending.lock().await;
+        guard.insert(inner_id, tx);
+        rx
+    }
+
+    /// Remove a pending waiter for a given id (used to cleanup on timeout).
+    pub async fn remove_waiter(&self, inner_id: u64) -> Option<oneshot::Sender<Value>> {
+        let mut guard = self.pending.lock().await;
+        guard.remove(&inner_id)
+    }
+
+    /// Access to underlying Mycelium client (for pushMessage).
+    pub fn mycelium(&self) -> Arc<MyceliumClient> {
+        self.mycelium.clone()
+    }
+
+    /// Access configured topic.
+    pub fn topic(&self) -> &str {
+        &self.topic
+    }
+}
+
+impl std::fmt::Debug for SupervisorHub {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("SupervisorHub")
+            .field("topic", &self.topic)
+            .finish()
+    }
+}

src/dag.rs

@@ -3,7 +3,7 @@ use std::collections::{HashMap, HashSet, VecDeque};
 use std::fmt;
 
 use crate::{
-    models::{Flow, Job, ScriptType},
+    models::{Flow, Job, JobStatus, ScriptType},
     storage::RedisDriver,
 };
 
@@ -212,6 +212,41 @@ pub async fn build_flow_dag(
     edges.sort_unstable();
     reverse_edges.sort_unstable();
 
+    // Populate runtime execution state from persisted Job.status()
+    let mut started_set: HashSet<u32> = HashSet::new();
+    let mut completed_set: HashSet<u32> = HashSet::new();
+    let mut error_ids: Vec<u32> = Vec::new();
+
+    for (&jid, job) in &jobs {
+        match job.status() {
+            JobStatus::Finished => {
+                completed_set.insert(jid);
+            }
+            JobStatus::Started => {
+                started_set.insert(jid);
+            }
+            JobStatus::Dispatched => {
+                // Consider Dispatched as "in-flight" for DAG runtime started set,
+                // so queued/running work is visible in periodic snapshots.
+                started_set.insert(jid);
+            }
+            JobStatus::Error => {
+                error_ids.push(jid);
+            }
+            JobStatus::WaitingForPrerequisites => {
+                // Neither started nor completed
+            }
+        }
+    }
+
+    // Choose a deterministic failed job if any errors exist (smallest job id)
+    let failed_job = if error_ids.is_empty() {
+        None
+    } else {
+        error_ids.sort_unstable();
+        Some(error_ids[0])
+    };
+
     let dag = FlowDag {
         flow_id,
         caller_id,
@@ -222,9 +257,9 @@ pub async fn build_flow_dag(
         roots,
         leaves,
         levels,
-        started: HashSet::new(),
-        completed: HashSet::new(),
-        failed_job: None,
+        started: started_set,
+        completed: completed_set,
+        failed_job,
     };
 
     Ok(dag)

src/main.rs

@@ -99,21 +99,24 @@ async fn main() {
     // Shared application state
     let state = Arc::new(herocoordinator::rpc::AppState::new(service));
 
-    // Start router workers (auto-discovered contexts) and a single global inbound listener
+    // Start router workers (auto-discovered contexts) using a single global SupervisorHub (no separate inbound listener)
     {
         let base_url = format!("http://{}:{}", cli.mycelium_ip, cli.mycelium_port);
+        let hub = herocoordinator::clients::SupervisorHub::new(
+            base_url.clone(),
+            "supervisor.rpc".to_string(),
+        )
+        .expect("Failed to initialize SupervisorHub");
         let cfg = herocoordinator::router::RouterConfig {
             context_ids: Vec::new(), // ignored by start_router_auto
             concurrency: 32,
             base_url,
             topic: "supervisor.rpc".to_string(),
+            sup_hub: hub.clone(),
             transport_poll_interval_secs: 2,
             transport_poll_timeout_secs: 300,
         };
-        // Global inbound listener for supervisor replies via Mycelium popMessage
-        let _inbound_handle =
-            herocoordinator::router::start_inbound_listener(service_for_router.clone(), cfg.clone());
-        // Per-context outbound delivery loops
+        // Per-context outbound delivery loops (replies handled by SupervisorHub)
         let _auto_handle = herocoordinator::router::start_router_auto(service_for_router, cfg);
     }
 

src/service.rs

@@ -672,10 +672,16 @@ impl AppService {
         let allowed = match current {
             JobStatus::Dispatched => matches!(
                 new_status,
-                JobStatus::WaitingForPrerequisites | JobStatus::Started | JobStatus::Error
+                JobStatus::WaitingForPrerequisites
+                    | JobStatus::Started
+                    | JobStatus::Finished
+                    | JobStatus::Error
             ),
             JobStatus::WaitingForPrerequisites => {
-                matches!(new_status, JobStatus::Started | JobStatus::Error)
+                matches!(
+                    new_status,
+                    JobStatus::Started | JobStatus::Finished | JobStatus::Error
+                )
             }
             JobStatus::Started => matches!(new_status, JobStatus::Finished | JobStatus::Error),
             JobStatus::Finished | JobStatus::Error => false,
@@ -714,10 +720,16 @@ impl AppService {
         let allowed = match current {
             JobStatus::Dispatched => matches!(
                 new_status,
-                JobStatus::WaitingForPrerequisites | JobStatus::Started | JobStatus::Error
+                JobStatus::WaitingForPrerequisites
+                    | JobStatus::Started
+                    | JobStatus::Finished
+                    | JobStatus::Error
             ),
             JobStatus::WaitingForPrerequisites => {
-                matches!(new_status, JobStatus::Started | JobStatus::Error)
+                matches!(
+                    new_status,
+                    JobStatus::Started | JobStatus::Finished | JobStatus::Error
+                )
             }
             JobStatus::Started => matches!(new_status, JobStatus::Finished | JobStatus::Error),
             JobStatus::Finished | JobStatus::Error => false,
@@ -1182,10 +1194,7 @@ impl AppService {
         &self,
         inner_id: u64,
     ) -> Result<Option<(u32, u32, u32, u32)>, BoxError> {
-        self.redis
-            .supcorr_get(inner_id)
-            .await
-            .map_err(Into::into)
+        self.redis.supcorr_get(inner_id).await.map_err(Into::into)
     }
 
     /// Correlation map: delete mapping by inner supervisor JSON-RPC id.

src/storage/redis.rs

@@ -789,10 +789,7 @@ impl RedisDriver {
         Ok(())
     }
 
-    pub async fn supcorr_get(
-        &self,
-        inner_id: u64,
-    ) -> Result<Option<(u32, u32, u32, u32)>> {
+    pub async fn supcorr_get(&self, inner_id: u64) -> Result<Option<(u32, u32, u32, u32)>> {
         let mut cm = self.manager_for_db(0).await?;
         let key = format!("supcorr:{}", inner_id);
         let res: Option<String> = redis::cmd("GET")