fix: redact private paths from identity receipts

docs/leo-reproducible-identity.md

@@ -170,6 +170,12 @@ python -m scripts.leo_identity_profile compile \
 All commands require a clean Git worktree because a commit plus an unretained
 dirty source tree is not reproducible.
 
+Child processes execute the real interpreter and temporary-profile paths only
+in process-local memory. Retained receipts persist a repo-relative
+`logical_command`, the stable `<python>` and `<temporary-profile>` placeholders,
+and an explicit `profile_role`; they never serialize the checkout, interpreter,
+or temporary directory path.
+
 ## State inventory and transitions
 
 | State | Meaning | Next valid transition |

scripts/run_leo_identity_reconstruction_canary.py

@@ -13,7 +13,7 @@ import sys
 import tempfile
 import time
 from datetime import UTC, datetime
-from pathlib import Path
+from pathlib import Path, PurePosixPath, PureWindowsPath
 from typing import Any
 
 if __package__ in {None, ""}:
@@ -24,6 +24,7 @@ from scripts import leo_identity_manifest as identity
 from scripts import leo_identity_profile as profile_runtime
 
 SCHEMA = "livingip.leoIdentityReconstructionReceipt.v1"
+PRIVATE_PATH_PREFIXES = tuple("/" + value for value in ("Users/", "private/tmp/", "var/folders/"))
 
 
 class CanaryError(RuntimeError):
@@ -35,6 +36,25 @@ def _require(condition: bool, message: str) -> None:
         raise CanaryError(message)
 
 
+def _receipt_strings(value: object) -> list[str]:
+    if isinstance(value, dict):
+        keys = [key for key in value if isinstance(key, str)]
+        return keys + [text for item in value.values() for text in _receipt_strings(item)]
+    if isinstance(value, list):
+        return [text for item in value for text in _receipt_strings(item)]
+    return [value] if isinstance(value, str) else []
+
+
+def _validate_portable_receipt(value: object) -> None:
+    for text in _receipt_strings(value):
+        if (
+            any(prefix in text for prefix in PRIVATE_PATH_PREFIXES)
+            or PurePosixPath(text).is_absolute()
+            or PureWindowsPath(text).is_absolute()
+        ):
+            raise CanaryError("receipt contains a private or absolute filesystem path")
+
+
 def _process_group_absent(process_group: int) -> bool:
     try:
         os.killpg(process_group, 0)
@@ -72,15 +92,55 @@ def _cleanup_remaining_process_group(process_group: int) -> str | None:
     return "SIGKILL"
 
 
+def _logical_repo_path(path: Path, *, deployment_root: Path, placeholder: str) -> str:
+    """Return a stable repo-relative path or a non-identifying placeholder."""
+
+    try:
+        relative = path.resolve().relative_to(deployment_root.resolve())
+    except (OSError, ValueError):
+        return placeholder
+    value = relative.as_posix()
+    return value if value != "." else "."
+
+
+def _logical_query_command(
+    *,
+    deployment_root: Path,
+    source_root: Path,
+    hermes_root: Path,
+) -> list[str]:
+    return [
+        "<python>",
+        "-m",
+        "scripts.leo_identity_profile",
+        "query",
+        "--profile",
+        "<temporary-profile>",
+        "--source-root",
+        _logical_repo_path(source_root, deployment_root=deployment_root, placeholder="<source-root>"),
+        "--hermes-root",
+        _logical_repo_path(hermes_root, deployment_root=deployment_root, placeholder="<hermes-root>"),
+        "--deployment-root",
+        ".",
+        "--query",
+        profile_runtime.FIXED_QUERY,
+    ]
+
+
 def _run_query_child(
     *,
     deployment_root: Path,
     profile: Path,
     source_root: Path,
     hermes_root: Path,
+    profile_role: str,
     timeout_seconds: float = 60,
 ) -> dict[str, Any]:
