test(claude): pin the MCP complete_task validation gate coverage

tests/test_goal_mode_mcp_completion_validation.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import json
+import shlex
+import sys
+from pathlib import Path
+
+from loopx.goal_mode_mcp import GoalModeMCPConfig, GoalModeMCPControlPlane
+from loopx.status import parse_active_state_todos
+from loopx.todos import add_goal_todo
+
+GOAL_ID = "mcp-completion-validation"
+AGENT = "claude"
+
+_FAIL_COMMAND = f'{shlex.quote(sys.executable)} -c "raise SystemExit(1)"'
+
+
+def _write_fixture(tmp_path: Path) -> tuple[Path, Path]:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    state = repo / "ACTIVE_GOAL_STATE.md"
+    state.write_text(
+        "\n".join(
+            [
+                "---",
+                f"goal_id: {GOAL_ID}",
+                "updated_at: 2026-08-19T00:00:00+00:00",
+                "---",
+                "",
+                "## Agent Todo",
+                "",
+            ]
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    registry = tmp_path / "registry.global.json"
+    registry.write_text(
+        json.dumps(
+            {
+                "common_runtime_root": str(tmp_path / "runtime"),
+                "goals": [
+                    {
+                        "id": GOAL_ID,
+                        "domain": "harness_self_improvement",
+                        "status": "active",
+                        "repo": str(repo),
+                        "state_file": state.name,
+                        "adapter": {"kind": "harness_self_improvement"},
+                        "coordination": {
+                            "agent_model": "peer_v1",
+                            "registered_agents": [AGENT],
+                        },
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    return registry, state
+
+
+def _add_todo(registry: Path, *, validation_command: str | None = None) -> str:
+    todo = add_goal_todo(
+        registry_path=registry,
+        goal_id=GOAL_ID,
+        role="agent",
+        text="Deliver one bounded change.",
+        task_class="advancement_task",
+        claimed_by=AGENT,
+        validation_command=validation_command,
+    )
+    return str(todo["todo_id"])
+
+
+def _control(registry: Path) -> GoalModeMCPControlPlane:
+    control = GoalModeMCPControlPlane(
+        GoalModeMCPConfig(
+            server_name="loopx-test",
+            runtime_profile="claude_code",
+            legacy_host_surface="claude_code",
+        ),
+        lambda: {
+            "goal_id": GOAL_ID,
+            "registry": str(registry),
+            "agent_id": AGENT,
+        },
+    )
+    control.command_prefix = lambda: [sys.executable, "-m", "loopx.cli"]
+    return control
+
+
+def _agent_todo_status(state: Path, todo_id: str) -> str:
+    todos = parse_active_state_todos(state.read_text(encoding="utf-8"))
+    item = next(
+        item
+        for item in todos["agent_todos"]["items"]
+        if item["todo_id"] == todo_id
+    )
+    return str(item["status"])
+
+
+def _first_json_blob(output: str) -> dict:
+    return json.loads(output.split("\n--- ", 1)[0])
+
+
+def test_mcp_complete_task_fails_closed_on_failing_declared_validation(
+    tmp_path: Path,
+) -> None:
+    """The MCP route inherits the declared-command gate: a failing command
+    blocks the completion with a typed receipt and skips settlement."""
+    registry, state = _write_fixture(tmp_path)
+    todo_id = _add_todo(registry, validation_command=_FAIL_COMMAND)
+
+    output = _control(registry).complete_task(todo_id, AGENT, "claimed done")
+
+    payload = _first_json_blob(output)
+    assert payload["ok"] is False
+    assert payload["completed"] is False
+    assert payload["validation_blocked_completion"] is True
+    assert payload["validation"]["passed"] is False
+    assert "spend-slot" not in output
+    assert "refresh-state" not in output
+    assert _agent_todo_status(state, todo_id) != "done"