feat: add --transport-dir file-backed mode to run-local-flow

README.md

@@ -176,6 +176,14 @@ PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=src python -m aethermesh_core.cli run-local
 
 The flow writes `dispatch-message-log.json`, a merged run-level `flow-message-log.json`, one shared `ledger.json`, deterministic per-job `receipts.json`, per-node state files under `node-state/`, and per-node replay/output logs under `worker-message-logs/`. The flow log is a deterministic local audit transcript: dispatch messages appear once, followed by worker-emitted result/contribution messages in stable available-node order. Receipts compactly tie each processed assignment to its result, validation decision, contribution record, credited units, and output summary. Offline manifest nodes stay visible in the JSON summary but are not processed as workers.
 
+To route the same local flow through file-backed per-node inboxes before worker processing, opt in with `--transport-dir`:
+
+```bash
+PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=src python -m aethermesh_core.cli run-local-flow --manifest examples/local-batch.json --output-dir /tmp/aethermesh-local-flow --transport-dir /tmp/aethermesh-local-transport
+```
+
+This still writes the dispatch log under `--output-dir`, then materializes addressed dispatch messages to version 1 inbox files under the transport directory and has each available worker consume its own local inbox. The JSON result includes deterministic transport metadata (`transport_dir`, `transport_inbox_count`, and `transport_inbox_paths`). This is local file-backed transport only, not real networking or peer discovery.
+
 Audit an existing local flow artifact directory without modifying it:
 
 ```bash

src/aethermesh_core/cli.py

@@ -28,8 +28,10 @@
 )
 from aethermesh_core.local_transport import (
     LocalTransportError,
+    local_inbox_path,
     load_local_inbox,
     materialize_local_inboxes,
+    write_local_inbox,
 )
 from aethermesh_core.message_bus import LocalMessageBus
 from aethermesh_core.message_log import (
@@ -186,6 +188,11 @@ def build_parser() -> argparse.ArgumentParser:
         required=True,
         help="Directory for deterministic local flow artifacts.",
     )
+    local_flow.add_argument(
+        "--transport-dir",
+        default=None,
+        help="Opt in to file-backed local transport inboxes for worker processing.",
+    )
 
     audit_local = subcommands.add_parser(
         "audit-local-flow",
@@ -500,7 +507,9 @@ def summarize_peers(message_log_path: str) -> dict[str, object]:
     return peer_summary_document(message_log_path)
 
 
-def run_local_flow(manifest_path: str, output_dir: str) -> dict[str, object]:
+def run_local_flow(
+    manifest_path: str, output_dir: str, transport_dir: str | None = None
+) -> dict[str, object]:
     """Run dispatch and all available local worker inboxes as one local flow."""
 
     batch = load_job_manifest(manifest_path)
@@ -535,6 +544,23 @@ def run_local_flow(manifest_path: str, output_dir: str) -> dict[str, object]:
     dispatch_payload = dispatch_local_batch_command(
         manifest_path, str(dispatch_message_log_path)
     )
+    transport_payload: dict[str, object] | None = None
+    if transport_dir is not None:
+        transport_payload = materialize_local_inboxes(
+            message_log_path=dispatch_message_log_path,
+            transport_dir=transport_dir,
+        )
+        materialized_inbox_paths = transport_payload.get("inbox_paths", {})
+        if not isinstance(materialized_inbox_paths, dict):
+            raise ValueError("materialize-local-inboxes returned invalid inbox paths")
+        for node_id in available_node_ids:
+            if node_id not in materialized_inbox_paths:
+                write_local_inbox(
+                    transport_dir=transport_dir,
+                    node_id=node_id,
+                    messages=[],
+                    source_message_log_path=dispatch_message_log_path,
+                )
 
     per_node_results: list[dict[str, object]] = []
     processed_assignments = []
@@ -547,7 +573,10 @@ def run_local_flow(manifest_path: str, output_dir: str) -> dict[str, object]:
         node_payload, inbox_result = _process_local_inbox(
             InboxReplayRequest(
                 node_id=node_id,
-                message_log_path=str(dispatch_message_log_path),
+                message_log_path=(
+                    str(dispatch_message_log_path) if transport_dir is None else None
+                ),
+                transport_dir=transport_dir,
                 ledger_path=str(ledger_path),
                 output_message_log_path=str(worker_message_log_path),
                 node_state_path=str(node_state_path),
@@ -609,7 +638,7 @@ def run_local_flow(manifest_path: str, output_dir: str) -> dict[str, object]:
         existing_document=existing_receipt_document,
     )
     write_receipt_document(receipts_path, receipt_document)
-    return {
+    result: dict[str, object] = {
         "command": "run-local-flow",
         "manifest_path": manifest_path,
         "output_dir": output_dir,
@@ -636,6 +665,19 @@ def run_local_flow(manifest_path: str, output_dir: str) -> dict[str, object]:
         "dispatch_summary": dispatch_payload,
         "node_results": per_node_results,
     }
+    if transport_payload is not None:
+        transport_inbox_paths = transport_payload.get("inbox_paths", {})
+        if not isinstance(transport_inbox_paths, dict):
+            raise ValueError("materialize-local-inboxes returned invalid inbox paths")
+        transport_dir_value = str(transport_dir)
+        result["transport_dir"] = str(transport_dir)
+        result["transport_inbox_count"] = transport_payload["inbox_count"]
+        result["transport_inbox_paths"] = {
+            node_id: str(local_inbox_path(transport_dir_value, node_id))
+            for node_id in sorted(available_node_ids)
+            if node_id in transport_inbox_paths
+        }
+    return result
 
 
 def _require_int_result_field(result: dict[str, object], field_name: str) -> int:
@@ -912,12 +954,13 @@ def main(argv: Sequence[str] | None = None) -> int:
 
     if args.command == "run-local-flow":
         try:
-            payload = run_local_flow(args.manifest, args.output_dir)
+            payload = run_local_flow(args.manifest, args.output_dir, args.transport_dir)
         except ManifestError as exc:
             print(f"error: {exc}", file=sys.stderr)
             return 1
         except (
             MessageLogPersistenceError,
+            LocalTransportError,
             LedgerPersistenceError,
             NodeStatePersistenceError,
             ValueError,

tests/test_cli.py

@@ -184,7 +184,7 @@ def test_build_parser_help_snapshots_stay_stable(self) -> None:
                 "run-local-batch": "usage: aethermesh-core run-local-batch [-h] --manifest MANIFEST [--ledger-path LEDGER_PATH] [--message-log-path MESSAGE_LOG_PATH] options: -h, --help show this help message and exit --manifest MANIFEST Path to a version 1 local job-batch JSON manifest. --ledger-path LEDGER_PATH Opt in to JSON-file-backed local contribution ledger persistence. --message-log-path MESSAGE_LOG_PATH Opt in to overwriting a local JSON audit log of deterministic mesh messages.",
                 "dispatch-local-batch": "usage: aethermesh-core dispatch-local-batch [-h] --manifest MANIFEST --message-log-path MESSAGE_LOG_PATH options: -h, --help show this help message and exit --manifest MANIFEST Path to a version 1 local job-batch JSON manifest. --message-log-path MESSAGE_LOG_PATH Path to write the version 1 assignment-only local message log.",
                 "dispatch-peer-batch": "usage: aethermesh-core dispatch-peer-batch [-h] --peer-log-path PEER_LOG_PATH --manifest MANIFEST --message-log-path MESSAGE_LOG_PATH options: -h, --help show this help message and exit --peer-log-path PEER_LOG_PATH Path to an existing version 1 local heartbeat message log. --manifest MANIFEST Path to a version 1 manifest whose jobs should be dispatched. --message-log-path MESSAGE_LOG_PATH Path to write the version 1 assignment-only local message log.",
-                "run-local-flow": "usage: aethermesh-core run-local-flow [-h] --manifest MANIFEST --output-dir OUTPUT_DIR options: -h, --help show this help message and exit --manifest MANIFEST Path to a version 1 local job-batch JSON manifest. --output-dir OUTPUT_DIR Directory for deterministic local flow artifacts.",
+                "run-local-flow": "usage: aethermesh-core run-local-flow [-h] --manifest MANIFEST --output-dir OUTPUT_DIR [--transport-dir TRANSPORT_DIR] options: -h, --help show this help message and exit --manifest MANIFEST Path to a version 1 local job-batch JSON manifest. --output-dir OUTPUT_DIR Directory for deterministic local flow artifacts. --transport-dir TRANSPORT_DIR Opt in to file-backed local transport inboxes for worker processing.",
                 "audit-local-flow": "usage: aethermesh-core audit-local-flow [-h] --output-dir OUTPUT_DIR options: -h, --help show this help message and exit --output-dir OUTPUT_DIR Directory containing deterministic local flow artifacts to audit.",
                 "ledger-summary": "usage: aethermesh-core ledger-summary [-h] --ledger-path LEDGER_PATH options: -h, --help show this help message and exit --ledger-path LEDGER_PATH Path to an existing version 1 local contribution ledger JSON file.",
                 "peer-summary": "usage: aethermesh-core peer-summary [-h] --message-log-path MESSAGE_LOG_PATH options: -h, --help show this help message and exit --message-log-path MESSAGE_LOG_PATH Path to an existing version 1 local message log.",
@@ -2574,6 +2574,9 @@ def test_run_local_flow_happy_path_with_example_manifest(self) -> None:
             payload["flow_message_log_path"], str(output_dir / "flow-message-log.json")
         )
         self.assertEqual(payload["receipts_path"], str(output_dir / "receipts.json"))
+        self.assertNotIn("transport_dir", payload)
+        self.assertNotIn("transport_inbox_count", payload)
+        self.assertNotIn("transport_inbox_paths", payload)
         self.assertEqual(payload["receipt_count"], 6)
         self.assertEqual(payload["flow_message_count"], 20)
         self.assertEqual(payload["flow_emitted_message_count"], 12)

tests/test_cli_transport.py

@@ -9,6 +9,191 @@
 
 
 class LocalTransportCliTests(unittest.TestCase):
+    def test_run_local_flow_transport_handles_available_node_without_assignments(
+        self,
+    ) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            manifest_path = Path(temp_dir) / "local-batch.json"
+            output_dir = Path(temp_dir) / "flow"
+            transport_dir = Path(temp_dir) / "transport"
+            manifest_path.write_text(
+                json.dumps(
+                    {
+                        "version": 1,
+                        "nodes": ["local-node-a", "local-node-b"],
+                        "jobs": [
+                            {
+                                "job_id": "echo-1",
+                                "job_type": "echo",
+                                "payload": {"message": "one"},
+                            }
+                        ],
+                    }
+                ),
+                encoding="utf-8",
+            )
+
+            stdout = io.StringIO()
+            with contextlib.redirect_stdout(stdout):
+                exit_code = main(
+                    [
+                        "run-local-flow",
+                        "--manifest",
+                        str(manifest_path),
+                        "--output-dir",
+                        str(output_dir),
+                        "--transport-dir",
+                        str(transport_dir),
+                    ]
+                )
+
+            payload = json.loads(stdout.getvalue())
+            node_b_inbox = json.loads(
+                (transport_dir / "inboxes" / "local-node-b.json").read_text(
+                    encoding="utf-8"
+                )
+            )
+
+        self.assertEqual(exit_code, 0)
+        self.assertEqual(
+            payload["available_node_ids"], ["local-node-a", "local-node-b"]
+        )
+        self.assertEqual(
+            payload["processed_node_ids"], ["local-node-a", "local-node-b"]
+        )
+        self.assertEqual(payload["processed_assignment_count"], 1)
+        self.assertEqual(payload["transport_inbox_count"], 1)
+        self.assertEqual(node_b_inbox["node_id"], "local-node-b")
+        self.assertEqual(node_b_inbox["messages"], [])
+
+    def test_run_local_flow_uses_transport_inboxes_and_audits(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            manifest_path = Path(temp_dir) / "local-batch.json"
+            output_dir = Path(temp_dir) / "flow"
+            transport_dir = Path(temp_dir) / "transport"
+            manifest_path.write_text(
+                json.dumps(
+                    {
+                        "version": 1,
+                        "nodes": [
+                            {"node_id": "local-node-a", "status": "available"},
+                            {"node_id": "local-node-b", "status": "offline"},
+                            {"node_id": "local-node-c", "status": "available"},
+                        ],
+                        "jobs": [
+                            {
+                                "job_id": "echo-1",
+                                "job_type": "echo",
+                                "payload": {"message": "one"},
+                            },
+                            {
+                                "job_id": "echo-2",
+                                "job_type": "echo",
+                                "payload": {"message": "two"},
+                            },
+                            {
+                                "job_id": "echo-3",
+                                "job_type": "echo",
+                                "payload": {"message": "three"},
+                            },
+                        ],
+                    }
+                ),
+                encoding="utf-8",
+            )
+
+            first_stdout = io.StringIO()
+            with contextlib.redirect_stdout(first_stdout):
+                first_exit = main(
+                    [
+                        "run-local-flow",
+                        "--manifest",
+                        str(manifest_path),
+                        "--output-dir",
+                        str(output_dir),
+                        "--transport-dir",
+                        str(transport_dir),
+                    ]
+                )
+            audit_stdout = io.StringIO()
+            with contextlib.redirect_stdout(audit_stdout):
+                audit_exit = main(["audit-local-flow", "--output-dir", str(output_dir)])
+            second_stdout = io.StringIO()
+            with contextlib.redirect_stdout(second_stdout):
+                second_exit = main(
+                    [
+                        "run-local-flow",
+                        "--manifest",
+                        str(manifest_path),
+                        "--output-dir",
+                        str(output_dir),
+                        "--transport-dir",
+                        str(transport_dir),
+                    ]
+                )
+
+            first_payload = json.loads(first_stdout.getvalue())
+            audit_payload = json.loads(audit_stdout.getvalue())
+            second_payload = json.loads(second_stdout.getvalue())
+            ledger = json.loads(
+                (output_dir / "ledger.json").read_text(encoding="utf-8")
+            )
+            node_a_inbox = json.loads(
+                (transport_dir / "inboxes" / "local-node-a.json").read_text(
+                    encoding="utf-8"
+                )
+            )
+            node_c_inbox = json.loads(
+                (transport_dir / "inboxes" / "local-node-c.json").read_text(
+                    encoding="utf-8"
+                )
+            )
+            offline_inbox_exists = (
+                transport_dir / "inboxes" / "local-node-b.json"
+            ).exists()
+            offline_worker_log_exists = (
+                output_dir / "worker-message-logs" / "local-node-b.json"
+            ).exists()
+
+        self.assertEqual(first_exit, 0)
+        self.assertEqual(audit_exit, 0)
+        self.assertEqual(second_exit, 0)
+        self.assertEqual(first_payload["transport_dir"], str(transport_dir))
+        self.assertEqual(first_payload["transport_inbox_count"], 2)
+        self.assertEqual(
+            first_payload["transport_inbox_paths"],
+            {
+                "local-node-a": str(transport_dir / "inboxes" / "local-node-a.json"),
+                "local-node-c": str(transport_dir / "inboxes" / "local-node-c.json"),
+            },
+        )
+        self.assertEqual(
+            first_payload["available_node_ids"], ["local-node-a", "local-node-c"]
+        )
+        self.assertEqual(first_payload["offline_node_ids"], ["local-node-b"])
+        self.assertEqual(
+            first_payload["processed_node_ids"], ["local-node-a", "local-node-c"]
+        )
+        self.assertEqual(first_payload["processed_assignment_count"], 3)
+        self.assertEqual(second_payload["processed_assignment_count"], 0)
+        self.assertEqual(second_payload["skipped_processed_assignment_count"], 3)
+        self.assertEqual(second_payload["transport_inbox_count"], 2)
+        self.assertEqual(audit_payload["ok"], True)
+        self.assertEqual(audit_payload["counts"]["ledger_records"], 3)
+        self.assertEqual(len(ledger["records"]), 3)
+        self.assertEqual(node_a_inbox["node_id"], "local-node-a")
+        self.assertEqual(node_c_inbox["node_id"], "local-node-c")
+        self.assertEqual(
+            [message["recipient_node_id"] for message in node_a_inbox["messages"]],
+            ["local-node-a", "local-node-a"],
+        )
+        self.assertEqual(
+            [message["recipient_node_id"] for message in node_c_inbox["messages"]],
+            ["local-node-c"],
+        )
+        self.assertFalse(offline_inbox_exists)
+        self.assertFalse(offline_worker_log_exists)
+
     def test_materialize_local_inboxes_and_process_transport_inbox(self) -> None:
         with tempfile.TemporaryDirectory() as temp_dir:
             manifest_path = Path(temp_dir) / "local-batch.json"