Check job status in redis db as well before sending rpc call

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

scripts/jsonrpc_demo.py

@@ -0,0 +1,376 @@
+#!/usr/bin/env python3
+"""
+Demo script for HeroCoordinator JSON-RPC API.
+- Creates an actor
+- Verifies by loading the actor
+- Creates a context with the actor as admin/reader/executor
+- Creates three jobs with dependencies
+- Creates a flow referencing those jobs
+- Fetches and prints the flow DAG
+
+Usage:
+  COORDINATOR_URL=http://127.0.0.1:9652 python3 scripts/jsonrpc_demo.py
+Defaults to http://127.0.0.1:9652 if COORDINATOR_URL is not set.
+"""
+import os
+import json
+import sys
+from urllib import request, error
+from typing import Any, Dict, List, Tuple
+
+JSONRPC_VERSION = "2.0"
+
+class JsonRpcClient:
+    def __init__(self, url: str):
+        self.url = url.rstrip("/")
+        self._id = 0
+
+    def call(self, method: str, params: Dict[str, Any]) -> Any:
+        self._id += 1
+        payload = {
+            "jsonrpc": JSONRPC_VERSION,
+            "id": self._id,
+            "method": method,
+            "params": params,
+        }
+        data = json.dumps(payload).encode("utf-8")
+        req = request.Request(self.url, data=data, headers={"Content-Type": "application/json"})
+        try:
+            with request.urlopen(req) as resp:
+                body = resp.read()
+        except error.HTTPError as e:
+            try:
+                details = e.read().decode("utf-8", "ignore")
+            except Exception:
+                details = ""
+            raise RuntimeError(f"HTTP error {e.code}: {details}") from e
+        except error.URLError as e:
+            raise RuntimeError(f"URL error: {e.reason}") from e
+
+        try:
+            obj = json.loads(body.decode("utf-8"))
+        except Exception as e:
+            raise RuntimeError(f"Invalid JSON response: {body!r}") from e
+
+        # JSON-RPC single response expected
+        if isinstance(obj, list):
+            raise RuntimeError("Batch responses are not supported in this demo")
+
+        if obj.get("error"):
+            raise RuntimeError(f"RPC error: {json.dumps(obj['error'])}")
+
+        return obj.get("result")
+
+def print_header(title: str):
+    print("\n" + "=" * 80)
+    print(title)
+    print("=" * 80)
+
+def pretty_print(obj: Any):
+    print(json.dumps(obj, indent=2, sort_keys=True))
+
+def summarize_dag(dag: Dict[str, Any]):
+    print_header("Flow DAG Summary")
+    flow_id = dag.get("flow_id")
+    caller_id = dag.get("caller_id")
+    context_id = dag.get("context_id")
+    print(f"flow_id={flow_id} caller_id={caller_id} context_id={context_id}")
+    edges: List[Tuple[int, int]] = dag.get("edges", [])
+    roots: List[int] = dag.get("roots", [])
+    leaves: List[int] = dag.get("leaves", [])
+    levels: List[List[int]] = dag.get("levels", [])
+    nodes: Dict[str, Any] = dag.get("nodes", {})
+
+    print("Edges:")
+    for a, b in edges:
+        print(f"  {a} -> {b}")
+
+    print(f"Roots: {roots}")
+    print(f"Leaves: {leaves}")
+    print("Levels:")
+    for i, lvl in enumerate(levels):
+        print(f"  L{i}: {lvl}")
+
+    # Show nodes and their dependencies (from JobSummary)
+    print("Nodes:")
+    for k, v in nodes.items():
+        depends = v.get("depends", [])
+        prerequisites = v.get("prerequisites", [])
+        stype = v.get("script_type")
+        print(f"  Job {k}: depends={depends} prerequisites={prerequisites} script_type={stype}")
+
+def assert_edges(edges: List[Tuple[int, int]], required: List[Tuple[int, int]]):
+    edge_set = {(int(a), int(b)) for a, b in edges}
+    missing = [e for e in required if e not in edge_set]
+    if missing:
+        raise AssertionError(f"Missing expected edges in DAG: {missing}; got={sorted(edge_set)}")
+
+def main():
+    url = os.getenv("COORDINATOR_URL", "http://127.0.0.1:9652")
+    client = JsonRpcClient(url)
+
+    # Deterministic demo IDs; change if collisions happen
+    actor_id = 1001
+    context_id = 1  # Redis DB indices are 0-15; keep <= 15
+    job_a = 3001
+    job_b = 3002
+    job_c = 3003
+    job_d = 3004
+    job_e = 3005
+    job_f = 3006
+    job_g = 3007
+    job_h = 3008
+    job_i = 3009
+    flow_id = 4001
+
+runner_id = 2001
+    print_header("actor.create")
+    actor = client.call("actor.create", {
+        "actor": {
+            "id": actor_id,
+            "pubkey": "demo-pubkey",
+            "address": ["127.0.0.1"]
+        }
+    })
+    pretty_print(actor)
+
+    print_header("actor.load")
+    actor_loaded = client.call("actor.load", {"id": actor_id})
+    pretty_print(actor_loaded)
+
+    print_header("context.create")
+    context = client.call("context.create", {
+        "context": {
+            "id": context_id,
+            "admins": [actor_id],
+            "readers": [actor_id],
+            "executors": [actor_id]
+        }
+    })
+    pretty_print(context)
+print_header("runner.create")
+    runner = client.call("runner.create", {
+        "context_id": context_id,
+        "runner": {
+            "id": runner_id,
+            "pubkey": "",  # leave empty to route by IP
+            "address": "127.0.0.1",
+            "topic": f"runner{runner_id}",
+            "script_type": "Python",
+            "local": True,
+            "secret": "demo-secret"
+        }
+    })
+    pretty_print(runner)
+
+    print_header("job.create - A (root)")
+    jobA = client.call("job.create", {
+        "context_id": context_id,
+        "job": {
+            "id": job_a,
+            "caller_id": actor_id,
+            "context_id": context_id,
+            "script": "print('A')",
+            "script_type": "Python",
+            "timeout": 30,
+            "retries": 0,
+            "env_vars": {},
+            "prerequisites": [],
+            "depends": []
+        }
+    })
+    pretty_print(jobA)
+
+    print_header("job.create - B (root)")
+    jobB = client.call("job.create", {
+        "context_id": context_id,
+        "job": {
+            "id": job_b,
+            "caller_id": actor_id,
+            "context_id": context_id,
+            "script": "print('B')",
+            "script_type": "Python",
+            "timeout": 30,
+            "retries": 0,
+            "env_vars": {},
+            "prerequisites": [],
+            "depends": []
+        }
+    })
+    pretty_print(jobB)
+
+    print_header("job.create - C (depends on A and B)")
+    jobC = client.call("job.create", {
+        "context_id": context_id,
+        "job": {
+            "id": job_c,
+            "caller_id": actor_id,
+            "context_id": context_id,
+            "script": "print('C')",
+            "script_type": "Python",
+            "timeout": 30,
+            "retries": 0,
+            "env_vars": {},
+            "prerequisites": [],
+            "depends": [job_a, job_b]
+        }
+    })
+    pretty_print(jobC)
+
+    print_header("job.create - D (depends on A)")
+    jobD = client.call("job.create", {
+        "context_id": context_id,
+        "job": {
+            "id": job_d,
+            "caller_id": actor_id,
+            "context_id": context_id,
+            "script": "print('D')",
+            "script_type": "Python",
+            "timeout": 30,
+            "retries": 0,
+            "env_vars": {},
+            "prerequisites": [],
+            "depends": [job_a]
+        }
+    })
+    pretty_print(jobD)
+
+    print_header("job.create - E (depends on B)")
+    jobE = client.call("job.create", {
+        "context_id": context_id,
+        "job": {
+            "id": job_e,
+            "caller_id": actor_id,
+            "context_id": context_id,
+            "script": "print('E')",
+            "script_type": "Python",
+            "timeout": 30,
+            "retries": 0,
+            "env_vars": {},
+            "prerequisites": [],
+            "depends": [job_b]
+        }
+    })
+    pretty_print(jobE)
+
+    print_header("job.create - F (depends on C and D)")
+    jobF = client.call("job.create", {
+        "context_id": context_id,
+        "job": {
+            "id": job_f,
+            "caller_id": actor_id,
+            "context_id": context_id,
+            "script": "print('F')",
+            "script_type": "Python",
+            "timeout": 30,
+            "retries": 0,
+            "env_vars": {},
+            "prerequisites": [],
+            "depends": [job_c, job_d]
+        }
+    })
+    pretty_print(jobF)
+
+    print_header("job.create - G (depends on C and E)")
+    jobG = client.call("job.create", {
+        "context_id": context_id,
+        "job": {
+            "id": job_g,
+            "caller_id": actor_id,
+            "context_id": context_id,
+            "script": "print('G')",
+            "script_type": "Python",
+            "timeout": 30,
+            "retries": 0,
+            "env_vars": {},
+            "prerequisites": [],
+            "depends": [job_c, job_e]
+        }
+    })
+    pretty_print(jobG)
+
+    print_header("job.create - H (leaf; depends on F and G)")
+    jobH = client.call("job.create", {
+        "context_id": context_id,
+        "job": {
+            "id": job_h,
+            "caller_id": actor_id,
+            "context_id": context_id,
+            "script": "print('H')",
+            "script_type": "Python",
+            "timeout": 30,
+            "retries": 0,
+            "env_vars": {},
+            "prerequisites": [],
+            "depends": [job_f, job_g]
+        }
+    })
+    pretty_print(jobH)
+
+    print_header("job.create - I (leaf; depends on F and G)")
+    jobI = client.call("job.create", {
+        "context_id": context_id,
+        "job": {
+            "id": job_i,
+            "caller_id": actor_id,
+            "context_id": context_id,
+            "script": "print('I')",
+            "script_type": "Python",
+            "timeout": 30,
+            "retries": 0,
+            "env_vars": {},
+            "prerequisites": [],
+            "depends": [job_f, job_g]
+        }
+    })
+    pretty_print(jobI)
+
+    print_header("flow.create")
+    flow = client.call("flow.create", {
+        "context_id": context_id,
+        "flow": {
+            "id": flow_id,
+            "caller_id": actor_id,
+            "context_id": context_id,
+            "jobs": [job_a, job_b, job_c, job_d, job_e, job_f, job_g, job_h, job_i],
+            "env_vars": {}
+        }
+    })
+    pretty_print(flow)
+
+    print_header("flow.dag")
+    dag = client.call("flow.dag", {"context_id": context_id, "id": flow_id})
+    summarize_dag(dag)
+
+    # Validate roots and leaves
+    got_roots = list(map(int, dag.get("roots", [])))
+    if got_roots != sorted([job_a, job_b]):
+        print("WARNING: Unexpected roots:", got_roots, file=sys.stderr)
+
+    got_leaves = {int(x) for x in dag.get("leaves", [])}
+    expected_leaves = {job_h, job_i}
+    if got_leaves != expected_leaves:
+        print("WARNING: Unexpected leaves:", got_leaves, "expected:", expected_leaves, file=sys.stderr)
+
+    # Check edges reflect the expanded DAG
+    expected_edges = [
+        (job_a, job_c), (job_b, job_c),
+        (job_a, job_d), (job_b, job_e),
+        (job_c, job_f), (job_d, job_f),
+        (job_c, job_g), (job_e, job_g),
+        (job_f, job_h), (job_g, job_h),
+        (job_f, job_i), (job_g, job_i),
+    ]
+    try:
+        assert_edges(dag.get("edges", []), expected_edges)
+        print("DAG edges contain expected dependencies:", expected_edges)
+    except AssertionError as e:
+        print("WARNING:", e, file=sys.stderr)
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as e:
+        print_header("Error")
+        print(str(e))
+        sys.exit(1)

