Merge 319a740913 into 851743adcf

strix/interface/utils.py

@@ -5,15 +5,18 @@ import os
 import re
 import secrets
 import shutil
+import stat
 import subprocess
 import sys
 import tempfile
+from collections.abc import Iterator
 from dataclasses import dataclass, field
-from pathlib import Path
-from typing import Any
-from urllib.parse import parse_qs, urlparse
+from pathlib import Path, PurePosixPath
+from typing import Any, cast
+from urllib.parse import parse_qs, unquote_to_bytes, urlparse, urlsplit
 
 import requests
+import yaml
 from rich.console import Console
 from rich.panel import Panel
 from rich.text import Text
@@ -1500,6 +1503,313 @@ def rewrite_localhost_targets(targets_info: list[dict[str, Any]], host_gateway:
 #: from wherever they happen to live on the host.
 API_SPEC_WORKSPACE_SUBDIR = "api-specs"
 
+_API_SPEC_EXTENSIONS = frozenset({".json", ".yaml", ".yml"})
+_API_SPEC_MAX_FILES = 100
+_API_SPEC_MAX_TOTAL_BYTES = 25 * 1024 * 1024
+_API_SPEC_MAX_REF_DEPTH = 25
+_API_SPEC_MAX_DOCUMENT_NODES = 100_000
+
+
+@dataclass(frozen=True)
+class _ApiSpecBundle:
+    main_name: str
+    files: dict[Path, bytes]
+
+
+def _safe_api_spec_label(path: Path) -> str:
+    label = re.sub(r"[^A-Za-z0-9._-]", "?", path.name)
+    return label[:80] or "referenced file"
+
+
+def _parse_api_spec_document(content: bytes, label: str) -> Any:
+    try:
+        text = content.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise ValueError(f"API spec file '{label}' must contain UTF-8 text.") from exc
+    try:
+        return json.loads(text)
+    except (json.JSONDecodeError, RecursionError):
+        try:
+            return yaml.safe_load(text)
+        except (yaml.YAMLError, RecursionError) as exc:
+            raise ValueError(f"API spec file '{label}' is not valid JSON or YAML.") from exc
+
+
+def _iter_api_spec_refs(document: Any, label: str) -> Iterator[str]:
+    stack = [document]
+    seen_containers: set[int] = set()
+    nodes = 0
+    while stack:
+        node = stack.pop()
+        nodes += 1
+        if nodes > _API_SPEC_MAX_DOCUMENT_NODES:
+            raise ValueError(f"API spec file '{label}' is too complex to stage safely.")
+        if isinstance(node, dict):
+            mapping = cast("dict[str, Any]", node)
+            identity = id(mapping)
+            if identity in seen_containers:
+                continue
+            seen_containers.add(identity)
+            reference = mapping.get("$ref")
+            if isinstance(reference, str):
+                yield reference
+            stack.extend(mapping.values())
+        elif isinstance(node, list):
+            sequence = cast("list[Any]", node)  # type: ignore[redundant-cast]
+            identity = id(sequence)
+            if identity in seen_containers:
+                continue
+            seen_containers.add(identity)
+            stack.extend(sequence)
+
+
+def _path_has_symlink(root: Path, path: Path) -> bool:
+    current = root
+    if current.is_symlink():
+        return True
+    for part in path.relative_to(root).parts:
+        current /= part
+        if current.is_symlink():
+            return True
+    return False
+
+
+def _read_api_spec_file(path: Path, root: Path, remaining_bytes: int) -> bytes:
+    label = _safe_api_spec_label(path)
+    if _path_has_symlink(root, path):
+        raise ValueError(f"API spec file '{label}' is a symbolic link and cannot be staged.")
+    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except OSError as exc:
+        raise ValueError(f"Cannot safely read referenced API spec file '{label}'.") from exc
+    try:
+        file_stat = os.fstat(descriptor)
+        if not stat.S_ISREG(file_stat.st_mode):
+            raise ValueError(f"Referenced API spec file '{label}' is not a regular file.")
+        with os.fdopen(descriptor, "rb", closefd=False) as stream:
+            content = stream.read(remaining_bytes + 1)
+    finally:
+        os.close(descriptor)
+    if len(content) > remaining_bytes:
+        raise ValueError(
+            f"API spec dependencies exceed the {_API_SPEC_MAX_TOTAL_BYTES}-byte staging limit."
+        )
+    return content
+
+
+def _resolve_local_api_spec_ref(reference: str, referrer: Path, root: Path) -> Path | None:
+    try:
+        parsed = urlsplit(reference)
+    except ValueError as exc:
+        raise ValueError(
+            f"API spec file '{_safe_api_spec_label(referrer)}' has an invalid reference URI."
+        ) from exc
+    if parsed.scheme or parsed.netloc:
+        if parsed.scheme in {"http", "https"} and parsed.netloc:
+            return None
+        raise ValueError(
+            f"API spec file '{_safe_api_spec_label(referrer)}' uses an unsupported reference URI."
+        )
+    if not parsed.path:
+        return None
+    if parsed.query:
+        raise ValueError(
+            f"API spec file '{_safe_api_spec_label(referrer)}' uses an unsupported local reference."
+        )
+    try:
+        decoded = unquote_to_bytes(parsed.path).decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise ValueError(
+            f"API spec file '{_safe_api_spec_label(referrer)}' has an invalid local reference."
+        ) from exc
+    if not decoded or "\x00" in decoded or "\\" in decoded:
+        raise ValueError(
+            f"API spec file '{_safe_api_spec_label(referrer)}' has an invalid local reference."
+        )
+    relative = PurePosixPath(decoded)
+    if relative.is_absolute():
+        raise ValueError(
+            f"A local reference from '{_safe_api_spec_label(referrer)}' points outside the "
+            "API spec directory. Bundle the specification under one directory."
+        )
+    if relative.suffix.lower() not in _API_SPEC_EXTENSIONS:
+        raise ValueError(
+            f"Local API spec dependency '{_safe_api_spec_label(Path(relative.name))}' must be "
+            "a JSON or YAML file."
+        )
+
+    candidate = referrer.parent.joinpath(*relative.parts).absolute()
+    if not candidate.is_relative_to(root):
+        raise ValueError(
+            f"A local reference from '{_safe_api_spec_label(referrer)}' points outside the "
+            "API spec directory. Bundle the specification under one directory."
+        )
+    if _path_has_symlink(root, candidate):
+        raise ValueError(
+            f"API spec file '{_safe_api_spec_label(candidate)}' is a symbolic link and "
+            "cannot be staged."
+        )
+    try:
+        resolved = candidate.resolve(strict=True)
+    except OSError as exc:
+        raise ValueError(
+            f"Referenced API spec file '{_safe_api_spec_label(candidate)}' does not exist "
+            "or cannot be read."
+        ) from exc
+    if not resolved.is_relative_to(root.resolve()):
+        raise ValueError(
+            f"A local reference from '{_safe_api_spec_label(referrer)}' points outside the "
+            "API spec directory. Bundle the specification under one directory."
+        )
+    return resolved
+
+
+def _collect_api_spec_bundle(
+    source: Path,
+    *,
+    follow_refs: bool,
+    available_files: int,
+    available_bytes: int,
+) -> _ApiSpecBundle:
+    source = source.absolute()
+    if source.is_symlink():
+        raise ValueError(
+            f"API spec file '{_safe_api_spec_label(source)}' is a symbolic link and "
+            "cannot be staged."
+        )
+    try:
+        root = source.parent.resolve(strict=True)
+        source = source.resolve(strict=True)
+    except OSError as exc:
+        raise ValueError(
+            f"Cannot safely read API spec file '{_safe_api_spec_label(source)}'."
+        ) from exc
+    if source.suffix.lower() not in _API_SPEC_EXTENSIONS:
+        raise ValueError("API specifications and their dependencies must be JSON or YAML files.")
+
+    queued: list[tuple[Path, int]] = [(source, 0)]
+    visited: set[Path] = set()
+    files: dict[Path, bytes] = {}
+    total_bytes = 0
+    while queued:
+        path, depth = queued.pop()
+        if path in visited:
+            continue
+        if depth > _API_SPEC_MAX_REF_DEPTH:
+            raise ValueError(
+                f"API spec reference depth exceeds the {_API_SPEC_MAX_REF_DEPTH}-level limit."
+            )
+        if len(visited) >= available_files:
+            raise ValueError(
+                f"API spec dependencies exceed the {_API_SPEC_MAX_FILES}-file staging limit."
+            )
+        content = _read_api_spec_file(path, root, available_bytes - total_bytes)
+        total_bytes += len(content)
+        visited.add(path)
+        relative_path = path.relative_to(root)
+        files[relative_path] = content
+        if not follow_refs:
+            continue
+        label = _safe_api_spec_label(relative_path)
+        document = _parse_api_spec_document(content, label)
+        for reference in _iter_api_spec_refs(document, label):
+            dependency = _resolve_local_api_spec_ref(reference, path, root)
+            if dependency is not None and dependency not in visited:
+                queued.append((dependency, depth + 1))
+    return _ApiSpecBundle(main_name=source.name, files=files)
+
+
+def _unique_api_spec_name(name: str, used: set[str]) -> str:
+    stem, suffix = Path(name).stem, Path(name).suffix
+    candidate = name
+    count = 1
+    while candidate.casefold() in used:
+        count += 1
+        candidate = f"{stem}-{count}{suffix}"
+    used.add(candidate.casefold())
+    return candidate
+
+
+def _collect_api_spec_bundles(specs: list[dict[str, Any]]) -> list[_ApiSpecBundle]:
+    bundles: list[_ApiSpecBundle] = []
+    remaining_files = _API_SPEC_MAX_FILES
+    remaining_bytes = _API_SPEC_MAX_TOTAL_BYTES
+    for target in specs:
+        bundle = _collect_api_spec_bundle(
+            Path(str(target["details"]["target_spec"])),
+            follow_refs=target["details"].get("spec_format") in {"openapi", "swagger"},
+            available_files=remaining_files,
+            available_bytes=remaining_bytes,
+        )
+        bundles.append(bundle)
+        remaining_files -= len(bundle.files)
+        remaining_bytes -= sum(len(content) for content in bundle.files.values())
+    return bundles
+
+
+def _layout_api_spec_bundles(
+    bundles: list[_ApiSpecBundle],
+) -> list[tuple[Path, dict[Path, bytes]]]:
+    used_names: set[str] = set()
+    allocated_names = [_unique_api_spec_name(bundle.main_name, used_names) for bundle in bundles]
+    used_root_entries = {
+        name.casefold()
+        for name, bundle in zip(allocated_names, bundles, strict=True)
+        if len(bundle.files) == 1
+    }
+    layouts: list[tuple[Path, dict[Path, bytes]]] = []
+    for name, bundle in zip(allocated_names, bundles, strict=True):
+        if len(bundles) == 1 or len(bundle.files) == 1:
+            main_path = Path(name)
+            files = (
+                {main_path: next(iter(bundle.files.values()))}
+                if len(bundle.files) == 1
+                else dict(bundle.files)
+            )
+        else:
+            namespace = _unique_api_spec_name(Path(name).stem, used_root_entries)
+            main_path = Path(namespace) / bundle.main_name
+            files = {
+                Path(namespace) / relative: content for relative, content in bundle.files.items()
+            }
+        layouts.append((main_path, files))
+    return layouts
+
+
+def _publish_api_spec_layouts(layouts: list[tuple[Path, dict[Path, bytes]]], run_name: str) -> Path:
+    temporary_root = Path(tempfile.gettempdir())
+    try:
+        temporary_root.mkdir(parents=True, exist_ok=True)
+        temporary_root = temporary_root.resolve(strict=True)
+        staging_parent = temporary_root / "strix_api_specs"
+        if staging_parent.is_symlink():
+            raise ValueError("API specification staging directory cannot be a symbolic link.")
+        staging_parent.mkdir(parents=True, exist_ok=True)
+        if staging_parent.resolve(strict=True) != staging_parent:
+            raise ValueError("API specification staging directory is not safe to use.")
+    except OSError as exc:
+        raise ValueError("Cannot create a safe API specification staging directory.") from exc
+
+    staging = staging_parent / run_name
+    temporary = Path(tempfile.mkdtemp(prefix=f".{run_name}-", dir=staging_parent))
+    try:
+        for _main_path, files in layouts:
+            for relative, content in files.items():
+                destination = temporary / relative
+                destination.parent.mkdir(parents=True, exist_ok=True)
+                destination.write_bytes(content)
+        if staging.is_symlink() or staging.is_file():
+            staging.unlink()
+        elif staging.exists():
+            shutil.rmtree(staging)
+        temporary.rename(staging)
+    except Exception:
+        if temporary.exists():
+            shutil.rmtree(temporary)
+        raise
+    return staging
+
 
 def write_fetched_collection(collection: dict[str, Any], collection_uid: str) -> str:
     """Write a collection fetched from the Postman API to a local file.
@@ -1515,33 +1825,29 @@ def write_fetched_collection(collection: dict[str, Any], collection_uid: str) ->
 
 
 def stage_api_specs(targets_info: list[dict[str, Any]], run_name: str) -> list[dict[str, Any]]:
-    """Copy every ``api_spec`` target into one directory for the sandbox.
+    """Stage API specs and their bounded local ``$ref`` closure for the sandbox.
 
-    A spec is a single file the agent reads, not a tree it works in, so it is
-    copied to a per-run staging directory that is exposed at
-    ``/workspace/api-specs`` instead of mounting its host location. Each target's
-    ``workspace_path`` records where the agent will find it.
+    Only explicit relative JSON/YAML references below each spec directory are
+    followed. Remote references remain unchanged, and nothing is published until
+    every local dependency has been validated and copied.
     """
     specs = [t for t in targets_info if t.get("type") == "api_spec"]
     if not specs:
         return []
-
-    staging = Path(tempfile.gettempdir()) / "strix_api_specs" / run_name
-    staging.mkdir(parents=True, exist_ok=True)
-
-    used: set[str] = set()
     for target in specs:
-        details = target["details"]
-        source = Path(str(details["target_spec"]))
-        name = source.name
-        stem, suffix = source.stem, source.suffix
-        count = 1
-        while name in used:
-            count += 1
-            name = f"{stem}-{count}{suffix}"
-        used.add(name)
-        shutil.copy2(source, staging / name)
-        details["workspace_path"] = f"/workspace/{API_SPEC_WORKSPACE_SUBDIR}/{name}"
+        target["details"].pop("workspace_path", None)
+
+    safe_run_name = sanitize_name(run_name)
+    if safe_run_name != run_name or safe_run_name in {".", ".."}:
+        raise ValueError("Invalid run name for API specification staging.")
+
+    layouts = _layout_api_spec_bundles(_collect_api_spec_bundles(specs))
+    staging = _publish_api_spec_layouts(layouts, safe_run_name)
+
+    for target, (main_path, _files) in zip(specs, layouts, strict=True):
+        target["details"]["workspace_path"] = (
+            f"/workspace/{API_SPEC_WORKSPACE_SUBDIR}/{main_path.as_posix()}"
+        )
 
     return [
         {

tests/test_api_spec_targets.py

@@ -36,6 +36,23 @@ def _write_spec(directory: Path, name: str = "openapi.json") -> Path:
     return path
 
 
+def _write_spec_with_ref(directory: Path, reference: str, name: str = "openapi.json") -> Path:
+    spec = {
+        **OPENAPI,
+        "components": {"schemas": {"User": {"$ref": reference}}},
+    }
+    directory.mkdir(parents=True, exist_ok=True)
+    path = directory / name
+    path.write_text(json.dumps(spec), encoding="utf-8")
+    return path
+
+
+@pytest.fixture(autouse=True)
+def _isolate_staging(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    staging_root = tmp_path / "temporary"
+    monkeypatch.setattr("strix.interface.utils.tempfile.gettempdir", lambda: str(staging_root))
+
+
 def _resolved_targets(*spec_paths: Path) -> list[dict[str, Any]]:
     """Run spec targets through the real setup path (detection + spec resolution)."""
     args = argparse.Namespace(target=[str(p) for p in spec_paths], target_list=None)
@@ -105,12 +122,14 @@ def test_build_targets_info_rejects_unparseable_spec(tmp_path: Path) -> None:
 
 
 def test_stage_api_specs_copies_spec_into_workspace_dir(tmp_path: Path) -> None:
-    targets = _resolved_targets(_write_spec(tmp_path / "src"))
+    original = _write_spec(tmp_path / "src")
+    targets = _resolved_targets(original)
     (source,) = stage_api_specs(targets, "stage-run")
 
     assert source["workspace_subdir"] == "api-specs"
     staged = Path(source["source_path"]) / "openapi.json"
     assert json.loads(staged.read_text(encoding="utf-8"))["info"]["title"] == "Shop API"
+    assert staged.read_bytes() == original.read_bytes()
     assert targets[0]["details"]["workspace_path"] == "/workspace/api-specs/openapi.json"
 
 
@@ -129,6 +148,256 @@ def test_stage_api_specs_disambiguates_same_filename(tmp_path: Path) -> None:
     assert (Path(source["source_path"]) / "openapi-2.json").is_file()
 
 
+def test_stage_api_specs_copies_transitive_json_and_yaml_refs(tmp_path: Path) -> None:
+    source_dir = tmp_path / "src"
+    spec = _write_spec_with_ref(source_dir, "./schemas/User.json#/definitions/User")
+    schemas = source_dir / "schemas"
+    schemas.mkdir()
+    (schemas / "User.json").write_text(
+        json.dumps({"definitions": {"User": {"$ref": "../common.yaml#/User"}}}),
+        encoding="utf-8",
+    )
+    (source_dir / "common.yaml").write_text("User:\n  type: object\n", encoding="utf-8")
+    (source_dir / "unused.yaml").write_text("Secret: unreferenced\n", encoding="utf-8")
+
+    targets = _resolved_targets(spec)
+    (staged_source,) = stage_api_specs(targets, "relative-refs")
+    staged = Path(staged_source["source_path"])
+
+    assert (staged / "openapi.json").is_file()
+    assert (staged / "schemas" / "User.json").is_file()
+    assert (staged / "common.yaml").is_file()
+    assert not (staged / "unused.yaml").exists()
+    assert (staged / "openapi.json").read_bytes() == spec.read_bytes()
+    assert staged_source["workspace_subdir"] == "api-specs"
+    assert targets[0]["details"]["workspace_path"] == "/workspace/api-specs/openapi.json"
+
+
+def test_stage_api_specs_terminates_local_ref_cycles(tmp_path: Path) -> None:
+    source_dir = tmp_path / "src"
+    spec = _write_spec_with_ref(source_dir, "schemas/User.yaml#/User")
+    schemas = source_dir / "schemas"
+    schemas.mkdir()
+    (schemas / "User.yaml").write_text(
+        "User:\n  allOf:\n    - $ref: ../openapi.json#/components/schemas/User\n",
+        encoding="utf-8",
+    )
+
+    (staged_source,) = stage_api_specs(_resolved_targets(spec), "cyclic-refs")
+    staged = Path(staged_source["source_path"])
+
+    assert sorted(path.relative_to(staged) for path in staged.rglob("*") if path.is_file()) == [
+        Path("openapi.json"),
+        Path("schemas/User.yaml"),
+    ]
+
+
+def test_stage_api_specs_keeps_fragment_only_refs_internal(tmp_path: Path) -> None:
+    spec = _write_spec_with_ref(tmp_path / "src", "#/components/schemas/Local")
+
+    (staged_source,) = stage_api_specs(_resolved_targets(spec), "fragment-ref")
+    staged = Path(staged_source["source_path"])
+
+    assert [path.relative_to(staged) for path in staged.rglob("*") if path.is_file()] == [
+        Path("openapi.json")
+    ]
+
+
+def test_stage_api_specs_does_not_fetch_remote_refs(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    spec = _write_spec_with_ref(tmp_path / "src", "https://schemas.example.test/User.json")
+
+    def unexpected_fetch(*_args: object, **_kwargs: object) -> None:
+        raise AssertionError("remote reference was fetched")
+
+    monkeypatch.setattr("strix.interface.utils.requests.get", unexpected_fetch)
+    (staged_source,) = stage_api_specs(_resolved_targets(spec), "remote-ref")
+    staged = Path(staged_source["source_path"])
+
+    assert [path.relative_to(staged) for path in staged.rglob("*") if path.is_file()] == [
+        Path("openapi.json")
+    ]
+
+
+def test_stage_api_specs_keeps_postman_collections_single_file(tmp_path: Path) -> None:
+    collection = {
+        "info": {"name": "Collection", "_postman_id": "fixture"},
+        "item": [],
+        "metadata": {"$ref": "missing.yaml"},
+    }
+    path = tmp_path / "collection.json"
+    path.write_text(json.dumps(collection), encoding="utf-8")
+    targets = _resolved_targets(path)
+
+    (staged_source,) = stage_api_specs(targets, "postman-single-file")
+    staged = Path(staged_source["source_path"])
+
+    assert (staged / "collection.json").read_bytes() == path.read_bytes()
+    assert targets[0]["details"]["workspace_path"] == "/workspace/api-specs/collection.json"
+
+
+def test_stage_api_specs_rejects_missing_local_ref_without_partial_stage(tmp_path: Path) -> None:
+    spec = _write_spec_with_ref(tmp_path / "src", "schemas/Missing.yaml#/User")
+    targets = _resolved_targets(spec)
+
+    with pytest.raises(ValueError, match=r"Missing\.yaml"):
+        stage_api_specs(targets, "missing-ref")
+
+    assert not (tmp_path / "temporary" / "strix_api_specs" / "missing-ref").exists()
+    assert "workspace_path" not in targets[0]["details"]
+
+
+def test_stage_api_specs_rejects_ref_outside_spec_root(tmp_path: Path) -> None:
+    source_dir = tmp_path / "src"
+    spec = _write_spec_with_ref(source_dir, "../outside.yaml#/User")
+    (tmp_path / "outside.yaml").write_text("User: {type: object}\n", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="outside the API spec directory"):
+        stage_api_specs(_resolved_targets(spec), "escaping-ref")
+
+
+def test_stage_api_specs_rejects_encoded_ref_escape(tmp_path: Path) -> None:
+    source_dir = tmp_path / "src"
+    spec = _write_spec_with_ref(source_dir, "..%2Foutside.yaml#/User")
+    (tmp_path / "outside.yaml").write_text("User: {type: object}\n", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="outside the API spec directory"):
+        stage_api_specs(_resolved_targets(spec), "encoded-escape")
+
+
+def test_stage_api_specs_rejects_symlinked_ref(tmp_path: Path) -> None:
+    source_dir = tmp_path / "src"
+    spec = _write_spec_with_ref(source_dir, "schemas/User.yaml#/User")
+    schemas = source_dir / "schemas"
+    schemas.mkdir()
+    real_schema = schemas / "real.yaml"
+    real_schema.write_text("User: {type: object}\n", encoding="utf-8")
+    (schemas / "User.yaml").symlink_to(real_schema)
+
+    with pytest.raises(ValueError, match="symbolic link"):
+        stage_api_specs(_resolved_targets(spec), "symlink-ref")
+
+
+def test_stage_api_specs_isolates_colliding_multifile_specs(tmp_path: Path) -> None:
+    specs: list[Path] = []
+    for directory, marker in (("a", "first"), ("b", "second")):
+        source_dir = tmp_path / directory
+        specs.append(_write_spec_with_ref(source_dir, "schemas/User.json"))
+        schemas = source_dir / "schemas"
+        schemas.mkdir()
+        (schemas / "User.json").write_text(json.dumps({"marker": marker}), encoding="utf-8")
+
+    targets = _resolved_targets(*specs)
+    (staged_source,) = stage_api_specs(targets, "isolated-refs")
+    staged = Path(staged_source["source_path"])
+
+    assert [target["details"]["workspace_path"] for target in targets] == [
+        "/workspace/api-specs/openapi/openapi.json",
+        "/workspace/api-specs/openapi-2/openapi.json",
+    ]
+    assert json.loads((staged / "openapi" / "schemas" / "User.json").read_text())["marker"] == (
+        "first"
+    )
+    assert (
+        json.loads((staged / "openapi-2" / "schemas" / "User.json").read_text())["marker"]
+        == "second"
+    )
+
+
+def test_stage_api_specs_rejects_unsupported_local_ref_type(tmp_path: Path) -> None:
+    source_dir = tmp_path / "src"
+    spec = _write_spec_with_ref(source_dir, "schemas/User.txt")
+    schemas = source_dir / "schemas"
+    schemas.mkdir()
+    (schemas / "User.txt").write_text("not a JSON or YAML document", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="JSON or YAML"):
+        stage_api_specs(_resolved_targets(spec), "unsupported-ref")
+
+
+def test_stage_api_specs_validates_all_targets_before_publishing(tmp_path: Path) -> None:
+    good = _write_spec(tmp_path / "good")
+    bad = _write_spec_with_ref(tmp_path / "bad", "missing.yaml")
+    targets = _resolved_targets(good, bad)
+    targets[0]["details"]["workspace_path"] = "/workspace/api-specs/stale.json"
+
+    with pytest.raises(ValueError, match=r"missing\.yaml"):
+        stage_api_specs(targets, "atomic-stage")
+
+    assert not (tmp_path / "temporary" / "strix_api_specs" / "atomic-stage").exists()
+    assert all("workspace_path" not in target["details"] for target in targets)
+
+
+def test_stage_api_specs_enforces_file_limit(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    source_dir = tmp_path / "src"
+    spec = _write_spec_with_ref(source_dir, "one.yaml")
+    (source_dir / "one.yaml").write_text("$ref: two.yaml\n", encoding="utf-8")
+    (source_dir / "two.yaml").write_text("type: object\n", encoding="utf-8")
+    monkeypatch.setattr("strix.interface.utils._API_SPEC_MAX_FILES", 2)
+
+    with pytest.raises(ValueError, match="2-file staging limit"):
+        stage_api_specs(_resolved_targets(spec), "file-limit")
+
+
+def test_stage_api_specs_enforces_limits_across_targets(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    specs = [_write_spec(tmp_path / str(index)) for index in range(3)]
+    monkeypatch.setattr("strix.interface.utils._API_SPEC_MAX_FILES", 2)
+
+    with pytest.raises(ValueError, match="2-file staging limit"):
+        stage_api_specs(_resolved_targets(*specs), "aggregate-file-limit")
+
+
+def test_stage_api_specs_enforces_depth_limit(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    source_dir = tmp_path / "src"
+    spec = _write_spec_with_ref(source_dir, "one.yaml")
+    (source_dir / "one.yaml").write_text("$ref: two.yaml\n", encoding="utf-8")
+    (source_dir / "two.yaml").write_text("type: object\n", encoding="utf-8")
+    monkeypatch.setattr("strix.interface.utils._API_SPEC_MAX_REF_DEPTH", 1)
+
+    with pytest.raises(ValueError, match="1-level limit"):
+        stage_api_specs(_resolved_targets(spec), "depth-limit")
+
+
+def test_stage_api_specs_enforces_total_byte_limit(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    spec = _write_spec(tmp_path / "src")
+    monkeypatch.setattr("strix.interface.utils._API_SPEC_MAX_TOTAL_BYTES", 10)
+
+    with pytest.raises(ValueError, match="10-byte staging limit"):
+        stage_api_specs(_resolved_targets(spec), "byte-limit")
+
+
+def test_stage_api_specs_rejects_unsafe_run_name(tmp_path: Path) -> None:
+    targets = _resolved_targets(_write_spec(tmp_path / "src"))
+
+    with pytest.raises(ValueError, match="Invalid run name"):
+        stage_api_specs(targets, "../outside")
+
+    assert not (tmp_path / "outside").exists()
+
+
+def test_stage_api_specs_rejects_symlinked_staging_directory(tmp_path: Path) -> None:
+    source = _write_spec(tmp_path / "src")
+    temporary_root = tmp_path / "temporary"
+    temporary_root.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (temporary_root / "strix_api_specs").symlink_to(outside, target_is_directory=True)
+
+    with pytest.raises(ValueError, match="staging directory cannot be a symbolic link"):
+        stage_api_specs(_resolved_targets(source), "symlinked-stage")
+
+    assert list(outside.iterdir()) == []
+
+
 def test_stage_api_specs_without_specs_returns_nothing() -> None:
     assert stage_api_specs([{"type": "web_application", "details": {}}], "run") == []