Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
41 KiB
Sandboxed Sessions (bubblewrap) Implementation Plan
For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: Let a user start a harness session inside a bubblewrap (bwrap) sandbox, chosen per session, with read-only/read-write project access and an internet on/off toggle; sandboxed sessions automatically get the harness's "skip permissions" flag.
Architecture: A new pure module hqt/sandbox.py builds the bwrap … -- <command> argv from a SandboxPolicy plus per-harness Binds. The HarnessConfigurator ABC gains the per-harness skip-permission flags and the credential/config paths to bind. SessionService wraps the spawn/resume command when a session is sandboxed and persists the policy in a new Session.sandbox_json column. The New Session dialog exposes the toggles and disables them when bwrap is unavailable (the same check hqt doctor uses).
Tech Stack: Python 3.14, SQLAlchemy (SQLite, user_version migrations), Textual (TUI), Click (CLI), pytest + pytest-asyncio.
File Structure
- Create
src/hqt/sandbox.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— addsandbox_skip_permission_flagsclass attr,sandbox_binds()method, and asandboxed: bool = Falseparameter to the two abstract build methods. - Modify
src/hqt/harnesses/configurators/{claude,codex,kiro,generic}.py— acceptsandboxed, append flags, declare binds. - Modify
src/hqt/db/models.py— addSession.sandbox_jsoncolumn. - 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 tocreate_session. - Modify
src/hqt/cli.py—doctorreports bubblewrap availability. - Tests:
tests/test_sandbox.py(new), and additions totests/test_harnesses.py,tests/test_db.py,tests/test_sessions.py,tests/test_tui.py,tests/test_cli.py(new or existing).
Run the full suite with uv run pytest -q and lint with uv run ruff check src tests at checkpoints.
Task 1: Sandbox primitives — Bind, SandboxPolicy, is_available()
Files:
-
Create:
src/hqt/sandbox.py -
Test:
tests/test_sandbox.py -
Step 1: Write the failing tests
# tests/test_sandbox.py
import sys
from pathlib import Path
from hqt import sandbox
from hqt.sandbox import Bind, SandboxPolicy
def test_bind_defaults_readonly():
b = Bind(Path("/home/u/.claude"))
assert b.src == Path("/home/u/.claude")
assert b.writable is False
def test_sandbox_policy_fields():
p = SandboxPolicy(fs="rw", net=True)
assert p.fs == "rw"
assert p.net is True
def test_is_available_true_on_linux_with_bwrap(monkeypatch):
monkeypatch.setattr(sandbox.sys, "platform", "linux")
monkeypatch.setattr(sandbox.shutil, "which", lambda name: "/usr/bin/bwrap")
assert sandbox.is_available() is True
def test_is_available_false_without_bwrap(monkeypatch):
monkeypatch.setattr(sandbox.sys, "platform", "linux")
monkeypatch.setattr(sandbox.shutil, "which", lambda name: None)
assert sandbox.is_available() is False
def test_is_available_false_off_linux(monkeypatch):
monkeypatch.setattr(sandbox.sys, "platform", "darwin")
monkeypatch.setattr(sandbox.shutil, "which", lambda name: "/usr/bin/bwrap")
assert sandbox.is_available() is False
- Step 2: Run tests to verify they fail
Run: uv run pytest tests/test_sandbox.py -q
Expected: FAIL — ModuleNotFoundError: No module named 'hqt.sandbox'.
- Step 3: Create the module with primitives
# src/hqt/sandbox.py
import os
import shutil
import sys
from dataclasses import dataclass
from pathlib import Path
@dataclass(frozen=True)
class Bind:
"""A host path to expose inside the sandbox (read-only unless writable)."""
src: Path
writable: bool = False
@dataclass(frozen=True)
class SandboxPolicy:
"""Per-session sandbox settings.
fs: "rw" binds the project dir writable; "ro" binds it read-only.
net: True shares the host network namespace; False = no connectivity.
"""
fs: str
net: bool
def is_available() -> bool:
"""True when bubblewrap can be used: Linux with `bwrap` on PATH.
The single source of truth for both `hqt doctor` and the New Session dialog.
"""
return sys.platform.startswith("linux") and shutil.which("bwrap") is not None
(os is imported now; Task 2 uses it for env passthrough.)
- Step 4: Run tests to verify they pass
Run: uv run pytest tests/test_sandbox.py -q
Expected: PASS (5 passed).
- Step 5: Commit
git add src/hqt/sandbox.py tests/test_sandbox.py
git commit -m "feat: sandbox primitives — Bind, SandboxPolicy, is_available"
Task 2: wrap() — build the bwrap argv
Files:
-
Modify:
src/hqt/sandbox.py -
Test:
tests/test_sandbox.py -
Step 1: Write the failing tests
# tests/test_sandbox.py (append)
def _split(args, flag):
"""Return list of (src, dst) pairs that follow each occurrence of flag."""
out = []
for i, a in enumerate(args):
if a == flag:
out.append((args[i + 1], args[i + 2]))
return out
def test_wrap_command_after_double_dash(tmp_path):
cmd = ["claude", "--session-id", "x"]
args = sandbox.wrap(cmd, tmp_path, SandboxPolicy(fs="rw", net=True), [])
assert args[0] == "bwrap"
assert "--" in args
assert args[args.index("--") + 1 :] == cmd
def test_wrap_base_flags(tmp_path):
args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [])
assert "--die-with-parent" in args
assert "--unshare-all" in args
assert "--proc" in args and "--dev" in args
def test_wrap_net_on_shares_net(tmp_path):
args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [])
assert "--share-net" in args
def test_wrap_net_off_does_not_share(tmp_path):
args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=False), [])
assert "--share-net" not in args
def test_wrap_cwd_rw_is_bind(tmp_path):
args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [])
assert (str(tmp_path), str(tmp_path)) in _split(args, "--bind")
def test_wrap_cwd_ro_is_ro_bind(tmp_path):
args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="ro", net=True), [])
assert (str(tmp_path), str(tmp_path)) in _split(args, "--ro-bind")
def test_wrap_writable_bind_spliced(tmp_path):
cred = tmp_path / "cred"
cred.mkdir()
args = sandbox.wrap(
["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [Bind(cred, writable=True)]
)
assert (str(cred), str(cred)) in _split(args, "--bind")
def test_wrap_readonly_bind_spliced(tmp_path):
cred = tmp_path / "cred"
cred.mkdir()
args = sandbox.wrap(
["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [Bind(cred)]
)
assert (str(cred), str(cred)) in _split(args, "--ro-bind")
def test_wrap_skips_missing_binds(tmp_path):
missing = tmp_path / "nope"
args = sandbox.wrap(
["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [Bind(missing)]
)
assert str(missing) not in args
- Step 2: Run tests to verify they fail
Run: uv run pytest tests/test_sandbox.py -k wrap -q
Expected: FAIL — AttributeError: module 'hqt.sandbox' has no attribute 'wrap'.
- Step 3: Implement
wrap()
Append to src/hqt/sandbox.py:
# System paths bound read-only so binaries and shared libraries resolve.
_SYSTEM_RO_PATHS = ["/usr", "/bin", "/sbin", "/lib", "/lib64", "/etc", "/opt"]
# Environment variables passed through into the jail.
_PASSTHROUGH_ENV = ["PATH", "HOME", "TERM", "LANG", "LC_ALL", "USER", "SHELL"]
def wrap(
command: list[str],
cwd: Path,
policy: SandboxPolicy,
binds: list[Bind],
) -> list[str]:
"""Build the `bwrap … -- <command>` argv for a sandboxed harness.
Minimal-allowlist model: $HOME and other user data are NOT exposed. System
dirs are read-only; the project dir (cwd) and each existing harness bind are
bound explicitly. Network is all-or-nothing: shared only when policy.net.
Binds whose source does not exist are skipped (bwrap would otherwise error).
"""
args: list[str] = ["bwrap", "--die-with-parent", "--unshare-all"]
if policy.net:
args.append("--share-net")
for path in _SYSTEM_RO_PATHS:
if Path(path).exists():
args += ["--ro-bind", path, path]
args += ["--proc", "/proc", "--dev", "/dev", "--tmpfs", "/tmp"]
for bind in binds:
if not bind.src.exists():
continue
flag = "--bind" if bind.writable else "--ro-bind"
args += [flag, str(bind.src), str(bind.src)]
cwd_flag = "--bind" if policy.fs == "rw" else "--ro-bind"
args += [cwd_flag, str(cwd), str(cwd)]
for name in _PASSTHROUGH_ENV:
value = os.environ.get(name)
if value is not None:
args += ["--setenv", name, value]
args.append("--")
args += command
return args
- Step 4: Run tests to verify they pass
Run: uv run pytest tests/test_sandbox.py -q
Expected: PASS (all wrap tests + Task 1 tests).
- Step 5: Commit
git add src/hqt/sandbox.py tests/test_sandbox.py
git commit -m "feat: sandbox.wrap builds bwrap argv from policy and binds"
Task 3: Configurator ABC — skip-permission flags and binds
Files:
-
Modify:
src/hqt/harnesses/base.py -
Modify:
src/hqt/harnesses/configurators/claude.py:24-38 -
Modify:
src/hqt/harnesses/configurators/codex.py:22-36 -
Modify:
src/hqt/harnesses/configurators/kiro.py:13-21 -
Modify:
src/hqt/harnesses/configurators/generic.py:16-24 -
Test:
tests/test_harnesses.py -
Step 1: Write the failing tests
# tests/test_harnesses.py (append)
from pathlib import Path
from hqt.harnesses.base import HarnessConfigurator
from hqt.harnesses.configurators.kiro import KiroConfigurator
from hqt.harnesses.configurators.generic import GenericConfigurator
def test_base_skip_flags_default_empty():
assert HarnessConfigurator.sandbox_skip_permission_flags == []
def test_base_sandbox_binds_default_empty():
assert KiroConfigurator().sandbox_binds() == []
def test_claude_skip_flag_added_when_sandboxed():
c = ClaudeConfigurator()
sid = c.generate_session_id(1)
cfg = c.build_spawn_config(Path("/tmp"), sid, sandboxed=True)
assert "--dangerously-skip-permissions" in cfg.command
def test_claude_skip_flag_absent_when_not_sandboxed():
c = ClaudeConfigurator()
sid = c.generate_session_id(1)
cfg = c.build_spawn_config(Path("/tmp"), sid, sandboxed=False)
assert "--dangerously-skip-permissions" not in cfg.command
def test_claude_resume_skip_flag_when_sandboxed():
c = ClaudeConfigurator()
sid = c.generate_session_id(1)
cfg = c.build_resume_config(Path("/tmp"), sid, sandboxed=True)
assert "--dangerously-skip-permissions" in cfg.command
def test_claude_sandbox_binds_includes_dot_claude():
c = ClaudeConfigurator()
binds = c.sandbox_binds()
assert any(b.src == Path.home() / ".claude" and b.writable for b in binds)
def test_codex_skip_flag_added_when_sandboxed():
c = CodexConfigurator()
cfg = c.build_spawn_config(Path("/tmp"), "1", sandboxed=True)
assert "--dangerously-bypass-approvals-and-sandbox" in cfg.command
def test_generic_sandboxed_is_noop_flagwise():
g = GenericConfigurator(["mytool"])
cfg = g.build_spawn_config(Path("/tmp"), "1", sandboxed=True)
assert cfg.command == ["mytool"]
- Step 2: Run tests to verify they fail
Run: uv run pytest tests/test_harnesses.py -k "sandbox or skip or sandboxed or binds" -q
Expected: FAIL — TypeError: build_spawn_config() got an unexpected keyword argument 'sandboxed' / missing attributes.
- Step 3: Update the ABC
In src/hqt/harnesses/base.py, add the import and the two members, and add sandboxed to both abstract signatures:
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from hqt.sandbox import Bind
@dataclass
class SpawnConfig:
command: list[str]
env: dict[str, str] = field(default_factory=dict)
cwd: Path = Path(".")
class HarnessConfigurator(ABC):
name: str
display_name: str
captures_session_id: bool = False
# Flags appended to the harness command when the session is sandboxed.
sandbox_skip_permission_flags: list[str] = []
@abstractmethod
def generate_session_id(self, hqt_session_id: int) -> str: ...
@abstractmethod
def build_spawn_config(
self,
project_path: Path,
harness_session_id: str,
model: str | None = None,
sandboxed: bool = False,
) -> SpawnConfig: ...
@abstractmethod
def build_resume_config(
self,
project_path: Path,
harness_session_id: str,
model: str | None = None,
sandboxed: bool = False,
) -> SpawnConfig: ...
def sandbox_binds(self) -> list[Bind]:
"""Host paths this harness needs inside the jail (credentials, config)."""
return []
def capture_session_id(self, project_path: Path, since: float) -> str | None:
return None
def parse_status(self, pane_text: str) -> str | None:
"""Parse visible pane text to determine harness status.
Returns "working", "waiting", or None (unknown — caller uses activity-based fallback).
Default implementation returns None for all harnesses that don't override.
"""
return None
- Step 4: Update Claude configurator
Replace the body of src/hqt/harnesses/configurators/claude.py build methods and add binds. Add from hqt.sandbox import Bind to the imports:
class ClaudeConfigurator(HarnessConfigurator):
name = "claude"
display_name = "Claude Code"
sandbox_skip_permission_flags = ["--dangerously-skip-permissions"]
# ... _WORKING_MARKERS / _WAITING_MARKERS unchanged ...
def sandbox_binds(self) -> list[Bind]:
return [Bind(Path.home() / ".claude", writable=True)]
def build_spawn_config(
self,
project_path: Path,
harness_session_id: str,
model: str | None = None,
sandboxed: bool = False,
) -> SpawnConfig:
cmd = ["claude", "--session-id", harness_session_id]
if model:
cmd += ["--model", model]
if sandboxed:
cmd += self.sandbox_skip_permission_flags
return SpawnConfig(command=cmd, cwd=project_path)
def build_resume_config(
self,
project_path: Path,
harness_session_id: str,
model: str | None = None,
sandboxed: bool = False,
) -> SpawnConfig:
cmd = ["claude", "--resume", harness_session_id]
if model:
cmd += ["--model", model]
if sandboxed:
cmd += self.sandbox_skip_permission_flags
return SpawnConfig(command=cmd, cwd=project_path)
- Step 5: Update Codex configurator
In src/hqt/harnesses/configurators/codex.py add from hqt.sandbox import Bind, set the flag, add binds, and thread sandboxed:
class CodexConfigurator(HarnessConfigurator):
name = "codex"
display_name = "Codex"
captures_session_id = True
sandbox_skip_permission_flags = ["--dangerously-bypass-approvals-and-sandbox"]
def sandbox_binds(self) -> list[Bind]:
return [Bind(Path.home() / ".codex", writable=True)]
def build_spawn_config(
self,
project_path: Path,
harness_session_id: str,
model: str | None = None,
sandboxed: bool = False,
) -> SpawnConfig:
cmd = ["codex"]
if model:
cmd += ["-m", model]
if sandboxed:
cmd += self.sandbox_skip_permission_flags
return SpawnConfig(command=cmd, cwd=project_path)
def build_resume_config(
self,
project_path: Path,
harness_session_id: str,
model: str | None = None,
sandboxed: bool = False,
) -> SpawnConfig:
cmd = ["codex", "resume", harness_session_id]
if model:
cmd += ["-m", model]
if sandboxed:
cmd += self.sandbox_skip_permission_flags
return SpawnConfig(command=cmd, cwd=project_path)
(Leave capture_session_id, parse_status, and _meta_timestamp unchanged.)
- Step 6: Update Kiro and Generic configurators
src/hqt/harnesses/configurators/kiro.py — add from hqt.sandbox import Bind, declare binds, accept sandboxed (no flag to append; kiro has none):
class KiroConfigurator(HarnessConfigurator):
name = "kiro"
display_name = "Kiro"
def sandbox_binds(self) -> list[Bind]:
return [Bind(Path.home() / ".kiro", writable=True)]
def generate_session_id(self, hqt_session_id: int) -> str:
return str(hqt_session_id)
def build_spawn_config(
self,
project_path: Path,
harness_session_id: str,
model: str | None = None,
sandboxed: bool = False,
) -> SpawnConfig:
return SpawnConfig(command=["kiro-cli", "chat"], cwd=project_path)
def build_resume_config(
self,
project_path: Path,
harness_session_id: str,
model: str | None = None,
sandboxed: bool = False,
) -> SpawnConfig:
return SpawnConfig(command=["kiro-cli", "chat"], cwd=project_path)
src/hqt/harnesses/configurators/generic.py — accept sandboxed (no flags, no binds):
def build_spawn_config(
self,
project_path: Path,
harness_session_id: str,
model: str | None = None,
sandboxed: bool = False,
) -> SpawnConfig:
return SpawnConfig(command=self._command, cwd=project_path)
def build_resume_config(
self,
project_path: Path,
harness_session_id: str,
model: str | None = None,
sandboxed: bool = False,
) -> SpawnConfig:
return SpawnConfig(command=self._command, cwd=project_path)
- Step 7: Run tests to verify they pass
Run: uv run pytest tests/test_harnesses.py -q
Expected: PASS (existing + new). Existing positional calls like build_spawn_config(Path("/tmp"), sid, model="opus") still work because sandboxed defaults to False.
- Step 8: Commit
git add src/hqt/harnesses tests/test_harnesses.py
git commit -m "feat: configurators declare sandbox skip-permission flags and binds"
Task 4: Persist the policy — Session.sandbox_json column + migration
Files:
-
Modify:
src/hqt/db/models.py:32-45 -
Modify:
src/hqt/db/migrations.py:15 -
Test:
tests/test_db.py -
Step 1: Write the failing tests
# tests/test_db.py (append)
def test_latest_version_is_two():
assert migrations.LATEST_VERSION == 2
def test_migration_adds_sandbox_json_column(tmp_path):
# Simulate a baseline DB whose sessions table predates the column.
settings = _tmp_settings(tmp_path)
settings.db_path.parent.mkdir(parents=True, exist_ok=True)
engine = get_engine(settings)
with engine.begin() as conn:
conn.exec_driver_sql(
"CREATE TABLE sessions ("
"id INTEGER PRIMARY KEY, project_id INTEGER, harness_id INTEGER, "
"tmux_session_name TEXT)"
)
conn.exec_driver_sql("PRAGMA user_version = 1")
migrate(engine)
cols = [c["name"] for c in inspect(engine).get_columns("sessions")]
assert "sandbox_json" in cols
assert _user_version(engine) == 2
def test_session_round_trips_sandbox_json(tmp_path):
settings = _tmp_settings(tmp_path)
ensure_db(settings)
factory = get_session_factory(get_engine(settings))
with factory() as session:
p = Project(name="proj", path="/tmp/proj")
h = Harness(name="claude")
session.add_all([p, h])
session.commit()
s = Session(
project_id=p.id,
harness_id=h.id,
tmux_session_name="hqt-9",
sandbox_json='{"fs": "rw", "net": true}',
)
session.add(s)
session.commit()
assert session.get(Session, s.id).sandbox_json == '{"fs": "rw", "net": true}'
- Step 2: Run tests to verify they fail
Run: uv run pytest tests/test_db.py -k "sandbox or version_is_two" -q
Expected: FAIL — LATEST_VERSION == 1, no sandbox_json attribute/column.
- Step 3: Add the model column
In src/hqt/db/models.py, add to Session after model:
model: Mapped[str | None] = mapped_column(default=None)
sandbox_json: Mapped[str | None] = mapped_column(default=None)
- Step 4: Add the migration
In src/hqt/db/migrations.py, replace the empty MIGRATIONS list:
MIGRATIONS: list[tuple[int, Callable[[Connection], None]]] = [
(
2,
lambda conn: conn.exec_driver_sql(
"ALTER TABLE sessions ADD COLUMN sandbox_json TEXT"
),
),
]
- Step 5: Run tests to verify they pass
Run: uv run pytest tests/test_db.py -q
Expected: PASS (existing migration tests still pass; LATEST_VERSION is now 2 and fresh DBs are stamped 2).
- Step 6: Commit
git add src/hqt/db/models.py src/hqt/db/migrations.py tests/test_db.py
git commit -m "feat: add Session.sandbox_json column and migration"
Task 5: Service integration — wrap on spawn and resume
Files:
-
Modify:
src/hqt/sessions/service.py -
Test:
tests/test_sessions.py -
Step 1: Write the failing tests
# tests/test_sessions.py (append)
from hqt.sandbox import SandboxPolicy
@pytest.fixture
def sandbox_harness():
h = MagicMock()
h.captures_session_id = False
h.generate_session_id.return_value = "sess-1"
h.build_spawn_config.return_value = MagicMock(
command=["claude", "--dangerously-skip-permissions"],
env={},
cwd=Path("/tmp/myproj"),
)
h.sandbox_binds.return_value = []
h.parse_status = MagicMock(return_value=None)
return {"claude-code": h}
@pytest.fixture
def sandbox_service(factory, db, tmux, sandbox_harness):
return SessionService(factory=factory, tmux=tmux, harnesses=sandbox_harness)
@pytest.mark.asyncio
async def test_sandboxed_create_wraps_command(sandbox_service, db, tmux, sandbox_harness):
with patch("hqt.sessions.service.sandbox.is_available", return_value=True), patch(
"hqt.sessions.service.sandbox.wrap", return_value=["bwrap", "--", "claude"]
) as mock_wrap:
result = await sandbox_service.create_session(
project_id=1,
harness_name="claude-code",
sandbox=SandboxPolicy(fs="rw", net=True),
)
mock_wrap.assert_called_once()
sandbox_harness["claude-code"].build_spawn_config.assert_called_once()
# build_spawn_config was called with sandboxed=True
assert sandbox_harness["claude-code"].build_spawn_config.call_args.kwargs[
"sandboxed"
] is True
# tmux.spawn received the wrapped command
assert tmux.spawn.call_args.args[0].command == ["bwrap", "--", "claude"]
# policy persisted on the row
assert result.session.sandbox_json == '{"fs": "rw", "net": true}'
@pytest.mark.asyncio
async def test_unsandboxed_create_does_not_wrap(service, db, tmux, harnesses):
with patch("hqt.sessions.service.sandbox.wrap") as mock_wrap:
result = await service.create_session(project_id=1, harness_name="claude-code")
mock_wrap.assert_not_called()
assert result.session.sandbox_json is None
@pytest.mark.asyncio
async def test_sandboxed_create_raises_when_bwrap_missing(sandbox_service, db):
with patch("hqt.sessions.service.sandbox.is_available", return_value=False):
with pytest.raises(ServiceError, match="bubblewrap"):
await sandbox_service.create_session(
project_id=1,
harness_name="claude-code",
sandbox=SandboxPolicy(fs="rw", net=True),
)
- Step 2: Run tests to verify they fail
Run: uv run pytest tests/test_sessions.py -k "sandbox" -q
Expected: FAIL — create_session() got an unexpected keyword argument 'sandbox'.
- Step 3: Add imports and helpers to the service
In src/hqt/sessions/service.py, add near the top imports:
import json
from hqt import sandbox
from hqt.sandbox import SandboxPolicy
Add these helpers to SessionService (e.g. after _project_path):
def _sandbox_policy(self, sess: Session) -> SandboxPolicy | None:
"""Decode the persisted sandbox policy, or None if the session is plain."""
if not sess.sandbox_json:
return None
data = json.loads(sess.sandbox_json)
return SandboxPolicy(fs=data["fs"], net=data["net"])
def _wrap_command(
self,
configurator: HarnessConfigurator,
command: list[str],
cwd: Path,
policy: SandboxPolicy | None,
) -> list[str]:
"""Wrap a harness command in bwrap when a policy is set.
Raises ServiceError if a sandbox is required but bwrap is unavailable —
a backstop; the dialog already disables the toggle in that case.
"""
if policy is None:
return command
if not sandbox.is_available():
raise ServiceError(
"bubblewrap (bwrap) is not available; cannot start a sandboxed session"
)
return sandbox.wrap(command, cwd, policy, configurator.sandbox_binds())
- Step 4: Thread
sandboxthroughcreate_session
Change the signature and the spawn-config construction in create_session:
async def create_session(
self,
project_id: int,
harness_name: str,
nickname: str | None = None,
model: str | None = None,
sandbox: SandboxPolicy | None = None,
) -> "CreateSessionResult":
Inside, after computing harness_session_id, persist the policy, build the spawn config with sandboxed=, and wrap. The wrapping (and the hard "bwrap missing" failure) is delegated to _wrap_command, so this method never references the sandbox module directly — important because the parameter sandbox shadows it within this method body:
sess.sandbox_json = (
json.dumps({"fs": sandbox.fs, "net": sandbox.net})
if sandbox is not None
else None
)
spawn_cfg = configurator.build_spawn_config(
project_path, harness_session_id, model, sandboxed=sandbox is not None
)
command = self._wrap_command(
configurator, spawn_cfg.command, spawn_cfg.cwd, sandbox
)
Then pass command (not spawn_cfg.command) into the SpawnRequest:
spawn_result = await self.tmux.spawn(
SpawnRequest(
window_name=sess.tmux_session_name,
command=command,
cwd=str(spawn_cfg.cwd),
env=spawn_cfg.env,
)
)
Why no separate availability pre-check here: _wrap_command already raises
ServiceError when a policy is set and bwrap is unavailable. Because the row is
added/flushed but not committed before _wrap_command runs, that exception
propagates out of the with self.factory() as db: block without a commit, so no
partial session row is persisted. Keeping the single check inside
_wrap_command also means it resolves the sandbox module in its own scope
(its only parameter is policy), so patch("hqt.sessions.service.sandbox.is_available", ...)
in tests takes effect — a module-level alias would not.
- Step 5: Wrap the resume and fallback paths
Update _get_resume_config to read the policy, pass sandboxed=, and wrap:
def _get_resume_config(self, db: DBSession, sess: Session) -> SpawnConfig:
configurator = self._configurator(sess.harness.name)
harness_session_id = (
sess.harness_session_id or configurator.generate_session_id(sess.id)
)
project_path = self._project_path(db, sess.project_id)
policy = self._sandbox_policy(sess)
cfg = configurator.build_resume_config(
project_path, harness_session_id, sess.model, sandboxed=policy is not None
)
command = self._wrap_command(configurator, cfg.command, cfg.cwd, policy)
return SpawnConfig(command=command, env=cfg.env, cwd=cfg.cwd)
In _respawn_with_fallback, the rung-2 fresh spawn must also wrap. Replace the rung-2 block that builds spawn_cfg and calls respawn_verified:
configurator = self._configurator(sess.harness.name)
project_path = self._project_path(db, sess.project_id)
policy = self._sandbox_policy(sess)
harness_session_id = (
sess.harness_session_id or configurator.generate_session_id(sess.id)
)
spawn_cfg = configurator.build_spawn_config(
project_path, harness_session_id, sess.model, sandboxed=policy is not None
)
command = self._wrap_command(
configurator, spawn_cfg.command, spawn_cfg.cwd, policy
)
since = time.time()
result = await self.tmux.respawn_verified(
window_name, command, str(spawn_cfg.cwd), env=spawn_cfg.env
)
(The rung-1 resume already goes through _get_resume_config, which now wraps.)
- Step 6: Run tests to verify they pass
Run: uv run pytest tests/test_sessions.py -q
Expected: PASS — new sandbox tests pass and the existing test_create_session etc. still pass (the harnesses fixture's build_spawn_config accepts the extra sandboxed kwarg via MagicMock, and sandbox defaults to None so _wrap_command returns the command unchanged).
- Step 7: Run the full suite + lint
Run: uv run pytest -q && uv run ruff check src tests
Expected: PASS / no lint errors.
- Step 8: Commit
git add src/hqt/sessions/service.py tests/test_sessions.py
git commit -m "feat: wrap sandboxed sessions in bwrap on spawn and resume"
Task 6: New Session dialog — toggles, gating, and app wiring
Files:
-
Modify:
src/hqt/tui/screens/new_session.py -
Modify:
src/hqt/tui/app.py:239-274 -
Test:
tests/test_tui.py -
Step 1: Write the failing tests
# tests/test_tui.py (append)
import pytest
from textual.app import App
from textual.widgets import Switch
from hqt.sandbox import SandboxPolicy
from hqt.tui.screens.new_session import NewSessionScreen
def test_policy_from_disabled_returns_none():
assert NewSessionScreen._policy_from(False, "rw", True) is None
def test_policy_from_enabled_builds_policy():
p = NewSessionScreen._policy_from(True, "ro", False)
assert p == SandboxPolicy(fs="ro", net=False)
class _Host(App):
def __init__(self, screen: NewSessionScreen) -> None:
self._scr = screen
super().__init__()
async def on_mount(self) -> None:
await self.push_screen(self._scr)
@pytest.mark.asyncio
async def test_sandbox_switch_disabled_when_unavailable():
screen = NewSessionScreen(["claude"], sandbox_available=False)
async with _Host(screen).run_test():
assert screen.query_one("#sandbox-switch", Switch).disabled is True
@pytest.mark.asyncio
async def test_sandbox_switch_enabled_when_available():
screen = NewSessionScreen(["claude"], sandbox_available=True)
async with _Host(screen).run_test():
assert screen.query_one("#sandbox-switch", Switch).disabled is False
- Step 2: Run tests to verify they fail
Run: uv run pytest tests/test_tui.py -k "policy_from or sandbox_switch" -q
Expected: FAIL — NewSessionScreen has no _policy_from and its __init__ rejects sandbox_available.
- Step 3: Rewrite the dialog
Replace src/hqt/tui/screens/new_session.py with:
from textual.app import ComposeResult
from textual.containers import Horizontal, Vertical
from textual.screen import ModalScreen
from textual.widgets import Button, Input, Label, Select, Switch
from hqt.sandbox import SandboxPolicy
SessionResult = tuple[str, str, str | None, SandboxPolicy | None]
class NewSessionScreen(ModalScreen[SessionResult | None]):
def __init__(
self, harness_names: list[str], sandbox_available: bool = False
) -> None:
self.harness_names = harness_names
self.sandbox_available = sandbox_available
super().__init__()
def compose(self) -> ComposeResult:
with Vertical(id="new-session-dialog"):
yield Label("New Session")
yield Label("Harness:")
yield Select(
[(n, n) for n in self.harness_names],
id="harness-select",
allow_blank=False,
)
yield Label("Nickname:")
yield Input(placeholder="session nickname", id="nickname-input")
yield Label("Model (optional):")
yield Input(placeholder="model name", id="model-input")
with Horizontal(classes="sandbox-row"):
yield Label("Sandboxed:")
yield Switch(value=False, id="sandbox-switch", disabled=not self.sandbox_available)
if not self.sandbox_available:
yield Label(
"bubblewrap not found — run `hqt doctor`",
id="sandbox-warning",
)
yield Label("Filesystem:")
yield Select(
[("Read-write", "rw"), ("Read-only", "ro")],
id="fs-select",
value="rw",
allow_blank=False,
)
with Horizontal(classes="sandbox-row"):
yield Label("Network:")
yield Switch(value=True, id="net-switch")
with Horizontal(classes="dialog-actions"):
yield Button("OK", variant="primary", id="ok-btn")
yield Button("Cancel", id="cancel-btn")
@staticmethod
def _policy_from(enabled: bool, fs: str, net: bool) -> SandboxPolicy | None:
"""Pure mapping from widget values to a policy (None when disabled)."""
if not enabled:
return None
return SandboxPolicy(fs=fs, net=net)
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "ok-btn":
self._submit()
else:
self.dismiss(None)
def on_input_submitted(self, event: Input.Submitted) -> None:
self._submit()
def _submit(self) -> None:
harness = self.query_one("#harness-select", Select).value
nickname = self.query_one("#nickname-input", Input).value
model = self.query_one("#model-input", Input).value or None
enabled = self.query_one("#sandbox-switch", Switch).value
fs = self.query_one("#fs-select", Select).value
net = self.query_one("#net-switch", Switch).value
policy = self._policy_from(bool(enabled), str(fs), bool(net))
if harness and harness != Select.BLANK:
self.dismiss((str(harness), nickname, model, policy))
else:
self.dismiss(None)
- Step 4: Wire the app
In src/hqt/tui/app.py, add from hqt import sandbox to the imports, then update action_new_session:
def action_new_session(self) -> None:
if self._selected_project_id is None:
self.notify("Select a project first", severity="warning")
return
project_id = self._selected_project_id
harness_names = list(discover_harnesses().keys())
if not harness_names:
self.notify("No harnesses found", severity="error")
return
def on_dismiss(result: tuple[str, str, str | None, object] | None) -> None:
if result:
harness_name, nickname, model, sandbox_policy = result
async def _do() -> None:
create_result = await self._sessions().create_session(
project_id, harness_name, nickname, model, sandbox=sandbox_policy
)
if not create_result.spawn_ok:
first_line = next(
(
ln
for ln in create_result.spawn_error.splitlines()
if ln.strip()
),
create_result.spawn_error,
)[:120]
self.notify(
f"Session created but harness failed to start: {first_line}",
severity="error",
)
await self._refresh_sessions()
self._run_service_worker(_do())
self.push_screen(
NewSessionScreen(harness_names, sandbox_available=sandbox.is_available()),
on_dismiss,
)
- Step 5: Run tests to verify they pass
Run: uv run pytest tests/test_tui.py -q
Expected: PASS (new dialog tests + existing TUI tests).
- Step 6: Commit
git add src/hqt/tui/screens/new_session.py src/hqt/tui/app.py tests/test_tui.py
git commit -m "feat: sandbox toggles in New Session dialog, gated on bwrap availability"
Task 7: doctor reports bubblewrap
Files:
-
Modify:
src/hqt/cli.py:105-122 -
Test:
tests/test_cli.py(create if absent) -
Step 1: Write the failing tests
# tests/test_cli.py
from unittest.mock import patch
from click.testing import CliRunner
from hqt.cli import main
def _which(found: set[str]):
return lambda name: f"/usr/bin/{name}" if name in found else None
def test_doctor_reports_bwrap_present():
with patch("hqt.cli.shutil.which", side_effect=_which({"tmux", "bwrap"})), patch(
"hqt.harnesses.registry.shutil.which", side_effect=_which(set())
):
result = CliRunner().invoke(main, ["doctor"])
assert result.exit_code == 0
assert "bubblewrap: ✓" in result.output
def test_doctor_reports_bwrap_missing():
with patch("hqt.cli.shutil.which", side_effect=_which({"tmux"})), patch(
"hqt.harnesses.registry.shutil.which", side_effect=_which(set())
):
result = CliRunner().invoke(main, ["doctor"])
assert result.exit_code == 0
assert "bubblewrap: ✗" in result.output
- Step 2: Run tests to verify they fail
Run: uv run pytest tests/test_cli.py -q
Expected: FAIL — output contains no "bubblewrap" line.
- Step 3: Add the doctor check
In src/hqt/cli.py, inside doctor(), after the harness loop (before/after is fine), add a non-fatal bubblewrap report:
bwrap = shutil.which("bwrap")
if bwrap:
click.echo(f"bubblewrap: ✓ {bwrap}")
else:
click.echo("bubblewrap: ✗ not found — sandboxed sessions unavailable")
- Step 4: Run tests to verify they pass
Run: uv run pytest tests/test_cli.py -q
Expected: PASS (2 passed).
- Step 5: Full suite + lint
Run: uv run pytest -q && uv run ruff check src tests
Expected: PASS / clean.
- Step 6: Commit
git add src/hqt/cli.py tests/test_cli.py
git commit -m "feat: doctor reports bubblewrap availability"
Self-Review (completed by plan author)
Spec coverage:
- Data model (
sandbox_json+ migration) → Task 4. - ABC additions (skip-permission flags,
sandbox_binds,sandboxedparam) → 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_commandand 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.