scripts/supervisor_flow_demo.py

@@ -0,0 +1,502 @@
+#!/usr/bin/env python3
+"""
+Supervisor flow demo for HeroCoordinator.
+
+This script:
+- Optionally pre-registers and starts a Python runner on the target Supervisor over Mycelium using an admin secret (--admin-secret). If the flag is not set, this step is skipped.
+- Creates an actor
+- Creates a context granting the actor admin/reader/executor privileges
+- Registers a Runner in the context targeting a Supervisor reachable via Mycelium (by public key or IP)
+- Creates simple Python jobs (text jobs) with a small dependency chain
+- Creates a flow referencing those jobs
+- Starts the flow and polls until it finishes (or errors)
+
+Transport: JSON-RPC over HTTP to the Coordinator (default COORDINATOR_URL=http://127.0.0.1:9652).
+
+Example usage:
+  COORDINATOR_URL=http://127.0.0.1:9652 python3 scripts/supervisor_flow_demo.py --dst-ip 2001:db8::1 [--secret your-secret]
+  COORDINATOR_URL=http://127.0.0.1:9652 python3 scripts/supervisor_flow_demo.py --dst-pk bb39b4a3a4efd70f3e05e37887677e02efbda14681d0acd3882bc0f754792c32 [--secret your-secret]
+
+Notes:
+- Exactly one of --dst-ip or --dst-pk must be provided.
+- Runner.topic defaults to "supervisor.rpc" (see main.rs).
+- The router auto-discovers contexts and will deliver job.run messages to the supervisor.
+- Mycelium URL is read from MYCELIUM_URL (default http://127.0.0.1:8990).
+- supervisor.register_runner uses static name="python" and queue="python".
+"""
+
+import argparse
+import json
+import base64
+import os
+import sys
+import time
+from typing import Any, Dict, List, Optional, Tuple
+from urllib import request, error
+
+JSONRPC_VERSION = "2.0"
+
+
+def env_url() -> str:
+    return os.getenv("COORDINATOR_URL", "http://127.0.0.1:9652").rstrip("/")
+
+def env_mycelium_url() -> str:
+    return os.getenv("MYCELIUM_URL", "http://127.0.0.1:8990").rstrip("/")
+
+
+class JsonRpcClient:
+    def __init__(self, url: str):
+        self.url = url
+        self._id = 0
+
+    def call(self, method: str, params: Dict[str, Any]) -> Any:
+        self._id += 1
+        payload = {
+            "jsonrpc": JSONRPC_VERSION,
+            "id": self._id,
+            "method": method,
+            "params": params,
+        }
+        data = json.dumps(payload).encode("utf-8")
+        req = request.Request(self.url, data=data, headers={"Content-Type": "application/json"})
+        try:
+            with request.urlopen(req) as resp:
+                body = resp.read()
+        except error.HTTPError as e:
+            try:
+                details = e.read().decode("utf-8", "ignore")
+            except Exception:
+                details = ""
+            raise RuntimeError(f"HTTP error {e.code}: {details}") from e
+        except error.URLError as e:
+            raise RuntimeError(f"URL error: {e.reason}") from e
+
+        try:
+            obj = json.loads(body.decode("utf-8"))
+        except Exception as e:
+            raise RuntimeError(f"Invalid JSON response: {body!r}") from e
+
+        if isinstance(obj, list):
+            raise RuntimeError("Batch responses are not supported")
+
+        if obj.get("error"):
+            raise RuntimeError(f"RPC error: {json.dumps(obj['error'])}")
+
+        return obj.get("result")
+
+
+def print_header(title: str):
+    print("\n" + "=" * 80)
+    print(title)
+    print("=" * 80)
+
+
+def pretty(obj: Any):
+    print(json.dumps(obj, indent=2, sort_keys=True))
+
+def mycelium_register_runner(
+    myc: "JsonRpcClient",
+    dst_pk: Optional[str],
+    dst_ip: Optional[str],
+    topic: str,
+    admin_secret: str,
+    name: str = "python",
+    queue: str = "python",
+    timeout: int = 15,
+) -> Any:
+    """
+    Send supervisor.register_runner over Mycelium using pushMessage and wait for the reply.
+    - myc: JsonRpcClient for the Mycelium API (MYCELIUM_URL)
+    - dst_pk/dst_ip: destination on the overlay; one of them must be provided
+    - topic: message topic (defaults to supervisor.rpc from args)
+    - admin_secret: supervisor admin secret to authorize the registration
+    - name/queue: static identifiers for the python runner on the supervisor
+    - timeout: seconds to wait for a reply
+    Returns the JSON-RPC 'result' from the supervisor or raises on error/timeout.
+    """
+    envelope = {
+        "jsonrpc": JSONRPC_VERSION,
+        "id": 1,
+        "method": "register_runner",
+        "params": [{"secret": admin_secret, "name": name, "queue": queue}],
+    }
+    payload_b64 = base64.b64encode(json.dumps(envelope).encode("utf-8")).decode("ascii")
+    topic_b64 = base64.b64encode(topic.encode("utf-8")).decode("ascii")
+
+    if dst_pk:
+        dst = {"pk": dst_pk}
+    elif dst_ip:
+        dst = {"ip": dst_ip}
+    else:
+        raise RuntimeError("Either dst_pk or dst_ip must be provided for Mycelium destination")
+
+    params = {
+        "message": {"dst": dst, "topic": topic_b64, "payload": payload_b64},
+    }
+    resp = myc.call("pushMessage", params)
+    time.sleep(15)
+
+    # Expect an InboundMessage with a payload if a reply was received
+    # if isinstance(resp, dict) and "payload" in resp:
+    #     try:
+    #         reply = json.loads(base64.b64decode(resp["payload"]).decode("utf-8"))
+    #     except Exception as e:
+    #         raise RuntimeError(f"Invalid supervisor reply payload: {e}")
+    #     if isinstance(reply, dict) and reply.get("error"):
+    #         raise RuntimeError(f"Supervisor register_runner error: {json.dumps(reply['error'])}")
+    #     return reply.get("result")
+    #
+    # raise RuntimeError("No reply received from supervisor for register_runner (timeout)")
+
+
+def mycelium_start_runner(
+    myc: "JsonRpcClient",
+    dst_pk: Optional[str],
+    dst_ip: Optional[str],
+    topic: str,
+    secret: str,
+    actor_id: str = "python",
+    timeout: int = 15,
+) -> Any:
+    """
+    Send supervisor.start_runner over Mycelium using pushMessage and wait for the reply.
+    - actor_id is set to the static name "python" by default to start the registered python runner.
+    Returns the JSON-RPC 'result' or raises on error/timeout.
+    """
+    envelope = {
+        "jsonrpc": JSONRPC_VERSION,
+        "id": 1,
+        "method": "start_runner",
+        "params": [actor_id],
+    }
+    payload_b64 = base64.b64encode(json.dumps(envelope).encode("utf-8")).decode("ascii")
+    topic_b64 = base64.b64encode(topic.encode("utf-8")).decode("ascii")
+
+    if dst_pk:
+        dst = {"pk": dst_pk}
+    elif dst_ip:
+        dst = {"ip": dst_ip}
+    else:
+        raise RuntimeError("Either dst_pk or dst_ip must be provided for Mycelium destination")
+
+    params = {
+        "message": {"dst": dst, "topic": topic_b64, "payload": payload_b64},
+    }
+    resp = myc.call("pushMessage", params)
+
+    time.sleep(15)
+    # if isinstance(resp, dict) and "payload" in resp:
+    #     try:
+    #         reply = json.loads(base64.b64decode(resp["payload"]).decode("utf-8"))
+    #     except Exception as e:
+    #         raise RuntimeError(f"Invalid supervisor reply payload (start_runner): {e}")
+    #     if isinstance(reply, dict) and reply.get("error"):
+    #         raise RuntimeError(f"Supervisor start_runner error: {json.dumps(reply['error'])}")
+    #     return reply.get("result")
+    #
+    # raise RuntimeError("No reply received from supervisor for start_runner (timeout)")
+
+
+def try_create_or_load(client: JsonRpcClient, create_method: str, create_params: Dict[str, Any],
+                       load_method: str, load_params: Dict[str, Any]) -> Any:
+    """Attempt a create; if it fails due to existence, try load."""
+    try:
+        return client.call(create_method, create_params)
+    except RuntimeError as e:
+        msg = str(e)
+        # Server maps AlreadyExists to StorageError, we don't have a structured error code here.
+        if "Already exists" in msg or "Storage Error" in msg or "Invalid params" in msg:
+            # Fall back to load
+            return client.call(load_method, load_params)
+        raise
+
+
+def parse_args() -> argparse.Namespace:
+    p = argparse.ArgumentParser(description="Create actor/context/runner/jobs/flow; start and wait until completion.")
+    group = p.add_mutually_exclusive_group(required=True)
+    group.add_argument("--dst-ip", help="Supervisor Mycelium IP address (IPv4 or IPv6)")
+    group.add_argument("--dst-pk", help="Supervisor public key (64-hex)")
+
+    p.add_argument("--context-id", type=int, default=2, help="Context id (Redis DB index; 0-15). Default: 2")
+    p.add_argument("--actor-id", type=int, default=11001, help="Actor id. Default: 11001")
+    p.add_argument("--runner-id", type=int, default=12001, help="Runner id. Default: 12001")
+    p.add_argument("--flow-id", type=int, default=13001, help="Flow id. Default: 13001")
+    p.add_argument("--base-job-id", type=int, default=20000, help="Base job id for first job; subsequent jobs increment. Default: 20000")
+    p.add_argument("--jobs", type=int, default=3, help="Number of jobs to create (>=1). Forms a simple chain. Default: 3")
+    p.add_argument("--timeout-secs", type=int, default=60, help="Per-job timeout seconds. Default: 60")
+    p.add_argument("--retries", type=int, default=0, help="Per-job retries (0-255). Default: 0")
+    p.add_argument(
+        "--script-type",
+        choices=["Python", "V", "Osis", "Sal"],
+        default="Python",
+        help="ScriptType for jobs/runner. Default: Python"
+    )
+    p.add_argument("--topic", default="supervisor.rpc", help="Supervisor topic. Default: supervisor.rpc")
+    p.add_argument("--secret", help="Optional supervisor secret used for authenticated supervisor calls")
+    p.add_argument("--admin-secret", help="Supervisor admin secret to pre-register a Python runner over Mycelium. If omitted, pre-registration is skipped.")
+    p.add_argument("--poll-interval", type=float, default=2.0, help="Flow poll interval seconds. Default: 2.0")
+    p.add_argument("--poll-timeout", type=int, default=600, help="Max seconds to wait for flow completion. Default: 600")
+    return p.parse_args()
+
+
+def main():
+    args = parse_args()
+    if args.jobs < 1:
+        print("ERROR: --jobs must be >= 1", file=sys.stderr)
+        sys.exit(2)
+
+    url = env_url()
+    client = JsonRpcClient(url)
+
+    mycelium_url = env_mycelium_url()
+    mycelium_client = JsonRpcClient(mycelium_url) if getattr(args, "admin_secret", None) else None
+
+    actor_id = int(args.actor_id)
+    context_id = int(args.context_id)
+    runner_id = int(args.runner_id)
+    flow_id = int(args.flow_id)
+    base_job_id = int(args.base_job_id)
+    script_type = args.script_type
+    timeout = int(args.timeout_secs)
+    retries = int(args.retries)
+    topic = args.topic
+
+    # 1) Actor
+    print_header("actor.create (or load)")
+    actor = try_create_or_load(
+        client,
+        "actor.create",
+        {
+            "actor": {
+                "id": actor_id,
+                "pubkey": "demo-pubkey",
+                "address": ["127.0.0.1"],
+            }
+        },
+        "actor.load",
+        {"id": actor_id},
+    )
+    pretty(actor)
+
+    # 2) Context
+    print_header("context.create (or load)")
+    context = try_create_or_load(
+        client,
+        "context.create",
+        {
+            "context": {
+                "id": context_id,
+                "admins": [actor_id],
+                "readers": [actor_id],
+                "executors": [actor_id],
+            }
+        },
+        "context.load",
+        {"id": context_id},
+    )
+    pretty(context)
+
+    # 3) Runner in this context
+    # Router picks pubkey if non-empty, else IP address.
+    # However, RunnerCreate requires both fields; we fill both and control routing via pubkey empty/non-empty.
+    runner_pubkey = args.dst_pk if args.dst_pk else ""
+    runner_address = args.dst_ip if args.dst_ip else "127.0.0.1"
+
+    # Optional: pre-register a Python runner on the Supervisor over Mycelium using an admin secret
+    if getattr(args, "admin_secret", None):
+        print_header("supervisor.register_runner (pre-register via Mycelium)")
+        try:
+            mycelium_result = mycelium_register_runner(
+                mycelium_client,
+                args.dst_pk if args.dst_pk else None,
+                args.dst_ip if args.dst_ip else None,
+                topic,
+                args.admin_secret,
+                name="Python",
+                queue="Python",
+                timeout=15,
+            )
+            print("Supervisor register_runner ->", mycelium_result)
+        except Exception as e:
+            print(f"ERROR: Supervisor pre-registration failed: {e}", file=sys.stderr)
+            sys.exit(1)
+
+        print_header("supervisor.start_runner (start via Mycelium)")
+        try:
+            mycelium_result = mycelium_start_runner(
+                mycelium_client,
+                args.dst_pk if args.dst_pk else None,
+                args.dst_ip if args.dst_ip else None,
+                topic,
+                args.admin_secret,
+                actor_id="Python",
+                timeout=15,
+            )
+            print("Supervisor start_runner ->", mycelium_result)
+        except Exception as e:
+            print(f"ERROR: Supervisor start failed: {e}", file=sys.stderr)
+            sys.exit(1)
+
+    print_header("runner.create (or load)")
+    # runner.load requires both context_id and id
+    try:
+        runner_payload = {
+            "id": runner_id,
+            "pubkey": runner_pubkey,
+            "address": runner_address,
+            "topic": topic,
+            "script_type": script_type,
+            "local": False,
+        }
+        # Optional supervisor secret used by router for authenticated supervisor calls
+        if getattr(args, "secret", None):
+            runner_payload["secret"] = args.secret
+        runner = client.call("runner.create", {
+            "context_id": context_id,
+            "runner": runner_payload
+        })
+    except RuntimeError as e:
+        msg = str(e)
+        if "Already exists" in msg or "Storage Error" in msg or "Invalid params" in msg:
+            runner = client.call("runner.load", {"context_id": context_id, "id": runner_id})
+        else:
+            raise
+    pretty(runner)
+
+    # 4) Jobs
+    # Build a simple chain: J0 (root), J1 depends on J0, J2 depends on J1, ... up to N-1
+    job_ids: List[int] = []
+    for i in range(args.jobs):
+        jid = base_job_id + i
+        depends = [] if i == 0 else [base_job_id + (i - 1)]
+        job_payload = {
+            "id": jid,
+            "caller_id": actor_id,
+            "context_id": context_id,
+            "script": f"print('Job {i} running')",
+            "script_type": script_type,
+            "timeout": timeout,
+            "retries": retries,
+            "env_vars": {},
+            "prerequisites": [],
+            "depends": depends,
+        }
+        print_header(f"job.create - {jid} {'(root)' if not depends else f'(depends on {depends})'}")
+        try:
+            job = client.call("job.create", {
+                "context_id": context_id,
+                "job": job_payload
+            })
+        except RuntimeError as e:
+            msg = str(e)
+            if "Already exists" in msg or "Storage Error" in msg or "Invalid params" in msg:
+                job = client.call("job.load", {
+                    "context_id": context_id,
+                    "caller_id": actor_id,
+                    "id": jid
+                })
+            else:
+                raise
+        pretty(job)
+        job_ids.append(jid)
+
+    # 5) Flow
+    print_header("flow.create (or load)")
+    try:
+        flow = client.call("flow.create", {
+            "context_id": context_id,
+            "flow": {
+                "id": flow_id,
+                "caller_id": actor_id,
+                "context_id": context_id,
+                "jobs": job_ids,
+                "env_vars": {}
+            }
+        })
+    except RuntimeError as e:
+        msg = str(e)
+        if "Already exists" in msg or "Storage Error" in msg or "Invalid params" in msg:
+            flow = client.call("flow.load", {"context_id": context_id, "id": flow_id})
+        else:
+            raise
+    pretty(flow)
+
+    # Optional: show DAG
+    try:
+        print_header("flow.dag")
+        dag = client.call("flow.dag", {"context_id": context_id, "id": flow_id})
+        pretty(dag)
+    except Exception as e:
+        print(f"WARN: flow.dag failed: {e}", file=sys.stderr)
+
+    # 6) Start flow (idempotent; returns bool whether scheduler started)
+    print_header("flow.start")
+    started = client.call("flow.start", {"context_id": context_id, "id": flow_id})
+    print(f"flow.start -> {started}")
+
+    # 7) Poll until Finished or Error (or timeout)
+    print_header("Polling flow.load until completion")
+    t0 = time.time()
+    status = None
+    last_status_print = 0.0
+    poll_count = 0
+    while True:
+        poll_count += 1
+        flow = client.call("flow.load", {"context_id": context_id, "id": flow_id})
+        status = flow.get("status")
+        now = time.time()
+        if now - last_status_print >= max(1.0, float(args.poll_interval)):
+            print(f"[{int(now - t0)}s] flow.status = {status}")
+            last_status_print = now
+
+        # Every 5th poll, print the current flow DAG
+        if (poll_count % 5) == 0:
+            try:
+                print_header("flow.dag (periodic)")
+                dag = client.call("flow.dag", {"context_id": context_id, "id": flow_id})
+                pretty(dag)
+            except Exception as e:
+                print(f"WARN: periodic flow.dag failed: {e}", file=sys.stderr)
+
+        if status in ("Finished", "Error"):
+            break
+        if (now - t0) > args.poll_timeout:
+            print(f"ERROR: Flow did not complete within {args.poll_timeout}s (status={status})", file=sys.stderr)
+            break
+
+        time.sleep(float(args.poll_interval))
+
+    # 8) Final summary: job statuses
+    print_header("Final job statuses")
+    for jid in job_ids:
+        try:
+            j = client.call("job.load", {
+                "context_id": context_id,
+                "caller_id": actor_id,
+                "id": jid
+            })
+            print(f"Job {jid}: status={j.get('status')} result={j.get('result')}")
+        except Exception as e:
+            print(f"Job {jid}: load failed: {e}", file=sys.stderr)
+
+    # Exit code
+    if status == "Finished":
+        print_header("Result")
+        print("Flow finished successfully.")
+        sys.exit(0)
+    else:
+        print_header("Result")
+        print(f"Flow ended with status={status}")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except KeyboardInterrupt:
+        print("\nInterrupted.")
+        sys.exit(130)
+    except Exception as e:
+        print_header("Error")
+        print(str(e))
+        sys.exit(1)

