Files
hqt/docs/superpowers/plans/2026-06-10-sandboxed-sessions.md
T
2026-06-10 21:05:16 -04:00

41 KiB
Raw Blame History

Sandboxed Sessions (bubblewrap) Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Let a user start a harness session inside a bubblewrap (bwrap) sandbox, chosen per session, with read-only/read-write project access and an internet on/off toggle; sandboxed sessions automatically get the harness's "skip permissions" flag.

Architecture: A new pure module hqt/sandbox.py builds the bwrap … -- <command> argv from a SandboxPolicy plus per-harness Binds. The HarnessConfigurator ABC gains the per-harness skip-permission flags and the credential/config paths to bind. SessionService wraps the spawn/resume command when a session is sandboxed and persists the policy in a new Session.sandbox_json column. The New Session dialog exposes the toggles and disables them when bwrap is unavailable (the same check hqt doctor uses).

Tech Stack: Python 3.14, SQLAlchemy (SQLite, user_version migrations), Textual (TUI), Click (CLI), pytest + pytest-asyncio.


File Structure

  • Create src/hqt/sandbox.pyBind, SandboxPolicy, is_available(), wrap(). Pure (no subprocess); the single source of truth for "can we sandbox?" and "how do we build the bwrap argv".
  • Modify src/hqt/harnesses/base.py — add sandbox_skip_permission_flags class attr, sandbox_binds() method, and a sandboxed: bool = False parameter to the two abstract build methods.
  • Modify src/hqt/harnesses/configurators/{claude,codex,kiro,generic}.py — accept sandboxed, append flags, declare binds.
  • Modify src/hqt/db/models.py — add Session.sandbox_json column.
  • Modify src/hqt/db/migrations.py — add version-2 migration adding the column.
  • Modify src/hqt/sessions/service.pycreate_session(sandbox=...), wrap command, persist policy, re-wrap on resume/respawn.
  • Modify src/hqt/tui/screens/new_session.py — sandbox toggle + sub-controls + availability gating; extend the result tuple.
  • Modify src/hqt/tui/app.py — pass availability to the screen; unpack the sandbox config; pass it to create_session.
  • Modify src/hqt/cli.pydoctor reports bubblewrap availability.
  • Tests: tests/test_sandbox.py (new), and additions to tests/test_harnesses.py, tests/test_db.py, tests/test_sessions.py, tests/test_tui.py, tests/test_cli.py (new or existing).

Run the full suite with uv run pytest -q and lint with uv run ruff check src tests at checkpoints.


Task 1: Sandbox primitives — Bind, SandboxPolicy, is_available()

Files:

  • Create: src/hqt/sandbox.py

  • Test: tests/test_sandbox.py

  • Step 1: Write the failing tests

# tests/test_sandbox.py
import sys
from pathlib import Path

from hqt import sandbox
from hqt.sandbox import Bind, SandboxPolicy


def test_bind_defaults_readonly():
    b = Bind(Path("/home/u/.claude"))
    assert b.src == Path("/home/u/.claude")
    assert b.writable is False


def test_sandbox_policy_fields():
    p = SandboxPolicy(fs="rw", net=True)
    assert p.fs == "rw"
    assert p.net is True


def test_is_available_true_on_linux_with_bwrap(monkeypatch):
    monkeypatch.setattr(sandbox.sys, "platform", "linux")
    monkeypatch.setattr(sandbox.shutil, "which", lambda name: "/usr/bin/bwrap")
    assert sandbox.is_available() is True


def test_is_available_false_without_bwrap(monkeypatch):
    monkeypatch.setattr(sandbox.sys, "platform", "linux")
    monkeypatch.setattr(sandbox.shutil, "which", lambda name: None)
    assert sandbox.is_available() is False


def test_is_available_false_off_linux(monkeypatch):
    monkeypatch.setattr(sandbox.sys, "platform", "darwin")
    monkeypatch.setattr(sandbox.shutil, "which", lambda name: "/usr/bin/bwrap")
    assert sandbox.is_available() is False
  • Step 2: Run tests to verify they fail

Run: uv run pytest tests/test_sandbox.py -q Expected: FAIL — ModuleNotFoundError: No module named 'hqt.sandbox'.

  • Step 3: Create the module with primitives
# src/hqt/sandbox.py
import os
import shutil
import sys
from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class Bind:
    """A host path to expose inside the sandbox (read-only unless writable)."""

    src: Path
    writable: bool = False


@dataclass(frozen=True)
class SandboxPolicy:
    """Per-session sandbox settings.

    fs:  "rw" binds the project dir writable; "ro" binds it read-only.
    net: True shares the host network namespace; False = no connectivity.
    """

    fs: str
    net: bool


def is_available() -> bool:
    """True when bubblewrap can be used: Linux with `bwrap` on PATH.

    The single source of truth for both `hqt doctor` and the New Session dialog.
    """
    return sys.platform.startswith("linux") and shutil.which("bwrap") is not None

(os is imported now; Task 2 uses it for env passthrough.)

  • Step 4: Run tests to verify they pass

