diff --git a/backend/cortex_backend/execution/__init__.py b/backend/cortex_backend/execution/__init__.py
index 27de659..dc53e53 100644
--- a/backend/cortex_backend/execution/__init__.py
+++ b/backend/cortex_backend/execution/__init__.py
@@ -21,6 +21,16 @@
     encode_frame,
     encode_message,
 )
+from .artifact_boundary import (
+    ArtifactBoundary,
+    ArtifactBoundaryError,
+    ArtifactSourceGrant,
+    MAX_ARTIFACT_PATH_CHARS,
+    MAX_OUTPUT_COUNT,
+    OutputClaim,
+    PublishedArtifact,
+    sniff_artifact_mime,
+)
 from .fake import FakeExecutionPlan, FakeExecutionProvider
 from .bundle_installer import (
     BundleInstallError,
@@ -80,6 +90,9 @@
 
 __all__ = [
     "ArtifactLimitError",
+    "ArtifactBoundary",
+    "ArtifactBoundaryError",
+    "ArtifactSourceGrant",
     "ApprovalPolicyError",
     "ApprovalTransitionError",
     "BrokerAclPolicy",
@@ -108,6 +121,8 @@
     "ManifestEntry",
     "ManifestState",
     "ManifestVerificationError",
+    "MAX_ARTIFACT_PATH_CHARS",
+    "MAX_OUTPUT_COUNT",
     "MAX_NATIVE_PIPE_NAME_LENGTH",
     "NativeBrokerClient",
     "NativeBrokerClientConfig",
@@ -116,6 +131,8 @@
     "NativeBrokerServer",
     "NativeBrokerServerConfig",
     "PrimitiveEvaluationError",
+    "OutputClaim",
+    "PublishedArtifact",
     "RecipeValidationError",
     "RuntimeHealth",
     "RollbackAuthorizer",
@@ -129,6 +146,7 @@
     "parse_calculator",
     "parse_check",
     "parse_image_transform",
+    "sniff_artifact_mime",
     "verify_bundle_files",
     "verify_manifest_signature",
     "parse_keyring_update",
diff --git a/backend/cortex_backend/execution/artifact_boundary.py b/backend/cortex_backend/execution/artifact_boundary.py
new file mode 100644
index 0000000..26ea091
--- /dev/null
+++ b/backend/cortex_backend/execution/artifact_boundary.py
@@ -0,0 +1,487 @@
+"""Trusted artifact copy-in, validation, quarantine, and publication.
+
+This module is the only boundary that may move a user-selected file into the
+execution artifact store or publish a guest output.  It accepts explicit source
+grants, reads files with identity/size/hash checks, MIME-sniffs bytes instead of
+trusting extensions or model claims, and publishes only after every output has
+passed validation.  It never executes, decodes, or overwrites a source file.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from hashlib import sha256
+import json
+import os
+from pathlib import Path
+import re
+import secrets
+import stat
+from typing import Any
+
+from .models import ExecutionArtifact
+from .repository import ExecutionRepository, ExecutionRepositoryError
+
+
+MAX_ARTIFACT_PATH_CHARS = 4096
+MAX_OUTPUT_COUNT = 16
+_SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$")
+_SAFE_MIME = re.compile(r"^[a-z0-9][a-z0-9.+-]{0,31}/[a-z0-9][a-z0-9.+-]{0,63}$")
+_SAFE_RELATIVE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,255}$")
+_CONTROL_BYTES = frozenset(range(0, 9)) | frozenset(range(11, 13)) | frozenset(range(14, 32))
+_ARCHIVE_MAGICS = (
+    b"PK\x03\x04",
+    b"PK\x05\x06",
+    b"PK\x07\x08",
+    b"Rar!\x1a\x07",
+    b"7z\xbc\xaf\x27\x1c",
+    b"\x1f\x8b",
+)
+
+
+class ArtifactBoundaryError(ValueError):
+    """Stable artifact boundary failure without paths or raw OS details."""
+
+    def __init__(self, code: str) -> None:
+        if re.fullmatch(r"[a-z][a-z0-9_]{0,63}", code) is None:
+            raise ValueError("invalid artifact boundary code")
+        self.code = code
+        super().__init__("The artifact could not be transferred safely.")
+
+
+class _NonFiniteJSON(ValueError):
+    """Internal marker for JSON extensions that represent non-finite numbers."""
+
+
+@dataclass(frozen=True, slots=True)
+class ArtifactSourceGrant:
+    """Authenticated, one-turn authorization for one explicit source path."""
+
+    owner: str
+    job_id: str
+    source_turn_id: str
+    source_path: Path
+
+    def __post_init__(self) -> None:
+        for value in (self.owner, self.job_id, self.source_turn_id):
+            if not isinstance(value, str) or _SAFE_ID.fullmatch(value) is None:
+                raise ValueError("artifact source grant identifier is invalid")
+        object.__setattr__(self, "source_path", Path(self.source_path))
+
+
+@dataclass(frozen=True, slots=True)
+class OutputClaim:
+    """Provider declaration for one relative output in its private staging root."""
+
+    relative_path: str
+    mime_type: str | None = None
+
+    def __post_init__(self) -> None:
+        parts = self.relative_path.split("/") if isinstance(self.relative_path, str) else []
+        if (
+            not isinstance(self.relative_path, str)
+            or len(self.relative_path) == 0
+            or len(self.relative_path) > 256
+            or _SAFE_RELATIVE.fullmatch(self.relative_path) is None
+            or "\\" in self.relative_path
+            or ":" in self.relative_path
+            or any(part in {"", ".", ".."} for part in parts)
+        ):
+            raise ValueError("artifact output path is invalid")
+        if self.mime_type is not None and _SAFE_MIME.fullmatch(self.mime_type) is None:
+            raise ValueError("artifact output MIME type is invalid")
+
+
+@dataclass(frozen=True, slots=True)
+class PublishedArtifact:
+    artifact: ExecutionArtifact
+    relative_path: str
+    mime_type: str
+
+
+def _is_reparse_point(path: Path) -> bool:
+    if path.is_symlink():
+        return True
+    is_junction = getattr(path, "is_junction", None)
+    return bool(is_junction is not None and is_junction())
+
+
+def _has_reparse_parent(path: Path) -> bool:
+    try:
+        return any(parent.exists() and _is_reparse_point(parent) for parent in path.parents)
+    except OSError:
+        return True
+
+
+def _validate_absolute_path(path: Path) -> None:
+    text = str(path)
+    if (
+        not path.is_absolute()
+        or len(text) > MAX_ARTIFACT_PATH_CHARS
+        or "\x00" in text
+        or any(":" in part for part in path.parts[1:])
+    ):
+        raise ArtifactBoundaryError("artifact_path_invalid")
+
+
+def _validate_components(path: Path) -> None:
+    """Reject links/reparse points in every existing parent component."""
+
+    try:
+        for component in reversed(path.parents):
+            if component.exists() and _is_reparse_point(component):
+                raise ArtifactBoundaryError("artifact_reparse_point")
+        if _is_reparse_point(path):
+            raise ArtifactBoundaryError("artifact_reparse_point")
+    except ArtifactBoundaryError:
+        raise
+    except OSError:
+        raise ArtifactBoundaryError("artifact_path_unavailable") from None
+
+
+def _is_sparse(info: os.stat_result) -> bool:
+    sparse_flag = getattr(stat, "FILE_ATTRIBUTE_SPARSE_FILE", 0x200)
+    if int(getattr(info, "st_file_attributes", 0)) & sparse_flag:
+        return True
+    blocks = int(getattr(info, "st_blocks", 0))
+    return blocks > 0 and int(info.st_size) > blocks * 512
+
+
+def _file_identity(path: Path) -> tuple[int, int, int, int, int]:
+    try:
+        info = path.lstat()
+    except OSError:
+        raise ArtifactBoundaryError("artifact_source_unavailable") from None
+    if _is_reparse_point(path):
+        raise ArtifactBoundaryError("artifact_reparse_point")
+    if not stat.S_ISREG(info.st_mode):
+        raise ArtifactBoundaryError("artifact_not_regular_file")
+    if int(getattr(info, "st_nlink", 1)) != 1:
+        raise ArtifactBoundaryError("artifact_hardlink_rejected")
+    if _is_sparse(info):
+        raise ArtifactBoundaryError("artifact_sparse_file")
+    return (
+        int(getattr(info, "st_dev", 0)),
+        int(getattr(info, "st_ino", 0)),
+        int(info.st_size),
+        int(getattr(info, "st_mtime_ns", 0)),
+        int(getattr(info, "st_ctime_ns", 0)),
+    )
+
+
+def _read_stable(path: Path, maximum: int) -> bytes:
+    before = _file_identity(path)
+    if before[2] > maximum:
+        raise ArtifactBoundaryError("artifact_too_large")
+    try:
+        with path.open("rb") as stream:
+            content = stream.read(maximum + 1)
+    except OSError:
+        raise ArtifactBoundaryError("artifact_source_unavailable") from None
+    after = _file_identity(path)
+    if before != after or len(content) > maximum:
+        raise ArtifactBoundaryError("artifact_source_changed")
+    return content
+
+
+def _is_printable_text(content: bytes) -> bool:
+    try:
+        text = content.decode("utf-8")
+    except UnicodeDecodeError:
+        return False
+    return not any(ord(character) in _CONTROL_BYTES for character in text)
+
+
+def _reject_json_constant(_value: str) -> Any:
+    """Keep non-standard NaN/Infinity JSON extensions outside the boundary."""
+
+    raise _NonFiniteJSON("non-finite JSON number")
+
+
+def sniff_artifact_mime(content: bytes) -> str:
+    """Return a conservative MIME type or reject active/archive content."""
+
+    if not isinstance(content, bytes):
+        raise ArtifactBoundaryError("invalid_artifact")
+    prefix = content[:4096]
+    lower = prefix.lstrip(b"\xef\xbb\xbf \t\r\n").lower()
+    if content.startswith(b"\x89PNG\r\n\x1a\n"):
+        return "image/png"
+    if content.startswith(b"\xff\xd8\xff"):
+        return "image/jpeg"
+    if len(content) >= 12 and content[:4] == b"RIFF" and content[8:12] == b"WEBP":
+        return "image/webp"
+    if content.startswith(b"MZ") or content.startswith(b"\x7fELF"):
+        raise ArtifactBoundaryError("invalid_artifact")
+    if content.startswith(b"\x4c\x00\x00\x00") or content.startswith(b"\xd0\xcf\x11\xe0"):
+        raise ArtifactBoundaryError("invalid_artifact")
+    if content.startswith(b"#!") or lower.startswith(b"[internetshortcut]"):
+        raise ArtifactBoundaryError("invalid_artifact")
+    if any(content.startswith(magic) for magic in _ARCHIVE_MAGICS) or content[257:262] == b"ustar":
+        raise ArtifactBoundaryError("invalid_artifact")
+    if (
+        lower.startswith((b"<!doctype html", b"<html", b"<script", b"<svg", b"javascript:"))
+        or b"<svg" in lower[:1024]
+    ):
+        raise ArtifactBoundaryError("invalid_artifact")
+    if lower.startswith((b"@echo off", b"powershell ", b"cmd.exe ")):
+        raise ArtifactBoundaryError("invalid_artifact")
+    if _is_printable_text(content):
+        try:
+            parsed = json.loads(content.decode("utf-8"), parse_constant=_reject_json_constant)
+        except _NonFiniteJSON:
+            raise ArtifactBoundaryError("invalid_artifact") from None
+        except (UnicodeDecodeError, json.JSONDecodeError):
+            return "text/plain"
+        if isinstance(parsed, (dict, list, str, int, float, bool)) or parsed is None:
+            return "application/json"
+    return "application/octet-stream"
+
+
+class ArtifactBoundary:
+    """Owner-scoped copy-in and all-or-nothing output publication."""
+
+    def __init__(
+        self,
+        repository: ExecutionRepository,
+        *,
+        max_input_bytes: int | None = None,
+        max_output_count: int = MAX_OUTPUT_COUNT,
+        max_total_output_bytes: int | None = None,
+    ) -> None:
+        if not isinstance(repository, ExecutionRepository):
+            raise TypeError("repository must be an ExecutionRepository")
+        maximum = repository.max_artifact_bytes if max_input_bytes is None else max_input_bytes
+        if not 1 <= maximum <= repository.max_artifact_bytes:
+            raise ValueError("max_input_bytes is invalid")
+        if not 1 <= max_output_count <= MAX_OUTPUT_COUNT:
+            raise ValueError("max_output_count is invalid")
+        total = repository.max_artifact_bytes * max_output_count
+        if max_total_output_bytes is not None:
+            if not 1 <= max_total_output_bytes <= total:
+                raise ValueError("max_total_output_bytes is invalid")
+            total = max_total_output_bytes
+        self.repository = repository
+        self.max_input_bytes = maximum
+        self.max_output_count = max_output_count
+        self.max_total_output_bytes = total
+        self.quarantine_root = repository.artifact_root / ".artifact_quarantine"
+        try:
+            if _is_reparse_point(repository.artifact_root) or _has_reparse_parent(repository.artifact_root):
+                raise ArtifactBoundaryError("artifact_root_unavailable")
+            self.quarantine_root.mkdir(parents=True, exist_ok=True)
+            if _is_reparse_point(self.quarantine_root):
+                raise ArtifactBoundaryError("artifact_root_unavailable")
+        except ArtifactBoundaryError:
+            raise
+        except OSError:
+            raise ArtifactBoundaryError("artifact_root_unavailable") from None
+
+    def _authorize_job(self, job_id: str, owner: str) -> None:
+        if self.repository.get_job(job_id, owner=owner) is None:
+            raise ArtifactBoundaryError("artifact_owner_mismatch")
+
+    @staticmethod
+    def _generated_name(digest: str) -> str:
+        return f"artifact-{digest[:32]}"
+
+    def copy_in(
+        self,
+        grant: ArtifactSourceGrant,
+        *,
+        retention_seconds: int = 86_400,
+    ) -> ExecutionArtifact:
+        """Copy one explicitly granted source into the owner-scoped artifact store."""
+
+        self._validate_retention(retention_seconds)
+        self._authorize_job(grant.job_id, grant.owner)
+        source = grant.source_path
+        _validate_absolute_path(source)
+        _validate_components(source)
+        content = _read_stable(source, self.max_input_bytes)
+        mime_type = sniff_artifact_mime(content)
+        digest = sha256(content).hexdigest()
+        try:
+            return self.repository.publish_artifact(
+                grant.job_id,
+                name=self._generated_name(digest),
+                content=content,
+                mime_type=mime_type,
+                retention_seconds=retention_seconds,
+            )
+        except ExecutionRepositoryError:
+            raise ArtifactBoundaryError("artifact_publish_failed") from None
+
+    @staticmethod
+    def _validate_retention(retention_seconds: int) -> None:
+        if (
+            isinstance(retention_seconds, bool)
+            or not isinstance(retention_seconds, int)
+            or retention_seconds <= 0
+        ):
+            raise ArtifactBoundaryError("artifact_retention_invalid")
+
+    def _output_root(self, root: str | os.PathLike[str]) -> Path:
+        value = Path(root)
+        _validate_absolute_path(value)
+        _validate_components(value)
+        try:
+            resolved = value.resolve(strict=True)
+        except (OSError, RuntimeError):
+            raise ArtifactBoundaryError("artifact_output_unavailable") from None
+        if not resolved.is_dir() or _is_reparse_point(resolved):
+            raise ArtifactBoundaryError("artifact_output_unavailable")
+        return resolved
+
+    @staticmethod
+    def _claim_path(root: Path, claim: OutputClaim) -> Path:
+        candidate = root / claim.relative_path
+        cursor = root
+        for component in Path(claim.relative_path).parts:
+            cursor = cursor / component
+            if _is_reparse_point(cursor):
+                raise ArtifactBoundaryError("artifact_reparse_point")
+        try:
+            resolved = candidate.resolve(strict=True)
+        except (OSError, RuntimeError):
+            raise ArtifactBoundaryError("artifact_output_unavailable") from None
+        if not resolved.is_relative_to(root) or _is_reparse_point(candidate):
+            raise ArtifactBoundaryError("artifact_path_invalid")
+        return candidate
+
+    def _quarantine(self, root: Path, path: Path) -> None:
+        if _is_reparse_point(path):
+            raise ArtifactBoundaryError("artifact_cleanup_pending")
+        quarantine = root / ".quarantine"
+        try:
+            if _is_reparse_point(quarantine):
+                raise ArtifactBoundaryError("artifact_cleanup_pending")
+            quarantine.mkdir(parents=True, exist_ok=True)
+            if _is_reparse_point(quarantine):
+                raise ArtifactBoundaryError("artifact_cleanup_pending")
+            target = quarantine / f"artifact-{secrets.token_hex(16)}"
+            os.replace(path, target)
+        except ArtifactBoundaryError:
+            raise
+        except OSError:
+            raise ArtifactBoundaryError("artifact_cleanup_pending") from None
+
+    def _output_files(self, root: Path) -> list[tuple[str, Path]]:
+        found: list[tuple[str, Path]] = []
+        try:
+            for path in root.rglob("*"):
+                relative = path.relative_to(root).as_posix()
+                if _is_reparse_point(path):
+                    raise ArtifactBoundaryError("artifact_reparse_point")
+                if relative == ".quarantine" or relative.startswith(".quarantine/"):
+                    continue
+                if path.is_dir():
+                    continue
+                if not path.is_file():
+                    raise ArtifactBoundaryError("artifact_not_regular_file")
+                found.append((relative, path))
+        except ArtifactBoundaryError:
+            raise
+        except OSError:
+            raise ArtifactBoundaryError("artifact_output_unavailable") from None
+        return found
+
+    def collect_outputs(
+        self,
+        job_id: str,
+        owner: str,
+        output_root: str | os.PathLike[str],
+        claims: Sequence[OutputClaim],
+        *,
+        retention_seconds: int = 86_400,
+    ) -> tuple[PublishedArtifact, ...]:
+        """Validate every declared output before publishing any artifact."""
+
+        self._validate_retention(retention_seconds)
+        self._authorize_job(job_id, owner)
+        if not isinstance(claims, Sequence) or not claims or len(claims) > self.max_output_count:
+            raise ArtifactBoundaryError("artifact_output_count_invalid")
+        if any(not isinstance(claim, OutputClaim) for claim in claims):
+            raise ArtifactBoundaryError("artifact_output_claim_invalid")
+        claim_map: dict[str, OutputClaim] = {}
+        for claim in claims:
+            if claim.relative_path in claim_map:
+                raise ArtifactBoundaryError("artifact_output_claim_invalid")
+            claim_map[claim.relative_path] = claim
+        root = self._output_root(output_root)
+        files = self._output_files(root)
+        if {relative for relative, _ in files} != set(claim_map):
+            for relative, path in files:
+                if relative not in claim_map:
+                    self._quarantine(root, path)
+            raise ArtifactBoundaryError("artifact_unclaimed_output")
+        for claim in claims:
+            self._claim_path(root, claim)
+        prepared: list[tuple[OutputClaim, Path, bytes, str, str]] = []
+        total = 0
+        for relative, path in files:
+            claim = claim_map[relative]
+            try:
+                content = _read_stable(path, self.repository.max_artifact_bytes)
+                mime_type = sniff_artifact_mime(content)
+                if claim.mime_type is not None and claim.mime_type != mime_type:
+                    raise ArtifactBoundaryError("artifact_mime_mismatch")
+            except ArtifactBoundaryError:
+                self._quarantine(root, path)
+                raise
+            total += len(content)
+            if total > self.max_total_output_bytes:
+                self._quarantine(root, path)
+                raise ArtifactBoundaryError("artifact_output_limit")
+            prepared.append((claim, path, content, mime_type, sha256(content).hexdigest()))
+        published: list[PublishedArtifact] = []
+        try:
+            for claim, _path, content, mime_type, digest in prepared:
+                artifact = self.repository.publish_artifact(
+                    job_id,
+                    name=self._generated_name(digest),
+                    content=content,
+                    mime_type=mime_type,
+                    retention_seconds=retention_seconds,
+                )
+                published.append(
+                    PublishedArtifact(
+                        artifact=artifact,
+                        relative_path=claim.relative_path,
+                        mime_type=mime_type,
+                    )
+                )
+        except ExecutionRepositoryError:
+            try:
+                for item in published:
+                    self.repository.delete_artifact(item.artifact.artifact_id)
+                for _claim, path, _content, _mime, _digest in prepared:
+                    if path.exists() or path.is_symlink():
+                        self._quarantine(root, path)
+            except (ExecutionRepositoryError, OSError):
+                raise ArtifactBoundaryError("artifact_cleanup_pending") from None
+            raise ArtifactBoundaryError("artifact_publish_failed") from None
+        try:
+            for _claim, path, _content, _mime, _digest in prepared:
+                path.unlink()
+        except OSError:
+            try:
+                for item in published:
+                    self.repository.delete_artifact(item.artifact.artifact_id)
+            except ExecutionRepositoryError:
+                raise ArtifactBoundaryError("artifact_cleanup_pending") from None
+            raise ArtifactBoundaryError("artifact_cleanup_pending") from None
+        return tuple(published)
+
+
+__all__ = [
+    "ArtifactBoundary",
+    "ArtifactBoundaryError",
+    "ArtifactSourceGrant",
+    "MAX_ARTIFACT_PATH_CHARS",
+    "MAX_OUTPUT_COUNT",
+    "OutputClaim",
+    "PublishedArtifact",
+    "sniff_artifact_mime",
+]
diff --git a/backend/cortex_backend/execution/repository.py b/backend/cortex_backend/execution/repository.py
index 37e2a6e..8596dd6 100644
--- a/backend/cortex_backend/execution/repository.py
+++ b/backend/cortex_backend/execution/repository.py
@@ -6,10 +6,12 @@
 from datetime import datetime, timedelta, timezone
 import hashlib
 import json
+import os
 from pathlib import Path
 import re
 import secrets
 import sqlite3
+import stat
 from collections.abc import Iterator, Mapping
 from threading import RLock
 from typing import Any, Literal
@@ -31,9 +33,28 @@
 _SAFE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,199}$")
 _SAFE_PROFILE = re.compile(r"^[a-z][a-z0-9._-]{0,99}$")
 _SAFE_INSTALLATION_PRINCIPAL = re.compile(r"^[0-9a-f]{64}$")
+_SAFE_MIME = re.compile(r"^[a-z0-9][a-z0-9.+-]{0,31}/[a-z0-9][a-z0-9.+-]{0,63}$")
 _SCHEMA_LOCK = RLock()
 
 
+def _is_reparse_point(path: Path) -> bool:
+    """Treat symbolic links and Windows junctions as untrusted path hops."""
+
+    if path.is_symlink():
+        return True
+    is_junction = getattr(path, "is_junction", None)
+    return bool(is_junction is not None and is_junction())
+
+
+def _has_reparse_parent(path: Path) -> bool:
+    """Reject a configured root that would traverse a link in an ancestor."""
+
+    try:
+        return any(parent.exists() and _is_reparse_point(parent) for parent in path.parents)
+    except OSError:
+        return True
+
+
 class ExecutionRepositoryError(RuntimeError):
     """Safe repository boundary error."""
 
@@ -104,6 +125,8 @@ def connect(self) -> Iterator[sqlite3.Connection]:
     def _ensure_schema(self) -> None:
         self.db_path.parent.mkdir(parents=True, exist_ok=True)
         self.artifact_root.mkdir(parents=True, exist_ok=True)
+        if _is_reparse_point(self.artifact_root) or _has_reparse_parent(self.artifact_root):
+            raise ExecutionRepositoryError("Artifact root is unavailable.")
         with _SCHEMA_LOCK:
             self._ensure_schema_locked()
 
@@ -766,6 +789,10 @@ def publish_artifact(
     ) -> ExecutionArtifact:
         if not _SAFE_NAME.fullmatch(name):
             raise ExecutionRepositoryError("Artifact name is invalid.")
+        if not isinstance(content, bytes):
+            raise ExecutionRepositoryError("Artifact content is invalid.")
+        if not isinstance(mime_type, str) or _SAFE_MIME.fullmatch(mime_type) is None:
+            raise ExecutionRepositoryError("Artifact MIME type is invalid.")
         if len(content) > self.max_artifact_bytes:
             raise ArtifactLimitError("Artifact exceeds the configured size limit.")
         if retention_seconds <= 0:
@@ -774,18 +801,35 @@ def publish_artifact(
             raise ExecutionRepositoryError("Execution job does not exist.")
         artifact_id = uuid4().hex
         job_root = self.artifact_root / job_id
+        if job_root.exists() and _is_reparse_point(job_root):
+            raise ExecutionRepositoryError("Artifact root is unavailable.")
         job_root.mkdir(parents=True, exist_ok=True)
-        target = (job_root / f"{artifact_id}-{name}").resolve()
         root = self.artifact_root.resolve()
-        if not target.is_relative_to(root):
+        resolved_job_root = job_root.resolve(strict=True)
+        if not resolved_job_root.is_relative_to(root) or _is_reparse_point(resolved_job_root):
             raise ExecutionRepositoryError("Artifact path escaped the artifact root.")
+        target = job_root / f"{artifact_id}-{name}"
         temporary = target.with_name(f".tmp-{artifact_id}")
-        temporary.write_bytes(content)
         digest = hashlib.sha256(content).hexdigest()
         now = datetime.now(timezone.utc)
         expires = now + timedelta(seconds=retention_seconds)
         try:
-            temporary.replace(target)
+            with temporary.open("xb") as stream:
+                stream.write(content)
+                stream.flush()
+                os.fsync(stream.fileno())
+            os.replace(temporary, target)
+            try:
+                directory = os.open(str(job_root), os.O_RDONLY)
+            except OSError:
+                directory = None
+            if directory is not None:
+                try:
+                    os.fsync(directory)
+                except OSError:
+                    pass
+                finally:
+                    os.close(directory)
             with self.connect() as connection:
                 connection.execute(
                     """
@@ -821,6 +865,33 @@ def publish_artifact(
             expires_at=expires.isoformat(),
         )
 
+    def delete_artifact(self, artifact_id: str) -> None:
+        """Remove one unpublished/rolled-back artifact record and file safely."""
+
+        with self.connect() as connection:
+            row = connection.execute(
+                "SELECT path FROM execution_artifacts WHERE artifact_id = ?",
+                (artifact_id,),
+            ).fetchone()
+            if row is None:
+                return
+            path = Path(row["path"])
+            root = self.artifact_root.resolve()
+            try:
+                resolved = path.resolve(strict=False)
+            except (OSError, RuntimeError):
+                raise ExecutionRepositoryError("Artifact path is unavailable.") from None
+            if not resolved.is_relative_to(root) or _is_reparse_point(path):
+                raise ExecutionRepositoryError("Artifact path is unavailable.")
+            try:
+                path.unlink(missing_ok=True)
+            except OSError as exc:
+                raise ExecutionRepositoryError("Artifact cleanup failed.") from exc
+            connection.execute(
+                "DELETE FROM execution_artifacts WHERE artifact_id = ?",
+                (artifact_id,),
+            )
+
     def read_artifact(self, artifact_id: str) -> bytes:
         with self.connect() as connection:
             row = connection.execute(
@@ -831,10 +902,25 @@ def read_artifact(self, artifact_id: str) -> bytes:
             raise ExecutionRepositoryError("Artifact does not exist.")
         if datetime.fromisoformat(row["expires_at"]) <= datetime.now(timezone.utc):
             raise ExecutionRepositoryError("Artifact retention has expired.")
-        path = Path(row["path"]).resolve()
-        if not path.is_relative_to(self.artifact_root.resolve()) or not path.is_file():
+        original_path = Path(row["path"])
+        if _is_reparse_point(original_path):
+            raise ExecutionRepositoryError("Artifact path is unavailable.")
+        try:
+            path = original_path.resolve(strict=True)
+            info = path.lstat()
+        except (OSError, RuntimeError):
+            raise ExecutionRepositoryError("Artifact path is unavailable.") from None
+        if (
+            not path.is_relative_to(self.artifact_root.resolve())
+            or _is_reparse_point(path)
+            or not stat.S_ISREG(info.st_mode)
+            or getattr(info, "st_nlink", 1) != 1
+        ):
             raise ExecutionRepositoryError("Artifact path is unavailable.")
-        content = path.read_bytes()
+        try:
+            content = path.read_bytes()
+        except OSError:
+            raise ExecutionRepositoryError("Artifact path is unavailable.") from None
         if hashlib.sha256(content).hexdigest() != row["sha256"]:
             raise ExecutionRepositoryError("Artifact integrity check failed.")
         return content
@@ -848,7 +934,19 @@ def purge_expired(self, *, now: str | None = None) -> int:
                 (cutoff,),
             ).fetchall()
             for row in artifacts:
-                Path(row["path"]).unlink(missing_ok=True)
+                path = Path(row["path"])
+                if _is_reparse_point(path):
+                    raise ExecutionRepositoryError("Artifact path is unavailable.")
+                try:
+                    resolved = path.resolve(strict=False)
+                except (OSError, RuntimeError):
+                    raise ExecutionRepositoryError("Artifact path is unavailable.") from None
+                if not resolved.is_relative_to(self.artifact_root.resolve()):
+                    raise ExecutionRepositoryError("Artifact path is unavailable.")
+                try:
+                    path.unlink(missing_ok=True)
+                except OSError:
+                    raise ExecutionRepositoryError("Artifact cleanup failed.") from None
                 connection.execute(
                     "DELETE FROM execution_artifacts WHERE artifact_id = ?",
                     (row["artifact_id"],),
diff --git a/docs/adr/0001-capability-tiered-agentic-execution-harness.md b/docs/adr/0001-capability-tiered-agentic-execution-harness.md
index c32855f..c6c5a58 100644
--- a/docs/adr/0001-capability-tiered-agentic-execution-harness.md
+++ b/docs/adr/0001-capability-tiered-agentic-execution-harness.md
@@ -703,7 +703,8 @@ application-data exhaustion.
 An image request such as “make this grayscale and increase contrast” follows this
 path:
 
-1. The attached image is snapshotted and assigned an opaque artifact ID.
+1. The attached image is snapshotted through the trusted artifact boundary and assigned
+   an opaque artifact ID; the source remains untouched.
 2. The model or deterministic request adapter emits a typed `image_transform` plan.
 3. Policy verifies that the user requested transformation of that exact attached
    artifact; no general filesystem grant is inferred.
@@ -714,8 +715,10 @@ path:
    arbitrary filter/plugin name or file path.
 6. The result is re-encoded as a new image. Metadata is stripped by default; preserving
    selected color/orientation metadata must be explicit and sanitized.
-7. The artifact validator decodes the output again, verifies dimensions/type/size,
-   creates a thumbnail in trusted code, and atomically publishes both.
+7. The future artifact validator decodes the output again, verifies dimensions/type/size,
+   creates a thumbnail in trusted code, and atomically publishes both. The current
+   boundary validates bytes, type, size, claims, and hashes but deliberately does not
+   decode images.
 8. The UI presents the new artifact, operation summary, dimensions, size, and download
    action. The source is untouched.
 
@@ -936,12 +939,15 @@ pass without executing code.
 
 ### Phase 2 — signed image recipes and calculator/check primitives
 
-- Release only fixed-function typed operations in the OS sandbox.
-- Validate the complete copy-in/validate/publish path and collect opt-in aggregate
-  reliability metrics, never content.
+- Keep the typed recipe, signed bundle, broker transport, and trusted artifact boundary
+  provider-independent until the sandbox qualification gate passes.
+- The owner-bound copy-in, exact-claim output validation, quarantine, hashing, and
+  atomic publication boundary is implemented; collect opt-in aggregate reliability
+  metrics, never content, after the provider is qualified.
 
-**Exit gate:** parser fuzzing and artifact security review pass; no source overwrite is
-possible.
+**Implementation gate:** parser and artifact-boundary adversarial suites pass; no source
+overwrite is possible. **Release gate:** parser fuzzing, artifact security review, and
+fixed-function provider qualification must pass before any provider is enabled.
 
 ### Phase 3 — `scratch.auto.v1` arbitrary WebAssembly code
 
diff --git a/docs/adr/0001-phase2-artifact-boundary.md b/docs/adr/0001-phase2-artifact-boundary.md
new file mode 100644
index 0000000..e6d9be8
--- /dev/null
+++ b/docs/adr/0001-phase2-artifact-boundary.md
@@ -0,0 +1,104 @@
+# ADR-0001 Phase 2 trusted artifact boundary
+
+- **Status:** Implemented and verified
+- **Phase:** 2 - trusted user-artifact copy-in, output validation, and publication
+- **Parent:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md)
+- **Related:** [Phase 2 typed recipe contract](0001-phase2-recipe-contract.md), [Phase 2 evidence log](0001-phase2-evidence.md)
+- **Scope:** Boundary code only. No image codec, provider, sandbox, or automatic execution is enabled by this ADR.
+
+## Context
+
+The signed bundle installer protects trusted runtime bytes, but it must not be used
+as the user-artifact transfer path. User-selected inputs and provider outputs have a
+different trust model: paths are attacker-controlled, files can change during a read,
+and a provider can return files that were not declared by the plan. A copy or
+publication helper that trusts a filename, caller MIME type, or a resolved path can
+create path traversal, link/reparse-point, time-of-check/time-of-use, active-content,
+partial-publication, or source-overwrite failures.
+
+Windows reparse points deliberately change ordinary file-operation behavior, so this
+boundary treats symbolic links and junctions as untrusted path hops. The implementation
+also uses exclusive temporary files and an atomic replacement for the repository commit;
+the operating-system behavior and limitations are documented by Microsoft in
+[Reparse Points and File Operations](https://learn.microsoft.com/en-us/windows/win32/fileio/reparse-points-and-file-operations),
+[Reparse Point Operations](https://learn.microsoft.com/en-us/windows/win32/fileio/reparse-point-operations),
+and [MoveFileEx](https://learn.microsoft.com/en-us/windows/win32/api/winbase/nf-winbase-movefileexa).
+
+## Decision
+
+`ArtifactBoundary` is the only Phase 2 API allowed to copy a user source into the
+artifact store or publish provider output. It is owner-scoped, fail-closed, and
+provider-independent.
+
+### 1. Explicit source grants and immutable snapshots
+
+Copy-in requires an `ArtifactSourceGrant(owner, job_id, source_turn_id, source_path)`.
+The repository job owner is checked before the path is inspected; an owner mismatch
+does not disclose the source path. The source path must be absolute and bounded, with
+no NUL, alternate-data-stream component, link, junction, or other reparse component.
+
+The source must be a regular, non-sparse file with exactly one directory entry
+(hardlinks are rejected). Its device/inode, size, and timestamps are captured before
+and after a bounded read. Any mutation, disappearance, oversize read, or non-regular
+object aborts the transfer. The source is read-only from this boundary and is never
+renamed, replaced, or deleted.
+
+### 2. Byte-derived content policy
+
+MIME is derived from bytes, never from an extension or model declaration. PNG, JPEG,
+WebP, finite JSON, and ordinary UTF-8 text are recognized. Portable executables,
+ELF/OLE files, shortcuts, shebang scripts, archives, active HTML/SVG/JavaScript,
+and common shell/PowerShell launchers are rejected. Unknown non-active bytes may be
+stored as `application/octet-stream`; this is metadata safety, not permission to decode
+or execute them. Image decoding remains a later provider qualification gate.
+
+### 3. Private output staging and exact claims
+
+The provider receives a private staging directory and returns a finite list of
+`OutputClaim(relative_path, mime_type?)`. Relative paths use a narrow forward-slash
+grammar and cannot contain dot segments, backslashes, drive/ADS syntax, or reparse
+components. The staged file set must equal the claim set exactly. Extra files are
+moved into a randomized `.quarantine` directory and no output is published. Missing
+claims, invalid paths, links, mutation, active/archive content, MIME mismatches,
+per-file limits, output-count limits, and aggregate-size limits fail closed.
+
+Every declared output is fully read, identity-checked, MIME-sniffed, and hashed before
+the first output is published. This prevents a late invalid file from leaving a
+partially visible result set. On validation failure, the offending staged file is
+quarantined. On publication failure, already-published records are deleted and all
+remaining staged files are quarantined; inability to complete cleanup reports the
+stable `artifact_cleanup_pending` category for supervisor recovery.
+
+### 4. Safe repository publication
+
+The repository generates an opaque digest-derived artifact name. It writes bytes to
+an exclusive temporary file, flushes and fsyncs it, atomically replaces the final
+path, and inserts the SQLite record only after the file commit succeeds. Artifact
+read, delete, and expiry purge re-check artifact-root confinement, regular-file/link
+state, and the stored SHA-256. A database row can never authorize deletion or reading
+of a path outside the configured artifact root.
+
+### 5. Stable failure categories
+
+The boundary exposes only categories such as `artifact_owner_mismatch`,
+`artifact_path_invalid`, `artifact_reparse_point`, `artifact_hardlink_rejected`,
+`artifact_source_changed`, `artifact_too_large`, `invalid_artifact`,
+`artifact_unclaimed_output`, `artifact_mime_mismatch`, `artifact_output_limit`,
+`artifact_publish_failed`, and `artifact_cleanup_pending`. Raw paths and operating
+system details do not cross the application boundary.
+
+## Verification and evidence
+
+`tests/test_phase2_artifact_boundary.py` covers owner binding, source preservation,
+ADS/link/reparse rejection, source mutation, exact claims, quarantine, active/archive
+rejection, executable rejection, non-finite JSON rejection, MIME mismatch, aggregate
+limits, and all-or-nothing publication rollback. The repository tests continue to
+cover hash/size/retention behavior. The complete matrix is recorded in the
+[Phase 2 evidence log](0001-phase2-evidence.md).
+
+## Explicit non-goals
+
+This ADR does not authorize image decoding, thumbnail generation, archive extraction,
+provider loading, code execution, model tool exposure, network access, or automatic
+execution. The next gate is fixed-function provider qualification inside the already
+planned OS sandbox, followed by a lifecycle health check and external security review.
diff --git a/docs/adr/0001-phase2-evidence.md b/docs/adr/0001-phase2-evidence.md
index 4d824bb..7288d7d 100644
--- a/docs/adr/0001-phase2-evidence.md
+++ b/docs/adr/0001-phase2-evidence.md
@@ -1,7 +1,7 @@
 # ADR-0001 Phase 2 evidence log
 
 - **Phase:** 2 — signed image recipes and calculator/check primitives
-- **Status:** Typed contract, signed-manifest verification, native broker transport, and signed bundle installation complete; artifact staging, provider, and release gates remain open
+- **Status:** Typed contract, signed-manifest verification, native broker transport, signed bundle installation, and trusted artifact boundary complete; provider and release gates remain open
 - **Scope:** Provider-independent validation and deterministic trusted primitives only
 - **Source decision:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md)
 - **Contract ADR:** [Phase 2 typed recipe and primitive contract](0001-phase2-recipe-contract.md)
@@ -18,7 +18,8 @@
 | Signed bundle installation/update | **Complete (storage-only)** | Digest-named immutable generations, exclusive staging, atomic activation state, chained keyring rotation, explicit rollback authorization, and previous-generation recovery are covered by installer tests. No provider is loaded. |
 | Authenticated broker contract | **Complete (transport-neutral)** | Bounded versioned frames, direction-specific HMAC keys, canonical messages, peer ACL/integrity policy, and owner-scoped authorization are covered by adversarial tests. |
 | Native named-pipe adapter/DACL/peer-token binding | **Complete (transport-only)** | Protected local pipe, expected PID, OS token identity, X25519/HKDF handshake, direction keys, and close-on-error lifecycle are covered by native broker tests. |
-| User-artifact copy-in, image decoding, output validation, publication | **Blocked / next gate** | The signed bundle installer copies only its declared bundle files; user-owned artifact snapshotting, codec validation, output publication, and source non-overwrite proofs remain unimplemented. |
+| User-artifact copy-in, output validation, and publication | **Complete (boundary only)** | Explicit owner/turn grants, bounded stable snapshots, link/reparse/hardlink/sparse/ADS rejection, byte-derived MIME policy, exact output claims, quarantine, hash/size limits, atomic repository publication, rollback, and cleanup categories are covered by `tests/test_phase2_artifact_boundary.py`. |
+| Image decoding and provider-produced image outputs | **Blocked / next gate** | The boundary records safe bytes and metadata only; codecs, decompression, thumbnails, and provider execution remain intentionally unimplemented. |
 | Production sandbox provider and execution route | **Blocked / next gate** | The native pipe is transport-only; no Wasmtime, AppContainer, Job Object, subprocess, lifecycle provider, or production execution route was added. |
 
 ## Security invariants
@@ -55,6 +56,16 @@
 14. The activation pointer is atomically replaced only after the generation is
     verified; keyring updates are signature-chained and rollback/recovery always need
     a separate trusted local decision.
+15. User copy-in requires an owner-bound source grant and never mutates or overwrites
+    the selected source; source identity, size, and timestamps must be stable across
+    the bounded read.
+16. Reparse points, hardlinks, sparse files, devices, ADS/path ambiguity, active
+    content, archives, and non-finite JSON are rejected before artifact publication.
+17. Provider output claims must exactly match the private staging file set; all files
+    are validated before any publication, and publication failure rolls back records
+    while quarantine/cleanup failures surface for supervisor recovery.
+18. Artifact records are opaque IDs; repository read/delete/purge operations remain
+    confined to the configured artifact root and verify the stored SHA-256.
 
 ## Re-run target
 
@@ -64,6 +75,7 @@ python -m pytest tests/test_phase2_manifest.py -q
 python -m pytest tests/test_phase2_broker.py -q
 python -m pytest tests/test_phase2_native_broker.py -q
 python -m pytest tests/test_phase2_bundle_installer.py -q
+python -m pytest tests/test_phase2_artifact_boundary.py -q
 python -m compileall -q backend\cortex_backend\execution tests
 python -m pytest -q
 python tools/generate_contracts.py
@@ -74,9 +86,9 @@ npm.cmd test --prefix frontend -- --run
 ```
 
 **Validation result (2026-07-21):** 16 Phase 2 contract tests, 9 signed-manifest tests,
-7 broker-contract tests, 9 native-broker tests, 7 bundle-installer tests, and the full
-Python suite passed (171 tests total) with one native-platform skip and one pre-existing
-`pytest-asyncio` deprecation warning.
+7 broker-contract tests, 9 native-broker tests, 7 bundle-installer tests, and 16
+artifact-boundary tests passed; the full Python suite passed (187 tests total) with one
+native-platform skip and one pre-existing `pytest-asyncio` deprecation warning.
 Frontend lint, typecheck, production build, and all 39 frontend tests passed. Contract
 generation, compileall, and `git diff --check` passed. No production execution
 provider is enabled.
diff --git a/docs/adr/0001-phase2-recipe-contract.md b/docs/adr/0001-phase2-recipe-contract.md
index 499e3fe..79b76ca 100644
--- a/docs/adr/0001-phase2-recipe-contract.md
+++ b/docs/adr/0001-phase2-recipe-contract.md
@@ -1,6 +1,6 @@
 # ADR-0001 Phase 2 typed recipe and primitive contract
 
-- **Status:** Typed contract, signed-manifest verification, native broker transport, and signed bundle installation implemented and verified; copy-in/output and provider enablement remain blocked
+- **Status:** Typed contract, signed-manifest verification, native broker transport, signed bundle installation, and trusted artifact boundary implemented and verified; provider enablement remains blocked
 - **Parent:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md)
 - **Depends on:** [Phase 1 production lifecycle gate](0001-phase1-production-lifecycle.md)
 - **Scope:** Typed fixed-function image plans, calculator/check primitives, canonical
@@ -54,8 +54,9 @@ This ADR does not authorize:
   verification plus storage installation are implemented separately in
   [the signed-manifest ADR](0001-phase2-signed-manifest.md) and
   [the bundle installation ADR](0001-phase2-bundle-installation.md);
-- image codecs, thumbnails, or decompression handling;
-- artifact copy-in, output validation, atomic publication, or source ownership binding;
+- image decoding, codecs, thumbnails, archive extraction, or provider-produced content
+  handling beyond the
+  trusted [artifact boundary](0001-phase2-artifact-boundary.md);
 - production execution beyond the transport-only
   [native broker adapter](0001-phase2-native-broker.md);
 - Wasmtime/WASI, AppContainer/LPAC, Job Object, host process, or any other provider;
@@ -66,9 +67,7 @@ packaged application remains on the explicitly disabled lifecycle from Phase 1.
 
 ## Required next gates
 
-1. Implement trusted user-artifact copy-in/output validation and artifact publication tests,
-   including parser fuzzing and source non-overwrite proofs.
-2. Qualify the fixed-function provider inside the OS sandbox and wire it only through
+1. Qualify the fixed-function provider inside the OS sandbox and wire it only through
    a passing lifecycle health check after external review.
 
 ## Verification
diff --git a/tests/test_phase2_artifact_boundary.py b/tests/test_phase2_artifact_boundary.py
new file mode 100644
index 0000000..600810f
--- /dev/null
+++ b/tests/test_phase2_artifact_boundary.py
@@ -0,0 +1,273 @@
+"""Adversarial trusted copy-in, MIME validation, quarantine, and publication tests."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+import cortex_backend.execution.artifact_boundary as boundary_module
+from cortex_backend.execution import (
+    ArtifactBoundary,
+    ArtifactBoundaryError,
+    ArtifactSourceGrant,
+    ExecutionRepository,
+    ExecutionRepositoryError,
+    OutputClaim,
+)
+from cortex_backend.execution.artifact_boundary import sniff_artifact_mime
+
+
+def _repository(tmp_path: Path, *, maximum: int = 128) -> tuple[ExecutionRepository, str]:
+    repository = ExecutionRepository(
+        tmp_path / "execution.sqlite",
+        tmp_path / "artifacts",
+        max_artifact_bytes=maximum,
+    )
+    job, _created = repository.create_job(
+        job_id="job-artifact-boundary",
+        owner="session-a",
+        request_id="request-artifact-boundary",
+        profile="artifact.transform.v1",
+        payload={},
+    )
+    return repository, job.job_id
+
+
+def _grant(path: Path, job_id: str = "job-artifact-boundary") -> ArtifactSourceGrant:
+    return ArtifactSourceGrant(
+        owner="session-a",
+        job_id=job_id,
+        source_turn_id="turn-1",
+        source_path=path,
+    )
+
+
+def test_copy_in_is_owner_bound_mime_sniffed_and_never_overwrites_source(tmp_path: Path):
+    repository, job_id = _repository(tmp_path)
+    source = tmp_path / "uploaded.bin"
+    content = b"\x89PNG\r\n\x1a\ntrusted bytes"
+    source.write_bytes(content)
+    boundary = ArtifactBoundary(repository)
+
+    artifact = boundary.copy_in(_grant(source, job_id))
+
+    assert artifact.mime_type == "image/png"
+    assert artifact.sha256
+    assert source.read_bytes() == content
+    assert repository.read_artifact(artifact.artifact_id) == content
+    assert Path(artifact.path).is_relative_to(repository.artifact_root.resolve())
+
+
+def test_copy_in_rejects_wrong_owner_and_alternate_data_stream_paths(tmp_path: Path):
+    repository, job_id = _repository(tmp_path)
+    source = tmp_path / "input.txt"
+    source.write_text("hello", encoding="utf-8")
+    boundary = ArtifactBoundary(repository)
+
+    with pytest.raises(ArtifactBoundaryError) as owner_error:
+        boundary.copy_in(
+            ArtifactSourceGrant(
+                owner="session-b",
+                job_id=job_id,
+                source_turn_id="turn-1",
+                source_path=source,
+            )
+        )
+    assert owner_error.value.code == "artifact_owner_mismatch"
+
+    with pytest.raises(ArtifactBoundaryError) as ads_error:
+        boundary.copy_in(_grant(Path(f"{source}:secret"), job_id))
+    assert ads_error.value.code == "artifact_path_invalid"
+
+
+def test_copy_in_rejects_source_mutation_during_snapshot(tmp_path: Path, monkeypatch):
+    repository, job_id = _repository(tmp_path)
+    source = tmp_path / "input.txt"
+    source.write_text("stable", encoding="utf-8")
+    boundary = ArtifactBoundary(repository)
+    original = boundary_module._file_identity
+    calls = 0
+
+    def changed(path: Path):
+        nonlocal calls
+        calls += 1
+        identity = original(path)
+        return identity if calls == 1 else (*identity[:-1], identity[-1] + 1)
+
+    monkeypatch.setattr(boundary_module, "_file_identity", changed)
+    with pytest.raises(ArtifactBoundaryError) as error:
+        boundary.copy_in(_grant(source, job_id))
+    assert error.value.code == "artifact_source_changed"
+
+
+def test_copy_in_rejects_hardlinks_and_symlinks_when_platform_allows_them(tmp_path: Path):
+    repository, job_id = _repository(tmp_path)
+    external = tmp_path / "external.bin"
+    external.write_bytes(b"hardlink")
+    hardlink = tmp_path / "hardlink.bin"
+    try:
+        hardlink.hardlink_to(external)
+    except (OSError, NotImplementedError):
+        pytest.skip("hard links are unavailable on this platform")
+    boundary = ArtifactBoundary(repository)
+    with pytest.raises(ArtifactBoundaryError) as hardlink_error:
+        boundary.copy_in(_grant(hardlink, job_id))
+    assert hardlink_error.value.code == "artifact_hardlink_rejected"
+
+    symlink = tmp_path / "symlink.bin"
+    try:
+        symlink.symlink_to(external)
+    except (OSError, NotImplementedError):
+        pytest.skip("symbolic links are unavailable on this platform")
+    with pytest.raises(ArtifactBoundaryError) as symlink_error:
+        boundary.copy_in(_grant(symlink, job_id))
+    assert symlink_error.value.code == "artifact_reparse_point"
+
+
+def test_output_collection_requires_exact_claims_and_publishes_only_after_validation(tmp_path: Path):
+    repository, job_id = _repository(tmp_path)
+    output_root = tmp_path / "output"
+    output_root.mkdir()
+    (output_root / "result.txt").write_text("safe output", encoding="utf-8")
+    boundary = ArtifactBoundary(repository)
+
+    published = boundary.collect_outputs(
+        job_id,
+        "session-a",
+        output_root,
+        [OutputClaim("result.txt", "text/plain")],
+    )
+
+    assert len(published) == 1
+    assert published[0].mime_type == "text/plain"
+    assert repository.read_artifact(published[0].artifact.artifact_id) == b"safe output"
+    assert not (output_root / "result.txt").exists()
+
+
+def test_output_extra_file_is_quarantined_without_partial_publication(tmp_path: Path):
+    repository, job_id = _repository(tmp_path)
+    output_root = tmp_path / "output"
+    output_root.mkdir()
+    (output_root / "declared.txt").write_text("declared", encoding="utf-8")
+    (output_root / "secret.txt").write_text("unclaimed", encoding="utf-8")
+    boundary = ArtifactBoundary(repository)
+
+    with pytest.raises(ArtifactBoundaryError) as error:
+        boundary.collect_outputs(
+            job_id,
+            "session-a",
+            output_root,
+            [OutputClaim("declared.txt", "text/plain")],
+        )
+    assert error.value.code == "artifact_unclaimed_output"
+    with repository.connect() as connection:
+        assert connection.execute("SELECT COUNT(*) FROM execution_artifacts").fetchone()[0] == 0
+    assert list((output_root / ".quarantine").iterdir())
+
+
+@pytest.mark.parametrize(
+    ("content", "expected"),
+    [
+        (b"<!doctype html><script>alert(1)</script>", "invalid_artifact"),
+        (b"MZ\x90\x00unsafe executable", "invalid_artifact"),
+        (b"PK\x03\x04archive", "invalid_artifact"),
+        (b"0" * 257 + b"ustar" + b"tar", "invalid_artifact"),
+        (b'{"value": NaN}', "invalid_artifact"),
+        (b"plain text", "artifact_mime_mismatch"),
+    ],
+)
+def test_output_rejects_active_archive_or_mismatched_content(
+    tmp_path: Path,
+    content: bytes,
+    expected: str,
+):
+    repository, job_id = _repository(tmp_path, maximum=max(128, len(content)))
+    output_root = tmp_path / "output"
+    output_root.mkdir()
+    (output_root / "result.bin").write_bytes(content)
+    boundary = ArtifactBoundary(repository)
+
+    with pytest.raises(ArtifactBoundaryError) as error:
+        boundary.collect_outputs(
+            job_id,
+            "session-a",
+            output_root,
+            [OutputClaim("result.bin", "image/png" if expected == "artifact_mime_mismatch" else None)],
+        )
+    assert error.value.code == expected
+
+
+def test_output_publication_rolls_back_prior_artifacts_on_batch_failure(tmp_path: Path, monkeypatch):
+    repository, job_id = _repository(tmp_path)
+    output_root = tmp_path / "output"
+    output_root.mkdir()
+    (output_root / "one.txt").write_text("one", encoding="utf-8")
+    (output_root / "two.txt").write_text("two", encoding="utf-8")
+    boundary = ArtifactBoundary(repository)
+    original = repository.publish_artifact
+    calls = 0
+
+    def fail_second(*args, **kwargs):
+        nonlocal calls
+        calls += 1
+        if calls == 2:
+            raise ExecutionRepositoryError("injected publication failure")
+        return original(*args, **kwargs)
+
+    monkeypatch.setattr(repository, "publish_artifact", fail_second)
+    with pytest.raises(ArtifactBoundaryError) as error:
+        boundary.collect_outputs(
+            job_id,
+            "session-a",
+            output_root,
+            [OutputClaim("one.txt", "text/plain"), OutputClaim("two.txt", "text/plain")],
+        )
+    assert error.value.code == "artifact_publish_failed"
+    with repository.connect() as connection:
+        assert connection.execute("SELECT COUNT(*) FROM execution_artifacts").fetchone()[0] == 0
+    assert list((output_root / ".quarantine").iterdir())
+
+
+def test_json_non_finite_numbers_are_not_misclassified_as_text():
+    with pytest.raises(ArtifactBoundaryError) as error:
+        sniff_artifact_mime(b'{"value": Infinity}')
+    assert error.value.code == "invalid_artifact"
+
+
+def test_repository_expiry_cleanup_fails_closed_for_external_database_paths(tmp_path: Path):
+    repository, job_id = _repository(tmp_path)
+    source = tmp_path / "input.txt"
+    source.write_text("safe", encoding="utf-8")
+    artifact = ArtifactBoundary(repository).copy_in(_grant(source, job_id))
+    external = tmp_path / "outside.txt"
+    external.write_text("must remain", encoding="utf-8")
+    with repository.connect() as connection:
+        connection.execute(
+            "UPDATE execution_artifacts SET path = ? WHERE artifact_id = ?",
+            (str(external), artifact.artifact_id),
+        )
+
+    with pytest.raises(ExecutionRepositoryError):
+        repository.purge_expired(now="9999-01-01T00:00:00+00:00")
+    assert external.read_text(encoding="utf-8") == "must remain"
+
+
+def test_output_limits_and_invalid_claims_fail_before_publication(tmp_path: Path):
+    repository, job_id = _repository(tmp_path, maximum=8)
+    output_root = tmp_path / "output"
+    output_root.mkdir()
+    (output_root / "result.txt").write_text("too large", encoding="utf-8")
+    boundary = ArtifactBoundary(repository, max_total_output_bytes=8)
+
+    with pytest.raises(ArtifactBoundaryError) as error:
+        boundary.collect_outputs(
+            job_id,
+            "session-a",
+            output_root,
+            [OutputClaim("result.txt", "text/plain")],
+        )
+    assert error.value.code == "artifact_too_large"
+
+    with pytest.raises(ValueError):
+        OutputClaim("../escape.txt")