Merge branch 'tool-windows'

Add tool windows: tool registry, styled aux tmux windows
(TmuxRunner.new_aux_window / TmuxManager.open_aux_window),
SessionService tool-window + clone-for-window methods, and
hqt tool / hqt palette CLI subcommands (fzf tool launcher).

# Conflicts:
#	src/hqt/cli.py
#	src/hqt/tmux/manager.py
#	tests/test_cli.py
#	tests/test_sessions.py

docs/superpowers/specs/2026-06-11-delayed-prompt-injection-design.md

@@ -0,0 +1,198 @@
+# Delayed prompt injection
+
+**Date:** 2026-06-11
+**Status:** Approved
+
+## Goal
+
+Add a TUI-only feature that schedules one delayed prompt for a specific hqt
+session and injects it when the selected time arrives. The primary use case is
+recovering a harness that stopped because it ran out of budget: the user can
+schedule a restart/continue prompt and let hqt deliver it later.
+
+Scheduled prompts fire only while the hqt TUI is running. If hqt is closed at
+the due time, the prompt remains pending and fires on the next TUI poll after
+hqt starts again.
+
+## User Behavior
+
+The session list gets a new keybinding, `p`, labeled "Prompt Later". Pressing
+it for the selected session opens a modal that collects:
+
+- Prompt text.
+- A relative delay such as `30m`, `2h`, `1h30m`, or `90s`.
+- A confirmation action.
+
+Each session supports exactly one active delayed prompt. Scheduling a new prompt
+for that session replaces the existing pending, sent, or failed prompt. A
+companion cancel action, `P` / `Shift+P`, removes the selected session's active
+prompt.
+
+Visualization is intentionally compact:
+
+- The session row shows the next pending trigger, for example
+  `◉ waiting  @ 14:30` or `● idle  @ 12m`.
+- After scheduling, canceling, sending, or failing a prompt, the TUI shows a
+  short notification such as `Prompt scheduled for hqt-4 at 14:30`.
+
+The row indicator is shown only for pending prompts. Sent prompts disappear from
+the active indicator. Failed prompts are not retried automatically; they can be
+replaced or canceled from the selected session.
+
+## Architecture
+
+Use a persisted database table and drive dispatch from the existing TUI poller.
+This matches the current service architecture and keeps scheduling state visible
+across TUI restarts without introducing a background daemon.
+
+### Data Model
+
+Add a `ScheduledPrompt` model and migration:
+
+- `id` primary key.
+- `session_id` foreign key to `sessions.id`, unique.
+- `prompt` text, required.
+- `due_at` datetime, required.
+- `status` string, one of `pending`, `sending`, `sent`, or `failed`.
+- `error` text, nullable.
+- `created_at` datetime.
+- `sent_at` datetime, nullable.
+
+The unique `session_id` constraint enforces one active prompt record per
+session. Scheduling a prompt replaces any existing prompt row for that session,
+including failed or sent rows.
+
+### Service Layer
+
+`SessionService` owns delayed prompt behavior because it already owns session
+lookup, resume/fallback behavior, status polling, and tmux interaction.
+
+Add service methods along these lines:
+
+```python
+def schedule_prompt(self, session_id: int, prompt: str, due_at: datetime) -> ScheduledPrompt:
+    ...
+
+def cancel_scheduled_prompt(self, session_id: int) -> None:
+    ...
+
+def get_scheduled_prompt(self, session_id: int) -> ScheduledPrompt | None:
+    ...
+
+async def dispatch_due_prompts(self, now: datetime) -> list[DispatchResult]:
+    ...
+```
+
+Extend `SessionInfo` with an optional `scheduled_prompt` field so `SessionList`
+can render the compact row indicator from the same object it already receives
+for each row.
+
+### Tmux Layer
+
+`TmuxRunner` already has literal `send_text()` and `send_enter()` methods. Add
+thin `TmuxManager` delegates and make both send paths return success/failure
+instead of fire-and-forget so the service can record failed injection.
+
+Prompt injection sends the prompt body as literal text, then sends Enter as a
+separate tmux key event. Prompt content such as `-flag`, quotes, semicolons, or
+the word `Enter` must be treated as text, not as tmux commands or key names.
+
+### TUI Runtime
+
+The existing 3-second TUI poll loop becomes the scheduler runtime:
+
+1. Call `dispatch_due_prompts(now)`.
+2. Surface any sent/failed dispatch results as notifications.
+3. Refresh session rows and tmux labels using the existing status flow.
+
+No prompt fires while the TUI process is not running. Overdue prompts fire on
+the first poll after startup.
+
+## Dispatch Semantics
+
+When a pending prompt is due:
+
+1. Load the target session row.
+2. Change the prompt status to `sending` and commit before interacting with
+   tmux, so overlapping poll workers cannot inject the same prompt twice.
+3. If the session row is gone, mark the prompt `failed`.
+4. If the tmux window is alive, inject directly.
+5. If the window is dead or missing, resume the session without switching tmux
+   focus, then inject.
+6. On successful text + Enter injection, mark the row `sent` and set `sent_at`.
+7. On any failure, mark the row `failed` and store an error message.
+
+The current `attach_session()` method resumes and then selects the tmux window.
+Factor an internal helper, `_ensure_session_alive(db, sess) -> bool`, so due
+prompt dispatch can reuse the resume/fallback ladder without changing the
+user's active tmux window.
+
+Failed prompts must not retry automatically on every poll. This avoids sending
+duplicate prompts or repeatedly restarting a harness after a transient failure.
+The user can replace the failed prompt by scheduling a new one, or clear it with
+the cancel action.
+
+## TUI Details
+
+Add a new modal screen, `SchedulePromptScreen`, under `src/hqt/tui/screens/`.
+It should follow existing modal conventions from the project:
+
+- Plain Textual form controls.
+- Compact dialog styling using existing CSS patterns.
+- Inline validation for empty prompt or invalid delay.
+- OK and Cancel actions.
+
+Delay parsing is relative-only for the first version. Accepted units are
+seconds, minutes, and hours. Combined values such as `1h30m` are accepted.
+Absolute timestamps, recurring prompts, and multiple queued prompts are out of
+scope.
+
+`SessionList` row rendering includes pending prompt information when present.
+Show `@ <countdown>` while the due time is less than 60 minutes away, rounded
+up to whole minutes, for example `@ 12m`. Show local 24-hour time for prompts
+that are at least 60 minutes away, for example `@ 14:30`.
+
+## Error Handling
+
+- Scheduling with no selected session shows a warning notification.
+- Empty prompt text or invalid delay keeps the modal open with an inline error.
+- Scheduling for a deleted session raises `ServiceError` and shows a TUI error.
+- Deleted target session at dispatch time marks the prompt failed.
+- Resume/fallback failure marks the prompt failed with the existing tmux/spawn
+  error when available.
+- Text or Enter injection failure marks the prompt failed.
+- Database errors are allowed to surface through existing service error handling.
+
+## Testing
+
+Add focused tests at the same layers the current code uses:
+
+- Model/migration test for the `scheduled_prompts` table and unique
+  `session_id`.
+- Service tests for schedule, replace, cancel, and retrieving pending prompt
+  metadata in session rows.
+- Service dispatch tests for alive-window injection, dead-window auto-resume
+  without attach/select, missing session failure, injection failure, and failed
+  prompts not retrying on the next poll.
+- Tmux tests for manager send delegates and literal send behavior returning
+  success/failure.
+- TUI tests for the keybindings, modal validation, scheduling action, cancel
+  action, row indicator, and poll-time dispatch notifications.
+
+After implementation changes, the project checks must pass:
+
+```bash
+uv run ruff format src tests
+uv run ruff check src tests
+uv run ty check
+```
+
+## Out of Scope
+
+- Firing delayed prompts while hqt is closed.
+- CLI scheduling.
+- Multiple queued prompts per session.
+- Absolute date/time scheduling.
+- Recurring prompts.
+- Prompt history or a full queue-management screen.
+- Harness-specific prompt templates.

docs/superpowers/specs/2026-06-11-tmux-window-index-reset-design.md

@@ -0,0 +1,134 @@
+# Reset tmux window indexes
+
+**Date:** 2026-06-11
+**Status:** Approved
+
+## Goal
+
+Add reusable functionality that compacts tmux window indexes inside the `hqt`
+tmux session so keyboard shortcuts such as `Alt+1`, `Alt+2`, and `Alt+3` stay
+useful after windows have been closed.
+
+This reset affects only tmux window indexes. It does not change database session
+IDs, `tmux_session_name`, harness conversation IDs, nicknames, panes, or running
+processes.
+
+## Behavior
+
+The reset operation works on the configured `hqt` tmux session:
+
+- The home/TUI window always ends at tmux index `0`.
+- Every non-home window keeps its current name, pane, process, and content.
+- Non-home windows are sorted by their current tmux index.
+- Those non-home windows are moved into contiguous indexes starting at `1`.
+
+Example:
+
+```text
+0: ⌂ HQT
+1: hqt-foo
+4: hqt-bar
+9: hqt-baz
+```
+
+becomes:
+
+```text
+0: ⌂ HQT
+1: hqt-foo
+2: hqt-bar
+3: hqt-baz
+```
+
+If the home window is not already at `0`, it is moved to `0` first, then the
+remaining windows are compacted after it in their original index order.
+
+## Architecture
+
+Add the reset at the tmux layer rather than the session or database layer.
+Existing app and service code targets windows by name, so changing tmux indexes
+should not disturb attach, stop, delete, status polling, or label syncing.
+
+### Runner
+
+Add a low-level method on `TmuxRunner`, for example:
+
+```python
+async def reset_window_indexes(self, home_window: str) -> bool:
+    ...
+```
+
+The method should:
+
+1. List windows for `self.session_name` with `#{window_id}`,
+   `#{window_index}`, and `#{window_name}`.
+2. Find the row whose name matches `home_window`.
+3. Move the home window to index `0` if needed.
+4. Sort all other windows by their original index.
+5. Move each non-home window to `1..N` in that order.
+
+Use `move-window` with `window_id` as the source target so changing one index
+cannot make a later move target the wrong window. The destination should target
+the configured tmux session and explicit index.
+
+The implementation should be idempotent: running it against an already compact
+session should return success without changing anything meaningful.
+
+### Manager
+
+Add a thin delegate on `TmuxManager`, for example:
+
+```python
+async def reset_window_indexes(self, home_window: str) -> bool:
+    return await self.runner.reset_window_indexes(home_window)
+```
+
+No `SessionService` change is needed for the initial implementation. This is a
+tmux maintenance operation, not a session lifecycle operation.
+
+### Command Surface
+
+Implement the reusable operation now and defer user-facing wiring. A command
+palette is planned separately, so this feature should expose a callable method
+that the future command can invoke.
+
+Do not add a TUI keybinding or footer item in this change.
+
+## Error Handling
+
+- If the configured tmux session does not exist or cannot be listed, return
+  failure and leave the database untouched.
+- If the home window is missing, return failure rather than guessing which
+  window should become index `0`.
+- If a `move-window` command fails, return failure. The operation may have
+  partially moved tmux windows before the failure; this is acceptable because
+  tmux indexes are maintenance metadata and no persisted hqt state is changed.
+- Log tmux stderr for failed operations to aid debugging.
+
+## Testing
+
+Add focused tests in `tests/test_tmux.py`:
+
+- Missing home window returns failure and performs no moves.
+- Already compact windows return success.
+- Sparse indexes compact in current index order:
+  `0 home, 1 foo, 4 bar, 9 baz` moves `bar` to `2` and `baz` to `3`.
+- Home initially not at `0` is moved to `0`, and all other windows compact to
+  `1..N` in original order.
+- Moves target stable window identities, not mutable index-only targets.
+
+The implementation must pass the project checks after code changes:
+
+```bash
+uv run ruff format src tests
+uv run ruff check src tests
+uv run ty check
+```
+
+## Out of Scope
+
+- Changing database primary keys.
+- Changing `Session.tmux_session_name`.
+- Changing harness session IDs or resume behavior.
+- Adding command palette UI, keybindings, or footer labels.
+- Enabling permanent tmux automatic renumbering.

