Skip to content

Commit 4d98407

Browse files
committed
chore: implement next AetherMesh step
1 parent 37a717c commit 4d98407

7 files changed

Lines changed: 452 additions & 10 deletions

File tree

‎README.md‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,6 +134,14 @@ PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=src python -m aethermesh_core.cli dispatch-
134134

135135
The dispatch log contains deterministic `node_heartbeat` messages for available nodes and `job_assigned` messages for scheduled jobs. Heartbeat payloads include each available node's manifest capabilities. Use `process-local-inbox` to consume the dispatch log later for one node and record validation-gated contribution.
136136

137+
Dispatch jobs to peers discovered from an existing local heartbeat message log instead of from manifest `nodes`:
138+
139+
```bash
140+
PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=src python -m aethermesh_core.cli dispatch-local-peers --peer-message-log-path ./local-peers.json --manifest examples/local-batch.json --message-log-path ./local-peer-dispatch.json
141+
```
142+
143+
This remains local message-log discovery only. The command reads jobs from the manifest, builds the scheduling roster from latest `node_heartbeat` messages, ignores offline peers for assignment, and writes the same assignment-only dispatch log shape used by local inbox processing.
144+
137145
To test the file-backed local transport seam, materialize addressed dispatch messages into per-node inbox files and have a node consume only its own inbox:
138146

