test_shell_drive_finalize.py

"""Tests for koru.autonomy.shell_drive_finalize."""

from __future__ import annotations

import subprocess
from pathlib import Path

import pytest

import koru.queue.ticket as ticket_mod
import koru.tillm_bridge as bridge_mod
from koru.autonomy import post_run_verify as verify_mod
from koru.autonomy.shell_drive_finalize import finalize_shell_drive_ticket


class _PlanfileRecorder:
    def __init__(self, done_rc: int = 0) -> None:
        self.calls: list[list[str]] = []
        self.done_rc = done_rc

    def __call__(self, project: Path, args, runner=None):
        self.calls.append(list(args))
        rc = self.done_rc if args[:2] == ["ticket", "done"] else 0
        return subprocess.CompletedProcess(list(args), rc, stdout="", stderr="")

    def commands(self) -> list[str]:
        return [" ".join(c[:2]) for c in self.calls]


@pytest.fixture()
def planfile(monkeypatch: pytest.MonkeyPatch) -> _PlanfileRecorder:
    rec = _PlanfileRecorder()
    monkeypatch.setattr(ticket_mod, "planfile_command", rec)
    monkeypatch.setattr(bridge_mod, "shell_drive_client_id", lambda ide: "claude-code")
    monkeypatch.delenv("KORU_SHELL_DRIVE_AUTODONE", raising=False)
    return rec


def _hp(_msg: str) -> None:
    pass


def _finalize(tmp_path: Path, **overrides):
    kwargs = dict(
        project=tmp_path,
        autopilot_ide="claude",
        ticket_id="STARTER-1",
        reply={"ok": True, "message": "refactor applied", "client_id": "claude-code"},
        ok=True,
        decision_kind="escalation_prompt",
        _hp=_hp,
    )
    kwargs.update(overrides)
    return finalize_shell_drive_ticket(**kwargs)


def test_skips_non_shell_client(tmp_path, planfile, monkeypatch):
    monkeypatch.setattr(bridge_mod, "shell_drive_client_id", lambda ide: None)
    assert _finalize(tmp_path) == "skipped"
    assert planfile.calls == []


def test_skips_failed_drive_and_wrong_kind(tmp_path, planfile):
    assert _finalize(tmp_path, ok=False) == "skipped"
    assert _finalize(tmp_path, decision_kind="idle_no_ticket") == "skipped"
    assert planfile.calls == []


def test_verified_without_config_only_notes(tmp_path, planfile, monkeypatch):
    monkeypatch.setattr(verify_mod, "load_post_run_verify_config", lambda p: None)
    assert _finalize(tmp_path) == "noted"
    assert planfile.commands() == ["ticket show", "ticket update"]


def test_verified_green_marks_done(tmp_path, planfile, monkeypatch):
    config = verify_mod.PostRunVerifyConfig(enabled=True, commands=("true",))
    monkeypatch.setattr(verify_mod, "load_post_run_verify_config", lambda p: config)
    monkeypatch.setattr(
        verify_mod,
        "verify_completed_tickets",
        lambda project, ids, **kw: [{"ticket_id": i, "ok": True, "action": "verified"} for i in ids],
    )
    assert _finalize(tmp_path) == "done_verified"
    assert planfile.commands() == ["ticket show", "ticket update", "ticket done"]


def test_verified_red_reports_reopen(tmp_path, planfile, monkeypatch):
    config = verify_mod.PostRunVerifyConfig(enabled=True, commands=("false",))
    monkeypatch.setattr(verify_mod, "load_post_run_verify_config", lambda p: config)
    monkeypatch.setattr(
        verify_mod,
        "verify_completed_tickets",
        lambda project, ids, **kw: [{"ticket_id": i, "ok": False, "action": "reopened"} for i in ids],
    )
    assert _finalize(tmp_path) == "verify_failed:reopened"


