"""Exercise tier planning, dispatch refusal, verified handoff, and old ledgers."""

import copy
import hashlib
import json
import os
import shutil
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from foreman.herdr import HerdrClient
from foreman.planner import plan
from foreman.state import STATE_SCHEMA_VERSION, add_assignment, empty_state, load_state_checked, role_counts
from tests.fakes import FakeRunner, ScriptedReads, agent_json, composer_reads, composer_screen, ok_json
from tests.test_cli import CliCase, CONFIG
from tests.tier_fixture import AT, tier_row


class TierIntegrationTest(CliCase):
    def setUp(self):
        super().setUp()
        self.settings = copy.deepcopy(CONFIG)
        self.settings["schema_version"] = 2
        self.settings["agents"] = self.settings["agents"][:1]
        self.settings["agents"][0]["tiers"] = {"build": tier_row()}
        self.write_config()

    def write_config(self):
        self.config.write_text(json.dumps(self.settings), encoding="utf-8")

    def apply_args(self, document=None):
        return ["apply", *self.base(), "--assignments", json.dumps(document or {"developer": "claude"}),
                "--common", str(self.common), *self.brief_args("developer"), "--now", AT, "--composer-settle", "0"]

    def test_planner_ranks_tiered_seats_by_headroom(self):
        other = copy.deepcopy(self.settings["agents"][0])
        other["name"] = "spare"
        self.settings["agents"].append(other)
        self.write_config()
        self.snapshot.write_text(json.dumps({"agents": {"claude": {"headroom_pct": 50}, "spare": {"headroom_pct": 99}}}))
        rc, output, error = self.run_cli(["plan", *self.base(), "--roles", "developer",
                                        "--snapshot", str(self.snapshot), "--now", AT])
        self.assertEqual(rc, 0, error)
        self.assertEqual(json.loads(output)["assignments"], {"developer": "spare"})

    def test_a_scarce_plan_records_its_pressure_and_apply_agrees_with_it(self):
        # coding-policy#477: `apply` measures nothing, so it re-reads the
        # headroom the PLAN resolved against. Without that, a de-escalated plan
        # would recompute un-de-escalated at dispatch and refuse itself with
        # "Plan tiers differ from current config or fix context".
        self.snapshot.write_text(json.dumps({"agents": {"claude": {"headroom_pct": 8}}}))
        context = self.tmp / "round-context.json"
        context.write_text(json.dumps({"developer": {"risk_flags": ["network", "persistence"]}}))
        rc, output, error = self.run_cli(["plan", *self.base(), "--roles", "developer",
                                          "--snapshot", str(self.snapshot),
                                          "--round-context", str(context), "--now", AT])
        self.assertEqual(rc, 0, error)
        document = json.loads(output)
        tier = document["tiers"]["developer"]
        self.assertTrue(tier["de_escalated"])
        self.assertEqual(tier["pressure_headroom"], 8)
        self.assertEqual((tier["tier_row"], tier["effort"]), ("build", "high"))

        rc, _output, error = self.run_cli(self.apply_args(document) + ["--dry-run"],
                                          client=HerdrClient("herdr", FakeRunner()))
        self.assertEqual(rc, 0, error)
        self.assertNotIn("Plan tiers differ", error)

    def fresh_dispatch_runner(self):
        """A fake Herdr through which a cleared, tier-verified claude dispatch succeeds."""
        runner = FakeRunner()
        info = json.loads(agent_json("claude", "idle", "w1:p2"))["result"]["agent"]
        info["terminal_id"] = "term-2"
        runner.set("agent get claude", json.dumps({"result": {"agent": info}}))
        process = {"pane_id": "w1:p2", "shell_pid": 100,
                   "foreground_processes": [{"name": "claude", "pid": 200, "argv": ["claude"]}]}
        shell = {**process, "foreground_processes": [{"name": "bash", "pid": 100}]}
        argv = ["claude", "--dangerously-skip-permissions", "--model", "sonnet-5", "--effort", "high"]
        started = {**process, "foreground_processes": [{"name": "claude", "pid": 300, "argv": argv}]}
        runner.responses["pane process-info"] = ScriptedReads([
            json.dumps({"result": {"process_info": item}}) for item in (process, process, shell, started)])
        runner.set("-TERM 200")
        runner.set("agent start", json.dumps({"result": {"agent": info, "argv": argv}}))
        runner.responses["agent read"] = composer_reads("claude", ("ready", "ready", "> New assignment from the team lead."))
        runner.set("agent prompt", ok_json())
        runner.set("agent wait", ok_json())
        runner.set("pane rename", ok_json())
        return runner

    def reset_streams(self):
        self.out.seek(0)
        self.out.truncate()
        self.err.seek(0)
        self.err.truncate()

    def test_plan_pins_the_oracle_apply_binds_it_and_verify_refuses_edits(self):
        # coding-policy#488: the licence is the bytes the oracle file held when
        # the plan was written, not whatever the path holds at the gate.
        # coding-policy#585: and it is the pin the round was DISPATCHED with,
        # not whatever the plan file says at the gate.
        task = "owner/repo#585"
        self.settings["agents"][0]["tiers"]["mechanical"] = tier_row()
        self.write_config()
        record = self.tmp / "task.json"
        record.write_text(json.dumps({"task": task, "base_revision": "a" * 40, "scope": "Apply the exact patch",
                                      "allowed_paths": ["src/*"], "authorization": {
                                          "source": "fixture operator message", "quote": "Approve this task."}}))
        rc, _output, error = self.run_cli(["task", *self.base(), "--record", str(record), "--now", AT])
        self.assertEqual(rc, 0, error)
        expected = self.tmp / "exact.patch"
        expected.write_bytes(b"+licensed\n")
        context = self.tmp / "round-context.json"
        context.write_text(json.dumps({"developer": {"oracle": {"kind": "patch", "path": str(expected)}}}))
        self.reset_streams()
        rc, output, error = self.run_cli(["plan", *self.base(), "--roles", "developer",
                                          "--snapshot", str(self.snapshot), "--round", "developer=mechanical",
                                          "--round-context", str(context), "--task", task, "--now", AT])
        self.assertEqual(rc, 0, error)
        document = json.loads(output)
        pin = hashlib.sha256(b"+licensed\n").hexdigest()
        self.assertEqual(document["oracle_pins"], {"developer": {"path": str(expected), "sha256": pin}})
        plan_file = self.tmp / "plan.json"
        plan_file.write_text(output)
        apply = self.apply_args(document) + ["--task", task, "--dispatch-id", "oracle-round"]
        self.reset_streams()
        rc, _output, error = self.run_cli(apply, client=HerdrClient("herdr", self.fresh_dispatch_runner()))
        self.assertEqual(rc, 0, error)
        stored, usable = load_state_checked(self.state)
        self.assertTrue(usable)
        self.assertEqual(stored["recovery"]["dispatches"][-1]["oracle"],
                         {"kind": "patch", "path": str(expected), "sha256": pin})

        result = self.tmp / "result.diff"
        result.write_bytes(b"+licensed\n")
        verify = ["verify-oracle", *self.base(), "--plan", str(plan_file), "--role", "developer",
                  "--result", str(result), "--task", task]
        self.reset_streams()
        rc, output, error = self.run_cli(verify)
        self.assertEqual(rc, 0, error)
        self.assertTrue(json.loads(output)["match"])

        # The file alone rewritten: its bytes no longer hash to the pin.
        expected.write_bytes(b"+rewritten\n")
        result.write_bytes(b"+rewritten\n")
        self.reset_streams()
        rc, output, error = self.run_cli(verify)
        self.assertEqual((rc, output), (1, ""))
        self.assertIn("changed since the plan was written", error)

        # The plan's pin rewritten to match: the plan alone checks clean, the
        # dispatch still holds the pin the round was sent with.
        edited = json.loads(plan_file.read_text())
        edited["oracle_pins"]["developer"]["sha256"] = hashlib.sha256(b"+rewritten\n").hexdigest()
        plan_file.write_text(json.dumps(edited))
        self.reset_streams()
        rc, output, error = self.run_cli(verify)
        self.assertEqual((rc, output), (1, ""))
        self.assertIn("edited after dispatch", error)

        # The pin is part of the dispatch identity: the edited plan is new
        # inputs under the recorded id, never a replay of it.
        self.reset_streams()
        rc, _output, error = self.run_cli(self.apply_args(edited) + ["--task", task, "--dispatch-id", "oracle-round"],
                                          client=HerdrClient("herdr", FakeRunner()))
        self.assertEqual(rc, 1)
        self.assertIn("already names different inputs", error)

    def test_a_retired_qualification_field_is_refused_with_the_allowed_fields(self):
        self.settings["agents"][0]["tiers"]["build"]["qualification"] = []
        self.write_config()
        rc, _, error = self.run_cli(["plan", *self.base(), "--roles", "developer",
                                    "--snapshot", str(self.snapshot), "--now", AT])
        self.assertEqual(rc, 1)
        self.assertIn("must contain only", error)

    def test_invalid_utf8_json_inputs_report_the_path_without_worker_calls(self):
        invalid = self.tmp / "invalid.json"
        invalid.write_bytes(b"\xff\xfe")
        cases = [
            ["plan", *self.base(), "--roles", "developer", "--snapshot", str(self.snapshot), "--round-context", str(invalid), "--now", AT],
            ["plan", *self.base(), "--roles", "developer", "--snapshot", str(invalid), "--now", AT],
            ["apply", *self.base(), "--assignments", str(invalid), "--common", str(self.common)],
            ["plan", *self.base(), "--config", str(invalid), "--roles", "developer", "--snapshot", str(self.snapshot), "--now", AT],
        ]
        for args in cases:
            with self.subTest(args=args):
                self.out.seek(0)
                self.out.truncate()
                self.err.seek(0)
                self.err.truncate()
                runner = FakeRunner()
                rc, output, error = self.run_cli(args, client=HerdrClient("herdr", runner))
                self.assertEqual(rc, 1)
                self.assertEqual(output, "")
                self.assertIn(str(invalid), error)
                self.assertIn("UTF-8 JSON", error)
                self.assertNotIn("Traceback", error)
                self.assertEqual(runner.calls, [])
                self.assertEqual(invalid.read_bytes(), b"\xff\xfe")
        warnings = []
        state, usable = load_state_checked(invalid, warn=warnings.append)
        self.assertFalse(usable)
        self.assertEqual(state, empty_state())
        self.assertIn("UTF-8 JSON", warnings[0])
        self.assertEqual(invalid.read_bytes(), b"\xff\xfe")

    def test_default_plan_selects_a_configured_tier_row(self):
        rc, output, error = self.run_cli(["plan", *self.base(), "--roles", "developer",
                                        "--snapshot", str(self.snapshot), "--now", AT])
        self.assertEqual(rc, 0, error)
        self.assertEqual(json.loads(output)["tiers"]["developer"]["model"], "sonnet-5")

    def test_dry_run_prints_flags_without_herdr(self):
        runner = FakeRunner()
        rc, output, error = self.run_cli(self.apply_args() + ["--dry-run"], client=HerdrClient("herdr", runner))
        self.assertEqual(rc, 0, error)
        step = json.loads(output)["steps"][0]
        launch = next(command["argv"] for command in step["commands"] if command["argv"][1:3] == ["agent", "start"])
        self.assertEqual(launch[-6:], ["--", "--dangerously-skip-permissions", "--model", "sonnet-5", "--effort", "high"])
        self.assertEqual(len(step["prompt_hash"]), 64)
        self.assertEqual(runner.calls, [])
        self.assertFalse(self.state.exists())

    def test_edited_plan_tier_is_rejected_before_dispatch(self):
        runner = FakeRunner()
        document = {"assignments": {"developer": "claude"}, "tiers": {"developer": {"model": "haiku-4.5"}}}
        rc, _, error = self.run_cli(self.apply_args(document), client=HerdrClient("herdr", runner))
        self.assertEqual(rc, 1)
        self.assertIn("Plan tiers differ", error)
        self.assertEqual(runner.calls, [])

    def test_unreadable_dispatch_input_names_recovery_and_sends_nothing(self):
        original_open = os.open
        for unreadable in (self.common, self.briefs["developer"]):
            with self.subTest(path=unreadable):
                self.out.seek(0)
                self.out.truncate()
                self.err.seek(0)
                self.err.truncate()
                runner = FakeRunner().set("agent get claude", agent_json("claude", "idle", "w1:p2"))

                # The freeze opens the brief by name under its directory's
                # descriptor; a frozen copy's name carries a digest instead.
                def opener(name, *args, unreadable=unreadable, **kwargs):
                    if str(name) in (unreadable.name, str(unreadable)):
                        raise PermissionError(13, "Permission denied", str(name))
                    return original_open(name, *args, **kwargs)

                with patch("foreman.assign.os.open", side_effect=opener):
                    rc, output, error = self.run_cli(self.apply_args(), client=HerdrClient("herdr", runner))
                self.assertEqual(rc, 1)
                self.assertEqual(output, "")
                self.assertIn(str(unreadable), error)
                self.assertIn("Restore readability", error)
                self.assertNotIn("Traceback", error)
                self.assertEqual(runner.writes(), [])
                self.assertFalse(any(command.startswith(("agent start", "-TERM")) for command in runner.commands()))
                self.assertFalse(self.state.exists())

    def test_fresh_dispatch_records_verified_tier_and_readable_state(self):
        runner = FakeRunner()
        info = json.loads(agent_json("claude", "idle", "w1:p2"))["result"]["agent"]
        info["terminal_id"] = "term-2"
        runner.set("agent get claude", json.dumps({"result": {"agent": info}}))
        process = {"pane_id": "w1:p2", "shell_pid": 100,
                   "foreground_processes": [{"name": "claude", "pid": 200, "argv": ["claude"]}]}
        shell = {**process, "foreground_processes": [{"name": "bash", "pid": 100}]}
        argv = ["claude", "--dangerously-skip-permissions", "--model", "sonnet-5", "--effort", "high"]
        started = {**process, "foreground_processes": [{"name": "claude", "pid": 300, "argv": argv}]}
        runner.responses["pane process-info"] = ScriptedReads([
            json.dumps({"result": {"process_info": item}}) for item in (process, process, shell, started)])
        runner.set("-TERM 200")
        runner.set("agent start", json.dumps({"result": {"agent": info, "argv": argv}}))
        runner.responses["agent read"] = composer_reads("claude", ("ready", "ready", "> New assignment from the team lead."))
        runner.set("agent prompt", ok_json())
        runner.set("agent wait", ok_json())
        runner.set("pane rename", ok_json())
        rc, output, error = self.run_cli(self.apply_args(), client=HerdrClient("herdr", runner))
        self.assertEqual(rc, 0, error)
        applied = json.loads(output)["applied"][0]
        self.assertTrue(applied["cleared"])
        self.assertEqual(applied["tier"]["verified"]["argv"], argv)
        self.assertIn(applied["tier"]["prompt_hash"], runner.pasted_prompts()[0])
        stored, usable = load_state_checked(self.state)
        self.assertTrue(usable)
        self.assertEqual(stored["schema_version"], STATE_SCHEMA_VERSION)
        self.assertEqual(stored["assignments"][0]["tier"], applied["tier"])
        # The capability verdict explains a plan; the recorded tier keeps its schema (#520).
        self.assertFalse({"capability", "cheaper_adequate"} & set(stored["assignments"][0]["tier"]))
        self.assertEqual(role_counts(stored), {"developer": {"claude": 1}})

    def test_tiered_worker_changed_during_composer_read_receives_no_prompt(self):
        runner = FakeRunner().set("agent get claude", agent_json("claude", "idle", "w1:p2"))
        process = {"pane_id": "w1:p2", "foreground_processes": [{"name": "claude", "pid": 200,
            "argv": ["claude", "--dangerously-skip-permissions", "--model", "sonnet-5", "--effort", "high"]}]}
        runner.set("pane process-info", json.dumps({"result": {"process_info": process}}))
        client = HerdrClient("herdr", runner)

        def read_replaced_worker(*_args, **_kwargs):
            process["foreground_processes"] = [{"name": "claude", "pid": 999, "argv": ["claude"]}]
            runner.set("pane process-info", json.dumps({"result": {"process_info": process}}))
            return composer_screen("claude")

        with patch.object(client, "agent_read", side_effect=read_replaced_worker):
            rc, output, error = self.run_cli(self.apply_args() + ["--no-clear"], client=client)
        self.assertEqual(rc, 1)
        self.assertEqual(output, "")
        self.assertIn("launch options", error)
        self.assertEqual(runner.writes(), [])

    def test_schema_three_migration_preserves_task_session_and_fix_counter(self):
        session = {"pane_id": "w1:p2", "source": "herdr:claude", "agent": "claude", "kind": "id", "value": "s1"}
        row = {"schema_version": 3, "at": AT, "role": "developer", "agent": "claude", "status": "applied",
               "task": "owner/repo#322", "fix_round": 2, "cleared": False, "clear_reason": "retained", "context_session": session}
        self.state.write_text(json.dumps({"schema_version": 3, "snapshots": [{"schema_version": 2,
            "agents": {"claude": {"window_group": "shared"}}}], "assignments": [row]}))
        migrated, usable = load_state_checked(self.state)
        self.assertTrue(usable)
        self.assertEqual(migrated["assignments"][0], {**row, "schema_version": STATE_SCHEMA_VERSION,
                         "tier": None, "requirements": None, "reviewer_scope": None, "judge_mode": None})
        self.assertEqual(migrated["snapshots"][0]["agents"]["claude"], {"window_group": "shared", "tier_billing": {}})
        self.assertEqual(role_counts(migrated), {"developer": {"claude": 1}})

    def test_apply_reads_pressure_from_the_named_snapshot_not_the_plan(self):
        # coding-policy#477 review: a plan's own `pressure_headroom` is not
        # evidence. Apply re-reads the snapshot the plan names, so a plan whose
        # claimed scarcity that snapshot no longer shows is refused before any
        # worker call, rather than dispatching its downgrade.
        self.snapshot.write_text(json.dumps({"agents": {"claude": {"headroom_pct": 8}}}))
        context = self.tmp / "round-context.json"
        context.write_text(json.dumps({"developer": {"risk_flags": ["network", "persistence"]}}))
        rc, output, error = self.run_cli(["plan", *self.base(), "--roles", "developer",
                                          "--snapshot", str(self.snapshot),
                                          "--round-context", str(context), "--now", AT])
        self.assertEqual(rc, 0, error)
        document = json.loads(output)
        self.assertTrue(document["tiers"]["developer"]["de_escalated"])
        self.snapshot.write_text(json.dumps({"agents": {"claude": {"headroom_pct": 90}}}))
        runner = FakeRunner()
        rc, output, error = self.run_cli(self.apply_args(document) + ["--dry-run"],
                                         client=HerdrClient("herdr", runner))
        self.assertEqual(rc, 1)
        self.assertRegex(error, "eligible tier|Plan tiers differ")
        self.assertEqual(runner.calls, [])

    def test_a_numeric_string_headroom_resolves_as_the_planner_ranks_it(self):
        self.snapshot.write_text(json.dumps({"agents": {"claude": {"headroom_pct": "8"}}}))
        context = self.tmp / "round-context.json"
        context.write_text(json.dumps({"developer": {"risk_flags": ["network", "persistence"]}}))
        rc, output, error = self.run_cli(["plan", *self.base(), "--roles", "developer",
                                          "--snapshot", str(self.snapshot),
                                          "--round-context", str(context), "--now", AT])
        self.assertEqual(rc, 0, error)
        tier = json.loads(output)["tiers"]["developer"]
        self.assertTrue(tier["de_escalated"])
        self.assertEqual(tier["pressure_headroom"], 8.0)

    def test_a_non_finite_stored_pressure_is_refused(self):
        state = empty_state()
        argv = ["claude", "--model", "sonnet-5", "--effort", "high"]
        add_assignment(state, AT, "developer", "claude", tier={
            "kind": "claude", "model": "sonnet-5", "effort": "high", "launch_args": [],
            "verified": {"source": "launch_argv", "model": "sonnet-5", "effort": "high", "pane_id": "w1:p2", "argv": argv}})
        state["assignments"][0]["tier"]["pressure_headroom"] = float("nan")
        self.state.write_text(json.dumps(state))
        _stored, usable = load_state_checked(self.state)
        self.assertFalse(usable)

    def test_an_overflowing_stored_pressure_is_refused_not_raised(self):
        # coding-policy#490: `math.isfinite(10**1000)` raises OverflowError,
        # which escaped the unusable-state path as a traceback.
        state = empty_state()
        argv = ["claude", "--model", "sonnet-5", "--effort", "high"]
        add_assignment(state, AT, "developer", "claude", tier={
            "kind": "claude", "model": "sonnet-5", "effort": "high", "launch_args": [],
            "verified": {"source": "launch_argv", "model": "sonnet-5", "effort": "high", "pane_id": "w1:p2", "argv": argv}})
        state["assignments"][0]["tier"]["pressure_headroom"] = 10 ** 1000
        self.state.write_text(json.dumps(state))
        _stored, usable = load_state_checked(self.state, warn=lambda _: None)
        self.assertFalse(usable)

    def test_a_relative_snapshot_path_survives_a_directory_change(self):
        # coding-policy#490: apply re-reads the snapshot the plan names. A
        # relative name resolved against apply's own directory reads as
        # unmeasured, and a legitimate de-escalation is refused as stale.
        self.snapshot.write_text(json.dumps({"agents": {"claude": {"headroom_pct": 8}}}))
        context = self.tmp / "round-context.json"
        context.write_text(json.dumps({"developer": {"risk_flags": ["network", "persistence"]}}))
        elsewhere = Path(tempfile.mkdtemp(prefix="foreman-elsewhere-"))
        self.addCleanup(shutil.rmtree, elsewhere)
        self.addCleanup(os.chdir, os.getcwd())
        os.chdir(self.tmp)
        rc, output, error = self.run_cli(["plan", *self.base(), "--roles", "developer",
                                          "--snapshot", self.snapshot.name,
                                          "--round-context", str(context), "--now", AT])
        self.assertEqual(rc, 0, error)
        document = json.loads(output)
        self.assertTrue(document["tiers"]["developer"]["de_escalated"])
        self.assertTrue(Path(document["snapshot_ref"]["source"]).is_absolute())
        os.chdir(elsewhere)
        rc, _output, error = self.run_cli(self.apply_args(document) + ["--dry-run"],
                                          client=HerdrClient("herdr", FakeRunner()))
        self.assertEqual(rc, 0, error)

    def test_the_pinned_judge_entry_has_the_shape_of_every_other_tier(self):
        # coding-policy#490: the judge branch returned before the projection
        # that adds the pressure fields, so a plan's `tiers.judge` lacked them.
        judge = copy.deepcopy(self.settings["agents"][0])
        judge["name"] = "judge"
        del judge["tiers"]
        self.settings["agents"].append(judge)
        self.settings["judge"] = {"agent": "judge", "model": "opus-5", "effort": "high"}
        self.write_config()
        self.snapshot.write_text(json.dumps({"agents": {"claude": {"headroom_pct": 50},
                                                        "judge": {"headroom_pct": 50}}}))
        rc, output, error = self.run_cli(["plan", *self.base(), "--roles", "developer,judge",
                                          "--judge-mode", "adjudication",
                                          "--snapshot", str(self.snapshot), "--now", AT])
        self.assertEqual(rc, 0, error)
        tiers = json.loads(output)["tiers"]
        self.assertEqual((tiers["judge"]["pressure_headroom"], tiers["judge"]["de_escalated"]), (None, False))
        self.assertEqual(set(tiers["judge"]), set(tiers["developer"]))

    def test_a_schema_six_tier_row_migrates_as_never_de_escalated(self):
        # coding-policy#477: nothing could de-escalate before this version, so
        # an older tier row says so explicitly instead of lacking the fields.
        state = empty_state()
        argv = ["claude", "--model", "sonnet-5", "--effort", "high"]
        add_assignment(state, AT, "developer", "claude", tier={
            "kind": "claude", "model": "sonnet-5", "effort": "high", "launch_args": [],
            "verified": {"source": "launch_argv", "model": "sonnet-5", "effort": "high", "pane_id": "w1:p2", "argv": argv}})
        state["schema_version"] = 6
        row = state["assignments"][0]
        row["schema_version"] = 6
        del row["judge_mode"]  # a pre-9 row never carried it
        for key in ("pressure_headroom", "de_escalated"):
            del row["tier"][key]
        self.state.write_text(json.dumps(state))
        migrated, usable = load_state_checked(self.state)
        self.assertTrue(usable)
        tier = migrated["assignments"][0]["tier"]
        self.assertEqual((tier["pressure_headroom"], tier["de_escalated"]), (None, False))
        self.assertEqual(migrated["assignments"][0]["schema_version"], STATE_SCHEMA_VERSION)

    def test_a_schema_seven_row_loses_the_retired_qualification_summary(self):
        state = empty_state()
        argv = ["claude", "--model", "sonnet-5", "--effort", "high"]
        add_assignment(state, AT, "developer", "claude", tier={
            "kind": "claude", "model": "sonnet-5", "effort": "high", "launch_args": [],
            "verified": {"source": "launch_argv", "model": "sonnet-5", "effort": "high", "pane_id": "w1:p2", "argv": argv}})
        state["schema_version"] = 7
        row = state["assignments"][0]
        row["schema_version"] = 7
        del row["judge_mode"]  # a pre-9 row never carried it
        row["tier"]["qualification"] = {"role": "developer", "promotion_cases": 20}
        self.state.write_text(json.dumps(state))
        migrated, usable = load_state_checked(self.state)
        self.assertTrue(usable)
        self.assertNotIn("qualification", migrated["assignments"][0]["tier"])
        self.assertEqual(migrated["assignments"][0]["schema_version"], STATE_SCHEMA_VERSION)

    def test_historical_permission_modes_stay_readable_without_becoming_live_proof(self):
        for options in ([], ["--permission-mode", "acceptEdits"]):
            with self.subTest(options=options):
                state = empty_state()
                argv = ["claude"] + options + ["--model", "sonnet-5", "--effort", "high"]
                add_assignment(state, AT, "developer", "claude", tier={
                    "kind": "claude", "model": "sonnet-5", "effort": "high", "launch_args": options,
                    "verified": {"source": "launch_argv", "model": "sonnet-5", "effort": "high", "pane_id": "w1:p2", "argv": argv}})
                original = json.dumps(state)
                self.state.write_text(original)
                stored, usable = load_state_checked(self.state)
                self.assertTrue(usable)
                self.assertEqual(stored, state)
                self.assertEqual(self.state.read_text(), original)

    def test_retained_fix_preserves_verified_higher_effort_without_restart(self):
        self.settings["agents"][0]["tiers"]["fix"] = {"model": "sonnet-5", "effort": "medium", "multiplier": 1}
        self.write_config()
        session = {"pane_id": "w1:p2", "source": "herdr:claude", "agent": "claude", "kind": "id", "value": "s1"}
        argv = ["claude", "--dangerously-skip-permissions", "--model", "sonnet-5", "--effort", "high"]
        tier = {"kind": "claude", "model": "sonnet-5", "effort": "high", "effective_multiplier": 2,
                "launch_args": ["--dangerously-skip-permissions"],
                "verified": {"model": "sonnet-5", "effort": "high", "source": "launch_argv", "pane_id": "w1:p2", "argv": argv}}
        self.state.write_text(json.dumps({"schema_version": 4, "snapshots": [], "assignments": [{
            "schema_version": 4, "at": AT, "role": "developer", "agent": "claude", "status": "applied",
            "task": "owner/repo#324", "fix_round": None, "cleared": True, "clear_reason": "automatic",
            "context_session": session, "tier": tier}]}))
        runner = FakeRunner().set("agent get claude", agent_json("claude", "idle", "w1:p2", session_id="s1"))
        runner.set("pane process-info", json.dumps({"result": {"process_info": {"pane_id": "w1:p2",
            "foreground_processes": [{"name": "claude", "pid": 200, "argv": argv}]}}}))
        runner.responses["agent read"] = composer_reads("claude", ("ready", "> New assignment from the team lead."))
        runner.set("agent prompt", ok_json()).set("agent wait", ok_json()).set("pane rename", ok_json())
        rc, output, error = self.run_cli(self.apply_args() + ["--retain-context", "--task", "owner/repo#324", "--fix-round", "1"],
                                        client=HerdrClient("herdr", runner))
        self.assertEqual(rc, 0, error)
        record = json.loads(output)["applied"][0]
        self.assertFalse(record["cleared"])
        self.assertEqual(record["tier"]["effort"], "high")
        self.assertEqual(record["tier"]["effective_multiplier"], 2)
        self.assertEqual(record["tier"]["verified"]["source"], "process_argv")
        self.assertFalse(any(command.startswith(("agent start", "-TERM")) for command in runner.commands()))

    def test_measure_records_unknown_tier_billing_even_when_worker_is_skipped(self):
        runner = FakeRunner().set("agent get claude", agent_json("claude", "blocked", "w1:p2"))
        rc, output, error = self.run_cli(["measure", *self.base(), "--now", AT], client=HerdrClient("herdr", runner))
        self.assertEqual(rc, 0, error)
        result = json.loads(output)
        self.assertEqual(result["schema_version"], 3)
        self.assertEqual(result["agents"]["claude"]["tier_billing"]["build"]["window"], "unknown")
        self.assertTrue(result["agents"]["claude"]["skipped"])

    def test_invalid_saved_tier_never_discards_the_ledger(self):
        for effort, argv, source in (("high", ["claude", "--model", "sonnet-5"], "launch_argv"),
                                    ({}, [], "launch_argv"), ("high", [], [])):
            with self.subTest(effort=effort, source=source):
                state = empty_state()
                add_assignment(state, AT, "developer", "claude", tier={"kind": "claude", "model": "sonnet-5", "effort": effort,
                    "verified": {"source": source, "model": "sonnet-5", "effort": effort, "pane_id": "w1:p2", "argv": argv}})
                original = json.dumps(state)
                self.state.write_text(original)
                result, usable = load_state_checked(self.state, warn=lambda _: None)
                self.assertFalse(usable)
                self.assertEqual(result, empty_state())
                self.assertEqual(self.state.read_text(), original)


