@@ -30,8 +30,23 @@ def config(tmp_path):
 
 
 @pytest.fixture
-def native_double(monkeypatch):
+def native_double(monkeypatch, tmp_path):
     calls = []
+    # Metadata now streams through Popen rather than the captured-log helper.
+    # Keep this explicitly inert command fixture at the existing boundary.
+    import os
+    import sys
+    executable = tmp_path / "fixture-gmx"
+    executable.write_text(
+        f"#!{sys.executable}\nimport sys\n"
+        "if sys.argv[1:3] == ['dump', '-s']:\n"
+        " print('dt = 0.001\\ntinit = 7\\nnsteps = 7\\nld-seed = 73')\n"
+        "elif sys.argv[1:3] == ['dump', '-cp']:\n"
+        " print('step = 7\\nt = 7.007')\n"
+        "else: sys.exit(2)\n"
+    )
+    executable.chmod(0o755)
+    monkeypatch.setenv("PATH", str(tmp_path) + os.pathsep + os.environ["PATH"])
 
     def run(command, *, cwd, log_path, stdin_text=None):
         calls.append(command)
@@ -51,10 +66,6 @@ def run(command, *, cwd, log_path, stdin_text=None):
                 (prefix.parent / "pullx.xvg").write_text("7.007 0.6\n")
         elif command[1] == "editconf":
             Path(command[command.index("-o") + 1]).write_text("inert PDB\n")
-        elif command[1:3] == ["dump", "-s"]:
-            text = "    dt = 0.001\n    tinit = 7\n    nsteps = 7\n    ld-seed = 73\n"
-        elif command[1:3] == ["dump", "-cp"]:
-            text = "step = 7\nt = 7.007\n"
         else:
             pytest.fail(f"unexpected native command: {command}")
         log_path.write_text(text)
@@ -80,6 +91,7 @@ def test_ordered_native_roles_and_no_invented_trajectory(tmp_path, native_double
         assert "-maxwarn" not in command
     assert (tmp_path / "sample/sample.mdp").read_text() == render_mdp("sample", cfg, 0)
     assert manifest["final_stage"] == "sample"
+    assert not list(tmp_path.glob("*/*.dump.txt"))
     assert manifest["window_id"] == "w0" and manifest["replicate_index"] == 0
     assert manifest["native_endpoints"]["sample"]["step"] == 7
     assert manifest["native_endpoints"]["sample"]["time_ps"] == 7.007
 
@@ -17,7 +17,23 @@ def _preprocess_pdb(source: Path, destination: Path) -> list[str]:
     records: list[str] = []
     residue_order: list[tuple[str, str, str]] = []
     sulfurs: dict[tuple[str, str, str], tuple[float, float, float]] = {}
-    for raw in source.read_text(encoding="utf-8", errors="strict").splitlines():
+    source_text = source.read_text(encoding="utf-8", errors="strict")
+    if any(line.lstrip().startswith("_atom_site.") for line in source_text.splitlines()):
+        # Managed source bytes remain unchanged in source.pdb and its manifest.
+        # Normalize CIF only for Amber's PDB reader, using the installed parser.
+        # Retain every model; choosing an ensemble member is not preparation's job.
+        import io
+        from openmm.app import PDBFile, PDBxFile
+
+        parsed = PDBxFile(io.StringIO(source_text))
+        normalized = io.StringIO()
+        PDBFile.writeHeader(parsed.topology, normalized)
+        for frame in range(parsed.getNumFrames()):
+            PDBFile.writeModel(parsed.topology, parsed.getPositions(frame=frame), normalized,
+                               modelIndex=frame + 1, keepIds=True)
+        PDBFile.writeFooter(parsed.topology, normalized)
+        source_text = normalized.getvalue()
+    for raw in source_text.splitlines():
         if raw.startswith("TER"):
             if records and not records[-1].startswith("TER"):
                 records.append(raw.ljust(80))
 
@@ -199,7 +199,7 @@ def _expected_snapshot_metadata(input_config: Mapping[str, Any], field: str) ->
 
 
 def _verify_open_snapshot(source: Path, expected_digest: str, expected_bytes: int) -> None:
-    from lib.portable_inputs import resolve_input_path
+    from scripts.lib.portable_inputs import resolve_input_path
     source = resolve_input_path(source)
     digest = hashlib.sha256()
     consumed = 0
@@ -296,7 +296,7 @@ def _copy_verified_snapshot(
     expected_digest: str,
     expected_bytes: int,
 ) -> None:
-    from lib.portable_inputs import resolve_input_path
+    from scripts.lib.portable_inputs import resolve_input_path
     source = resolve_input_path(source)
     destination.parent.mkdir(parents=True, exist_ok=True)
     descriptor, temporary_name = tempfile.mkstemp(prefix=".verified.", suffix=".tmp", dir=destination.parent)
 
@@ -82,9 +82,26 @@ def run_native_stages(config: Mapping[str, Any], *, config_path: Path, output_di
     publication_errors = {}
 
     def observe_dump(flag: str, path: Path, log_path: Path) -> str:
+        # TPR/CPT dumps contain entire coordinate/state arrays. Consume them
+        # incrementally and retain only the native scalar metadata used below.
+        from collections import deque
+
+        scalars = []
+        tail = deque(maxlen=20)
+        scalar = re.compile(r"^\s*(?:integrator|nsteps|init-step|dt|tinit|ld-seed|pull-ncoords|step|t)\s*=")
         try:
-            return _run_command([gmx_binary, "dump", flag, str(path)], cwd=path.parent, log_path=log_path)
-        except RuntimeError as exc:
+            with subprocess.Popen([gmx_binary, "dump", flag, str(path)], cwd=path.parent,
+                                  stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) as process:
+                assert process.stdout is not None
+                for line in process.stdout:
+                    tail.append(line)
+                    if scalar.match(line):
+                        scalars.append(line)
+                returncode = process.wait()
+            if returncode:
+                raise RuntimeError(f"GROMACS metadata dump failed with exit code {returncode}: " + "".join(tail))
+            return "".join(scalars)
+        except (RuntimeError, OSError) as exc:
             publication_errors[str(log_path.relative_to(output_dir))] = str(exc)
             return ""
     for stage in stages: