hqt/tests/test_sessions.py

import pytest
from datetime import datetime, timedelta
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from hqt.db.models import Base, Harness, Project, ScheduledPrompt, Session
from hqt.errors import ServiceError
from hqt.sandbox import SandboxPolicy
from hqt.sessions.service import SessionInfo, SessionService
from hqt.tmux.manager import SpawnResult, TmuxManager


@pytest.fixture
def factory():
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(engine)
    return sessionmaker(bind=engine, expire_on_commit=False)


@pytest.fixture
def db(factory):
    # Seeding/assertion handle onto the same in-memory DB the service uses.
    session = factory()
    session.add(Harness(name="claude-code", display_name="Claude Code"))
    session.add(Project(name="myproj", path="/tmp/myproj"))
    session.commit()
    yield session
    session.close()


@pytest.fixture
def tmux():
    m = MagicMock(spec=TmuxManager)
    m.spawn = AsyncMock(return_value=SpawnResult(ok=True, window_id="@1", error=""))
    m.kill = AsyncMock()
    m.is_alive = AsyncMock(return_value=True)
    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="")
    )
    return m


@pytest.fixture
def harnesses():
    h = MagicMock()
    h.captures_session_id = False
    h.generate_session_id.return_value = "sess-1"
    h.build_spawn_config.return_value = MagicMock(
        command=["claude"], env={}, cwd=Path("/tmp/myproj")
    )
    h.parse_status = MagicMock(return_value=None)
    return {"claude-code": h}


@pytest.fixture
def service(factory, db, tmux, harnesses):
    # depends on db so the seed rows exist before the service runs
    return SessionService(factory=factory, tmux=tmux, harnesses=harnesses)


@pytest.mark.asyncio
async def test_create_session(service, db, tmux, harnesses):
    result = await service.create_session(
        project_id=1, harness_name="claude-code", nickname="test"
    )
    sess = result.session
    assert sess.id is not None
    assert sess.tmux_session_name == f"hqt-{sess.id}"
    assert sess.harness_session_id == "sess-1"
    tmux.spawn.assert_called_once()
    harnesses["claude-code"].build_spawn_config.assert_called_once()


@pytest.mark.asyncio
async def test_list_sessions(service, db, tmux):
    from hqt.tmux.runner import WindowInfo

    await service.create_session(project_id=1, harness_name="claude-code")
    tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1000)}
    result = await service.list_sessions(project_id=1)
    assert len(result) == 1
    assert isinstance(result[0], SessionInfo)
    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")
    await service.stop_session(session_id=1)
    tmux.kill.assert_called_once_with("hqt-1")


@pytest.mark.asyncio
async def test_stop_session_no_kill_when_window_missing(service, db, tmux):
    """stop_session must NOT call kill when the window doesn't exist."""
    await service.create_session(project_id=1, harness_name="claude-code")
    tmux.window_exists = AsyncMock(return_value=False)
    tmux.kill.reset_mock()
    await service.stop_session(session_id=1)
    tmux.kill.assert_not_called()


@pytest.mark.asyncio
async def test_create_session_logs_failed_spawn(service, db, tmux, caplog):
    """create_session logs error including pane text when spawn fails, but still commits."""
    import logging

    tmux.spawn.return_value = SpawnResult(
        ok=False, window_id=None, error="bash: command not found"
    )
    with caplog.at_level(logging.ERROR):
        result = await service.create_session(project_id=1, harness_name="claude-code")
    assert result.session.id is not None  # row was committed
    # Error was logged and includes captured pane text
    error_messages = [r.message for r in caplog.records if r.levelno == logging.ERROR]
    assert any("command not found" in m for m in error_messages)


@pytest.mark.asyncio
async def test_attach_session_passes_env(service, db, tmux, harnesses):
    """attach_session passes cfg.env to tmux.respawn_verified when the window needs respawning."""
    from unittest.mock import AsyncMock

    # Create a session first
    await service.create_session(project_id=1, harness_name="claude-code")

    # Set up harness to return a config with env
    env_cfg = MagicMock(
        command=["claude"], env={"SESSION_VAR": "abc"}, cwd=Path("/tmp/myproj")
    )
    harnesses["claude-code"].build_resume_config = MagicMock(return_value=env_cfg)

    # Simulate window is not alive → respawn path
    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)

    await service.attach_session(session_id=1)

    tmux.respawn_verified.assert_called()
    first_call = tmux.respawn_verified.call_args_list[0]
    # env must be forwarded (passed as keyword or positional)
    assert first_call.kwargs.get("env") == {"SESSION_VAR": "abc"} or (
        len(first_call.args) >= 4 and first_call.args[3] == {"SESSION_VAR": "abc"}
    )


# ---------------------------------------------------------------------------
# Task 3: fallback ladder + model preserved on resume
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_attach_session_dead_resume_ok_attaches_once(
    service, db, tmux, harnesses
):
    """attach_session: resume ok → attach called once, no fallback spawn attempted."""
    await service.create_session(project_id=1, harness_name="claude-code")

    resume_cfg = MagicMock(
        command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj")
    )
    harnesses["claude-code"].build_resume_config = MagicMock(return_value=resume_cfg)
    harnesses["claude-code"].build_spawn_config = MagicMock(
        return_value=MagicMock(
            command=["claude", "--session-id", "sess-1"],
            env={},
            cwd=Path("/tmp/myproj"),
        )
    )

    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)

    result = await service.attach_session(session_id=1)

    assert result is True
    tmux.attach.assert_called_once()
    # Only one respawn_verified call (the resume rung), no fallback
    assert tmux.respawn_verified.call_count == 1
    first_cmd = tmux.respawn_verified.call_args_list[0].args[1]
    assert "--resume" in first_cmd


@pytest.mark.asyncio
async def test_attach_session_dead_resume_fails_fallback_spawn(
    service, db, tmux, harnesses
):
    """attach_session: resume dies → fallback spawn with session-id, attach called once."""
    await service.create_session(project_id=1, harness_name="claude-code")

    resume_cfg = MagicMock(
        command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj")
    )
    spawn_cfg = MagicMock(
        command=["claude", "--session-id", "sess-1"], env={}, cwd=Path("/tmp/myproj")
    )
    harnesses["claude-code"].build_resume_config = MagicMock(return_value=resume_cfg)
    harnesses["claude-code"].build_spawn_config = MagicMock(return_value=spawn_cfg)

    tmux.is_alive = AsyncMock(return_value=False)
    tmux.respawn_verified = AsyncMock(
        side_effect=[
            SpawnResult(
                ok=False, window_id=None, error="no transcript"
            ),  # resume fails
            SpawnResult(ok=True, window_id="@2", error=""),  # fresh spawn ok
        ]
    )
    tmux.attach = AsyncMock(return_value=True)

    result = await service.attach_session(session_id=1)

    assert result is True
    # Two respawn_verified calls: first resume, then spawn
    assert tmux.respawn_verified.call_count == 2
    first_cmd = tmux.respawn_verified.call_args_list[0].args[1]
    second_cmd = tmux.respawn_verified.call_args_list[1].args[1]
    assert "--resume" in first_cmd
    assert "--session-id" in second_cmd
    tmux.attach.assert_called_once()


@pytest.mark.asyncio
async def test_attach_session_both_rungs_fail_returns_false(
    service, db, tmux, harnesses
):
    """attach_session: both resume and spawn fail → False, no attach."""
    await service.create_session(project_id=1, harness_name="claude-code")

    resume_cfg = MagicMock(
        command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj")
    )
    spawn_cfg = MagicMock(
        command=["claude", "--session-id", "sess-1"], env={}, cwd=Path("/tmp/myproj")
    )
    harnesses["claude-code"].build_resume_config = MagicMock(return_value=resume_cfg)
    harnesses["claude-code"].build_spawn_config = MagicMock(return_value=spawn_cfg)

    tmux.is_alive = AsyncMock(return_value=False)
    tmux.respawn_verified = AsyncMock(
        return_value=SpawnResult(ok=False, window_id=None, error="fatal")
    )
    tmux.attach = AsyncMock(return_value=True)

    result = await service.attach_session(session_id=1)

    assert result is False
    tmux.attach.assert_not_called()


# ---------------------------------------------------------------------------
# Task 4: captures_session_id flag + bounded async retry in create_session
# ---------------------------------------------------------------------------


@pytest.fixture
def harnesses_no_capture():
    """Configurator with captures_session_id=False (like claude-code)."""
    h = MagicMock()
    h.captures_session_id = False
    h.generate_session_id.return_value = "sess-nc"
    h.build_spawn_config.return_value = MagicMock(
        command=["claude"], env={}, cwd=Path("/tmp/myproj")
    )
    return {"claude-code": h}


@pytest.fixture
def harnesses_capture():
    """Configurator with captures_session_id=True (like codex)."""
    h = MagicMock()
    h.captures_session_id = True
    h.generate_session_id.return_value = "placeholder-id"
    h.build_spawn_config.return_value = MagicMock(
        command=["codex"], env={}, cwd=Path("/tmp/myproj")
    )
    return {"claude-code": h}


@pytest.mark.asyncio
async def test_create_session_no_capture_flag_never_calls_capture(
    factory, db, tmux, harnesses_no_capture
):
    """When captures_session_id=False, capture_session_id is never called."""
    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_no_capture)
    await service.create_session(project_id=1, harness_name="claude-code")
    harnesses_no_capture["claude-code"].capture_session_id.assert_not_called()


@pytest.mark.asyncio
async def test_create_session_capture_retries_until_success(
    factory, db, tmux, harnesses_capture
):
    """captures_session_id=True: returns None twice then an id → retries and stores captured id."""
    import hqt.sessions.service as svc_mod

    harnesses_capture["claude-code"].capture_session_id.side_effect = [
        None,
        None,
        "captured-id-xyz",
    ]

    sleep_calls = []

    async def fake_sleep(t):
        sleep_calls.append(t)

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
    with patch.object(svc_mod, "_capture_sleep", fake_sleep):
        result = await service.create_session(project_id=1, harness_name="claude-code")

    assert result.session.harness_session_id == "captured-id-xyz"
    # capture was called 3 times: first immediate, then twice after sleeps
    assert harnesses_capture["claude-code"].capture_session_id.call_count == 3
    assert len(sleep_calls) == 2


@pytest.mark.asyncio
async def test_create_session_capture_all_none_keeps_placeholder(
    factory, db, tmux, harnesses_capture, caplog
):
    """All capture attempts return None → placeholder id kept, no crash, warning logged."""
    import logging
    import hqt.sessions.service as svc_mod

    harnesses_capture["claude-code"].capture_session_id.return_value = None

    async def fake_sleep(t):
        pass

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
    with caplog.at_level(logging.WARNING):
        with patch.object(svc_mod, "_capture_sleep", fake_sleep):
            result = await service.create_session(
                project_id=1, harness_name="claude-code"
            )

    assert result.session.harness_session_id == "placeholder-id"
    # Should have attempted up to 6 times (CAPTURE_MAX_ATTEMPTS)
    assert (
        harnesses_capture["claude-code"].capture_session_id.call_count
        == svc_mod.CAPTURE_MAX_ATTEMPTS
    )
    # Exhaustion must emit a warning
    warning_messages = [
        r.message for r in caplog.records if r.levelno == logging.WARNING
    ]
    assert any("exhausted" in m for m in warning_messages)


@pytest.mark.asyncio
async def test_create_session_capture_first_attempt_succeeds(
    factory, db, tmux, harnesses_capture
):
    """captures_session_id=True: first attempt succeeds → no sleeps needed."""
    import hqt.sessions.service as svc_mod

    harnesses_capture["claude-code"].capture_session_id.return_value = "immediate-id"

    sleep_calls = []

    async def fake_sleep(t):
        sleep_calls.append(t)

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
    with patch.object(svc_mod, "_capture_sleep", fake_sleep):
        result = await service.create_session(project_id=1, harness_name="claude-code")

    assert result.session.harness_session_id == "immediate-id"
    assert len(sleep_calls) == 0  # no sleep needed — succeeded on first try


# ---------------------------------------------------------------------------
# Task 7: list_sessions status computation
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_list_sessions_dead_window(factory, db, tmux, harnesses):
    """list_sessions: dead window → status='dead', alive=False."""
    from hqt.tmux.runner import WindowInfo

    await SessionService(
        factory=factory, tmux=tmux, harnesses=harnesses
    ).create_session(project_id=1, harness_name="claude-code")
    tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=False, last_activity=0)}
    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    result = await service.list_sessions(project_id=1)
    assert len(result) == 1
    assert result[0].alive is False
    assert result[0].status == "dead"


@pytest.mark.asyncio
async def test_list_sessions_missing_window_is_dead(factory, db, tmux, harnesses):
    """list_sessions: window not in poll_info result → status='dead', alive=False."""
    await SessionService(
        factory=factory, tmux=tmux, harnesses=harnesses
    ).create_session(project_id=1, harness_name="claude-code")
    tmux.poll_info.return_value = {}  # no windows at all
    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    result = await service.list_sessions(project_id=1)
    assert result[0].alive is False
    assert result[0].status == "dead"


@pytest.mark.asyncio
async def test_list_sessions_alive_parse_working(factory, db, tmux, harnesses):
    """Alive window + harness parse returns 'working' → status='working'."""
    from hqt.tmux.runner import WindowInfo

    await SessionService(
        factory=factory, tmux=tmux, harnesses=harnesses
    ).create_session(project_id=1, harness_name="claude-code")

    now = 2000.0
    tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1999)}
    tmux.capture_pane = AsyncMock(return_value="Esc to interrupt\n")
    harnesses["claude-code"].parse_status = MagicMock(return_value="working")

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    result = await service.list_sessions(project_id=1, now=now)

    assert result[0].status == "working"
    assert result[0].alive is True


@pytest.mark.asyncio
async def test_list_sessions_alive_parse_none_recent_activity(
    factory, db, tmux, harnesses
):
    """Alive window + parse_status None + recent activity → status='active'."""
    from hqt.tmux.runner import WindowInfo

    await SessionService(
        factory=factory, tmux=tmux, harnesses=harnesses
    ).create_session(project_id=1, harness_name="claude-code")

    now = 2000.0
    tmux.poll_info.return_value = {
        "hqt-1": WindowInfo(alive=True, last_activity=1998)
    }  # 2s ago
    tmux.capture_pane = AsyncMock(return_value="some text")
    harnesses["claude-code"].parse_status = MagicMock(return_value=None)

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    result = await service.list_sessions(project_id=1, now=now)

    assert result[0].status == "active"