src/hqt/cli.py

@@ -1,6 +1,7 @@
 import logging
 import os
 import shlex
+import shutil
 import subprocess
 import sys
 import traceback
@@ -106,7 +107,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")
@@ -120,6 +121,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")
 
 
 def _build_session_service():

src/hqt/db/migrations.py

@@ -18,8 +18,32 @@ def _migrate_v2(conn: Connection) -> None:
     conn.exec_driver_sql("ALTER TABLE sessions ADD COLUMN worktree_branch TEXT")
 
 
+def _migrate_v3(conn: Connection) -> None:
+    conn.exec_driver_sql(
+        """
+        CREATE TABLE scheduled_prompts (
+            id INTEGER NOT NULL PRIMARY KEY,
+            session_id INTEGER NOT NULL UNIQUE,
+            prompt TEXT NOT NULL,
+            due_at DATETIME NOT NULL,
+            status VARCHAR NOT NULL,
+            error TEXT,
+            created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            sent_at DATETIME,
+            FOREIGN KEY(session_id) REFERENCES sessions (id)
+        )
+        """
+    )
+
+
+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

src/hqt/db/models.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from sqlalchemy import ForeignKey, String, func
+from sqlalchemy import ForeignKey, String, Text, func
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 
 
@@ -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())
@@ -45,6 +46,25 @@ class Session(Base):
     archived: Mapped[bool] = mapped_column(default=False)
     project: Mapped["Project"] = relationship(back_populates="sessions")
     harness: Mapped["Harness"] = relationship(back_populates="sessions")
+    scheduled_prompt: Mapped["ScheduledPrompt | None"] = relationship(
+        back_populates="session"
+    )
+
+
+class ScheduledPrompt(Base):
+    __tablename__ = "scheduled_prompts"
+    id: Mapped[int] = mapped_column(primary_key=True)
+    session_id: Mapped[int] = mapped_column(
+        ForeignKey("sessions.id"),
+        unique=True,
+    )
+    prompt: Mapped[str] = mapped_column(Text)
+    due_at: Mapped[datetime]
+    status: Mapped[str] = mapped_column(String, default="pending")
+    error: Mapped[str | None] = mapped_column(Text, default=None)
+    created_at: Mapped[datetime] = mapped_column(insert_default=func.now())
+    sent_at: Mapped[datetime | None] = mapped_column(default=None)
+    session: Mapped["Session"] = relationship(back_populates="scheduled_prompt")
 
 
 class McpServer(Base):

src/hqt/harnesses/base.py

@@ -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
 

src/hqt/harnesses/configurators/claude.py

@@ -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:

src/hqt/harnesses/configurators/codex.py

@@ -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:

src/hqt/harnesses/configurators/generic.py

@@ -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)

src/hqt/harnesses/configurators/kiro.py

@@ -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)

src/hqt/sandbox.py

@@ -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

src/hqt/sessions/service.py

@@ -1,9 +1,10 @@
 import asyncio
 import contextlib
+import json
 import logging
 import shutil
 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
@@ -11,11 +12,13 @@ from pathlib import Path
 from sqlalchemy.orm import Session as DBSession
 from sqlalchemy.orm import selectinload, sessionmaker
 
-from hqt.db.models import Harness, Project, Session
+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.tools import TOOLS
 from hqt.tmux.manager import SpawnRequest, TmuxManager
@@ -26,6 +29,10 @@ log = logging.getLogger(__name__)
 CAPTURE_MAX_ATTEMPTS = 6
 CAPTURE_RETRY_INTERVAL = 0.5
 ACTIVITY_RECENT_SECS = 5
+PROMPT_PENDING = "pending"
+PROMPT_SENDING = "sending"
+PROMPT_SENT = "sent"
+PROMPT_FAILED = "failed"
 
 
 async def _capture_sleep(t: float) -> None:
@@ -37,6 +44,7 @@ class SessionInfo:
     session: Session
     alive: bool
     status: str = "dead"
+    scheduled_prompt: ScheduledPrompt | None = None
 
 
 @dataclass
@@ -48,6 +56,14 @@ class CreateSessionResult:
     spawn_error: str = ""
 
 
+@dataclass
+class DispatchResult:
+    session_id: int
+    window_name: str
+    ok: bool
+    error: str = ""
+
+
 class SessionService:
     """Session lifecycle operations.
 
@@ -161,6 +177,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:
@@ -194,6 +255,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.
@@ -240,13 +302,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):
@@ -254,7 +328,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,
                     )
@@ -292,44 +366,64 @@ class SessionService:
                 raise ServiceError("Session not found")
             window_name = sess.tmux_session_name
 
-            # Rung 0: a worktree session whose directory was removed from disk
+            await self._prepare_session_for_resume(db, sess)
-            # but whose branch still exists can be recreated before anything
+            ok = await self._ensure_session_alive(db, sess, window_name)
-            # else runs. If recreation fails, the branch is gone too.
-            if sess.worktree_path and not Path(sess.worktree_path).exists():
-                branch = sess.worktree_branch
-                if branch is None:
-                    raise ServiceError(
-                        "Worktree session is missing its branch; delete the session"
-                    )
-                project_path = self._project_path(db, sess.project_id)
-                try:
-                    await worktree.add_worktree_for_branch(project_path, branch)
-                except ServiceError as exc:
-                    raise ServiceError(
-                        "Worktree and branch are gone; delete the session"
-                    ) from exc
-
-            self._maybe_capture_missing_session_id(db, sess)
-
-            if await self.tmux.is_alive(window_name):
-                # Window exists, process running — just switch
-                return await self.tmux.attach(window_name)
-
-            # Window exists but pane is dead, OR window is completely gone
-            log.info(
-                "Session %s not alive, respawning with fallback ladder", window_name
-            )
-            ok = await self._respawn_with_fallback(db, sess, window_name)
             if not ok:
                 return False
             return await self.tmux.attach(window_name)
 
+    async def _prepare_session_for_resume(self, db: DBSession, sess: Session) -> None:
+        """Run pre-resume repair/capture work without selecting the window."""
+        # Rung 0: a worktree session whose directory was removed from disk
+        # but whose branch still exists can be recreated before anything
+        # else runs. If recreation fails, the branch is gone too.
+        if sess.worktree_path and not Path(sess.worktree_path).exists():
+            branch = sess.worktree_branch
+            if branch is None:
+                raise ServiceError(
+                    "Worktree session is missing its branch; delete the session"
+                )
+            project_path = self._project_path(db, sess.project_id)
+            try:
+                await worktree.add_worktree_for_branch(project_path, branch)
+            except ServiceError as exc:
+                raise ServiceError(
+                    "Worktree and branch are gone; delete the session"
+                ) from exc
+
+        self._maybe_capture_missing_session_id(db, sess)
+
+    async def _ensure_session_alive(
+        self, db: DBSession, sess: Session, window_name: str
+    ) -> bool:
+        """Ensure a session's tmux pane is alive without selecting the window."""
+        ok, _error = await self._ensure_session_alive_result(db, sess, window_name)
+        return ok
+
+    async def _ensure_session_alive_result(
+        self, db: DBSession, sess: Session, window_name: str
+    ) -> tuple[bool, str]:
+        """Ensure a session is alive, returning the respawn error on failure."""
+        if await self.tmux.is_alive(window_name):
+            return True, ""
+        log.info(
+            "Session %s not alive, respawning with fallback ladder",
+            window_name,
+        )
+        return await self._respawn_with_fallback_result(db, sess, window_name)
+
     async def _respawn_with_fallback(
         self, db: DBSession, sess: Session, window_name: str
     ) -> bool:
+        ok, _error = await self._respawn_with_fallback_result(db, sess, window_name)
+        return ok
+
+    async def _respawn_with_fallback_result(
+        self, db: DBSession, sess: Session, window_name: str
+    ) -> tuple[bool, str]:
         """Try resume config first; if it dies, fall back to fresh spawn config.
 
-        Returns True if either rung succeeded, False if both failed.
+        Returns (True, "") if either rung succeeded, otherwise (False, error).
 
         On rung-2 success, if the harness captures_session_id, a capture-retry
         loop updates sess.harness_session_id so later resumes target the fresh
@@ -341,7 +435,7 @@ class SessionService:
             window_name, resume_cfg.command, str(resume_cfg.cwd), env=resume_cfg.env
         )
         if result.ok:
-            return True
+            return True, ""
 
         log.warning(
             "Resume failed for %s (%s), falling back to fresh spawn",
@@ -355,21 +449,30 @@ 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:
+                error = result.error or "respawn failed"
                 log.error(
                     "Fallback spawn also failed for %s: %s",
                     window_name,
                     result.error or "(no output)",
                 )
-                return False
+                return False, error
 
             # Rung-2 succeeded — capture the new session id if the harness
             # supports it so that future resumes target this fresh conversation.
@@ -386,7 +489,7 @@ class SessionService:
                         window_name,
                         sess.harness_session_id,
                     )
-        return True
+        return True, ""
 
     def session_id_for_window(self, window_name: str) -> int | None:
         """Resolve a tmux window name to its active hqt session id, or None.
@@ -495,6 +598,173 @@ class SessionService:
             sess.nickname = (nickname or "").strip() or None
             db.commit()
 