specs/openrpc.json

@@ -563,6 +563,9 @@
           "local": {
             "type": "boolean"
           },
+          "secret": {
+            "type": "string"
+          },
           "created_at": {
             "type": "integer",
             "format": "int64"
@@ -1001,6 +1004,9 @@
           },
           "local": {
             "type": "boolean"
+          },
+          "secret": {
+            "type": "string"
           }
         }
       },

src/clients/mycelium_client.rs

@@ -3,6 +3,8 @@ use std::sync::atomic::{AtomicU64, Ordering};
 
 use reqwest::Client as HttpClient;
 
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
 use serde_json::{Value, json};
 use thiserror::Error;
 
@@ -53,6 +55,8 @@ impl MyceliumClient {
             "method": method,
             "params": [ params ]
         });
+
+        tracing::info!(%req, "jsonrpc");
         let resp = self.http.post(&self.base_url).json(&req).send().await?;
         let status = resp.status();
         let body: Value = resp.json().await?;
@@ -82,13 +86,13 @@ impl MyceliumClient {
         &self,
         id_hex: &str,
     ) -> Result<TransportStatus, MyceliumClientError> {
-        let params = json!({ "id": id_hex });
+        let params = json!(id_hex);
-        let body = self.jsonrpc("messageStatus", params).await?;
+        let body = self.jsonrpc("getMessageInfo", params).await?;
         let result = body.get("result").ok_or_else(|| {
             MyceliumClientError::InvalidResponse(format!("missing result in response: {body}"))
         })?;
-        // Accept both { status: "..."} and bare "..."
+        // Accept both { state: "..."} and bare "..."
-        let status_str = if let Some(s) = result.get("status").and_then(|v| v.as_str()) {
+        let status_str = if let Some(s) = result.get("state").and_then(|v| v.as_str()) {
             s.to_string()
         } else if let Some(s) = result.as_str() {
             s.to_string()
@@ -97,18 +101,19 @@ impl MyceliumClient {
                 "unexpected result shape: {result}"
             )));
         };
