54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
#!/usr/bin/env python3
|
|
"""DeltaTrace CLI - lightweight MVP launcher"""
|
|
|
|
import argparse
|
|
import json
|
|
from typing import List, Dict
|
|
|
|
from .trace import LocalEvent, PlanDelta, Metadata
|
|
from .adapters import FixFeedAdapter, ExchangeGatewaySandbox
|
|
from .replay import deterministic_replay
|
|
|
|
|
|
def _toy_delta_stream() -> List[Dict[str, object]]:
|
|
# A tiny toy delta stream for demo purposes
|
|
return [
|
|
{
|
|
"version": "0.1",
|
|
"timestamp": 1.0,
|
|
"nonce": "n1",
|
|
"author": "tester",
|
|
"contract_id": "CON-1",
|
|
"delta_id": "D-1",
|
|
"payload": {"decision": "BUY"},
|
|
"signature": "sig1",
|
|
}
|
|
]
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(prog="deltatrace", description="DeltaTrace MVP CLI")
|
|
sub = parser.add_subparsers(dest="cmd", required=False)
|
|
sub.add_parser("demo", help="Run a quick demo with toy delta stream and replay")
|
|
sub.add_parser("replay", help="Run deterministic replay on a provided delta stream (JSON file)")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.cmd == "demo" or args.cmd is None:
|
|
# Quick demo: create toy events and run a tiny replay
|
|
delta = _toy_delta_stream()
|
|
# simple baseline log mirroring delta ids
|
|
event_log = [{"delta_id": d["delta_id"]} for d in delta]
|
|
result = deterministic_replay(delta, event_log)
|
|
print(json.dumps(result, indent=2))
|
|
return
|
|
|
|
if args.cmd == "replay":
|
|
# In a full implementation, we'd parse a JSON file from argv; here we provide a friendly fallback
|
|
print("Replay command not wired to a file in this MVP stub.")
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|