PR #8599 — qualify Windows file identity and bind backup evidence

loopx/control_plane/coordination/cold_source_backup.py

@@ -14,6 +14,7 @@
 import json
 import os
 from pathlib import Path, PurePosixPath
+import posixpath
 import shutil
 import sqlite3
 import stat
@@ -85,9 +86,13 @@ def _open_regular(path: Path) -> tuple[BinaryIO, os.stat_result]:
 
 
 def _same_file(before: os.stat_result, after: os.stat_result) -> bool:
+    identity_fields: tuple[str, ...] = ("st_dev", "st_ino", "st_size", "st_mtime_ns")
+    # Only a complete Windows volume/file identity can tolerate ctime skew.
+    if os.name != "nt" or not all(info.st_dev and info.st_ino for info in (before, after)):
+        identity_fields += ("st_ctime_ns",)
     return all(
         getattr(before, field) == getattr(after, field)
-        for field in ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns")
+        for field in identity_fields
     )
 
 
@@ -114,7 +119,7 @@ def _safe_link(name: str, target: str, *, hardlink: bool) -> str:
         raise ColdSourceBackupError("cold_import_backup_link_unsafe")
     if hardlink:
         return _member_name(target)
-    resolved = os.path.normpath(str(PurePosixPath(name).parent / target))
+    resolved = posixpath.normpath(str(PurePosixPath(name).parent / target))
     if resolved == ".." or resolved.startswith("../"):
         raise ColdSourceBackupError("cold_import_backup_link_unsafe")
     return _member_name(resolved)
@@ -190,8 +195,8 @@ def _matches(name: str, prefixes: list[str] | None) -> bool:
 
 
 def _scan_archive(
-    archive: BinaryIO, *, selected: list[str] | None, workspace: Path
-) -> tuple[dict[str, Any], list[dict[str, Any]], int, int, int]:
+    archive: BinaryIO, *, execution: dict[str, Any], selected: list[str] | None, workspace: Path
+) -> tuple[dict[str, Any], list[dict[str, Any]], int, int, int, str, int]:
     workspace.mkdir(mode=0o700, parents=True, exist_ok=True)
     workspace_info = workspace.lstat()
     if not stat.S_ISDIR(workspace_info.st_mode):