Run: uv run pytest tests/test_sandbox.py -q Expected: PASS (5 passed).

  • Step 5: Commit
git add src/hqt/sandbox.py tests/test_sandbox.py
git commit -m "feat: sandbox primitives — Bind, SandboxPolicy, is_available"

Task 2: wrap() — build the bwrap argv

Files:

  • Modify: src/hqt/sandbox.py

  • Test: tests/test_sandbox.py

  • Step 1: Write the failing tests

# tests/test_sandbox.py  (append)
def _split(args, flag):
    """Return list of (src, dst) pairs that follow each occurrence of flag."""
    out = []
    for i, a in enumerate(args):
        if a == flag:
            out.append((args[i + 1], args[i + 2]))
    return out


def test_wrap_command_after_double_dash(tmp_path):
    cmd = ["claude", "--session-id", "x"]
    args = sandbox.wrap(cmd, tmp_path, SandboxPolicy(fs="rw", net=True), [])
    assert args[0] == "bwrap"
    assert "--" in args
    assert args[args.index("--") + 1 :] == cmd


def test_wrap_base_flags(tmp_path):
    args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [])
    assert "--die-with-parent" in args
    assert "--unshare-all" in args
    assert "--proc" in args and "--dev" in args


def test_wrap_net_on_shares_net(tmp_path):
    args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [])
    assert "--share-net" in args


def test_wrap_net_off_does_not_share(tmp_path):
    args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=False), [])
    assert "--share-net" not in args


def test_wrap_cwd_rw_is_bind(tmp_path):
    args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [])
    assert (str(tmp_path), str(tmp_path)) in _split(args, "--bind")


def test_wrap_cwd_ro_is_ro_bind(tmp_path):
    args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="ro", net=True), [])
    assert (str(tmp_path), str(tmp_path)) in _split(args, "--ro-bind")


