feat: wrap sandboxed sessions in bwrap on spawn and resume

- Import `hqt.sandbox` module and `SandboxPolicy` in service.py
- Add `_sandbox_policy` helper to decode persisted JSON policy from a session row
- Add `_wrap_command` helper that calls `sandbox.is_available` / `sandbox.wrap` and raises `ServiceError` when bwrap is unavailable for a sandboxed session
- Add `sandbox: SandboxPolicy | None = None` parameter to `create_session`; persist as `sandbox_json`, pass `sandboxed=` to `build_spawn_config`, wrap command before spawning
- Update `_get_resume_config` to pass `sandboxed=` to `build_resume_config` and wrap the resulting command
- Update rung-2 path in `_respawn_with_fallback` to pass `sandboxed=` and wrap the fresh-spawn command
- Add three new tests covering: sandboxed create wraps command, unsandboxed create does not wrap, and ServiceError on missing bwrap

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/hqt/sessions/service.py

@@ -1,5 +1,6 @@
 import asyncio
 import contextlib
+import json
 import logging
 import time
 from collections.abc import Mapping
@@ -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, 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 SandboxPolicy
 from hqt.status import window_label
 from hqt.tmux.manager import SpawnRequest, TmuxManager
 from hqt.tmux.runner import WindowInfo
@@ -159,6 +162,33 @@ 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 _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())
+
     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:
@@ -192,6 +222,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.
@@ -238,13 +269,21 @@ 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
             )
             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):
@@ -252,7 +291,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,
                     )
@@ -353,13 +392,15 @@ 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)
         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(
@@ -552,6 +593,10 @@ class SessionService:
         harness_session_id = (
             sess.harness_session_id or configurator.generate_session_id(sess.id)
         )
-        return configurator.build_resume_config(
+        policy = self._sandbox_policy(sess)
-            self._session_path(db, sess), harness_session_id, sess.model
+        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)
+        return SpawnConfig(command=command, env=cfg.env, cwd=cfg.cwd)

tests/test_sessions.py

@@ -8,6 +8,7 @@ from sqlalchemy.pool import StaticPool
 
 from hqt.db.models import Base, Harness, Project
 from hqt.errors import ServiceError
+from hqt.sandbox import SandboxPolicy
 from hqt.sessions.service import SessionInfo, SessionService
 from hqt.tmux.manager import SpawnResult, TmuxManager
 
@@ -816,7 +817,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
 
@@ -1518,3 +1519,66 @@ 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),
+            )