@@ -205,9 +210,20 @@ def _scan_archive(
             archive.seek(0)
             # One disposable transaction avoids per-prefix durable writes.
             # This scratch index is not a Goal authority or recovery store.
-            with closing(sqlite3.connect(database_path)) as database:
+            with (
+                closing(sqlite3.connect(database_path)) as database,
+                (Path(temporary) / "archive.tar.gz").open("w+b") as snapshot,
+            ):
+                # Hash and scan the same private bytes even if the source is overwritten and restored.
+                shutil.copyfileobj(archive, snapshot, _CHUNK)
+                original_sha, archive_size = _sha(snapshot)
+                if original_sha != execution.get("archive_sha256"):
+                    raise ColdSourceBackupError("cold_import_backup_archive_changed")
+                declared_size = execution.get("archive_size_bytes")
+                if declared_size is not None and declared_size != archive_size:
+                    raise ColdSourceBackupError("cold_import_backup_archive_changed")
                 database.execute("CREATE TABLE members (name BLOB PRIMARY KEY, marker BLOB NOT NULL) WITHOUT ROWID")
-                with tarfile.open(fileobj=archive, mode="r|gz") as tar:
+                with tarfile.open(fileobj=snapshot, mode="r|gz") as tar:
                     # r|gz streams bytes but still caches every TarInfo on
                     # supported Python versions. No random member lookup is
                     # needed: hardlink coverage already lives in the disk DB.
@@ -282,7 +298,7 @@ def _scan_archive(
         raise
     if embedded is None:
         raise ColdSourceBackupError("cold_import_backup_manifest_missing")
-    return embedded, members, member_count, logical_bytes, workspace_bytes
+    return embedded, members, member_count, logical_bytes, workspace_bytes, original_sha, archive_size
 
 
 def read_cold_source_backup(
@@ -319,15 +335,10 @@ def read_cold_source_backup(
     archive_path = Path(str(execution.get("archive_path") or "")).expanduser().absolute()
     archive, before = _open_regular(archive_path)
     with archive:
-        original_sha, archive_size = _sha(archive)
-        if original_sha != execution.get("archive_sha256"):
-            raise ColdSourceBackupError("cold_import_backup_archive_changed")
-        declared_size = execution.get("archive_size_bytes")
-        if declared_size is not None and declared_size != archive_size:
-            raise ColdSourceBackupError("cold_import_backup_archive_changed")
         scan_workspace = (workspace or manifest_path.parent / ".cold-import-work").expanduser().absolute()
-        embedded, members, member_count, logical_bytes, workspace_bytes = _scan_archive(
-            archive, selected=selected, workspace=scan_workspace
+        (embedded, members, member_count, logical_bytes, workspace_bytes,
+            original_sha, archive_size) = _scan_archive(
+            archive, execution=execution, selected=selected, workspace=scan_workspace
         )
         after = os.fstat(archive.fileno())
         if not _same_file(before, after):

loopx/materials.py

@@ -222,7 +222,12 @@ def read_bounded_material_file(path: Path, max_bytes: int) -> tuple[Path, bytes,
         after = os.fstat(handle.fileno())
         resolved = path.resolve()
         current = resolved.stat()
-        identity_fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns")
+        identity_fields: tuple[str, ...] = ("st_dev", "st_ino", "st_size", "st_mtime_ns")
+        # Tolerate Windows ctime skew only with a volume and file index in every observation.
+        if os.name != "nt" or not all(
+            observed.st_dev and observed.st_ino for observed in (before, after, current)
+        ):
+            identity_fields += ("st_ctime_ns",)
         identity = tuple(getattr(before, field) for field in identity_fields)
         if len(content) != before.st_size or any(
             tuple(getattr(observed, field) for field in identity_fields) != identity

tests/capabilities/test_decision_context_profile.py

@@ -411,6 +411,78 @@ def stale_preflight(path, *args, **kwargs):
     assert grew
 
 
+@pytest.mark.parametrize("platform", ["nt", "posix"])
+@pytest.mark.parametrize("descriptor_change", [False, True])
+def test_local_file_ctime_skew(tmp_path, monkeypatch, platform, descriptor_change):
+    import os
+    from types import SimpleNamespace
+
+    from loopx import materials
+
+    authority = tmp_path / "authority.txt"
+    authority.write_text("stable source", encoding="utf-8")
+    profile = normalize_decision_context_profile(profile_payload(authority))
+    provider = build_decision_source_provider(profile.provider_binding_map()["local-authority"])
+    original_fstat = os.fstat
+    calls = 0
+
+    def descriptor_stat(fd):
+        nonlocal calls
+        observed = original_fstat(fd)
+        calls += 1
+        fields = {name: getattr(observed, name) for name in dir(observed) if name.startswith("st_")}
+        fields["st_ctime_ns"] = authority.stat().st_ctime_ns + 499_700
+        if descriptor_change and calls == 2:
+            fields["st_ctime_ns"] += 1
+        return SimpleNamespace(**fields)
+
+    monkeypatch.setattr(materials, "os", SimpleNamespace(**{
+        **vars(os), "name": platform, "fstat": descriptor_stat,
+    }))
+    args = dict(source=profile.sources[0], after_cursor=None,
+        before="2100-01-01T00:00:00+00:00", limit=8,
+        timeout_seconds=2, observed_at=OBSERVED_AT)
+    if platform == "posix":
+        with pytest.raises(ValueError, match="changed during bounded read"):
+            provider.scan(**args)
+    else:
+        scan = provider.scan(**args)
+        exact = provider.exact_read(source=profile.sources[0], item=scan.items[0],
+            timeout_seconds=2, observed_at=OBSERVED_AT)
+        assert scan.status == "completed"
+        assert exact.content == "stable source"
+
+
+@pytest.mark.parametrize("change", ["mtime", "size"])
+def test_local_file_path_change_after_descriptor_snapshot(tmp_path, monkeypatch, change):
+    import os
+
+    authority = tmp_path / "authority.txt"
+    authority.write_bytes(b"before")
+    profile = normalize_decision_context_profile(profile_payload(authority))
+    provider = build_decision_source_provider(profile.provider_binding_map()["local-authority"])
+    original_fstat = os.fstat
+    calls = 0
+
+    def change_after_snapshot(fd):
+        nonlocal calls
+        observed = original_fstat(fd)
+        calls += 1
+        if calls == 2:
+            if change == "size":
+                authority.write_bytes(b"longer source")
+            os.utime(authority, ns=(observed.st_atime_ns,
+                observed.st_mtime_ns + (1_000_000_000 if change == "mtime" else 0)))
+        return observed
+
+    monkeypatch.setattr(os, "fstat", change_after_snapshot)
+    with pytest.raises(ValueError, match="changed during bounded read"):
+        provider.scan(source=profile.sources[0], after_cursor=None,
+            before="2100-01-01T00:00:00+00:00", limit=8,
+            timeout_seconds=2, observed_at=OBSERVED_AT)
+    assert calls == 2
+
+
 def test_local_file_growth_during_read_is_bounded(tmp_path, monkeypatch):
     import os
 
@@ -460,6 +532,86 @@ def read(self, size=-1):
     assert read_sizes == [17]
 
 
+@pytest.mark.parametrize("missing_identity", [None, "st_ino", "st_dev"])
+def test_local_file_same_metadata_replacement_has_no_verified_read(tmp_path, monkeypatch, missing_identity):
+    import os
+    import subprocess
+    from types import SimpleNamespace
+
+    from loopx import materials
+
+    original_dir, replacement_dir = tmp_path / "original", tmp_path / "replacement"
+    original_dir.mkdir()
+    replacement_dir.mkdir()
+    original = original_dir / "authority.txt"
+    original.write_bytes(b"original")
+    os.utime(original, (1_700_000_000, 1_700_000_000))
+    before = original.stat()
+    replacement = replacement_dir / "authority.txt"
+    replacement.write_bytes(b"replaced")
+    os.utime(replacement, ns=(before.st_atime_ns, before.st_mtime_ns))
+    assert replacement.stat().st_ino != before.st_ino
+    assert replacement.stat().st_size == before.st_size
+    assert replacement.stat().st_mtime_ns == before.st_mtime_ns
+    alias = tmp_path / "alias"
+
+    def link_to(target):
+        if os.name == "nt":
+            created = subprocess.run(
+                ["cmd.exe", "/d", "/c", "mklink", "/J", str(alias), str(target)],
+                capture_output=True,
+            )
+            assert created.returncode == 0, created.stderr or created.stdout
+        else:
+            alias.symlink_to(target, target_is_directory=True)
+
+    link_to(original_dir)
+    authority = alias / "authority.txt"
+    profile = normalize_decision_context_profile(profile_payload(authority))
+    provider = build_decision_source_provider(profile.provider_binding_map()["local-authority"])
+    original_fstat, original_stat = os.fstat, Path.stat
+    observations = 0
+
+    def windows_stat(observed):
+        fields = {name: getattr(observed, name) for name in dir(observed) if name.startswith("st_")}
+        fields["st_ctime_ns"] = before.st_ctime_ns + (observed.st_ino != before.st_ino)
+        if missing_identity:
+            fields[missing_identity] = 0
+        if missing_identity == "st_dev":
+            # File indexes may collide across volumes; inject only the OS metadata.
+            fields["st_ino"] = before.st_ino
+        return SimpleNamespace(**fields)
+
+    def descriptor_stat(fd):
+        nonlocal observations
+        observed = original_fstat(fd)
+        observations += 1
+        if observations == 2:
+            # Retarget the directory while the original file is open on Windows.
+            if os.name == "nt":
+                alias.rmdir()
+            else:
+                alias.unlink()
+            link_to(replacement_dir)
+            assert authority.read_bytes() == b"replaced"
+        return windows_stat(observed)
+
+    def path_stat(path, *args, **kwargs):
+        observed = original_stat(path, *args, **kwargs)
+        return windows_stat(observed) if path in (authority, original, replacement) else observed
+
+    monkeypatch.setattr(materials, "os", SimpleNamespace(**{
+        **vars(os), "name": "nt", "fstat": descriptor_stat,
+    }))
+    monkeypatch.setattr(Path, "stat", path_stat)
+    with pytest.raises(ValueError, match="changed during bounded read"):
+        provider.scan(source=profile.sources[0], after_cursor=None,
+            before="2100-01-01T00:00:00+00:00", limit=8,
+            timeout_seconds=2, observed_at=OBSERVED_AT)
+    assert observations == 2
+    assert authority.read_bytes() == b"replaced"
+
+
 def test_local_file_replacement_after_open_has_no_verified_read(tmp_path, monkeypatch):
     import os
 

tests/control_plane/test_cold_source_backup_scope.py

@@ -4,9 +4,11 @@
 import hashlib
 import io
 import json
+import os
 from pathlib import Path
 import sqlite3
 import tarfile
+from types import SimpleNamespace
 
 import pytest
 
@@ -140,6 +142,66 @@ def test_scoping_does_not_hide_unsafe_unrelated_members_or_changed_artifact(tmp_
         read_cold_source_backup(manifest, source_paths=[tmp_path / "runtime/goals/example"])
 
 
+@pytest.mark.parametrize("unavailable_identity", [None, "st_dev", "st_ino"])
+@pytest.mark.parametrize("platform", ["nt", "posix"])
+@pytest.mark.parametrize("field", ["st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns"])
+def test_archive_identity_change_policy(tmp_path, monkeypatch, platform, field, unavailable_identity):
+    from loopx.control_plane.coordination import cold_source_backup
+
+    manifest = backup(tmp_path)
+    archive = tmp_path / "state.tar.gz"
+    archive_stat = archive.stat()
+    assert archive_stat.st_dev and archive_stat.st_ino
+    original_fstat = os.fstat
+    original_lstat = Path.lstat
+    observations = 0
+
+    def partial_identity(observed):
+        if unavailable_identity is None or (observed.st_dev, observed.st_ino) != (
+            archive_stat.st_dev, archive_stat.st_ino
+        ):
+            return observed
+        # Unavailable volume/file indexes are injected only at the stat boundary.
+        fields = {name: getattr(observed, name) for name in dir(observed) if name.startswith("st_")}
+        fields[unavailable_identity] = 0
+        return SimpleNamespace(**fields)
+
+    def path_stat(path, *args, **kwargs):
+        return partial_identity(original_lstat(path, *args, **kwargs))
+
+    def descriptor_stat(fd):
+        nonlocal observations
+        observed = original_fstat(fd)
+        if (observed.st_dev, observed.st_ino) != (archive_stat.st_dev, archive_stat.st_ino):
+            return observed
+        observations += 1
+        observed = partial_identity(observed)
+        if observations != 2:
+            return observed
+        fields = {name: getattr(observed, name) for name in dir(observed) if name.startswith("st_")}
+        fields[field] += 1
+        return SimpleNamespace(**fields)
+
+    monkeypatch.setattr(Path, "lstat", path_stat)
+    monkeypatch.setattr(cold_source_backup, "os", SimpleNamespace(**{
+        **vars(os), "name": platform, "fstat": descriptor_stat,
+    }))
+    if platform == "nt" and field == "st_ctime_ns" and unavailable_identity is None:
+        witness = read_cold_source_backup(manifest)
+        assert witness["archive_sha256"] == hashlib.sha256(archive.read_bytes()).hexdigest()
+        assert witness["archive_size_bytes"] == archive_stat.st_size
+        assert witness["members"] == [
+            {"archive_path": "runtime/goals/example/state.md",
+             "sha256": hashlib.sha256(b"retained source").hexdigest()},
+            {"archive_path": "runtime/goals/example/task-leases/settled.json",
+             "sha256": hashlib.sha256(b"retained lease").hexdigest()},
+        ]
+    else:
+        with pytest.raises(ColdSourceBackupError, match="cold_import_backup_changed_retry"):
+            read_cold_source_backup(manifest)
+    assert observations == 2
+
+
 def test_streaming_witness_has_no_old_expanded_byte_ceiling(tmp_path):
     archive = tmp_path / "large-expanded.tar.gz"
     source = tmp_path / "source.bin"
@@ -200,6 +262,8 @@ def unsafe_manifest(name: str, members: list[tarfile.TarInfo]) -> Path:
 
     link = tarfile.TarInfo("root/link")
     link.type, link.linkname = tarfile.SYMTYPE, "target"
+    assert read_cold_source_backup(unsafe_manifest("first-link", [link]),
+        source_paths=[tmp_path])["members"] == []
     descendant = tarfile.TarInfo("root/link/file")
     descendant.size = 1
     with pytest.raises(ValueError, match="member_unsafe"):