1207 lines
41 KiB
Markdown
1207 lines
41 KiB
Markdown
# 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 `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 … -- <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**
|
||
|
||
```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.
|
||
```
|