-        Self::map_status(&status_str).ok_or_else(|| {
+        let status = Self::map_status(&status_str).ok_or_else(|| {
             MyceliumClientError::InvalidResponse(format!("unknown status: {status_str}"))
-        })
+        });
+        tracing::info!(%id_hex, status = %status.as_ref().unwrap(), "queried messages status");
+        status
     }
 
     fn map_status(s: &str) -> Option<TransportStatus> {
         match s {
-            "queued" => Some(TransportStatus::Queued),
+            "pending" => Some(TransportStatus::Queued),
-            "sent" => Some(TransportStatus::Sent),
+            "received" => Some(TransportStatus::Delivered),
-            "delivered" => Some(TransportStatus::Delivered),
             "read" => Some(TransportStatus::Read),
-            "failed" => Some(TransportStatus::Failed),
+            "aborted" => Some(TransportStatus::Failed),
             _ => None,
         }
     }
@@ -125,16 +130,15 @@ impl MyceliumClient {
             Destination::Ip(ip) => json!({ "ip": ip.to_string() }),
             Destination::Pk(pk) => json!({ "pk": pk }),
         };
-        let message = json!({
+        let mut message = json!({
             "dst": dst_v,
             "topic": topic,
             "payload": payload_b64,
         });
-        let mut params = json!({ "message": message });
         if let Some(rt) = reply_timeout {
-            params["reply_timeout"] = json!(rt);
+            message["reply_timeout"] = json!(rt);
         }
-        params
+        message
     }
 
     /// pushMessage: send a message with dst/topic/payload. Optional reply_timeout for sync replies.
@@ -160,6 +164,83 @@ impl MyceliumClient {
             .and_then(|v| v.as_str())
             .map(|s| s.to_string())
     }
+    /// popMessage: retrieve an inbound message if available (optionally filtered by topic).
+    /// - peek: if true, do not remove the message from the queue
+    /// - timeout_secs: seconds to wait for a message (0 returns immediately)
+    /// - topic_plain: optional plain-text topic which will be base64-encoded per Mycelium spec
+    /// Returns:
+    ///   - Ok(Some(result_json)) on success, where result_json matches InboundMessage schema
+    ///   - Ok(None) when there is no message ready (Mycelium returns error code 204)
+    pub async fn pop_message(
+        &self,
+        peek: Option<bool>,
+        timeout_secs: Option<u64>,
+        topic_plain: Option<&str>,
+    ) -> Result<Option<Value>, MyceliumClientError> {
+        // Build params array
+        let mut params_array = vec![];
+        if let Some(p) = peek {
+            params_array.push(serde_json::Value::Bool(p));
+        } else {
+            params_array.push(serde_json::Value::Null)
+        }
+        if let Some(t) = timeout_secs {
+            params_array.push(serde_json::Value::Number(t.into()));
+        } else {
+            params_array.push(serde_json::Value::Null)
+        }
+        if let Some(tp) = topic_plain {
+            let topic_b64 = BASE64_STANDARD.encode(tp.as_bytes());
+            params_array.push(serde_json::Value::String(topic_b64));
+        } else {
+            params_array.push(serde_json::Value::Null)
+        }
+
+        let req = json!({
+            "jsonrpc": "2.0",
+            "id": self.next_id(),
+            "method": "popMessage",
+            "params": serde_json::Value::Array(params_array),
+        });
+
+        tracing::info!(%req, "calling popMessage");
+
+        let resp = self.http.post(&self.base_url).json(&req).send().await?;
+        let status = resp.status();
+        let body: Value = resp.json().await?;
+
+        // Handle JSON-RPC error envelope specially for code 204 (no message ready)
+        if let Some(err) = body.get("error") {
+            let code = err.get("code").and_then(|v| v.as_i64()).unwrap_or(0);
+            let msg = err
+                .get("message")
+                .and_then(|v| v.as_str())
+                .unwrap_or("unknown error");
+
+            if code == 204 {
+                // No message ready
+                return Ok(None);
+            }
+            if code == 408 {
+                // Align with other transport timeout mapping
+                return Err(MyceliumClientError::TransportTimeout);
+            }
+            return Err(MyceliumClientError::RpcError(format!(
+                "code={code} msg={msg}"
+            )));
+        }
+
+        if !status.is_success() {
+            return Err(MyceliumClientError::RpcError(format!(
+                "HTTP {status}, body {body}"
+            )));
+        }
+
+        let result = body.get("result").ok_or_else(|| {
+            MyceliumClientError::InvalidResponse(format!("missing result in response: {body}"))
+        })?;
+        Ok(Some(result.clone()))
+    }
 }
 
 #[cfg(test)]

src/clients/supervisor_client.rs

@@ -99,11 +99,25 @@ impl SupervisorClient {
         })
     }
 
+    /// Build a supervisor JSON-RPC payload but force a specific id (used for correlation).
+    fn build_supervisor_payload_with_id(&self, method: &str, params: Value, id: u64) -> Value {
+        json!({
+            "jsonrpc": "2.0",
+            "id": id,
+            "method": method,
+            "params": params,
+        })
+    }
+
     fn encode_payload(payload: &Value) -> Result<String, SupervisorClientError> {
         let s = serde_json::to_string(payload)?;
         Ok(BASE64_STANDARD.encode(s.as_bytes()))
     }
 
+    fn encode_topic(topic: &[u8]) -> String {
+        BASE64_STANDARD.encode(topic)
+    }
+
     fn extract_message_id_from_result(result: &Value) -> Option<String> {
         // Two possibilities per Mycelium spec oneOf:
         // - PushMessageResponseId: { "id": "0123456789abcdef" }
@@ -120,7 +134,12 @@ impl SupervisorClient {
         let payload_b64 = Self::encode_payload(&inner)?;
         let result = self
             .mycelium
-            .push_message(&self.destination, &self.topic, &payload_b64, None)
+            .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) {
@@ -138,66 +157,42 @@ impl SupervisorClient {
         )))
     }
 