class TierCostTest(unittest.TestCase):
    def test_candidate_multiplier_changes_winner(self):
        snapshot = {"agents": {"costly": {"headroom_pct": 95}, "cheap": {"headroom_pct": 90}}}
        tiers = {"developer": {"costly": {"effective_multiplier": 2}, "cheap": {"effective_multiplier": 1}}}
        selected = plan(["developer"], snapshot, tier_candidates=tiers)
        self.assertEqual(selected["assignments"], {"developer": "cheap"})
        self.assertEqual(selected["tiers"]["developer"]["effective_multiplier"], 1)

    def test_multiplier_burns_the_shared_pool_before_next_seat(self):
        snapshot = {"agents": {"a": {"headroom_pct": 90, "window_group": "shared"},
                               "b": {"headroom_pct": 90, "window_group": "shared"},
                               "c": {"headroom_pct": 80}}}
        tiers = {role: {name: {"effective_multiplier": 2 if role == "developer" else 1}
                       for name in "abc"} for role in ("developer", "reviewer")}
        selected = plan(["developer", "reviewer"], snapshot, tier_candidates=tiers, exclude={"developer": ["b", "c"]})
        self.assertEqual(selected["assignments"], {"developer": "a", "reviewer": "c"})


if __name__ == "__main__":
    unittest.main()