139147
```bash

‎src/aethermesh_core/cli.py‎

Lines changed: 74 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,11 @@
1414
from aethermesh_core.dispatch import dispatch_local_batch
1515
from aethermesh_core.flow_audit import FlowAuditError, audit_local_flow
1616
from aethermesh_core.identity import IdentityPersistenceError, load_or_create_identity
17-
from aethermesh_core.job_manifest import ManifestError, load_job_manifest
17+
from aethermesh_core.job_manifest import (
18+
ManifestError,
19+
load_job_manifest,
20+
load_job_manifest_jobs,
21+
)
1822
from aethermesh_core.ledger import (
1923
ContributionLedger,
2024
LedgerPersistenceError,
@@ -48,7 +52,11 @@
4852
load_node_processing_state,
4953
save_node_processing_state,
5054
)
51-
from aethermesh_core.peer_registry import PeerRegistryError, peer_summary_document
55+
from aethermesh_core.peer_registry import (
56+
PeerRegistryError,
57+
peer_summary_document,
58+
scheduled_nodes_from_heartbeat_log,
59+
)
5260
from aethermesh_core.receipts import (
5361
ReceiptPersistenceError,
5462
build_receipt_document,
@@ -144,6 +152,26 @@ def build_parser() -> argparse.ArgumentParser:
144152
help="Path to write the version 1 assignment-only local message log.",
145153
)
146154

155+
peer_dispatch = subcommands.add_parser(
156+
"dispatch-local-peers",
157+
help="Write local dispatch messages using heartbeat-discovered peers.",
158+
)
159+
peer_dispatch.add_argument(
160+
"--peer-message-log-path",
161+
required=True,
162+
help="Path to an existing version 1 message log with node heartbeats.",
163+
)
164+
peer_dispatch.add_argument(
165+
"--manifest",
166+
required=True,
167+
help="Path to a version 1 local job manifest whose jobs should be dispatched.",
168+
)
169+
peer_dispatch.add_argument(
170+
"--message-log-path",
171+
required=True,
172+
help="Path to write the version 1 assignment-only local message log.",
173+
)
174+
147175
local_flow = subcommands.add_parser(
148176
"run-local-flow",
149177
help="Run dispatch plus all available local worker inboxes for a manifest.",
@@ -431,6 +459,36 @@ def dispatch_local_batch_command(
431459
return dispatch.to_dict()
432460

433461

462+
def dispatch_local_peers_command(
463+
peer_message_log_path: str,
464+
manifest_path: str,
465+
message_log_path: str,
466+
) -> dict[str, object]:
467+
"""Dispatch manifest jobs to nodes discovered from heartbeat messages."""
468+
469+
nodes = scheduled_nodes_from_heartbeat_log(peer_message_log_path)
470+
jobs = load_job_manifest_jobs(manifest_path)
471+
dispatch = dispatch_local_batch(
472+
manifest_path=manifest_path,
473+
message_log_path=message_log_path,
474+
nodes=nodes,
475+
jobs=jobs,
476+
)
477+
message_log_document = build_dispatch_message_log_document(
478+
messages=dispatch.messages,
479+
jobs=dispatch.jobs,
480+
nodes=dispatch.nodes,
481+
assignments=dispatch.assignments,
482+
manifest_path=manifest_path,
483+
source="dispatch-local-peers",
484+
)
485+
write_message_log(message_log_path, message_log_document)
486+
payload = dispatch.to_dict()
487+
payload["command"] = "dispatch-local-peers"
488+
payload["peer_message_log_path"] = peer_message_log_path
489+
return payload
490+
491+
434492
def summarize_ledger(ledger_path: str) -> dict[str, object]:
435493
"""Load an existing ledger and return read-only aggregate totals."""
436494

@@ -837,6 +895,20 @@ def main(argv: Sequence[str] | None = None) -> int:
837895
print(json.dumps(payload, sort_keys=True))
838896
return 0
839897

898+
if args.command == "dispatch-local-peers":
899+
try:
900+
payload = dispatch_local_peers_command(
901+
args.peer_message_log_path, args.manifest, args.message_log_path
902+
)
903+
except (ManifestError, PeerRegistryError, MessageLogPersistenceError) as exc:
904+
print(f"error: {exc}", file=sys.stderr)
905+
return 1
906+
except ValueError as exc:
907+
print(f"error: local peer dispatch failed: {exc}", file=sys.stderr)
908+
return 1
909+
print(json.dumps(payload, sort_keys=True))
910+
return 0
911+
840912
if args.command == "run-local-flow":
841913
try:
842914
payload = run_local_flow(args.manifest, args.output_dir)

‎src/aethermesh_core/job_manifest.py‎

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,21 @@ def node_ids(self) -> list[str]:
3636
def load_job_manifest(path: str | Path) -> LocalJobBatch:
3737
"""Load and validate a version 1 local job-batch manifest."""
3838

39+
document = _load_manifest_document(path)
40+
return LocalJobBatch(
41+
nodes=_parse_nodes(document.get("nodes")),
42+
jobs=_parse_jobs(document.get("jobs")),
43+
)
44+
45+
46+
def load_job_manifest_jobs(path: str | Path) -> list[Job]:
47+
"""Load only jobs from a version 1 manifest, ignoring any node roster."""
48+
49+
document = _load_manifest_document(path)
50+
return _parse_jobs(document.get("jobs"))
51+
52+
53+
def _load_manifest_document(path: str | Path) -> dict[str, Any]:
3954
manifest_path = Path(path)
4055
try:
4156
raw_manifest = manifest_path.read_text(encoding="utf-8")
@@ -55,11 +70,7 @@ def load_job_manifest(path: str | Path) -> LocalJobBatch:
5570
raise ManifestError("manifest version must be integer 1")
5671
if version != 1:
5772
raise ManifestError(f"unsupported manifest version: {version}")
58-
59-
return LocalJobBatch(
60-
nodes=_parse_nodes(document.get("nodes")),
61-
jobs=_parse_jobs(document.get("jobs")),
62-
)
73+
return document
6374

6475

6576
def _parse_nodes(value: Any) -> list[ScheduledNode]:

‎src/aethermesh_core/message_log.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -88,13 +88,14 @@ def build_dispatch_message_log_document(
8888
nodes: list[ScheduledNode],
8989
assignments: list[JobAssignment],
9090
manifest_path: str | Path,
91+
source: str = "dispatch-local-batch",
9192
) -> dict[str, Any]:
9293
"""Build a deterministic version 1 assignment-only dispatch document."""
9394

9495
return {
9596
"version": 1,
9697
"metadata": {
97-
"source": "dispatch-local-batch",
98+
"source": source,
9899
"manifest_path": str(manifest_path),
99100
"message_count": len(messages),
100101
"node_count": len(nodes),

‎src/aethermesh_core/peer_registry.py‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
load_message_log_messages,
1212
)
1313
from aethermesh_core.messages import MeshMessage
14+
from aethermesh_core.scheduler import NodeStatus, ScheduledNode
1415

1516

1617
class PeerRegistryError(ValueError):
@@ -81,6 +82,50 @@ def peer_summary_document(message_log_path: str | Path) -> dict[str, object]:
8182
}
8283

8384

85+
def scheduled_nodes_from_heartbeat_log(
86+
message_log_path: str | Path,
87+
) -> list[ScheduledNode]:
88+
"""Build deterministic scheduler nodes from latest local heartbeats.
89+
90+
The input message log is read-only. Returned nodes are sorted by node id so a
91+
repeated dispatch from the same heartbeat log produces stable assignment
92+
ordering independent of the original message ordering.
93+
"""
94+
95+
document = peer_summary_document(message_log_path)
96+
raw_peers = document["peers"]
97+
if not isinstance(raw_peers, list):
98+
raise PeerRegistryError("peer summary peers must be a list")
99+
100+
nodes: list[ScheduledNode] = []
101+
for peer in raw_peers:
102+
if not isinstance(peer, dict):
103+
raise PeerRegistryError("peer summary entries must be objects")
104+
node_id = peer.get("node_id")
105+
raw_status = peer.get("status")
106+
raw_capabilities = peer.get("capabilities")
107+
if not isinstance(node_id, str) or node_id == "":
108+
raise PeerRegistryError("peer node_id must be a non-empty string")
109+
if not isinstance(raw_status, str):
110+
raise PeerRegistryError(f"peer {node_id} status must be a string")
111+
try:
112+
status = NodeStatus(raw_status)
113+
except ValueError as exc:
114+
supported = ", ".join(status.value for status in NodeStatus)
115+
raise PeerRegistryError(
116+
f"peer {node_id} status must be one of: {supported}"
117+
) from exc
118+
capabilities = _parse_capabilities(raw_capabilities, f"peer {node_id}")
119+
nodes.append(
120+
ScheduledNode(
121+
node_id=node_id,
122+
status=status,
123+
capabilities=tuple(capabilities),
124+
)
125+
)
126+
return nodes
127+
128+
84129
def _parse_heartbeat(message: MeshMessage) -> PeerSummary:
85130
payload = message.payload
86131
node_id = payload.get("node_id")

0 commit comments

Comments
 (0)