-    command = [
+    deployment_root = deployment_root.resolve()
+    profile = profile.resolve()
+    source_root = source_root.resolve()
+    hermes_root = hermes_root.resolve()
+    actual_argv = [
         sys.executable,
         "-m",
         "scripts.leo_identity_profile",
@@ -97,7 +157,7 @@ def _run_query_child(
         profile_runtime.FIXED_QUERY,
     ]
     process = subprocess.Popen(
-        command,
+        actual_argv,
         cwd=deployment_root,
         env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
         text=True,
@@ -143,7 +203,14 @@ def _run_query_child(
     except json.JSONDecodeError as exc:
         raise CanaryError("identity child returned invalid JSON") from exc
     return {
-        "command": command,
+        "logical_command": _logical_query_command(
+            deployment_root=deployment_root,
+            source_root=source_root,
+            hermes_root=hermes_root,
+        ),
+        "command_serialization": "logical_redacted_v1",
+        "profile_role": profile_role,
+        "actual_argv_persisted": False,
         "launcher_pid": process.pid,
         "returncode": process.returncode,
         "timed_out": timed_out,
@@ -181,6 +248,7 @@ def _drift_attempt(
         profile=drift_profile,
         source_root=source_root,
         hermes_root=hermes_root,
+        profile_role="drift_negative",
     )
     child["fail_closed"] = bool(
         child["returncode"] == 3
@@ -250,6 +318,7 @@ def run_canary(args: argparse.Namespace) -> tuple[dict[str, Any], int]:
             profile=compiled_profile,
             source_root=args.source_root,
             hermes_root=args.hermes_root,
+            profile_role="first_start",
         )
         report["first_start"] = first
         _require(_query_passed(first), "first disposable identity process did not answer successfully")
@@ -271,6 +340,7 @@ def run_canary(args: argparse.Namespace) -> tuple[dict[str, Any], int]:
                 profile=compiled_profile,
                 source_root=args.source_root,
                 hermes_root=args.hermes_root,
+                profile_role="pre_restart_drift",
             )
             report["pre_restart_drift"] = rejected
             report["drift_target"] = "compiled SOUL.md generated view"
@@ -315,6 +385,7 @@ def run_canary(args: argparse.Namespace) -> tuple[dict[str, Any], int]:
                 profile=restarted_profile,
                 source_root=args.source_root,
                 hermes_root=args.hermes_root,
+                profile_role="restart",
             )
             report["restart"] = second
             _require(_query_passed(second), "restarted disposable identity process did not answer successfully")
@@ -370,7 +441,13 @@ def run_canary(args: argparse.Namespace) -> tuple[dict[str, Any], int]:
             exit_code = 0
     except (CanaryError, identity.IdentityManifestError, OSError, subprocess.SubprocessError) as exc:
         report["status"] = "fail"
-        report["errors"].append({"type": type(exc).__name__, "message": str(exc)})
+        report["errors"].append(
+            {
+                "type": type(exc).__name__,
+                "message": "details_redacted",
+                "detail_sha256": behavior.sha256_bytes(str(exc).encode("utf-8")),
+            }
+        )
     finally:
         shutil.rmtree(temporary_root, ignore_errors=True)
         report["cleanup"] = {
@@ -402,6 +479,7 @@ def run_canary(args: argparse.Namespace) -> tuple[dict[str, Any], int]:
             else "Local first-process and pre-restart drift-refusal component only; no successful restart proof, "
             "GatewayRunner, hosted model, production database, VPS, or T3 claim."
         )
+        _validate_portable_receipt(report)
         stable = {key: value for key, value in report.items() if key != "receipt_sha256"}
         report["receipt_sha256"] = behavior.canonical_sha256(stable)
         identity.write_json(output, report)

tests/test_leo_identity_manifest.py

@@ -6,7 +6,7 @@ import json
 import os
 import subprocess
 import time
-from pathlib import Path
+from pathlib import Path, PureWindowsPath
 
 import pytest
 
@@ -16,6 +16,10 @@ from scripts import leo_identity_profile as profile_runtime
 from scripts import run_leo_identity_reconstruction_canary as canary
 
 ROOT = Path(__file__).resolve().parents[1]
+COMMITTED_RECEIPTS = (
+    ROOT / "docs/reports/leo-working-state-20260709/leo-reproducible-identity-t2-current.json",
+    ROOT / "docs/reports/leo-working-state-20260709/leo-reproducible-identity-drift-negative-current.json",
+)
 
 
 def _write(path: Path, value: str) -> None:
@@ -27,6 +31,21 @@ def _write_json(path: Path, value: dict) -> None:
     _write(path, json.dumps(value, indent=2, sort_keys=True) + "\n")
 
 
+def _assert_path_portable_receipt(value: object) -> None:
+    canary._validate_portable_receipt(value)
+
+
+def _assert_logical_child_execution(child: dict, *, profile_role: str) -> None:
+    logical_command = child["logical_command"]
+    assert "command" not in child
+    assert child["profile_role"] == profile_role
+    assert child["command_serialization"] == "logical_redacted_v1"
+    assert child["actual_argv_persisted"] is False
+    assert logical_command[0] == "<python>"
+    assert logical_command[logical_command.index("--profile") + 1] == "<temporary-profile>"
+    _assert_path_portable_receipt(child)
+
+
 def _fixture(tmp_path: Path) -> dict[str, Path | dict]:
     repo = tmp_path / "repo"
     profile = repo / "profile-template"
@@ -224,6 +243,50 @@ def _canary_args(fixture: dict[str, Path | dict], output: Path, *, inject_drift:
     )
 
 
+@pytest.mark.parametrize(
+    "private_path",
+    [
+        "/" + "Users/operator/checkout/profile",
+        "/" + "private/tmp/lane/.venv/bin/python",
+        "/" + "var/folders/cache/temporary-profile",
+        "/" + "tmp/temporary-profile",
+        "/" + "opt/checkout/interpreter",
+        "/" + "workspace/checkout/temporary-profile",
+        str(PureWindowsPath("C:/operator/checkout/temporary-profile")),
+        "\\\\" + "server\\share\\checkout\\temporary-profile",
+    ],
+)
+def test_receipt_path_guard_recursively_rejects_private_and_absolute_paths(private_path: str) -> None:
+    with pytest.raises(canary.CanaryError, match="private or absolute filesystem path"):
+        _assert_path_portable_receipt({"outer": [{"logical_command": ["<python>", private_path]}]})
+
+
+def test_receipt_path_guard_rejects_an_absolute_path_in_a_nested_key() -> None:
+    private_key = "/" + "tmp/checkout/temporary-profile"
+    with pytest.raises(canary.CanaryError, match="private or absolute filesystem path"):
+        _assert_path_portable_receipt({"outer": [{private_key: "<temporary-profile>"}]})
+
+
+def test_logical_command_uses_placeholders_for_paths_outside_the_deployment_root(tmp_path: Path) -> None:
+    logical_command = canary._logical_query_command(
+        deployment_root=tmp_path / "checkout",
+        source_root=tmp_path / "outside-source",
+        hermes_root=tmp_path / "outside-hermes",
+    )
+
+    assert logical_command[logical_command.index("--source-root") + 1] == "<source-root>"
+    assert logical_command[logical_command.index("--hermes-root") + 1] == "<hermes-root>"
+    _assert_path_portable_receipt(logical_command)
+
+
+@pytest.mark.parametrize("receipt_path", COMMITTED_RECEIPTS, ids=lambda path: path.name)
+def test_committed_identity_receipts_are_path_portable_and_self_hashed(receipt_path: Path) -> None:
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    _assert_path_portable_receipt(receipt)
+    stable = {key: value for key, value in receipt.items() if key != "receipt_sha256"}
+    assert behavior.canonical_sha256(stable) == receipt["receipt_sha256"]
+
+
 def test_manifest_compiles_generated_views_and_reproduces_identity_across_queries(tmp_path: Path) -> None:
     fixture = _fixture(tmp_path)
     manifest = fixture["manifest"]
@@ -493,7 +556,8 @@ def test_fully_rehashed_model_claim_must_match_observed_runtime(tmp_path: Path)
 
 def test_restart_canary_uses_distinct_processes_and_cleans_up(tmp_path: Path) -> None:
     fixture = _fixture(tmp_path)
-    report, exit_code = canary.run_canary(_canary_args(fixture, tmp_path / "restart-receipt.json"))
+    output = tmp_path / "restart-receipt.json"
+    report, exit_code = canary.run_canary(_canary_args(fixture, output))
 
     assert exit_code == 0
     assert report["status"] == "pass"
@@ -501,9 +565,10 @@ def test_restart_canary_uses_distinct_processes_and_cleans_up(tmp_path: Path) ->
     assert report["gates"]["restart_profile_recompiled_from_manifest"] is True
     assert report["restart_compile"]["session_directories_started_empty"] is True
     assert report["drift_compile"]["session_directories_started_empty"] is True
-    first_profile = report["first_start"]["command"][report["first_start"]["command"].index("--profile") + 1]
+    _assert_logical_child_execution(report["first_start"], profile_role="first_start")
-    restart_profile = report["restart"]["command"][report["restart"]["command"].index("--profile") + 1]
+    _assert_logical_child_execution(report["restart"], profile_role="restart")
-    assert first_profile != restart_profile
+    _assert_logical_child_execution(report["drift_negative"], profile_role="drift_negative")
+    _assert_path_portable_receipt(report)
     assert report["first_start"]["launcher_pid"] != report["restart"]["launcher_pid"]
     assert report["first_start"]["process_group_absent_after_wait"] is True
     assert report["restart"]["process_group_absent_after_wait"] is True
@@ -513,22 +578,28 @@ def test_restart_canary_uses_distinct_processes_and_cleans_up(tmp_path: Path) ->
     assert report["current_tier"] == "T2_runtime"
     assert report["cleanup"]["temporary_root_removed"] is True
     assert behavior.git_state(fixture["repo"])["worktree_clean"] is True
+    assert json.loads(output.read_text(encoding="utf-8")) == report
 
 
 def test_restart_canary_rejects_drift_before_second_start(tmp_path: Path) -> None:
     fixture = _fixture(tmp_path)
-    report, exit_code = canary.run_canary(_canary_args(fixture, tmp_path / "drift-receipt.json", inject_drift=True))
+    output = tmp_path / "drift-receipt.json"
+    report, exit_code = canary.run_canary(_canary_args(fixture, output, inject_drift=True))
 
     assert exit_code == 0
     assert report["status"] == "pass"
     assert report["gates"]["drift_detected_before_restart"] is True
     assert report["second_start_attempted"] is False
     assert "restart" not in report
+    _assert_logical_child_execution(report["first_start"], profile_role="first_start")
+    _assert_logical_child_execution(report["pre_restart_drift"], profile_role="pre_restart_drift")
+    _assert_path_portable_receipt(report)
     assert report["pre_restart_drift"]["process_group_absent_after_wait"] is True
     assert all(report["gates"].values())
     assert report["goal_tier_satisfied"] is False
     assert report["current_tier"] == "T1_model"
     assert report["cleanup"]["temporary_root_removed"] is True
+    assert json.loads(output.read_text(encoding="utf-8")) == report
 
 
 def test_query_child_timeout_terminates_process_group(tmp_path: Path) -> None:
@@ -550,6 +621,7 @@ time.sleep(60)
         profile=fixture["compiled"],
         source_root=fixture["repo"],
         hermes_root=fixture["hermes"],
+        profile_role="timeout_test",
         timeout_seconds=0.5,
     )
 
@@ -594,6 +666,7 @@ print(json.dumps({query_payload!r}))
         profile=fixture["compiled"],
         source_root=fixture["repo"],
         hermes_root=fixture["hermes"],
+        profile_role="orphan_cleanup_test",
         timeout_seconds=5,
     )