145 lines
5.9 KiB
Python
145 lines
5.9 KiB
Python
"""CLI for HopeMesh registry and governance snapshots."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
|
|
from .contracts import GraphOfContracts, GovernanceLedger
|
|
|
|
|
|
DEFAULT_STORE = Path(".hopemesh/state.json")
|
|
|
|
|
|
def _load_state(store: Path) -> Dict[str, Any]:
|
|
if not store.exists():
|
|
return {"registry": {"records": [], "links": []}, "ledger": {"entries": []}}
|
|
return json.loads(store.read_text(encoding="utf-8"))
|
|
|
|
|
|
def _save_state(store: Path, state: Dict[str, Any]) -> None:
|
|
store.parent.mkdir(parents=True, exist_ok=True)
|
|
store.write_text(json.dumps(state, indent=2, sort_keys=True), encoding="utf-8")
|
|
|
|
|
|
def _graph_from_state(state: Dict[str, Any]) -> GraphOfContracts:
|
|
return GraphOfContracts.from_snapshot(state.get("registry", {}))
|
|
|
|
|
|
def _ledger_from_state(state: Dict[str, Any]) -> GovernanceLedger:
|
|
return GovernanceLedger.from_snapshot(state.get("ledger", {}))
|
|
|
|
|
|
def _store_snapshot(store: Path, graph: GraphOfContracts, ledger: GovernanceLedger) -> None:
|
|
_save_state(store, {"registry": graph.snapshot(), "ledger": ledger.snapshot()})
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(prog="hopemesh", description="HopeMesh 2.0 registry and governance CLI")
|
|
parser.add_argument("--version", dest="show_version", action="store_true", help="Show version")
|
|
subparsers = parser.add_subparsers(dest="cmd")
|
|
|
|
registry = subparsers.add_parser("registry", help="Manage the Graph-of-Contracts")
|
|
registry_sub = registry.add_subparsers(dest="registry_cmd", required=True)
|
|
|
|
registry_init = registry_sub.add_parser("init", help="Create an empty registry store")
|
|
registry_init.add_argument("--store", type=Path, default=DEFAULT_STORE)
|
|
|
|
registry_schema = registry_sub.add_parser("add-schema", help="Register a schema contract")
|
|
registry_schema.add_argument("--store", type=Path, default=DEFAULT_STORE)
|
|
registry_schema.add_argument("--name", required=True)
|
|
registry_schema.add_argument("--schema-version", required=True)
|
|
registry_schema.add_argument("--definition-json", required=True)
|
|
|
|
registry_adapter = registry_sub.add_parser("add-adapter", help="Register an adapter contract")
|
|
registry_adapter.add_argument("--store", type=Path, default=DEFAULT_STORE)
|
|
registry_adapter.add_argument("--name", required=True)
|
|
registry_adapter.add_argument("--adapter-version", required=True)
|
|
registry_adapter.add_argument("--definition-json", required=True)
|
|
|
|
registry_link = registry_sub.add_parser("link", help="Link two contracts")
|
|
registry_link.add_argument("--store", type=Path, default=DEFAULT_STORE)
|
|
registry_link.add_argument("--source-id", required=True)
|
|
registry_link.add_argument("--target-id", required=True)
|
|
registry_link.add_argument("--relation", default="implements")
|
|
|
|
registry_show = registry_sub.add_parser("snapshot", help="Print the registry snapshot")
|
|
registry_show.add_argument("--store", type=Path, default=DEFAULT_STORE)
|
|
|
|
ledger = subparsers.add_parser("ledger", help="Manage the governance ledger")
|
|
ledger_sub = ledger.add_subparsers(dest="ledger_cmd", required=True)
|
|
|
|
ledger_append = ledger_sub.add_parser("append", help="Append a governance event")
|
|
ledger_append.add_argument("--store", type=Path, default=DEFAULT_STORE)
|
|
ledger_append.add_argument("--event-type", required=True)
|
|
ledger_append.add_argument("--payload-json", required=True)
|
|
ledger_append.add_argument("--actor", default="system")
|
|
|
|
ledger_verify = ledger_sub.add_parser("verify", help="Verify ledger integrity")
|
|
ledger_verify.add_argument("--store", type=Path, default=DEFAULT_STORE)
|
|
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
|
|
if args.show_version:
|
|
print("hopemesh-sdk 0.1.0")
|
|
return 0
|
|
|
|
if not args.cmd:
|
|
parser.print_help()
|
|
return 0
|
|
|
|
state = _load_state(args.store)
|
|
graph = _graph_from_state(state)
|
|
ledger = _ledger_from_state(state)
|
|
|
|
if args.cmd == "registry":
|
|
if args.registry_cmd == "init":
|
|
_store_snapshot(args.store, graph, ledger)
|
|
print(str(args.store))
|
|
return 0
|
|
if args.registry_cmd in {"add-schema", "add-adapter"}:
|
|
definition = json.loads(args.definition_json)
|
|
if args.registry_cmd == "add-schema":
|
|
record = graph.register_schema(args.name, args.schema_version, definition)
|
|
else:
|
|
record = graph.register_adapter(args.name, args.adapter_version, definition)
|
|
ledger.append("registry.register", {"contract_id": record.contract_id, "kind": record.kind})
|
|
_store_snapshot(args.store, graph, ledger)
|
|
print(record.contract_id)
|
|
return 0
|
|
if args.registry_cmd == "link":
|
|
link = graph.link(args.source_id, args.target_id, args.relation)
|
|
ledger.append("registry.link", link.to_dict())
|
|
_store_snapshot(args.store, graph, ledger)
|
|
print(json.dumps(link.to_dict(), sort_keys=True))
|
|
return 0
|
|
if args.registry_cmd == "snapshot":
|
|
print(json.dumps(graph.snapshot(), indent=2, sort_keys=True))
|
|
return 0
|
|
|
|
if args.cmd == "ledger":
|
|
if args.ledger_cmd == "append":
|
|
payload = json.loads(args.payload_json)
|
|
entry = ledger.append(args.event_type, payload, actor=args.actor)
|
|
_store_snapshot(args.store, graph, ledger)
|
|
print(json.dumps(entry.to_dict(), indent=2, sort_keys=True))
|
|
return 0
|
|
if args.ledger_cmd == "verify":
|
|
issues = ledger.verify()
|
|
print(json.dumps({"ok": not issues, "issues": issues}, indent=2, sort_keys=True))
|
|
return 0
|
|
|
|
parser.error("unknown command")
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|