@pytest.mark.asyncio
async def test_list_sessions_alive_parse_none_stale_activity(
    factory, db, tmux, harnesses
):
    """Alive window + parse_status None + stale activity → status='idle'."""
    from hqt.tmux.runner import WindowInfo

    await SessionService(
        factory=factory, tmux=tmux, harnesses=harnesses
    ).create_session(project_id=1, harness_name="claude-code")

    now = 2000.0
    tmux.poll_info.return_value = {
        "hqt-1": WindowInfo(alive=True, last_activity=1990)
    }  # 10s ago
    tmux.capture_pane = AsyncMock(return_value="some text")
    harnesses["claude-code"].parse_status = MagicMock(return_value=None)

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    result = await service.list_sessions(project_id=1, now=now)

    assert result[0].status == "idle"


@pytest.mark.asyncio
async def test_list_sessions_capture_empty_falls_back_to_activity(
    factory, db, tmux, harnesses
):
    """capture_pane returns empty text → activity-based fallback, no crash."""
    from hqt.tmux.runner import WindowInfo

    await SessionService(
        factory=factory, tmux=tmux, harnesses=harnesses
    ).create_session(project_id=1, harness_name="claude-code")

    now = 2000.0
    tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1999)}
    tmux.capture_pane = AsyncMock(return_value="")  # empty — treat as capture failure
    harnesses["claude-code"].parse_status = MagicMock(return_value=None)

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    result = await service.list_sessions(project_id=1, now=now)

    # Should still produce a valid status via activity
    assert result[0].status in ("active", "idle")
    assert result[0].alive is True


@pytest.mark.asyncio
async def test_list_sessions_alive_parse_waiting(factory, db, tmux, harnesses):
    """Alive window + parse_status 'waiting' → status='waiting'."""
    from hqt.tmux.runner import WindowInfo

    await SessionService(
        factory=factory, tmux=tmux, harnesses=harnesses
    ).create_session(project_id=1, harness_name="claude-code")

    now = 2000.0
    tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1999)}
    tmux.capture_pane = AsyncMock(return_value="Do you want to proceed?\n❯ 1. Yes\n")
    harnesses["claude-code"].parse_status = MagicMock(return_value="waiting")

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    result = await service.list_sessions(project_id=1, now=now)

    assert result[0].status == "waiting"


# ---------------------------------------------------------------------------
# sync_window_labels: status-reflecting tmux window titles (session-wide)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_sync_window_labels_sets_status_symbol(factory, db, tmux, harnesses):
    """sync_window_labels pushes a '<symbol> <nickname>' label per session."""
    from hqt.tmux.runner import WindowInfo

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    await service.create_session(
        project_id=1, harness_name="claude-code", nickname="mywork"
    )

    now = 2000.0
    tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1999)}
    tmux.capture_pane = AsyncMock(return_value="Esc to interrupt\n")
    harnesses["claude-code"].parse_status = MagicMock(return_value="working")
    tmux.set_window_label = AsyncMock()

    infos = await service.sync_window_labels(now=now)

    assert len(infos) == 1
    assert infos[0].status == "working"
    # Label uses the working glyph (◐) and the nickname.
    tmux.set_window_label.assert_awaited_once_with("hqt-1", "◐ mywork")


@pytest.mark.asyncio
async def test_sync_window_labels_dead_window_uses_dead_glyph(
    factory, db, tmux, harnesses
):
    """A dead/missing window gets the ○ glyph and falls back to the window name."""
    from hqt.tmux.runner import WindowInfo

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    await service.create_session(
        project_id=1, harness_name="claude-code"
    )  # no nickname

    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].status == "dead"
    # No nickname → falls back to the tmux window name.
    tmux.set_window_label.assert_awaited_once_with("hqt-1", "○ hqt-1")


@pytest.mark.asyncio
async def test_sync_window_labels_covers_all_projects(factory, db, tmux, harnesses):
    """sync_window_labels is session-wide: it labels sessions across all projects."""
    from hqt.db.models import Project
    from hqt.tmux.runner import WindowInfo

    db.add(Project(name="proj2", path="/tmp/proj2"))
    db.commit()

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    await service.create_session(project_id=1, harness_name="claude-code", nickname="a")
    await service.create_session(project_id=2, harness_name="claude-code", nickname="b")

    tmux.poll_info.return_value = {
        "hqt-1": WindowInfo(alive=False, last_activity=0),
        "hqt-2": WindowInfo(alive=False, last_activity=0),
    }
    tmux.set_window_label = AsyncMock()

    infos = await service.sync_window_labels()

    assert len(infos) == 2  # both projects' sessions processed
    labelled = {c.args[0] for c in tmux.set_window_label.await_args_list}
    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
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_list_sessions_real_configurator_waiting(factory, db):
    """list_sessions with real ClaudeConfigurator + spec'd TmuxManager mock → status='waiting'.

    Uses the REAL ClaudeConfigurator (not a mock) to close the decorative-fixture
    gap and ensure parse_status is correctly wired end-to-end.

    TDD story: before the capture_pane delegate is added to TmuxManager,
    MagicMock(spec=TmuxManager).capture_pane raises AttributeError, which propagates
    as an AttributeError from list_sessions.  After the delegate is added the spec
    allows the attribute and this test passes.
    """
    from hqt.tmux.runner import WindowInfo
    from hqt.tmux.manager import TmuxManager
    from hqt.harnesses.configurators.claude import ClaudeConfigurator

    # spec=TmuxManager: only methods that exist on TmuxManager are accessible.
    # capture_pane MUST be on TmuxManager (via the delegate) for the spec to allow it.
    tmux_spec = MagicMock(spec=TmuxManager)
    tmux_spec.spawn = AsyncMock(
        return_value=SpawnResult(ok=True, window_id="@1", error="")
    )
    tmux_spec.kill = AsyncMock()
    tmux_spec.is_alive = AsyncMock(return_value=True)
    tmux_spec.poll_info = AsyncMock(return_value={})
    # capture_pane returns realistic "waiting" pane text; the spec must allow this
    # attribute — fails with AttributeError before TmuxManager.capture_pane delegate exists.
    tmux_spec.capture_pane = AsyncMock(
        return_value="Do you want to proceed?\n❯ 1. Yes\n"
    )
    tmux_spec.attach = AsyncMock()
    tmux_spec.window_exists = AsyncMock(return_value=True)
    tmux_spec.respawn_verified = AsyncMock(
        return_value=SpawnResult(ok=True, window_id=None, error="")
    )

    configurator = ClaudeConfigurator()
    harnesses = {"claude-code": configurator}
    service = SessionService(factory=factory, tmux=tmux_spec, harnesses=harnesses)

    # Create a session so there is a row in the DB
    await service.create_session(project_id=1, harness_name="claude-code")

    now = 2000.0
    window_name = "hqt-1"
    tmux_spec.poll_info.return_value = {
        window_name: WindowInfo(alive=True, last_activity=1999)
    }

    result = await service.list_sessions(project_id=1, now=now)

    assert len(result) == 1
    assert result[0].status == "waiting"
    assert result[0].alive is True
    tmux_spec.capture_pane.assert_called_once_with(window_name)


# ---------------------------------------------------------------------------
# Item 4: capture_pane NOT called for dead/missing windows
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_list_sessions_dead_window_no_capture(factory, db, tmux):
    """list_sessions: dead window → capture_pane must NOT be called."""
    from hqt.tmux.runner import WindowInfo
    from hqt.harnesses.configurators.claude import ClaudeConfigurator

    harnesses = {"claude-code": ClaudeConfigurator()}
    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    await service.create_session(project_id=1, harness_name="claude-code")

    tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=False, last_activity=0)}
    tmux.capture_pane.reset_mock()

    await service.list_sessions(project_id=1)

    tmux.capture_pane.assert_not_called()


