267 lines
7.9 KiB
Python
267 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
|
"""CLI for durable external-task truth and notification receipts.
|
|
|
|
The CLI is suitable for OpenClaw workers, systemd jobs and release gates. It
|
|
never infers execution from the presence of a manifest: each lifecycle action
|
|
must be recorded explicitly in the shared SQLite registry.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Mapping, Sequence
|
|
|
|
from .external_task_control import ExternalTaskCoordinator
|
|
from .external_task_query import list_records
|
|
|
|
|
|
def _json_mapping(value: str | None, path: str | None) -> Mapping[str, Any]:
|
|
if value and path:
|
|
raise ValueError("inline JSON and JSON file are mutually exclusive")
|
|
if path:
|
|
parsed = json.loads(Path(path).read_text(encoding="utf-8"))
|
|
else:
|
|
parsed = json.loads(value or "{}")
|
|
if not isinstance(parsed, Mapping):
|
|
raise ValueError("JSON value must be an object")
|
|
return dict(parsed)
|
|
|
|
|
|
def _coordinator(args: argparse.Namespace) -> ExternalTaskCoordinator:
|
|
return ExternalTaskCoordinator.from_file(args.store_file)
|
|
|
|
|
|
def _print(value: Any) -> None:
|
|
print(json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2, default=str))
|
|
|
|
|
|
def cmd_declare(args: argparse.Namespace) -> int:
|
|
record = _coordinator(args).declare_file(args.manifest)
|
|
_print(record.as_dict())
|
|
return 0
|
|
|
|
|
|
def cmd_discover(args: argparse.Namespace) -> int:
|
|
records = list_records(
|
|
args.store_file,
|
|
executor=args.executor,
|
|
states=tuple(args.state or ()),
|
|
)
|
|
_print({"count": len(records), "items": [item.as_dict() for item in records]})
|
|
return 0
|
|
|
|
|
|
def cmd_register(args: argparse.Namespace) -> int:
|
|
_print(_coordinator(args).register(args.task_id, executor_run_id=args.executor_run_id).as_dict())
|
|
return 0
|
|
|
|
|
|
def cmd_claim(args: argparse.Namespace) -> int:
|
|
_print(
|
|
_coordinator(args).claim(
|
|
args.task_id,
|
|
executor_run_id=args.executor_run_id,
|
|
executor_id=args.executor_id,
|
|
).as_dict()
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_start(args: argparse.Namespace) -> int:
|
|
_print(
|
|
_coordinator(args).start(
|
|
args.task_id,
|
|
executor_run_id=args.executor_run_id,
|
|
executor_id=args.executor_id,
|
|
).as_dict()
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_heartbeat(args: argparse.Namespace) -> int:
|
|
_print(
|
|
_coordinator(args).heartbeat(
|
|
args.task_id,
|
|
executor_run_id=args.executor_run_id,
|
|
executor_id=args.executor_id,
|
|
current_step=args.current_step,
|
|
).as_dict()
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_succeed(args: argparse.Namespace) -> int:
|
|
evidence = _json_mapping(args.evidence_json, args.evidence_file)
|
|
_print(
|
|
_coordinator(args).succeed(
|
|
args.task_id,
|
|
executor_run_id=args.executor_run_id,
|
|
executor_id=args.executor_id,
|
|
evidence={str(key): str(value) for key, value in evidence.items()},
|
|
).as_dict()
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_fail(args: argparse.Namespace) -> int:
|
|
error = _json_mapping(args.error_json, args.error_file)
|
|
evidence = None
|
|
if args.evidence_json or args.evidence_file:
|
|
evidence_value = _json_mapping(args.evidence_json, args.evidence_file)
|
|
evidence = {str(key): str(value) for key, value in evidence_value.items()}
|
|
_print(
|
|
_coordinator(args).fail(
|
|
args.task_id,
|
|
executor_run_id=args.executor_run_id,
|
|
executor_id=args.executor_id,
|
|
error=error,
|
|
evidence=evidence,
|
|
).as_dict()
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_status(args: argparse.Namespace) -> int:
|
|
_print(_coordinator(args).status(args.task_id))
|
|
return 0
|
|
|
|
|
|
def cmd_gate(args: argparse.Namespace) -> int:
|
|
value = _coordinator(args).gate(args.task_id)
|
|
_print(value)
|
|
return 0 if value["passed"] else 3
|
|
|
|
|
|
def cmd_notifications(args: argparse.Namespace) -> int:
|
|
items = _coordinator(args).notifications.list(task_id=args.task_id, status=args.status)
|
|
_print({"count": len(items), "items": [item.as_dict() for item in items]})
|
|
return 0
|
|
|
|
|
|
def cmd_ack(args: argparse.Namespace) -> int:
|
|
_print(
|
|
_coordinator(args).notifications.acknowledge(
|
|
args.notification_id,
|
|
receipt_sha256=args.receipt_sha256,
|
|
).as_dict()
|
|
)
|
|
return 0
|
|
|
|
|
|
def common(parser: argparse.ArgumentParser) -> None:
|
|
parser.add_argument(
|
|
"--store-file",
|
|
default=os.getenv(
|
|
"AIOS_EXTERNAL_TASK_STORE_FILE",
|
|
"/var/lib/aios-v2/external-task-truth.sqlite3",
|
|
),
|
|
)
|
|
|
|
|
|
def lifecycle(parser: argparse.ArgumentParser) -> None:
|
|
parser.add_argument("--task-id", required=True)
|
|
parser.add_argument("--executor-run-id", required=True)
|
|
parser.add_argument("--executor-id", required=True)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="AIOS external task truth CLI")
|
|
commands = parser.add_subparsers(dest="command", required=True)
|
|
|
|
declare = commands.add_parser("declare")
|
|
common(declare)
|
|
declare.add_argument("--manifest", required=True)
|
|
declare.set_defaults(handler=cmd_declare)
|
|
|
|
discover = commands.add_parser("discover")
|
|
common(discover)
|
|
discover.add_argument("--executor", default="openclaw")
|
|
discover.add_argument("--state", action="append")
|
|
discover.set_defaults(handler=cmd_discover)
|
|
|
|
register = commands.add_parser("register")
|
|
common(register)
|
|
register.add_argument("--task-id", required=True)
|
|
register.add_argument("--executor-run-id", required=True)
|
|
register.set_defaults(handler=cmd_register)
|
|
|
|
claim = commands.add_parser("claim")
|
|
common(claim)
|
|
lifecycle(claim)
|
|
claim.set_defaults(handler=cmd_claim)
|
|
|
|
start = commands.add_parser("start")
|
|
common(start)
|
|
lifecycle(start)
|
|
start.set_defaults(handler=cmd_start)
|
|
|
|
heartbeat = commands.add_parser("heartbeat")
|
|
common(heartbeat)
|
|
lifecycle(heartbeat)
|
|
heartbeat.add_argument("--current-step", type=int, required=True)
|
|
heartbeat.set_defaults(handler=cmd_heartbeat)
|
|
|
|
succeed = commands.add_parser("succeed")
|
|
common(succeed)
|
|
lifecycle(succeed)
|
|
evidence = succeed.add_mutually_exclusive_group(required=True)
|
|
evidence.add_argument("--evidence-json")
|
|
evidence.add_argument("--evidence-file")
|
|
succeed.set_defaults(handler=cmd_succeed)
|
|
|
|
fail = commands.add_parser("fail")
|
|
common(fail)
|
|
lifecycle(fail)
|
|
error = fail.add_mutually_exclusive_group(required=True)
|
|
error.add_argument("--error-json")
|
|
error.add_argument("--error-file")
|
|
fail_evidence = fail.add_mutually_exclusive_group()
|
|
fail_evidence.add_argument("--evidence-json")
|
|
fail_evidence.add_argument("--evidence-file")
|
|
fail.set_defaults(handler=cmd_fail)
|
|
|
|
status = commands.add_parser("status")
|
|
common(status)
|
|
status.add_argument("--task-id", required=True)
|
|
status.set_defaults(handler=cmd_status)
|
|
|
|
gate = commands.add_parser("gate")
|
|
common(gate)
|
|
gate.add_argument("--task-id", required=True)
|
|
gate.set_defaults(handler=cmd_gate)
|
|
|
|
notifications = commands.add_parser("notifications")
|
|
common(notifications)
|
|
notifications.add_argument("--task-id")
|
|
notifications.add_argument("--status")
|
|
notifications.set_defaults(handler=cmd_notifications)
|
|
|
|
ack = commands.add_parser("ack-notification")
|
|
common(ack)
|
|
ack.add_argument("--notification-id", required=True)
|
|
ack.add_argument("--receipt-sha256", required=True)
|
|
ack.set_defaults(handler=cmd_ack)
|
|
return parser
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
try:
|
|
return int(args.handler(args))
|
|
except Exception as exc:
|
|
print(
|
|
json.dumps(
|
|
{"status": "error", "error": type(exc).__name__, "message": str(exc)},
|
|
ensure_ascii=False,
|
|
),
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|