def test_always_policy_trusts_agent(tmp_path, planfile, monkeypatch):
    monkeypatch.setenv("KORU_SHELL_DRIVE_AUTODONE", "always")
    assert _finalize(tmp_path) == "done"
    assert planfile.commands() == ["ticket show", "ticket update", "ticket done"]


def test_off_policy_only_notes(tmp_path, planfile, monkeypatch):
    monkeypatch.setenv("KORU_SHELL_DRIVE_AUTODONE", "off")
    assert _finalize(tmp_path) == "noted"
    assert planfile.commands() == ["ticket show", "ticket update"]


class _PlanfileWithShowStatus:
    """Records calls like _PlanfileRecorder, but answers `ticket show` with a
    fixed status payload -- used to simulate a ticket already resolved by a
    concurrent actor before this drive's finalize step runs."""

    def __init__(self, show_status: str) -> None:
        self.calls: list[list[str]] = []
        self.show_status = show_status

    def __call__(self, project: Path, args, runner=None):
        self.calls.append(list(args))
        if args[:2] == ["ticket", "show"]:
            import json

            return subprocess.CompletedProcess(
                list(args), 0, stdout=json.dumps({"status": self.show_status}), stderr=""
            )
        return subprocess.CompletedProcess(list(args), 0, stdout="", stderr="")

    def commands(self) -> list[str]:
        return [" ".join(c[:2]) for c in self.calls]


def test_skips_done_and_verify_when_already_done_by_another_actor(tmp_path, monkeypatch):
    rec = _PlanfileWithShowStatus(show_status="done")
    monkeypatch.setattr(ticket_mod, "planfile_command", rec)
    monkeypatch.setattr(bridge_mod, "shell_drive_client_id", lambda ide: "claude-code")
    monkeypatch.delenv("KORU_SHELL_DRIVE_AUTODONE", raising=False)

    assert _finalize(tmp_path) == "already_resolved"
    # Only the status lookup happened -- no note, no done, no verify.
    assert rec.commands() == ["ticket show"]


def test_proceeds_normally_when_ticket_still_open(tmp_path, monkeypatch):
    rec = _PlanfileWithShowStatus(show_status="open")
    monkeypatch.setattr(ticket_mod, "planfile_command", rec)
    monkeypatch.setattr(bridge_mod, "shell_drive_client_id", lambda ide: "claude-code")
    monkeypatch.setenv("KORU_SHELL_DRIVE_AUTODONE", "off")

    assert _finalize(tmp_path) == "noted"
    assert rec.commands() == ["ticket show", "ticket update"]


def test_done_failure_reported(tmp_path, monkeypatch):
    rec = _PlanfileRecorder(done_rc=1)
    monkeypatch.setattr(ticket_mod, "planfile_command", rec)
    monkeypatch.setattr(bridge_mod, "shell_drive_client_id", lambda ide: "claude-code")
    monkeypatch.setenv("KORU_SHELL_DRIVE_AUTODONE", "always")
    assert _finalize(tmp_path) == "done_failed"


def test_bytes_reply_message_survives(tmp_path, planfile, monkeypatch):
    monkeypatch.setenv("KORU_SHELL_DRIVE_AUTODONE", "off")
    result = _finalize(tmp_path, reply={"ok": True, "message": b"\xffraw", "client_id": "codex"})
    assert result == "noted"
    note = planfile.calls[-1][-1]
    assert "raw" in note


