diff --git a/assert_ai/core/runtime_path_policy.py b/assert_ai/core/runtime_path_policy.py new file mode 100644 index 00000000..fd82af9b --- /dev/null +++ b/assert_ai/core/runtime_path_policy.py @@ -0,0 +1,499 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Workspace-aware runtime path resolution and containment policy.""" + +from __future__ import annotations + +import os +import stat +from dataclasses import dataclass +from enum import StrEnum +from pathlib import Path +from typing import Iterable + + +class RuntimePathErrorCode(StrEnum): + """Stable machine-readable categories for runtime path failures.""" + + INVALID_ROOT = "invalid_root" + OUTSIDE_CONFIG_ROOT = "outside_config_root" + OUTSIDE_INPUT_ROOT = "outside_input_root" + OUTSIDE_WORKSPACE = "outside_workspace" + OUTSIDE_ARTIFACTS_ROOT = "outside_artifacts_root" + OUTSIDE_EXPECTED_ROOT = "outside_expected_root" + MANAGED_ROOT_OVERRIDE = "managed_root_override" + MANAGED_PATH_LINK = "managed_path_link" + PATH_NOT_FOUND = "path_not_found" + NOT_A_FILE = "not_a_file" + + +class RuntimePathError(ValueError): + """Typed path-policy failure suitable for application error mapping.""" + + def __init__( + self, + code: RuntimePathErrorCode, + message: str, + *, + field_name: str, + path: Path | None = None, + expected_root: Path | None = None, + ) -> None: + super().__init__(message) + self.code = code + self.field_name = field_name + self.path = path + self.expected_root = expected_root + + +def _is_within(path: Path, root: Path) -> bool: + path = _comparison_path(path) + root = _comparison_path(root) + try: + path.relative_to(root) + return True + except ValueError: + return False + + +def _comparison_path(path: Path) -> Path: + """Normalize equivalent Windows extended-length paths for comparison.""" + value = os.path.normpath(os.fspath(path)) + if os.name == "nt": + value = value.replace("/", "\\") + if value.startswith("\\\\?\\UNC\\"): + value = "\\\\" + value[8:] + elif value.startswith("\\\\?\\"): + value = value[4:] + value = os.path.normcase(value) + return Path(value) + + +def _resolved(path: str | Path) -> Path: + return Path(path).expanduser().resolve() + + +def _deduplicate_paths(paths: Iterable[Path]) -> tuple[Path, ...]: + unique: list[Path] = [] + for path in paths: + if path not in unique: + unique.append(path) + return tuple(unique) + + +@dataclass(frozen=True, slots=True) +class RuntimePathPolicy: + """Resolve runtime paths against explicit workspace roots.""" + + workspace_root: Path + config_root: Path + artifacts_root: Path + results_root: Path + additional_read_roots: tuple[Path, ...] = () + allow_absolute_inputs: bool = False + force_managed_outputs: bool = True + + def __post_init__(self) -> None: + workspace_root = _resolved(self.workspace_root) + if not workspace_root.is_dir(): + raise RuntimePathError( + RuntimePathErrorCode.INVALID_ROOT, + f"Workspace root is not a directory: {workspace_root}", + field_name="workspace_root", + path=workspace_root, + ) + + config_root = _resolved(self.config_root) + artifacts_root = _resolved(self.artifacts_root) + results_root = _resolved(self.results_root) + additional_read_roots = _deduplicate_paths( + _resolved(root) for root in self.additional_read_roots + ) + + for field_name, root in ( + ("config_root", config_root), + ("artifacts_root", artifacts_root), + ("results_root", results_root), + ): + if not _is_within(root, workspace_root): + raise RuntimePathError( + RuntimePathErrorCode.INVALID_ROOT, + f"{field_name} must be inside workspace_root", + field_name=field_name, + path=root, + expected_root=workspace_root, + ) + + if self.force_managed_outputs and not _is_within(results_root, artifacts_root): + raise RuntimePathError( + RuntimePathErrorCode.INVALID_ROOT, + "results_root must be inside artifacts_root", + field_name="results_root", + path=results_root, + expected_root=artifacts_root, + ) + + object.__setattr__(self, "workspace_root", workspace_root) + object.__setattr__(self, "config_root", config_root) + object.__setattr__(self, "artifacts_root", artifacts_root) + object.__setattr__(self, "results_root", results_root) + object.__setattr__(self, "additional_read_roots", additional_read_roots) + + @property + def read_roots(self) -> tuple[Path, ...]: + return _deduplicate_paths( + ( + self.workspace_root, + self.config_root, + self.artifacts_root, + *self.additional_read_roots, + ) + ) + + def resolve_config_path( + self, + path: str | Path, + *, + must_exist: bool = False, + reject_links: bool = False, + ) -> Path: + """Resolve a config path strictly under ``config_root``.""" + candidate = Path(path).expanduser() + if candidate.is_absolute(): + unresolved = candidate + else: + parts = candidate.parts + if parts and parts[0] == self.config_root.name: + candidate = Path(*parts[1:]) if len(parts) > 1 else Path() + unresolved = self.config_root / candidate + self._require_within( + Path(os.path.abspath(unresolved)), + self.config_root, + field_name="config", + code=RuntimePathErrorCode.OUTSIDE_CONFIG_ROOT, + ) + if reject_links: + self._require_no_links( + unresolved, + self.config_root, + field_name="config", + ) + resolved = unresolved.resolve() + self._require_within( + resolved, + self.config_root, + field_name="config", + code=RuntimePathErrorCode.OUTSIDE_CONFIG_ROOT, + ) + self._require_kind( + resolved, + field_name="config", + must_exist=must_exist, + file_only=must_exist, + ) + return resolved + + def resolve_input( + self, + path: str | Path, + *, + base_dir: Path, + field_name: str, + must_exist: bool = False, + file_only: bool = False, + ) -> Path: + """Resolve an input path without allowing relative root escapes.""" + candidate = Path(path).expanduser() + if candidate.is_absolute(): + resolved = candidate.resolve() + if not self.allow_absolute_inputs: + self._require_within_any_read_root(resolved, field_name=field_name) + else: + artifact_relative = self._artifact_relative(candidate) + root = self.artifacts_root if artifact_relative is not None else _resolved(base_dir) + self._require_within_any_read_root(root, field_name=f"{field_name} base directory") + suffix = artifact_relative if artifact_relative is not None else candidate + resolved = (root / suffix).resolve() + self._require_within( + resolved, + root, + field_name=field_name, + code=RuntimePathErrorCode.OUTSIDE_INPUT_ROOT, + ) + self._require_kind( + resolved, + field_name=field_name, + must_exist=must_exist, + file_only=file_only, + ) + return resolved + + def resolve_output( + self, + path: str | Path, + *, + field_name: str, + ) -> Path: + """Resolve an output path under the managed artifacts root.""" + resolved = self._output_candidate(path).resolve() + if self.force_managed_outputs: + self._require_within( + resolved, + self.artifacts_root, + field_name=field_name, + code=RuntimePathErrorCode.OUTSIDE_ARTIFACTS_ROOT, + ) + return resolved + + def resolve_managed_output( + self, + path: str | Path, + *, + field_name: str, + expected_root: str | Path, + reject_links: bool = False, + ) -> Path: + """Resolve an output within one operation-specific managed root.""" + expected_candidate = Path(expected_root).expanduser() + if not expected_candidate.is_absolute(): + expected_candidate = self._output_candidate(expected_candidate) + expected = expected_candidate.resolve() + raw_candidate = Path(path).expanduser() + if ( + raw_candidate.is_absolute() + or self._artifact_relative(raw_candidate) is not None + ): + candidate = self._output_candidate(raw_candidate) + else: + candidate = expected / raw_candidate + self._require_within( + expected, + self.artifacts_root, + field_name=f"{field_name} expected root", + code=RuntimePathErrorCode.OUTSIDE_ARTIFACTS_ROOT, + ) + resolved = candidate.resolve() + self._require_within( + resolved, + self.artifacts_root, + field_name=field_name, + code=RuntimePathErrorCode.OUTSIDE_ARTIFACTS_ROOT, + ) + self._require_within( + resolved, + expected, + field_name=field_name, + code=RuntimePathErrorCode.OUTSIDE_EXPECTED_ROOT, + ) + if reject_links: + self._require_no_links( + expected_candidate, + self.artifacts_root, + field_name=f"{field_name} expected root", + ) + self._require_no_links( + candidate, + expected, + field_name=field_name, + ) + return resolved + + def resolve_workspace_path( + self, + path: str | Path, + *, + field_name: str, + must_exist: bool = False, + file_only: bool = False, + ) -> Path: + """Resolve a path relative to the workspace and keep it contained.""" + candidate = Path(path).expanduser() + resolved = ( + candidate.resolve() + if candidate.is_absolute() + else (self.workspace_root / candidate).resolve() + ) + self.require_workspace_path(resolved, field_name=field_name) + self._require_kind( + resolved, + field_name=field_name, + must_exist=must_exist, + file_only=file_only, + ) + return resolved + + def require_managed_tree( + self, + path: str | Path, + *, + field_name: str, + expected_root: str | Path, + ) -> Path: + """Reject links or junctions anywhere in an existing managed tree.""" + root = self.resolve_managed_output( + path, + field_name=field_name, + expected_root=expected_root, + reject_links=True, + ) + if not root.is_dir(): + return root + for current_root, dir_names, file_names in os.walk( + root, + followlinks=False, + ): + current = Path(current_root) + for name in (*dir_names, *file_names): + self.resolve_managed_output( + current / name, + field_name=f"{field_name} entry", + expected_root=root, + reject_links=True, + ) + return root + + def require_workspace_path(self, path: str | Path, *, field_name: str) -> Path: + """Re-resolve and require a path to remain inside the workspace.""" + resolved = _resolved(path) + self._require_within( + resolved, + self.workspace_root, + field_name=field_name, + code=RuntimePathErrorCode.OUTSIDE_WORKSPACE, + ) + return resolved + + def module_search_roots(self, config_path: Path | None) -> tuple[tuple[str, Path], ...]: + """Return the only roots strict dynamic imports may add to ``sys.path``.""" + roots: list[tuple[str, Path]] = [] + if config_path is not None: + config_dir = self.require_workspace_path( + config_path.parent, + field_name="config module root", + ) + roots.append(("Relative to config", config_dir)) + if self.workspace_root not in {root for _, root in roots}: + roots.append(("Relative to workspace", self.workspace_root)) + return tuple(roots) + + def require_managed_root( + self, + configured: Path, + expected: Path, + *, + field_name: str, + ) -> None: + """Reject a config root override that differs from the managed root.""" + if configured != expected: + raise RuntimePathError( + RuntimePathErrorCode.MANAGED_ROOT_OVERRIDE, + f"{field_name} is managed by the runtime and cannot be overridden", + field_name=field_name, + path=configured, + expected_root=expected, + ) + + def _artifact_relative(self, path: Path) -> Path | None: + parts = path.parts + if not parts or parts[0] not in {"artifacts", self.artifacts_root.name}: + return None + return Path(*parts[1:]) if len(parts) > 1 else Path() + + def _output_candidate(self, path: str | Path) -> Path: + candidate = Path(path).expanduser() + if candidate.is_absolute(): + return candidate + artifact_relative = self._artifact_relative(candidate) + suffix = artifact_relative if artifact_relative is not None else candidate + return self.artifacts_root / suffix + + @staticmethod + def _require_no_links( + path: Path, + root: Path, + *, + field_name: str, + ) -> None: + normalized = _comparison_path(Path(os.path.abspath(path))) + comparison_root = _comparison_path(root) + try: + relative = normalized.relative_to(comparison_root) + except ValueError: + return + current = root + for part in relative.parts: + current /= part + is_junction = getattr(current, "is_junction", None) + is_reparse_point = False + if os.name == "nt": + try: + attributes = os.lstat(current).st_file_attributes + except (AttributeError, FileNotFoundError, OSError): + attributes = 0 + is_reparse_point = bool( + attributes & stat.FILE_ATTRIBUTE_REPARSE_POINT + ) + if ( + current.is_symlink() + or (callable(is_junction) and is_junction()) + or is_reparse_point + ): + raise RuntimePathError( + RuntimePathErrorCode.MANAGED_PATH_LINK, + f"{field_name} cannot traverse a symbolic link or junction", + field_name=field_name, + path=current, + expected_root=root, + ) + + def _require_within_any_read_root(self, path: Path, *, field_name: str) -> None: + if any(_is_within(path, root) for root in self.read_roots): + return + raise RuntimePathError( + RuntimePathErrorCode.OUTSIDE_INPUT_ROOT, + f"{field_name} is outside the configured read roots", + field_name=field_name, + path=path, + ) + + @staticmethod + def _require_within( + path: Path, + root: Path, + *, + field_name: str, + code: RuntimePathErrorCode, + ) -> None: + if _is_within(path, root): + return + raise RuntimePathError( + code, + f"{field_name} escapes its expected root directory", + field_name=field_name, + path=path, + expected_root=root, + ) + + @staticmethod + def _require_kind( + path: Path, + *, + field_name: str, + must_exist: bool, + file_only: bool, + ) -> None: + if must_exist and not path.exists(): + raise RuntimePathError( + RuntimePathErrorCode.PATH_NOT_FOUND, + f"{field_name} does not exist: {path}", + field_name=field_name, + path=path, + ) + if file_only and path.exists() and not path.is_file(): + raise RuntimePathError( + RuntimePathErrorCode.NOT_A_FILE, + f"{field_name} is not a file: {path}", + field_name=field_name, + path=path, + ) diff --git a/assert_ai/core/workspace.py b/assert_ai/core/workspace.py new file mode 100644 index 00000000..028a604f --- /dev/null +++ b/assert_ai/core/workspace.py @@ -0,0 +1,69 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Workspace layout and safe path references for application services.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable + +from assert_ai.core.runtime_path_policy import RuntimePathPolicy + + +@dataclass(frozen=True, slots=True) +class WorkspaceService: + """Canonical workspace roots shared by application services.""" + + root: Path + configs_root: Path + artifacts_root: Path + results_root: Path + path_policy: RuntimePathPolicy + + @classmethod + def create( + cls, + root: str | Path, + *, + additional_read_roots: Iterable[str | Path] = (), + ) -> "WorkspaceService": + workspace_root = Path(root).expanduser().resolve(strict=True) + configs_root = workspace_root / "evals" + artifacts_root = workspace_root / "artifacts" + results_root = artifacts_root / "results" + policy = RuntimePathPolicy( + workspace_root=workspace_root, + config_root=configs_root, + artifacts_root=artifacts_root, + results_root=results_root, + additional_read_roots=tuple(Path(path) for path in additional_read_roots), + allow_absolute_inputs=False, + force_managed_outputs=True, + ) + return cls( + root=workspace_root, + configs_root=policy.config_root, + artifacts_root=policy.artifacts_root, + results_root=policy.results_root, + path_policy=policy, + ) + + def resolve_file(self, path: str | Path, *, field_name: str) -> Path: + """Resolve an existing workspace-contained file.""" + return self.path_policy.resolve_workspace_path( + path, + field_name=field_name, + must_exist=True, + file_only=True, + ) + + def reference(self, path: str | Path) -> str: + """Return a workspace-relative, forward-slash reference.""" + resolved = self.path_policy.require_workspace_path( + path, + field_name="workspace reference", + ) + relative = resolved.relative_to(self.root) + return "." if not relative.parts else relative.as_posix() diff --git a/tests/test_runtime_path_policy.py b/tests/test_runtime_path_policy.py new file mode 100644 index 00000000..9f685b92 --- /dev/null +++ b/tests/test_runtime_path_policy.py @@ -0,0 +1,390 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from assert_ai.core.runtime_path_policy import ( + RuntimePathError, + RuntimePathErrorCode, + RuntimePathPolicy, + _is_within, +) +from assert_ai.core.workspace import WorkspaceService + + +def _workspace( + tmp_path: Path, + *, + additional_read_roots: tuple[Path, ...] = (), +) -> tuple[WorkspaceService, Path]: + root = tmp_path / "workspace" + configs = root / "evals" + configs.mkdir(parents=True) + config_path = configs / "eval_config.yaml" + config_path.write_text("pipeline: {}\n", encoding="utf-8") + return ( + WorkspaceService.create( + root, + additional_read_roots=additional_read_roots, + ), + config_path, + ) + + +def test_workspace_service_exposes_canonical_relative_roots( + tmp_path: Path, +) -> None: + workspace, _ = _workspace(tmp_path) + + assert workspace.reference(workspace.root) == "." + assert workspace.reference(workspace.configs_root) == "evals" + assert workspace.reference(workspace.artifacts_root) == "artifacts" + assert workspace.reference(workspace.results_root) == "artifacts/results" + + +def test_workspace_reference_rejects_external_paths(tmp_path: Path) -> None: + workspace, _ = _workspace(tmp_path) + + with pytest.raises(RuntimePathError) as exc: + workspace.reference(tmp_path / "outside") + + assert exc.value.code is RuntimePathErrorCode.OUTSIDE_WORKSPACE + assert exc.value.field_name == "workspace reference" + + +def test_workspace_requires_an_existing_root(tmp_path: Path) -> None: + with pytest.raises(FileNotFoundError): + WorkspaceService.create(tmp_path / "missing") + + +@pytest.mark.skipif(os.name != "nt", reason="Windows path representation") +def test_extended_length_path_is_compared_as_the_same_windows_path() -> None: + root = Path("C:/workspace") + + assert _is_within( + Path("//?/C:/workspace/artifacts/results"), + root, + ) + assert not _is_within(Path("//?/C:/outside"), root) + + +def test_policy_rejects_managed_roots_outside_workspace( + tmp_path: Path, +) -> None: + root = tmp_path / "workspace" + root.mkdir() + + with pytest.raises(RuntimePathError) as exc: + RuntimePathPolicy( + workspace_root=root, + config_root=tmp_path / "evals", + artifacts_root=root / "artifacts", + results_root=root / "artifacts" / "results", + ) + + assert exc.value.code is RuntimePathErrorCode.INVALID_ROOT + assert exc.value.field_name == "config_root" + + +def test_config_paths_are_contained_under_config_root( + tmp_path: Path, +) -> None: + workspace, config_path = _workspace(tmp_path) + + assert ( + workspace.path_policy.resolve_config_path("eval_config.yaml") + == config_path + ) + assert ( + workspace.path_policy.resolve_config_path("evals/eval_config.yaml") + == config_path + ) + + for candidate in ("../outside.yaml", tmp_path / "outside.yaml"): + with pytest.raises(RuntimePathError) as exc: + workspace.path_policy.resolve_config_path(candidate) + assert exc.value.code is RuntimePathErrorCode.OUTSIDE_CONFIG_ROOT + + +def test_config_path_can_require_an_existing_file(tmp_path: Path) -> None: + workspace, _ = _workspace(tmp_path) + directory = workspace.configs_root / "nested" + directory.mkdir() + + with pytest.raises(RuntimePathError) as missing: + workspace.path_policy.resolve_config_path( + "missing.yaml", + must_exist=True, + ) + with pytest.raises(RuntimePathError) as not_file: + workspace.path_policy.resolve_config_path( + "nested", + must_exist=True, + ) + + assert missing.value.code is RuntimePathErrorCode.PATH_NOT_FOUND + assert not_file.value.code is RuntimePathErrorCode.NOT_A_FILE + + +def test_relative_input_cannot_escape_its_base_directory( + tmp_path: Path, +) -> None: + workspace, config_path = _workspace(tmp_path) + shared = workspace.root / "shared.jsonl" + shared.write_text("{}\n", encoding="utf-8") + + with pytest.raises(RuntimePathError) as exc: + workspace.path_policy.resolve_input( + "../shared.jsonl", + base_dir=config_path.parent, + field_name="pipeline.inference.test_set_path", + ) + + assert exc.value.code is RuntimePathErrorCode.OUTSIDE_INPUT_ROOT + + +def test_artifact_relative_input_uses_artifacts_root(tmp_path: Path) -> None: + workspace, config_path = _workspace(tmp_path) + artifact = workspace.artifacts_root / "inputs" / "cases.jsonl" + artifact.parent.mkdir(parents=True) + artifact.write_text("{}\n", encoding="utf-8") + + assert ( + workspace.path_policy.resolve_input( + "artifacts/inputs/cases.jsonl", + base_dir=config_path.parent, + field_name="test cases", + must_exist=True, + file_only=True, + ) + == artifact + ) + + +def test_absolute_input_requires_an_approved_read_root( + tmp_path: Path, +) -> None: + external_root = tmp_path / "approved-inputs" + external_root.mkdir() + external = external_root / "cases.jsonl" + external.write_text("{}\n", encoding="utf-8") + workspace, config_path = _workspace(tmp_path) + + with pytest.raises(RuntimePathError) as exc: + workspace.path_policy.resolve_input( + external, + base_dir=config_path.parent, + field_name="test cases", + ) + assert exc.value.code is RuntimePathErrorCode.OUTSIDE_INPUT_ROOT + + approved, approved_config = _workspace( + tmp_path / "approved", + additional_read_roots=(external_root,), + ) + assert ( + approved.path_policy.resolve_input( + external, + base_dir=approved_config.parent, + field_name="test cases", + must_exist=True, + ) + == external + ) + + +def test_legacy_policy_can_allow_external_absolute_inputs( + tmp_path: Path, +) -> None: + workspace, config_path = _workspace(tmp_path) + external = tmp_path / "external.jsonl" + external.write_text("{}\n", encoding="utf-8") + policy = RuntimePathPolicy( + workspace_root=workspace.root, + config_root=workspace.configs_root, + artifacts_root=workspace.artifacts_root, + results_root=workspace.results_root, + allow_absolute_inputs=True, + ) + + assert ( + policy.resolve_input( + external, + base_dir=config_path.parent, + field_name="legacy input", + must_exist=True, + ) + == external + ) + + +def test_outputs_are_contained_under_artifacts_root(tmp_path: Path) -> None: + workspace, _ = _workspace(tmp_path) + + assert ( + workspace.path_policy.resolve_output( + "reports/summary.json", + field_name="report", + ) + == workspace.artifacts_root / "reports" / "summary.json" + ) + assert ( + workspace.path_policy.resolve_output( + "artifacts/results/suite-a", + field_name="suite", + ) + == workspace.results_root / "suite-a" + ) + + for candidate in ("../outside", tmp_path / "outside"): + with pytest.raises(RuntimePathError) as exc: + workspace.path_policy.resolve_output( + candidate, + field_name="report", + ) + assert exc.value.code is RuntimePathErrorCode.OUTSIDE_ARTIFACTS_ROOT + + +def test_managed_output_is_confined_to_operation_root( + tmp_path: Path, +) -> None: + workspace, _ = _workspace(tmp_path) + suite_root = workspace.results_root / "suite-a" + + assert ( + workspace.path_policy.resolve_managed_output( + "run-a", + field_name="run root", + expected_root=suite_root, + ) + == suite_root / "run-a" + ) + + with pytest.raises(RuntimePathError) as exc: + workspace.path_policy.resolve_managed_output( + workspace.results_root / "suite-b" / "run-b", + field_name="run root", + expected_root=suite_root, + ) + + assert exc.value.code is RuntimePathErrorCode.OUTSIDE_EXPECTED_ROOT + + +def test_workspace_file_resolution_checks_kind(tmp_path: Path) -> None: + workspace, config_path = _workspace(tmp_path) + + assert ( + workspace.resolve_file("evals/eval_config.yaml", field_name="config") + == config_path + ) + + with pytest.raises(RuntimePathError) as outside: + workspace.resolve_file( + tmp_path / "outside.yaml", + field_name="config", + ) + with pytest.raises(RuntimePathError) as directory: + workspace.resolve_file("evals", field_name="config") + + assert outside.value.code is RuntimePathErrorCode.OUTSIDE_WORKSPACE + assert directory.value.code is RuntimePathErrorCode.NOT_A_FILE + + +def test_config_path_can_reject_links_inside_config_root( + tmp_path: Path, + symlink_or_skip, +) -> None: + workspace, config_path = _workspace(tmp_path) + link = workspace.configs_root / "linked.yaml" + symlink_or_skip(link, config_path) + + assert workspace.path_policy.resolve_config_path(link) == config_path + with pytest.raises(RuntimePathError) as exc: + workspace.path_policy.resolve_config_path( + link, + reject_links=True, + ) + + assert exc.value.code is RuntimePathErrorCode.MANAGED_PATH_LINK + + +def test_managed_output_can_reject_links( + tmp_path: Path, + symlink_or_skip, +) -> None: + workspace, _ = _workspace(tmp_path) + suite_root = workspace.results_root / "suite-a" + suite_root.mkdir(parents=True) + target = suite_root / "result.json" + target.write_text("{}\n", encoding="utf-8") + link = suite_root / "linked.json" + symlink_or_skip(link, target) + + with pytest.raises(RuntimePathError) as exc: + workspace.path_policy.resolve_managed_output( + link, + field_name="result", + expected_root=suite_root, + reject_links=True, + ) + + assert exc.value.code is RuntimePathErrorCode.MANAGED_PATH_LINK + + +def test_managed_tree_rejects_nested_links( + tmp_path: Path, + symlink_or_skip, +) -> None: + workspace, _ = _workspace(tmp_path) + suite_root = workspace.results_root / "suite-a" + run_root = suite_root / "run-a" + run_root.mkdir(parents=True) + target = run_root / "scores.jsonl" + target.write_text("{}\n", encoding="utf-8") + symlink_or_skip(run_root / "scores-link.jsonl", target) + + with pytest.raises(RuntimePathError) as exc: + workspace.path_policy.require_managed_tree( + suite_root, + field_name="suite tree", + expected_root=workspace.results_root, + ) + + assert exc.value.code is RuntimePathErrorCode.MANAGED_PATH_LINK + + +def test_module_search_roots_stay_inside_workspace(tmp_path: Path) -> None: + workspace, config_path = _workspace(tmp_path) + + assert workspace.path_policy.module_search_roots(config_path) == ( + ("Relative to config", workspace.configs_root), + ("Relative to workspace", workspace.root), + ) + + with pytest.raises(RuntimePathError) as exc: + workspace.path_policy.module_search_roots(tmp_path / "outside.yaml") + + assert exc.value.code is RuntimePathErrorCode.OUTSIDE_WORKSPACE + + +def test_managed_root_override_is_rejected(tmp_path: Path) -> None: + workspace, _ = _workspace(tmp_path) + + workspace.path_policy.require_managed_root( + workspace.artifacts_root, + workspace.artifacts_root, + field_name="artifacts_root", + ) + with pytest.raises(RuntimePathError) as exc: + workspace.path_policy.require_managed_root( + workspace.results_root, + workspace.artifacts_root, + field_name="artifacts_root", + ) + + assert exc.value.code is RuntimePathErrorCode.MANAGED_ROOT_OVERRIDE