def test_wrap_writable_bind_spliced(tmp_path):
    cred = tmp_path / "cred"
    cred.mkdir()
    args = sandbox.wrap(
        ["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [Bind(cred, writable=True)]
    )
    assert (str(cred), str(cred)) in _split(args, "--bind")


def test_wrap_readonly_bind_spliced(tmp_path):
    cred = tmp_path / "cred"
    cred.mkdir()
    args = sandbox.wrap(
        ["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [Bind(cred)]
    )
    assert (str(cred), str(cred)) in _split(args, "--ro-bind")


def test_wrap_skips_missing_binds(tmp_path):
    missing = tmp_path / "nope"
    args = sandbox.wrap(
        ["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [Bind(missing)]
    )
    assert str(missing) not in args
  • Step 2: Run tests to verify they fail

Run: uv run pytest tests/test_sandbox.py -k wrap -q Expected: FAIL — AttributeError: module 'hqt.sandbox' has no attribute 'wrap'.

  • Step 3: Implement wrap()

Append to src/hqt/sandbox.py:

# System paths bound read-only so binaries and shared libraries resolve.
_SYSTEM_RO_PATHS = ["/usr", "/bin", "/sbin", "/lib", "/lib64", "/etc", "/opt"]

# Environment variables passed through into the jail.
_PASSTHROUGH_ENV = ["PATH", "HOME", "TERM", "LANG", "LC_ALL", "USER", "SHELL"]


def wrap(
    command: list[str],
    cwd: Path,
    policy: SandboxPolicy,
    binds: list[Bind],
) -> list[str]:
    """Build the `bwrap … -- <command>` argv for a sandboxed harness.

    Minimal-allowlist model: $HOME and other user data are NOT exposed. System
    dirs are read-only; the project dir (cwd) and each existing harness bind are
    bound explicitly. Network is all-or-nothing: shared only when policy.net.
    Binds whose source does not exist are skipped (bwrap would otherwise error).
    """
    args: list[str] = ["bwrap", "--die-with-parent", "--unshare-all"]
    if policy.net:
        args.append("--share-net")
    for path in _SYSTEM_RO_PATHS:
        if Path(path).exists():
            args += ["--ro-bind", path, path]
    args += ["--proc", "/proc", "--dev", "/dev", "--tmpfs", "/tmp"]
    for bind in binds:
        if not bind.src.exists():
            continue
        flag = "--bind" if bind.writable else "--ro-bind"
        args += [flag, str(bind.src), str(bind.src)]
    cwd_flag = "--bind" if policy.fs == "rw" else "--ro-bind"
    args += [cwd_flag, str(cwd), str(cwd)]
    for name in _PASSTHROUGH_ENV:
        value = os.environ.get(name)
        if value is not None:
            args += ["--setenv", name, value]
    args.append("--")
    args += command
    return args
  • Step 4: Run tests to verify they pass

Run: uv run pytest tests/test_sandbox.py -q Expected: PASS (all wrap tests + Task 1 tests).

  • Step 5: Commit
git add src/hqt/sandbox.py tests/test_sandbox.py
git commit -m "feat: sandbox.wrap builds bwrap argv from policy and binds"

Task 3: Configurator ABC — skip-permission flags and binds

Files:

  • Modify: src/hqt/harnesses/base.py

  • Modify: src/hqt/harnesses/configurators/claude.py:24-38

  • Modify: src/hqt/harnesses/configurators/codex.py:22-36

  • Modify: src/hqt/harnesses/configurators/kiro.py:13-21

  • Modify: src/hqt/harnesses/configurators/generic.py:16-24

  • Test: tests/test_harnesses.py

  • Step 1: Write the failing tests

# tests/test_harnesses.py  (append)
from pathlib import Path

from hqt.harnesses.base import HarnessConfigurator
from hqt.harnesses.configurators.kiro import KiroConfigurator
from hqt.harnesses.configurators.generic import GenericConfigurator


def test_base_skip_flags_default_empty():
    assert HarnessConfigurator.sandbox_skip_permission_flags == []


def test_base_sandbox_binds_default_empty():
    assert KiroConfigurator().sandbox_binds() == []


def test_claude_skip_flag_added_when_sandboxed():
    c = ClaudeConfigurator()
    sid = c.generate_session_id(1)
    cfg = c.build_spawn_config(Path("/tmp"), sid, sandboxed=True)
    assert "--dangerously-skip-permissions" in cfg.command


def test_claude_skip_flag_absent_when_not_sandboxed():
    c = ClaudeConfigurator()
    sid = c.generate_session_id(1)
    cfg = c.build_spawn_config(Path("/tmp"), sid, sandboxed=False)
    assert "--dangerously-skip-permissions" not in cfg.command


def test_claude_resume_skip_flag_when_sandboxed():
    c = ClaudeConfigurator()
    sid = c.generate_session_id(1)
    cfg = c.build_resume_config(Path("/tmp"), sid, sandboxed=True)
    assert "--dangerously-skip-permissions" in cfg.command


def test_claude_sandbox_binds_includes_dot_claude():
    c = ClaudeConfigurator()
    binds = c.sandbox_binds()
    assert any(b.src == Path.home() / ".claude" and b.writable for b in binds)


def test_codex_skip_flag_added_when_sandboxed():
    c = CodexConfigurator()
    cfg = c.build_spawn_config(Path("/tmp"), "1", sandboxed=True)
    assert "--dangerously-bypass-approvals-and-sandbox" in cfg.command


def test_generic_sandboxed_is_noop_flagwise():
    g = GenericConfigurator(["mytool"])
    cfg = g.build_spawn_config(Path("/tmp"), "1", sandboxed=True)
    assert cfg.command == ["mytool"]
  • Step 2: Run tests to verify they fail

Run: uv run pytest tests/test_harnesses.py -k "sandbox or skip or sandboxed or binds" -q Expected: FAIL — TypeError: build_spawn_config() got an unexpected keyword argument 'sandboxed' / missing attributes.

  • Step 3: Update the ABC

In src/hqt/harnesses/base.py, add the import and the two members, and add sandboxed to both abstract signatures:

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path

from hqt.sandbox import Bind


@dataclass
class SpawnConfig:
    command: list[str]
    env: dict[str, str] = field(default_factory=dict)
    cwd: Path = Path(".")


class HarnessConfigurator(ABC):
    name: str
    display_name: str
    captures_session_id: bool = False
    # Flags appended to the harness command when the session is sandboxed.
    sandbox_skip_permission_flags: list[str] = []

    @abstractmethod
    def generate_session_id(self, hqt_session_id: int) -> str: ...

    @abstractmethod
    def build_spawn_config(
        self,
        project_path: Path,
        harness_session_id: str,
        model: str | None = None,
        sandboxed: bool = False,
    ) -> SpawnConfig: ...

    @abstractmethod
    def build_resume_config(
        self,
        project_path: Path,
        harness_session_id: str,
        model: str | None = None,
        sandboxed: bool = False,
    ) -> SpawnConfig: ...

    def sandbox_binds(self) -> list[Bind]:
        """Host paths this harness needs inside the jail (credentials, config)."""
        return []

    def capture_session_id(self, project_path: Path, since: float) -> str | None:
        return None

    def parse_status(self, pane_text: str) -> str | None:
        """Parse visible pane text to determine harness status.

        Returns "working", "waiting", or None (unknown — caller uses activity-based fallback).
        Default implementation returns None for all harnesses that don't override.
        """
        return None
  • Step 4: Update Claude configurator

Replace the body of src/hqt/harnesses/configurators/claude.py build methods and add binds. Add from hqt.sandbox import Bind to the imports:

class ClaudeConfigurator(HarnessConfigurator):
    name = "claude"
    display_name = "Claude Code"
    sandbox_skip_permission_flags = ["--dangerously-skip-permissions"]

    # ... _WORKING_MARKERS / _WAITING_MARKERS unchanged ...

    def sandbox_binds(self) -> list[Bind]:
        return [Bind(Path.home() / ".claude", writable=True)]

    def build_spawn_config(
        self,
        project_path: Path,
        harness_session_id: str,
        model: str | None = None,
        sandboxed: bool = False,
    ) -> SpawnConfig:
        cmd = ["claude", "--session-id", harness_session_id]
        if model:
            cmd += ["--model", model]
        if sandboxed:
            cmd += self.sandbox_skip_permission_flags
        return SpawnConfig(command=cmd, cwd=project_path)

    def build_resume_config(
        self,
        project_path: Path,
        harness_session_id: str,
        model: str | None = None,
        sandboxed: bool = False,
    ) -> SpawnConfig:
        cmd = ["claude", "--resume", harness_session_id]
        if model:
            cmd += ["--model", model]
        if sandboxed:
            cmd += self.sandbox_skip_permission_flags
        return SpawnConfig(command=cmd, cwd=project_path)
  • Step 5: Update Codex configurator

In src/hqt/harnesses/configurators/codex.py add from hqt.sandbox import Bind, set the flag, add binds, and thread sandboxed:

class CodexConfigurator(HarnessConfigurator):
    name = "codex"
    display_name = "Codex"
    captures_session_id = True
    sandbox_skip_permission_flags = ["--dangerously-bypass-approvals-and-sandbox"]

    def sandbox_binds(self) -> list[Bind]:
        return [Bind(Path.home() / ".codex", writable=True)]

    def build_spawn_config(
        self,
        project_path: Path,
        harness_session_id: str,
        model: str | None = None,
        sandboxed: bool = False,
    ) -> SpawnConfig:
        cmd = ["codex"]
        if model:
            cmd += ["-m", model]
        if sandboxed:
            cmd += self.sandbox_skip_permission_flags
        return SpawnConfig(command=cmd, cwd=project_path)

    def build_resume_config(
        self,
        project_path: Path,
        harness_session_id: str,
        model: str | None = None,
        sandboxed: bool = False,
    ) -> SpawnConfig:
        cmd = ["codex", "resume", harness_session_id]
        if model:
            cmd += ["-m", model]
        if sandboxed:
            cmd += self.sandbox_skip_permission_flags
        return SpawnConfig(command=cmd, cwd=project_path)

(Leave capture_session_id, parse_status, and _meta_timestamp unchanged.)

  • Step 6: Update Kiro and Generic configurators

src/hqt/harnesses/configurators/kiro.py — add from hqt.sandbox import Bind, declare binds, accept sandboxed (no flag to append; kiro has none):

class KiroConfigurator(HarnessConfigurator):
    name = "kiro"
    display_name = "Kiro"

    def sandbox_binds(self) -> list[Bind]:
        return [Bind(Path.home() / ".kiro", writable=True)]

    def generate_session_id(self, hqt_session_id: int) -> str:
        return str(hqt_session_id)

    def build_spawn_config(
        self,
        project_path: Path,
        harness_session_id: str,
        model: str | None = None,
        sandboxed: bool = False,
    ) -> SpawnConfig:
        return SpawnConfig(command=["kiro-cli", "chat"], cwd=project_path)

    def build_resume_config(
        self,
        project_path: Path,
        harness_session_id: str,
        model: str | None = None,
        sandboxed: bool = False,
    ) -> SpawnConfig:
        return SpawnConfig(command=["kiro-cli", "chat"], cwd=project_path)

src/hqt/harnesses/configurators/generic.py — accept sandboxed (no flags, no binds):

    def build_spawn_config(
        self,
        project_path: Path,
        harness_session_id: str,
        model: str | None = None,
        sandboxed: bool = False,
    ) -> SpawnConfig:
        return SpawnConfig(command=self._command, cwd=project_path)

    def build_resume_config(
        self,
        project_path: Path,
        harness_session_id: str,
        model: str | None = None,
        sandboxed: bool = False,
    ) -> SpawnConfig:
        return SpawnConfig(command=self._command, cwd=project_path)
  • Step 7: Run tests to verify they pass

Run: uv run pytest tests/test_harnesses.py -q Expected: PASS (existing + new). Existing positional calls like build_spawn_config(Path("/tmp"), sid, model="opus") still work because sandboxed defaults to False.

  • Step 8: Commit
git add src/hqt/harnesses tests/test_harnesses.py
git commit -m "feat: configurators declare sandbox skip-permission flags and binds"

Task 4: Persist the policy — Session.sandbox_json column + migration

Files:

  • Modify: src/hqt/db/models.py:32-45

  • Modify: src/hqt/db/migrations.py:15

  • Test: tests/test_db.py

  • Step 1: Write the failing tests

# tests/test_db.py  (append)
def test_latest_version_is_two():
    assert migrations.LATEST_VERSION == 2


def test_migration_adds_sandbox_json_column(tmp_path):
    # Simulate a baseline DB whose sessions table predates the column.
    settings = _tmp_settings(tmp_path)
    settings.db_path.parent.mkdir(parents=True, exist_ok=True)
    engine = get_engine(settings)
    with engine.begin() as conn:
        conn.exec_driver_sql(
            "CREATE TABLE sessions ("
            "id INTEGER PRIMARY KEY, project_id INTEGER, harness_id INTEGER, "
            "tmux_session_name TEXT)"
        )
        conn.exec_driver_sql("PRAGMA user_version = 1")
    migrate(engine)
    cols = [c["name"] for c in inspect(engine).get_columns("sessions")]
    assert "sandbox_json" in cols
    assert _user_version(engine) == 2


def test_session_round_trips_sandbox_json(tmp_path):
    settings = _tmp_settings(tmp_path)
    ensure_db(settings)
    factory = get_session_factory(get_engine(settings))
    with factory() as session:
        p = Project(name="proj", path="/tmp/proj")
        h = Harness(name="claude")
        session.add_all([p, h])
        session.commit()
        s = Session(
            project_id=p.id,
            harness_id=h.id,
            tmux_session_name="hqt-9",
            sandbox_json='{"fs": "rw", "net": true}',
        )
        session.add(s)
        session.commit()
        assert session.get(Session, s.id).sandbox_json == '{"fs": "rw", "net": true}'
  • Step 2: Run tests to verify they fail

Run: uv run pytest tests/test_db.py -k "sandbox or version_is_two" -q Expected: FAIL — LATEST_VERSION == 1, no sandbox_json attribute/column.

  • Step 3: Add the model column

In src/hqt/db/models.py, add to Session after model:

    model: Mapped[str | None] = mapped_column(default=None)
    sandbox_json: Mapped[str | None] = mapped_column(default=None)
  • Step 4: Add the migration

In src/hqt/db/migrations.py, replace the empty MIGRATIONS list:

MIGRATIONS: list[tuple[int, Callable[[Connection], None]]] = [
    (
        2,
        lambda conn: conn.exec_driver_sql(
            "ALTER TABLE sessions ADD COLUMN sandbox_json TEXT"
        ),
    ),
]
  • Step 5: Run tests to verify they pass

Run: uv run pytest tests/test_db.py -q Expected: PASS (existing migration tests still pass; LATEST_VERSION is now 2 and fresh DBs are stamped 2).

  • Step 6: Commit
git add src/hqt/db/models.py src/hqt/db/migrations.py tests/test_db.py
git commit -m "feat: add Session.sandbox_json column and migration"

Task 5: Service integration — wrap on spawn and resume

Files:

  • Modify: src/hqt/sessions/service.py

  • Test: tests/test_sessions.py

  • Step 1: Write the failing tests

# tests/test_sessions.py  (append)
from hqt.sandbox import SandboxPolicy


@pytest.fixture
def sandbox_harness():
    h = MagicMock()
    h.captures_session_id = False
    h.generate_session_id.return_value = "sess-1"
    h.build_spawn_config.return_value = MagicMock(
        command=["claude", "--dangerously-skip-permissions"],
        env={},
        cwd=Path("/tmp/myproj"),
    )
    h.sandbox_binds.return_value = []
    h.parse_status = MagicMock(return_value=None)
    return {"claude-code": h}


@pytest.fixture
def sandbox_service(factory, db, tmux, sandbox_harness):
    return SessionService(factory=factory, tmux=tmux, harnesses=sandbox_harness)


@pytest.mark.asyncio
async def test_sandboxed_create_wraps_command(sandbox_service, db, tmux, sandbox_harness):
    with patch("hqt.sessions.service.sandbox.is_available", return_value=True), patch(
        "hqt.sessions.service.sandbox.wrap", return_value=["bwrap", "--", "claude"]
    ) as mock_wrap:
        result = await sandbox_service.create_session(
            project_id=1,
            harness_name="claude-code",
            sandbox=SandboxPolicy(fs="rw", net=True),
        )
    mock_wrap.assert_called_once()
    sandbox_harness["claude-code"].build_spawn_config.assert_called_once()
    # build_spawn_config was called with sandboxed=True
    assert sandbox_harness["claude-code"].build_spawn_config.call_args.kwargs[
        "sandboxed"
    ] is True
    # tmux.spawn received the wrapped command
    assert tmux.spawn.call_args.args[0].command == ["bwrap", "--", "claude"]
    # policy persisted on the row
    assert result.session.sandbox_json == '{"fs": "rw", "net": true}'


@pytest.mark.asyncio
async def test_unsandboxed_create_does_not_wrap(service, db, tmux, harnesses):
    with patch("hqt.sessions.service.sandbox.wrap") as mock_wrap:
        result = await service.create_session(project_id=1, harness_name="claude-code")
    mock_wrap.assert_not_called()
    assert result.session.sandbox_json is None


@pytest.mark.asyncio
async def test_sandboxed_create_raises_when_bwrap_missing(sandbox_service, db):
    with patch("hqt.sessions.service.sandbox.is_available", return_value=False):
        with pytest.raises(ServiceError, match="bubblewrap"):
            await sandbox_service.create_session(
                project_id=1,
                harness_name="claude-code",
                sandbox=SandboxPolicy(fs="rw", net=True),
            )
  • Step 2: Run tests to verify they fail

Run: uv run pytest tests/test_sessions.py -k "sandbox" -q Expected: FAIL — create_session() got an unexpected keyword argument 'sandbox'.

  • Step 3: Add imports and helpers to the service

In src/hqt/sessions/service.py, add near the top imports:

import json

from hqt import sandbox
from hqt.sandbox import SandboxPolicy

Add these helpers to SessionService (e.g. after _project_path):

    def _sandbox_policy(self, sess: Session) -> SandboxPolicy | None:
        """Decode the persisted sandbox policy, or None if the session is plain."""
        if not sess.sandbox_json:
            return None
        data = json.loads(sess.sandbox_json)
        return SandboxPolicy(fs=data["fs"], net=data["net"])

    def _wrap_command(
        self,
        configurator: HarnessConfigurator,
        command: list[str],
        cwd: Path,
        policy: SandboxPolicy | None,
    ) -> list[str]:
        """Wrap a harness command in bwrap when a policy is set.

        Raises ServiceError if a sandbox is required but bwrap is unavailable —
        a backstop; the dialog already disables the toggle in that case.
        """
        if policy is None:
            return command
        if not sandbox.is_available():
            raise ServiceError(
                "bubblewrap (bwrap) is not available; cannot start a sandboxed session"
            )
        return sandbox.wrap(command, cwd, policy, configurator.sandbox_binds())
  • Step 4: Thread sandbox through create_session

Change the signature and the spawn-config construction in create_session:

    async def create_session(
        self,
        project_id: int,
        harness_name: str,
        nickname: str | None = None,
        model: str | None = None,
        sandbox: SandboxPolicy | None = None,
    ) -> "CreateSessionResult":

Inside, after computing harness_session_id, persist the policy, build the spawn config with sandboxed=, and wrap. The wrapping (and the hard "bwrap missing" failure) is delegated to _wrap_command, so this method never references the sandbox module directly — important because the parameter sandbox shadows it within this method body:

            sess.sandbox_json = (
                json.dumps({"fs": sandbox.fs, "net": sandbox.net})
                if sandbox is not None
                else None
            )
            spawn_cfg = configurator.build_spawn_config(
                project_path, harness_session_id, model, sandboxed=sandbox is not None
            )
            command = self._wrap_command(
                configurator, spawn_cfg.command, spawn_cfg.cwd, sandbox
            )

Then pass command (not spawn_cfg.command) into the SpawnRequest:

            spawn_result = await self.tmux.spawn(
                SpawnRequest(
                    window_name=sess.tmux_session_name,
                    command=command,
                    cwd=str(spawn_cfg.cwd),
                    env=spawn_cfg.env,
                )
            )

Why no separate availability pre-check here: _wrap_command already raises ServiceError when a policy is set and bwrap is unavailable. Because the row is added/flushed but not committed before _wrap_command runs, that exception propagates out of the with self.factory() as db: block without a commit, so no partial session row is persisted. Keeping the single check inside _wrap_command also means it resolves the sandbox module in its own scope (its only parameter is policy), so patch("hqt.sessions.service.sandbox.is_available", ...) in tests takes effect — a module-level alias would not.

  • Step 5: Wrap the resume and fallback paths

Update _get_resume_config to read the policy, pass sandboxed=, and wrap:

    def _get_resume_config(self, db: DBSession, sess: Session) -> SpawnConfig:
        configurator = self._configurator(sess.harness.name)
        harness_session_id = (
            sess.harness_session_id or configurator.generate_session_id(sess.id)
        )
        project_path = self._project_path(db, sess.project_id)
        policy = self._sandbox_policy(sess)
        cfg = configurator.build_resume_config(
            project_path, harness_session_id, sess.model, sandboxed=policy is not None
        )
        command = self._wrap_command(configurator, cfg.command, cfg.cwd, policy)
        return SpawnConfig(command=command, env=cfg.env, cwd=cfg.cwd)

In _respawn_with_fallback, the rung-2 fresh spawn must also wrap. Replace the rung-2 block that builds spawn_cfg and calls respawn_verified:

        configurator = self._configurator(sess.harness.name)
        project_path = self._project_path(db, sess.project_id)
        policy = self._sandbox_policy(sess)
        harness_session_id = (
            sess.harness_session_id or configurator.generate_session_id(sess.id)
        )
        spawn_cfg = configurator.build_spawn_config(
            project_path, harness_session_id, sess.model, sandboxed=policy is not None
        )
        command = self._wrap_command(
            configurator, spawn_cfg.command, spawn_cfg.cwd, policy
        )
        since = time.time()
        result = await self.tmux.respawn_verified(
            window_name, command, str(spawn_cfg.cwd), env=spawn_cfg.env
        )

(The rung-1 resume already goes through _get_resume_config, which now wraps.)

  • Step 6: Run tests to verify they pass

Run: uv run pytest tests/test_sessions.py -q Expected: PASS — new sandbox tests pass and the existing test_create_session etc. still pass (the harnesses fixture's build_spawn_config accepts the extra sandboxed kwarg via MagicMock, and sandbox defaults to None so _wrap_command returns the command unchanged).

  • Step 7: Run the full suite + lint

Run: uv run pytest -q && uv run ruff check src tests Expected: PASS / no lint errors.

  • Step 8: Commit
git add src/hqt/sessions/service.py tests/test_sessions.py
git commit -m "feat: wrap sandboxed sessions in bwrap on spawn and resume"

Task 6: New Session dialog — toggles, gating, and app wiring

Files:

  • Modify: src/hqt/tui/screens/new_session.py

  • Modify: src/hqt/tui/app.py:239-274

  • Test: tests/test_tui.py

  • Step 1: Write the failing tests

# tests/test_tui.py  (append)
import pytest
from textual.app import App
from textual.widgets import Switch

from hqt.sandbox import SandboxPolicy
from hqt.tui.screens.new_session import NewSessionScreen


def test_policy_from_disabled_returns_none():
    assert NewSessionScreen._policy_from(False, "rw", True) is None


def test_policy_from_enabled_builds_policy():
    p = NewSessionScreen._policy_from(True, "ro", False)
    assert p == SandboxPolicy(fs="ro", net=False)


class _Host(App):
    def __init__(self, screen: NewSessionScreen) -> None:
        self._scr = screen
        super().__init__()

    async def on_mount(self) -> None:
        await self.push_screen(self._scr)


@pytest.mark.asyncio
async def test_sandbox_switch_disabled_when_unavailable():
    screen = NewSessionScreen(["claude"], sandbox_available=False)
    async with _Host(screen).run_test():
        assert screen.query_one("#sandbox-switch", Switch).disabled is True


@pytest.mark.asyncio
async def test_sandbox_switch_enabled_when_available():
    screen = NewSessionScreen(["claude"], sandbox_available=True)
    async with _Host(screen).run_test():
        assert screen.query_one("#sandbox-switch", Switch).disabled is False
  • Step 2: Run tests to verify they fail

Run: uv run pytest tests/test_tui.py -k "policy_from or sandbox_switch" -q Expected: FAIL — NewSessionScreen has no _policy_from and its __init__ rejects sandbox_available.

  • Step 3: Rewrite the dialog

Replace src/hqt/tui/screens/new_session.py with:

from textual.app import ComposeResult
from textual.containers import Horizontal, Vertical
from textual.screen import ModalScreen
from textual.widgets import Button, Input, Label, Select, Switch

from hqt.sandbox import SandboxPolicy

SessionResult = tuple[str, str, str | None, SandboxPolicy | None]


class NewSessionScreen(ModalScreen[SessionResult | None]):
    def __init__(
        self, harness_names: list[str], sandbox_available: bool = False
    ) -> None:
        self.harness_names = harness_names
        self.sandbox_available = sandbox_available
        super().__init__()

    def compose(self) -> ComposeResult:
        with Vertical(id="new-session-dialog"):
            yield Label("New Session")
            yield Label("Harness:")
            yield Select(
                [(n, n) for n in self.harness_names],
                id="harness-select",
                allow_blank=False,
            )
            yield Label("Nickname:")
            yield Input(placeholder="session nickname", id="nickname-input")
            yield Label("Model (optional):")
            yield Input(placeholder="model name", id="model-input")
            with Horizontal(classes="sandbox-row"):
                yield Label("Sandboxed:")
                yield Switch(value=False, id="sandbox-switch", disabled=not self.sandbox_available)
            if not self.sandbox_available:
                yield Label(
                    "bubblewrap not found — run `hqt doctor`",
                    id="sandbox-warning",
                )
            yield Label("Filesystem:")
            yield Select(
                [("Read-write", "rw"), ("Read-only", "ro")],
                id="fs-select",
                value="rw",
                allow_blank=False,
            )
            with Horizontal(classes="sandbox-row"):
                yield Label("Network:")
                yield Switch(value=True, id="net-switch")
            with Horizontal(classes="dialog-actions"):
                yield Button("OK", variant="primary", id="ok-btn")
                yield Button("Cancel", id="cancel-btn")

    @staticmethod
    def _policy_from(enabled: bool, fs: str, net: bool) -> SandboxPolicy | None:
        """Pure mapping from widget values to a policy (None when disabled)."""
        if not enabled:
            return None
        return SandboxPolicy(fs=fs, net=net)

    def on_button_pressed(self, event: Button.Pressed) -> None:
        if event.button.id == "ok-btn":
            self._submit()
        else:
            self.dismiss(None)

    def on_input_submitted(self, event: Input.Submitted) -> None:
        self._submit()

    def _submit(self) -> None:
        harness = self.query_one("#harness-select", Select).value
        nickname = self.query_one("#nickname-input", Input).value
        model = self.query_one("#model-input", Input).value or None
        enabled = self.query_one("#sandbox-switch", Switch).value
        fs = self.query_one("#fs-select", Select).value
        net = self.query_one("#net-switch", Switch).value
        policy = self._policy_from(bool(enabled), str(fs), bool(net))
        if harness and harness != Select.BLANK:
            self.dismiss((str(harness), nickname, model, policy))
        else:
            self.dismiss(None)
  • Step 4: Wire the app

In src/hqt/tui/app.py, add from hqt import sandbox to the imports, then update action_new_session:

    def action_new_session(self) -> None:
        if self._selected_project_id is None:
            self.notify("Select a project first", severity="warning")
            return
        project_id = self._selected_project_id
        harness_names = list(discover_harnesses().keys())
        if not harness_names:
            self.notify("No harnesses found", severity="error")
            return

        def on_dismiss(result: tuple[str, str, str | None, object] | None) -> None:
            if result:
                harness_name, nickname, model, sandbox_policy = result

                async def _do() -> None:
                    create_result = await self._sessions().create_session(
                        project_id, harness_name, nickname, model, sandbox=sandbox_policy
                    )
                    if not create_result.spawn_ok:
                        first_line = next(
                            (
                                ln
                                for ln in create_result.spawn_error.splitlines()
                                if ln.strip()
                            ),
                            create_result.spawn_error,
                        )[:120]
                        self.notify(
                            f"Session created but harness failed to start: {first_line}",
                            severity="error",
                        )
                    await self._refresh_sessions()

                self._run_service_worker(_do())

        self.push_screen(
            NewSessionScreen(harness_names, sandbox_available=sandbox.is_available()),
            on_dismiss,
        )
  • Step 5: Run tests to verify they pass

Run: uv run pytest tests/test_tui.py -q Expected: PASS (new dialog tests + existing TUI tests).

  • Step 6: Commit
git add src/hqt/tui/screens/new_session.py src/hqt/tui/app.py tests/test_tui.py
git commit -m "feat: sandbox toggles in New Session dialog, gated on bwrap availability"

Task 7: doctor reports bubblewrap

Files:

  • Modify: src/hqt/cli.py:105-122

  • Test: tests/test_cli.py (create if absent)

  • Step 1: Write the failing tests

# tests/test_cli.py
from unittest.mock import patch

from click.testing import CliRunner

from hqt.cli import main


def _which(found: set[str]):
    return lambda name: f"/usr/bin/{name}" if name in found else None


def test_doctor_reports_bwrap_present():
    with patch("hqt.cli.shutil.which", side_effect=_which({"tmux", "bwrap"})), patch(
        "hqt.harnesses.registry.shutil.which", side_effect=_which(set())
    ):
        result = CliRunner().invoke(main, ["doctor"])
    assert result.exit_code == 0
    assert "bubblewrap: ✓" in result.output


def test_doctor_reports_bwrap_missing():
    with patch("hqt.cli.shutil.which", side_effect=_which({"tmux"})), patch(
        "hqt.harnesses.registry.shutil.which", side_effect=_which(set())
    ):
        result = CliRunner().invoke(main, ["doctor"])
    assert result.exit_code == 0
    assert "bubblewrap: ✗" in result.output
  • Step 2: Run tests to verify they fail

Run: uv run pytest tests/test_cli.py -q Expected: FAIL — output contains no "bubblewrap" line.

  • Step 3: Add the doctor check

In src/hqt/cli.py, inside doctor(), after the harness loop (before/after is fine), add a non-fatal bubblewrap report:

    bwrap = shutil.which("bwrap")
    if bwrap:
        click.echo(f"bubblewrap: ✓ {bwrap}")
    else:
        click.echo("bubblewrap: ✗ not found — sandboxed sessions unavailable")
  • Step 4: Run tests to verify they pass

Run: uv run pytest tests/test_cli.py -q Expected: PASS (2 passed).

  • Step 5: Full suite + lint

Run: uv run pytest -q && uv run ruff check src tests Expected: PASS / clean.

  • Step 6: Commit
git add src/hqt/cli.py tests/test_cli.py
git commit -m "feat: doctor reports bubblewrap availability"

Self-Review (completed by plan author)

Spec coverage:

  • Data model (sandbox_json + migration) → Task 4.
  • ABC additions (skip-permission flags, sandbox_binds, sandboxed param) → Task 3.
  • Bubblewrap policy module (is_available, wrap) → Tasks 12.
  • Spawn/resume integration (create + both respawn rungs + _get_resume_config) → Task 5.
  • UI toggles + gating → Task 6.
  • doctor + availability helper as single source of truth → Tasks 1 & 7 (both use is_available / shutil.which("bwrap")).
  • Failure handling (hard ServiceError, no silent downgrade) → Task 5 (_wrap_command and create-time check).
  • Testing (wrap combos, configurator flags, service wrap + bwrap-missing, availability/UI gating, migration) → Tasks 17.

Network all-or-nothing & minimal-allowlist are realized in wrap() (Task 2): --share-net only when net, $HOME not bound, system dirs read-only, explicit cwd + harness binds.

Placeholder scan: none — every code step shows complete code; harness flags/paths are concrete (verify against real binaries when implementing, per the spec note).

Type/name consistency: SandboxPolicy(fs, net), Bind(src, writable), is_available(), wrap(command, cwd, policy, binds), sandbox_binds(), sandbox_skip_permission_flags, _sandbox_policy, _wrap_command, _policy_from, result tuple (harness, nickname, model, policy) are used identically across tasks.