diff --git a/tartarus/jail.py b/tartarus/jail.py
new file mode 100644
index 0000000..4f5f47a
--- /dev/null
+++ b/tartarus/jail.py
@@ -0,0 +1,475 @@
+"""The JailBuilder: bwrap confinement for brokered commands (PLAN.md §6.7).
+
+Builds a JailSpec from a capability grant and executes the command inside
+bubblewrap. The jail binds only the declared shell and capability closures
+read-only, mounts the work tree read-only, clears the host env, and sets PATH
+explicitly. Writable grants re-bind only declared work-tree paths as writable;
+package grants append package bin directories only for that one invocation.
+Network grants route proxy-aware commands through a per-call filtering proxy;
+plain raw-socket containment is a later namespace/firewall step. Unrestricted
+grants skip bwrap entirely after policy approval, but still use the shell PATH.
+"""
+
+import os
+import signal
+import shlex
+import shutil
+import subprocess
+import threading
+import time
+import uuid
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from queue import Empty, Queue
+
+from tartarus.manifest import Grant
+from tartarus.network_proxy import FilteringProxy
+
+DEFAULT_JAIL_TIMEOUT_SECONDS = 30
+TIMEOUT_EXIT_CODE = 124  # matches coreutils `timeout`
+
+
+class JailError(Exception):
+    """Raised when a jail cannot be built or run as requested."""
+
+
+@dataclass(frozen=True)
+class ExecResult:
+    code: int
+    stdout: str
+    stderr: str
+    network_summary: str | None = None
+
+
+@dataclass
+class BackgroundHandle:
+    """A detached jailed process, handed to the BackgroundRegistry.
+
+    `proc` is the launched bwrap process (its own session leader, so `pgid`
+    addresses the whole tree for signalling). `log_path` is the combined
+    stdout+stderr sink the registry tails. `proxy`, when present, is the
+    per-task filtering proxy whose lifetime the registry owns — it is stopped
+    when the task exits or the harness shuts down.
+    """
+
+    proc: subprocess.Popen
+    pgid: int
+    log_path: str
+    proxy: FilteringProxy | None = None
+
+
+@dataclass(frozen=True)
+class JailSpec:
+    work_tree: str
+    shell_path: str
+    base_env: dict[str, str]
+    writable: list[str] = field(default_factory=list)
+    extra_path: list[str] = field(default_factory=list)  # granted package bin dirs
+    # The store paths bound read-only into the jail: the agent's baseline closure
+    # (shell PATH + CA bundle) plus this grant's package closure. The jail sees
+    # exactly these store paths and nothing else, so a capability reaches only its
+    # declared closure (PLAN.md §13).
+    bind_paths: list[str] = field(default_factory=list)
+    allowed_hosts: list[str] = field(default_factory=list)
+    network: str = "none"  # "none" | "proxy"
+    unrestricted: bool = False
+
+
+class JailBuilder:
+    def __init__(
+        self,
+        work_tree: str,
+        shell_path: str,
+        base_env: dict[str, str] | None = None,
+        proxy_factory: Callable[[list[str]], FilteringProxy] | None = None,
+        shell_closure: list[str] | None = None,
+    ):
+        self._work_tree = os.path.abspath(work_tree)
+        self._shell_path = shell_path
+        self._base_env = dict(base_env or {})
+        self._bwrap_path = shutil.which("bwrap") or "bwrap"
+        self._proxy_factory = proxy_factory or FilteringProxy
+        # The baseline store paths every jailed call binds (shell PATH + CA
+        # bundle closure), before this call's own grant closure is added.
+        self._shell_closure = list(shell_closure or [])
+
+    def build(self, grant: Grant) -> JailSpec:
+        return JailSpec(
+            work_tree=self._work_tree,
+            shell_path=self._shell_path,
+            base_env=self._base_env,
+            writable=_validated_writable_paths(grant.writable),
+            extra_path=list(grant.package_bins),
+            bind_paths=_dedup(self._shell_closure + list(grant.closure_paths)),
+            allowed_hosts=list(grant.allowed_hosts),
+            network="proxy" if grant.allowed_hosts else "none",
+            unrestricted=grant.unrestricted,
+        )
+
+    def exec(
+        self,
+        spec: JailSpec,
+        command: str,
+        timeout: int | None = DEFAULT_JAIL_TIMEOUT_SECONDS,
+        output_callback: Callable[[str], None] | None = None,
+        cancellation: threading.Event | None = None,
+    ) -> ExecResult:
+        if spec.unrestricted:
+            return self._exec_unrestricted(
+                spec, command, timeout, output_callback, cancellation
+            )
+
+        if spec.network == "proxy":
+            with self._proxy_factory(spec.allowed_hosts) as proxy:
+                result = self._exec_argv(
+                    self._bwrap_argv(spec, command, proxy.url),
+                    timeout,
+                    output_callback,
+                    cancellation,
+                )
+                return _append_stderr(result, proxy.summary())
+
+        return self._exec_argv(
+            self._bwrap_argv(spec, command), timeout, output_callback, cancellation
+        )
+
+    def exec_background(self, spec: JailSpec, command: str) -> BackgroundHandle:
+        """Launch a jailed command detached and return immediately.
+
+        Unlike `exec`, output is not captured into pipes (which would deadlock a
+        long-lived task) — it is redirected to a per-task log file the registry
+        tails. When the spec carries network grants, a filtering proxy is started
+        here and handed to the caller, which owns stopping it when the task ends.
+        """
+        if spec.unrestricted:
+            raise JailError("background execution does not support unrestricted grants")
+
+        proxy: FilteringProxy | None = None
+        proxy_url: str | None = None
+        log_file: object | None = None
+
+        try:
+            if spec.network == "proxy":
+                proxy = self._proxy_factory(spec.allowed_hosts)
+                proxy.start()
+                proxy_url = proxy.url
+
+            argv = self._bwrap_argv(spec, command, proxy_url)
+            log_path = self._background_log_path()
+            log_file = open(log_path, "wb")
+            try:
+                proc = subprocess.Popen(
+                    argv,
+                    env={},
+                    stdout=log_file,
+                    stderr=subprocess.STDOUT,
+                    start_new_session=True,  # own session leader: pgid addresses the tree
+                )
+            finally:
+                # The child holds its own dup of the fd; the parent's copy is not
+                # needed once the process is spawned.
+                log_file.close()
+
+            return BackgroundHandle(
+                proc=proc, pgid=os.getpgid(proc.pid), log_path=log_path, proxy=proxy
+            )
+        except FileNotFoundError as missing:
+            if proxy is not None:
+                proxy.stop()
+            raise JailError(f"jail runtime not found: {missing}") from missing
+        except OSError as error:
+            if proxy is not None:
+                proxy.stop()
+            raise JailError(
+                f"cannot open background log or run jail: {error}"
+            ) from error
+
+    def _background_log_path(self) -> str:
+        bg_dir = os.path.join(self._work_tree, ".tartarus", "bg")
+        os.makedirs(bg_dir, exist_ok=True)
+        return os.path.join(bg_dir, f"{uuid.uuid4().hex}.log")
+
+    def _exec_argv(
+        self,
+        argv: list[str],
+        timeout: int | None,
+        output_callback: Callable[[str], None] | None = None,
+        cancellation: threading.Event | None = None,
+    ) -> ExecResult:
+        try:
+            proc = subprocess.Popen(
+                argv,
+                env={},
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                start_new_session=True,  # isolate process group for timeout kills
+            )
+        except FileNotFoundError as missing:
+            raise JailError(f"jail runtime not found: {missing}") from missing
+
+        return _wait_for_process(proc, timeout, output_callback, cancellation)
+
+    def _exec_unrestricted(
+        self,
+        spec: JailSpec,
+        command: str,
+        timeout: int | None,
+        output_callback: Callable[[str], None] | None = None,
+        cancellation: threading.Event | None = None,
+    ) -> ExecResult:
+        env = {"PATH": self._compose_path(spec), **spec.base_env}
+        try:
+            proc = subprocess.Popen(
+                shlex.split(command),
+                cwd=spec.work_tree,
+                env=env,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                start_new_session=True,
+            )
+        except FileNotFoundError as missing:
+            raise JailError(
+                f"unrestricted command runtime not found: {missing}"
+            ) from missing
+
+        return _wait_for_process(proc, timeout, output_callback, cancellation)
+
+    def _bwrap_argv(
+        self,
+        spec: JailSpec,
+        command: str,
+        proxy_url: str | None = None,
+    ) -> list[str]:
+        env_args = ["--setenv", "PATH", self._compose_path(spec)]
+        env_args += self._network_env_args(proxy_url)
+        for key, value in spec.base_env.items():
+            env_args += ["--setenv", key, value]
+
+        return [
+            self._bwrap_path,
+            # Only this call's closure is visible, read-only — not the whole
+            # store. bwrap synthesizes the /nix/store parent, so the in-jail
+            # store contains exactly the bound closure (PLAN.md §13).
+            *_store_bind_args(spec.bind_paths),
+            *self._work_tree_bind_args(spec),
+            *self._writable_bind_args(spec),
+            "--chdir",
+            "/work",
+            "--unshare-all",  # new net/pid/ipc/uts/mount/cgroup namespaces
+            *self._network_namespace_args(spec),
+            "--die-with-parent",
+            "--dir",
+            "/proc",
+            "--dev",
+            "/dev",
+            "--clearenv",  # start from empty env, then set PATH explicitly
+            *env_args,
+            "--",
+            *shlex.split(command),
+        ]
+
+    @staticmethod
+    def _network_namespace_args(spec: JailSpec) -> list[str]:
+        if spec.network == "proxy":
+            return ["--share-net"]
+        return []
+
+    @staticmethod
+    def _network_env_args(proxy_url: str | None) -> list[str]:
+        if proxy_url is None:
+            return []
+        return [
+            "--setenv",
+            "HTTP_PROXY",
+            proxy_url,
+            "--setenv",
+            "HTTPS_PROXY",
+            proxy_url,
+            "--setenv",
+            "ALL_PROXY",
+            proxy_url,
+            "--setenv",
+            "http_proxy",
+            proxy_url,
+            "--setenv",
+            "https_proxy",
+            proxy_url,
+            "--setenv",
+            "all_proxy",
+            proxy_url,
+            "--setenv",
+            "NO_PROXY",
+            "",
+            "--setenv",
+            "no_proxy",
+            "",
+        ]
+
+    @staticmethod
+    def _work_tree_bind_args(spec: JailSpec) -> list[str]:
+        if "." in spec.writable:
+            return ["--bind", spec.work_tree, "/work"]
+        return ["--ro-bind", spec.work_tree, "/work"]
+
+    def _writable_bind_args(self, spec: JailSpec) -> list[str]:
+        args = []
+        for relative_path in spec.writable:
+            if relative_path == ".":
+                continue
+            host_path = _host_writable_path(spec.work_tree, relative_path)
+            jail_path = "/work" if relative_path == "." else f"/work/{relative_path}"
+            args += ["--bind", host_path, jail_path]
+        return args
+
+    @staticmethod
+    def _compose_path(spec: JailSpec) -> str:
+        if not spec.extra_path:
+            return spec.shell_path
+        return ":".join([spec.shell_path, *spec.extra_path])
+
+
+def _wait_for_process(
+    proc: subprocess.Popen,
+    timeout: int | None,
+    output_callback: Callable[[str], None] | None,
+    cancellation: threading.Event | None,
+) -> ExecResult:
+    output_queue: Queue[tuple[str, str | None]] = Queue()
+    stdout_parts: list[str] = []
+    stderr_parts: list[str] = []
+    threads = [
+        _start_pipe_reader(proc.stdout, "stdout", output_queue),
+        _start_pipe_reader(proc.stderr, "stderr", output_queue),
+    ]
+    deadline = None if timeout is None else time.monotonic() + timeout
+    termination_error: str | None = None
+
+    while proc.poll() is None:
+        _drain_output_queue(output_queue, stdout_parts, stderr_parts, output_callback)
+        if cancellation is not None and cancellation.is_set():
+            termination_error = "command cancelled"
+            _terminate_process_group(proc)
+            break
+        if deadline is not None and time.monotonic() >= deadline:
+            termination_error = f"command timed out after {timeout}s"
+            _terminate_process_group(proc)
+            break
+        time.sleep(0.01)
+
+    proc.wait()
+    for thread in threads:
+        thread.join()
+    _drain_output_queue(output_queue, stdout_parts, stderr_parts, output_callback)
+
+    stdout = "".join(stdout_parts)
+    stderr = "".join(stderr_parts)
+    if termination_error is not None:
+        stderr = "\n".join(part for part in (stderr.strip(), termination_error) if part)
+        if stderr:
+            stderr += "\n"
+        return ExecResult(TIMEOUT_EXIT_CODE, stdout, stderr)
+    return ExecResult(proc.returncode, stdout, stderr)
+
+
+def _start_pipe_reader(pipe, stream_name: str, output_queue: Queue):
+    def read_lines() -> None:
+        if pipe is None:
+            output_queue.put((stream_name, None))
+            return
+        try:
+            for line in pipe:
+                output_queue.put((stream_name, line))
+        finally:
+            pipe.close()
+            output_queue.put((stream_name, None))
+
+    thread = threading.Thread(
+        target=read_lines,
+        name=f"tartarus-{stream_name}-reader",
+        daemon=True,
+    )
+    thread.start()
+    return thread
+
+
+def _drain_output_queue(
+    output_queue: Queue[tuple[str, str | None]],
+    stdout_parts: list[str],
+    stderr_parts: list[str],
+    output_callback: Callable[[str], None] | None,
+) -> None:
+    while True:
+        try:
+            stream_name, text = output_queue.get_nowait()
+        except Empty:
+            return
+        if text is None:
+            continue
+        if stream_name == "stdout":
+            stdout_parts.append(text)
+        else:
+            stderr_parts.append(text)
+        if output_callback is not None:
+            output_callback(text)
+
+
+def _terminate_process_group(proc: subprocess.Popen) -> None:
+    try:
+        os.killpg(proc.pid, signal.SIGTERM)
+    except ProcessLookupError:
+        # The process group may already be gone; fall through so we still reap
+        # any stragglers below.
+        pass
+    try:
+        proc.wait(timeout=1)
+    except subprocess.TimeoutExpired:
+        try:
+            os.killpg(proc.pid, signal.SIGKILL)
+        except ProcessLookupError:
+            pass
+
+
+def _append_stderr(result: ExecResult, message: str) -> ExecResult:
+    stderr = "\n".join(part for part in (result.stderr.strip(), message) if part)
+    if stderr:
+        stderr += "\n"
+    return ExecResult(result.code, result.stdout, stderr, network_summary=message)
+
+
+def _store_bind_args(bind_paths: list[str]) -> list[str]:
+    """`--ro-bind p p` for each closure path, replacing the whole-store mount."""
+    args: list[str] = []
+    for path in bind_paths:
+        args += ["--ro-bind", path, path]
+    return args
+
+
+def _dedup(paths: list[str]) -> list[str]:
+    """De-duplicate while preserving order (closures overlap on shared deps)."""
+    seen: set[str] = set()
+    unique: list[str] = []
+    for path in paths:
+        if path not in seen:
+            seen.add(path)
+            unique.append(path)
+    return unique
+
+
+def _validated_writable_paths(paths: list[str]) -> list[str]:
+    for path in paths:
+        if path.startswith("/"):
+            raise JailError(f"writable path '{path}' must be relative")
+        if ".." in path.split("/"):
+            raise JailError(f"writable path '{path}' escapes the work tree")
+    return list(paths)
+
+
+def _host_writable_path(work_tree: str, relative_path: str) -> str:
+    host_path = os.path.abspath(os.path.join(work_tree, relative_path))
+    if host_path != work_tree and not host_path.startswith(work_tree + os.sep):
+        raise JailError(f"writable path '{relative_path}' escapes the work tree")
+    if not os.path.exists(host_path):
+        os.makedirs(host_path, exist_ok=True)
+    return host_path
diff --git a/tartarus/shell.py b/tartarus/shell.py
new file mode 100644
index 0000000..3433b2d
--- /dev/null
+++ b/tartarus/shell.py
@@ -0,0 +1,48 @@
+"""Minimal shell-path helper for tests and ad-hoc store resolution.
+
+The harness no longer resolves a live shell: an agent's baseline PATH is baked
+into its bundle manifest (`shellPath`) at build time (PLAN.md §14). What remains
+here is `resolve_minimal_shell_path`, used by the jail integration tests to build
+a small PATH from named packages.
+"""
+
+import os
+
+from tartarus.process import ProcessError, run_checked
+
+DEFAULT_SHELL_PACKAGES = ("nixpkgs#coreutils", "nixpkgs#bash")
+
+
+class ShellError(Exception):
+    """Raised when a shell package cannot be resolved to a bin directory."""
+
+
+def resolve_minimal_shell_path(
+    packages: tuple[str, ...] = DEFAULT_SHELL_PACKAGES,
+) -> str:
+    """Realize each package into the store and join their `bin` dirs into a PATH."""
+    return ":".join(_resolve_bin_dir(package) for package in packages)
+
+
+def _resolve_bin_dir(package: str) -> str:
+    store_paths = _build_package(package)
+    if not store_paths:
+        raise ShellError(f"`nix build {package}` produced no store path")
+
+    # A package may have several outputs (out, man, dev, ...); pick the one that
+    # actually carries executables.
+    for store_path in store_paths:
+        bin_dir = os.path.join(store_path, "bin")
+        if os.path.isdir(bin_dir):
+            return bin_dir
+    raise ShellError(f"no output of `{package}` has a bin directory")
+
+
+def _build_package(package: str) -> list[str]:
+    command = ["nix", "build", package, "--no-link", "--print-out-paths"]
+    try:
+        stdout = run_checked(command)
+    except ProcessError as error:
+        raise ShellError(f"cannot build `{package}`: {error}") from error
+
+    return [line for line in stdout.splitlines() if line.strip()]
diff --git a/tests/test_jail.py b/tests/test_jail.py
new file mode 100644
index 0000000..1e689f7
--- /dev/null
+++ b/tests/test_jail.py
@@ -0,0 +1,440 @@
+"""Integration tests for the bwrap jail (PLAN.md §11).
+
+These require Linux with `bwrap` and `nix` present, so they skip elsewhere. They
+prove the security invariants: confinement, content purity, and reach
+isolation.
+"""
+
+import shutil
+import shlex
+import subprocess
+import sys
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+import tartarus.jail
+import pytest
+
+from tartarus.shell import ShellError, resolve_minimal_shell_path
+from tartarus.jail import JailBuilder, JailError
+from tartarus.manifest import Grant
+
+_NEEDS_SANDBOX = pytest.mark.skipif(
+    shutil.which("bwrap") is None or shutil.which("nix") is None,
+    reason="requires bwrap and nix",
+)
+
+
+def _store_root(path: str) -> str:
+    """/nix/store/<hash-name>/anything -> /nix/store/<hash-name>."""
+    return "/".join(path.split("/")[:4])
+
+
+def _closure_of(store_paths: list[str]) -> list[str]:
+    """The transitive closure of the given store paths' roots, via nix-store.
+
+    The harness gets closures from Nix (`closureInfo`); these jail tests compute
+    them independently so the bind set under test mirrors what Nix would emit.
+    """
+    roots = sorted({_store_root(path) for path in store_paths})
+    out = subprocess.run(
+        ["nix-store", "--query", "--requisites", *roots],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    return [line for line in out.stdout.splitlines() if line]
+
+
+@pytest.fixture(scope="module")
+def shell_path():
+    return resolve_minimal_shell_path()
+
+
+@pytest.fixture(scope="module")
+def shell_closure(shell_path):
+    """The baseline bind set: the closure of the minimal shell PATH."""
+    return _closure_of(shell_path.split(":"))
+
+
+@_NEEDS_SANDBOX
+def test_echo_runs_confined(tmp_path, shell_path, shell_closure):
+    jail = JailBuilder(str(tmp_path), shell_path, shell_closure=shell_closure)
+    result = jail.exec(jail.build(Grant()), "echo banana")
+
+    assert result.code == 0
+    assert "banana" in result.stdout
+
+
+@_NEEDS_SANDBOX
+def test_bwrap_parent_environment_does_not_leak_into_proc(
+    tmp_path, shell_path, shell_closure, monkeypatch
+):
+    monkeypatch.setenv("TARTARUS_TEST_HOST_SECRET", "secret-from-host-env")
+    jail = JailBuilder(str(tmp_path), shell_path, shell_closure=shell_closure)
+
+    result = jail.exec(jail.build(Grant()), "cat /proc/1/environ")
+
+    assert "TARTARUS_TEST_HOST_SECRET" not in result.stdout
+    assert "secret-from-host-env" not in result.stdout
+    assert "TARTARUS_TEST_HOST_SECRET" not in result.stderr
+    assert "secret-from-host-env" not in result.stderr
+
+
+@_NEEDS_SANDBOX
+def test_proc_file_descriptors_are_not_available_for_output_injection(
+    tmp_path, shell_path, shell_closure
+):
+    jail = JailBuilder(str(tmp_path), shell_path, shell_closure=shell_closure)
+
+    result = jail.exec(
+        jail.build(Grant()),
+        "bash -c 'echo injected-output > /proc/1/fd/1; echo normal-output'",
+    )
+
+    assert result.code == 0
+    assert result.stdout == "normal-output\n"
+    assert "injected-output" not in result.stdout
+
+
+@_NEEDS_SANDBOX
+def test_host_only_tool_is_absent_inside_jail(tmp_path, shell_path, shell_closure):
+    jail = JailBuilder(str(tmp_path), shell_path, shell_closure=shell_closure)
+    # git is not in the shell closure and was not granted, so it cannot resolve
+    # by name even though the baseline shell binaries do.
+    result = jail.exec(jail.build(Grant()), "git --version")
+
+    assert result.code != 0
+    assert "no such file" in result.stderr.lower()
+
+
+@_NEEDS_SANDBOX
+def test_ungranted_tool_unreachable_by_absolute_store_path(
+    tmp_path, shell_path, shell_closure
+):
+    # The store-bind purity gap (PLAN.md §13): with the whole store mounted, an
+    # un-granted binary was reachable by absolute path. Now only the closure is
+    # bound, so git's own store path does not exist inside the jail even though
+    # its dependencies (bash, coreutils) are part of the shell closure.
+    try:
+        git_bin = resolve_minimal_shell_path(("nixpkgs#git",))
+    except ShellError as error:
+        pytest.skip(f"cannot resolve git package: {error}")
+
+    jail = JailBuilder(str(tmp_path), shell_path, shell_closure=shell_closure)
+    result = jail.exec(jail.build(Grant()), f"{shlex.quote(git_bin)}/git --version")
+
+    assert result.code != 0
+    assert "no such file" in result.stderr.lower()
+
+
+@_NEEDS_SANDBOX
+def test_no_host_filesystem_beyond_work_tree(tmp_path, shell_path, shell_closure):
+    jail = JailBuilder(str(tmp_path), shell_path, shell_closure=shell_closure)
+    result = jail.exec(jail.build(Grant()), "ls /")
+
+    visible = set(result.stdout.split())
+    assert visible <= {"dev", "nix", "proc", "work"}
+    assert "home" not in visible
+    assert "usr" not in visible
+
+
+@_NEEDS_SANDBOX
+def test_no_network_interfaces_inside_jail(tmp_path, shell_path, shell_closure):
+    jail = JailBuilder(str(tmp_path), shell_path, shell_closure=shell_closure)
+    # --unshare-all removed the network namespace, so /sys/class/net is gone.
+    result = jail.exec(jail.build(Grant()), "ls /sys/class/net")
+
+    assert result.code != 0
+
+
+def test_network_grant_builds_proxy_spec(tmp_path):
+    jail = JailBuilder(str(tmp_path), "/unused/bin")
+    spec = jail.build(Grant(allowed_hosts=["example.com:443"]))
+
+    assert spec.network == "proxy"
+    assert spec.allowed_hosts == ["example.com:443"]
+
+
+def test_proxy_jail_sets_proxy_environment(tmp_path):
+    jail = JailBuilder(str(tmp_path), "/unused/bin")
+    spec = jail.build(Grant(allowed_hosts=["example.com:443"]))
+
+    argv = jail._bwrap_argv(spec, "true", proxy_url="http://127.0.0.1:12345")
+
+    assert "--share-net" in argv
+    assert "HTTP_PROXY" in argv
+    assert "http://127.0.0.1:12345" in argv
+
+
+@_NEEDS_SANDBOX
+def test_proxy_jail_routes_curl_through_allowed_host(
+    tmp_path, shell_path, shell_closure
+):
+    with _HttpServer() as upstream:
+        upstream_host, upstream_port = upstream.server_address
+        curl_bins = _curl_bin_dirs()
+        jail = JailBuilder(
+            str(tmp_path),
+            shell_path,
+            shell_closure=shell_closure,
+        )
+        spec = jail.build(
+            Grant(
+                package_bins=curl_bins,
+                closure_paths=_closure_of(curl_bins),
+                allowed_hosts=[f"{upstream_host}:{upstream_port}"],
+            )
+        )
+
+        result = jail.exec(spec, f"curl -fsS http://{upstream_host}:{upstream_port}/")
+
+        assert result.code == 0
+        assert result.stdout == "hello"
+        assert "proxy decisions: 1 allowed, 0 blocked" in result.stderr
+        assert result.network_summary is not None
+        assert "1 allowed, 0 blocked" in result.network_summary
+
+
+@_NEEDS_SANDBOX
+def test_proxy_jail_blocks_unlisted_host(tmp_path, shell_path, shell_closure):
+    with _HttpServer() as upstream:
+        upstream_host, upstream_port = upstream.server_address
+        curl_bins = _curl_bin_dirs()
+        jail = JailBuilder(
+            str(tmp_path),
+            shell_path,
+            shell_closure=shell_closure,
+        )
+        spec = jail.build(
+            Grant(
+                package_bins=curl_bins,
+                closure_paths=_closure_of(curl_bins),
+                allowed_hosts=["example.com:80"],
+            )
+        )
+
+        result = jail.exec(spec, f"curl -fsS http://{upstream_host}:{upstream_port}/")
+
+        assert result.code != 0
+        assert "proxy decisions: 0 allowed, 1 blocked" in result.stderr
+        assert result.network_summary is not None
+        assert "0 allowed, 1 blocked" in result.network_summary
+
+
+@_NEEDS_SANDBOX
+def test_writable_grant_allows_only_declared_path(tmp_path, shell_path, shell_closure):
+    writable_dir = tmp_path / "allowed"
+    readonly_dir = tmp_path / "readonly"
+    writable_dir.mkdir()
+    readonly_dir.mkdir()
+
+    jail = JailBuilder(str(tmp_path), shell_path, shell_closure=shell_closure)
+    spec = jail.build(Grant(writable=["allowed"]))
+
+    allowed = jail.exec(spec, "bash -c 'echo yes > allowed/file.txt'")
+    denied = jail.exec(spec, "bash -c 'echo no > readonly/file.txt'")
+
+    assert allowed.code == 0
+    assert (writable_dir / "file.txt").read_text().strip() == "yes"
+    assert denied.code != 0
+    assert not (readonly_dir / "file.txt").exists()
+
+
+def test_package_bins_add_extra_path_for_one_spec(tmp_path):
+    jail = JailBuilder(str(tmp_path), "/shell/bin")
+
+    spec = jail.build(Grant(package_bins=["/nix/store/jq/bin"]))
+    baseline = jail.build(Grant())
+
+    assert spec.extra_path == ["/nix/store/jq/bin"]
+    assert baseline.extra_path == []
+
+
+def test_build_unions_shell_and_grant_closure(tmp_path):
+    jail = JailBuilder(
+        str(tmp_path),
+        "/shell/bin",
+        shell_closure=["/nix/store/bash", "/nix/store/coreutils"],
+    )
+
+    spec = jail.build(Grant(closure_paths=["/nix/store/coreutils", "/nix/store/jq"]))
+    baseline = jail.build(Grant())
+
+    # Baseline carries the shell closure; the grant adds its own, de-duplicated.
+    assert baseline.bind_paths == ["/nix/store/bash", "/nix/store/coreutils"]
+    assert spec.bind_paths == [
+        "/nix/store/bash",
+        "/nix/store/coreutils",
+        "/nix/store/jq",
+    ]
+
+
+def test_bwrap_argv_binds_each_closure_path_not_whole_store(tmp_path):
+    jail = JailBuilder(str(tmp_path), "/shell/bin", shell_closure=["/nix/store/bash"])
+    spec = jail.build(Grant(closure_paths=["/nix/store/jq"]))
+
+    argv = jail._bwrap_argv(spec, "true")
+
+    # Each closure path is bound individually; the whole store is never mounted.
+    assert _ro_bind_pairs(argv) >= {
+        ("/nix/store/bash", "/nix/store/bash"),
+        ("/nix/store/jq", "/nix/store/jq"),
+    }
+    assert ("/nix/store", "/nix/store") not in _ro_bind_pairs(argv)
+
+
+def _ro_bind_pairs(argv: list[str]) -> set[tuple[str, str]]:
+    pairs = set()
+    for index, token in enumerate(argv):
+        if token == "--ro-bind":
+            pairs.add((argv[index + 1], argv[index + 2]))
+    return pairs
+
+
+def test_unrestricted_grant_bypasses_bwrap_after_approval_path(tmp_path):
+    work_tree = tmp_path / "work"
+    work_tree.mkdir()
+    secret = tmp_path / "secret.txt"
+    secret.write_text("outside work tree")
+    bin_dir = tmp_path / "bin"
+    bin_dir.mkdir()
+    reader = bin_dir / "read-secret"
+    reader.write_text(
+        "#!/bin/sh\nIFS= read -r line < ../secret.txt || true\nprintf '%s' \"$line\"\n"
+    )
+    reader.chmod(0o755)
+
+    jail = JailBuilder(str(work_tree), str(bin_dir))
+    spec = jail.build(Grant(unrestricted=True))
+
+    result = jail.exec(spec, "read-secret")
+
+    assert result.code == 0
+    assert result.stdout == "outside work tree"
+
+
+def test_exec_streams_unrestricted_output_lines(tmp_path):
+    jail = JailBuilder(str(tmp_path), "/unused/bin")
+    spec = jail.build(Grant(unrestricted=True))
+    lines: list[str] = []
+    code = "import sys; print('one', flush=True); print('two', flush=True)"
+
+    result = jail.exec(
+        spec,
+        f"{shlex.quote(sys.executable)} -c {shlex.quote(code)}",
+        output_callback=lines.append,
+    )
+
+    assert result.code == 0
+    assert result.stdout == "one\ntwo\n"
+    assert lines == ["one\n", "two\n"]
+
+
+def test_exec_cancellation_stops_unrestricted_process(tmp_path):
+    jail = JailBuilder(str(tmp_path), "/unused/bin")
+    spec = jail.build(Grant(unrestricted=True))
+    cancellation = threading.Event()
+    lines: list[str] = []
+    code = (
+        "import time; "
+        "print('started', flush=True); "
+        "time.sleep(10); "
+        "print('finished', flush=True)"
+    )
+
+    def capture(line: str) -> None:
+        lines.append(line)
+        cancellation.set()
+
+    result = jail.exec(
+        spec,
+        f"{shlex.quote(sys.executable)} -c {shlex.quote(code)}",
+        output_callback=capture,
+        cancellation=cancellation,
+    )
+
+    assert result.code == 124
+    assert lines == ["started\n"]
+    assert "command cancelled" in result.stderr
+
+
+class _HelloHandler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.0"
+
+    def do_GET(self):
+        body = b"hello"
+        self.send_response(200)
+        self.send_header("Content-Length", str(len(body)))
+        self.send_header("Connection", "close")
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, format: str, *args) -> None:
+        pass
+
+
+class _HttpServer:
+    def __enter__(self):
+        self._server = ThreadingHTTPServer(("127.0.0.1", 0), _HelloHandler)
+        self._thread = threading.Thread(
+            target=self._server.serve_forever,
+            name="tartarus-nix-jail-test-http-server",
+            daemon=True,
+        )
+        self._thread.start()
+        return self._server
+
+    def __exit__(self, *_args):
+        self._server.shutdown()
+        self._server.server_close()
+        self._thread.join(timeout=5)
+
+
+def _curl_bin_dirs() -> list[str]:
+    try:
+        return [resolve_minimal_shell_path(("nixpkgs#curl",))]
+    except ShellError as error:
+        pytest.skip(f"cannot resolve curl package: {error}")
+
+
+def test_exec_background_stops_proxy_on_popen_failure(tmp_path, monkeypatch):
+    """A network-enabled background task must clean up its proxy even if Popen fails."""
+
+    class FakeProxy:
+        def __init__(self, allowed_hosts):
+            self.allowed_hosts = allowed_hosts
+            self.stopped = False
+
+        def start(self) -> None:
+            pass
+
+        @property
+        def url(self) -> str:
+            return "http://127.0.0.1:9999"
+
+        def stop(self) -> None:
+            self.stopped = True
+
+        def summary(self) -> str:
+            return "fake summary"
+
+    captured: dict[str, FakeProxy] = {}
+
+    def factory(allowed_hosts):
+        proxy = FakeProxy(allowed_hosts)
+        captured["proxy"] = proxy
+        return proxy
+
+    jail = JailBuilder(str(tmp_path), "/bin/sh", proxy_factory=factory)
+    spec = jail.build(Grant(allowed_hosts=["example.com:80"]))
+
+    def raising_popen(*_args, **_kwargs):
+        raise OSError("popen failed")
+
+    monkeypatch.setattr(tartarus.jail.subprocess, "Popen", raising_popen)
+
+    with pytest.raises(JailError):
+        jail.exec_background(spec, "true")
+
+    assert captured["proxy"].stopped is True