Make v2 manifest artifact paths portable

docs/source/user_guide/agent_cli.rst

@@ -141,6 +141,26 @@ high-pass cleaning criteria off unless the YAML config sets them explicitly.
 Set ``highpass`` as a two-value transition band, for example
 ``highpass: [0.25, 0.75]``.
 
+Manifest Portability And Migration
+==================================
+
+The ``eegprep.manifest.v2`` schema stores the ``path`` fields in
+``input_files`` and ``output_files`` relative to the manifest file. Stored
+relative paths always use forward slashes. Runtime manifests created by
+``build_manifest()`` use absolute paths, and ``read_manifest()`` restores v2
+relative paths to absolute paths for the current machine.
+
+This preserves artifact relationships when the manifest and files move
+together. References using ``..`` only remain valid when the referenced files
+move with the same directory layout. A path on another Windows drive cannot be
+made relative and remains absolute, so it is not portable to another machine.
+
+Version 1 and unknown schema versions keep their original path strings when
+read or written; EEGPrep cannot infer what their relative paths were relative
+to. Consumers that need resolved v2 paths should use ``read_manifest()``
+instead of loading the JSON directly. Path serialization does not alter or
+verify the recorded SHA-256 values.
+
 QC And Reports
 ==============
 

src/eegprep/cli/commands/transforms.py