@pytest.mark.asyncio
async def test_list_sessions_missing_window_no_capture(factory, db, tmux):
    """list_sessions: window missing from poll_info → capture_pane must NOT be called."""
    from hqt.harnesses.configurators.claude import ClaudeConfigurator

    harnesses = {"claude-code": ClaudeConfigurator()}
    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    await service.create_session(project_id=1, harness_name="claude-code")

    tmux.poll_info.return_value = {}  # window not present at all
    tmux.capture_pane.reset_mock()

    await service.list_sessions(project_id=1)

    tmux.capture_pane.assert_not_called()


# ---------------------------------------------------------------------------
# Fix 1: _respawn_with_fallback rung-2 updates harness_session_id
# ---------------------------------------------------------------------------


@pytest.fixture
def harnesses_capture_fallback():
    """Configurator with captures_session_id=True for fallback ladder tests."""
    h = MagicMock()
    h.name = "claude-code"
    h.captures_session_id = True
    h.generate_session_id.return_value = "placeholder-id"
    h.build_spawn_config.return_value = MagicMock(
        command=["codex"], env={}, cwd=Path("/tmp/myproj")
    )
    h.build_resume_config.return_value = MagicMock(
        command=["codex", "resume", "placeholder-id"], env={}, cwd=Path("/tmp/myproj")
    )
    return {"claude-code": h}


@pytest.mark.asyncio
async def test_respawn_fallback_rung2_updates_harness_session_id(
    factory, db, tmux, harnesses_capture_fallback
):
    """Rung-2 success + captures_session_id=True + capture returns new id → sess.harness_session_id updated and committed."""
    import hqt.sessions.service as svc_mod

    harnesses_capture_fallback[
        "claude-code"
    ].capture_session_id.return_value = "new-codex-id"

    async def fake_sleep(t):
        pass

    service = SessionService(
        factory=factory, tmux=tmux, harnesses=harnesses_capture_fallback
    )
    with patch.object(svc_mod, "_capture_sleep", fake_sleep):
        result = await service.create_session(project_id=1, harness_name="claude-code")

    sess_id = result.session.id

    # Simulate rung-1 resume fails, rung-2 fresh spawn succeeds
    harnesses_capture_fallback["claude-code"].capture_session_id.reset_mock()
    harnesses_capture_fallback[
        "claude-code"
    ].capture_session_id.return_value = "new-codex-id-after-rung2"

    tmux.is_alive = AsyncMock(return_value=False)
    tmux.respawn_verified = AsyncMock(
        side_effect=[
            SpawnResult(
                ok=False, window_id=None, error="no transcript"
            ),  # rung-1 resume fails
            SpawnResult(ok=True, window_id="@2", error=""),  # rung-2 fresh spawn ok
        ]
    )
    tmux.attach = AsyncMock(return_value=True)

    with patch.object(svc_mod, "_capture_sleep", fake_sleep):
        ok = await service.attach_session(session_id=sess_id)

    assert ok is True
    # harness_session_id must now be the newly captured id
    from hqt.db.models import Session as DBSession

    updated = db.get(DBSession, sess_id)
    assert updated.harness_session_id == "new-codex-id-after-rung2"


@pytest.mark.asyncio
async def test_attach_late_capture_updates_id_before_resume(
    factory, db, tmux, harnesses_capture_fallback
):
    """If create-time Codex capture missed, attach captures the real id before resuming."""
    import hqt.sessions.service as svc_mod

    harness = harnesses_capture_fallback["claude-code"]
    harness.capture_session_id.side_effect = [None] * svc_mod.CAPTURE_MAX_ATTEMPTS + [
        "late-codex-id"
    ]

    async def fake_sleep(t):
        pass

    service = SessionService(
        factory=factory, tmux=tmux, harnesses=harnesses_capture_fallback
    )
    with patch.object(svc_mod, "_capture_sleep", fake_sleep):
        result = await service.create_session(project_id=1, harness_name="claude-code")

    assert result.session.harness_session_id == "placeholder-id"

    tmux.is_alive = AsyncMock(return_value=False)
    tmux.respawn_verified = AsyncMock(
        return_value=SpawnResult(ok=True, window_id="@2", error="")
    )
    tmux.attach = AsyncMock(return_value=True)

    ok = await service.attach_session(session_id=result.session.id)

    assert ok is True
    harness.build_resume_config.assert_called_with(
        Path("/tmp/myproj"), "late-codex-id", result.session.model, sandboxed=False
    )
    from hqt.db.models import Session as DBSession

    updated = db.get(DBSession, result.session.id)
    assert updated.harness_session_id == "late-codex-id"


@pytest.mark.asyncio
async def test_respawn_fallback_rung2_capture_exhausted_keeps_old_id(
    factory, db, tmux, harnesses_capture_fallback, caplog
):
    """Rung-2 success + capture exhausted → old id kept + warning logged."""
    import logging
    import hqt.sessions.service as svc_mod

    harnesses_capture_fallback[
        "claude-code"
    ].capture_session_id.return_value = "placeholder-id"

    async def fake_sleep(t):
        pass

    service = SessionService(
        factory=factory, tmux=tmux, harnesses=harnesses_capture_fallback
    )
    with patch.object(svc_mod, "_capture_sleep", fake_sleep):
        result = await service.create_session(project_id=1, harness_name="claude-code")

    sess_id = result.session.id
    old_id = result.session.harness_session_id

    # After create, reset capture to always return None
    harnesses_capture_fallback["claude-code"].capture_session_id.reset_mock()
    harnesses_capture_fallback["claude-code"].capture_session_id.return_value = None

    tmux.is_alive = AsyncMock(return_value=False)
    tmux.respawn_verified = AsyncMock(
        side_effect=[
            SpawnResult(ok=False, window_id=None, error="no transcript"),
            SpawnResult(ok=True, window_id="@2", error=""),
        ]
    )
    tmux.attach = AsyncMock(return_value=True)

    with caplog.at_level(logging.WARNING):
        with patch.object(svc_mod, "_capture_sleep", fake_sleep):
            ok = await service.attach_session(session_id=sess_id)

    assert ok is True
    from hqt.db.models import Session as DBSession

    updated = db.get(DBSession, sess_id)
    # Old id preserved
    assert updated.harness_session_id == old_id
    # Warning emitted
    warning_messages = [
        r.message for r in caplog.records if r.levelno == logging.WARNING
    ]
    assert any("exhausted" in m for m in warning_messages)


@pytest.mark.asyncio
async def test_respawn_fallback_rung2_no_capture_when_flag_false(factory, db, tmux):
    """Rung-2 success + captures_session_id=False → capture_session_id never called."""
    h = MagicMock()
    h.name = "claude-code"
    h.captures_session_id = False
    h.generate_session_id.return_value = "sess-nc"
    h.build_spawn_config.return_value = MagicMock(
        command=["claude"], env={}, cwd=Path("/tmp/myproj")
    )
    h.build_resume_config.return_value = MagicMock(
        command=["claude", "--resume", "sess-nc"], env={}, cwd=Path("/tmp/myproj")
    )
    harnesses_nc = {"claude-code": h}

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_nc)
    result = await service.create_session(project_id=1, harness_name="claude-code")
    sess_id = result.session.id

    h.capture_session_id.reset_mock()

    tmux.is_alive = AsyncMock(return_value=False)
    tmux.respawn_verified = AsyncMock(
        side_effect=[
            SpawnResult(ok=False, window_id=None, error="no transcript"),
            SpawnResult(ok=True, window_id="@2", error=""),
        ]
    )
    tmux.attach = AsyncMock(return_value=True)

    await service.attach_session(session_id=sess_id)

    h.capture_session_id.assert_not_called()


