# 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. ```