"""CLI wrapper for atomic mutations of `approved-tasks.json`. Shell scripts (ralph.sh) and cron-job prompts cannot import `src.jsonlock` directly. This module is the bridge: every subcommand serialises its mutation through ``write_locked`` so external writers honour the same flock invariant as the in-process code in ``src/router.py`` / ``src/planning_session.py``. Run via: python3 -m src.approved_tasks_cli [args] Subcommands: set-status --slug SLUG --status STATUS set-field --slug SLUG --key KEY --value VALUE [--int|--null|--now|--json] add-project --slug SLUG --description DESC [--status STATUS] mark-running --slug SLUG --pid PID mark-failed --slug SLUG [--error MSG] show [--slug SLUG] # read-only inspection All mutators bump ``last_updated`` to the current UTC ISO timestamp. Exit codes: 0 success 1 bad usage / invalid argument 2 slug not found 3 lock timeout / IO error """ from __future__ import annotations import argparse import json import sys from datetime import datetime, timezone from pathlib import Path # Make `from src.jsonlock import ...` work whether invoked as # `python -m src.approved_tasks_cli` (sys.path already correct) or as # `python3 src/approved_tasks_cli.py` (need to prepend project root). _PROJECT_ROOT = Path(__file__).resolve().parent.parent if str(_PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(_PROJECT_ROOT)) from src.jsonlock import read_locked, write_locked, LockTimeoutError # noqa: E402 APPROVED_TASKS_FILE = _PROJECT_ROOT / "approved-tasks.json" def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() def _bump_timestamp(data: dict) -> None: data["last_updated"] = _now_iso() def _find_project(data: dict, slug: str) -> dict | None: for p in data.get("projects", []): if p.get("name", "").lower() == slug.lower(): return p return None def _coerce_value(value: str, *, as_int: bool, as_null: bool, as_now: bool, as_json: bool): if as_null: return None if as_now: return _now_iso() if as_int: try: return int(value) except ValueError as exc: raise SystemExit(f"value '{value}' is not a valid int: {exc}") from exc if as_json: try: return json.loads(value) except json.JSONDecodeError as exc: raise SystemExit(f"value is not valid JSON: {exc}") from exc return value # ---- subcommand implementations ------------------------------------------- def cmd_set_status(args) -> int: def mutator(data: dict) -> dict: proj = _find_project(data, args.slug) if proj is None: raise KeyError(args.slug) proj["status"] = args.status _bump_timestamp(data) return data try: write_locked(str(APPROVED_TASKS_FILE), mutator) except KeyError: print(f"slug '{args.slug}' not found", file=sys.stderr) return 2 print(f"set status of '{args.slug}' = '{args.status}'") return 0 def cmd_set_field(args) -> int: value = _coerce_value( args.value or "", as_int=args.int, as_null=args.null, as_now=args.now, as_json=args.json_value, ) def mutator(data: dict) -> dict: proj = _find_project(data, args.slug) if proj is None: raise KeyError(args.slug) proj[args.key] = value _bump_timestamp(data) return data try: write_locked(str(APPROVED_TASKS_FILE), mutator) except KeyError: print(f"slug '{args.slug}' not found", file=sys.stderr) return 2 print(f"set {args.slug}.{args.key} = {value!r}") return 0 def cmd_add_project(args) -> int: def mutator(data: dict) -> dict: data.setdefault("projects", []) if _find_project(data, args.slug) is not None: raise FileExistsError(args.slug) entry = { "name": args.slug, "description": args.description, "status": args.status, "planning_session_id": None, "final_plan_path": None, "proposed_at": _now_iso(), "approved_at": None, "started_at": None, "pid": None, } data["projects"].append(entry) _bump_timestamp(data) return data try: write_locked(str(APPROVED_TASKS_FILE), mutator) except FileExistsError: print(f"slug '{args.slug}' already exists — refusing to overwrite", file=sys.stderr) return 1 print(f"added project '{args.slug}' (status={args.status})") return 0 def cmd_mark_running(args) -> int: """Convenience: status=running, started_at=now, pid= in ONE locked write.""" def mutator(data: dict) -> dict: proj = _find_project(data, args.slug) if proj is None: raise KeyError(args.slug) proj["status"] = "running" proj["started_at"] = _now_iso() proj["pid"] = args.pid _bump_timestamp(data) return data try: write_locked(str(APPROVED_TASKS_FILE), mutator) except KeyError: print(f"slug '{args.slug}' not found", file=sys.stderr) return 2 print(f"marked '{args.slug}' running (pid={args.pid})") return 0 def cmd_mark_failed(args) -> int: def mutator(data: dict) -> dict: proj = _find_project(data, args.slug) if proj is None: raise KeyError(args.slug) proj["status"] = "failed" if args.error: proj["error"] = args.error _bump_timestamp(data) return data try: write_locked(str(APPROVED_TASKS_FILE), mutator) except KeyError: print(f"slug '{args.slug}' not found", file=sys.stderr) return 2 print(f"marked '{args.slug}' failed") return 0 def cmd_show(args) -> int: try: data = read_locked(str(APPROVED_TASKS_FILE)) except FileNotFoundError: print(f"file not found: {APPROVED_TASKS_FILE}", file=sys.stderr) return 3 if args.slug: proj = _find_project(data, args.slug) if proj is None: print(f"slug '{args.slug}' not found", file=sys.stderr) return 2 print(json.dumps(proj, indent=2, ensure_ascii=False)) else: print(json.dumps(data, indent=2, ensure_ascii=False)) return 0 # ---- argparse setup ------------------------------------------------------- def _build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="approved-tasks", description="Atomic CLI for approved-tasks.json (uses src.jsonlock.write_locked).", ) sub = p.add_subparsers(dest="command", required=True) sp = sub.add_parser("set-status", help="Set the status field of a project.") sp.add_argument("--slug", required=True) sp.add_argument("--status", required=True) sp.set_defaults(func=cmd_set_status) sp = sub.add_parser("set-field", help="Set an arbitrary field on a project.") sp.add_argument("--slug", required=True) sp.add_argument("--key", required=True) sp.add_argument("--value", default="", help="String value (or use --null/--now).") g = sp.add_mutually_exclusive_group() g.add_argument("--int", action="store_true", help="Coerce --value to int.") g.add_argument("--null", action="store_true", help="Set value to JSON null.") g.add_argument("--now", action="store_true", help="Set value to current UTC ISO timestamp.") g.add_argument("--json-value", action="store_true", dest="json_value", help="Parse --value as JSON.") sp.set_defaults(func=cmd_set_field) sp = sub.add_parser("add-project", help="Append a new project entry.") sp.add_argument("--slug", required=True) sp.add_argument("--description", required=True) sp.add_argument("--status", default="pending") sp.set_defaults(func=cmd_add_project) sp = sub.add_parser("mark-running", help="Atomic: status=running, started_at=now, pid=.") sp.add_argument("--slug", required=True) sp.add_argument("--pid", type=int, required=True) sp.set_defaults(func=cmd_mark_running) sp = sub.add_parser("mark-failed", help="Set status=failed (and optionally an error message).") sp.add_argument("--slug", required=True) sp.add_argument("--error", default=None) sp.set_defaults(func=cmd_mark_failed) sp = sub.add_parser("show", help="Print approved-tasks.json (or one project) to stdout.") sp.add_argument("--slug", default=None) sp.set_defaults(func=cmd_show) return p def main(argv: list[str] | None = None) -> int: parser = _build_parser() args = parser.parse_args(argv) try: return args.func(args) except LockTimeoutError as exc: print(f"lock timeout: {exc}", file=sys.stderr) return 3 except OSError as exc: print(f"io error: {exc}", file=sys.stderr) return 3 if __name__ == "__main__": sys.exit(main())