feat(todos): wire caller-approved validation_command into todo completion

loopx/canary/module_metric_baseline.json

@@ -93,7 +93,7 @@
     "loopx/todos.py": {
       "any_count": 48,
       "dict_any_count": 0,
-      "lines": 2142
+      "lines": 2318
     },
     "loopx/worker_bridge.py": {
       "any_count": 28,

loopx/capabilities/issue_fix/acceptance_loop.py

@@ -2,7 +2,6 @@
 
 import hashlib
 import os
-import shlex
 import shutil
 import subprocess
 import sys
@@ -13,6 +12,10 @@
 from typing import Any
 
 from .intake_surface import build_content_ops_issue_fix_metadata_preview_packet
+from ...control_plane.runtime.validation_command import (
+    run_caller_validation as _run_caller_validation,
+    require_validation_passed as _require_passed,
+)
 
 
 ISSUE_FIX_ACCEPTANCE_LOOP_SCHEMA_VERSION = "issue_fix_acceptance_loop_v0"
@@ -247,39 +250,9 @@ def add_lines(lines: list[str]) -> None:
     return files[:20], truncated
 
 
-def _run_caller_validation(
-    workspace: Path,
-    *,
-    validation_command: str,
-    validation_label: str,
-    timeout_seconds: int,
-) -> dict[str, Any]:
-    argv = shlex.split(validation_command)
-    if not argv:
-        raise ValueError("validation_command must not be empty")
-    result = subprocess.run(
-        argv,
-        cwd=workspace,
-        env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
-        text=True,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        timeout=timeout_seconds,
-    )
-    return {
-        "schema_version": "issue_fix_validation_command_v0",
-        "command_label": validation_label or "caller-declared validation",
-        "exit_code": result.returncode,
-        "passed": result.returncode == 0,
-        "stdout_captured": False,
-        "stderr_captured": False,
-        "local_path_captured": False,
-    }
-
-
-def _require_passed(step: Mapping[str, Any]) -> None:
-    if step.get("passed") is not True:
-        raise RuntimeError(f"{step.get('command_label')} failed with exit code {step.get('exit_code')}")
+# Caller-validation execution (_run_caller_validation / _require_passed) is
+# imported from control_plane.runtime.validation_command so the todo-completion
+# path can reuse the same privacy-safe handler.
 
 
 def _remove_temporary_git_workspace(workspace: Path) -> None:

loopx/cli_commands/todo.py

@@ -111,6 +111,19 @@ def register_todo_command(
     todo_parser.add_argument("--status", choices=["open", "done", "blocked", "deferred"], help="For todo add/update, set the lifecycle status.")
     todo_parser.add_argument("--note", help="Public-safe note to attach to a lifecycle transition.")
     todo_parser.add_argument("--evidence", help="Public-safe evidence pointer or short result for complete/update.")
+    todo_parser.add_argument(
+        "--validation-command",
+        help=(
+            "Caller-approved validation command (no shell) to run before a "
+            "todo's completion commits, e.g. 'pytest -q tests/test_x.py'. Set "
+            "on `todo add`; completion runs it independently and blocks on a "
+            "non-zero exit."
+        ),
+    )
+    todo_parser.add_argument(
+        "--validation-label",
+        help="Optional public-safe label for the validation receipt.",
+    )
     todo_parser.add_argument("--reason", help="Public-safe reason for blocked/deferred/supersede transitions.")
     todo_parser.add_argument(
         "--authority-reason",
@@ -502,6 +515,8 @@ def handle_todo_command(
                 agent_id=args.agent_id,
                 unblocks_todo_id=args.unblocks_todo_id,
                 resume_when=args.resume_when,
+                validation_command=args.validation_command,
+                validation_label=args.validation_label,
                 monitor_metadata={
                     "target_key": args.monitor_target_key,
                     "cadence": args.cadence,

loopx/cli_commands/todo_argument_validation.py

@@ -45,6 +45,8 @@
     ("--unblocks-todo-id", "unblocks_todo_id"),
     ("--successor-todo-id", "successor_todo_ids"),
     ("--resume-when", "resume_when"),
+    ("--validation-command", "validation_command"),
+    ("--validation-label", "validation_label"),
     ("--clear-resume-when", "clear_resume_when"),
     ("--target-key", "monitor_target_key"),
     ("--cadence", "cadence"),

loopx/control_plane/runtime/validation_command.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import os
+import shlex
+import subprocess
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any
+
+# Frozen wire id shared with capabilities/issue_fix. Kept verbatim for wire
+# compatibility with existing caller-repo-branch receipts.
+CALLER_VALIDATION_RECEIPT_SCHEMA_VERSION = "issue_fix_validation_command_v0"
+
+
+def run_caller_validation(
+    workspace: Path,
+    *,
+    validation_command: str,
+    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``.
+    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 not argv:
+        raise ValueError("validation_command must not be empty")
+    result = subprocess.run(
+        argv,
+        cwd=workspace,
+        env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        timeout=timeout_seconds,
+    )
+    return {
+        "schema_version": CALLER_VALIDATION_RECEIPT_SCHEMA_VERSION,
+        "command_label": validation_label or "caller-declared validation",
+        "exit_code": result.returncode,
+        "passed": result.returncode == 0,
+        "stdout_captured": False,
+        "stderr_captured": False,
+        "local_path_captured": False,
+    }
+
+
+def require_validation_passed(step: Mapping[str, Any]) -> None:
+    """Raise ``RuntimeError`` unless the validation step ``passed``."""
+    if step.get("passed") is not True:
+        raise RuntimeError(
+            f"{step.get('command_label')} failed with exit code {step.get('exit_code')}"
+        )

loopx/control_plane/todos/contract.py

@@ -1090,6 +1090,8 @@ def _metadata_value_is_present(value: Any) -> bool:
             "completed_at",
             "updated_at",
             "completion_turn_key",
+            "validation_command",
+            "validation_label",
         )
     ),
     _TodoMetadataField(
@@ -1225,6 +1227,8 @@ def format_todo_metadata_line(
     note: str | None = None,
     evidence: str | None = None,
     completion_turn_key: str | None = None,
+    validation_command: str | None = None,
+    validation_label: str | None = None,
     reason: str | None = None,
     completed_at: str | None = None,
     updated_at: str | None = None,

loopx/control_plane/todos/markdown.py

@@ -179,4 +179,21 @@ def render_todo_markdown(payload: dict[str, Any]) -> str:
                 f"- status: `{apply_result.get('status')}`",
             ]
         )
+    validation = payload.get("validation")
+    if isinstance(validation, dict):
+        lines.extend(
+            [
+                "",
+                "## Validation",
+                "",
+                f"- validation_blocked_completion: `{payload.get('validation_blocked_completion')}`",
+                f"- command_label: `{validation.get('command_label')}`",
+                f"- passed: `{validation.get('passed')}`",
+                f"- status: `{validation.get('status')}`",
+                f"- exit_code: `{validation.get('exit_code')}`",
+            ]
+        )
+        summary = validation.get("summary")
+        if summary:
+            lines.append(f"- summary: {summary}")
     return "\n".join(lines)