diff --git a/loopx/control_plane/testing/canary_harness.py b/loopx/control_plane/testing/canary_harness.py
index 1991cb0c96..c1599112f1 100644
--- a/loopx/control_plane/testing/canary_harness.py
+++ b/loopx/control_plane/testing/canary_harness.py
@@ -65,6 +65,11 @@ def run_json_cli_result(
     runtime_root: Path | None = None,
     cwd: Path | None = None,
 ) -> tuple[int, dict[str, Any]]:
+    """Return process status separately from the producer's unchanged JSON.
+
+    Callers can validate or replay the returned wire without test metadata
+    changing content commitments or overwriting a producer-owned field.
+    """
     command = [
         sys.executable,
         "-m",
@@ -89,7 +94,6 @@ def run_json_cli_result(
     payload = json.loads(result.stdout)
     if not isinstance(payload, dict):
         raise AssertionError(f"expected JSON object for {command!r}: {payload!r}")
-    payload["_returncode"] = result.returncode
     return result.returncode, payload
 
 
diff --git a/tests/control_plane/test_canary_cli_transport.py b/tests/control_plane/test_canary_cli_transport.py
new file mode 100644
index 0000000000..a1da7b10cc
--- /dev/null
+++ b/tests/control_plane/test_canary_cli_transport.py
@@ -0,0 +1,46 @@
+"""Tuple-returning CLI transport preserves producer JSON, including unknown fields."""
+from __future__ import annotations
+
+import hashlib
+import json
+import subprocess
+import sys
+from types import SimpleNamespace
+
+import pytest
+
+from loopx.control_plane.testing import canary_harness
+
+
+@pytest.mark.parametrize("exit_code", [0, 7])
+@pytest.mark.parametrize("producer_field", [False, True])
+def test_result_transport_does_not_inject_or_overwrite_source_fields(tmp_path, monkeypatch, exit_code, producer_field):
+    source = {"number": 1.0, "unknown": {"allowed": False, "evidence": None}}
+    if producer_field:
+        source["_returncode"] = {"producer_fact": "Preserve this exact value."}
+    raw = json.dumps(source, ensure_ascii=False)
+    monkeypatch.setattr(canary_harness.subprocess, "run", lambda *args, **kwargs:
+        SimpleNamespace(stdout=raw, stderr="", returncode=exit_code))
+    status, observed = canary_harness.run_json_cli_result("probe", registry_path=tmp_path / "absent.json")
+    assert status == exit_code
+    assert json.dumps(observed, ensure_ascii=False) == raw
+
+
+@pytest.mark.parametrize("valid_digest", [False, True])
+def test_real_capture_cli_returns_the_same_wire_through_the_helper(tmp_path, monkeypatch, valid_digest):
+    monkeypatch.setenv("LOOPX_USAGE_PING", "0")
+    capture = tmp_path / "source.json"
+    raw = b'{"goal":"Keep the original source and stop before publication.","value":1.0}'
+    capture.write_bytes(raw)
+    digest = hashlib.sha256(raw).hexdigest() if valid_digest else "0" * 64
+    registry, runtime = tmp_path / "absent-registry.json", tmp_path / "absent-runtime"
+    arguments = ["quota", "context-read", "--capture-file", str(capture), "--capture-sha256", digest]
+    direct = subprocess.run([sys.executable, "-m", "loopx.cli", "--registry", str(registry),
+        "--runtime-root", str(runtime), "--format", "json", *arguments],
+        cwd=tmp_path, capture_output=True, text=True, check=False)
+    status, observed = canary_harness.run_json_cli_result(*arguments,
+        registry_path=registry, runtime_root=runtime, cwd=tmp_path)
+    assert status == direct.returncode == (0 if valid_digest else 1)
+    assert observed == json.loads(direct.stdout)
+    assert capture.read_bytes() == raw
+    assert not registry.exists() and not runtime.exists()
diff --git a/tests/control_plane/test_todo_source_runtime_route.py b/tests/control_plane/test_todo_source_runtime_route.py
index de9b3394d2..a4da9b975c 100644
--- a/tests/control_plane/test_todo_source_runtime_route.py
+++ b/tests/control_plane/test_todo_source_runtime_route.py
@@ -70,7 +70,6 @@ def test_global_cli_check_and_manager_read_the_original_provider(routed):
     expected = list_goal_todos(registry_path=source, goal_id=GOAL, json_references=True)
     code, actual = run_json_cli_result("todo", "list", "--goal-id", GOAL, registry_path=registry)
     assert code == 0, actual
-    assert actual.pop("_returncode") == 0  # Test harness metadata, outside the CLI payload.
     assert_todo_read_unchanged(actual, expected)
     assert actual["todo_count"] == 5
     assert {row["todo_id"] for row in actual["todos"]} == {f"todo_source_{i}" for i in range(5)}