+    def schedule_prompt(
+        self, session_id: int, prompt: str, due_at: datetime
+    ) -> ScheduledPrompt:
+        """Create or replace the pending scheduled prompt for a session."""
+        prompt = prompt.strip()
+        if not prompt:
+            raise ServiceError("Prompt cannot be empty")
+        with self.factory() as db:
+            sess = db.get(Session, session_id)
+            if sess is None:
+                raise ServiceError("Session not found")
+            scheduled = (
+                db.query(ScheduledPrompt).filter_by(session_id=session_id).one_or_none()
+            )
+            if scheduled is None:
+                scheduled = ScheduledPrompt(session_id=session_id, prompt=prompt)
+                db.add(scheduled)
+            else:
+                scheduled.prompt = prompt
+            scheduled.due_at = due_at
+            scheduled.status = PROMPT_PENDING
+            scheduled.error = None
+            scheduled.sent_at = None
+            db.commit()
+            db.refresh(scheduled)
+            db.expunge(scheduled)
+            return scheduled
+
+    def _mark_prompt_failed(
+        self, db: DBSession, row: ScheduledPrompt, error: str
+    ) -> DispatchResult:
+        row.status = PROMPT_FAILED
+        row.error = error
+        db.commit()
+        session = db.get(Session, row.session_id)
+        window_name = session.tmux_session_name if session is not None else ""
+        return DispatchResult(
+            session_id=row.session_id,
+            window_name=window_name,
+            ok=False,
+            error=error,
+        )
+
+    def _dispatch_error_message(self, exc: Exception) -> str:
+        return str(exc) or exc.__class__.__name__
+
+    async def dispatch_due_prompts(
+        self, now: datetime | None = None
+    ) -> list[DispatchResult]:
+        """Dispatch due pending prompts once; failed prompts are not retried."""
+        if now is None:
+            now = datetime.now()
+        results: list[DispatchResult] = []
+        with self.factory() as db:
+            rows = (
+                db.query(ScheduledPrompt)
+                .filter(
+                    ScheduledPrompt.status == PROMPT_PENDING,
+                    ScheduledPrompt.due_at <= now,
+                )
+                .all()
+            )
+            for row in rows:
+                row.status = PROMPT_SENDING
+                row.error = None
+                db.commit()
+
+                sess = db.get(
+                    Session,
+                    row.session_id,
+                    options=[selectinload(Session.harness)],
+                )
+                if sess is None:
+                    results.append(
+                        self._mark_prompt_failed(db, row, "Session not found")
+                    )
+                    continue
+
+                window_name = sess.tmux_session_name
+                try:
+                    await self._prepare_session_for_resume(db, sess)
+                    alive, error = await self._ensure_session_alive_result(
+                        db, sess, window_name
+                    )
+                except Exception as exc:
+                    results.append(
+                        self._mark_prompt_failed(
+                            db, row, self._dispatch_error_message(exc)
+                        )
+                    )
+                    continue
+                if not alive:
+                    results.append(
+                        self._mark_prompt_failed(
+                            db,
+                            row,
+                            error or "Failed to resume session before prompt",
+                        )
+                    )
+                    continue
+
+                try:
+                    send_text_ok = await self.tmux.send_text(window_name, row.prompt)
+                except Exception as exc:
+                    results.append(
+                        self._mark_prompt_failed(
+                            db, row, self._dispatch_error_message(exc)
+                        )
+                    )
+                    continue
+                if not send_text_ok:
+                    results.append(
+                        self._mark_prompt_failed(db, row, "send text failed")
+                    )
+                    continue
+                try:
+                    send_enter_ok = await self.tmux.send_enter(window_name)
+                except Exception as exc:
+                    results.append(
+                        self._mark_prompt_failed(
+                            db, row, self._dispatch_error_message(exc)
+                        )
+                    )
+                    continue
+                if not send_enter_ok:
+                    results.append(
+                        self._mark_prompt_failed(db, row, "send enter failed")
+                    )
+                    continue
+
+                row.status = PROMPT_SENT
+                row.sent_at = now
+                row.error = None
+                db.commit()
+                results.append(
+                    DispatchResult(
+                        session_id=sess.id,
+                        window_name=window_name,
+                        ok=True,
+                    )
+                )
+        return results
+
+    def cancel_scheduled_prompt(self, session_id: int) -> None:
+        """Delete any scheduled prompt for a session."""
+        with self.factory() as db:
+            scheduled = (
+                db.query(ScheduledPrompt).filter_by(session_id=session_id).one_or_none()
+            )
+            if scheduled is None:
+                return
+            db.delete(scheduled)
+            db.commit()
+
+    def get_scheduled_prompt(self, session_id: int) -> ScheduledPrompt | None:
+        """Return a session's pending scheduled prompt, if one exists."""
+        with self.factory() as db:
+            scheduled = (
+                db.query(ScheduledPrompt)
+                .filter_by(session_id=session_id, status=PROMPT_PENDING)
+                .one_or_none()
+            )
+            if scheduled is None:
+                return None
+            db.expunge(scheduled)
+            return scheduled
+
     async def delete_session(
         self, session_id: int, remove_worktree: bool = False, force: bool = False
     ) -> None:
@@ -529,6 +799,11 @@ class SessionService:
                     force,
                 )
 
+            scheduled = (
+                db.query(ScheduledPrompt).filter_by(session_id=session_id).one_or_none()
+            )
+            if scheduled is not None:
+                db.delete(scheduled)
             db.delete(sess)
             db.commit()
 
@@ -571,6 +846,21 @@ class SessionService:
         age = now - wi.last_activity
         return "active" if age <= ACTIVITY_RECENT_SECS else "idle"
 
+    def _pending_prompt_map(
+        self, db: DBSession, session_ids: list[int]
+    ) -> dict[int, ScheduledPrompt]:
+        if not session_ids:
+            return {}
+        prompts = (
+            db.query(ScheduledPrompt)
+            .filter(
+                ScheduledPrompt.session_id.in_(session_ids),
+                ScheduledPrompt.status == PROMPT_PENDING,
+            )
+            .all()
+        )
+        return {prompt.session_id: prompt for prompt in prompts}
+
     async def list_sessions(
         self, project_id: int, now: float | None = None
     ) -> list[SessionInfo]:
@@ -585,13 +875,19 @@ class SessionService:
             )
             names = [s.tmux_session_name for s in sessions]
             info_map = await self.tmux.poll_info(names)
+            prompt_map = self._pending_prompt_map(db, [s.id for s in sessions])
             result: list[SessionInfo] = []
             for s in sessions:
                 self._maybe_capture_missing_session_id(db, s)
                 wi = info_map.get(s.tmux_session_name)
                 status = await self._status_for(s, wi, now)
                 result.append(
-                    SessionInfo(session=s, alive=bool(wi and wi.alive), status=status)
+                    SessionInfo(
+                        session=s,
+                        alive=bool(wi and wi.alive),
+                        status=status,
+                        scheduled_prompt=prompt_map.get(s.id),
+                    )
                 )
             return result
 
@@ -615,13 +911,21 @@ class SessionService:
             )
             names = [s.tmux_session_name for s in sessions]
             info_map = await self.tmux.poll_info(names)
+            prompt_map = self._pending_prompt_map(db, [s.id for s in sessions])
             result: list[SessionInfo] = []
             for s in sessions:
                 self._maybe_capture_missing_session_id(db, s)
                 wi = info_map.get(s.tmux_session_name)
                 alive = bool(wi and wi.alive)
                 status = await self._status_for(s, wi, now)
-                result.append(SessionInfo(session=s, alive=alive, status=status))
+                result.append(
+                    SessionInfo(
+                        session=s,
+                        alive=alive,
+                        status=status,
+                        scheduled_prompt=prompt_map.get(s.id),
+                    )
+                )
                 name = s.nickname or s.tmux_session_name
                 await self.tmux.set_window_label(
                     s.tmux_session_name, window_label(status, name, alive=alive)
@@ -633,6 +937,18 @@ 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,
+            self._worktree_git_binds(db, sess),
+        )
+        return SpawnConfig(command=command, env=cfg.env, cwd=cfg.cwd)

src/hqt/tmux/manager.py

@@ -113,10 +113,22 @@ class TmuxManager:
         """Return the visible text of the named window's pane."""
         return await self.runner.capture_pane(window_name)
 
+    async def send_text(self, window_name: str, text: str) -> bool:
+        """Send literal text to a harness window."""
+        return await self.runner.send_text(window_name, text)
+
+    async def send_enter(self, window_name: str) -> bool:
+        """Send Enter to a harness window."""
+        return await self.runner.send_enter(window_name)
+
     async def set_window_label(self, window_name: str, label: str) -> None:
         """Set the status-bar label for a window (status symbol + name)."""
         await self.runner.set_window_label(window_name, label)
 
+    async def reset_window_indexes(self, home_window: str) -> bool:
+        """Compact tmux window indexes while keeping the home window at 0."""
+        return await self.runner.reset_window_indexes(home_window)
+
     async def open_aux_window(
         self, name: str, cwd: str, command: list[str], label: str
     ) -> str | None:

src/hqt/tmux/runner.py

@@ -117,6 +117,15 @@ class WindowInfo:
     last_activity: int
 
 
+@dataclass(frozen=True)
+class WindowIndex:
+    """Stable tmux window identity plus its current numeric index."""
+
+    window_id: str
+    index: int
+    name: str
+
+
 class TmuxRunner:
     """Low-level async tmux command wrapper. All operations target windows within a single session."""
 
@@ -232,6 +241,104 @@ class TmuxRunner:
             flags.extend(["-e", f"{key}={value}"])
         return flags
 