# ---------------------------------------------------------------------------
# Fix 2 (Minor): capture NOT called when spawn_result.ok is False
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_create_session_no_capture_when_spawn_fails(
    factory, db, tmux, harnesses_capture
):
    """create_session: spawn fails → capture_session_id must NOT be called."""
    import hqt.sessions.service as svc_mod

    tmux.spawn.return_value = SpawnResult(
        ok=False, window_id=None, error="spawn failed"
    )

    async def fake_sleep(t):
        pass

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
    with patch.object(svc_mod, "_capture_sleep", fake_sleep):
        result = await service.create_session(project_id=1, harness_name="claude-code")

    harnesses_capture["claude-code"].capture_session_id.assert_not_called()
    assert result.spawn_ok is False
    assert "spawn failed" in result.spawn_error


# ---------------------------------------------------------------------------
# Fix 3: create_session returns CreateSessionResult
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_create_session_returns_result_dataclass(service, db, tmux, harnesses):
    """create_session returns a CreateSessionResult with spawn_ok=True on success."""
    from hqt.sessions.service import CreateSessionResult

    result = await service.create_session(
        project_id=1, harness_name="claude-code", nickname="test"
    )
    assert isinstance(result, CreateSessionResult)
    assert result.spawn_ok is True
    assert result.spawn_error == ""
    assert result.session.id is not None


@pytest.mark.asyncio
async def test_create_session_result_propagates_spawn_failure(
    factory, db, tmux, harnesses
):
    """create_session result reflects spawn failure."""
    from hqt.sessions.service import CreateSessionResult

    tmux.spawn.return_value = SpawnResult(
        ok=False, window_id=None, error="command not found: claude"
    )
    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    result = await service.create_session(project_id=1, harness_name="claude-code")

    assert isinstance(result, CreateSessionResult)
    assert result.spawn_ok is False
    assert "command not found" in result.spawn_error
    # Session row was still committed
    assert result.session.id is not None


# ---------------------------------------------------------------------------
# rename_session / get_session
# ---------------------------------------------------------------------------


def _seed_session(db, project_id: int = 1, nickname=None) -> Session:
    """Create a session for the seeded 'claude-code' harness."""
    harness = db.query(Harness).filter_by(name="claude-code").first()
    sess = Session(
        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)
    db.commit()
    return sess


def test_rename_session_sets_nickname(service, db):
    from hqt.db.models import Session

    sess = _seed_session(db, nickname="old")
    service.rename_session(sess.id, "new-name")
    db.expire_all()
    assert db.get(Session, sess.id).nickname == "new-name"


def test_rename_session_empty_clears_to_none(service, db):
    from hqt.db.models import Session

    sess = _seed_session(db, nickname="old")
    service.rename_session(sess.id, "")
    db.expire_all()
    assert db.get(Session, sess.id).nickname is None


def test_rename_session_whitespace_clears_to_none(service, db):
    from hqt.db.models import Session

    sess = _seed_session(db, nickname="old")
    service.rename_session(sess.id, "   ")
    db.expire_all()
    assert db.get(Session, sess.id).nickname is None


def test_get_session_returns_row(service, db):
    sess = _seed_session(db, nickname="x")
    assert service.get_session(sess.id).id == sess.id


def test_get_session_missing_returns_none(service):
    assert service.get_session(99999) is None


# ---------------------------------------------------------------------------
# Task 4: per-operation sessions, ServiceError, eager-load harness
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_create_session_unknown_project_raises(service):
    with pytest.raises(ServiceError, match="no longer exists"):
        await service.create_session(9999, "claude-code")


@pytest.mark.asyncio
async def test_create_session_uninstalled_harness_raises(factory, db, tmux):
    # Harness row exists in the DB but the binary is gone from PATH
    # (self.harnesses has no configurator for it).
    service = SessionService(factory=factory, tmux=tmux, harnesses={})
    with pytest.raises(ServiceError, match="not installed"):
        await service.create_session(1, "claude-code")


@pytest.mark.asyncio
async def test_attach_session_missing_row_raises(service):
    with pytest.raises(ServiceError, match="not found"):
        await service.attach_session(9999)


@pytest.mark.asyncio
async def test_list_sessions_rows_usable_after_return(service, db, tmux):
    await service.create_session(1, "claude-code")
    infos = await service.list_sessions(1)
    # The widget reads session.harness.name on detached rows; selectinload
    # must have eager-loaded the relationship.
    assert infos[0].session.harness.name == "claude-code"


@pytest.mark.asyncio
async def test_list_sessions_skips_capture_for_deleted_project(
    factory, db, tmux, harnesses_capture
):
    # A session whose project row was deleted must not crash the poller.
    # Capture returns None at create time so the placeholder id is kept; the
    # poller then re-enters the capture path and must SKIP on the missing
    # project rather than raise.
    import hqt.sessions.service as svc_mod
    from hqt.db.models import Session

    harnesses_capture["claude-code"].capture_session_id.return_value = None

    async def fake_sleep(t):
        pass

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
    with patch.object(svc_mod, "_capture_sleep", fake_sleep):
        await service.create_session(1, "claude-code")

    # The placeholder id must still be in place so the poller attempts capture.
    db.expire_all()
    fresh = db.query(Session).filter_by(project_id=1).first()
    assert fresh.harness_session_id == "placeholder-id"

    db.query(Project).delete()
    db.commit()

    # Capture would now find an id, but the project lookup must short-circuit
    # before it is consulted; either way no ServiceError may escape.
    harnesses_capture["claude-code"].capture_session_id.return_value = "would-capture"
    infos = await service.list_sessions(1)
    assert len(infos) == 1  # no ServiceError escaped


# ---------------------------------------------------------------------------
# Task 3: worktree-aware session lifecycle
# ---------------------------------------------------------------------------


WT_PATH = Path("/tmp/myproj/.worktrees/feature-x")


@pytest.fixture
def fake_worktree(monkeypatch):
    """Monkeypatch hqt.git.worktree functions with recording async fakes.

    Returns a namespace exposing the recorded call args and the WorktreeState
    class so tests can assert ordering and arguments.
    """
    import hqt.git.worktree as wt_mod

    class Recorder:
        def __init__(self):
            self.create_calls = []
            self.add_calls = []
            self.remove_calls = []
            self.state_calls = []
            self.create_path = WT_PATH
            self.create_error: Exception | None = None
            self.add_error: Exception | None = None
            self.remove_error: Exception | None = None
            self.state_result = wt_mod.WorktreeState(
                exists=True, dirty=False, unique_commits=2
            )

        async def create_worktree(self, repo, branch):
            self.create_calls.append((repo, branch))
            if self.create_error is not None:
                raise self.create_error
            return self.create_path

        async def add_worktree_for_branch(self, repo, branch):
            self.add_calls.append((repo, branch))
            if self.add_error is not None:
                raise self.add_error
            return self.create_path

        async def remove_worktree(self, repo, path, branch, force):
            self.remove_calls.append((repo, path, branch, force))
            if self.remove_error is not None:
                raise self.remove_error

        async def worktree_state(self, repo, path, branch):
            self.state_calls.append((repo, path, branch))
            return self.state_result

    rec = Recorder()
    monkeypatch.setattr(wt_mod, "create_worktree", rec.create_worktree)
    monkeypatch.setattr(wt_mod, "add_worktree_for_branch", rec.add_worktree_for_branch)
    monkeypatch.setattr(wt_mod, "remove_worktree", rec.remove_worktree)
    monkeypatch.setattr(wt_mod, "worktree_state", rec.worktree_state)
    return rec


