From 3fef47b74700f1fdc2401e39491e0077f3423ec4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 27 Aug 2026 02:15:48 +0900 Subject: [PATCH 1/8] feat(nim): add transactional evidence boundary --- CHANGELOG.md | 2 + contextual_orchestrator/nim_evidence.py | 167 ++++++++++++++++++++ docs/nim-benchmark-evidence-boundary.md | 27 ++++ tests/test_nim_evidence.py | 202 ++++++++++++++++++++++++ 4 files changed, 398 insertions(+) create mode 100644 contextual_orchestrator/nim_evidence.py create mode 100644 docs/nim-benchmark-evidence-boundary.md create mode 100644 tests/test_nim_evidence.py diff --git a/CHANGELOG.md b/CHANGELOG.md index eef15604d..cf8c7f616 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,8 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html) ### Added +- A fail-closed, transactional evidence boundary for the optional NVIDIA NIM + benchmark with immutable task/scorer identities and complete provenance. - Bounded first-valid-completion racing for operator-declared equivalent model group endpoints across text and media capabilities, with fail-closed contract comparison and winner/cancellation provenance. diff --git a/contextual_orchestrator/nim_evidence.py b/contextual_orchestrator/nim_evidence.py new file mode 100644 index 000000000..3f56d4a90 --- /dev/null +++ b/contextual_orchestrator/nim_evidence.py @@ -0,0 +1,167 @@ +"""Validate and atomically publish evidence from optional NIM benchmarks. + +This module deliberately contains no scoring, routing, vector, uncertainty, or +Pareto arithmetic. It is the small trust boundary shared by future benchmark +adapters: immutable task identity in, complete provenance-bearing artifacts out. +""" + +from __future__ import annotations + +import json +import os +import re +import shutil +import tempfile +import uuid +from collections.abc import Mapping +from pathlib import Path + +NIM_EVIDENCE_SCHEMA_VERSION = "1.0.0" +NIM_ARTIFACT_NAMES = frozenset( + { + "benchmark_report.json", + "benchmark_cells.csv", + "benchmark_summary.md", + "run_provenance.json", + } +) +_SHA256 = re.compile(r"[0-9a-f]{64}") +_PROVENANCE_FIELDS = ( + "source_commit", + "catalog_snapshot_sha256", + "task_manifest_sha256", + "pricing_scenario_sha256", + "workflow_run_id", + "evidence_status", +) + + +class NimEvidenceError(ValueError): + """Raised when benchmark evidence is malformed or cannot be published safely.""" + + +def canonical_json_sha256(value: object) -> str: + """Return the SHA-256 of deterministic UTF-8 JSON without doing model arithmetic.""" + import hashlib + + encoded = json.dumps( + value, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ).encode() + return hashlib.sha256(encoded).hexdigest() + + +def validate_task_manifest(manifest: object) -> dict[str, object]: + """Return a validated manifest with unique immutable task/scorer identities.""" + if ( + not isinstance(manifest, dict) + or manifest.get("schema_version") != NIM_EVIDENCE_SCHEMA_VERSION + ): + raise NimEvidenceError( + f"task manifest schema_version must be {NIM_EVIDENCE_SCHEMA_VERSION}" + ) + tasks = manifest.get("tasks") + if not isinstance(tasks, list) or not tasks: + raise NimEvidenceError("task manifest requires a non-empty tasks list") + task_ids: set[str] = set() + for task in tasks: + if not isinstance(task, dict): + raise NimEvidenceError("each task must be an object") + task_id = task.get("task_id") + prompt = task.get("prompt") + scorer = task.get("scorer") + if not isinstance(task_id, str) or not task_id or task_id in task_ids: + raise NimEvidenceError("task_id must be a unique non-empty string") + if not isinstance(prompt, str) or not prompt: + raise NimEvidenceError(f"task {task_id} requires a non-empty prompt") + if not isinstance(scorer, dict): + raise NimEvidenceError(f"task {task_id} requires a scorer object") + identity = (scorer.get("name"), scorer.get("version")) + if not all(isinstance(item, str) and item for item in identity): + raise NimEvidenceError(f"task {task_id} requires scorer name and version") + task_ids.add(task_id) + return manifest + + +def validate_provenance(provenance: object) -> dict[str, str]: + """Return complete, secret-free provenance or fail closed before publication.""" + if not isinstance(provenance, dict) or set(provenance) != set(_PROVENANCE_FIELDS): + raise NimEvidenceError("provenance fields are incomplete or unexpected") + normalized: dict[str, str] = {} + for field in _PROVENANCE_FIELDS: + value = provenance[field] + if not isinstance(value, str) or not value: + raise NimEvidenceError(f"provenance {field} must be a non-empty string") + if ( + field == "source_commit" + and re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", value) is None + ): + raise NimEvidenceError("provenance source_commit must be a Git object ID") + if ( + field.endswith("_sha256") + and not (field == "pricing_scenario_sha256" and value == "unknown") + and _SHA256.fullmatch(value) is None + ): + raise NimEvidenceError(f"provenance {field} must be a lowercase SHA-256") + normalized[field] = value + return normalized + + +def _residue(final: Path, kind: str) -> list[Path]: + """Return private publication residue for one final directory.""" + return sorted(final.parent.glob(f".{final.name}.{kind}-*")) + + +def _recover_publication(final: Path) -> None: + """Remove abandoned staging and restore one unambiguous crash backup.""" + for staging in _residue(final, "staging"): + shutil.rmtree(staging) + backups = _residue(final, "backup") + if len(backups) > 1: + raise NimEvidenceError("multiple publication backups require operator review") + if backups: + if final.exists(): + shutil.rmtree(backups[0]) + else: + os.replace(backups[0], final) + + +def publish_artifact_set( + output_directory: str | os.PathLike[str], artifacts: Mapping[str, bytes] +) -> None: + """Publish exactly one complete artifact set, restoring the prior set on failure.""" + if set(artifacts) != NIM_ARTIFACT_NAMES or any( + not isinstance(value, bytes) or not value for value in artifacts.values() + ): + raise NimEvidenceError( + "artifact set must contain exactly four non-empty byte payloads" + ) + provenance = json.loads(artifacts["run_provenance.json"].decode("utf-8")) + validate_provenance(provenance) + + final = Path(output_directory).expanduser() + if final.name in {"", ".", ".."}: + raise NimEvidenceError("output directory must name a dedicated directory") + final = final.parent.resolve() / final.name + if final.is_symlink() or (final.exists() and not final.is_dir()): + raise NimEvidenceError("output directory must be a real directory") + final.parent.mkdir(parents=True, exist_ok=True) + _recover_publication(final) + staging = Path(tempfile.mkdtemp(dir=final.parent, prefix=f".{final.name}.staging-")) + backup: Path | None = None + try: + for name, payload in artifacts.items(): + (staging / name).write_bytes(payload) + if final.exists(): + backup = final.parent / f".{final.name}.backup-{uuid.uuid4().hex}" + os.replace(final, backup) + try: + os.replace(staging, final) + except BaseException: + if backup is not None and backup.exists(): + os.replace(backup, final) + raise + if backup is not None: + shutil.rmtree(backup) + finally: + if staging.exists(): + shutil.rmtree(staging) diff --git a/docs/nim-benchmark-evidence-boundary.md b/docs/nim-benchmark-evidence-boundary.md new file mode 100644 index 000000000..5f78f2cef --- /dev/null +++ b/docs/nim-benchmark-evidence-boundary.md @@ -0,0 +1,27 @@ +# NVIDIA NIM benchmark evidence boundary + +Issue #86 requires evidence before production routing can change. This first +slice freezes task and scorer identities, requires complete run provenance, and +publishes JSON, CSV, Markdown, and provenance as one replaceable directory. + +It deliberately performs no scoring, uncertainty, vector, Pareto, or routing +arithmetic. Those calculations remain a later Rust-owned slice. A dry-run +artifact proves only schema and publication behavior; it does not prove model +quality, cost, or production readiness. + +The implementation reuses the valid transactional design from closed PR #90, +but its checks, reviews, and runtime evidence are not transferred. It corrects +that branch's duplicate CSV-field defect by keeping this slice independent of +CSV enrichment. + +Operators must supply these exact non-empty artifacts: + +- `benchmark_report.json` +- `benchmark_cells.csv` +- `benchmark_summary.md` +- `run_provenance.json` + +The provenance object accepts only the protected Git source identity, catalog +and task-manifest SHA-256 identities, pricing-scenario SHA-256 or the explicit +value `unknown`, plus workflow-run and evidence status. Unexpected fields fail +closed so secrets cannot be silently serialized. diff --git a/tests/test_nim_evidence.py b/tests/test_nim_evidence.py new file mode 100644 index 000000000..06e6984b4 --- /dev/null +++ b/tests/test_nim_evidence.py @@ -0,0 +1,202 @@ +"""Contract tests for the NIM benchmark evidence boundary.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +from contextual_orchestrator.nim_evidence import ( + NIM_EVIDENCE_SCHEMA_VERSION, + NimEvidenceError, + canonical_json_sha256, + publish_artifact_set, + validate_provenance, + validate_task_manifest, +) + + +def _manifest() -> dict[str, object]: + return { + "schema_version": NIM_EVIDENCE_SCHEMA_VERSION, + "tasks": [ + { + "task_id": "locked_task", + "prompt": "Answer.", + "scorer": {"name": "exact_match", "version": "1"}, + } + ], + } + + +def _provenance() -> dict[str, str]: + digest = "a" * 64 + return { + "source_commit": "a" * 40, + "catalog_snapshot_sha256": digest, + "task_manifest_sha256": digest, + "pricing_scenario_sha256": "unknown", + "workflow_run_id": "offline_fixture", + "evidence_status": "dry_run", + } + + +def _artifacts() -> dict[str, bytes]: + return { + "benchmark_report.json": b"{}", + "benchmark_cells.csv": b"task_id\nlocked_task\n", + "benchmark_summary.md": b"# Evidence\n", + "run_provenance.json": json.dumps(_provenance()).encode(), + } + + +def test_manifest_and_provenance_are_deterministic_and_strict() -> None: + manifest = _manifest() + assert validate_task_manifest(manifest) is manifest + assert canonical_json_sha256(manifest) == canonical_json_sha256( + dict(reversed(list(manifest.items()))) + ) + assert validate_provenance(_provenance())["evidence_status"] == "dry_run" + for invalid in ( + {}, + {**manifest, "schema_version": "future"}, + {**manifest, "tasks": []}, + ): + with pytest.raises(NimEvidenceError): + validate_task_manifest(invalid) + duplicate = _manifest() + duplicate["tasks"] = [duplicate["tasks"][0], duplicate["tasks"][0]] # type: ignore[index] + with pytest.raises(NimEvidenceError): + validate_task_manifest(duplicate) + with pytest.raises(NimEvidenceError): + validate_provenance({**_provenance(), "source_commit": "not-a-hash"}) + with pytest.raises(NimEvidenceError): + validate_provenance({**_provenance(), "pricing_scenario_sha256": "not-a-hash"}) + + +@pytest.mark.parametrize( + "task", + [ + 42, + {"task_id": "", "prompt": "x", "scorer": {"name": "n", "version": "1"}}, + {"task_id": "id", "prompt": "", "scorer": {"name": "n", "version": "1"}}, + {"task_id": "id", "prompt": "x", "scorer": None}, + {"task_id": "id", "prompt": "x", "scorer": {"name": "", "version": "1"}}, + ], +) +def test_manifest_rejects_malformed_task_fields(task: object) -> None: + with pytest.raises(NimEvidenceError): + validate_task_manifest( + {"schema_version": NIM_EVIDENCE_SCHEMA_VERSION, "tasks": [task]} + ) + + +def test_provenance_rejects_shape_and_empty_values() -> None: + with pytest.raises(NimEvidenceError): + validate_provenance([]) + with pytest.raises(NimEvidenceError): + validate_provenance({**_provenance(), "workflow_run_id": ""}) + + +def test_complete_set_replaces_prior_set_and_rejects_partial_set( + tmp_path: Path, +) -> None: + target = tmp_path / "nim_evidence" + target.mkdir() + (target / "old.txt").write_text("old", encoding="utf-8") + publish_artifact_set(target, _artifacts()) + assert {path.name for path in target.iterdir()} == set(_artifacts()) + with pytest.raises(NimEvidenceError): + publish_artifact_set(tmp_path / "partial", {"benchmark_report.json": b"{}"}) + + fresh = tmp_path / "fresh" + publish_artifact_set(fresh, _artifacts()) + assert fresh.is_dir() + + +def test_publication_rejects_invalid_payloads_and_targets(tmp_path: Path) -> None: + empty = _artifacts() + empty["benchmark_report.json"] = b"" + with pytest.raises(NimEvidenceError): + publish_artifact_set(tmp_path / "empty", empty) + malformed = _artifacts() + malformed["run_provenance.json"] = b"not-json" + with pytest.raises(json.JSONDecodeError): + publish_artifact_set(tmp_path / "malformed", malformed) + regular_file = tmp_path / "regular" + regular_file.write_text("x", encoding="utf-8") + with pytest.raises(NimEvidenceError): + publish_artifact_set(regular_file, _artifacts()) + symlink = tmp_path / "linked" + symlink.symlink_to(regular_file) + with pytest.raises(NimEvidenceError): + publish_artifact_set(symlink, _artifacts()) + with pytest.raises(NimEvidenceError): + publish_artifact_set(Path("."), _artifacts()) + + +def test_publication_failure_restores_prior_set( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + target = tmp_path / "nim_evidence" + target.mkdir() + (target / "old.txt").write_text("old", encoding="utf-8") + real_replace = os.replace + + def fail_staging( + source: str | os.PathLike[str], destination: str | os.PathLike[str] + ) -> None: + if Path(source).name.startswith(".nim_evidence.staging-"): + raise OSError("simulated publication failure") + real_replace(source, destination) + + monkeypatch.setattr("contextual_orchestrator.nim_evidence.os.replace", fail_staging) + with pytest.raises(OSError): + publish_artifact_set(target, _artifacts()) + assert (target / "old.txt").read_text(encoding="utf-8") == "old" + + +def test_fresh_publication_failure_leaves_no_target( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + target = tmp_path / "nim_evidence" + + def fail_replace( + source: str | os.PathLike[str], destination: str | os.PathLike[str] + ) -> None: + raise OSError("simulated publication failure") + + monkeypatch.setattr("contextual_orchestrator.nim_evidence.os.replace", fail_replace) + with pytest.raises(OSError): + publish_artifact_set(target, _artifacts()) + assert not target.exists() + + +def test_crash_residue_is_recovered_or_fails_closed(tmp_path: Path) -> None: + target = tmp_path / "nim_evidence" + abandoned = tmp_path / ".nim_evidence.staging-old" + abandoned.mkdir() + backup = tmp_path / ".nim_evidence.backup-old" + backup.mkdir() + (backup / "old.txt").write_text("old", encoding="utf-8") + publish_artifact_set(target, _artifacts()) + assert not abandoned.exists() + assert not backup.exists() + + first = tmp_path / ".ambiguous.backup-one" + second = tmp_path / ".ambiguous.backup-two" + first.mkdir() + second.mkdir() + with pytest.raises(NimEvidenceError): + publish_artifact_set(tmp_path / "ambiguous", _artifacts()) + + +def test_existing_final_discards_stale_backup(tmp_path: Path) -> None: + target = tmp_path / "nim_evidence" + target.mkdir() + backup = tmp_path / ".nim_evidence.backup-old" + backup.mkdir() + publish_artifact_set(target, _artifacts()) + assert not backup.exists() From a23e330c00c11a3599568593dd416b02a96dd5f3 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 27 Aug 2026 03:20:07 +0900 Subject: [PATCH 2/8] fix(nim): finalize readable evidence publication --- contextual_orchestrator/nim_evidence.py | 7 ++++++- tests/test_nim_evidence.py | 20 ++++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/contextual_orchestrator/nim_evidence.py b/contextual_orchestrator/nim_evidence.py index 3f56d4a90..295476eb7 100644 --- a/contextual_orchestrator/nim_evidence.py +++ b/contextual_orchestrator/nim_evidence.py @@ -11,6 +11,7 @@ import os import re import shutil +import stat import tempfile import uuid from collections.abc import Mapping @@ -147,6 +148,7 @@ def publish_artifact_set( final.parent.mkdir(parents=True, exist_ok=True) _recover_publication(final) staging = Path(tempfile.mkdtemp(dir=final.parent, prefix=f".{final.name}.staging-")) + staging.chmod(stat.S_IMODE(final.stat().st_mode) if final.exists() else 0o755) backup: Path | None = None try: for name, payload in artifacts.items(): @@ -161,7 +163,10 @@ def publish_artifact_set( os.replace(backup, final) raise if backup is not None: - shutil.rmtree(backup) + try: + shutil.rmtree(backup) + except OSError: + pass finally: if staging.exists(): shutil.rmtree(staging) diff --git a/tests/test_nim_evidence.py b/tests/test_nim_evidence.py index 06e6984b4..67e5f850c 100644 --- a/tests/test_nim_evidence.py +++ b/tests/test_nim_evidence.py @@ -4,6 +4,7 @@ import json import os +import shutil from pathlib import Path import pytest @@ -114,6 +115,25 @@ def test_complete_set_replaces_prior_set_and_rejects_partial_set( fresh = tmp_path / "fresh" publish_artifact_set(fresh, _artifacts()) assert fresh.is_dir() + assert fresh.stat().st_mode & 0o777 == 0o755 + + +def test_successful_replacement_preserves_mode_and_ignores_backup_cleanup_failure( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + target = tmp_path / "nim_evidence" + target.mkdir(mode=0o750) + real_rmtree = shutil.rmtree + + def fail_backup(path: str | os.PathLike[str]) -> None: + if Path(path).name.startswith(".nim_evidence.backup-"): + raise OSError("simulated cleanup failure") + real_rmtree(path) + + monkeypatch.setattr("contextual_orchestrator.nim_evidence.shutil.rmtree", fail_backup) + publish_artifact_set(target, _artifacts()) + assert target.stat().st_mode & 0o777 == 0o750 + assert (target / "benchmark_report.json").read_bytes() == b"{}" def test_publication_rejects_invalid_payloads_and_targets(tmp_path: Path) -> None: From 7c1d5c24965954263e5bbe20bf2fda81fc914a7c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 27 Aug 2026 03:48:06 +0900 Subject: [PATCH 3/8] fix(nim): harden evidence publication inputs --- contextual_orchestrator/nim_evidence.py | 15 ++++++++++----- tests/test_nim_evidence.py | 24 +++++++++++++++++++++--- 2 files changed, 31 insertions(+), 8 deletions(-) diff --git a/contextual_orchestrator/nim_evidence.py b/contextual_orchestrator/nim_evidence.py index 295476eb7..82a68a416 100644 --- a/contextual_orchestrator/nim_evidence.py +++ b/contextual_orchestrator/nim_evidence.py @@ -12,7 +12,6 @@ import re import shutil import stat -import tempfile import uuid from collections.abc import Mapping from pathlib import Path @@ -109,7 +108,8 @@ def validate_provenance(provenance: object) -> dict[str, str]: def _residue(final: Path, kind: str) -> list[Path]: """Return private publication residue for one final directory.""" - return sorted(final.parent.glob(f".{final.name}.{kind}-*")) + prefix = f".{final.name}.{kind}-" + return sorted(path for path in final.parent.iterdir() if path.name.startswith(prefix)) def _recover_publication(final: Path) -> None: @@ -136,7 +136,10 @@ def publish_artifact_set( raise NimEvidenceError( "artifact set must contain exactly four non-empty byte payloads" ) - provenance = json.loads(artifacts["run_provenance.json"].decode("utf-8")) + try: + provenance = json.loads(artifacts["run_provenance.json"].decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise NimEvidenceError("run_provenance.json must contain valid UTF-8 JSON") from exc validate_provenance(provenance) final = Path(output_directory).expanduser() @@ -147,8 +150,10 @@ def publish_artifact_set( raise NimEvidenceError("output directory must be a real directory") final.parent.mkdir(parents=True, exist_ok=True) _recover_publication(final) - staging = Path(tempfile.mkdtemp(dir=final.parent, prefix=f".{final.name}.staging-")) - staging.chmod(stat.S_IMODE(final.stat().st_mode) if final.exists() else 0o755) + staging = final.parent / f".{final.name}.staging-{uuid.uuid4().hex}" + staging.mkdir(mode=0o777) + if final.exists(): + staging.chmod(stat.S_IMODE(final.stat().st_mode)) backup: Path | None = None try: for name, payload in artifacts.items(): diff --git a/tests/test_nim_evidence.py b/tests/test_nim_evidence.py index 67e5f850c..fed43fe2e 100644 --- a/tests/test_nim_evidence.py +++ b/tests/test_nim_evidence.py @@ -113,9 +113,13 @@ def test_complete_set_replaces_prior_set_and_rejects_partial_set( publish_artifact_set(tmp_path / "partial", {"benchmark_report.json": b"{}"}) fresh = tmp_path / "fresh" - publish_artifact_set(fresh, _artifacts()) + previous_umask = os.umask(0o027) + try: + publish_artifact_set(fresh, _artifacts()) + finally: + os.umask(previous_umask) assert fresh.is_dir() - assert fresh.stat().st_mode & 0o777 == 0o755 + assert fresh.stat().st_mode & 0o777 == 0o750 def test_successful_replacement_preserves_mode_and_ignores_backup_cleanup_failure( @@ -143,7 +147,7 @@ def test_publication_rejects_invalid_payloads_and_targets(tmp_path: Path) -> Non publish_artifact_set(tmp_path / "empty", empty) malformed = _artifacts() malformed["run_provenance.json"] = b"not-json" - with pytest.raises(json.JSONDecodeError): + with pytest.raises(NimEvidenceError, match="valid UTF-8 JSON"): publish_artifact_set(tmp_path / "malformed", malformed) regular_file = tmp_path / "regular" regular_file.write_text("x", encoding="utf-8") @@ -213,6 +217,20 @@ def test_crash_residue_is_recovered_or_fails_closed(tmp_path: Path) -> None: publish_artifact_set(tmp_path / "ambiguous", _artifacts()) +def test_crash_recovery_treats_glob_metacharacters_as_literal(tmp_path: Path) -> None: + target = tmp_path / "nim[evidence]" + backup = tmp_path / ".nim[evidence].backup-old" + backup.mkdir() + unrelated = tmp_path / ".nime.backup-unrelated" + unrelated.mkdir() + + publish_artifact_set(target, _artifacts()) + + assert target.is_dir() + assert not backup.exists() + assert unrelated.is_dir() + + def test_existing_final_discards_stale_backup(tmp_path: Path) -> None: target = tmp_path / "nim_evidence" target.mkdir() From 7a7054a5414a04475cef3695354673494e42bd11 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 27 Aug 2026 04:07:49 +0900 Subject: [PATCH 4/8] fix(nim): preserve read-only publication mode safely --- contextual_orchestrator/nim_evidence.py | 5 +++-- tests/test_nim_evidence.py | 10 ++++++++++ 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/contextual_orchestrator/nim_evidence.py b/contextual_orchestrator/nim_evidence.py index 82a68a416..d56c0c2bc 100644 --- a/contextual_orchestrator/nim_evidence.py +++ b/contextual_orchestrator/nim_evidence.py @@ -152,12 +152,13 @@ def publish_artifact_set( _recover_publication(final) staging = final.parent / f".{final.name}.staging-{uuid.uuid4().hex}" staging.mkdir(mode=0o777) - if final.exists(): - staging.chmod(stat.S_IMODE(final.stat().st_mode)) + final_mode = stat.S_IMODE(final.stat().st_mode) if final.exists() else None backup: Path | None = None try: for name, payload in artifacts.items(): (staging / name).write_bytes(payload) + if final_mode is not None: + staging.chmod(final_mode) if final.exists(): backup = final.parent / f".{final.name}.backup-{uuid.uuid4().hex}" os.replace(final, backup) diff --git a/tests/test_nim_evidence.py b/tests/test_nim_evidence.py index fed43fe2e..ba042b6a4 100644 --- a/tests/test_nim_evidence.py +++ b/tests/test_nim_evidence.py @@ -140,6 +140,16 @@ def fail_backup(path: str | os.PathLike[str]) -> None: assert (target / "benchmark_report.json").read_bytes() == b"{}" +def test_replacement_preserves_read_only_directory_mode(tmp_path: Path) -> None: + target = tmp_path / "nim_evidence" + target.mkdir(mode=0o500) + + publish_artifact_set(target, _artifacts()) + + assert target.stat().st_mode & 0o777 == 0o500 + assert (target / "benchmark_report.json").read_bytes() == b"{}" + + def test_publication_rejects_invalid_payloads_and_targets(tmp_path: Path) -> None: empty = _artifacts() empty["benchmark_report.json"] = b"" From 584dddfc632685e60c31fee5c2b393ed2d3f22f5 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 27 Aug 2026 04:49:20 +0900 Subject: [PATCH 5/8] fix(nim): tolerate stale backup cleanup failure --- contextual_orchestrator/nim_evidence.py | 5 ++++- tests/test_nim_evidence.py | 19 +++++++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/contextual_orchestrator/nim_evidence.py b/contextual_orchestrator/nim_evidence.py index d56c0c2bc..f12f99a0e 100644 --- a/contextual_orchestrator/nim_evidence.py +++ b/contextual_orchestrator/nim_evidence.py @@ -121,7 +121,10 @@ def _recover_publication(final: Path) -> None: raise NimEvidenceError("multiple publication backups require operator review") if backups: if final.exists(): - shutil.rmtree(backups[0]) + try: + shutil.rmtree(backups[0]) + except OSError: + pass else: os.replace(backups[0], final) diff --git a/tests/test_nim_evidence.py b/tests/test_nim_evidence.py index ba042b6a4..2ad27382f 100644 --- a/tests/test_nim_evidence.py +++ b/tests/test_nim_evidence.py @@ -140,6 +140,25 @@ def fail_backup(path: str | os.PathLike[str]) -> None: assert (target / "benchmark_report.json").read_bytes() == b"{}" +def test_recovery_ignores_stale_backup_cleanup_failure( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + target = tmp_path / "nim_evidence" + target.mkdir() + stale_backup = tmp_path / ".nim_evidence.backup-old" + stale_backup.mkdir() + real_rmtree = shutil.rmtree + + def fail_stale_backup(path: str | os.PathLike[str]) -> None: + if Path(path).resolve() == stale_backup.resolve(): + raise OSError("simulated cleanup failure") + real_rmtree(path) + + monkeypatch.setattr("contextual_orchestrator.nim_evidence.shutil.rmtree", fail_stale_backup) + publish_artifact_set(target, _artifacts()) + assert (target / "benchmark_report.json").read_bytes() == b"{}" + + def test_replacement_preserves_read_only_directory_mode(tmp_path: Path) -> None: target = tmp_path / "nim_evidence" target.mkdir(mode=0o500) From fe7aecfc76ebf6f5c2bdcda8d578aa505e4a8ef6 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 27 Aug 2026 05:38:04 +0900 Subject: [PATCH 6/8] fix(nim): serialize evidence publication --- contextual_orchestrator/nim_evidence.py | 102 +++++++++++++++++------- docs/nim-benchmark-evidence-boundary.md | 6 ++ tests/test_nim_evidence.py | 87 ++++++++++++++++++++ 3 files changed, 168 insertions(+), 27 deletions(-) diff --git a/contextual_orchestrator/nim_evidence.py b/contextual_orchestrator/nim_evidence.py index f12f99a0e..3d05a401f 100644 --- a/contextual_orchestrator/nim_evidence.py +++ b/contextual_orchestrator/nim_evidence.py @@ -13,9 +13,16 @@ import shutil import stat import uuid -from collections.abc import Mapping +from collections.abc import Iterator, Mapping +from contextlib import contextmanager from pathlib import Path +try: # pragma: no cover - selected by the host platform + import fcntl +except ImportError: # pragma: no cover - Windows compatibility + fcntl = None # type: ignore[assignment] + import msvcrt + NIM_EVIDENCE_SCHEMA_VERSION = "1.0.0" NIM_ARTIFACT_NAMES = frozenset( { @@ -112,10 +119,50 @@ def _residue(final: Path, kind: str) -> list[Path]: return sorted(path for path in final.parent.iterdir() if path.name.startswith(prefix)) +@contextmanager +def _publication_lock(final: Path) -> Iterator[None]: + """Serialize publishers that target the same evidence directory.""" + lock_path = final.parent / f".{final.name}.publish-lock" + if lock_path.is_symlink(): + raise NimEvidenceError("publication lock must not be a symbolic link") + flags = os.O_CREAT | os.O_RDWR + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + descriptor = os.open(lock_path, flags, 0o600) + except OSError as exc: + raise NimEvidenceError("publication lock could not be opened safely") from exc + try: + if fcntl is not None: + fcntl.flock(descriptor, fcntl.LOCK_EX) + else: # pragma: no cover - Windows compatibility + if os.fstat(descriptor).st_size == 0: + os.write(descriptor, b"\0") + os.lseek(descriptor, 0, os.SEEK_SET) + msvcrt.locking(descriptor, msvcrt.LK_LOCK, 1) + yield + finally: + if fcntl is not None: + fcntl.flock(descriptor, fcntl.LOCK_UN) + else: # pragma: no cover - Windows compatibility + os.lseek(descriptor, 0, os.SEEK_SET) + msvcrt.locking(descriptor, msvcrt.LK_UNLCK, 1) + os.close(descriptor) + + +def _remove_staging(path: Path) -> None: + """Remove private staging even when a crash preserved a read-only mode.""" + try: + path.chmod(stat.S_IMODE(path.stat().st_mode) | stat.S_IRWXU) + shutil.rmtree(path) + except OSError as exc: + raise NimEvidenceError("abandoned publication staging requires operator review") from exc + + def _recover_publication(final: Path) -> None: """Remove abandoned staging and restore one unambiguous crash backup.""" for staging in _residue(final, "staging"): - shutil.rmtree(staging) + _remove_staging(staging) backups = _residue(final, "backup") if len(backups) > 1: raise NimEvidenceError("multiple publication backups require operator review") @@ -152,30 +199,31 @@ def publish_artifact_set( if final.is_symlink() or (final.exists() and not final.is_dir()): raise NimEvidenceError("output directory must be a real directory") final.parent.mkdir(parents=True, exist_ok=True) - _recover_publication(final) - staging = final.parent / f".{final.name}.staging-{uuid.uuid4().hex}" - staging.mkdir(mode=0o777) - final_mode = stat.S_IMODE(final.stat().st_mode) if final.exists() else None - backup: Path | None = None - try: - for name, payload in artifacts.items(): - (staging / name).write_bytes(payload) - if final_mode is not None: - staging.chmod(final_mode) - if final.exists(): - backup = final.parent / f".{final.name}.backup-{uuid.uuid4().hex}" - os.replace(final, backup) + with _publication_lock(final): + _recover_publication(final) + staging = final.parent / f".{final.name}.staging-{uuid.uuid4().hex}" + staging.mkdir(mode=0o777) + final_mode = stat.S_IMODE(final.stat().st_mode) if final.exists() else None + backup: Path | None = None try: - os.replace(staging, final) - except BaseException: - if backup is not None and backup.exists(): - os.replace(backup, final) - raise - if backup is not None: + for name, payload in artifacts.items(): + (staging / name).write_bytes(payload) + if final_mode is not None: + staging.chmod(final_mode) + if final.exists(): + backup = final.parent / f".{final.name}.backup-{uuid.uuid4().hex}" + os.replace(final, backup) try: - shutil.rmtree(backup) - except OSError: - pass - finally: - if staging.exists(): - shutil.rmtree(staging) + os.replace(staging, final) + except BaseException: + if backup is not None and backup.exists(): + os.replace(backup, final) + raise + if backup is not None: + try: + shutil.rmtree(backup) + except OSError: + pass + finally: + if staging.exists(): + _remove_staging(staging) diff --git a/docs/nim-benchmark-evidence-boundary.md b/docs/nim-benchmark-evidence-boundary.md index 5f78f2cef..751bdc949 100644 --- a/docs/nim-benchmark-evidence-boundary.md +++ b/docs/nim-benchmark-evidence-boundary.md @@ -25,3 +25,9 @@ The provenance object accepts only the protected Git source identity, catalog and task-manifest SHA-256 identities, pricing-scenario SHA-256 or the explicit value `unknown`, plus workflow-run and evidence status. Unexpected fields fail closed so secrets cannot be silently serialized. + +Publication takes an advisory sibling lock per output directory. Concurrent +writers therefore commit complete artifact sets in order rather than deleting +one another's staging data. Crash recovery restores one unambiguous backup and +removes read-only staging residue; ambiguous backups still require operator +review. diff --git a/tests/test_nim_evidence.py b/tests/test_nim_evidence.py index 2ad27382f..9d78fcb0e 100644 --- a/tests/test_nim_evidence.py +++ b/tests/test_nim_evidence.py @@ -5,6 +5,9 @@ import json import os import shutil +import threading +import time +from concurrent.futures import ThreadPoolExecutor from pathlib import Path import pytest @@ -246,6 +249,90 @@ def test_crash_residue_is_recovered_or_fails_closed(tmp_path: Path) -> None: publish_artifact_set(tmp_path / "ambiguous", _artifacts()) +def test_read_only_crash_staging_is_recovered(tmp_path: Path) -> None: + target = tmp_path / "nim_evidence" + abandoned = tmp_path / ".nim_evidence.staging-old" + abandoned.mkdir(mode=0o500) + + publish_artifact_set(target, _artifacts()) + + assert target.is_dir() + assert not abandoned.exists() + + +def test_concurrent_publications_are_serialized( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + target = tmp_path / "nim_evidence" + first = _artifacts() + first["benchmark_report.json"] = b'{"run":"first"}' + second = _artifacts() + second["benchmark_report.json"] = b'{"run":"second"}' + first_write_started = threading.Event() + release_first = threading.Event() + real_write_bytes = Path.write_bytes + + def pause_first_report(path: Path, payload: bytes) -> int: + if payload == first["benchmark_report.json"]: + first_write_started.set() + assert release_first.wait(timeout=5) + return real_write_bytes(path, payload) + + monkeypatch.setattr(Path, "write_bytes", pause_first_report) + with ThreadPoolExecutor(max_workers=2) as executor: + first_future = executor.submit(publish_artifact_set, target, first) + assert first_write_started.wait(timeout=5) + second_future = executor.submit(publish_artifact_set, target, second) + time.sleep(0.05) + assert not second_future.done() + release_first.set() + first_future.result(timeout=5) + second_future.result(timeout=5) + + assert (target / "benchmark_report.json").read_bytes() == second["benchmark_report.json"] + assert {path.name for path in target.iterdir()} == set(second) + + +def test_publication_rejects_unsafe_or_unopenable_lock( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + target = tmp_path / "nim_evidence" + lock = tmp_path / ".nim_evidence.publish-lock" + lock.symlink_to(tmp_path / "elsewhere") + with pytest.raises(NimEvidenceError, match="symbolic link"): + publish_artifact_set(target, _artifacts()) + + lock.unlink() + real_open = os.open + + def fail_lock_open(path: str | os.PathLike[str], *args: object) -> int: + if Path(path) == lock: + raise OSError("simulated lock failure") + return real_open(path, *args) # type: ignore[arg-type] + + monkeypatch.setattr("contextual_orchestrator.nim_evidence.os.open", fail_lock_open) + with pytest.raises(NimEvidenceError, match="opened safely"): + publish_artifact_set(target, _artifacts()) + + +def test_unremovable_crash_staging_fails_closed( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + target = tmp_path / "nim_evidence" + abandoned = tmp_path / ".nim_evidence.staging-old" + abandoned.mkdir() + real_rmtree = shutil.rmtree + + def fail_abandoned(path: str | os.PathLike[str]) -> None: + if Path(path) == abandoned: + raise OSError("simulated staging cleanup failure") + real_rmtree(path) + + monkeypatch.setattr("contextual_orchestrator.nim_evidence.shutil.rmtree", fail_abandoned) + with pytest.raises(NimEvidenceError, match="operator review"): + publish_artifact_set(target, _artifacts()) + + def test_crash_recovery_treats_glob_metacharacters_as_literal(tmp_path: Path) -> None: target = tmp_path / "nim[evidence]" backup = tmp_path / ".nim[evidence].backup-old" From 05c0c99f03ab79909df45746901910749efd2d7e Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 27 Aug 2026 07:04:58 +0900 Subject: [PATCH 7/8] fix(nim): remove read-only publication backups --- contextual_orchestrator/nim_evidence.py | 8 ++++---- tests/test_nim_evidence.py | 13 +++++++++++++ 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/contextual_orchestrator/nim_evidence.py b/contextual_orchestrator/nim_evidence.py index 3d05a401f..92962f91e 100644 --- a/contextual_orchestrator/nim_evidence.py +++ b/contextual_orchestrator/nim_evidence.py @@ -169,8 +169,8 @@ def _recover_publication(final: Path) -> None: if backups: if final.exists(): try: - shutil.rmtree(backups[0]) - except OSError: + _remove_staging(backups[0]) + except NimEvidenceError: pass else: os.replace(backups[0], final) @@ -221,8 +221,8 @@ def publish_artifact_set( raise if backup is not None: try: - shutil.rmtree(backup) - except OSError: + _remove_staging(backup) + except NimEvidenceError: pass finally: if staging.exists(): diff --git a/tests/test_nim_evidence.py b/tests/test_nim_evidence.py index 9d78fcb0e..29a5f9488 100644 --- a/tests/test_nim_evidence.py +++ b/tests/test_nim_evidence.py @@ -260,6 +260,19 @@ def test_read_only_crash_staging_is_recovered(tmp_path: Path) -> None: assert not abandoned.exists() +def test_read_only_final_republishes_without_backup_residue(tmp_path: Path) -> None: + target = tmp_path / "nim_evidence" + target.mkdir() + (target / "old.txt").write_text("old", encoding="utf-8") + target.chmod(0o500) + + publish_artifact_set(target, _artifacts()) + publish_artifact_set(target, _artifacts()) + + assert target.stat().st_mode & 0o777 == 0o500 + assert not list(tmp_path.glob(".nim_evidence.backup-*")) + + def test_concurrent_publications_are_serialized( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: From 86124f4fa3e1d65589ef32b58d813f56eaffa3be Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 27 Aug 2026 09:18:13 +0900 Subject: [PATCH 8/8] fix(evidence): enforce secure default mode for new publication directories - Compare staging cleanup locks by resolving parent paths --- contextual_orchestrator/nim_evidence.py | 2 +- tests/test_nim_evidence.py | 8 +++++--- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/contextual_orchestrator/nim_evidence.py b/contextual_orchestrator/nim_evidence.py index 92962f91e..0f7e32eb2 100644 --- a/contextual_orchestrator/nim_evidence.py +++ b/contextual_orchestrator/nim_evidence.py @@ -202,7 +202,7 @@ def publish_artifact_set( with _publication_lock(final): _recover_publication(final) staging = final.parent / f".{final.name}.staging-{uuid.uuid4().hex}" - staging.mkdir(mode=0o777) + staging.mkdir(mode=0o700) final_mode = stat.S_IMODE(final.stat().st_mode) if final.exists() else None backup: Path | None = None try: diff --git a/tests/test_nim_evidence.py b/tests/test_nim_evidence.py index 29a5f9488..8940f2bdd 100644 --- a/tests/test_nim_evidence.py +++ b/tests/test_nim_evidence.py @@ -122,7 +122,7 @@ def test_complete_set_replaces_prior_set_and_rejects_partial_set( finally: os.umask(previous_umask) assert fresh.is_dir() - assert fresh.stat().st_mode & 0o777 == 0o750 + assert fresh.stat().st_mode & 0o777 == 0o700 def test_successful_replacement_preserves_mode_and_ignores_backup_cleanup_failure( @@ -319,7 +319,8 @@ def test_publication_rejects_unsafe_or_unopenable_lock( real_open = os.open def fail_lock_open(path: str | os.PathLike[str], *args: object) -> int: - if Path(path) == lock: + p = Path(path) + if p.parent.resolve() / p.name == lock.parent.resolve() / lock.name: raise OSError("simulated lock failure") return real_open(path, *args) # type: ignore[arg-type] @@ -337,7 +338,8 @@ def test_unremovable_crash_staging_fails_closed( real_rmtree = shutil.rmtree def fail_abandoned(path: str | os.PathLike[str]) -> None: - if Path(path) == abandoned: + p = Path(path) + if p.parent.resolve() / p.name == abandoned.parent.resolve() / abandoned.name: raise OSError("simulated staging cleanup failure") real_rmtree(path)