+    async def _list_window_indexes(self) -> list[WindowIndex] | None:
+        """Return window ids, indexes, and names for this tmux session.
+
+        None means tmux failed or returned malformed data; an empty list means
+        the session listed successfully but has no windows.
+        """
+        rc, stdout, err = await self._exec(
+            "list-windows",
+            "-t",
+            self.session_name,
+            "-F",
+            "#{window_id}\t#{window_index}\t#{window_name}",
+        )
+        if rc != 0:
+            log.error("list-windows failed for %s: %s", self.session_name, err)
+            return None
+
+        windows: list[WindowIndex] = []
+        for line in stdout.splitlines():
+            if not line.strip():
+                continue
+            parts = line.split("\t", 2)
+            if len(parts) != 3:
+                log.error("Malformed list-windows row: %r", line)
+                return None
+            window_id, index_text, name = parts
+            try:
+                index = int(index_text)
+            except ValueError:
+                log.error("Malformed window index in row: %r", line)
+                return None
+            windows.append(WindowIndex(window_id=window_id, index=index, name=name))
+        return windows
+
+    async def _move_window_id_to_index(self, window_id: str, index: int) -> bool:
+        rc, _, err = await self._exec(
+            "move-window",
+            "-d",
+            "-s",
+            window_id,
+            "-t",
+            f"{self.session_name}:{index}",
+        )
+        if rc != 0:
+            log.error("move-window failed for %s -> %s: %s", window_id, index, err)
+            return False
+        return True
+
+    async def reset_window_indexes(self, home_window: str) -> bool:
+        """Compact tmux window indexes while keeping the home window at 0.
+
+        The operation changes tmux indexes only. It preserves window names,
+        panes, processes, hqt database rows, and harness session ids.
+        """
+        windows = await self._list_window_indexes()
+        if windows is None:
+            return False
+
+        home = next((window for window in windows if window.name == home_window), None)
+        if home is None:
+            log.error(
+                "Home window %r not found in session %s",
+                home_window,
+                self.session_name,
+            )
+            return False
+
+        non_home = sorted(
+            (window for window in windows if window.window_id != home.window_id),
+            key=lambda window: window.index,
+        )
+        desired: list[tuple[WindowIndex, int]] = [(home, 0)]
+        desired.extend(
+            (window, index) for index, window in enumerate(non_home, start=1)
+        )
+
+        if all(window.index == target_index for window, target_index in desired):
+            return True
+
+        # Move every window out of the target range before placing final indexes.
+        # This avoids move-window failures when a destination index is occupied.
+        temp_start = max(window.index for window in windows) + len(windows)
+        for offset, window in enumerate(
+            sorted(windows, key=lambda item: item.index),
+            start=1,
+        ):
+            if not await self._move_window_id_to_index(
+                window.window_id,
+                temp_start + offset,
+            ):
+                return False
+
+        for window, target_index in desired:
+            if not await self._move_window_id_to_index(window.window_id, target_index):
+                return False
+
+        return True
+
     async def new_window(
         self,
         window_name: str,
@@ -351,8 +458,10 @@ class TmuxRunner:
             return None
         window_id = stdout.strip()
 
-        # Style by window_id: automatic-rename off, the @hqt_label, then the
+        # Style by window_id: automatic-rename + allow-rename off (nvim/lazygit
-        # Frappé per-window theme — one atomic invocation (";" argv separators).
+        # emit OSC title sequences that would otherwise scramble the label), the
+        # @hqt_label, then the Frappé per-window theme — one atomic invocation
+        # (";" argv separators). Mirrors new_window's window-option setup.
         rc, _, err = await self._exec(
             "set-option",
             "-w",
@@ -365,6 +474,13 @@ class TmuxRunner:
             "-w",
             "-t",
             window_id,
+            "allow-rename",
+            "off",
+            ";",
+            "set-option",
+            "-w",
+            "-t",
+            window_id,
             "@hqt_label",
             label,
             ";",
@@ -550,11 +666,29 @@ class TmuxRunner:
             for name in alive_map
         }
 
-    async def send_text(self, window_name: str, text: str) -> None:
+    async def send_text(self, window_name: str, text: str) -> bool:
-        await self._exec("send-keys", "-t", self._target(window_name), "-l", "--", text)
+        rc, _, err = await self._exec(
+            "send-keys",
+            "-t",
+            self._target(window_name),
+            "-l",
+            "--",
+            text,
+        )
+        if rc != 0:
+            log.error("send_text failed for %s: %s", window_name, err)
+        return rc == 0
 
-    async def send_enter(self, window_name: str) -> None:
+    async def send_enter(self, window_name: str) -> bool:
-        await self._exec("send-keys", "-t", self._target(window_name), "Enter")
+        rc, _, err = await self._exec(
+            "send-keys",
+            "-t",
+            self._target(window_name),
+            "Enter",
+        )
+        if rc != 0:
+            log.error("send_enter failed for %s: %s", window_name, err)
+        return rc == 0
 
     async def capture_pane(self, window_name: str) -> str:
         _, stdout, _ = await self._exec(

src/hqt/tui/app.py

@@ -1,5 +1,6 @@
 import logging
 from collections.abc import Coroutine
+from datetime import datetime
 from pathlib import Path
 from typing import Any
 
@@ -10,6 +11,8 @@ from textual.screen import ModalScreen
 from textual.theme import Theme
 from textual.widgets import Footer
 
+from hqt import sandbox
+from hqt.sandbox import SandboxPolicy
 from hqt.config import get_settings
 from hqt.git import worktree
 from hqt.db.engine import ensure_db, get_engine, get_session_factory
@@ -23,6 +26,7 @@ from hqt.tui.screens.confirm_delete import ConfirmDeleteScreen
 from hqt.tui.screens.project_form import ProjectFormScreen
 from hqt.tui.screens.new_session import NewSessionScreen
 from hqt.tui.screens.rename_session import RenameSessionScreen
+from hqt.tui.screens.schedule_prompt import SchedulePromptScreen
 from hqt.tui.widgets.project_list import ProjectList, ProjectSelected
 from hqt.tui.widgets.session_list import SessionList, SessionSelected
 
@@ -70,6 +74,8 @@ class HqtApp(App):
         Binding("enter", "attach_session", "Attach", priority=True),
         Binding("r", "rename_session", "Rename"),
         Binding("s", "stop_session", "Stop"),
+        Binding("p", "schedule_prompt", "Prompt Later"),
+        Binding("P", "cancel_scheduled_prompt", "Cancel Prompt"),
     ]
     CSS_PATH = "styles.tcss"
 
@@ -158,6 +164,16 @@ class HqtApp(App):
         # Sync every session's tmux window label (session-wide), then update the
         # UI for the selected project from the same computed status — no recompute.
         try:
+            dispatch_results = await self._sessions().dispatch_due_prompts()
+            for result in dispatch_results:
+                if result.ok:
+                    self.notify(f"Prompt sent to {result.window_name}")
+                else:
+                    target = result.window_name or f"session {result.session_id}"
+                    self.notify(
+                        f"Prompt failed for {target}: {result.error}",
+                        severity="error",
+                    )
             infos = await self._sessions().sync_window_labels()
         except ServiceError as err:
             log.warning("Session poll failed: %s", err)
@@ -255,10 +271,11 @@ 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, SandboxPolicy | None]
+            | 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 +284,7 @@ class HqtApp(App):
                         nickname,
                         model,
                         worktree_branch=worktree_branch,
+                        sandbox=sandbox_policy,
                     )
                     if not create_result.spawn_ok:
                         first_line = next(
@@ -296,7 +314,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())
 
@@ -402,3 +427,37 @@ class HqtApp(App):
                 await self._refresh_sessions()
 
             self._run_service_worker(_do())
+
+    def action_schedule_prompt(self) -> None:
+        sid = self.query_one(SessionList).get_selected_session_id()
+        if sid is None:
+            self.notify("Select a session first", severity="warning")
+            return
+
+        def on_dismiss(result) -> None:
+            if result is None:
+                return
+            prompt, delay = result
+            due_at = datetime.now() + delay
+            try:
+                self._sessions().schedule_prompt(sid, prompt, due_at)
+            except ServiceError as err:
+                self.notify(str(err), severity="error")
+                return
+            self.notify(f"Prompt scheduled for session {sid} at {due_at:%H:%M}")
+            self._run_service_worker(self._refresh_sessions())
+
+        self.push_screen(SchedulePromptScreen(), on_dismiss)
+
+    def action_cancel_scheduled_prompt(self) -> None:
+        sid = self.query_one(SessionList).get_selected_session_id()
+        if sid is None:
+            self.notify("Select a session first", severity="warning")
+            return
+        try:
+            self._sessions().cancel_scheduled_prompt(sid)
+        except ServiceError as err:
+            self.notify(str(err), severity="error")
+            return
+        self.notify(f"Scheduled prompt canceled for session {sid}")
+        self._run_service_worker(self._refresh_sessions())

src/hqt/tui/screens/new_session.py

@@ -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]):
+class NewSessionScreen(
-    def __init__(self, harness_names: list[str], is_git_repo: bool) -> None:
+    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)

src/hqt/tui/screens/schedule_prompt.py

@@ -0,0 +1,71 @@
+import re
+from datetime import timedelta
+
+from textual.app import ComposeResult
+from textual.containers import Horizontal, Vertical
+from textual.screen import ModalScreen
+from textual.widgets import Button, Input, Label
+
+
+_DELAY_RE = re.compile(
+    r"^(?:(?P<hours>\d+)h)?(?:(?P<minutes>\d+)m)?(?:(?P<seconds>\d+)s)?$"
+)
+
+
+def parse_delay(value: str) -> timedelta:
+    match = _DELAY_RE.fullmatch(value)
+    if match is None or not value:
+        raise ValueError("invalid delay")
+
+    total_seconds = (
+        int(match.group("hours") or 0) * 3600
+        + int(match.group("minutes") or 0) * 60
+        + int(match.group("seconds") or 0)
+    )
+    if total_seconds <= 0:
+        raise ValueError("invalid delay")
+    return timedelta(seconds=total_seconds)
+
+
+class SchedulePromptScreen(ModalScreen[tuple[str, timedelta] | None]):
+    def compose(self) -> ComposeResult:
+        with Vertical(id="schedule-prompt-dialog"):
+            yield Label("Schedule Prompt")
+            yield Label("Prompt:")
+            yield Input(placeholder="prompt to send later", id="prompt-input")
+            yield Label("Delay:")
+            yield Input(placeholder="30m", id="delay-input")
+            yield Label("", id="schedule-error")
+            with Horizontal(classes="dialog-actions"):
+                yield Button("OK", variant="primary", id="ok-btn")
+                yield Button("Cancel", id="cancel-btn")
+
+    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:
+        prompt_input = self.query_one("#prompt-input", Input)
+        prompt = prompt_input.value.strip()
+        if not prompt:
+            self._set_error("Prompt cannot be empty")
+            prompt_input.focus()
+            return
+
+        delay_input = self.query_one("#delay-input", Input)
+        try:
+            delay = parse_delay(delay_input.value.strip())
+        except ValueError:
+            self._set_error("Use a delay like 90s, 30m, 2h, or 1h30m")
+            delay_input.focus()
+            return
+
+        self.dismiss((prompt, delay))
+
+    def _set_error(self, message: str) -> None:
+        self.query_one("#schedule-error", Label).update(message)

src/hqt/tui/styles.tcss

@@ -23,11 +23,11 @@ SessionList {
     background: $background;
 }
 