@@ -20,6 +20,7 @@
 
 from eegprep.cli.core import (
     EEGPrepCLIError,
+    MANIFEST_SCHEMA_VERSION as _MANIFEST_SCHEMA_VERSION,
     build_manifest,
     file_sha256,
     utc_now,
@@ -38,7 +39,7 @@
 logger = logging.getLogger(__name__)
 
 RESULT_SCHEMA_VERSION = "eegprep.transform_result.v1"
-MANIFEST_SCHEMA_VERSION = "eegprep.manifest.v1"
+MANIFEST_SCHEMA_VERSION = _MANIFEST_SCHEMA_VERSION
 DATASET_OUTPUT_TYPE = "eeglab_set"
 
 

src/eegprep/cli/core.py

@@ -2,14 +2,16 @@
 
 from __future__ import annotations
 
+import copy
 import hashlib
 import json
 import math
+import os
 import platform
 import sys
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from pathlib import Path
+from pathlib import Path, PurePosixPath, PureWindowsPath
 from typing import Any
 
 import numpy as np
@@ -18,6 +20,7 @@
 
 
 COMMAND_RESULT_SCHEMA_VERSION = "eegprep.cli.result.v1"
+MANIFEST_SCHEMA_VERSION = "eegprep.manifest.v2"
 
 
 class EEGPrepCLIError(Exception):
@@ -238,11 +241,12 @@ def build_manifest(
     warnings: list[Any] | None = None,
 ) -> dict[str, Any]:
     stamp = runtime_stamp(started_at) if finished_at is None else RuntimeStamp(started_at, finished_at)
+    input_records = [_input_file_record(path) for path in input_files]
     manifest: dict[str, Any] = {
-        "schema_version": "eegprep.manifest.v1",
+        "schema_version": MANIFEST_SCHEMA_VERSION,
         "command": command,
-        "input_files": [_input_file_record(path) for path in input_files],
-        "output_files": output_files,
+        "input_files": _make_manifest_paths_absolute(input_records),
+        "output_files": _make_manifest_paths_absolute(output_files),
         "parameters": json_safe(parameters),
         "history": history,
         "software": software_info(),
@@ -254,12 +258,53 @@ def build_manifest(
     return manifest
 
 
+def _make_manifest_relative(manifest: dict[str, Any], manifest_path: Path) -> dict[str, Any]:
+    """Return the on-disk representation of a v2 manifest."""
+    prepared = copy.deepcopy(json_safe(manifest))
+    if prepared.get("schema_version") != MANIFEST_SCHEMA_VERSION:
+        return prepared
+
+    base_dir = manifest_path.resolve().parent
+    for item in _manifest_file_records(prepared):
+        value = item.get("path")
+        if not isinstance(value, str) or not value:
+            continue
+        native_path = Path(value).expanduser()
+        if not native_path.is_absolute():
+            continue
+        try:
+            item["path"] = Path(os.path.relpath(native_path, start=base_dir)).as_posix()
+        except ValueError:
+            # Windows cannot express a path on another drive as a relative path.
+            continue
+    return prepared
+
+
+def read_manifest(path: str | Path) -> dict[str, Any]:
+    """Read a manifest, resolving v2 artifact paths against its directory."""
+    resolved = Path(path).expanduser().resolve()
+    with resolved.open("r", encoding="utf-8") as stream:
+        manifest = json.load(stream)
+
+    if manifest.get("schema_version") != MANIFEST_SCHEMA_VERSION:
+        return manifest
+
+    for item in _manifest_file_records(manifest):
+        value = item.get("path")
+        if not isinstance(value, str) or not value or _is_absolute_on_any_platform(value):
+            continue
+        relative_path = Path(*PurePosixPath(value).parts)
+        item["path"] = str((resolved.parent / relative_path).resolve())
+    return manifest
+
+
 def write_manifest(path: str | Path | None, manifest: dict[str, Any]) -> Path | None:
     if path is None:
         return None
     resolved = Path(path).expanduser()
     resolved.parent.mkdir(parents=True, exist_ok=True)
-    resolved.write_text(json.dumps(json_safe(manifest), indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    rel_manifest = _make_manifest_relative(manifest, resolved)
+    resolved.write_text(json.dumps(json_safe(rel_manifest), indent=2, sort_keys=True) + "\n", encoding="utf-8")
     return resolved
 
 
@@ -284,7 +329,39 @@ def write_json_file(
 
 
 def write_manifest_file(path: str | Path, manifest: dict[str, Any], *, overwrite: bool = False) -> dict[str, Any]:
-    return write_json_file(path, manifest, overwrite=overwrite, output_type="manifest")
+    target = Path(path)
+    rel_manifest = _make_manifest_relative(manifest, target)
+    return write_json_file(target, rel_manifest, overwrite=overwrite, output_type="manifest")
+
+
+def _make_manifest_paths_absolute(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Normalize runtime artifact paths before they acquire manifest-relative semantics."""
+    prepared = copy.deepcopy(json_safe(records))
+    for item in prepared:
+        value = item.get("path")
+        if not isinstance(value, str) or not value or _is_foreign_absolute_path(value):
+            continue
+        item["path"] = str(Path(value).expanduser().resolve())
+    return prepared
+
+
+def _manifest_file_records(manifest: dict[str, Any]) -> list[dict[str, Any]]:
+    records: list[dict[str, Any]] = []
+    for key in ("input_files", "output_files"):
+        value = manifest.get(key)
+        if isinstance(value, list):
+            records.extend(item for item in value if isinstance(item, dict))
+    return records
+
+
+def _is_absolute_on_any_platform(value: str) -> bool:
+    return Path(value).is_absolute() or PurePosixPath(value).is_absolute() or PureWindowsPath(value).is_absolute()
+
+
+def _is_foreign_absolute_path(value: str) -> bool:
+    return not Path(value).is_absolute() and (
+        PurePosixPath(value).is_absolute() or PureWindowsPath(value).is_absolute()
+    )
 
 
 def _input_file_record(path: Path | dict[str, Any]) -> dict[str, Any]:

tests/test_cli_pipeline_qc_report.py

@@ -4,6 +4,7 @@
 import numpy as np
 import yaml
 
+from eegprep.cli.core import read_manifest
 from eegprep.cli.commands import transforms as transforms_cli
 from eegprep.cli.commands.pipeline import (
     _channel_indices,
@@ -56,10 +57,13 @@ def test_pipeline_run_writes_qc_report_and_manifest(tmp_path):
     assert (tmp_path / "out" / "report.html").is_file()
     manifest_path = tmp_path / "out" / "eegprep_manifest.json"
     assert manifest_path.is_file()
-    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
-    assert manifest["schema_version"] == "eegprep.manifest.v1"
+    stored_manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    assert {item["path"] for item in stored_manifest["output_files"]} == {"qc.json", "report.html"}
+    manifest = read_manifest(manifest_path)
+    assert manifest["schema_version"] == "eegprep.manifest.v2"
     assert manifest["command"] == "pipeline run"
     assert {item["type"] for item in manifest["output_files"]} == {"json", "html_report"}
+    assert all(Path(item["path"]).is_absolute() for item in manifest["input_files"] + manifest["output_files"])
     assert isinstance(result["qc"]["recommendations"], list)
 
 
@@ -73,7 +77,7 @@ def test_pipeline_run_resample_writes_dataset_manifest_and_history(tmp_path):
 
     assert result["status"] == "ok"
     assert (tmp_path / "out" / "eeglab_data_eegprep.set").is_file()
-    manifest = json.loads((tmp_path / "out" / "eegprep_manifest.json").read_text(encoding="utf-8"))
+    manifest = read_manifest(tmp_path / "out" / "eegprep_manifest.json")
     assert "pop_resample" in manifest["history"]
     assert any(item["type"] == "eeglab_set" for item in manifest["output_files"])
     assert any("pop_resample" in item for item in result["history"])
@@ -301,8 +305,8 @@ def test_report_and_qc_report_write_html_and_manifests(tmp_path):
     assert qc_result["status"] == "ok"
     assert "EEGPrep Report" in (tmp_path / "dataset_report.html").read_text(encoding="utf-8")
     assert "EEGPrep QC Report" in (tmp_path / "qc_report.html").read_text(encoding="utf-8")
-    assert json.loads((tmp_path / "dataset_report.manifest.json").read_text(encoding="utf-8"))["command"] == "report"
-    assert json.loads((tmp_path / "qc_report.manifest.json").read_text(encoding="utf-8"))["command"] == "qc report"
+    assert read_manifest(tmp_path / "dataset_report.manifest.json")["command"] == "report"
+    assert read_manifest(tmp_path / "qc_report.manifest.json")["command"] == "qc report"
 
 
 def _write_pipeline_config(tmp_path, *, steps, input_path=SAMPLE_SET):

tests/test_cli_transforms.py

@@ -7,6 +7,7 @@
 import subprocess
 import sys
 
+from eegprep.cli.core import read_manifest
 from eegprep.cli.commands import transforms
 from eegprep.functions.popfunc.pop_loadset import pop_loadset
 from eegprep.functions.popfunc.pop_saveset import pop_saveset
@@ -79,11 +80,14 @@ def test_resample_writes_dataset_manifest_and_clean_json_stdout(tmp_path):
 
     manifest_path = output.with_suffix(output.suffix + ".manifest.json")
     assert payload["manifest"]["path"] == str(manifest_path)
-    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    stored_manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    assert {record["path"] for record in stored_manifest["output_files"]} == {"resampled.set"}
+    manifest = read_manifest(manifest_path)
     assert manifest["schema_version"] == transforms.MANIFEST_SCHEMA_VERSION
     assert manifest["parameters"]["freq"] == 128
     assert "pop_resample" in manifest["history"]
     assert any(record["path"].endswith("eeglab_data.fdt") for record in manifest["input_files"])
+    assert all(Path(record["path"]).is_absolute() for record in manifest["input_files"] + manifest["output_files"])
 
     loaded = pop_loadset(str(output))
     assert loaded["srate"] == 128