diff --git a/docs/superpowers/plans/2026-06-10-sandboxed-sessions.md b/docs/superpowers/plans/2026-06-10-sandboxed-sessions.md new file mode 100644 index 0000000..5bf602f --- /dev/null +++ b/docs/superpowers/plans/2026-06-10-sandboxed-sessions.md @@ -0,0 +1,1206 @@ +# 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 … -- ` argv from a `SandboxPolicy` plus per-harness `Bind`s. 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.py` — `Bind`, `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.py` — `create_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.py` — `doctor` 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** + +```python +# 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** + +```python +# 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** + +```bash +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** + +```python +# 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`: + +```python +# 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 … -- ` 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** + +```bash +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** + +```python +# 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: + +```python +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: + +```python +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`: + +```python +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): + +```python +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): + +```python + 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** + +```bash +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** + +```python +# 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`: + +```python + 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: + +```python +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** + +```bash +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** + +```python +# 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: + +```python +import json + +from hqt import sandbox +from hqt.sandbox import SandboxPolicy +``` + +Add these helpers to `SessionService` (e.g. after `_project_path`): + +```python + 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`: + +```python + 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: + +```python + 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`: + +```python + 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: + +```python + 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`: + +```python + 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** + +```bash +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** + +```python +# 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: + +```python +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`: + +```python + 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** + +```bash +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** + +```python +# 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: + +```python + 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** + +```bash +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 1–2. +- 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 1–7. + +**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. +```