-ProjectFormScreen, NewSessionScreen {
+ProjectFormScreen, NewSessionScreen, SchedulePromptScreen {
     align: center middle;
 }
 
-#project-form-dialog, #new-session-dialog {
+#project-form-dialog, #new-session-dialog, #schedule-prompt-dialog {
     width: 60;
     height: auto;
     padding: 1 2;
@@ -35,6 +35,11 @@ ProjectFormScreen, NewSessionScreen {
     border: solid $border;
 }
 
+#schedule-error {
+    color: $error;
+    height: 1;
+}
+
 /* Action row: flat, single-row buttons in a centered row. Textual's default
    Button uses `tall` top/bottom borders, which force a 3-row beveled "block"
    that reads as heavy and dated. Dropping the bevel (border: none) collapses

src/hqt/tui/widgets/session_list.py

@@ -1,3 +1,6 @@
+from datetime import datetime
+from math import ceil
+
 from rich.markup import escape
 from textual.app import ComposeResult
 from textual.message import Message
@@ -28,12 +31,23 @@ _STATUS_COLORS: dict[str, str] = {
 }
 
 
+def format_scheduled_prompt(due_at: datetime, now: datetime | None = None) -> str:
+    if now is None:
+        now = datetime.now()
+    seconds = max(0, (due_at - now).total_seconds())
+    if seconds < 3600:
+        return f"@ {max(1, ceil(seconds / 60))}m"
+    return f"@ {due_at.strftime('%H:%M')}"
+
+
 def format_session_text(
     nickname: str,
     harness_name: str,
     status_text: str,
     symbol: str,
     worktree_branch: str | None = None,
+    scheduled_prompt=None,
+    now: datetime | None = None,
 ) -> str:
     color = _STATUS_COLORS.get(status_text, "#c6d0f5")  # default: Text
     name = nickname
@@ -47,7 +61,10 @@ def format_session_text(
         else:
             name = f"{nickname} ⎇ {worktree_branch}"
     label = escape(f"{name} [{harness_name}]")
-    return f"[{color}]{symbol}[/] {label} {status_text}"
+    schedule = ""
+    if scheduled_prompt is not None:
+        schedule = f"  {format_scheduled_prompt(scheduled_prompt.due_at, now=now)}"
+    return f"[{color}]{symbol}[/] {label} {status_text}{schedule}"
 
 
 class SessionList(Widget, can_focus=False):
@@ -71,6 +88,7 @@ class SessionList(Widget, can_focus=False):
                 status_text,
                 symbol,
                 getattr(s, "worktree_branch", None),
+                scheduled_prompt=getattr(info, "scheduled_prompt", None),
             )
             new_items.append((f"sess-{s.id}", text, s.id))
 

tests/test_cli.py

@@ -1,9 +1,39 @@
+from unittest.mock import patch
+
 from click.testing import CliRunner
 
-from hqt import cli
+from hqt import cli, sandbox
 from hqt.config import Settings
 
 
+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.shutil, "which", side_effect=_which({"tmux", "bwrap"})),
+        patch.object(sandbox, "is_available", return_value=True),
+    ):
+        result = CliRunner().invoke(cli.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.shutil, "which", side_effect=_which({"tmux"})),
+        patch.object(sandbox, "is_available", return_value=False),
+    ):
+        result = CliRunner().invoke(cli.main, ["doctor"])
+    assert result.exit_code == 0
+    assert "bubblewrap: ✗" in result.output
+
+
 def test_tool_cmd_opens_tool_window(monkeypatch, tmp_path):
     monkeypatch.setattr(
         "hqt.config.get_settings", lambda: Settings(db_path=tmp_path / "t.db")

tests/test_db.py

@@ -95,8 +95,10 @@ def test_unversioned_existing_db_stamped_latest(tmp_path):
     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 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
 
@@ -113,25 +115,26 @@ def test_db_from_newer_hqt_raises(tmp_path):
 
 def test_pending_migrations_applied_in_order_once(tmp_path, monkeypatch):
     settings = _tmp_settings(tmp_path)
-    ensure_db(settings)  # stamps migrations.LATEST_VERSION (currently 2)
+    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,
-            (3, lambda conn: applied.append(3)),
+            (next_version, lambda conn: applied.append(next_version)),
-            (4, lambda conn: applied.append(4)),
+            (next_version + 1, lambda conn: applied.append(next_version + 1)),
         ],
     )
-    monkeypatch.setattr(migrations, "LATEST_VERSION", 4)
+    monkeypatch.setattr(migrations, "LATEST_VERSION", next_version + 1)
     engine = get_engine(settings)
     migrate(engine)
-    assert applied == [3, 4]
+    assert applied == [next_version, next_version + 1]
-    assert _user_version(engine) == 4
+    assert _user_version(engine) == next_version + 1
     migrate(engine)  # re-run is a no-op
-    assert applied == [3, 4]
+    assert applied == [next_version, next_version + 1]
 
 
 def _session_column_names(engine) -> list[str]:
@@ -140,6 +143,12 @@ def _session_column_names(engine) -> list[str]:
     return [row[1] for row in rows]
 
 
+def _scheduled_prompt_column_names(engine) -> list[str]:
+    with engine.connect() as conn:
+        rows = conn.exec_driver_sql("PRAGMA table_info(scheduled_prompts)").fetchall()
+    return [row[1] for row in rows]
+
+
 def test_fresh_db_has_worktree_columns(tmp_path):
     settings = _tmp_settings(tmp_path)
     ensure_db(settings)
@@ -147,18 +156,21 @@ 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 _user_version(engine) == 2
+    assert "sandbox_json" in cols
+    assert _user_version(engine) == migrations.LATEST_VERSION
 
 
-def test_v1_db_upgraded_to_v2(tmp_path):
+def test_v1_db_upgraded_to_latest(tmp_path):
-    # Build a v1 database: create all tables, drop the two new columns, stamp v1.
+    # 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)
     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 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)
@@ -166,7 +178,79 @@ 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 _user_version(engine) == 2
+    assert "sandbox_json" in cols
+    assert "scheduled_prompts" in inspect(engine).get_table_names()
+    assert _user_version(engine) == migrations.LATEST_VERSION
+
+
+def test_fresh_db_has_scheduled_prompts_table(tmp_path):
+    settings = _tmp_settings(tmp_path)
+    ensure_db(settings)
+    engine = get_engine(settings)
+
+    tables = inspect(engine).get_table_names()
+    assert "scheduled_prompts" in tables
+    cols = _scheduled_prompt_column_names(engine)
+    assert cols == [
+        "id",
+        "session_id",
+        "prompt",
+        "due_at",
+        "status",
+        "error",
+        "created_at",
+        "sent_at",
+    ]
+    assert _user_version(engine) == migrations.LATEST_VERSION
+
+
+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 "sandbox_json" in _session_column_names(engine)
+    assert _user_version(engine) == migrations.LATEST_VERSION
+
+
+def test_scheduled_prompt_one_row_per_session(tmp_path):
+    from datetime import datetime, timedelta
+
+    from sqlalchemy.exc import IntegrityError
+
+    from hqt.db.models import ScheduledPrompt
+
+    settings = _tmp_settings(tmp_path)
+    ensure_db(settings)
+    factory = get_session_factory(get_engine(settings))
+
+    with factory() as session:
+        p = Project(name="sched-proj", path="/tmp/sched-proj")
+        h = Harness(name="claude-code-scheduled")
+        session.add_all([p, h])
+        session.flush()
+        s = Session(
+            project_id=p.id,
+            harness_id=h.id,
+            tmux_session_name="hqt-scheduled",
+        )
+        session.add(s)
+        session.flush()
+        due = datetime.now() + timedelta(minutes=30)
+        session.add(ScheduledPrompt(session_id=s.id, prompt="continue", due_at=due))
+        session.commit()
+
+        session.add(ScheduledPrompt(session_id=s.id, prompt="again", due_at=due))
+        with pytest.raises(IntegrityError):
+            session.commit()
 
 
 def test_session_worktree_fields_default_none(tmp_path):
@@ -215,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}'

tests/test_harnesses.py

@@ -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"]

tests/test_sandbox.py

@@ -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

tests/test_sessions.py

@@ -1,4 +1,5 @@
 import pytest
+from datetime import datetime, timedelta
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -6,8 +7,9 @@ from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy.pool import StaticPool
 
-from hqt.db.models import Base, Harness, Project
+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
 
@@ -43,6 +45,8 @@ def tmux():
     m.poll_info = AsyncMock(return_value={})
     m.capture_pane = AsyncMock(return_value="")
     m.attach = AsyncMock()
+    m.send_text = AsyncMock(return_value=True)
+    m.send_enter = AsyncMock(return_value=True)
     m.window_exists = AsyncMock(return_value=True)
     m.respawn_verified = AsyncMock(
         return_value=SpawnResult(ok=True, window_id=None, error="")
@@ -93,6 +97,367 @@ async def test_list_sessions(service, db, tmux):
     assert result[0].alive is True
 
 
+def test_schedule_prompt_creates_pending_row(service, db):
+    sess = _seed_session(db)
+    due_at = datetime(2026, 6, 11, 12)
+
+    prompt = service.schedule_prompt(sess.id, "  run the tests  ", due_at)
+
+    assert prompt.session_id == sess.id
+    assert prompt.prompt == "run the tests"
+    assert prompt.due_at == due_at
+    assert prompt.status == "pending"
+    assert prompt.error is None
+    rows = db.query(ScheduledPrompt).all()
+    assert len(rows) == 1
+
+
+def test_schedule_prompt_replaces_existing_row_for_session(service, db):
+    sess = _seed_session(db)
+    first_due = datetime(2026, 6, 11, 12)
+    second_due = datetime(2026, 6, 11, 13)
+    first = service.schedule_prompt(sess.id, "old prompt", first_due)
+
+    second = service.schedule_prompt(sess.id, "new prompt", second_due)
+
+    rows = db.query(ScheduledPrompt).all()
+    assert len(rows) == 1
+    assert second.id == first.id
+    assert rows[0].id == first.id
+    assert rows[0].prompt == "new prompt"
+    assert rows[0].due_at == second_due
+    assert rows[0].status == "pending"
+    assert rows[0].error is None
+
+
+def test_schedule_prompt_rejects_empty_prompt(service):
+    due_at = datetime(2026, 6, 11, 12)
+
+    with pytest.raises(ServiceError, match="Prompt cannot be empty"):
+        service.schedule_prompt(1, "   ", due_at)
+
+
+def test_schedule_prompt_rejects_missing_session(service):
+    due_at = datetime(2026, 6, 11, 12)
+
+    with pytest.raises(ServiceError, match="Session not found"):
+        service.schedule_prompt(999, "hello", due_at)
+
+
+def test_cancel_scheduled_prompt_deletes_row(service, db):
+    sess = _seed_session(db)
+    due_at = datetime(2026, 6, 11, 12)
+    service.schedule_prompt(sess.id, "run the tests", due_at)
+
+    service.cancel_scheduled_prompt(sess.id)
+
+    assert db.query(ScheduledPrompt).all() == []
+
+
+def test_get_scheduled_prompt_returns_pending_only(service, db):
+    sess = _seed_session(db)
+    due_at = datetime(2026, 6, 11, 12)
+    service.schedule_prompt(sess.id, "run the tests", due_at)
+
+    prompt = service.get_scheduled_prompt(sess.id)
+    assert prompt is not None
+    assert prompt.status == "pending"
+
+    row = db.query(ScheduledPrompt).filter_by(session_id=sess.id).one()
+    row.status = "failed"
+    db.commit()
+
+    assert service.get_scheduled_prompt(sess.id) is None
+
+
+@pytest.mark.asyncio
+async def test_list_sessions_includes_pending_scheduled_prompt(service, db, tmux):
+    from hqt.tmux.runner import WindowInfo
+
+    sess = _seed_session(db)
+    due_at = datetime(2026, 6, 11, 12)
+    service.schedule_prompt(sess.id, "run the tests", due_at)
+    tmux.poll_info.return_value = {
+        sess.tmux_session_name: WindowInfo(alive=True, last_activity=1000)
+    }
+
+    result = await service.list_sessions(project_id=1)
+
+    assert result[0].scheduled_prompt is not None
+    assert result[0].scheduled_prompt.prompt == "run the tests"
+    assert result[0].scheduled_prompt.status == "pending"
+
+
+@pytest.mark.asyncio
+async def test_dispatch_due_prompt_alive_window_sends_and_marks_sent(service, db, tmux):
+    sess = _seed_session(db)
+    due = datetime.now() - timedelta(seconds=1)
+    scheduled = service.schedule_prompt(sess.id, "continue", due)
+    tmux.is_alive = AsyncMock(return_value=True)
+
+    results = await service.dispatch_due_prompts(now=datetime.now())
+
+    assert len(results) == 1
+    assert results[0].ok is True
+    assert results[0].session_id == sess.id
+    assert results[0].window_name == "hqt-rename"
+    tmux.send_text.assert_awaited_once_with("hqt-rename", "continue")
+    tmux.send_enter.assert_awaited_once_with("hqt-rename")
+    db.expire_all()
+    row = db.get(ScheduledPrompt, scheduled.id)
+    assert row.status == "sent"
+    assert row.sent_at is not None
+
+
+@pytest.mark.asyncio
+async def test_dispatch_due_prompt_dead_window_resumes_without_attach(
+    service, db, tmux, harnesses
+):
+    sess = _seed_session(db)
+    service.schedule_prompt(sess.id, "continue", datetime.now() - timedelta(seconds=1))
+    tmux.is_alive = AsyncMock(return_value=False)
+    tmux.respawn_verified = AsyncMock(
+        return_value=SpawnResult(ok=True, window_id=None, error="")
+    )
+    tmux.attach = AsyncMock(return_value=True)
+    harnesses["claude-code"].build_resume_config = MagicMock(
+        return_value=MagicMock(
+            command=["claude", "--resume"], env={}, cwd=Path("/tmp/myproj")
+        )
+    )
+
+    results = await service.dispatch_due_prompts(now=datetime.now())
+
+    assert len(results) == 1
+    assert results[0].ok is True
+    tmux.respawn_verified.assert_awaited()
+    tmux.attach.assert_not_called()
+    tmux.send_text.assert_awaited_once_with("hqt-rename", "continue")
+    tmux.send_enter.assert_awaited_once_with("hqt-rename")
+
+
+@pytest.mark.asyncio
+async def test_dispatch_due_prompt_resume_failure_marks_failed(
+    service, db, tmux, harnesses
+):
+    sess = _seed_session(db)
+    scheduled = service.schedule_prompt(
+        sess.id, "continue", datetime.now() - timedelta(seconds=1)
+    )
+    tmux.is_alive = AsyncMock(return_value=False)
+    tmux.respawn_verified = AsyncMock(
+        return_value=SpawnResult(ok=False, window_id=None, error="resume failed")
+    )
+    harnesses["claude-code"].build_resume_config = MagicMock(
+        return_value=MagicMock(
+            command=["claude", "--resume"], env={}, cwd=Path("/tmp/myproj")
+        )
+    )
+    harnesses["claude-code"].build_spawn_config = MagicMock(
+        return_value=MagicMock(command=["claude"], env={}, cwd=Path("/tmp/myproj"))
+    )
+
+    results = await service.dispatch_due_prompts(now=datetime.now())
+
+    assert len(results) == 1
+    assert results[0].ok is False
+    assert "resume failed" in results[0].error
+    tmux.send_text.assert_not_called()
+    tmux.send_enter.assert_not_called()
+    db.expire_all()
+    row = db.get(ScheduledPrompt, scheduled.id)
+    assert row.status == "failed"
+    assert "resume failed" in row.error
+
+
+@pytest.mark.asyncio
+async def test_dispatch_due_prompt_send_failure_does_not_retry_next_poll(
+    service, db, tmux
+):
+    sess = _seed_session(db)
+    scheduled = service.schedule_prompt(
+        sess.id, "continue", datetime.now() - timedelta(seconds=1)
+    )
+    tmux.is_alive = AsyncMock(return_value=True)
+    tmux.send_text = AsyncMock(return_value=False)
+
+    first = await service.dispatch_due_prompts(now=datetime.now())
+    second = await service.dispatch_due_prompts(now=datetime.now())
+
+    assert len(first) == 1
+    assert first[0].ok is False
+    assert second == []
+    assert tmux.send_text.await_count == 1
+    tmux.send_enter.assert_not_called()
+    db.expire_all()
+    row = db.get(ScheduledPrompt, scheduled.id)
+    assert row.status == "failed"
+    assert "send text failed" in row.error
+
+
+@pytest.mark.asyncio
+async def test_dispatch_due_prompt_missing_worktree_recreates_without_attach(
+    factory, db, tmux, harnesses, fake_worktree
+):
+    missing = "/tmp/does-not-exist/.worktrees/feature-x"
+    sess = _seed_worktree_session(db, path=missing)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
+    service.schedule_prompt(sess.id, "continue", datetime.now() - timedelta(seconds=1))
+    harnesses["claude-code"].build_resume_config = MagicMock(
+        return_value=MagicMock(command=["claude"], env={}, cwd=Path(missing))
+    )
+    tmux.is_alive = AsyncMock(return_value=False)
+    tmux.respawn_verified = AsyncMock(
+        return_value=SpawnResult(ok=True, window_id=None, error="")
+    )
+    tmux.attach = AsyncMock(return_value=True)
+
+    results = await service.dispatch_due_prompts(now=datetime.now())
+
+    assert results[0].ok is True
+    assert fake_worktree.add_calls == [(Path("/tmp/myproj"), "feature-x")]
+    tmux.attach.assert_not_called()
+    tmux.send_text.assert_awaited_once_with("hqt-wt", "continue")
+
+
+@pytest.mark.asyncio
+async def test_dispatch_due_prompt_late_capture_updates_id_before_resume(
+    factory, db, tmux, harnesses_capture_fallback
+):
+    harness = harnesses_capture_fallback["claude-code"]
+    harness.capture_session_id.return_value = "late-codex-id"
+    sess = _seed_session(db)
+    sess.harness_session_id = "placeholder-id"
+    db.commit()
+    service = SessionService(
+        factory=factory, tmux=tmux, harnesses=harnesses_capture_fallback
+    )
+    service.schedule_prompt(sess.id, "continue", datetime.now() - timedelta(seconds=1))
+    tmux.is_alive = AsyncMock(return_value=False)
+    tmux.respawn_verified = AsyncMock(
+        return_value=SpawnResult(ok=True, window_id="@2", error="")
+    )
+
+    results = await service.dispatch_due_prompts(now=datetime.now())
+
+    assert results[0].ok is True
+    harness.build_resume_config.assert_called_with(
+        Path("/tmp/myproj"), "late-codex-id", sess.model, sandboxed=False
+    )
+    db.expire_all()
+    assert db.get(Session, sess.id).harness_session_id == "late-codex-id"
+
+
+@pytest.mark.asyncio
+async def test_dispatch_due_prompt_send_exception_marks_failed_and_does_not_retry(
+    service, db, tmux
+):
+    sess = _seed_session(db)
+    scheduled = service.schedule_prompt(
+        sess.id, "continue", datetime.now() - timedelta(seconds=1)
+    )
+    tmux.is_alive = AsyncMock(return_value=True)
+    tmux.send_text = AsyncMock(side_effect=RuntimeError("send exploded"))
+
+    first = await service.dispatch_due_prompts(now=datetime.now())
+    second = await service.dispatch_due_prompts(now=datetime.now())
+
+    assert len(first) == 1
+    assert first[0].ok is False
+    assert "send exploded" in first[0].error
+    assert second == []
+    tmux.send_text.assert_awaited_once()
+    tmux.send_enter.assert_not_called()
+    db.expire_all()
+    row = db.get(ScheduledPrompt, scheduled.id)
+    assert row.status == "failed"
+    assert "send exploded" in row.error
+
+
+@pytest.mark.asyncio
+async def test_dispatch_due_prompt_resume_exception_marks_failed_without_send(
+    service, db, tmux, harnesses
+):
+    sess = _seed_session(db)
+    scheduled = service.schedule_prompt(
+        sess.id, "continue", datetime.now() - timedelta(seconds=1)
+    )
+    harnesses["claude-code"].build_resume_config = MagicMock(
+        return_value=MagicMock(
+            command=["claude", "--resume"], env={}, cwd=Path("/tmp/myproj")
+        )
+    )
+    tmux.is_alive = AsyncMock(return_value=False)
+    tmux.respawn_verified = AsyncMock(side_effect=RuntimeError("resume exploded"))
+
+    results = await service.dispatch_due_prompts(now=datetime.now())
+
+    assert len(results) == 1
+    assert results[0].ok is False
+    assert "resume exploded" in results[0].error
+    tmux.send_text.assert_not_called()
+    tmux.send_enter.assert_not_called()
+    db.expire_all()
+    row = db.get(ScheduledPrompt, scheduled.id)
+    assert row.status == "failed"
+    assert "resume exploded" in row.error
+
+
+@pytest.mark.asyncio
+async def test_dispatch_due_prompt_send_enter_failure_marks_failed(service, db, tmux):
+    sess = _seed_session(db)
+    scheduled = service.schedule_prompt(
+        sess.id, "continue", datetime.now() - timedelta(seconds=1)
+    )
+    tmux.is_alive = AsyncMock(return_value=True)
+    tmux.send_enter = AsyncMock(return_value=False)
+
+    results = await service.dispatch_due_prompts(now=datetime.now())
+
+    assert len(results) == 1
+    assert results[0].ok is False
+    assert "send enter failed" in results[0].error
+    tmux.send_text.assert_awaited_once_with("hqt-rename", "continue")
+    tmux.send_enter.assert_awaited_once_with("hqt-rename")
+    db.expire_all()
+    row = db.get(ScheduledPrompt, scheduled.id)
+    assert row.status == "failed"
+    assert "send enter failed" in row.error
+
+
+@pytest.mark.asyncio
+async def test_dispatch_due_prompt_resume_fails_then_fallback_spawn_succeeds(
+    service, db, tmux, harnesses
+):
+    sess = _seed_session(db)
+    service.schedule_prompt(sess.id, "continue", datetime.now() - timedelta(seconds=1))
+    harnesses["claude-code"].build_resume_config = MagicMock(
+        return_value=MagicMock(
+            command=["claude", "--resume"], env={}, cwd=Path("/tmp/myproj")
+        )
+    )
+    harnesses["claude-code"].build_spawn_config = MagicMock(
+        return_value=MagicMock(
+            command=["claude", "--session-id", "sess-seeded"],
+            env={},
+            cwd=Path("/tmp/myproj"),
+        )
+    )
+    tmux.is_alive = AsyncMock(return_value=False)
+    tmux.respawn_verified = AsyncMock(
+        side_effect=[
+            SpawnResult(ok=False, window_id=None, error="resume failed"),
+            SpawnResult(ok=True, window_id="@2", error=""),
+        ]
+    )
+
+    results = await service.dispatch_due_prompts(now=datetime.now())
+
+    assert results[0].ok is True
+    assert tmux.respawn_verified.await_count == 2
+    tmux.send_text.assert_awaited_once_with("hqt-rename", "continue")
+
+
 @pytest.mark.asyncio
 async def test_stop_session(service, tmux):
     await service.create_session(project_id=1, harness_name="claude-code")
@@ -610,6 +975,27 @@ async def test_sync_window_labels_covers_all_projects(factory, db, tmux, harness
     assert labelled == {"hqt-1", "hqt-2"}
 
 
+@pytest.mark.asyncio
+async def test_sync_window_labels_includes_pending_scheduled_prompt(
+    factory, db, tmux, harnesses
+):
+    """sync_window_labels returns SessionInfo with pending scheduled prompt metadata."""
+    from hqt.tmux.runner import WindowInfo
+
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
+    result = await service.create_session(project_id=1, harness_name="claude-code")
+    due_at = datetime(2026, 6, 11, 12)
+    service.schedule_prompt(result.session.id, "run the tests", due_at)
+    tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=False, last_activity=0)}
+    tmux.set_window_label = AsyncMock()
+
+    infos = await service.sync_window_labels()
+
+    assert infos[0].scheduled_prompt is not None
+    assert infos[0].scheduled_prompt.prompt == "run the tests"
+    assert infos[0].scheduled_prompt.status == "pending"
+
+
 # ---------------------------------------------------------------------------
 # Item 3: real-configurator integration-seam test
 # ---------------------------------------------------------------------------
@@ -816,7 +1202,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
 
@@ -985,16 +1371,15 @@ async def test_create_session_result_propagates_spawn_failure(
 # ---------------------------------------------------------------------------
 
 
-def _seed_session(db, nickname=None):
+def _seed_session(db, project_id: int = 1, nickname=None) -> Session:
-    """Create a session for project 1 / harness 'claude-code' (seeded by the db fixture)."""
+    """Create a session for the seeded 'claude-code' harness."""
-    from hqt.db.models import Harness, Session
-
     harness = db.query(Harness).filter_by(name="claude-code").first()
     sess = Session(
-        project_id=1,
+        project_id=project_id,
         harness_id=harness.id,
         nickname=nickname,
         tmux_session_name="hqt-rename",
+        harness_session_id="sess-seeded",
         archived=False,
     )
     db.add(sess)
@@ -1406,6 +1791,20 @@ async def test_delete_session_remove_worktree_false_skips_removal(
     assert db.get(Session, sess_id) is None
 
 
+@pytest.mark.asyncio
+async def test_delete_session_deletes_scheduled_prompt(service, db):
+    sess = _seed_session(db)
+    sess_id = sess.id
+    due_at = datetime(2026, 6, 11, 12)
+    service.schedule_prompt(sess_id, "run the tests", due_at)
+
+    await service.delete_session(sess_id)
+
+    db.expire_all()
+    assert db.get(Session, sess_id) is None
+    assert db.query(ScheduledPrompt).filter_by(session_id=sess_id).one_or_none() is None
+
+
 @pytest.mark.asyncio
 async def test_worktree_state_for_plain_session_returns_none(
     service, db, tmux, harnesses, fake_worktree
@@ -1520,6 +1919,99 @@ async def test_create_session_no_lock_for_non_capturing_harness(service, db, tmu
     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
+
+
 # ---------------------------------------------------------------------------
 # Task 4: SessionService tool-window methods
 # ---------------------------------------------------------------------------

tests/test_tmux.py

@@ -139,6 +139,154 @@ async def test_list_windows(runner):
     assert result == ["hqt", "hqt-1", "hqt-2"]
 
 
+@pytest.mark.asyncio
+async def test_reset_window_indexes_missing_home_returns_false(runner):
+    runner._exec.return_value = (
+        0,
+        "@1\t1\thqt-foo\n@2\t4\thqt-bar\n",
+        "",
+    )
+
+    result = await runner.reset_window_indexes("⌂ HQT")
+
+    assert result is False
+    runner._exec.assert_called_once_with(
+        "list-windows",
+        "-t",
+        "hqt-main",
+        "-F",
+        "#{window_id}\t#{window_index}\t#{window_name}",
+    )
+
+
+@pytest.mark.asyncio
+async def test_reset_window_indexes_already_compact_no_moves(runner):
+    runner._exec.return_value = (
+        0,
+        "@0\t0\t⌂ HQT\n@1\t1\thqt-foo\n@2\t2\thqt-bar\n",
+        "",
+    )
+
+    result = await runner.reset_window_indexes("⌂ HQT")
+
+    assert result is True
+    assert runner._exec.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_reset_window_indexes_sparse_indexes_compact_in_current_order(runner):
+    runner._exec.side_effect = [
+        (
+            0,
+            "@0\t0\t⌂ HQT\n@1\t1\thqt-foo\n@4\t4\thqt-bar\n@9\t9\thqt-baz\n",
+            "",
+        ),
+        (0, "", ""),
+        (0, "", ""),
+        (0, "", ""),
+        (0, "", ""),
+        (0, "", ""),
+        (0, "", ""),
+        (0, "", ""),
+        (0, "", ""),
+    ]
+
+    result = await runner.reset_window_indexes("⌂ HQT")
+
+    assert result is True
+    assert runner._exec.call_count == 9
+    calls = [call.args for call in runner._exec.call_args_list]
+    assert calls[1] == ("move-window", "-d", "-s", "@0", "-t", "hqt-main:14")
+    assert calls[2] == ("move-window", "-d", "-s", "@1", "-t", "hqt-main:15")
+    assert calls[3] == ("move-window", "-d", "-s", "@4", "-t", "hqt-main:16")
+    assert calls[4] == ("move-window", "-d", "-s", "@9", "-t", "hqt-main:17")
+    assert calls[5] == ("move-window", "-d", "-s", "@0", "-t", "hqt-main:0")
+    assert calls[6] == ("move-window", "-d", "-s", "@1", "-t", "hqt-main:1")
+    assert calls[7] == ("move-window", "-d", "-s", "@4", "-t", "hqt-main:2")
+    assert calls[8] == ("move-window", "-d", "-s", "@9", "-t", "hqt-main:3")
+
+
+@pytest.mark.asyncio
+async def test_reset_window_indexes_home_not_zero_moves_home_first_in_final_layout(
+    runner,
+):
+    runner._exec.side_effect = [
+        (
+            0,
+            "@2\t2\thqt-foo\n@5\t5\t⌂ HQT\n@7\t7\thqt-bar\n",
+            "",
+        ),
+        (0, "", ""),
+        (0, "", ""),
+        (0, "", ""),
+        (0, "", ""),
+        (0, "", ""),
+        (0, "", ""),
+    ]
+
+    result = await runner.reset_window_indexes("⌂ HQT")
+
+    assert result is True
+    calls = [call.args for call in runner._exec.call_args_list]
+    assert calls[1] == ("move-window", "-d", "-s", "@2", "-t", "hqt-main:11")
+    assert calls[2] == ("move-window", "-d", "-s", "@5", "-t", "hqt-main:12")
+    assert calls[3] == ("move-window", "-d", "-s", "@7", "-t", "hqt-main:13")
+    assert calls[4] == ("move-window", "-d", "-s", "@5", "-t", "hqt-main:0")
+    assert calls[5] == ("move-window", "-d", "-s", "@2", "-t", "hqt-main:1")
+    assert calls[6] == ("move-window", "-d", "-s", "@7", "-t", "hqt-main:2")
+
+
+@pytest.mark.asyncio
+async def test_reset_window_indexes_move_failure_returns_false(runner):
+    runner._exec.side_effect = [
+        (
+            0,
+            "@0\t0\t⌂ HQT\n@1\t1\thqt-foo\n@4\t4\thqt-bar\n",
+            "",
+        ),
+        (0, "", ""),
+        (1, "", "index in use"),
+    ]
+
+    result = await runner.reset_window_indexes("⌂ HQT")
+
+    assert result is False
+    assert runner._exec.call_count == 3
+    calls = [call.args for call in runner._exec.call_args_list]
+    assert calls[1] == ("move-window", "-d", "-s", "@0", "-t", "hqt-main:8")
+    assert calls[2] == ("move-window", "-d", "-s", "@1", "-t", "hqt-main:9")
+
+
+@pytest.mark.asyncio
+async def test_reset_window_indexes_list_failure_returns_false(runner):
+    runner._exec.return_value = (1, "", "no server running")
+
+    result = await runner.reset_window_indexes("⌂ HQT")
+
+    assert result is False
+    runner._exec.assert_called_once_with(
+        "list-windows",
+        "-t",
+        "hqt-main",
+        "-F",
+        "#{window_id}\t#{window_index}\t#{window_name}",
+    )
+
+
+@pytest.mark.asyncio
+async def test_reset_window_indexes_malformed_list_line_returns_false(runner):
+    runner._exec.return_value = (
+        0,
+        "@0\t0\t⌂ HQT\nnot-enough-columns\n",
+        "",
+    )
+
+    result = await runner.reset_window_indexes("⌂ HQT")
+
+    assert result is False
+    assert runner._exec.call_count == 1
+
+
 @pytest.mark.asyncio
 async def test_manager_spawn(runner):
     runner._exec.side_effect = [
@@ -549,6 +697,46 @@ async def test_respawn_pane_no_env(runner):
     assert "-e" not in call_args
 
 
+@pytest.mark.asyncio
+async def test_send_text_returns_true_on_success(runner):
+    runner._exec.return_value = (0, "", "")
+
+    assert await runner.send_text("hqt-1", "hello") is True
+
+
+@pytest.mark.asyncio
+async def test_send_text_returns_false_on_failure(runner):
+    runner._exec.return_value = (1, "", "no such target")
+
+    assert await runner.send_text("hqt-1", "hello") is False
+
+
+@pytest.mark.asyncio
+async def test_send_enter_returns_true_on_success(runner):
+    runner._exec.return_value = (0, "", "")
+
+    assert await runner.send_enter("hqt-1") is True
+
+
+@pytest.mark.asyncio
+async def test_send_enter_returns_false_on_failure(runner):
+    runner._exec.return_value = (1, "", "no such target")
+
+    assert await runner.send_enter("hqt-1") is False
+
+
+@pytest.mark.asyncio
+async def test_manager_send_delegates(runner):
+    mgr = TmuxManager(runner)
+    runner.send_text = AsyncMock(return_value=True)
+    runner.send_enter = AsyncMock(return_value=True)
+
+    assert await mgr.send_text("hqt-1", "continue") is True
+    assert await mgr.send_enter("hqt-1") is True
+    runner.send_text.assert_awaited_once_with("hqt-1", "continue")
+    runner.send_enter.assert_awaited_once_with("hqt-1")
+
+
 @pytest.mark.asyncio
 async def test_spawn_verify_dead_empty_text_returns_fallback_error(runner):
     """When verify_window_alive returns (False, ''), spawn returns the fallback error message."""
@@ -896,6 +1084,17 @@ async def test_manager_set_window_label_delegates(runner):
     assert "○ deadproj" in args
 
 
+@pytest.mark.asyncio
+async def test_manager_reset_window_indexes_delegates(runner):
+    mgr = TmuxManager(runner)
+    runner.reset_window_indexes = AsyncMock(return_value=True)
+
+    result = await mgr.reset_window_indexes("⌂ HQT")
+
+    assert result is True
+    runner.reset_window_indexes.assert_awaited_once_with("⌂ HQT")
+
+
 @pytest.mark.asyncio
 async def test_apply_theme_sets_session_scoped_status_options(runner):
     """apply_theme themes the status bar in one atomic styling call.
@@ -1021,7 +1220,7 @@ async def test_new_aux_window_spawns_styles_and_selects(runner):
     runner._exec.side_effect = [
         (0, "0\n", ""),   # _next_window_index: indices [0] -> next index 1
         (0, "@7\n", ""),  # new-window -P -F '#{window_id}'
-        (0, "", ""),      # set-option (automatic-rename + @hqt_label + theme)
+        (0, "", ""),      # set-option (automatic-rename + allow-rename + @hqt_label + theme)
         (0, "", ""),      # select-window
     ]
     wid = await runner.new_aux_window("lazygit", "/proj", ["lazygit"], "lazygit · proj")
@@ -1037,6 +1236,8 @@ async def test_new_aux_window_spawns_styles_and_selects(runner):
     assert calls[2].args[:6] == (
         "set-option", "-w", "-t", "@7", "automatic-rename", "off",
     )
+    # allow-rename off too, so nvim/lazygit OSC titles can't scramble the label.
+    assert "allow-rename" in calls[2].args
     assert "@hqt_label" in calls[2].args
     assert "lazygit · proj" in calls[2].args
     assert "remain-on-exit" not in calls[2].args

tests/test_tui.py

@@ -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}"
 
 
 # ---------------------------------------------------------------------------
@@ -1027,6 +1032,51 @@ def test_format_session_text_status_colors():
     )  # Overlay0
 
 
+def _make_scheduled_prompt(due_at):
+    prompt = MagicMock()
+    prompt.due_at = due_at
+    prompt.status = "pending"
+    return prompt
+
+
+def test_format_session_text_shows_scheduled_prompt_countdown():
+    from datetime import datetime, timedelta
+    from hqt.tui.widgets.session_list import format_session_text
+
+    now = datetime(2026, 6, 11, 14, 0, 0)
+    scheduled = _make_scheduled_prompt(now + timedelta(minutes=12, seconds=1))
+
+    text = format_session_text(
+        "mywork",
+        "claude",
+        "idle",
+        "●",
+        scheduled_prompt=scheduled,
+        now=now,
+    )
+
+    assert "@ 13m" in text
+
+
+def test_format_session_text_shows_scheduled_prompt_clock_time():
+    from datetime import datetime, timedelta
+    from hqt.tui.widgets.session_list import format_session_text
+
+    now = datetime(2026, 6, 11, 14, 0, 0)
+    scheduled = _make_scheduled_prompt(now + timedelta(hours=2, minutes=30))
+
+    text = format_session_text(
+        "mywork",
+        "claude",
+        "idle",
+        "●",
+        scheduled_prompt=scheduled,
+        now=now,
+    )
+
+    assert "@ 16:30" in text
+
+
 @pytest.mark.asyncio
 async def test_session_list_renders_harness_name_brackets():
     """Regression: '[claude]' must be visible, not swallowed as a markup tag."""
@@ -1045,6 +1095,121 @@ async def test_session_list_renders_harness_name_brackets():
         assert any("[claude]" in t for t in texts), f"harness name missing: {texts}"
 
 
+# ---------------------------------------------------------------------------
+# SchedulePromptScreen: parse + submit
+# ---------------------------------------------------------------------------
+
+
+def test_parse_delay_accepts_compact_seconds_minutes_hours():
+    from datetime import timedelta
+
+    from hqt.tui.screens.schedule_prompt import parse_delay
+
+    assert parse_delay("90s") == timedelta(seconds=90)
+    assert parse_delay("30m") == timedelta(minutes=30)
+    assert parse_delay("2h") == timedelta(hours=2)
+    assert parse_delay("1h30m") == timedelta(hours=1, minutes=30)
+
+
+@pytest.mark.parametrize("value", ["", "0m", "abc", "1d", "1h 30m"])
+def test_parse_delay_rejects_invalid_values(value):
+    from hqt.tui.screens.schedule_prompt import parse_delay
+
+    with pytest.raises(ValueError):
+        parse_delay(value)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("submit_action", ["enter", "ok"])
+async def test_schedule_prompt_screen_submits_prompt_and_delay(submit_action):
+    from datetime import timedelta
+
+    from hqt.tui.screens.schedule_prompt import SchedulePromptScreen
+    from textual.widgets import Button, Input
+
+    app = HqtApp()
+    async with app.run_test(size=(120, 40)) as pilot:
+        results = []
+        app.push_screen(SchedulePromptScreen(), results.append)
+        await pilot.pause()
+
+        prompt = app.screen.query_one("#prompt-input", Input)
+        prompt.value = "continue work"
+        app.screen.query_one("#delay-input", Input).value = "1h30m"
+        prompt.focus()
+        await pilot.pause()
+        if submit_action == "enter":
+            await pilot.press("enter")
+        else:
+            app.screen.query_one("#ok-btn", Button).press()
+        await pilot.pause()
+
+        assert results == [("continue work", timedelta(hours=1, minutes=30))]
+
+
+@pytest.mark.asyncio
+async def test_schedule_prompt_screen_invalid_delay_stays_open_with_error():
+    from hqt.tui.screens.schedule_prompt import SchedulePromptScreen
+    from textual.widgets import Button, Input, Label
+
+    app = HqtApp()
+    async with app.run_test(size=(120, 40)) as pilot:
+        results = []
+        app.push_screen(SchedulePromptScreen(), results.append)
+        await pilot.pause()
+
+        app.screen.query_one("#prompt-input", Input).value = "continue work"
+        app.screen.query_one("#delay-input", Input).value = "1h 30m"
+        app.screen.query_one("#ok-btn", Button).press()
+        await pilot.pause()
+
+        assert isinstance(app.screen, SchedulePromptScreen)
+        assert results == []
+        error = app.screen.query_one("#schedule-error", Label)
+        assert "Use a delay like" in str(error.render())
+        assert app.screen.focused == app.screen.query_one("#delay-input", Input)
+
+
+@pytest.mark.asyncio
+async def test_schedule_prompt_screen_empty_prompt_stays_open_with_error():
+    from hqt.tui.screens.schedule_prompt import SchedulePromptScreen
+    from textual.widgets import Button, Input, Label
+
+    app = HqtApp()
+    async with app.run_test(size=(120, 40)) as pilot:
+        results = []
+        app.push_screen(SchedulePromptScreen(), results.append)
+        await pilot.pause()
+
+        app.screen.query_one("#prompt-input", Input).value = "   "
+        app.screen.query_one("#delay-input", Input).value = "30m"
+        app.screen.query_one("#ok-btn", Button).press()
+        await pilot.pause()
+
+        assert isinstance(app.screen, SchedulePromptScreen)
+        assert results == []
+        error = app.screen.query_one("#schedule-error", Label)
+        assert "Prompt cannot be empty" in str(error.render())
+        assert app.screen.focused == app.screen.query_one("#prompt-input", Input)
+
+
+@pytest.mark.asyncio
+async def test_schedule_prompt_screen_cancel_returns_none():
+    from hqt.tui.screens.schedule_prompt import SchedulePromptScreen
+    from textual.widgets import Button
+
+    app = HqtApp()
+    async with app.run_test(size=(120, 40)) as pilot:
+        results = []
+        app.push_screen(SchedulePromptScreen(), results.append)
+        await pilot.pause()
+
+        app.screen.query_one("#cancel-btn", Button).press()
+        await pilot.pause()
+
+        assert results == [None]
+
+
 # ---------------------------------------------------------------------------
 # RenameSessionScreen: prefill + submit
 # ---------------------------------------------------------------------------
@@ -1291,7 +1456,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 +1481,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 +1505,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 +1671,201 @@ 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
+
+
+# ---------------------------------------------------------------------------
+# delayed prompt: bindings, actions, poll dispatch
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_prompt_later_binding_exists():
+    bindings = {b.key: b for b in HqtApp.BINDINGS}
+    assert "p" in bindings
+    assert bindings["p"].action == "schedule_prompt"
+    assert "P" in bindings
+    assert bindings["P"].action == "cancel_scheduled_prompt"
+
+
+@pytest.mark.asyncio
+async def test_schedule_prompt_action_calls_service_and_refreshes():
+    from datetime import datetime
+    from unittest.mock import patch
+    from hqt.db.models import Harness, Project, Session
+    from hqt.tui.widgets.session_list import SessionList
+
+    app = HqtApp()
+    async with app.run_test(size=(120, 40)) as pilot:
+        seed_db = app._db_factory()
+        proj = Project(name="prompt-proj", path="/tmp/prompt-proj")
+        seed_db.add(proj)
+        seed_db.flush()
+        harness = seed_db.query(Harness).first()
+        sess = Session(
+            project_id=proj.id,
+            harness_id=harness.id,
+            tmux_session_name="hqt-prompt-test",
+            archived=False,
+        )
+        seed_db.add(sess)
+        seed_db.commit()
+        proj_id = proj.id
+        sess_id = sess.id
+        seed_db.close()
+
+        app._selected_project_id = proj_id
+        await app._refresh_sessions()
+        await pilot.pause()
+        app.query_one(SessionList).query_one("#session-list").focus()
+        await pilot.press("down")
+        await pilot.pause()
+
+        mock_service = MagicMock()
+        mock_service.schedule_prompt = MagicMock()
+        mock_service.list_sessions = AsyncMock(return_value=[])
+        mock_service.sync_window_labels = AsyncMock(return_value=[])
+        mock_service.dispatch_due_prompts = AsyncMock(return_value=[])
+        app._session_service = mock_service
+
+        notify_calls = []
+
+        def capture_notify(message, **kwargs):
+            notify_calls.append((message, kwargs))
+
+        with patch.object(app, "notify", side_effect=capture_notify):
+            app.action_schedule_prompt()
+            await pilot.pause()
+            app.screen.dismiss(
+                ("continue", __import__("datetime").timedelta(minutes=30))
+            )
+            await pilot.pause()
+            await app.workers.wait_for_complete()
+
+        assert mock_service.schedule_prompt.call_count == 1
+        args = mock_service.schedule_prompt.call_args.args
+        assert args[0] == sess_id
+        assert args[1] == "continue"
+        assert isinstance(args[2], datetime)
+        mock_service.list_sessions.assert_awaited()
+        assert any("Prompt scheduled" in call[0] for call in notify_calls)
+
+
+@pytest.mark.asyncio
+async def test_cancel_scheduled_prompt_action_calls_service_and_refreshes():
+    from hqt.db.models import Harness, Project, Session
+    from hqt.tui.widgets.session_list import SessionList
+
+    app = HqtApp()
+    async with app.run_test(size=(120, 40)) as pilot:
+        seed_db = app._db_factory()
+        proj = Project(name="cancel-proj", path="/tmp/cancel-proj")
+        seed_db.add(proj)
+        seed_db.flush()
+        harness = seed_db.query(Harness).first()
+        sess = Session(
+            project_id=proj.id,
+            harness_id=harness.id,
+            tmux_session_name="hqt-cancel-test",
+            archived=False,
+        )
+        seed_db.add(sess)
+        seed_db.commit()
+        proj_id = proj.id
+        sess_id = sess.id
+        seed_db.close()
+
+        app._selected_project_id = proj_id
+        await app._refresh_sessions()
+        await pilot.pause()
+        app.query_one(SessionList).query_one("#session-list").focus()
+        await pilot.press("down")
+        await pilot.pause()
+
+        mock_service = MagicMock()
+        mock_service.cancel_scheduled_prompt = MagicMock()
+        mock_service.list_sessions = AsyncMock(return_value=[])
+        mock_service.sync_window_labels = AsyncMock(return_value=[])
+        mock_service.dispatch_due_prompts = AsyncMock(return_value=[])
+        app._session_service = mock_service
+
+        await app.run_action("cancel_scheduled_prompt")
+        await pilot.pause()
+        await app.workers.wait_for_complete()
+
+        mock_service.cancel_scheduled_prompt.assert_called_once_with(sess_id)
+        mock_service.list_sessions.assert_awaited()
+
+
+@pytest.mark.asyncio
+async def test_poll_sessions_dispatches_due_prompts_and_notifies():
+    app = HqtApp()
+    async with app.run_test(size=(120, 40)):
+        result = MagicMock()
+        result.ok = True
+        result.window_name = "hqt-1"
+        result.error = ""
+
+        mock_service = MagicMock()
+        mock_service.dispatch_due_prompts = AsyncMock(return_value=[result])
+        mock_service.sync_window_labels = AsyncMock(return_value=[])
+        app._session_service = mock_service
+
+        notify_calls = []
+
+        def capture_notify(message, **kwargs):
+            notify_calls.append((message, kwargs))
+
+        from unittest.mock import patch
+
+        with patch.object(app, "notify", side_effect=capture_notify):
+            await app._poll_sessions()
+
+        mock_service.dispatch_due_prompts.assert_awaited_once()
+        assert any("Prompt sent" in call[0] for call in notify_calls)