@pytest.mark.asyncio
async def test_create_session_worktree_branch(
    service, db, tmux, harnesses, fake_worktree
):
    """worktree_branch set → create_worktree called with (project_path, branch);
    spawn cwd and capture use the worktree path; row stores both fields."""
    harnesses["claude-code"].captures_session_id = True
    harnesses["claude-code"].capture_session_id.return_value = "cap-id"
    harnesses["claude-code"].build_spawn_config.return_value = MagicMock(
        command=["claude"], env={}, cwd=WT_PATH
    )

    result = await service.create_session(
        project_id=1, harness_name="claude-code", worktree_branch="feature-x"
    )

    # create_worktree called with project path + branch
    assert fake_worktree.create_calls == [(Path("/tmp/myproj"), "feature-x")]
    # spawn request cwd is the worktree path
    spawn_req = tmux.spawn.call_args.args[0]
    assert spawn_req.cwd == str(WT_PATH)
    # build_spawn_config used the worktree path, not the project path
    assert harnesses["claude-code"].build_spawn_config.call_args.args[0] == WT_PATH
    # capture used the worktree path
    assert harnesses["claude-code"].capture_session_id.call_args.args[0] == WT_PATH
    # row carries both worktree fields
    sess = result.session
    assert sess.worktree_path == str(WT_PATH)
    assert sess.worktree_branch == "feature-x"


@pytest.mark.asyncio
async def test_create_session_worktree_blank_nickname_defaults_to_branch(
    service, db, tmux, harnesses, fake_worktree
):
    """Blank nickname + worktree_branch → nickname defaults to branch name."""
    result = await service.create_session(
        project_id=1,
        harness_name="claude-code",
        nickname="",
        worktree_branch="feature-x",
    )
    assert result.session.nickname == "feature-x"


@pytest.mark.asyncio
async def test_create_session_worktree_explicit_nickname_preserved(
    service, db, tmux, harnesses, fake_worktree
):
    """Explicit nickname + worktree_branch → nickname preserved."""
    result = await service.create_session(
        project_id=1,
        harness_name="claude-code",
        nickname="my-work",
        worktree_branch="feature-x",
    )
    assert result.session.nickname == "my-work"


@pytest.mark.asyncio
async def test_create_session_worktree_create_fails_no_row_no_spawn(
    factory, db, tmux, harnesses, fake_worktree
):
    """create_worktree raising ServiceError → propagates, no Session row, no spawn."""
    from hqt.db.models import Session

    fake_worktree.create_error = ServiceError("git worktree add failed")
    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)

    with pytest.raises(ServiceError, match="worktree add failed"):
        await service.create_session(
            project_id=1, harness_name="claude-code", worktree_branch="feature-x"
        )

    # No Session row committed
    db.expire_all()
    assert db.query(Session).count() == 0
    # tmux spawn never called
    tmux.spawn.assert_not_called()


@pytest.mark.asyncio
async def test_create_session_without_worktree_branch_regression(
    service, db, tmux, harnesses, fake_worktree
):
    """No worktree_branch → create_worktree not called, row worktree fields None."""
    result = await service.create_session(project_id=1, harness_name="claude-code")
    assert fake_worktree.create_calls == []
    assert result.session.worktree_path is None
    assert result.session.worktree_branch is None


def _seed_worktree_session(db, branch="feature-x", path=str(WT_PATH)):
    """Seed a worktree session for project 1 / harness claude-code."""
    from hqt.db.models import Harness, Session

    harness = db.query(Harness).filter_by(name="claude-code").first()
    sess = Session(
        project_id=1,
        harness_id=harness.id,
        tmux_session_name="hqt-wt",
        harness_session_id="sess-1",
        worktree_path=path,
        worktree_branch=branch,
        archived=False,
    )
    db.add(sess)
    db.commit()
    return sess


@pytest.mark.asyncio
async def test_attach_session_recreates_missing_worktree(
    factory, db, tmux, harnesses, fake_worktree
):
    """attach_session: worktree dir missing on disk → add_worktree_for_branch called."""
    # Use a path that does not exist on disk.
    missing = "/tmp/does-not-exist/.worktrees/feature-x"
    sess = _seed_worktree_session(db, path=missing)

    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)

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    await service.attach_session(session_id=sess.id)

    assert fake_worktree.add_calls == [(Path("/tmp/myproj"), "feature-x")]


@pytest.mark.asyncio
async def test_attach_session_missing_worktree_recreate_fails(
    factory, db, tmux, harnesses, fake_worktree
):
    """attach_session: recreate raises → ServiceError mentioning deleting the session."""
    missing = "/tmp/does-not-exist/.worktrees/feature-x"
    sess = _seed_worktree_session(db, path=missing)
    fake_worktree.add_error = ServiceError("branch gone")

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    with pytest.raises(ServiceError, match="delete the session"):
        await service.attach_session(session_id=sess.id)


@pytest.mark.asyncio
async def test_attach_session_worktree_resume_uses_worktree_path(
    factory, db, tmux, harnesses, fake_worktree
):
    """Resume config for a worktree session uses the worktree path as cwd."""
    # WT_PATH parent exists (/tmp/myproj/.worktrees won't, but use a real existing dir)
    import tempfile

    with tempfile.TemporaryDirectory() as tmpdir:
        sess = _seed_worktree_session(db, path=tmpdir)

        resume_cfg = MagicMock(command=["claude", "--resume"], env={}, cwd=Path(tmpdir))
        harnesses["claude-code"].build_resume_config = MagicMock(
            return_value=resume_cfg
        )
        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)

        service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
        await service.attach_session(session_id=sess.id)

        # build_resume_config received the worktree path as the project path arg
        assert harnesses["claude-code"].build_resume_config.call_args.args[0] == Path(
            tmpdir
        )
        # since the worktree exists on disk, recreate must not have been attempted
        assert fake_worktree.add_calls == []


@pytest.mark.asyncio
async def test_delete_session_remove_worktree_true(
    factory, db, tmux, harnesses, fake_worktree
):
    """delete_session(remove_worktree=True) calls remove_worktree and deletes the row."""
    from hqt.db.models import Session

    sess = _seed_worktree_session(db)
    sess_id = sess.id

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    await service.delete_session(sess_id, remove_worktree=True, force=True)

    assert fake_worktree.remove_calls == [
        (Path("/tmp/myproj"), WT_PATH, "feature-x", True)
    ]
    db.expire_all()
    assert db.get(Session, sess_id) is None


@pytest.mark.asyncio
async def test_delete_session_remove_worktree_raises_keeps_row(
    factory, db, tmux, harnesses, fake_worktree
):
    """remove_worktree raising → ServiceError propagates, row still present."""
    from hqt.db.models import Session

    sess = _seed_worktree_session(db)
    sess_id = sess.id
    fake_worktree.remove_error = ServiceError("worktree dirty")

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    with pytest.raises(ServiceError, match="worktree dirty"):
        await service.delete_session(sess_id, remove_worktree=True)

    db.expire_all()
    assert db.get(Session, sess_id) is not None