-    /// Synchronous variant: wait for a JSON-RPC reply via Mycelium reply_timeout, and return the inner JSON-RPC "result".
+    /// Variant of call that also returns the inner supervisor JSON-RPC id used in the payload.
-    /// If the supervisor returns an error object, map to RpcError.
+    /// This id is required to correlate asynchronous popMessage replies coming from Mycelium.
-    pub async fn call_sync(
+    pub async fn call_with_ids(
         &self,
         method: &str,
         params: Value,
-        reply_timeout_secs: u64,
+    ) -> Result<(String, u64), SupervisorClientError> {
-    ) -> Result<Value, SupervisorClientError> {
+        let inner_id = self.next_id();
-        let inner = self.build_supervisor_payload(method, params);
+        let inner = self.build_supervisor_payload_with_id(method, params, inner_id);
         let payload_b64 = Self::encode_payload(&inner)?;
-
         let result = self
             .mycelium
             .push_message(
                 &self.destination,
-                &self.topic,
+                &Self::encode_topic(self.topic.as_bytes()),
                 &payload_b64,
-                Some(reply_timeout_secs),
+                None,
             )
             .await?;
 
-        // Expect an InboundMessage-like with a base64 payload containing the supervisor JSON-RPC response
+        let out_id = if let Some(id) = MyceliumClient::extract_message_id_from_result(&result) {
-        let payload_field = if let Some(p) = result.get("payload").and_then(|v| v.as_str()) {
+            id
-            p.to_string()
+        } else if let Some(arr) = result.as_array()
-        } else if let Some(arr) = result.as_array() {
+            && arr.len() == 1
-            // Defensive: handle single-element array shape
+            && let Some(id) = MyceliumClient::extract_message_id_from_result(&arr[0])
-            if let Some(one) = arr.get(0) {
+        {
-                one.get("payload")
+            id
-                    .and_then(|v| v.as_str())
-                    .map(|s| s.to_string())
-                    .ok_or_else(|| {
-                        SupervisorClientError::InvalidResponse(format!(
-                            "missing payload in result: {result}"
-                        ))
-                    })?
         } else {
-                return Err(SupervisorClientError::TransportTimeout);
+            return Err(SupervisorClientError::InvalidResponse(format!(
-            }
+                "result did not contain message id: {result}"
-        } else {
+            )));
-            // No payload => no reply received within timeout (Mycelium would have returned just an id)
-            return Err(SupervisorClientError::TransportTimeout);
         };
 
-        let raw = BASE64_STANDARD
+        Ok((out_id, inner_id))
-            .decode(payload_field.as_bytes())
+    }
-            .map_err(|e| {
-                SupervisorClientError::InvalidResponse(format!("invalid base64 payload: {e}"))
-            })?;
-        let rpc_resp: Value = serde_json::from_slice(&raw)?;
 
-        if let Some(err) = rpc_resp.get("error") {
-            return Err(SupervisorClientError::RpcError(err.to_string()));
-        }
-        let res = rpc_resp.get("result").ok_or_else(|| {
-            SupervisorClientError::InvalidResponse(format!(
-                "missing result in supervisor reply: {rpc_resp}"
-            ))
-        })?;
-        Ok(res.clone())
-    }
 
     fn need_secret(&self) -> Result<&str, SupervisorClientError> {
         self.secret
@@ -299,6 +294,19 @@ impl SupervisorClient {
         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
+    }
+
     pub async fn job_start(
         &self,
         job_id: impl Into<String>,
@@ -318,28 +326,15 @@ impl SupervisorClient {
         self.call("job.status", json!([job_id.into()])).await
     }
 
-    /// Synchronous job.status: waits for the supervisor to reply and returns the status string.
+    /// Asynchronous job.status returning outbound and inner IDs for correlation
-    /// The supervisor result may be an object with { status: "..." } or a bare string.
+    pub async fn job_status_with_ids(
-    pub async fn job_status_sync(
         &self,
         job_id: impl Into<String>,
-        reply_timeout_secs: u64,
+    ) -> Result<(String, u64), SupervisorClientError> {
-    ) -> Result<String, SupervisorClientError> {
+        self.call_with_ids("job.status", json!([job_id.into()])).await
-        let res = self
-            .call_sync("job.status", json!([job_id.into()]), reply_timeout_secs)
-            .await?;
-        let status = if let Some(s) = res.get("status").and_then(|v| v.as_str()) {
-            s.to_string()
-        } else if let Some(s) = res.as_str() {
-            s.to_string()
-        } else {
-            return Err(SupervisorClientError::InvalidResponse(format!(
-                "unexpected job.status result shape: {res}"
-            )));
-        };
-        Ok(status)
     }
 
+
     pub async fn job_result(
         &self,
         job_id: impl Into<String>,
@@ -347,44 +342,14 @@ impl SupervisorClient {
         self.call("job.result", json!([job_id.into()])).await
     }
 
-    /// Synchronous job.result: waits for the supervisor to reply and returns a map
+    /// Asynchronous job.result returning outbound and inner IDs for correlation
-    /// containing exactly one of:
+    pub async fn job_result_with_ids(
-    /// - {"success": "..."} on success
-    /// - {"error": "..."} on error reported by the runner
-    /// Some servers may return a bare string; we treat that as {"success": "<string>"}.
-    pub async fn job_result_sync(
         &self,
         job_id: impl Into<String>,
-        reply_timeout_secs: u64,
+    ) -> Result<(String, u64), SupervisorClientError> {
-    ) -> Result<std::collections::HashMap<String, String>, SupervisorClientError> {
+        self.call_with_ids("job.result", json!([job_id.into()])).await
-        let res = self
-            .call_sync("job.result", json!([job_id.into()]), reply_timeout_secs)
-            .await?;
-
-        use std::collections::HashMap;
-        let mut out: HashMap<String, String> = HashMap::new();
-
-        if let Some(obj) = res.as_object() {
-            if let Some(s) = obj.get("success").and_then(|v| v.as_str()) {
-                out.insert("success".to_string(), s.to_string());
-                return Ok(out);
-            }
-            if let Some(s) = obj.get("error").and_then(|v| v.as_str()) {
-                out.insert("error".to_string(), s.to_string());
-                return Ok(out);
-            }
-            return Err(SupervisorClientError::InvalidResponse(format!(
-                "unexpected job.result result shape: {res}"
-            )));
-        } else if let Some(s) = res.as_str() {
-            out.insert("success".to_string(), s.to_string());
-            return Ok(out);
     }
 
-        Err(SupervisorClientError::InvalidResponse(format!(
-            "unexpected job.result result shape: {res}"
-        )))
-    }
 
     pub async fn job_stop(
         &self,

src/main.rs

@@ -2,8 +2,8 @@ use clap::Parser;
 use std::net::{IpAddr, SocketAddr};
 use std::sync::Arc;
 
-use tracing::{error, info, warn};
+use tracing::{error, info};
-use tracing_subscriber::{EnvFilter, fmt};
+use tracing_subscriber::EnvFilter;
 #[derive(Debug, Clone, Parser)]
 #[command(
     name = "herocoordinator",
@@ -25,8 +25,8 @@ struct Cli {
         long = "mycelium-port",
         short = 'p',
         env = "MYCELIUM_PORT",
-        default_value_t = 9651u16,
+        default_value_t = 8990u16,
-        help = "Port for Mycelium JSON-RPC (default: 9651)"
+        help = "Port for Mycelium JSON-RPC (default: 8990)"
     )]
     mycelium_port: u16,
 
@@ -99,7 +99,7 @@ async fn main() {
     // Shared application state
     let state = Arc::new(herocoordinator::rpc::AppState::new(service));
 
-    // Start router workers (auto-discovered contexts)
+    // Start router workers (auto-discovered contexts) and a single global inbound listener
     {
         let base_url = format!("http://{}:{}", cli.mycelium_ip, cli.mycelium_port);
         let cfg = herocoordinator::router::RouterConfig {
@@ -110,6 +110,10 @@ async fn main() {
             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
         let _auto_handle = herocoordinator::router::start_router_auto(service_for_router, cfg);
     }
 

src/models/message.rs

@@ -59,6 +59,18 @@ pub enum TransportStatus {
     Failed,
 }
 
+impl std::fmt::Display for TransportStatus {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            TransportStatus::Queued => f.write_str("queued"),
+            TransportStatus::Sent => f.write_str("sent"),
+            TransportStatus::Delivered => f.write_str("delivered"),
+            TransportStatus::Read => f.write_str("read"),
+            TransportStatus::Failed => f.write_str("failed"),
+        }
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum MessageFormatType {
     Html,

src/models/runner.rs

@@ -18,6 +18,8 @@ pub struct Runner {
     pub script_type: ScriptType,
     /// If this is true, the runner also listens on a local redis queue
     pub local: bool,
+    /// Optional secret used for authenticated supervisor calls (if required)
+    pub secret: Option<String>,
     pub created_at: Timestamp,
     pub updated_at: Timestamp,
 }

src/router.rs

@@ -1,14 +1,18 @@
-use std::{collections::HashSet, sync::Arc};
+use std::{collections::{HashSet, HashMap}, sync::Arc};
 
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
 use serde_json::{Value, json};
-use tokio::sync::Semaphore;
+use tokio::sync::{Semaphore, Mutex};
+use std::hash::{Hash, Hasher};
+use std::collections::hash_map::DefaultHasher;
 
 use crate::{
     clients::{Destination, MyceliumClient, SupervisorClient},
     models::{Job, JobStatus, Message, MessageStatus, ScriptType, TransportStatus},
     service::AppService,
 };
-use tracing::{error, info, warn};
+use tracing::{error, info};
 
 #[derive(Clone, Debug)]
 pub struct RouterConfig {
@@ -21,6 +25,88 @@ pub struct RouterConfig {
     pub transport_poll_timeout_secs: u64,  // e.g. 300 (5 minutes)
 }
 
+/*
+SupervisorClient reuse cache (Router-local):
+
+Rationale:
+- SupervisorClient maintains an internal JSON-RPC id_counter per instance.
+- Rebuilding a client for each message resets this counter, causing inner JSON-RPC ids to restart at 1.
+- We reuse one SupervisorClient per (destination, topic, secret) to preserve monotonically increasing ids.
+
+Scope:
+- Cache is per Router loop (and a separate one for the inbound listener).
+- If cross-loop/process reuse becomes necessary later, promote to a process-global cache.
+
+Keying:
+- Key: destination + topic + secret-presence (secret content hashed; not stored in plaintext).
+
+Concurrency:
+- tokio::Mutex protects a HashMap<String, Arc<SupervisorClient>>.
+- Values are Arc so call sites clone cheaply and share the same id_counter.
+*/
+#[derive(Clone)]
+struct SupervisorClientCache {
+    map: Arc<Mutex<HashMap<String, Arc<SupervisorClient>>>>,
+}
+
+impl SupervisorClientCache {
+    fn new() -> Self {
+        Self {
+            map: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+
+    fn make_key(dest: &Destination, topic: &str, secret: &Option<String>) -> String {
+        let dst = match dest {
+            Destination::Ip(ip) => format!("ip:{ip}"),
+            Destination::Pk(pk) => format!("pk:{pk}"),
+        };
+        // Hash the secret to avoid storing plaintext in keys while still differentiating values
+        let sec_hash = match secret {
+            Some(s) if !s.is_empty() => {
+                let mut hasher = DefaultHasher::new();
+                s.hash(&mut hasher);
+                format!("s:{}", hasher.finish())
+            }
+            _ => "s:none".to_string(),
+        };
+        format!("{dst}|t:{topic}|{sec_hash}")
+    }
+
+    async fn get_or_create(
+        &self,
+        mycelium: Arc<MyceliumClient>,
+        dest: Destination,
+        topic: String,
+        secret: Option<String>,
+    ) -> Arc<SupervisorClient> {
+        let key = Self::make_key(&dest, &topic, &secret);
+
+        {
+            let guard = self.map.lock().await;
+            if let Some(existing) = guard.get(&key) {
+                tracing::debug!(target: "router", cache="supervisor", hit=true, %topic, secret = %if secret.is_some() { "set" } else { "none" }, "SupervisorClient cache lookup");
+                return existing.clone();
+            }
+        }
+
+        let mut guard = self.map.lock().await;
+        if let Some(existing) = guard.get(&key) {
+            tracing::debug!(target: "router", cache="supervisor", hit=true, %topic, secret = %if secret.is_some() { "set" } else { "none" }, "SupervisorClient cache lookup (double-checked)");
+            return existing.clone();
+        }
+        let client = Arc::new(SupervisorClient::new_with_client(
+            mycelium,
+            dest,
+            topic.clone(),
+            secret.clone(),
+        ));
+        guard.insert(key, client.clone());
+        tracing::debug!(target: "router", cache="supervisor", hit=false, %topic, secret = %if secret.is_some() { "set" } else { "none" }, "SupervisorClient cache insert");
+        client
+    }
+}
+
 /// Start background router loops, one per context.
 /// Each loop:
 /// - BRPOP msg_out with 1s timeout
@@ -47,6 +133,8 @@ pub fn start_router(service: AppService, cfg: RouterConfig) -> Vec<tokio::task::
                 }
             };
 
+            let cache = Arc::new(SupervisorClientCache::new());
+
             loop {
                 // Pop next message key (blocking with timeout)
                 match service_cloned.brpop_msg_out(ctx_id, 1).await {
@@ -67,11 +155,12 @@ pub fn start_router(service: AppService, cfg: RouterConfig) -> Vec<tokio::task::
                         let cfg_task = cfg_cloned.clone();
                         tokio::spawn({
                             let mycelium = mycelium.clone();
+                            let cache = cache.clone();
                             async move {
                                 // Ensure permit is dropped at end of task
                                 let _permit = permit;
                                 if let Err(e) =
-                                    deliver_one(&service_task, &cfg_task, ctx_id, &key, mycelium)
+                                    deliver_one(&service_task, &cfg_task, ctx_id, &key, mycelium, cache.clone())
                                         .await
                                 {
                                     error!(context_id=ctx_id, key=%key, error=%e, "Delivery error");
@@ -102,6 +191,7 @@ async fn deliver_one(
     context_id: u32,
     msg_key: &str,
     mycelium: Arc<MyceliumClient>,
+    cache: Arc<SupervisorClientCache>,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     // Parse "message:{caller_id}:{id}"
     let (caller_id, id) = parse_message_key(msg_key)
@@ -140,19 +230,39 @@ async fn deliver_one(
     // Keep clones for poller usage
     let dest_for_poller = dest.clone();
     let topic_for_poller = cfg.topic.clone();
-    let client = SupervisorClient::new_with_client(
+    let secret_for_poller = runner.secret.clone();
+    let client = cache
+        .get_or_create(
             mycelium.clone(),
             dest.clone(),
             cfg.topic.clone(),
-        None, // secret
+            runner.secret.clone(),
-    );
+        )
+        .await;
 
     // Build supervisor method and params from Message
     let method = msg.message.clone();
     let params = build_params(&msg)?;
 
     // Send
-    let out_id = client.call(&method, params).await?;
+    // If this is a job.run and we have a secret configured on the client,
+    // prefer the typed wrapper that injects the secret into inner supervisor params,
+    // and also capture the inner supervisor JSON-RPC id for correlation.
+    let (out_id, inner_id_opt) = if method == "job.run" {
+        if let Some(j) = msg.job.first() {
+            let jv = job_to_json(j)?;
+            // Returns (outbound message id, inner supervisor JSON-RPC id)
+            let (out, inner) = client.job_run_with_ids(jv).await?;
+            (out, Some(inner))
+        } else {
+            // Fallback: no embedded job, use the generic call
+            let out = client.call(&method, params).await?;
+            (out, None)
+        }
+    } else {
+        let out = client.call(&method, params).await?;
+        (out, None)
+    };
 
     // Store transport id and initial Sent status
     let _ = service
@@ -170,6 +280,13 @@ async fn deliver_one(
         .update_message_status(context_id, caller_id, id, MessageStatus::Acknowledged)
         .await?;
 
+    // Record correlation (inner supervisor JSON-RPC id -> job/message) for inbound popMessage handling
+    if let (Some(inner_id), Some(job_id)) = (inner_id_opt, job_id_opt) {
+        let _ = service
+            .supcorr_set(inner_id, context_id, caller_id, job_id, id)
+            .await;
+    }
+
     // Spawn transport-status poller
     {
         let service_poll = service.clone();
@@ -177,12 +294,6 @@ async fn deliver_one(
         let poll_timeout = std::time::Duration::from_secs(cfg.transport_poll_timeout_secs);
         let out_id_cloned = out_id.clone();
         let mycelium = mycelium.clone();
-        // Determine reply timeout for supervisor job.result: prefer message.timeout_result, fallback to router config timeout
-        let job_result_reply_timeout: u64 = if msg.timeout_result > 0 {
-            msg.timeout_result as u64
-        } else {
-            cfg.transport_poll_timeout_secs
-        };
 
         tokio::spawn(async move {
             let start = std::time::Instant::now();
@@ -194,6 +305,8 @@ async fn deliver_one(
             let job_id_opt = job_id_opt;
 
             let mut last_status: Option<TransportStatus> = Some(TransportStatus::Sent);
+            // Ensure we only request supervisor job.status or job.result once per outbound message
+            let mut requested_job_check: bool = false;
 
             loop {
                 if start.elapsed() >= poll_timeout {
@@ -226,70 +339,51 @@ async fn deliver_one(
 
                         // Stop on terminal states
                         if matches!(s, TransportStatus::Delivered | TransportStatus::Read) {
-                            // On Read, fetch supervisor job.status and update local job/message if terminal
+                            // Only request a single job status/result per message
-                            if matches!(s, TransportStatus::Read) {
+                            if !requested_job_check {
                                 if let Some(job_id) = job_id_opt {
-                                    let sup = SupervisorClient::new_with_client(
+                                    // First consult Redis for the latest job state in case we already have a terminal update
-                                        client.clone(),
+                                    match service_poll.load_job(context_id, caller_id, job_id).await {
-                                        sup_dest.clone(),
+                                        Ok(job) => {
-                                        sup_topic.clone(),
+                                            match job.status() {
-                                        None,
+                                                JobStatus::Finished | JobStatus::Error => {
-                                    );
+                                                    // Local job is already terminal; skip supervisor job.status
-                                    match sup.job_status_sync(job_id.to_string(), 10).await {
-                                        Ok(remote_status) => {
-                                            if let Some((mapped, terminal)) =
-                                                map_supervisor_job_status(&remote_status)
-                                            {
-                                                if terminal {
-                                                    let _ = service_poll
-                                                        .update_job_status_unchecked(
-                                                            context_id,
-                                                            caller_id,
-                                                            job_id,
-                                                            mapped.clone(),
-                                                        )
-                                                        .await;
-
-                                                    // After terminal status, fetch supervisor job.result and store into Job.result
-                                                    let sup = SupervisorClient::new_with_client(
-                                                        client.clone(),
-                                                        sup_dest.clone(),
-                                                        sup_topic.clone(),
-                                                        None,
-                                                    );
-                                                    match sup
-                                                        .job_result_sync(
-                                                            job_id.to_string(),
-                                                            job_result_reply_timeout,
-                                                        )
-                                                        .await
-                                                    {
-                                                        Ok(result_map) => {
-                                                            // Persist the result into the Job.result map (merge)
-                                                            let _ = service_poll
-                                                                .update_job_result_merge_unchecked(
-                                                                    context_id,
-                                                                    caller_id,
-                                                                    job_id,
-                                                                    result_map.clone(),
-                                                                )
-                                                                .await;
-                                                            // Log which key was stored (success or error)
-                                                            let key = result_map
-                                                                .keys()
-                                                                .next()
-                                                                .cloned()
-                                                                .unwrap_or_else(|| {
-                                                                    "unknown".to_string()
-                                                                });
                                                     let _ = service_poll
                                                         .append_message_logs(
                                                             context_id,
                                                             caller_id,
                                                             id,
                                                             vec![format!(
-                                                                        "Stored supervisor job.result for job {} ({})",
+                                                                "Local job {} status is terminal ({:?}); skipping supervisor job.status",
-                                                                        job_id, key
+                                                                job_id,
+                                                                job.status()
+                                                            )],
+                                                        )
+                                                        .await;
+
+                                                    // If result is still empty, immediately request supervisor job.result
+                                                    if job.result.is_empty() {
+                                                        let sup = cache
+                                                            .get_or_create(
+                                                                client.clone(),
+                                                                sup_dest.clone(),
+                                                                sup_topic.clone(),
+                                                                secret_for_poller.clone(),
+                                                            )
+                                                            .await;
+                                                        match sup.job_result_with_ids(job_id.to_string()).await {
+                                                            Ok((_out2, inner2)) => {
+                                                                let _ = service_poll
+                                                                    .supcorr_set(inner2, context_id, caller_id, job_id, id)
+                                                                    .await;
+                                                                let _ = service_poll
+                                                                    .append_message_logs(
+                                                                        context_id,
+                                                                        caller_id,
+                                                                        id,
+                                                                        vec![format!(
+                                                                            "Requested supervisor job.result for job {} (local terminal w/ empty result)",
+                                                                            job_id
                                                                         )],
                                                                     )
                                                                     .await;
@@ -301,63 +395,119 @@ async fn deliver_one(
                                                                         caller_id,
                                                                         id,
                                                                         vec![format!(
-                                                                        "job.result fetch error for job {}: {}",
+                                                                            "job.result request error for job {}: {}",
                                                                             job_id, e
                                                                         )],
                                                                     )
                                                                     .await;
                                                             }
                                                         }
-
-                                                    // Mark message as processed
-                                                    let _ = service_poll
-                                                        .update_message_status(
-                                                            context_id,
-                                                            caller_id,
-                                                            id,
-                                                            MessageStatus::Processed,
-                                                        )
-                                                        .await;
-                                                    let _ = service_poll
-                                                        .append_message_logs(
-                                                            context_id,
-                                                            caller_id,
-                                                            id,
-                                                            vec![format!(
-                                                                "Supervisor job.status for job {} -> {} (mapped to {:?})",
-                                                                job_id, remote_status, mapped
-                                                            )],
-                                                        )
-                                                        .await;
-                                                }
                                                     } else {
+                                                        // Result already present; nothing to fetch
                                                         let _ = service_poll
                                                             .append_message_logs(
                                                                 context_id,
                                                                 caller_id,
                                                                 id,
                                                                 vec![format!(
-                                                            "Unknown supervisor status '{}' for job {}",
+                                                                    "Job {} already has result; no supervisor calls needed",
-                                                            remote_status, job_id
+                                                                    job_id
                                                                 )],
                                                             )
                                                             .await;
                                                     }
                                                 }
+                                                // Not terminal yet -> request supervisor job.status as before
+                                                _ => {
+                                                    let sup = cache
+                                                        .get_or_create(
+                                                            client.clone(),
+                                                            sup_dest.clone(),
+                                                            sup_topic.clone(),
+                                                            secret_for_poller.clone(),
+                                                        )
+                                                        .await;
+                                                    match sup.job_status_with_ids(job_id.to_string()).await {
+                                                        Ok((_out_id, inner_id)) => {
+                                                            // Correlate this status request to the message/job
+                                                            let _ = service_poll
+                                                                .supcorr_set(
+                                                                    inner_id, context_id, caller_id, job_id, id,
+                                                                )
+                                                                .await;
+                                                            let _ = service_poll
+                                                                .append_message_logs(
+                                                                    context_id,
+                                                                    caller_id,
+                                                                    id,
+                                                                    vec![format!(
+                                                                        "Requested supervisor job.status for job {}",
+                                                                        job_id
+                                                                    )],
+                                                                )
+                                                                .await;
+                                                        }
                                                         Err(e) => {
                                                             let _ = service_poll
                                                                 .append_message_logs(
                                                                     context_id,
                                                                     caller_id,
                                                                     id,
-                                                    vec![format!("job.status sync error: {}", e)],
+                                                                    vec![format!("job.status request error: {}", e)],
                                                                 )
                                                                 .await;
                                                         }
                                                     }
                                                 }
                                             }
-                            break;
+                                        }
+                                        // If we cannot load the job, fall back to requesting job.status
+                                        Err(_) => {
+                                            let sup = cache
+                                                .get_or_create(
+                                                    client.clone(),
+                                                    sup_dest.clone(),
+                                                    sup_topic.clone(),
+                                                    secret_for_poller.clone(),
+                                                )
+                                                .await;
+                                            match sup.job_status_with_ids(job_id.to_string()).await {
+                                                Ok((_out_id, inner_id)) => {
+                                                    let _ = service_poll
+                                                        .supcorr_set(
+                                                            inner_id, context_id, caller_id, job_id, id,
+                                                        )
+                                                        .await;
+                                                    let _ = service_poll
+                                                        .append_message_logs(
+                                                            context_id,
+                                                            caller_id,
+                                                            id,
+                                                            vec![format!(
+                                                                "Requested supervisor job.status for job {} (fallback; load_job failed)",
+                                                                job_id
+                                                            )],
+                                                        )
+                                                        .await;
+                                                }
+                                                Err(e) => {
+                                                    let _ = service_poll
+                                                        .append_message_logs(
+                                                            context_id,
+                                                            caller_id,
+                                                            id,
+                                                            vec![format!("job.status request error: {}", e)],
+                                                        )
+                                                        .await;
+                                                }
+                                            }
+                                        }
+                                    }
+                                    // Ensure we only do this once
+                                    requested_job_check = true;
+                                }
+                            }
+                            // break;
                         }
                         if matches!(s, TransportStatus::Failed) {
                             let _ = service_poll
@@ -473,3 +623,424 @@ pub fn start_router_auto(service: AppService, cfg: RouterConfig) -> tokio::task:
         }
     })
 }
+
+/// Start a single global inbound listener that reads Mycelium popMessage with topic filter,
+/// decodes supervisor JSON-RPC replies, and updates correlated jobs/messages.
+/// This listens for async replies like {"result":{"job_queued":...}} carrying the same inner JSON-RPC id.
+pub fn start_inbound_listener(
+    service: AppService,
+    cfg: RouterConfig,
+) -> tokio::task::JoinHandle<()> {
+    tokio::spawn(async move {
+        // Initialize Mycelium client (retry loop)
+        let mycelium = loop {
+            match MyceliumClient::new(cfg.base_url.clone()) {
+                Ok(c) => break Arc::new(c),
+                Err(e) => {
+                    error!(error=%e, "MyceliumClient init error (inbound listener)");
+                    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                }
+            }
+        };
+
+        let cache = Arc::new(SupervisorClientCache::new());
+
+        loop {
+            // Poll for inbound supervisor messages on the configured topic
+            match mycelium.pop_message(Some(false), Some(20), None).await {
+                Ok(Some(inb)) => {
+                    // Expect InboundMessage with base64 "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 {
+                        let _ = service
+                            .append_message_logs(
+                                0, // unknown context yet
+                                0,
+                                0,
+                                vec![
+                                    "Inbound payload base64 decode error (supervisor reply)".into(),
+                                ],
+                            )
+                            .await;
+                        continue;
+                    };
+                    tracing::info!(
+                        raw = %String::from_utf8_lossy(&raw),
+                        "Read raw messge from mycelium"
+                    );
+                    let Ok(rpc): Result<Value, _> = serde_json::from_slice(&raw) else {
+                        // Invalid JSON payload
+                        continue;
+                    };
+
+                    // Extract inner supervisor JSON-RPC id (number preferred; string fallback)
+                    let inner_id_u64 = match rpc.get("id") {
+                        Some(Value::Number(n)) => n.as_u64(),
+                        Some(Value::String(s)) => s.parse::<u64>().ok(),
+                        _ => None,
+                    };
+                    let Some(inner_id) = inner_id_u64 else {
+                        // Cannot correlate without id
+                        continue;
+                    };
+
+                    // Lookup correlation mapping
+                    match service.supcorr_get(inner_id).await {
+                        Ok(Some((context_id, caller_id, job_id, message_id))) => {
+                            // Determine success/error from supervisor JSON-RPC envelope
+                            // Inspect result/error to route job.run/job.status/job.result replies
+                            let result_opt = rpc.get("result");
+                            let error_opt = rpc.get("error");
+
+                            // Handle job.run success (job_queued)
+                            let is_job_queued = result_opt
+                                .and_then(|res| {
+                                    if res.get("job_queued").is_some() {
+                                        Some(true)
+                                    } else if let Some(s) = res.as_str() {
+                                        Some(s == "job_queued")
+                                    } else {
+                                        None
+                                    }
+                                })
+                                .unwrap_or(false);
+
+                            if is_job_queued {
+                                // Set to Dispatched (idempotent) per spec, and append log
+                                let _ = service
+                                    .update_job_status_unchecked(
+                                        context_id,
+                                        caller_id,
+                                        job_id,
+                                        JobStatus::Dispatched,
+                                    )
+                                    .await;
+                                let _ = service
+                                    .append_message_logs(
+                                        context_id,
+                                        caller_id,
+                                        message_id,
+                                        vec![format!(
+                                            "Supervisor reply for job {}: job_queued",
+                                            job_id
+                                        )],
+                                    )
+                                    .await;
+                                let _ = service.supcorr_del(inner_id).await;
+                                continue;
+                            }
+
+                            // Error envelope: set job Error and log
+                            if let Some(err_obj) = error_opt {
+                                let _ = service
+                                    .update_job_status_unchecked(
+                                        context_id,
+                                        caller_id,
+                                        job_id,
+                                        JobStatus::Error,
+                                    )
+                                    .await;
+                                let _ = service
+                                    .append_message_logs(
+                                        context_id,
+                                        caller_id,
+                                        message_id,
+                                        vec![format!(
+                                            "Supervisor error for job {}: {}",
+                                            job_id, err_obj
+                                        )],
+                                    )
+                                    .await;
+                                let _ = service.supcorr_del(inner_id).await;
+                                continue;
+                            }
+
+                            // If we have a result, try to interpret it as job.status or job.result
+                            if let Some(res) = result_opt {
+                                // Try job.status: object {status: "..."} or bare string
+                                let status_candidate = res
+                                    .get("status")
+                                    .and_then(|v| v.as_str())
+                                    .or_else(|| res.as_str());
+
+                                if let Some(remote_status) = status_candidate {
+                                    if let Some((mapped, terminal)) =
+                                        map_supervisor_job_status(remote_status)
+                                    {
+                                        // Update job status and log
+                                        let _ = service
+                                            .update_job_status_unchecked(
+                                                context_id,
+                                                caller_id,
+                                                job_id,
+                                                mapped.clone(),
+                                            )
+                                            .await;
+                                        let _ = service
+                                            .append_message_logs(
+                                                context_id,
+                                                caller_id,
+                                                message_id,
+                                                vec![format!(
+                                                    "Supervisor job.status for job {} -> {} (mapped to {:?})",
+                                                    job_id, remote_status, mapped
+                                                )],
+                                            )
+                                            .await;
+                                        // Done with this correlation id
+                                        let _ = service.supcorr_del(inner_id).await;
+
+                                        // If terminal, request job.result asynchronously now
+                                        if terminal {
+                                            // Load job to determine script_type for runner selection
+                                            match service
+                                                .load_job(context_id, caller_id, job_id)
+                                                .await
+                                            {
+                                                Ok(job) => {
+                                                    match service.scan_runners(context_id).await {
+                                                        Ok(runners) => {
+                                                            if let Some(runner) =
+                                                                runners.into_iter().find(|r| {
+                                                                    r.script_type == job.script_type
+                                                                })
+                                                            {
+                                                                let dest = if !runner
+                                                                    .pubkey
+                                                                    .trim()
+                                                                    .is_empty()
+                                                                {
+                                                                    Destination::Pk(
+                                                                        runner.pubkey.clone(),
+                                                                    )
+                                                                } else {
+                                                                    Destination::Ip(runner.address)
+                                                                };
+                                                                let sup = cache
+                                                                    .get_or_create(
+                                                                        mycelium.clone(),
+                                                                        dest,
+                                                                        cfg.topic.clone(),
+                                                                        runner.secret.clone(),
+                                                                    )
+                                                                    .await;
+                                                                match sup
+                                                                    .job_result_with_ids(
+                                                                        job_id.to_string(),
+                                                                    )
+                                                                    .await
+                                                                {
+                                                                    Ok((_out2, inner2)) => {
+                                                                        let _ = service
+                                                                            .supcorr_set(
+                                                                                inner2, context_id,
+                                                                                caller_id, job_id,
+                                                                                message_id,
+                                                                            )
+                                                                            .await;
+                                                                        let _ = service
+                                                                            .append_message_logs(
+                                                                                context_id,
+                                                                                caller_id,
+                                                                                message_id,
+                                                                                vec![format!(
+                                                                                    "Requested supervisor job.result for job {}",
+                                                                                    job_id
+                                                                                )],
+                                                                            )
+                                                                            .await;
+                                                                    }
+                                                                    Err(e) => {
+                                                                        let _ = service
+                                                                            .append_message_logs(
+                                                                                context_id,
+                                                                                caller_id,
+                                                                                message_id,
+                                                                                vec![format!(
+                                                                                    "job.result request error for job {}: {}",
+                                                                                    job_id, e
+                                                                                )],
+                                                                            )
+                                                                            .await;
+                                                                    }
+                                                                }
+                                                            } else {
+                                                                let _ = service
+                                                                    .append_message_logs(
+                                                                        context_id,
+                                                                        caller_id,
+                                                                        message_id,
+                                                                        vec![format!(
+                                                                            "No runner with matching script_type found to request job.result for job {}",
+                                                                            job_id
+                                                                        )],
+                                                                    )
+                                                                    .await;
+                                                            }
+                                                        }
+                                                        Err(e) => {
+                                                            let _ = service
+                                                                .append_message_logs(
+                                                                    context_id,
+                                                                    caller_id,
+                                                                    message_id,
+                                                                    vec![format!(
+                                                                        "scan_runners error while requesting job.result for job {}: {}",
+                                                                        job_id, e
+                                                                    )],
+                                                                )
+                                                                .await;
+                                                        }
+                                                    }
+                                                }
+                                                Err(e) => {
+                                                    let _ = service
+                                                        .append_message_logs(
+                                                            context_id,
+                                                            caller_id,
+                                                            message_id,
+                                                            vec![format!(
+                                                                "load_job error while requesting job.result for job {}: {}",
+                                                                job_id, e
+                                                            )],
+                                                        )
+                                                        .await;
+                                                }
+                                            }
+                                        }
+                                        continue;
+                                    }
+                                }
+
+                                // Try job.result: object with success/error or bare string treated as success
+                                if let Some(obj) = res.as_object() {
+                                    if let Some(s) = obj.get("success").and_then(|v| v.as_str()) {
+                                        let mut patch = std::collections::HashMap::new();
+                                        patch.insert("success".to_string(), s.to_string());
+                                        let _ = service
+                                            .update_job_result_merge_unchecked(
+                                                context_id, caller_id, job_id, patch,
+                                            )
+                                            .await;
+                                        let _ = service
+                                            .update_message_status(
+                                                context_id,
+                                                caller_id,
+                                                message_id,
+                                                MessageStatus::Processed,
+                                            )
+                                            .await;
+                                        let _ = service
+                                            .append_message_logs(
+                                                context_id,
+                                                caller_id,
+                                                message_id,
+                                                vec![format!(
+                                                    "Stored supervisor job.result for job {} (success)",
+                                                    job_id
+                                                )],
+                                            )
+                                            .await;
+                                        let _ = service.supcorr_del(inner_id).await;
+                                        continue;
+                                    }
+                                    if let Some(s) = obj.get("error").and_then(|v| v.as_str()) {
+                                        let mut patch = std::collections::HashMap::new();
+                                        patch.insert("error".to_string(), s.to_string());
+                                        let _ = service
+                                            .update_job_result_merge_unchecked(
+                                                context_id, caller_id, job_id, patch,
+                                            )
+                                            .await;
+                                        let _ = service
+                                            .update_message_status(
+                                                context_id,
+                                                caller_id,
+                                                message_id,
+                                                MessageStatus::Processed,
+                                            )
+                                            .await;
+                                        let _ = service
+                                            .append_message_logs(
+                                                context_id,
+                                                caller_id,
+                                                message_id,
+                                                vec![format!(
+                                                    "Stored supervisor job.result for job {} (error)",
+                                                    job_id
+                                                )],
+                                            )
+                                            .await;
+                                        let _ = service.supcorr_del(inner_id).await;
+                                        continue;
+                                    }
+                                } else if let Some(s) = res.as_str() {
+                                    // Bare string => treat as success
+                                    let mut patch = std::collections::HashMap::new();
+                                    patch.insert("success".to_string(), s.to_string());
+                                    let _ = service
+                                        .update_job_result_merge_unchecked(
+                                            context_id, caller_id, job_id, patch,
+                                        )
+                                        .await;
+                                    let _ = service
+                                        .update_message_status(
+                                            context_id,
+                                            caller_id,
+                                            message_id,
+                                            MessageStatus::Processed,
+                                        )
+                                        .await;
+                                    let _ = service
+                                        .append_message_logs(
+                                            context_id,
+                                            caller_id,
+                                            message_id,
+                                            vec![format!(
+                                                "Stored supervisor job.result for job {} (success)",
+                                                job_id
+                                            )],
+                                        )
+                                        .await;
+                                    let _ = service.supcorr_del(inner_id).await;
+                                    continue;
+                                }
+                            }
+
+                            // Unknown/unsupported supervisor reply; keep correlation for later
+                            let _ = service
+                                .append_message_logs(
+                                    context_id,
+                                    caller_id,
+                                    message_id,
+                                    vec![
+                                        "Supervisor reply did not contain recognizable job.run/status/result fields"
+                                            .to_string(),
+                                    ],
+                                )
+                                .await;
+                        }
+                        Ok(None) => {
+                            // No correlation found; ignore or log once
+                        }
+                        Err(e) => {
+                            error!(error=%e, "supcorr_get error");
+                            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+                        }
+                    }
+                }
+                Ok(None) => {
+                    // No message; continue polling
+                    continue;
+                }
+                Err(e) => {
+                    error!(error=%e, "popMessage error");
+                    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+                }
+            }
+        }
+    })
+}

src/rpc.rs

@@ -150,6 +150,8 @@ pub struct RunnerCreate {
     /// The script type this runner executes (used for routing)
     pub script_type: ScriptType,
     pub local: bool,
+    /// Optional secret used for authenticated supervisor calls (if required)
+    pub secret: Option<String>,
 }
 impl RunnerCreate {
     pub fn into_domain(self) -> Runner {
@@ -162,6 +164,7 @@ impl RunnerCreate {
             topic,
             script_type,
             local,
+            secret,
         } = self;
 
         Runner {
@@ -171,6 +174,7 @@ impl RunnerCreate {
             topic,
             script_type,
             local,
+            secret,
             created_at: ts,
             updated_at: ts,
         }

src/service.rs

@@ -1161,6 +1161,37 @@ impl AppService {
     pub async fn scan_runners(&self, context_id: u32) -> Result<Vec<Runner>, BoxError> {
         self.redis.scan_runners(context_id).await
     }
+
+    /// Correlation map: store mapping from inner supervisor JSON-RPC id to context/caller/job/message.
+    pub async fn supcorr_set(
+        &self,
+        inner_id: u64,
+        context_id: u32,
+        caller_id: u32,
+        job_id: u32,
+        message_id: u32,
+    ) -> Result<(), BoxError> {
+        self.redis
+            .supcorr_set(inner_id, context_id, caller_id, job_id, message_id)
+            .await
+            .map_err(Into::into)
+    }
+
+    /// Correlation map: load mapping by inner supervisor JSON-RPC id.
+    pub async fn supcorr_get(
+        &self,
+        inner_id: u64,
+    ) -> Result<Option<(u32, u32, u32, u32)>, BoxError> {
+        self.redis
+            .supcorr_get(inner_id)
+            .await
+            .map_err(Into::into)
+    }
+
+    /// Correlation map: delete mapping by inner supervisor JSON-RPC id.
+    pub async fn supcorr_del(&self, inner_id: u64) -> Result<(), BoxError> {
+        self.redis.supcorr_del(inner_id).await.map_err(Into::into)
+    }
 }
 
 /// Auto-discovery helpers for contexts (wrappers over RedisDriver)

src/storage/redis.rs

@@ -10,7 +10,7 @@ use crate::models::{
     Actor, Context, Flow, FlowStatus, Job, JobStatus, Message, MessageStatus, Runner,
     TransportStatus,
 };
-use tracing::{debug, error, info, trace, warn};
+use tracing::{error, warn};
 
 type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
 
@@ -122,7 +122,7 @@ impl RedisDriver {
             warn!(db=%db, key=%key, error=%e, "DEL before HSET failed");
         }
         // Write all fields
-        let _: usize = cm.hset_multiple(key, &pairs).await.map_err(|e| {
+        let _: () = cm.hset_multiple(key, &pairs).await.map_err(|e| {
             error!(db=%db, key=%key, error=%e, "HSET multiple failed");
             e
         })?;
@@ -323,7 +323,7 @@ impl RedisDriver {
             ("status".to_string(), status_str),
             ("updated_at".to_string(), ts.to_string()),
         ];
-        let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
+        let _: () = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
             error!(db=%db, key=%key, error=%e, "HSET update_job_status failed");
             e
         })?;
@@ -372,7 +372,7 @@ impl RedisDriver {
             ("status".to_string(), status_str),
             ("updated_at".to_string(), ts.to_string()),
         ];
-        let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
+        let _: () = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
             error!(db=%db, key=%key, error=%e, "HSET update_flow_status failed");
             e
         })?;
@@ -400,7 +400,7 @@ impl RedisDriver {
             ("status".to_string(), status_str),
             ("updated_at".to_string(), ts.to_string()),
         ];
-        let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
+        let _: () = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
             error!(db=%db, key=%key, error=%e, "HSET update_message_status failed");
             e
         })?;
@@ -437,7 +437,7 @@ impl RedisDriver {
         let ts = crate::time::current_timestamp();
         pairs.push(("updated_at".to_string(), ts.to_string()));
 
-        let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
+        let _: () = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
             error!(db=%db, key=%key, error=%e, "HSET update_message_transport failed");
             e
         })?;
@@ -473,7 +473,7 @@ impl RedisDriver {
             ("env_vars".to_string(), env_vars_str),
             ("updated_at".to_string(), ts.to_string()),
         ];
-        let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
+        let _: () = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
             error!(db=%db, key=%key, error=%e, "HSET update_flow_env_vars_merge failed");
             e
         })?;
@@ -509,7 +509,7 @@ impl RedisDriver {
             ("result".to_string(), result_str),
             ("updated_at".to_string(), ts.to_string()),
         ];
-        let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
+        let _: () = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
             error!(db=%db, key=%key, error=%e, "HSET update_flow_result_merge failed");
             e
         })?;
@@ -546,7 +546,7 @@ impl RedisDriver {
             ("env_vars".to_string(), env_vars_str),
             ("updated_at".to_string(), ts.to_string()),
         ];
-        let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
+        let _: () = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
             error!(db=%db, key=%key, error=%e, "HSET update_job_env_vars_merge failed");
             e
         })?;
@@ -583,7 +583,7 @@ impl RedisDriver {
             ("result".to_string(), result_str),
             ("updated_at".to_string(), ts.to_string()),
         ];
-        let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
+        let _: () = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
             error!(db=%db, key=%key, error=%e, "HSET update_job_result_merge failed");
             e
         })?;
@@ -601,7 +601,7 @@ impl RedisDriver {
             ("jobs".to_string(), jobs_str),
             ("updated_at".to_string(), ts.to_string()),
         ];
-        let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
+        let _: () = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
             error!(db=%db, key=%key, error=%e, "HSET update_flow_jobs_set failed");
             e
         })?;
@@ -635,7 +635,7 @@ impl RedisDriver {
             ("logs".to_string(), logs_str),
             ("updated_at".to_string(), ts.to_string()),
         ];
-        let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
+        let _: () = cm.hset_multiple(&key, &pairs).await.map_err(|e| {
             error!(db=%db, key=%key, error=%e, "HSET append_message_logs failed");
             e
         })?;
@@ -751,4 +751,80 @@ impl RedisDriver {
         out.sort_unstable();
         Ok(out)
     }
+
+    // -----------------------------
+    // Supervisor correlation mapping (DB 0)
+    // Key: "supcorr:{inner_id_decimal}"
+    // Value: JSON {"context_id":u32,"caller_id":u32,"job_id":u32,"message_id":u32}
+    // TTL: 1 hour to avoid leaks in case of crashes
+    pub async fn supcorr_set(
+        &self,
+        inner_id: u64,
+        context_id: u32,
+        caller_id: u32,
+        job_id: u32,
+        message_id: u32,
+    ) -> Result<()> {
+        let mut cm = self.manager_for_db(0).await?;
+        let key = format!("supcorr:{}", inner_id);
+        let val = serde_json::json!({
+            "context_id": context_id,
+            "caller_id": caller_id,
+            "job_id": job_id,
+            "message_id": message_id,
+        })
+        .to_string();
+        // SET key val EX 3600
+        let _: () = redis::cmd("SET")
+            .arg(&key)
+            .arg(&val)
+            .arg("EX")
+            .arg(3600)
+            .query_async(&mut cm)
+            .await
+            .map_err(|e| {
+                error!(db=0, key=%key, error=%e, "SET supcorr_set failed");
+                e
+            })?;
+        Ok(())
+    }
+
+    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")
+            .arg(&key)
+            .query_async(&mut cm)
+            .await
+            .map_err(|e| {
+                error!(db=0, key=%key, error=%e, "GET supcorr_get failed");
+                e
+            })?;
+        if let Some(s) = res {
+            let v: Value = serde_json::from_str(&s)?;
+            let ctx = v.get("context_id").and_then(|x| x.as_u64()).unwrap_or(0) as u32;
+            let caller = v.get("caller_id").and_then(|x| x.as_u64()).unwrap_or(0) as u32;
+            let job = v.get("job_id").and_then(|x| x.as_u64()).unwrap_or(0) as u32;
+            let msg = v.get("message_id").and_then(|x| x.as_u64()).unwrap_or(0) as u32;
+            return Ok(Some((ctx, caller, job, msg)));
+        }
+        Ok(None)
+    }
+
+    pub async fn supcorr_del(&self, inner_id: u64) -> Result<()> {
+        let mut cm = self.manager_for_db(0).await?;
+        let key = format!("supcorr:{}", inner_id);
+        let _: i64 = redis::cmd("DEL")
+            .arg(&key)
+            .query_async(&mut cm)
+            .await
+            .map_err(|e| {
+                error!(db=0, key=%key, error=%e, "DEL supcorr_del failed");
+                e
+            })?;
+        Ok(())
+    }
 }