From ad9025ed9fae29fd3c7627306005ab6e427f4d53 Mon Sep 17 00:00:00 2001 From: xiaobaifly7 <196511160+xiaobaifly7@users.noreply.github.com> Date: Wed, 11 Nov 2026 19:59:53 +0800 Subject: [PATCH 1/5] fix(materials): ignore Windows ctime-only skew Signed-off-by: xiaobaifly7 <196511160+xiaobaifly7@users.noreply.github.com> --- loopx/materials.py | 5 +- .../test_decision_context_profile.py | 72 +++++++++++++++++++ 2 files changed, 76 insertions(+), 1 deletion(-) diff --git a/loopx/materials.py b/loopx/materials.py index e4b3196f4e..11aa7cbe04 100644 --- a/loopx/materials.py +++ b/loopx/materials.py @@ -222,7 +222,10 @@ 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 = ("st_dev", "st_ino", "st_size", "st_mtime_ns") + # Windows ctime alone does not reliably identify a file change. + if os.name != "nt": + 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 diff --git a/tests/capabilities/test_decision_context_profile.py b/tests/capabilities/test_decision_context_profile.py index 7a8c9ac8d0..98dff2cb16 100644 --- a/tests/capabilities/test_decision_context_profile.py +++ b/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 From 6ac8672c9b2667426e5adbce5b728c2df8e11d7c Mon Sep 17 00:00:00 2001 From: xiaobaifly7 <196511160+xiaobaifly7@users.noreply.github.com> Date: Wed, 11 Nov 2026 20:23:25 +0800 Subject: [PATCH 2/5] fix(io): ignore Windows ctime-only skew in cold backup Signed-off-by: xiaobaifly7 <196511160+xiaobaifly7@users.noreply.github.com> --- .../coordination/cold_source_backup.py | 6 ++- .../test_cold_source_backup_scope.py | 44 +++++++++++++++++++ 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/loopx/control_plane/coordination/cold_source_backup.py b/loopx/control_plane/coordination/cold_source_backup.py index 2d0cb1613d..bc5e2eb9e3 100644 --- a/loopx/control_plane/coordination/cold_source_backup.py +++ b/loopx/control_plane/coordination/cold_source_backup.py @@ -85,9 +85,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 = ("st_dev", "st_ino", "st_size", "st_mtime_ns") + # Windows ctime alone does not reliably identify a file change. + if os.name != "nt": + 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 ) diff --git a/tests/control_plane/test_cold_source_backup_scope.py b/tests/control_plane/test_cold_source_backup_scope.py index d34f33cc17..1007f6c7e6 100644 --- a/tests/control_plane/test_cold_source_backup_scope.py +++ b/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,48 @@ 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("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): + from loopx.control_plane.coordination import cold_source_backup + + manifest = backup(tmp_path) + archive = tmp_path / "state.tar.gz" + archive_stat = archive.stat() + original_fstat = os.fstat + observations = 0 + + 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 + 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(cold_source_backup, "os", SimpleNamespace(**{ + **vars(os), "name": platform, "fstat": descriptor_stat, + })) + if platform == "nt" and field == "st_ctime_ns": + 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" From a611c850953ccd7b6c6167b8711dd7dcf8d77144 Mon Sep 17 00:00:00 2001 From: xiaobaifly7 <196511160+xiaobaifly7@users.noreply.github.com> Date: Thu, 12 Nov 2026 01:13:47 +0800 Subject: [PATCH 3/5] fix(materials): retain ctime checks without file identity Signed-off-by: xiaobaifly7 <196511160+xiaobaifly7@users.noreply.github.com> --- loopx/materials.py | 4 +- .../test_decision_context_profile.py | 50 +++++++++++++++++++ 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/loopx/materials.py b/loopx/materials.py index 11aa7cbe04..4f0af2913d 100644 --- a/loopx/materials.py +++ b/loopx/materials.py @@ -223,8 +223,8 @@ def read_bounded_material_file(path: Path, max_bytes: int) -> tuple[Path, bytes, resolved = path.resolve() current = resolved.stat() identity_fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns") - # Windows ctime alone does not reliably identify a file change. - if os.name != "nt": + # Tolerate Windows ctime skew only when every observation has a file index. + if os.name != "nt" or not all(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( diff --git a/tests/capabilities/test_decision_context_profile.py b/tests/capabilities/test_decision_context_profile.py index 98dff2cb16..d21962a2fd 100644 --- a/tests/capabilities/test_decision_context_profile.py +++ b/tests/capabilities/test_decision_context_profile.py @@ -532,6 +532,56 @@ def read(self, size=-1): assert read_sizes == [17] +@pytest.mark.skipif(__import__("os").name == "nt", reason="Open-file replacement requires POSIX sharing") +@pytest.mark.parametrize("file_index_available", [False, True]) +def test_local_file_same_metadata_replacement_has_no_verified_read(tmp_path, monkeypatch, file_index_available): + import os + from types import SimpleNamespace + + from loopx import materials + + authority = tmp_path / "authority.txt" + authority.write_bytes(b"original") + os.utime(authority, (1_700_000_000, 1_700_000_000)) + before = authority.stat() + replacement = tmp_path / "replacement.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 + 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 not file_index_available: + fields["st_ino"] = 0 + return SimpleNamespace(**fields) + + def descriptor_stat(fd): + nonlocal observations + observed = original_fstat(fd) + observations += 1 + if observations == 2: + replacement.replace(authority) + return windows_stat(observed) + + def path_stat(path, *args, **kwargs): + observed = original_stat(path, *args, **kwargs) + return windows_stat(observed) if path == authority 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"): + materials.read_bounded_material_file(authority, 64) + 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 From 8c46abc3ea199a44d7191838c0d09f77754eee7e Mon Sep 17 00:00:00 2001 From: xiaobaifly7 <196511160+xiaobaifly7@users.noreply.github.com> Date: Thu, 12 Nov 2026 01:14:03 +0800 Subject: [PATCH 4/5] fix(backup): scan members from the verified private archive Signed-off-by: xiaobaifly7 <196511160+xiaobaifly7@users.noreply.github.com> --- .../coordination/cold_source_backup.py | 32 +++++---- .../test_cold_source_import_cli.py | 72 +++++++++++++++++++ 2 files changed, 91 insertions(+), 13 deletions(-) diff --git a/loopx/control_plane/coordination/cold_source_backup.py b/loopx/control_plane/coordination/cold_source_backup.py index bc5e2eb9e3..194a90fdc1 100644 --- a/loopx/control_plane/coordination/cold_source_backup.py +++ b/loopx/control_plane/coordination/cold_source_backup.py @@ -194,8 +194,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): @@ -209,9 +209,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. @@ -286,7 +297,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( @@ -323,15 +334,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): diff --git a/tests/control_plane/test_cold_source_import_cli.py b/tests/control_plane/test_cold_source_import_cli.py index 5071d6bbd7..c49eaf5d30 100644 --- a/tests/control_plane/test_cold_source_import_cli.py +++ b/tests/control_plane/test_cold_source_import_cli.py @@ -6,6 +6,8 @@ """ from __future__ import annotations +from contextlib import contextmanager +import gzip import hashlib import io import json @@ -15,6 +17,7 @@ import subprocess import sys import tarfile +from types import SimpleNamespace import pytest @@ -94,6 +97,75 @@ def cli(action, *args, success=True, output_format="json", runtime_override=None return cli, state, backup, body, runtime, receiver, env +def test_prepare_rejects_archive_swap_and_restore_during_member_scan(tmp_path, monkeypatch, capsys): + from loopx.cli import main + from loopx.control_plane.coordination import cold_source_backup + + _, state, backup, _, runtime, _, _ = workspace(tmp_path, monkeypatch) + manifest_path = Path(backup["manifest_path"]) + manifest = json.loads(manifest_path.read_text()) + archive = Path(manifest["execution"]["archive_path"]) + original = archive.read_bytes() + source_a = state.read_bytes() + source_b = source_a.replace(b"Complete cold source requirement", b"Tampered cold source requirement") + assert source_a != source_b and len(source_a) == len(source_b) + expanded = gzip.decompress(original) + assert source_a in expanded + variant = gzip.compress(expanded.replace(source_a, source_b), mtime=0) + size = max(len(original), len(variant)) + original, variant = original.ljust(size, b"\0"), variant.ljust(size, b"\0") + archive.write_bytes(original) + manifest["execution"].update(archive_sha256=hashlib.sha256(original).hexdigest(), archive_size_bytes=size) + manifest_path.write_text(json.dumps(manifest)) + state.write_bytes(source_b) + os.utime(archive, (1_700_000_000, 1_700_000_000)) + before = archive.stat() + original_open, original_fstat = tarfile.open, os.fstat + scans = 0 + + def overwrite(data): + with archive.open("r+b") as handle: + handle.write(data) + os.utime(archive, ns=(before.st_atime_ns, before.st_mtime_ns)) + observed = archive.stat() + assert (observed.st_dev, observed.st_ino, observed.st_size, observed.st_mtime_ns) == ( + before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) + + @contextmanager + def swap_during_scan(*args, **kwargs): + nonlocal scans + scans += 1 + overwrite(variant) + try: + with original_open(*args, **kwargs) as tar: + yield tar + finally: + overwrite(original) + + def descriptor_stat(fd): + observed = original_fstat(fd) + if (observed.st_dev, observed.st_ino) != (before.st_dev, before.st_ino): + return observed + fields = {name: getattr(observed, name) for name in dir(observed) if name.startswith("st_")} + fields["st_ctime_ns"] = before.st_ctime_ns + scans + return SimpleNamespace(**fields) + + monkeypatch.setattr(tarfile, "open", swap_during_scan) + monkeypatch.setattr(cold_source_backup, "os", SimpleNamespace(**{ + **vars(os), "name": "nt", "fstat": descriptor_stat, + })) + code = main(["--registry", str(tmp_path / "project/.loopx/registry.json"), + "--runtime-root", str(runtime), "--format", "json", "coordination-shadow", "prepare-import", + "--goal-id", "cold", "--operation-id", "cold-original", "--backup-manifest", str(manifest_path), + "--provider", "file", "--target-handoff-mode", "hard_lease"]) + result = json.loads(capsys.readouterr().out)["cold_import"] + assert scans == 1 and archive.read_bytes() == original + assert code == 1, result + assert result["reason_code"] == "cold_import_backup_source_missing_or_changed" + assert result.get("coordination_source_backup_verified") is not True + assert not list((runtime / "authority-transition").rglob("writer-fence.json")) + + def test_coordination_command_help_and_rejections_match_full_cli(tmp_path): # Independently compare the existing compatibility parser/handler with the # selected dispatcher, including diagnostics that deliberately fall back. From 72167ad714f6da77b09f74080682153979a20e73 Mon Sep 17 00:00:00 2001 From: xiaobaifly7 <196511160+xiaobaifly7@users.noreply.github.com> Date: Thu, 12 Nov 2026 10:29:32 +0800 Subject: [PATCH 5/5] fix(io): require complete Windows identity for ctime tolerance Retain ctime checks when the volume or file index is unavailable in material and cold-backup observations. Exercise real Windows junction replacement with partial metadata and preserve complete-identity skew. Keep tar link normalization independent of host path syntax so valid first links remain accepted on Windows. Signed-off-by: xiaobaifly7 <196511160+xiaobaifly7@users.noreply.github.com> --- .../coordination/cold_source_backup.py | 9 +-- loopx/materials.py | 8 ++- .../test_decision_context_profile.py | 56 ++++++++++++++----- .../test_cold_source_backup_scope.py | 24 +++++++- 4 files changed, 75 insertions(+), 22 deletions(-) diff --git a/loopx/control_plane/coordination/cold_source_backup.py b/loopx/control_plane/coordination/cold_source_backup.py index 194a90fdc1..4bbd3390e6 100644 --- a/loopx/control_plane/coordination/cold_source_backup.py +++ b/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,9 @@ def _open_regular(path: Path) -> tuple[BinaryIO, os.stat_result]: def _same_file(before: os.stat_result, after: os.stat_result) -> bool: - identity_fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns") - # Windows ctime alone does not reliably identify a file change. - if os.name != "nt": + 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) @@ -118,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) diff --git a/loopx/materials.py b/loopx/materials.py index 4f0af2913d..dc81accc40 100644 --- a/loopx/materials.py +++ b/loopx/materials.py @@ -222,9 +222,11 @@ 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") - # Tolerate Windows ctime skew only when every observation has a file index. - if os.name != "nt" or not all(observed.st_ino for observed in (before, after, current)): + 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( diff --git a/tests/capabilities/test_decision_context_profile.py b/tests/capabilities/test_decision_context_profile.py index d21962a2fd..1964d1211a 100644 --- a/tests/capabilities/test_decision_context_profile.py +++ b/tests/capabilities/test_decision_context_profile.py @@ -532,32 +532,54 @@ def read(self, size=-1): assert read_sizes == [17] -@pytest.mark.skipif(__import__("os").name == "nt", reason="Open-file replacement requires POSIX sharing") -@pytest.mark.parametrize("file_index_available", [False, True]) -def test_local_file_same_metadata_replacement_has_no_verified_read(tmp_path, monkeypatch, file_index_available): +@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 - authority = tmp_path / "authority.txt" - authority.write_bytes(b"original") - os.utime(authority, (1_700_000_000, 1_700_000_000)) - before = authority.stat() - replacement = tmp_path / "replacement.txt" + 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 not file_index_available: - fields["st_ino"] = 0 + 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): @@ -565,19 +587,27 @@ def descriptor_stat(fd): observed = original_fstat(fd) observations += 1 if observations == 2: - replacement.replace(authority) + # 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 == authority else observed + 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"): - materials.read_bounded_material_file(authority, 64) + 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" diff --git a/tests/control_plane/test_cold_source_backup_scope.py b/tests/control_plane/test_cold_source_backup_scope.py index 1007f6c7e6..61ddb94e4c 100644 --- a/tests/control_plane/test_cold_source_backup_scope.py +++ b/tests/control_plane/test_cold_source_backup_scope.py @@ -142,33 +142,51 @@ 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): +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": + 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 @@ -244,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"):