@pytest.mark.asyncio
async def test_delete_session_remove_worktree_false_skips_removal(
    factory, db, tmux, harnesses, fake_worktree
):
    """remove_worktree=False → remove_worktree not called, row deleted."""
    from hqt.db.models import Session

    sess = _seed_worktree_session(db)
    sess_id = sess.id

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    await service.delete_session(sess_id, remove_worktree=False)

    assert fake_worktree.remove_calls == []
    db.expire_all()
    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
):
    """worktree_state_for: plain session → None."""
    sess = _seed_session(db)
    assert await service.worktree_state_for(sess.id) is None


@pytest.mark.asyncio
async def test_worktree_state_for_missing_id_returns_none(service, db, fake_worktree):
    """worktree_state_for: unknown session id → None."""
    assert await service.worktree_state_for(99999) is None


@pytest.mark.asyncio
async def test_worktree_state_for_worktree_session_returns_state(
    factory, db, tmux, harnesses, fake_worktree
):
    """worktree_state_for: worktree session → returns the module's WorktreeState."""
    sess = _seed_worktree_session(db)
    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)

    state = await service.worktree_state_for(sess.id)

    assert state is fake_worktree.state_result
    assert fake_worktree.state_calls == [(Path("/tmp/myproj"), WT_PATH, "feature-x")]


@pytest.mark.asyncio
async def test_repo_path_fallback_slash_branch_when_project_gone(
    factory, db, tmux, harnesses, fake_worktree
):
    """When the project row is gone, the repo is derived from the worktree
    layout even for slash-containing branch names (feature/foo).

    create_worktree builds the path as <repo>/.worktrees/<branch>, so for a
    branch like 'feature/foo' the path is <repo>/.worktrees/feature/foo. The
    fallback must strip '.worktrees/feature/foo' and recover <repo>, not the
    wrong parent.parent result (<repo>/.worktrees).
    """
    from hqt.db.models import Session

    repo = Path("/tmp/myproj")
    branch = "feature/foo"
    # Mirror how worktree.create_worktree builds the path.
    wt_path = repo / ".worktrees" / branch
    sess = _seed_worktree_session(db, branch=branch, path=str(wt_path))
    sess_id = sess.id

    # Delete the project row so the fallback derivation is exercised.
    db.query(Project).delete()
    db.commit()
    db.expire_all()
    # Session row outlives the project (no FK cascade in this seed).
    assert db.get(Session, sess_id) is not None

    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
    await service.worktree_state_for(sess_id)

    # The fallback must recover the real repo, not <repo>/.worktrees.
    assert fake_worktree.state_calls == [(repo, wt_path, branch)]


# ---------------------------------------------------------------------------
# Task 3 (P2): serialize the spawn->capture window with a lock
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_create_session_holds_capture_lock_for_capturing_harness(
    factory, db, tmux
):
    observed = {}

    def _capture(project_path, since):
        # capture runs inside the spawn->capture critical section
        observed["locked_during_capture"] = service._capture_lock.locked()
        return "real-codex-id"

    h = MagicMock()
    h.captures_session_id = True
    h.generate_session_id.return_value = "placeholder"
    h.build_spawn_config.return_value = MagicMock(
        command=["codex"], env={}, cwd=Path("/tmp/myproj")
    )
    h.capture_session_id.side_effect = _capture
    service = SessionService(factory=factory, tmux=tmux, harnesses={"claude-code": h})

    result = await service.create_session(project_id=1, harness_name="claude-code")

    assert observed["locked_during_capture"] is True
    assert result.session.harness_session_id == "real-codex-id"
    # Lock is released after create_session returns.
    assert service._capture_lock.locked() is False


@pytest.mark.asyncio
async def test_create_session_no_lock_for_non_capturing_harness(service, db, tmux):
    observed = {}

    async def _spawn(req):
        observed["locked_during_spawn"] = service._capture_lock.locked()
        return SpawnResult(ok=True, window_id="@1", error="")

    tmux.spawn.side_effect = _spawn

    await service.create_session(project_id=1, harness_name="claude-code")

    # The `service` fixture's harness has captures_session_id = False, so the
    # spawn must NOT be serialized under the capture lock.
    assert observed["locked_during_spawn"] is False


# ---------------------------------------------------------------------------
# Task 5: sandbox wrapping on spawn and resume
# ---------------------------------------------------------------------------


@pytest.fixture
def sandbox_harness():
    h = MagicMock()
    h.captures_session_id = False
    h.generate_session_id.return_value = "sess-1"
    h.build_spawn_config.return_value = MagicMock(
        command=["claude", "--dangerously-skip-permissions"],
        env={},
        cwd=Path("/tmp/myproj"),
    )
    h.sandbox_binds.return_value = []
    h.parse_status = MagicMock(return_value=None)
    return {"claude-code": h}


@pytest.fixture
def sandbox_service(factory, db, tmux, sandbox_harness):
    return SessionService(factory=factory, tmux=tmux, harnesses=sandbox_harness)


@pytest.mark.asyncio
async def test_sandboxed_create_wraps_command(
    sandbox_service, db, tmux, sandbox_harness
):
    with (
        patch("hqt.sessions.service.sandbox.is_available", return_value=True),
        patch(
            "hqt.sessions.service.sandbox.wrap", return_value=["bwrap", "--", "claude"]
        ) as mock_wrap,
    ):
        result = await sandbox_service.create_session(
            project_id=1,
            harness_name="claude-code",
            sandbox=SandboxPolicy(fs="rw", net=True),
        )
    mock_wrap.assert_called_once()
    sandbox_harness["claude-code"].build_spawn_config.assert_called_once()
    assert (
        sandbox_harness["claude-code"].build_spawn_config.call_args.kwargs["sandboxed"]
        is True
    )
    assert tmux.spawn.call_args.args[0].command == ["bwrap", "--", "claude"]
    assert result.session.sandbox_json == '{"fs": "rw", "net": true}'


@pytest.mark.asyncio
async def test_unsandboxed_create_does_not_wrap(service, db, tmux, harnesses):
    with patch("hqt.sessions.service.sandbox.wrap") as mock_wrap:
        result = await service.create_session(project_id=1, harness_name="claude-code")
    mock_wrap.assert_not_called()
    assert result.session.sandbox_json is None


@pytest.mark.asyncio
async def test_sandboxed_create_raises_when_bwrap_missing(sandbox_service, db):
    with patch("hqt.sessions.service.sandbox.is_available", return_value=False):
        with pytest.raises(ServiceError, match="bubblewrap"):
            await sandbox_service.create_session(
                project_id=1,
                harness_name="claude-code",
                sandbox=SandboxPolicy(fs="rw", net=True),
            )


@pytest.mark.asyncio
async def test_sandboxed_worktree_binds_repo_git_dir(
    sandbox_service, db, tmux, sandbox_harness, fake_worktree
):
    """A sandboxed worktree session binds the repo's common .git writable so
    git works inside the jail (the worktree's gitdir lives under <repo>/.git)."""
    from hqt.sandbox import Bind

    with (
        patch("hqt.sessions.service.sandbox.is_available", return_value=True),
        patch(
            "hqt.sessions.service.sandbox.wrap", return_value=["bwrap", "--", "claude"]
        ) as mock_wrap,
    ):
        await sandbox_service.create_session(
            project_id=1,
            harness_name="claude-code",
            worktree_branch="feature-x",
            sandbox=SandboxPolicy(fs="rw", net=True),
        )
    binds = mock_wrap.call_args.args[3]
    assert Bind(Path("/tmp/myproj/.git"), writable=True) in binds


