feat(todos): JSON argv form for the completion validation command

loopx/canary/module_metric_baseline.json

@@ -103,7 +103,7 @@
     "loopx/todos.py": {
       "any_count": 48,
       "dict_any_count": 0,
-      "lines": 2190
+      "lines": 2229
     },
     "loopx/worker_bridge.py": {
       "any_count": 28,

loopx/cli_commands/todo.py

@@ -221,14 +221,24 @@ def register_todo_command(
         "--validation-label",
         help="Optional public-safe label for the validation receipt.",
     )
+    todo_parser.add_argument(
+        "--validation-command-json",
+        help=(
+            "Trusted JSON string array (argv form, no shell parsing) for the "
+            "completion validation command, e.g. '[\"pytest\",\"-q\",\"tests/"
+            "test_x.py\"]'. Mutually exclusive with --validation-command; set "
+            "on `todo add`."
+        ),
+    )
     todo_parser.add_argument(
         "--validation-timeout-seconds",
         type=int,
         help=(
             "Per-todo timeout for the caller-approved validation command. "
-            "Only meaningful with --validation-command on `todo add`; must be "
-            "1-29 so a timed-out validation still produces a typed receipt "
-            "inside the 30s outer subprocess budget. Defaults to 20."
+            "Only meaningful with --validation-command or "
+            "--validation-command-json on `todo add`; must be 1-29 so a "
+            "timed-out validation still produces a typed receipt inside the "
+            "30s outer subprocess budget. Defaults to 20."
         ),
     )
     todo_parser.add_argument("--reason", help="Public-safe reason for blocked/deferred/supersede transitions.")
@@ -629,6 +639,7 @@ def handle_todo_command(
                 replan_obligation_id=replan_obligation_id,
                 resume_when=args.resume_when,
                 validation_command=args.validation_command,
+                validation_command_json=args.validation_command_json,
                 validation_label=args.validation_label,
                 validation_timeout_seconds=args.validation_timeout_seconds,
                 monitor_metadata={

loopx/cli_commands/todo_argument_validation.py

@@ -47,6 +47,7 @@
     ("--successor-todo-id", "successor_todo_ids"),
     ("--resume-when", "resume_when"),
     ("--validation-command", "validation_command"),
+    ("--validation-command-json", "validation_command_json"),
     ("--validation-label", "validation_label"),
     ("--validation-timeout-seconds", "validation_timeout_seconds"),
     ("--clear-resume-when", "clear_resume_when"),

loopx/control_plane/runtime/validation_command.py

@@ -15,21 +15,32 @@
 def run_caller_validation(
     workspace: Path,
     *,
-    validation_command: str,
+    validation_command: str | None = None,
+    validation_argv: list[str] | None = None,
     validation_label: str,
     timeout_seconds: int,
 ) -> dict[str, Any]:
     """Run a caller-approved validation command and return a privacy-safe receipt.
 
-    The command is ``shlex.split`` (no shell) and run with ``cwd=workspace``.
+    Exactly one of ``validation_command`` (shell-free ``shlex.split``) or
+    ``validation_argv`` (pre-split JSON argv array, no shell parsing) selects
+    the command form. The command runs with ``cwd=workspace``.
     Only ``exit_code`` and the boolean ``passed`` are recorded; command stdout,
     stderr, and local paths are deliberately not captured. A timeout raises
     ``subprocess.TimeoutExpired``; callers decide whether to convert that into
     a failure receipt.
     """
-    argv = shlex.split(validation_command)
+    if (validation_command is None) == (validation_argv is None):
+        raise ValueError(
+            "exactly one of validation_command or validation_argv is required"
+        )
+    argv = (
+        [str(item) for item in validation_argv]
+        if validation_argv is not None
+        else shlex.split(validation_command or "")
+    )
     if not argv:
-        raise ValueError("validation_command must not be empty")
+        raise ValueError("validation command must not be empty")
     result = subprocess.run(
         argv,
         cwd=workspace,

loopx/control_plane/todos/completion_validation.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import subprocess
 from pathlib import Path
 from typing import Any
@@ -34,27 +35,32 @@ def _resolve_goal_repo_workspace(registry_path: Path, goal_id: str) -> Path | No
 
 def _read_declared_validation(
     *, state_file: Path, todo_id: str, role: str | None
-) -> tuple[str | None, str | None, int | None, bool]:
+) -> tuple[str | None, list[str] | None, str | None, int | None, bool]:
     """Pre-read a todo's declared validation command without the mutation lock.
 
-    Returns ``(validation_command, validation_label, validation_timeout_seconds,
-    already_completed)`` from the markdown state file, or ``(None, None, None,
-    False)`` when the todo is not materialized in markdown. Read-only; safe to
-    call before acquiring the state-file lock so a slow validation command does
-    not block concurrent todo operations on the same goal (the MUTATION lock
-    deadline is 5s). ``validation_command`` and ``validation_timeout_seconds``
-    are set only at ``todo add`` and have no update path, so the values cannot
-    drift between this pre-read and the in-lock commit. A stored timeout that
-    fails to parse as an int falls back to ``None`` (the default), matching the
-    writer-side range check that guarantees a well-formed value.
+    Returns ``(validation_command, validation_argv, validation_label,
+    validation_timeout_seconds, already_completed)`` from the markdown state
+    file, or ``(None, None, None, None, False)`` when the todo is not
+    materialized in markdown. Read-only; safe to call before acquiring the
+    state-file lock so a slow validation command does not block concurrent todo
+    operations on the same goal (the MUTATION lock deadline is 5s).
+    ``validation_command``, ``validation_command_argv`` and
+    ``validation_timeout_seconds`` are set only at ``todo add`` and have no
+    update path, so the values cannot drift between this pre-read and the
+    in-lock commit. A stored timeout that fails to parse as an int falls back
+    to ``None`` (the default), matching the writer-side range check that
+    guarantees a well-formed value. A stored argv that fails to parse as a
+    non-empty string list collapses to ``[]`` — never to ``None`` — so a
+    corrupted declaration still runs the gate and fails closed as a malformed
+    command instead of silently skipping validation.
     """
     try:
         lines = state_file.read_text(encoding="utf-8").splitlines()
     except FileNotFoundError:
-        return None, None, None, False
+        return None, None, None, None, False
     match = find_todo_block(lines, todo_id=todo_id, role=role)
     if not match:
-        return None, None, None, False
+        return None, None, None, None, False
     _role, _section, _start, _end, block = match
     try:
         timeout_seconds: int | None = (
@@ -64,8 +70,23 @@ def _read_declared_validation(
         )
     except (TypeError, ValueError):
         timeout_seconds = None
+    validation_argv: list[str] | None = None
+    if block.get("validation_command_argv"):
+        try:
+            parsed_argv = json.loads(block["validation_command_argv"])
+        except ValueError:
+            parsed_argv = None
+        if (
+            isinstance(parsed_argv, list)
+            and parsed_argv
+            and all(isinstance(item, str) and item for item in parsed_argv)
+        ):
+            validation_argv = parsed_argv
+        else:
+            validation_argv = []
     return (
         block.get("validation_command") or None,
+        validation_argv,
         block.get("validation_label") or None,
         timeout_seconds,
         block.get("status") == TODO_STATUS_DONE,
@@ -75,23 +96,26 @@ def _read_declared_validation(
 def _run_declared_completion_validation(
     *,
     validation_command: str | None,
+    validation_argv: list[str] | None,
     validation_label: str | None,
     validation_timeout_seconds: int | None,
     registry_path: Path,
     goal_id: str,
 ) -> dict[str, Any] | None:
     """Run a todo's declared caller-approved validation command.
 
-    Returns ``None`` when no ``validation_command`` is declared (the unchanged
-    fast path). ``validation_timeout_seconds`` overrides the module default
-    when declared on ``todo add``; ``None`` keeps the default. Otherwise always
-    returns a privacy-safe receipt whose ``passed`` is True only when the
-    command ran and exited zero; setup failures (no repository workspace),
-    timeouts, missing executables, and malformed commands are all reported as
-    ``passed=False`` receipts rather than raised, so completion can surface a
-    typed failure without committing.
+    Returns ``None`` when no command form is declared (the unchanged fast
+    path). ``validation_argv`` (the JSON argv form declared on ``todo add``)
+    takes the run-once no-shell path; ``validation_command`` keeps the
+    legacy shlex form. ``validation_timeout_seconds`` overrides the module
+    default when declared on ``todo add``; ``None`` keeps the default.
+    Otherwise always returns a privacy-safe receipt whose ``passed`` is True
+    only when the command ran and exited zero; setup failures (no repository
+    workspace), timeouts, missing executables, and malformed commands are all
+    reported as ``passed=False`` receipts rather than raised, so completion
+    can surface a typed failure without committing.
     """
-    if not validation_command:
+    if not validation_command and validation_argv is None:
         return None
     timeout_seconds = (
         validation_timeout_seconds
@@ -116,6 +140,13 @@ def _run_declared_completion_validation(
             "local_path_captured": False,
         }
     try:
+        if validation_argv is not None:
+            return run_caller_validation(
+                workspace,
+                validation_argv=validation_argv,
+                validation_label=label,
+                timeout_seconds=timeout_seconds,
+            )
         return run_caller_validation(
             workspace,
             validation_command=str(validation_command),
@@ -149,7 +180,9 @@ def _run_declared_completion_validation(
             "local_path_captured": False,
         }
     except ValueError as exc:
-        # shlex.split rejects malformed (e.g. unbalanced-quote) commands.
+        # shlex.split rejects malformed (e.g. unbalanced-quote) commands, and
+        # an argv form that collapsed to [] (corrupted stored declaration)
+        # fails the runner's own empty-command check.
         return {
             "schema_version": CALLER_VALIDATION_RECEIPT_SCHEMA_VERSION,
             "command_label": label,
@@ -184,13 +217,15 @@ def run_completion_validation_gate(
     """
     (
         validation_command,
+        validation_argv,
         validation_label,
         validation_timeout_seconds,
         already_completed,
     ) = _read_declared_validation(state_file=state_file, todo_id=todo_id, role=role)
     completion_validation = (
         _run_declared_completion_validation(
             validation_command=validation_command,
+            validation_argv=validation_argv,
             validation_label=validation_label,
             validation_timeout_seconds=validation_timeout_seconds,
             registry_path=registry_path,

loopx/control_plane/todos/contract.py

@@ -1137,6 +1137,7 @@ def _metadata_value_is_present(value: Any) -> bool:
             "updated_at",
             "completion_turn_key",
             "validation_command",
+            "validation_command_argv",
             "validation_label",
             "validation_timeout_seconds",
         )
@@ -1276,6 +1277,7 @@ def format_todo_metadata_line(
     evidence: str | None = None,
     completion_turn_key: str | None = None,
     validation_command: str | None = None,
+    validation_command_argv: str | None = None,
     validation_label: str | None = None,
     validation_timeout_seconds: str | None = None,
     reason: str | None = None,

loopx/todos.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 from contextlib import ExitStack
+from json import dumps as json_dumps
+from json import loads as json_loads
 from pathlib import Path
 from typing import Any
 
@@ -548,6 +550,28 @@ def list_goal_todos(
     return payload
 
 
+def _normalize_validation_command_json(raw: str | None) -> list[str] | None:
+    """Validate a ``--validation-command-json`` payload (run-once precedent).
+
+    Returns the argv list, or ``None`` when no JSON form is declared. Raises
+    ``ValueError`` when the payload is not a non-empty JSON string array —
+    the same shape rule the Turn-level ``--validation-command-json`` applies.
+    """
+    if raw is None:
+        return None
+    try:
+        argv = json_loads(raw)
+    except ValueError as exc:
+        raise ValueError(
+            "--validation-command-json must be a JSON string array"
+        ) from exc
+    if not isinstance(argv, list) or not argv or not all(
+        isinstance(item, str) and item for item in argv
+    ):
+        raise ValueError("--validation-command-json must be a JSON string array")
+    return argv
+
+
 def add_todo_to_lines(
     lines: list[str],
     *,
@@ -576,16 +600,24 @@ def add_todo_to_lines(
     replan_obligation_id: str | None = None,
     resume_when: str | None = None,
     validation_command: str | None = None,
+    validation_command_json: str | None = None,
     validation_label: str | None = None,
     validation_timeout_seconds: int | None = None,
     monitor_metadata: dict[str, Any] | None = None,
     evidence: str | None = None,
     updated_at: str | None = None,
 ) -> dict[str, Any]:
+    if validation_command and validation_command_json:
+        raise ValueError(
+            "--validation-command and --validation-command-json are mutually "
+            "exclusive; declare the validation command in exactly one form"
+        )
+    validation_argv = _normalize_validation_command_json(validation_command_json)
     if validation_timeout_seconds is not None:
-        if not validation_command:
+        if not validation_command and validation_argv is None:
             raise ValueError(
-                "--validation-timeout-seconds requires --validation-command"
+                "--validation-timeout-seconds requires --validation-command "
+                "or --validation-command-json"
             )
         if not (
             1
@@ -674,6 +706,11 @@ def add_todo_to_lines(
             replan_obligation_id=replan_obligation_id,
             resume_when=normalized_resume_when,
             validation_command=validation_command,
+            validation_command_argv=(
+                json_dumps(validation_argv)
+                if validation_argv is not None
+                else None
+            ),
             validation_label=validation_label,
             validation_timeout_seconds=(
                 str(validation_timeout_seconds)
@@ -867,6 +904,7 @@ def add_goal_todo(
     replan_obligation_id: str | None = None,
     resume_when: str | None = None,
     validation_command: str | None = None,
+    validation_command_json: str | None = None,
     validation_label: str | None = None,
     validation_timeout_seconds: int | None = None,
     monitor_metadata: dict[str, Any] | None = None,
@@ -1068,6 +1106,7 @@ def add_goal_todo(
             replan_obligation_id=replan_obligation_id,
             resume_when=normalized_resume_when,
             validation_command=validation_command,
+            validation_command_json=validation_command_json,
             validation_label=validation_label,
             validation_timeout_seconds=validation_timeout_seconds,
             monitor_metadata=normalized_monitor_metadata,