From c0ef66eff095be1fc9455885f161507fcd9d50ef Mon Sep 17 00:00:00 2001 From: AaronAspinwall123 Date: Fri, 28 Aug 2026 15:29:19 -0700 Subject: [PATCH] Add workspace-scoped runtime path policy Introduce canonical workspace roots and typed containment for configs, inputs, outputs, and managed trees. Cover traversal, external roots, linked paths, and Windows path normalization with focused tests. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 11b8494c-d35b-46ff-b5d5-bbe3dbff0586 --- assert_ai/core/runtime_path_policy.py | 499 ++++++++++++++++++++++++++ assert_ai/core/workspace.py | 69 ++++ tests/test_runtime_path_policy.py | 390 ++++++++++++++++++++ 3 files changed, 958 insertions(+) create mode 100644 assert_ai/core/runtime_path_policy.py create mode 100644 assert_ai/core/workspace.py create mode 100644 tests/test_runtime_path_policy.py diff --git a/assert_ai/core/runtime_path_policy.py b/assert_ai/core/runtime_path_policy.py new file mode 100644 index 000000000..fd82af9bc --- /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 000000000..028a604fd --- /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 000000000..9f685b92a --- /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