# ---------------------------------------------------------------------------
# Task 4: SessionService tool-window methods
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_open_tool_window_spawns_styled_window(service, db, tmux, monkeypatch):
    await service.create_session(project_id=1, harness_name="claude-code")
    monkeypatch.setattr("hqt.sessions.service.shutil.which", lambda b: "/usr/bin/" + b)
    tmux.open_aux_window = AsyncMock(return_value="@5")

    wid = await service.open_tool_window(1, "lazygit")

    assert wid == "@5"
    tmux.open_aux_window.assert_awaited_once_with(
        "lazygit", "/tmp/myproj", ["lazygit"], "lazygit · myproj"
    )


@pytest.mark.asyncio
async def test_open_tool_window_shell_skips_which_check(service, db, tmux, monkeypatch):
    await service.create_session(project_id=1, harness_name="claude-code")
    monkeypatch.setattr("hqt.sessions.service.shutil.which", lambda b: None)
    tmux.open_aux_window = AsyncMock(return_value="@3")

    wid = await service.open_tool_window(1, "shell")

    assert wid == "@3"
    tmux.open_aux_window.assert_awaited_once_with(
        "shell", "/tmp/myproj", [], "shell · myproj"
    )


@pytest.mark.asyncio
async def test_open_tool_window_unknown_tool_raises(service):
    with pytest.raises(ServiceError):
        await service.open_tool_window(1, "emacs")


@pytest.mark.asyncio
async def test_open_tool_window_missing_binary_raises(service, db, monkeypatch):
    await service.create_session(project_id=1, harness_name="claude-code")
    monkeypatch.setattr("hqt.sessions.service.shutil.which", lambda b: None)
    with pytest.raises(ServiceError):
        await service.open_tool_window(1, "lazygit")


@pytest.mark.asyncio
async def test_open_tool_window_unknown_session_raises(service, monkeypatch):
    monkeypatch.setattr("hqt.sessions.service.shutil.which", lambda b: "/usr/bin/" + b)
    with pytest.raises(ServiceError):
        await service.open_tool_window(999, "nvim")


@pytest.mark.asyncio
async def test_session_id_for_window_resolves_and_misses(service, db):
    await service.create_session(project_id=1, harness_name="claude-code")
    assert service.session_id_for_window("hqt-1") == 1
    assert service.session_id_for_window("not-an-hqt-window") is None


@pytest.mark.asyncio
async def test_require_session_id_for_window_resolves_or_raises(service, db):
    # The shared guard behind tool/stop/rename: id on a hit, ServiceError on a miss.
    await service.create_session(project_id=1, harness_name="claude-code")
    assert service.require_session_id_for_window("hqt-1") == 1
    with pytest.raises(ServiceError):
        service.require_session_id_for_window("not-an-hqt-window")


@pytest.mark.asyncio
async def test_open_tool_window_for_window_resolves_by_name(
    service, db, tmux, monkeypatch
):
    await service.create_session(project_id=1, harness_name="claude-code")
    monkeypatch.setattr("hqt.sessions.service.shutil.which", lambda b: "/usr/bin/" + b)
    tmux.open_aux_window = AsyncMock(return_value="@2")

    wid = await service.open_tool_window_for_window("hqt-1", "nvim")

    assert wid == "@2"
    tmux.open_aux_window.assert_awaited_once_with(
        "nvim", "/tmp/myproj", ["nvim"], "nvim · myproj"
    )


@pytest.mark.asyncio
async def test_open_tool_window_for_window_unknown_window_raises(service):
    with pytest.raises(ServiceError):
        await service.open_tool_window_for_window("not-an-hqt-window", "nvim")


# ---------------------------------------------------------------------------
# Task 5: clone_session_for_window
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_clone_session_for_window_reuses_project_harness_model(
    service, db, monkeypatch
):
    await service.create_session(
        project_id=1, harness_name="claude-code", nickname="orig", model="opus"
    )
    captured = {}

    async def fake_create(
        project_id, harness_name, nickname=None, model=None, worktree_branch=None
    ):
        captured["args"] = (project_id, harness_name, nickname, model)
        return "SENTINEL"

    monkeypatch.setattr(service, "create_session", fake_create)

    result = await service.clone_session_for_window("hqt-1")

    assert result == "SENTINEL"
    # Same project + harness + model; a fresh sibling, so no nickname.
    assert captured["args"] == (1, "claude-code", None, "opus")


@pytest.mark.asyncio
async def test_clone_session_for_window_unknown_window_raises(service):
    with pytest.raises(ServiceError):
        await service.clone_session_for_window("not-an-hqt-window")


@pytest.mark.asyncio
async def test_stop_session_for_window_kills_by_name(service, db, tmux):
    await service.create_session(project_id=1, harness_name="claude-code")
    await service.stop_session_for_window("hqt-1")
    tmux.kill.assert_called_once_with("hqt-1")


@pytest.mark.asyncio
async def test_stop_session_for_window_unknown_window_raises(service):
    with pytest.raises(ServiceError):
        await service.stop_session_for_window("not-an-hqt-window")


@pytest.mark.asyncio
async def test_list_sessions_archived_filter(service, db, tmux):
    h = db.query(Harness).filter_by(name="claude-code").first()
    active = Session(
        project_id=1, harness_id=h.id, tmux_session_name="hqt-active", archived=False
    )
    arch = Session(
        project_id=1, harness_id=h.id, tmux_session_name="hqt-arch", archived=True
    )
    db.add_all([active, arch])
    db.commit()
    active_id, arch_id = active.id, arch.id

    default = await service.list_sessions(project_id=1)
    assert [i.session.id for i in default] == [active_id]

    archived = await service.list_sessions(project_id=1, archived=True)
    assert [i.session.id for i in archived] == [arch_id]


@pytest.mark.asyncio
async def test_archive_session_kills_window_and_keeps_row(service, db, tmux):
    await service.create_session(project_id=1, harness_name="claude-code")
    await service.archive_session(1)

    tmux.kill.assert_called_once_with("hqt-1")
    db.expire_all()
    row = db.get(Session, 1)
    assert row is not None
    assert row.archived is True
    # Excluded from the default (active) listing.
    assert await service.list_sessions(project_id=1) == []


@pytest.mark.asyncio
async def test_archive_session_no_kill_when_window_missing(service, db, tmux):
    await service.create_session(project_id=1, harness_name="claude-code")
    tmux.window_exists = AsyncMock(return_value=False)
    tmux.kill.reset_mock()

    await service.archive_session(1)

    tmux.kill.assert_not_called()
    db.expire_all()
    assert db.get(Session, 1).archived is True


@pytest.mark.asyncio
async def test_unarchive_session_restores_to_active_list(service, db, tmux):
    await service.create_session(project_id=1, harness_name="claude-code")
    await service.archive_session(1)

    await service.unarchive_session(1)

    db.expire_all()
    assert db.get(Session, 1).archived is False
    active = await service.list_sessions(project_id=1)
    assert [i.session.id for i in active] == [1]