class TestProviderSwitchNote:
    """Autonomous provider fallback must be visible in the ticket note."""

    def test_no_note_without_fallback(self):
        from koru.autonomy.shell_drive_finalize import provider_switch_note

        assert provider_switch_note({"provider": "z.ai"}) is None
        assert (
            provider_switch_note(
                {"provider": "z.ai", "provider_attempts": ["z.ai", "minimax"]}
            )
            is None
        )
        assert provider_switch_note({}) is None

    def test_fallback_produces_switch_note(self):
        from koru.autonomy.shell_drive_finalize import provider_switch_note

        note = provider_switch_note(
            {
                "provider": "minimax",
                "provider_attempts": ["subscription", "z.ai", "minimax"],
            }
        )
        assert note is not None
        assert "subscription → z.ai → minimax" in note
        assert "unavailable/exhausted" in note
        assert "tillm provider order" in note

    def test_switch_note_lands_in_ticket_update(self, tmp_path, planfile, monkeypatch):
        monkeypatch.setattr(verify_mod, "load_post_run_verify_config", lambda p: None)
        reply = {
            "ok": True,
            "message": "done",
            "client_id": "claude-code",
            "provider": "minimax",
            "provider_attempts": ["z.ai", "minimax"],
        }
        assert _finalize(tmp_path, reply=reply) == "noted"
        update = next(c for c in planfile.calls if c[:2] == ["ticket", "update"])
        note = update[update.index("--note") + 1]
        assert "provider-switch: z.ai → minimax" in note
        assert "provider=minimax" in note

    def test_switch_reported_live_to_operator(self, tmp_path, planfile, monkeypatch):
        monkeypatch.setattr(verify_mod, "load_post_run_verify_config", lambda p: None)
        lines: list[str] = []
        reply = {
            "ok": True,
            "message": "done",
            "client_id": "claude-code",
            "provider": "minimax",
            "provider_attempts": ["z.ai", "minimax"],
        }
        _finalize(tmp_path, reply=reply, _hp=lines.append)
        assert any("provider-switch" in line for line in lines)


class TestProviderExhaustionNote:
    """A drive that failed on every provider must leave a ticket trace."""

    @pytest.fixture(autouse=True)
    def _fresh_dedupe(self):
        from koru.autonomy import shell_drive_finalize as mod

        mod._EXHAUSTION_NOTED.clear()
        yield
        mod._EXHAUSTION_NOTED.clear()

    def _note(self, tmp_path, reply, hp=None):
        from koru.autonomy.shell_drive_finalize import note_provider_exhaustion

        return note_provider_exhaustion(
            project=tmp_path,
            ticket_id="STARTER-1",
            reply=reply,
            _hp=hp or _hp,
        )

    def test_skipped_without_attempts_or_markers(self, tmp_path, planfile):
        assert self._note(tmp_path, {"ok": False, "stderr": "429 limit"}) == "skipped"
        assert (
            self._note(
                tmp_path,
                {"ok": False, "provider_attempts": ["z.ai"], "stderr": "file not found"},
            )
            == "skipped"
        )
        assert planfile.calls == []

    def test_skipped_without_real_ticket(self, tmp_path, planfile):
        from koru.autonomy.shell_drive_finalize import note_provider_exhaustion

        reply = {
            "ok": False,
            "provider_attempts": ["z.ai"],
            "stderr": "429 limit exceeded",
        }
        assert (
            note_provider_exhaustion(
                project=tmp_path,
                ticket_id="-",
                reply=reply,
                _hp=_hp,
            )
            == "skipped"
        )
        assert planfile.calls == []

    def test_exhaustion_writes_note_once(self, tmp_path, planfile):
        reply = {
            "ok": False,
            "provider_attempts": ["subscription", "z.ai", "openrouter"],
            "stderr": "429 rate limit exceeded",
        }
        assert self._note(tmp_path, reply) == "noted_exhaustion"
        assert self._note(tmp_path, reply) == "already_noted"
        updates = [c for c in planfile.calls if c[:2] == ["ticket", "update"]]
        assert len(updates) == 1
        note = updates[0][updates[0].index("--note") + 1]
        assert "provider-exhausted: tried subscription → z.ai → openrouter" in note
        assert "tillm provider order" in note

    def test_exhaustion_reported_live(self, tmp_path, planfile):
        lines: list[str] = []
        reply = {
            "ok": False,
            "provider_attempts": ["z.ai"],
            "message": "402 requires more credits",
        }
        assert self._note(tmp_path, reply, hp=lines.append) == "noted_exhaustion"
        assert any("provider-exhausted" in line for line in lines)