refactor(evidence): share inventory producer primitives

src/raes_adapters/_inventory.py

@@ -0,0 +1,61 @@
+"""Private byte-level inventory primitives shared by adapter producers.
+
+This module describes files, not experiment meaning. It deliberately imports
+neither simulators nor RAES contracts and defines no portable schema.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from collections.abc import Callable, Iterable
+from pathlib import Path
+
+
+def media_type(path: Path) -> str:
+    """Return the established portable media type for a produced artifact."""
+
+    return "application/json" if path.suffix == ".json" else "application/octet-stream"
+
+
+def inventory_entry(
+    root: Path,
+    path: Path,
+    *,
+    type_name: str | None = None,
+) -> dict[str, object]:
+    """Describe one regular contained file with the established entry shape."""
+
+    resolved_root = root.resolve()
+    if path.is_symlink() or not path.is_file():
+        raise ValueError("inventory member is not a regular file")
+    resolved_path = path.resolve()
+    if not resolved_path.is_relative_to(resolved_root):
+        raise ValueError("inventory member escapes its root")
+    try:
+        relative = path.relative_to(root).as_posix()
+    except ValueError as error:
+        raise ValueError("inventory member escapes its root") from error
+    content = path.read_bytes()
+    return {
+        "media_type": type_name if type_name is not None else media_type(path),
+        "path": relative,
+        "sha256": hashlib.sha256(content).hexdigest(),
+        "size_bytes": len(content),
+    }
+
+
+def inventory_document(
+    root: Path,
+    members: Iterable[Path],
+    *,
+    type_for: Callable[[Path], str] = media_type,
+) -> dict[str, object]:
+    """Build a deterministic inventory document without writing it."""
+
+    ordered = sorted(members)
+    return {
+        "artifacts": [inventory_entry(root, path, type_name=type_for(path)) for path in ordered]
+    }
+
+
+__all__ = ["inventory_document", "inventory_entry", "media_type"]

src/raes_adapters/cli.py

@@ -60,6 +60,7 @@
     atomic_write_json_artifact,
 )
 
+from raes_adapters._inventory import inventory_document, media_type
 from raes_adapters.cyberbattlesim import (
     load_qualification as load_cyberbattlesim_qualification,
 )
@@ -1332,26 +1333,16 @@ def _reserve_output(requested: Path) -> Path:
 def _media_type(path: Path) -> str:
     """Return the portable media type used in the artifact inventory."""
 
-    return "application/json" if path.suffix == ".json" else "application/octet-stream"
+    return media_type(path)
 
 
 def _seal_inventory(output: Path) -> dict[str, object]:
     """Hash every portable artifact and atomically seal the inventory."""
 
-    artifacts: list[dict[str, object]] = []
-    for path in sorted(item for item in output.rglob("*") if item.is_file()):
-        if path.name == _INVENTORY_NAME:
-            continue
-        payload = path.read_bytes()
-        artifacts.append(
-            {
-                "media_type": _media_type(path),
-                "path": path.relative_to(output).as_posix(),
-                "sha256": hashlib.sha256(payload).hexdigest(),
-                "size_bytes": len(payload),
-            }
-        )
-    inventory: dict[str, object] = {"artifacts": artifacts}
+    members = [
+        path for path in output.rglob("*") if path.is_file() and path.name != _INVENTORY_NAME
+    ]
+    inventory = inventory_document(output, members, type_for=_media_type)
     atomic_write_json_artifact(output / _INVENTORY_NAME, inventory)
     return inventory
 

src/raes_adapters/cyborg/reproduction.py

@@ -42,6 +42,9 @@
     atomic_write_json_artifact,
 )
 
+from raes_adapters._inventory import inventory_entry
+from raes_adapters.bundle_verifier import verify_bundle as verify_integrity_bundle
+
 from .driver import CyborgDriver, SourceInstalledCyborgDriver
 from .researcher import (
     EpisodeEvidence,
@@ -1393,17 +1396,14 @@ def _media_type(path: Path) -> str:
 def _inventory_entry(root: Path, path: Path) -> dict[str, object]:
     """Build one bounded relative inventory entry."""
 
-    if path.is_symlink() or not path.is_file() or not path.resolve().is_relative_to(root.resolve()):
-        raise ValueError("inventory member is invalid")
-    content = path.read_bytes()
-    if len(content) >= _MAX_PUBLIC_FILE_BYTES:
+    try:
+        entry = inventory_entry(root, path, type_name=_media_type(path))
+    except ValueError as error:
+        raise ValueError("inventory member is invalid") from error
+    size = entry["size_bytes"]
+    if not isinstance(size, int) or size >= _MAX_PUBLIC_FILE_BYTES:
         raise ValueError("public artifact exceeds the file-size limit")
-    return {
-        "path": path.relative_to(root).as_posix(),
-        "media_type": _media_type(path),
-        "sha256": hashlib.sha256(content).hexdigest(),
-        "size_bytes": len(content),
-    }
+    return entry
 
 
 def _seal_inventory(root: Path, members: Sequence[Path]) -> dict[str, object]:
@@ -2835,6 +2835,7 @@ def verify_bundle(
 ) -> dict[str, object]:
     """Offline-verify every transitive inventory and recompute the frozen result."""
 
+    verify_integrity_bundle(root)
     bundle = root.resolve()
     expected_root_directories = {"plans", "runs"}
     if (

tests/test_cyborg_reproduction.py

@@ -8,6 +8,7 @@
 import pytest
 
 from raes_adapters import cli
+from raes_adapters.bundle_verifier import BundleInvalid
 from raes_adapters.cyborg import reproduction
 from raes_adapters.cyborg.driver import (
     _NativeEvaluationContext,
@@ -20,6 +21,36 @@
 PROJECT_ROOT = Path(__file__).parents[1]
 
 
+def test_generic_integrity_precedes_backend_semantic_verification(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    bundle = tmp_path / "bundle"
+    bundle.mkdir()
+    (bundle / "inventory.json").write_text('{"artifacts":[]}\n', encoding="utf-8")
+    (bundle / "protocol.json").write_text("{}\n", encoding="utf-8")
+
+    monkeypatch.setattr(
+        reproduction,
+        "load_strict_json",
+        lambda _path: pytest.fail("backend semantic checks ran before generic integrity"),
+    )
+
+    with pytest.raises(BundleInvalid, match="bundle.inventory.membership-mismatch"):
+        reproduction.verify_bundle(bundle)
+
+
+def test_generic_integrity_rejects_cyborg_bundle_root_symlink(tmp_path: Path) -> None:
+    bundle = tmp_path / "bundle"
+    bundle.mkdir()
+    (bundle / "inventory.json").write_text('{"artifacts":[]}\n', encoding="utf-8")
+    link = tmp_path / "bundle-link"
+    link.symlink_to(bundle, target_is_directory=True)
+
+    with pytest.raises(BundleInvalid, match="bundle.root.invalid"):
+        reproduction.verify_bundle(link)
+
+
 class StudyDriver:
     """Bounded native seam for the real scheduler and persistence path."""
 
@@ -560,5 +591,5 @@ def test_offline_verifier_requires_complete_transitive_inventories(tmp_path: Pat
     root_members = [bundle / item["path"] for item in root_inventory["artifacts"]]
     reproduction._seal_inventory(bundle, root_members)
 
-    with pytest.raises(ValueError, match="inventory membership is incomplete"):
+    with pytest.raises(BundleInvalid, match="bundle.inventory.membership-mismatch"):
         reproduction.verify_bundle(bundle, selection=selection)

tests/test_inventory_primitives.py

@@ -0,0 +1,138 @@
+"""Parity checks for shared private inventory producer plumbing."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+
+import pytest
+
+from raes_adapters import cli
+from raes_adapters._inventory import inventory_document, inventory_entry
+from raes_adapters.cyborg import reproduction
+
+
+def _entry(path: str, content: bytes, media_type: str) -> dict[str, object]:
+    return {
+        "media_type": media_type,
+        "path": path,
+        "sha256": hashlib.sha256(content).hexdigest(),
+        "size_bytes": len(content),
+    }
+
+
+def _canonical_bytes(payload: object) -> bytes:
+    return (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode()
+
+
+def test_generic_producer_wrapper_preserves_exact_inventory_bytes(tmp_path: Path) -> None:
+    nested = tmp_path / "nested"
+    nested.mkdir()
+    json_content = b'{"value":1}\n'
+    binary_content = b"\x00portable"
+    (nested / "record.json").write_bytes(json_content)
+    (tmp_path / "artifact.bin").write_bytes(binary_content)
+    expected = {
+        "artifacts": [
+            _entry("artifact.bin", binary_content, "application/octet-stream"),
+            _entry("nested/record.json", json_content, "application/json"),
+        ]
+    }
+
+    assert cli._seal_inventory(tmp_path) == expected
+    assert (tmp_path / "inventory.json").read_bytes() == _canonical_bytes(expected)
+
+
+def test_generic_producer_wrapper_preserves_legacy_path_order(tmp_path: Path) -> None:
+    nested = tmp_path / "a"
+    nested.mkdir()
+    nested_content = b"nested\n"
+    sibling_content = b"sibling\n"
+    (nested / "b.json").write_bytes(nested_content)
+    (tmp_path / "a-b.json").write_bytes(sibling_content)
+    expected = {
+        "artifacts": [
+            _entry("a/b.json", nested_content, "application/json"),
+            _entry("a-b.json", sibling_content, "application/json"),
+        ]
+    }
+
+    assert cli._seal_inventory(tmp_path) == expected
+    assert (tmp_path / "inventory.json").read_bytes() == _canonical_bytes(expected)
+
+
+def test_cyborg_compatibility_wrappers_preserve_media_types_and_bytes(tmp_path: Path) -> None:
+    markdown_content = b"# Result\n"
+    gzip_content = b"compressed"
+    markdown = tmp_path / "report.md"
+    compressed = tmp_path / "attempt.json.gz"
+    markdown.write_bytes(markdown_content)
+    compressed.write_bytes(gzip_content)
+    expected = {
+        "artifacts": [
+            _entry("attempt.json.gz", gzip_content, "application/gzip"),
+            _entry("report.md", markdown_content, "text/markdown"),
+        ]
+    }
+
+    assert reproduction._inventory_entry(tmp_path, compressed) == expected["artifacts"][0]
+    assert reproduction._seal_inventory(tmp_path, [markdown, compressed]) == expected
+    assert (tmp_path / "inventory.json").read_bytes() == _canonical_bytes(expected)
+
+
+def test_private_primitives_have_no_simulator_or_contract_authority(tmp_path: Path) -> None:
+    artifact = tmp_path / "record.json"
+    artifact.write_bytes(b"{}\n")
+
+    assert inventory_entry(tmp_path, artifact)["path"] == "record.json"
+    assert inventory_document(tmp_path, [artifact]) == {
+        "artifacts": [inventory_entry(tmp_path, artifact)]
+    }
+
+    source = (Path(__file__).parents[1] / "src/raes_adapters/_inventory.py").read_text(
+        encoding="utf-8"
+    )
+    assert "raes_" not in source
+    assert "cyborg" not in source.casefold()
+    assert "nasim" not in source.casefold()
+    assert "cyberbattlesim" not in source.casefold()
+
+
+@pytest.mark.parametrize("member_kind", ["directory", "symlink"])
+def test_inventory_entry_rejects_non_regular_members(
+    tmp_path: Path,
+    member_kind: str,
+) -> None:
+    member = tmp_path / "member"
+    if member_kind == "directory":
+        member.mkdir()
+    else:
+        target = tmp_path / "target"
+        target.write_bytes(b"target")
+        member.symlink_to(target)
+
+    with pytest.raises(ValueError, match="inventory member is not a regular file"):
+        inventory_entry(tmp_path, member)
+
+
+def test_inventory_entry_rejects_member_outside_root(tmp_path: Path) -> None:
+    root = tmp_path / "root"
+    root.mkdir()
+    outside = tmp_path / "outside"
+    outside.write_bytes(b"outside")
+
+    with pytest.raises(ValueError, match="inventory member escapes its root"):
+        inventory_entry(root, outside)
+
+
+def test_inventory_entry_rejects_lexically_unrelated_member(tmp_path: Path) -> None:
+    target = tmp_path / "target"
+    target.mkdir()
+    member = target / "member"
+    member.write_bytes(b"member")
+    root_alias = tmp_path / "root-alias"
+    root_alias.symlink_to(target, target_is_directory=True)
+
+    with pytest.raises(ValueError, match="inventory member escapes its root"):
+        inventory_entry(root_alias, member)