felixm
4c36385045
Add sandbox_skip_permission_flags class attr and sandbox_binds() method to HarnessConfigurator ABC; thread sandboxed: bool = False through all build_spawn_config / build_resume_config signatures. Claude and Codex append their respective skip-permission flags when sandboxed=True; Kiro and Generic accept the param but add no flags. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
401 lines
13 KiB
Python
401 lines
13 KiB
Python
import json
|
||
import os
|
||
import time
|
||
from pathlib import Path
|
||
from unittest.mock import patch
|
||
|
||
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
|
||
|
||
|
||
def test_claude_spawn_config():
|
||
c = ClaudeConfigurator()
|
||
sid = c.generate_session_id(1)
|
||
cfg = c.build_spawn_config(Path("/tmp"), sid, model="opus")
|
||
assert cfg.command == ["claude", "--session-id", sid, "--model", "opus"]
|
||
|
||
|
||
def test_claude_resume_config():
|
||
c = ClaudeConfigurator()
|
||
sid = c.generate_session_id(1)
|
||
cfg = c.build_resume_config(Path("/tmp"), sid)
|
||
assert "--resume" in cfg.command
|
||
assert sid in cfg.command
|
||
|
||
|
||
def test_claude_resume_config_includes_model_when_set():
|
||
"""build_resume_config appends --model when model is provided."""
|
||
c = ClaudeConfigurator()
|
||
sid = c.generate_session_id(1)
|
||
cfg = c.build_resume_config(Path("/tmp"), sid, model="claude-opus-4-5")
|
||
assert "--resume" in cfg.command
|
||
assert sid in cfg.command
|
||
assert "--model" in cfg.command
|
||
assert "claude-opus-4-5" in cfg.command
|
||
|
||
|
||
def test_claude_resume_config_omits_model_when_none():
|
||
"""build_resume_config omits --model when model is None."""
|
||
c = ClaudeConfigurator()
|
||
sid = c.generate_session_id(1)
|
||
cfg = c.build_resume_config(Path("/tmp"), sid, model=None)
|
||
assert "--model" not in cfg.command
|
||
|
||
|
||
def test_codex_resume_config_includes_model_when_set():
|
||
"""build_resume_config appends -m <model> when model is provided."""
|
||
c = CodexConfigurator()
|
||
cfg = c.build_resume_config(Path("/tmp"), "abc-123", model="o3")
|
||
assert cfg.command == ["codex", "resume", "abc-123", "-m", "o3"]
|
||
|
||
|
||
def test_codex_resume_config_omits_model_when_none():
|
||
"""build_resume_config omits -m when model is None."""
|
||
c = CodexConfigurator()
|
||
cfg = c.build_resume_config(Path("/tmp"), "abc-123", model=None)
|
||
assert "-m" not in cfg.command
|
||
assert cfg.command == ["codex", "resume", "abc-123"]
|
||
|
||
|
||
def test_codex_spawn_with_model():
|
||
c = CodexConfigurator()
|
||
cfg = c.build_spawn_config(Path("/tmp"), "1", model="o3")
|
||
assert cfg.command == ["codex", "-m", "o3"]
|
||
|
||
|
||
def test_codex_resume_uses_uuid():
|
||
c = CodexConfigurator()
|
||
cfg = c.build_resume_config(Path("/tmp"), "abc-123")
|
||
assert cfg.command == ["codex", "resume", "abc-123"]
|
||
|
||
|
||
def test_codex_capture_session_id(tmp_path):
|
||
sessions_dir = tmp_path / ".codex" / "sessions" / "2026" / "06" / "09"
|
||
sessions_dir.mkdir(parents=True)
|
||
rollout = (
|
||
sessions_dir / "rollout-20260609-abcdef12-3456-7890-abcd-ef1234567890.jsonl"
|
||
)
|
||
meta = {
|
||
"type": "session_meta",
|
||
"payload": {
|
||
"id": "abcdef12-3456-7890-abcd-ef1234567890",
|
||
"cwd": "/projects/foo",
|
||
},
|
||
}
|
||
rollout.write_text(json.dumps(meta) + "\n")
|
||
|
||
since = time.time() - 60 # far enough back that freshly-written file qualifies
|
||
c = CodexConfigurator()
|
||
with patch("hqt.harnesses.configurators.codex.Path.home", return_value=tmp_path):
|
||
result = c.capture_session_id(Path("/projects/foo"), since)
|
||
assert result == "abcdef12-3456-7890-abcd-ef1234567890"
|
||
|
||
|
||
def test_codex_capture_session_id_wrong_cwd(tmp_path):
|
||
sessions_dir = tmp_path / ".codex" / "sessions" / "2026" / "06" / "09"
|
||
sessions_dir.mkdir(parents=True)
|
||
rollout = sessions_dir / "rollout-20260609-abcdef12.jsonl"
|
||
meta = {"type": "session_meta", "payload": {"id": "abcdef12", "cwd": "/other/path"}}
|
||
rollout.write_text(json.dumps(meta) + "\n")
|
||
|
||
since = time.time() - 60
|
||
c = CodexConfigurator()
|
||
with patch("hqt.harnesses.configurators.codex.Path.home", return_value=tmp_path):
|
||
result = c.capture_session_id(Path("/projects/foo"), since)
|
||
assert result is None
|
||
|
||
|
||
@patch("hqt.harnesses.registry.shutil.which")
|
||
def test_registry_discovers_harnesses(mock_which):
|
||
mock_which.side_effect = lambda x: (
|
||
"/usr/bin/" + x if x in ("claude", "codex") else None
|
||
)
|
||
result = discover_harnesses()
|
||
assert "claude" in result
|
||
assert "codex" in result
|
||
assert "kiro" not in result
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Task 4: time-fenced capture_session_id and captures_session_id flag
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_codex_captures_session_id_flag():
|
||
"""CodexConfigurator must expose captures_session_id = True."""
|
||
assert CodexConfigurator.captures_session_id is True
|
||
|
||
|
||
def test_base_captures_session_id_flag_false():
|
||
"""HarnessConfigurator base class default must be False."""
|
||
assert HarnessConfigurator.captures_session_id is False
|
||
|
||
|
||
def test_claude_captures_session_id_flag_false():
|
||
"""ClaudeConfigurator must NOT set captures_session_id (inherits False)."""
|
||
assert ClaudeConfigurator.captures_session_id is False
|
||
|
||
|
||
def _write_rollout(
|
||
path: Path, session_id: str, cwd: str, timestamp: str | None = None
|
||
) -> None:
|
||
meta = {"type": "session_meta", "payload": {"id": session_id, "cwd": cwd}}
|
||
if timestamp:
|
||
meta["payload"]["timestamp"] = timestamp
|
||
path.write_text(json.dumps(meta) + "\n")
|
||
|
||
|
||
def test_codex_capture_ignores_file_before_since(tmp_path):
|
||
"""Rollout with mtime BEFORE since must not be returned even if cwd matches."""
|
||
sessions_dir = tmp_path / ".codex" / "sessions" / "dir"
|
||
sessions_dir.mkdir(parents=True)
|
||
rollout = sessions_dir / "rollout-old.jsonl"
|
||
_write_rollout(rollout, "old-id", "/projects/foo")
|
||
|
||
# Set mtime to something clearly in the past
|
||
past = time.time() - 60
|
||
os.utime(rollout, (past, past))
|
||
|
||
since = time.time() - 10 # since is after the file's mtime
|
||
|
||
c = CodexConfigurator()
|
||
with patch("hqt.harnesses.configurators.codex.Path.home", return_value=tmp_path):
|
||
result = c.capture_session_id(Path("/projects/foo"), since)
|
||
assert result is None
|
||
|
||
|
||
def test_codex_capture_accepts_file_after_since(tmp_path):
|
||
"""Rollout with mtime >= since and matching cwd is returned."""
|
||
sessions_dir = tmp_path / ".codex" / "sessions" / "dir"
|
||
sessions_dir.mkdir(parents=True)
|
||
rollout = sessions_dir / "rollout-new.jsonl"
|
||
_write_rollout(rollout, "new-id", "/projects/foo")
|
||
|
||
since = time.time() - 60 # since is well before file was written
|
||
# mtime of a freshly written file should be >= since
|
||
|
||
c = CodexConfigurator()
|
||
with patch("hqt.harnesses.configurators.codex.Path.home", return_value=tmp_path):
|
||
result = c.capture_session_id(Path("/projects/foo"), since)
|
||
assert result == "new-id"
|
||
|
||
|
||
def test_codex_capture_ambiguous_returns_none(tmp_path):
|
||
"""Two rollouts match cwd + since window -> refuse to guess, return None."""
|
||
sessions_dir = tmp_path / ".codex" / "sessions" / "dir"
|
||
sessions_dir.mkdir(parents=True)
|
||
|
||
rollout_a = sessions_dir / "rollout-a.jsonl"
|
||
rollout_b = sessions_dir / "rollout-b.jsonl"
|
||
_write_rollout(rollout_a, "id-a", "/projects/foo")
|
||
_write_rollout(rollout_b, "id-b", "/projects/foo")
|
||
|
||
now = time.time()
|
||
since = now - 120
|
||
os.utime(rollout_a, (now - 30, now - 30))
|
||
os.utime(rollout_b, (now - 10, now - 10))
|
||
|
||
c = CodexConfigurator()
|
||
with patch("hqt.harnesses.configurators.codex.Path.home", return_value=tmp_path):
|
||
result = c.capture_session_id(Path("/projects/foo"), since)
|
||
assert result is None
|
||
|
||
|
||
def test_codex_capture_skips_non_matching_cwd_picks_older_matching(tmp_path):
|
||
"""Newest post-since file has wrong cwd; older post-since file with right cwd is returned."""
|
||
sessions_dir = tmp_path / ".codex" / "sessions" / "dir"
|
||
sessions_dir.mkdir(parents=True)
|
||
|
||
rollout_wrong = sessions_dir / "rollout-wrong.jsonl"
|
||
rollout_right = sessions_dir / "rollout-right.jsonl"
|
||
_write_rollout(rollout_wrong, "wrong-id", "/other/project")
|
||
_write_rollout(rollout_right, "right-id", "/projects/foo")
|
||
|
||
now = time.time()
|
||
since = now - 120
|
||
# wrong cwd file is newer
|
||
os.utime(rollout_wrong, (now - 5, now - 5))
|
||
# right cwd file is older but still after since
|
||
os.utime(rollout_right, (now - 20, now - 20))
|
||
|
||
c = CodexConfigurator()
|
||
with patch("hqt.harnesses.configurators.codex.Path.home", return_value=tmp_path):
|
||
result = c.capture_session_id(Path("/projects/foo"), since)
|
||
assert result == "right-id"
|
||
|
||
|
||
def test_codex_capture_missing_sessions_dir_returns_none(tmp_path):
|
||
"""Missing ~/.codex/sessions dir returns None without error."""
|
||
c = CodexConfigurator()
|
||
with patch("hqt.harnesses.configurators.codex.Path.home", return_value=tmp_path):
|
||
result = c.capture_session_id(Path("/projects/foo"), time.time())
|
||
assert result is None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Task 7: parse_status for harness configurators
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
CLAUDE_WORKING_SCREEN = """\
|
||
> Implement feature X
|
||
|
||
● Bash(git status)
|
||
⎿ On branch main
|
||
|
||
● Write(src/foo.py)
|
||
⎿ Wrote 42 lines
|
||
|
||
Esc to interrupt
|
||
"""
|
||
|
||
CLAUDE_WAITING_SCREEN = """\
|
||
> Do you want to proceed with these changes?
|
||
❯ 1. Yes
|
||
2. No
|
||
|
||
"""
|
||
|
||
CLAUDE_PERMISSION_PROMPT = """\
|
||
> Do you want to allow this tool to run?
|
||
❯ 1. Yes, allow once
|
||
2. No
|
||
|
||
"""
|
||
|
||
CLAUDE_IDLE_SCREEN = """\
|
||
>
|
||
|
||
✓ Done
|
||
|
||
"""
|
||
|
||
CODEX_WORKING_SCREEN = """\
|
||
Working...
|
||
|
||
Esc to interrupt
|
||
"""
|
||
|
||
CODEX_IDLE_SCREEN = """\
|
||
>
|
||
|
||
"""
|
||
|
||
|
||
def test_base_parse_status_returns_none():
|
||
"""Base HarnessConfigurator.parse_status always returns None."""
|
||
k = KiroConfigurator()
|
||
assert k.parse_status("anything") is None
|
||
assert k.parse_status("") is None
|
||
|
||
|
||
def test_claude_parse_status_working():
|
||
"""ClaudeConfigurator: 'esc to interrupt' (case-insensitive) → 'working'."""
|
||
c = ClaudeConfigurator()
|
||
assert c.parse_status(CLAUDE_WORKING_SCREEN) == "working"
|
||
|
||
|
||
def test_claude_parse_status_working_case_insensitive():
|
||
"""ClaudeConfigurator: 'ESC TO INTERRUPT' upper-case → 'working'."""
|
||
c = ClaudeConfigurator()
|
||
assert c.parse_status("ESC TO INTERRUPT") == "working"
|
||
|
||
|
||
def test_claude_parse_status_waiting_do_you_want():
|
||
"""ClaudeConfigurator: line starting with 'Do you want' → 'waiting'."""
|
||
c = ClaudeConfigurator()
|
||
assert c.parse_status(CLAUDE_WAITING_SCREEN) == "waiting"
|
||
|
||
|
||
def test_claude_parse_status_waiting_permission_prompt():
|
||
"""ClaudeConfigurator: '❯ 1.' style line → 'waiting'."""
|
||
c = ClaudeConfigurator()
|
||
assert c.parse_status(CLAUDE_PERMISSION_PROMPT) == "waiting"
|
||
|
||
|
||
def test_claude_parse_status_none_for_idle():
|
||
"""ClaudeConfigurator: idle/done screen → None (activity-based fallback)."""
|
||
c = ClaudeConfigurator()
|
||
assert c.parse_status(CLAUDE_IDLE_SCREEN) is None
|
||
|
||
|
||
def test_claude_parse_status_empty_returns_none():
|
||
"""ClaudeConfigurator: empty text → None."""
|
||
c = ClaudeConfigurator()
|
||
assert c.parse_status("") is None
|
||
|
||
|
||
def test_codex_parse_status_working():
|
||
"""CodexConfigurator: 'esc to interrupt' → 'working'."""
|
||
c = CodexConfigurator()
|
||
assert c.parse_status(CODEX_WORKING_SCREEN) == "working"
|
||
|
||
|
||
def test_codex_parse_status_none_for_idle():
|
||
"""CodexConfigurator: idle screen → None."""
|
||
c = CodexConfigurator()
|
||
assert c.parse_status(CODEX_IDLE_SCREEN) is None
|
||
|
||
|
||
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"]
|