#!/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())