Merge branch 'sandboxed-sessions'

This commit is contained in:
2026-06-11 08:35:14 -04:00
18 changed files with 750 additions and 51 deletions
+8 -1
View File
@@ -1,5 +1,6 @@
import logging
import os
import shutil
import subprocess
import sys
import traceback
@@ -105,7 +106,7 @@ def launch(inside_tmux: bool = False):
@main.command()
def doctor():
"""Check system requirements."""
import shutil
from hqt import sandbox
from hqt.harnesses.registry import discover_harnesses
tmux = shutil.which("tmux")
@@ -119,6 +120,12 @@ def doctor():
click.echo(f"{name}: ✓")
if not harnesses:
click.echo("No harnesses found on PATH")
# Use the shared helper so doctor agrees with the New Session dialog and the
# spawn backstop (single source of truth: Linux + bwrap on PATH).
if sandbox.is_available():
click.echo(f"bubblewrap: ✓ {shutil.which('bwrap')}")
else:
click.echo("bubblewrap: ✗ not found — sandboxed sessions unavailable")
@main.command(name="list")
+5
View File
@@ -36,9 +36,14 @@ def _migrate_v3(conn: Connection) -> None:
)
def _migrate_v4(conn: Connection) -> None:
conn.exec_driver_sql("ALTER TABLE sessions ADD COLUMN sandbox_json TEXT")
MIGRATIONS: list[tuple[int, Callable[[Connection], None]]] = [
(2, _migrate_v2),
(3, _migrate_v3),
(4, _migrate_v4),
]
BASELINE_VERSION = 1
+1
View File
@@ -38,6 +38,7 @@ class Session(Base):
tmux_session_name: Mapped[str] = mapped_column(unique=True)
harness_session_id: Mapped[str | None] = mapped_column(default=None)
model: Mapped[str | None] = mapped_column(default=None)
sandbox_json: Mapped[str | None] = mapped_column(default=None)
worktree_path: Mapped[str | None] = mapped_column(default=None)
worktree_branch: Mapped[str | None] = mapped_column(default=None)
created_at: Mapped[datetime] = mapped_column(insert_default=func.now())
+18 -2
View File
@@ -2,6 +2,8 @@ from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from hqt.sandbox import Bind
@dataclass
class SpawnConfig:
@@ -14,20 +16,34 @@ 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
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
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
+19 -2
View File
@@ -2,6 +2,7 @@ import uuid
from pathlib import Path
from hqt.harnesses.base import HarnessConfigurator, SpawnConfig
from hqt.sandbox import Bind
NAMESPACE = uuid.UUID("a1b2c3d4-e5f6-7890-abcd-ef1234567890")
@@ -9,6 +10,7 @@ NAMESPACE = uuid.UUID("a1b2c3d4-e5f6-7890-abcd-ef1234567890")
class ClaudeConfigurator(HarnessConfigurator):
name = "claude"
display_name = "Claude Code"
sandbox_skip_permission_flags = ["--dangerously-skip-permissions"]
# Markers for parse_status — conservative, case-insensitive substring checks.
# None returned when none match; caller falls back to activity-based status.
@@ -21,20 +23,35 @@ class ClaudeConfigurator(HarnessConfigurator):
def generate_session_id(self, hqt_session_id: int) -> str:
return str(uuid.uuid5(NAMESPACE, str(hqt_session_id)))
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
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
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)
def parse_status(self, pane_text: str) -> str | None:
+19 -2
View File
@@ -4,6 +4,7 @@ from datetime import datetime
from pathlib import Path
from hqt.harnesses.base import HarnessConfigurator, SpawnConfig
from hqt.sandbox import Bind
log = logging.getLogger(__name__)
@@ -12,6 +13,7 @@ class CodexConfigurator(HarnessConfigurator):
name = "codex"
display_name = "Codex"
captures_session_id = True
sandbox_skip_permission_flags = ["--dangerously-bypass-approvals-and-sandbox"]
# Markers for parse_status — conservative, case-insensitive substring checks.
_WORKING_MARKERS = ("esc to interrupt",)
@@ -19,20 +21,35 @@ class CodexConfigurator(HarnessConfigurator):
def generate_session_id(self, hqt_session_id: int) -> str:
return str(hqt_session_id)
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
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
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)
def parse_status(self, pane_text: str) -> str | None:
+10 -2
View File
@@ -14,11 +14,19 @@ class GenericConfigurator(HarnessConfigurator):
return str(hqt_session_id)
def build_spawn_config(
self, project_path: Path, harness_session_id: str, model: str | None = None
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
self,
project_path: Path,
harness_session_id: str,
model: str | None = None,
sandboxed: bool = False,
) -> SpawnConfig:
return SpawnConfig(command=self._command, cwd=project_path)
+14 -2
View File
@@ -1,6 +1,7 @@
from pathlib import Path
from hqt.harnesses.base import HarnessConfigurator, SpawnConfig
from hqt.sandbox import Bind
class KiroConfigurator(HarnessConfigurator):
@@ -10,12 +11,23 @@ class KiroConfigurator(HarnessConfigurator):
def generate_session_id(self, hqt_session_id: int) -> str:
return str(hqt_session_id)
def sandbox_binds(self) -> list[Bind]:
return [Bind(Path.home() / ".kiro", writable=True)]
def build_spawn_config(
self, project_path: Path, harness_session_id: str, model: str | None = None
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
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)
+81
View File
@@ -0,0 +1,81 @@
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
# 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]
# Git reads user.name/email from ~/.gitconfig; without it `git commit` fails
# inside the jail. Bound read-only since $HOME itself is hidden.
gitconfig = Path.home() / ".gitconfig"
if gitconfig.exists():
args += ["--ro-bind", str(gitconfig), str(gitconfig)]
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
+85 -8
View File
@@ -1,8 +1,9 @@
import asyncio
import contextlib
import json
import logging
import time
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
@@ -10,11 +11,13 @@ from pathlib import Path
from sqlalchemy.orm import Session as DBSession
from sqlalchemy.orm import selectinload, sessionmaker
from hqt import sandbox
from hqt.db.models import Harness, Project, ScheduledPrompt, Session
from hqt.errors import ServiceError
from hqt.git import worktree
from hqt.git.worktree import WorktreeState
from hqt.harnesses.base import HarnessConfigurator, SpawnConfig
from hqt.sandbox import Bind, SandboxPolicy
from hqt.status import window_label
from hqt.tmux.manager import SpawnRequest, TmuxManager
from hqt.tmux.runner import WindowInfo
@@ -164,6 +167,51 @@ class SessionService:
return created_at.timestamp()
return 0.0
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 _worktree_git_binds(self, db: DBSession, sess: Session) -> list[Bind]:
"""Extra binds a sandboxed worktree session needs for git to work.
A linked worktree's `.git` is a file pointing at the real gitdir under
the repo's common `.git` directory (`<repo>/.git/worktrees/<branch>`),
which lives OUTSIDE the bound cwd. Bind the repo's `.git` writable so
git operations (status, commit) resolve inside the jail. Plain sessions
need nothing extra.
"""
if not sess.worktree_path or sess.worktree_branch is None:
return []
repo = self._repo_path_for_worktree(
db, sess.project_id, sess.worktree_path, sess.worktree_branch
)
return [Bind(repo / ".git", writable=True)]
def _wrap_command(
self,
configurator: HarnessConfigurator,
command: list[str],
cwd: Path,
policy: SandboxPolicy | None,
extra_binds: Sequence[Bind] = (),
) -> 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"
)
binds = [*configurator.sandbox_binds(), *extra_binds]
return sandbox.wrap(command, cwd, policy, binds)
def _maybe_capture_missing_session_id(self, db: DBSession, sess: Session) -> None:
configurator = self.harnesses.get(sess.harness.name)
if configurator is None or not configurator.captures_session_id:
@@ -197,6 +245,7 @@ class SessionService:
nickname: str | None = None,
model: str | None = None,
worktree_branch: str | None = None,
sandbox: SandboxPolicy | None = None,
) -> "CreateSessionResult":
"""Create a new session row, spawn the harness window, and return a
CreateSessionResult with the Session row and spawn outcome.
@@ -243,13 +292,25 @@ class SessionService:
sess.tmux_session_name = f"hqt-{sess.id}"
harness_session_id = configurator.generate_session_id(sess.id)
sess.harness_session_id = harness_session_id
sess.sandbox_json = (
json.dumps({"fs": sandbox.fs, "net": sandbox.net})
if sandbox is not None
else None
)
spawn_cfg = configurator.build_spawn_config(
run_path, harness_session_id, model
run_path, harness_session_id, model, sandboxed=sandbox is not None
)
command = self._wrap_command(
configurator,
spawn_cfg.command,
spawn_cfg.cwd,
sandbox,
self._worktree_git_binds(db, sess),
)
log.info(
"Spawning window %s: cmd=%s cwd=%s",
sess.tmux_session_name,
spawn_cfg.command,
command,
spawn_cfg.cwd,
)
async with self._maybe_capture_lock(configurator.captures_session_id):
@@ -257,7 +318,7 @@ class SessionService:
spawn_result = await self.tmux.spawn(
SpawnRequest(
window_name=sess.tmux_session_name,
command=spawn_cfg.command,
command=command,
cwd=str(spawn_cfg.cwd),
env=spawn_cfg.env,
)
@@ -358,13 +419,21 @@ class SessionService:
harness_session_id = (
sess.harness_session_id or configurator.generate_session_id(sess.id)
)
policy = self._sandbox_policy(sess)
spawn_cfg = configurator.build_spawn_config(
run_path, harness_session_id, sess.model
run_path, harness_session_id, sess.model, sandboxed=policy is not None
)
command = self._wrap_command(
configurator,
spawn_cfg.command,
spawn_cfg.cwd,
policy,
self._worktree_git_binds(db, sess),
)
async with self._maybe_capture_lock(configurator.captures_session_id):
since = time.time()
result = await self.tmux.respawn_verified(
window_name, spawn_cfg.command, str(spawn_cfg.cwd), env=spawn_cfg.env
window_name, command, str(spawn_cfg.cwd), env=spawn_cfg.env
)
if not result.ok:
log.error(
@@ -643,6 +712,14 @@ class SessionService:
harness_session_id = (
sess.harness_session_id or configurator.generate_session_id(sess.id)
)
return configurator.build_resume_config(
self._session_path(db, sess), harness_session_id, sess.model
policy = self._sandbox_policy(sess)
cfg = configurator.build_resume_config(
self._session_path(db, sess),
harness_session_id,
sess.model,
sandboxed=policy is not None,
)
command = self._wrap_command(
configurator, cfg.command, cfg.cwd, policy, self._worktree_git_binds(db, sess)
)
return SpawnConfig(command=command, env=cfg.env, cwd=cfg.cwd)
+12 -3
View File
@@ -10,6 +10,7 @@ from textual.screen import ModalScreen
from textual.theme import Theme
from textual.widgets import Footer
from hqt import sandbox
from hqt.config import get_settings
from hqt.git import worktree
from hqt.db.engine import ensure_db, get_engine, get_session_factory
@@ -255,10 +256,10 @@ class HqtApp(App):
project_path = project.path
def on_dismiss(
result: tuple[str, str, str | None, str | None] | None,
result: tuple[str, str, str | None, str | None, object] | None,
) -> None:
if result:
harness_name, nickname, model, worktree_branch = result
harness_name, nickname, model, worktree_branch, sandbox_policy = result
async def _do() -> None:
create_result = await self._sessions().create_session(
@@ -267,6 +268,7 @@ class HqtApp(App):
nickname,
model,
worktree_branch=worktree_branch,
sandbox=sandbox_policy,
)
if not create_result.spawn_ok:
first_line = next(
@@ -296,7 +298,14 @@ class HqtApp(App):
except Exception as exc: # noqa: BLE001 - keep the worker from crashing the TUI
log.warning("is_git_repo check failed for %s: %s", project_path, exc)
is_git_repo = False
self.push_screen(NewSessionScreen(harness_names, is_git_repo), on_dismiss)
self.push_screen(
NewSessionScreen(
harness_names,
is_git_repo,
sandbox_available=sandbox.is_available(),
),
on_dismiss,
)
self.run_worker(_open())
+45 -4
View File
@@ -1,15 +1,24 @@
from textual.app import ComposeResult
from textual.containers import Horizontal, Vertical
from textual.screen import ModalScreen
from textual.widgets import Button, Checkbox, Input, Label, Select
from textual.widgets import Button, Checkbox, Input, Label, Select, Switch
from hqt.git import worktree
from hqt.sandbox import SandboxPolicy
class NewSessionScreen(ModalScreen[tuple[str, str, str | None, str | None] | None]):
def __init__(self, harness_names: list[str], is_git_repo: bool) -> None:
class NewSessionScreen(
ModalScreen[tuple[str, str, str | None, str | None, SandboxPolicy | None] | None]
):
def __init__(
self,
harness_names: list[str],
is_git_repo: bool = False,
sandbox_available: bool = False,
) -> None:
self.harness_names = harness_names
self._is_git_repo = is_git_repo
self.sandbox_available = sandbox_available
super().__init__()
def compose(self) -> ComposeResult:
@@ -43,6 +52,26 @@ class NewSessionScreen(ModalScreen[tuple[str, str, str | None, str | None] | Non
branch_input = Input(placeholder="branch name", id="branch-input")
branch_input.display = False
yield branch_input
yield Label("Sandboxed (bubblewrap):")
yield Switch(
id="sandbox-switch",
value=False,
disabled=not self.sandbox_available,
)
if not self.sandbox_available:
yield Label(
"bubblewrap not found — run `hqt doctor`",
id="sandbox-warning",
)
yield Label("Filesystem access:")
yield Select(
[("Read-write", "rw"), ("Read-only", "ro")],
id="fs-select",
value="rw",
allow_blank=False,
)
yield Label("Network:")
yield Switch(id="net-switch", value=True)
with Horizontal(classes="dialog-actions"):
yield Button("OK", variant="primary", id="ok-btn")
yield Button("Cancel", id="cancel-btn")
@@ -71,6 +100,13 @@ class NewSessionScreen(ModalScreen[tuple[str, str, str | None, str | None] | Non
# can be completed from the keyboard without reaching for the mouse.
self._submit()
@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 _submit(self) -> None:
harness = self.query_one("#harness-select", Select).value
nickname = self.query_one("#nickname-input", Input).value
@@ -86,7 +122,12 @@ class NewSessionScreen(ModalScreen[tuple[str, str, str | None, str | None] | Non
branch_input.focus()
return
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, worktree_branch))
self.dismiss((str(harness), nickname, model, worktree_branch, policy))
else:
self.dismiss(None)
+35
View File
@@ -0,0 +1,35 @@
from unittest.mock import patch
from click.testing import CliRunner
from hqt import cli as cli_module
from hqt import sandbox
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():
"""Test that doctor reports bubblewrap when present."""
# tmux is discovered via cli.shutil.which; bwrap availability comes from the
# shared sandbox.is_available() helper (the single source of truth).
with (
patch.object(cli_module.shutil, "which", side_effect=_which({"tmux", "bwrap"})),
patch.object(sandbox, "is_available", return_value=True),
):
result = CliRunner().invoke(main, ["doctor"])
assert result.exit_code == 0
assert "bubblewrap: ✓" in result.output
def test_doctor_reports_bwrap_missing():
"""Test that doctor reports bubblewrap as missing when not present."""
with (
patch.object(cli_module.shutil, "which", side_effect=_which({"tmux"})),
patch.object(sandbox, "is_available", return_value=False),
):
result = CliRunner().invoke(main, ["doctor"])
assert result.exit_code == 0
assert "bubblewrap: ✗" in result.output
+61 -11
View File
@@ -98,6 +98,7 @@ def test_unversioned_existing_db_stamped_latest(tmp_path):
conn.exec_driver_sql("DROP TABLE scheduled_prompts")
conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN worktree_path")
conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN worktree_branch")
conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN sandbox_json")
migrate(engine)
assert _user_version(engine) == migrations.LATEST_VERSION
@@ -117,22 +118,23 @@ def test_pending_migrations_applied_in_order_once(tmp_path, monkeypatch):
ensure_db(settings) # stamps migrations.LATEST_VERSION
applied: list[int] = []
# Use versions above current LATEST_VERSION so none have been applied yet.
next_version = migrations.LATEST_VERSION + 1
monkeypatch.setattr(
migrations,
"MIGRATIONS",
[
*migrations.MIGRATIONS,
(4, lambda conn: applied.append(4)),
(5, lambda conn: applied.append(5)),
(next_version, lambda conn: applied.append(next_version)),
(next_version + 1, lambda conn: applied.append(next_version + 1)),
],
)
monkeypatch.setattr(migrations, "LATEST_VERSION", 5)
monkeypatch.setattr(migrations, "LATEST_VERSION", next_version + 1)
engine = get_engine(settings)
migrate(engine)
assert applied == [4, 5]
assert _user_version(engine) == 5
assert applied == [next_version, next_version + 1]
assert _user_version(engine) == next_version + 1
migrate(engine) # re-run is a no-op
assert applied == [4, 5]
assert applied == [next_version, next_version + 1]
def _session_column_names(engine) -> list[str]:
@@ -154,11 +156,12 @@ def test_fresh_db_has_worktree_columns(tmp_path):
cols = _session_column_names(engine)
assert "worktree_path" in cols
assert "worktree_branch" in cols
assert "sandbox_json" in cols
assert _user_version(engine) == migrations.LATEST_VERSION
def test_v1_db_upgraded_to_v2(tmp_path):
# Build a v1 database: create all tables, drop the two new columns, stamp v1.
def test_v1_db_upgraded_to_latest(tmp_path):
# Build a v1 database: create all tables, drop v2+ objects, stamp v1.
settings = _tmp_settings(tmp_path)
settings.db_path.parent.mkdir(parents=True, exist_ok=True)
engine = get_engine(settings)
@@ -167,6 +170,7 @@ def test_v1_db_upgraded_to_v2(tmp_path):
conn.exec_driver_sql("DROP TABLE scheduled_prompts")
conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN worktree_path")
conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN worktree_branch")
conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN sandbox_json")
conn.exec_driver_sql("PRAGMA user_version = 1")
migrate(engine)
@@ -174,6 +178,8 @@ def test_v1_db_upgraded_to_v2(tmp_path):
cols = _session_column_names(engine)
assert "worktree_path" in cols
assert "worktree_branch" in cols
assert "sandbox_json" in cols
assert "scheduled_prompts" in inspect(engine).get_table_names()
assert _user_version(engine) == migrations.LATEST_VERSION
@@ -195,22 +201,24 @@ def test_fresh_db_has_scheduled_prompts_table(tmp_path):
"created_at",
"sent_at",
]
assert _user_version(engine) == 3
assert _user_version(engine) == migrations.LATEST_VERSION
def test_v2_db_upgraded_to_v3_adds_scheduled_prompts(tmp_path):
def test_v2_db_upgraded_to_latest_adds_scheduled_prompts_and_sandbox(tmp_path):
settings = _tmp_settings(tmp_path)
settings.db_path.parent.mkdir(parents=True, exist_ok=True)
engine = get_engine(settings)
Base.metadata.create_all(engine)
with engine.begin() as conn:
conn.exec_driver_sql("DROP TABLE scheduled_prompts")
conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN sandbox_json")
conn.exec_driver_sql("PRAGMA user_version = 2")
migrate(engine)
assert "scheduled_prompts" in inspect(engine).get_table_names()
assert _user_version(engine) == 3
assert "sandbox_json" in _session_column_names(engine)
assert _user_version(engine) == migrations.LATEST_VERSION
def test_scheduled_prompt_one_row_per_session(tmp_path):
@@ -291,3 +299,45 @@ def test_rows_stay_readable_after_session_closes(tmp_path):
session.add(p)
session.commit()
assert p.name == "x" # would raise DetachedInstanceError without the flag
def test_latest_version_is_four():
assert migrations.LATEST_VERSION == 4
def test_migration_adds_sandbox_json_column(tmp_path):
# Simulate a v3 DB whose sessions table predates the sandbox 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 = 3")
migrate(engine)
cols = [c["name"] for c in inspect(engine).get_columns("sessions")]
assert "sandbox_json" in cols
assert _user_version(engine) == migrations.LATEST_VERSION
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-sandbox")
h = Harness(name="claude-sandbox")
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}'
+57 -5
View File
@@ -4,8 +4,11 @@ import time
from pathlib import Path
from unittest.mock import patch
from hqt.harnesses.configurators.codex import CodexConfigurator
from hqt.harnesses.base import HarnessConfigurator
from hqt.harnesses.configurators.claude import ClaudeConfigurator
from hqt.harnesses.configurators.codex import CodexConfigurator
from hqt.harnesses.configurators.generic import GenericConfigurator
from hqt.harnesses.configurators.kiro import KiroConfigurator
from hqt.harnesses.registry import discover_harnesses
@@ -129,8 +132,6 @@ def test_codex_captures_session_id_flag():
def test_base_captures_session_id_flag_false():
"""HarnessConfigurator base class default must be False."""
from hqt.harnesses.base import HarnessConfigurator
assert HarnessConfigurator.captures_session_id is False
@@ -287,8 +288,6 @@ CODEX_IDLE_SCREEN = """\
def test_base_parse_status_returns_none():
"""Base HarnessConfigurator.parse_status always returns None."""
from hqt.harnesses.configurators.kiro import KiroConfigurator
k = KiroConfigurator()
assert k.parse_status("anything") is None
assert k.parse_status("") is None
@@ -346,3 +345,56 @@ def test_codex_parse_status_none_for_empty():
"""CodexConfigurator: empty text → None."""
c = CodexConfigurator()
assert c.parse_status("") is None
# ---------------------------------------------------------------------------
# Task 3: sandbox skip-permission flags and binds
# ---------------------------------------------------------------------------
def test_base_skip_flags_default_empty():
assert HarnessConfigurator.sandbox_skip_permission_flags == []
def test_base_sandbox_binds_default_empty():
# GenericConfigurator does not override sandbox_binds; verifies the base default.
assert GenericConfigurator(["mytool"]).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"]
+123
View File
@@ -0,0 +1,123 @@
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
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
def test_wrap_binds_gitconfig_readonly_when_present(tmp_path, monkeypatch):
"""~/.gitconfig is bound read-only so `git commit` works inside the jail."""
home = tmp_path / "home"
home.mkdir()
gitconfig = home / ".gitconfig"
gitconfig.write_text("[user]\n\tname = Test\n")
monkeypatch.setattr(sandbox.Path, "home", classmethod(lambda cls: home))
args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [])
assert (str(gitconfig), str(gitconfig)) in _split(args, "--ro-bind")
def test_wrap_skips_gitconfig_when_absent(tmp_path, monkeypatch):
home = tmp_path / "home"
home.mkdir()
monkeypatch.setattr(sandbox.Path, "home", classmethod(lambda cls: home))
args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [])
assert str(home / ".gitconfig") not in args
+95 -1
View File
@@ -9,6 +9,7 @@ from sqlalchemy.pool import StaticPool
from hqt.db.models import Base, Harness, Project, ScheduledPrompt, Session
from hqt.errors import ServiceError
from hqt.sandbox import SandboxPolicy
from hqt.sessions.service import SessionInfo, SessionService
from hqt.tmux.manager import SpawnResult, TmuxManager
@@ -931,7 +932,7 @@ async def test_attach_late_capture_updates_id_before_resume(
assert ok is True
harness.build_resume_config.assert_called_with(
Path("/tmp/myproj"), "late-codex-id", result.session.model
Path("/tmp/myproj"), "late-codex-id", result.session.model, sandboxed=False
)
from hqt.db.models import Session as DBSession
@@ -1646,3 +1647,96 @@ async def test_create_session_no_lock_for_non_capturing_harness(service, db, tmu
# The `service` fixture's harness has captures_session_id = False, so the
# spawn must NOT be serialized under the capture lock.
assert observed["locked_during_spawn"] is False
# ---------------------------------------------------------------------------
# Task 5: sandbox wrapping on spawn and resume
# ---------------------------------------------------------------------------
@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()
assert (
sandbox_harness["claude-code"].build_spawn_config.call_args.kwargs["sandboxed"]
is True
)
assert tmux.spawn.call_args.args[0].command == ["bwrap", "--", "claude"]
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),
)
@pytest.mark.asyncio
async def test_sandboxed_worktree_binds_repo_git_dir(
sandbox_service, db, tmux, sandbox_harness, fake_worktree
):
"""A sandboxed worktree session binds the repo's common .git writable so
git works inside the jail (the worktree's gitdir lives under <repo>/.git)."""
from hqt.sandbox import Bind
with (
patch("hqt.sessions.service.sandbox.is_available", return_value=True),
patch(
"hqt.sessions.service.sandbox.wrap", return_value=["bwrap", "--", "claude"]
) as mock_wrap,
):
await sandbox_service.create_session(
project_id=1,
harness_name="claude-code",
worktree_branch="feature-x",
sandbox=SandboxPolicy(fs="rw", net=True),
)
binds = mock_wrap.call_args.args[3]
assert Bind(Path("/tmp/myproj/.git"), writable=True) in binds
+62 -8
View File
@@ -292,7 +292,12 @@ async def test_new_session_on_dismiss_sequential_create_then_refresh():
call_order: list[str] = []
async def fake_create(
project_id, harness_name, nickname, model, worktree_branch=None
project_id,
harness_name,
nickname,
model,
worktree_branch=None,
sandbox=None,
):
# Brief yield so that, if two separate workers were used, the list
# worker would be able to start and record "list" before this
@@ -324,7 +329,7 @@ async def test_new_session_on_dismiss_sequential_create_then_refresh():
await pilot.pause()
# The NewSessionScreen is now on top; dismiss it with a valid result tuple.
app.screen.dismiss(("claude", "mynick", None, None))
app.screen.dismiss(("claude", "mynick", None, None, None))
await pilot.pause()
# Wait for the worker spawned by on_dismiss to finish deterministically.
@@ -396,7 +401,7 @@ async def test_new_session_on_dismiss_notifies_on_spawn_failure():
await app.workers.wait_for_complete()
await pilot.pause()
app.screen.dismiss(("claude", "mynick", None, None))
app.screen.dismiss(("claude", "mynick", None, None, None))
await pilot.pause()
await app.workers.wait_for_complete()
await pilot.pause()
@@ -467,7 +472,7 @@ async def test_new_session_on_dismiss_no_notify_on_success():
await app.workers.wait_for_complete()
await pilot.pause()
app.screen.dismiss(("claude", "mynick", None, None))
app.screen.dismiss(("claude", "mynick", None, None, None))
await pilot.pause()
await app.workers.wait_for_complete()
await pilot.pause()
@@ -582,7 +587,7 @@ async def test_new_session_enter_in_input_submits_dialog():
await pilot.press("enter")
await pilot.pause()
assert results == [("claude", "mywork", None, None)], f"got {results}"
assert results == [("claude", "mywork", None, None, None)], f"got {results}"
# ---------------------------------------------------------------------------
@@ -1291,7 +1296,7 @@ async def test_new_session_worktree_unchecked_returns_none_branch():
app.screen.query_one("#nickname-input", Input).value = "mywork"
app.screen.query_one("#ok-btn", Button).press()
await pilot.pause()
assert results == [("claude", "mywork", None, None)], f"got {results}"
assert results == [("claude", "mywork", None, None, None)], f"got {results}"
@pytest.mark.asyncio
@@ -1316,7 +1321,9 @@ async def test_new_session_worktree_checked_uses_typed_branch():
branch.value = "feature-x"
app.screen.query_one("#ok-btn", Button).press()
await pilot.pause()
assert results == [("claude", "My Work", None, "feature-x")], f"got {results}"
assert results == [("claude", "My Work", None, "feature-x", None)], (
f"got {results}"
)
@pytest.mark.asyncio
@@ -1338,7 +1345,7 @@ async def test_new_session_worktree_checked_blank_branch_slugifies_nickname():
app.screen.query_one("#branch-input", Input).value = ""
app.screen.query_one("#ok-btn", Button).press()
await pilot.pause()
assert results == [("claude", "Cool Feature!", None, "cool-feature")], (
assert results == [("claude", "Cool Feature!", None, "cool-feature", None)], (
f"got {results}"
)
@@ -1504,3 +1511,50 @@ async def test_delete_non_worktree_session_does_not_push_modal():
verify_db = app._db_factory()
assert verify_db.get(Session, sess_id) is None
verify_db.close()
# ---------------------------------------------------------------------------
# Task 6: Sandbox controls in NewSessionScreen
# ---------------------------------------------------------------------------
def test_policy_from_disabled_returns_none():
from hqt.tui.screens.new_session import NewSessionScreen
assert NewSessionScreen._policy_from(False, "rw", True) is None
def test_policy_from_enabled_builds_policy():
from hqt.sandbox import SandboxPolicy
from hqt.tui.screens.new_session import NewSessionScreen
p = NewSessionScreen._policy_from(True, "ro", False)
assert p == SandboxPolicy(fs="ro", net=False)
@pytest.mark.asyncio
async def test_sandbox_switch_disabled_when_unavailable():
from hqt.tui.screens.new_session import NewSessionScreen
from textual.widgets import Switch
app = HqtApp()
async with app.run_test(size=(120, 40)) as pilot:
app.push_screen(
NewSessionScreen(["claude"], is_git_repo=False, sandbox_available=False)
)
await pilot.pause()
assert app.screen.query_one("#sandbox-switch", Switch).disabled is True
@pytest.mark.asyncio
async def test_sandbox_switch_enabled_when_available():
from hqt.tui.screens.new_session import NewSessionScreen
from textual.widgets import Switch
app = HqtApp()
async with app.run_test(size=(120, 40)) as pilot:
app.push_screen(
NewSessionScreen(["claude"], is_git_repo=False, sandbox_available=True)
)
await pilot.pause()
assert app.screen.query_one("#sandbox-switch", Switch).disabled is False