"""Native-source fixtures distinguish authored formatting from UI decoration."""

import copy
import io
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from foreman import cli, recovery, report_delivery as delivery, state
from foreman.assign import assignment_text
from foreman.errors import UsageError
from foreman.herdr import HerdrClient
from tests.fakes import FakeRunner, pane_layout
from tests import test_recovery_cli as owner_fixture

AT = "2026-09-01T12:00:00+00:00"
SESSION = "native-session-361"
PANE = "w1:p1"


def identity(kind):
    return {"source": "herdr:" + kind, "agent": kind, "kind": "id", "value": SESSION}


def codex_rows(final):
    return [
        {"type": "session_meta", "payload": {"id": SESSION}},
        {"type": "event_msg", "payload": {"type": "task_started", "turn_id": "turn-1"}},
        {"type": "response_item", "payload": {"type": "message", "role": "assistant", "phase": "final_answer",
                                              "content": [{"type": "output_text", "text": final}]}},
        {"type": "event_msg", "payload": {"type": "task_complete", "turn_id": "turn-1", "last_agent_message": final}},
    ]


def grok_row(update, meta=None):
    return {"method": "session/update", "params": {"sessionId": SESSION, "update": update, "_meta": meta or {}}}


def grok_rows(final):
    return [
        grok_row({"sessionUpdate": "user_message_chunk", "content": {"type": "text", "text": "Write the fresh report"}}),
        grok_row({"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": final}},
                 {"promptId": "prompt-1", "streamStartMs": 42}),
        grok_row({"sessionUpdate": "turn_completed", "prompt_id": "prompt-1", "stop_reason": "end_turn"}),
    ]


def encode(rows):
    return "\n".join(json.dumps(row) for row in rows) + "\n"


class MarkerColumnsTests(unittest.TestCase):
    REPORT = "/Users/me/.local/state/fleet/r/architect.md"

    def test_each_kind_adds_its_prefix_and_right_reserve(self):
        text = len("REPORT: " + self.REPORT)
        self.assertEqual(delivery.marker_columns("grok", self.REPORT), 5 + text + 15)
        self.assertEqual(delivery.marker_columns("codex", self.REPORT), 2 + text + 2)
        self.assertEqual(delivery.marker_columns("claude", self.REPORT), 2 + text + 2)

    def test_unknown_kind_takes_the_widest_prefix_and_reserve(self):
        self.assertEqual(delivery.marker_columns("gemini", self.REPORT),
                         delivery.marker_columns("grok", self.REPORT))

    def test_wide_characters_count_two_columns(self):
        self.assertEqual(delivery.marker_columns("codex", "/r/\u5831\u544a.md"),
                         delivery.marker_columns("codex", "/r/abcd.md"))


class MarkerFitTests(unittest.TestCase):
    """`foreman marker-fit`: the live-width gate for a sender outside apply (#515)."""

    REPORT = "/Users/me/.local/state/fleet/standup/2026-09-27/worker-with-a-long-name.md"

    def runner(self, kind, width, pane=PANE, status: "str | None" = "idle"):
        runner = FakeRunner()
        record = {"agent": kind, "pane_id": pane, "name": "worker"}
        if status is not None:
            record["agent_status"] = status
        runner.set("agent get worker", json.dumps({"result": {"agent": record}}))
        runner.set("pane layout --pane " + pane, pane_layout(pane, width))
        return runner

    def run_cli(self, runner, report=None, homes=None):
        output, errors = io.StringIO(), io.StringIO()
        # Empty default homes unless a test supplies its own, so the host's own
        # foreman home never decides the outcome.
        with tempfile.TemporaryDirectory() as scratch:
            state, config = homes or (scratch + "/state", scratch + "/config")
            with patch.dict(os.environ, {"XDG_STATE_HOME": state, "XDG_CONFIG_HOME": config}):
                code = cli.main(["marker-fit", "--agent", "worker", "--report", report or self.REPORT],
                                stdout=output, stderr=errors, client=HerdrClient(runner=runner))
        return code, output.getvalue(), errors.getvalue()

    def test_a_home_awaiting_migration_neither_refuses_nor_gains_a_guard(self):
        with tempfile.TemporaryDirectory() as scratch:
            state, config = Path(scratch, "state"), Path(scratch, "config")
            (state / "teamlead").mkdir(parents=True)
            (config / "teamlead").mkdir(parents=True)
            code, out, errors = self.run_cli(self.runner("codex", 200), homes=(str(state), str(config)))
            self.assertEqual(code, 0, errors)
            self.assertTrue(json.loads(out)["fits"])
            self.assertEqual(sorted(p.name for p in state.iterdir()), ["teamlead"])

    def test_a_pane_one_column_short_does_not_fit(self):
        needed = delivery.marker_columns("grok", self.REPORT)
        code, out, _ = self.run_cli(self.runner("grok", needed - 1))
        self.assertEqual(code, 0)
        self.assertEqual(json.loads(out), {"agent": "worker", "pane_id": PANE, "kind": "grok", "agent_status": "idle",
                                           "report": self.REPORT,
                                           "pane_width": needed - 1, "needed": needed, "fits": False})

    def test_a_pane_exactly_as_wide_as_needed_fits(self):
        needed = delivery.marker_columns("codex", self.REPORT)
        code, out, _ = self.run_cli(self.runner("codex", needed))
        self.assertEqual(code, 0)
        self.assertTrue(json.loads(out)["fits"])

    def test_the_kind_herdr_reports_sets_the_reserve(self):
        width = delivery.marker_columns("codex", self.REPORT)
        _, claude, _ = self.run_cli(self.runner("claude", width))
        _, grok, _ = self.run_cli(self.runner("grok", width))
        self.assertTrue(json.loads(claude)["fits"])
        self.assertFalse(json.loads(grok)["fits"])

    def test_an_unknown_kind_is_measured_at_the_widest(self):
        needed = delivery.marker_columns("grok", self.REPORT)
        _, out, _ = self.run_cli(self.runner("gemini", needed - 1))
        self.assertEqual(json.loads(out)["needed"], needed)
        self.assertFalse(json.loads(out)["fits"])

    def test_the_measured_pane_is_the_one_herdr_names(self):
        runner = self.runner("codex", 200, pane="w9:p3")
        self.run_cli(runner)
        self.assertIn(["pane", "layout", "--pane", "w9:p3"], [call[1:] for call in runner.calls])

    def test_a_relative_report_is_refused_before_herdr(self):
        runner = self.runner("codex", 200)
        code, out, errors = self.run_cli(runner, report="reports/w.md")
        self.assertEqual((code, out), (1, ""))
        self.assertIn("absolute", errors)
        self.assertEqual(runner.calls, [])

    def test_the_fresh_status_rides_along_and_an_absent_one_is_unknown(self):
        _, working, _ = self.run_cli(self.runner("codex", 200, status="working"))
        _, absent, _ = self.run_cli(self.runner("codex", 200, status=None))
        self.assertEqual(json.loads(working)["agent_status"], "working")
        self.assertEqual(json.loads(absent)["agent_status"], "unknown")

    def test_every_control_character_is_refused_before_herdr(self):
        for char in ("\t", "\x7f", "\x85", "\x9b", "\u2028", "\u2029", "\u202e", "\ue000"):
            with self.subTest(char=hex(ord(char))):
                runner = self.runner("codex", 200)
                code, out, errors = self.run_cli(runner, report="/r/a" + char + "b.md")
                self.assertEqual((code, out), (1, ""))
                self.assertIn("one-row", errors)
                self.assertEqual(runner.calls, [])

    def test_probe_report_refuses_the_same_characters_before_herdr(self):
        # One rule for every report-path entry point (#578): probe-report
        # formerly refused only C0 controls.
        for char in ("\x7f", "\x85", "\u2028", "\u2029", "\u202e"):
            with self.subTest(char=hex(ord(char))):
                runner = self.runner("codex", 200)
                output, errors = io.StringIO(), io.StringIO()
                with patch.object(sys, "stdin", io.StringIO("visible")):
                    code = cli.main(["probe-report", "--agent", "worker", "--pane", PANE,
                                     "--report", "/r/a" + char + "b.md", "--lines", "40"],
                                    stdout=output, stderr=errors, client=HerdrClient(runner=runner))
                self.assertEqual((code, output.getvalue()), (1, ""))
                self.assertIn("one-row report path", errors.getvalue())
                self.assertEqual(runner.calls, [])

    def test_an_agent_record_without_a_pane_is_a_herdr_failure(self):
        runner = FakeRunner()
        runner.set("agent get worker", json.dumps({"result": {"agent": {"agent": "codex", "agent_status": "idle"}}}))
        code, out, errors = self.run_cli(runner)
        self.assertEqual((code, out), (1, ""))
        self.assertIn("pane_id", errors)


class NativeDeliveryTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.tmp = Path(self.temp.name)
        self.report = self.tmp / "report [361]+.md"
        self.report.write_text("Current report bytes.\n")
        self.marker = "REPORT: " + str(self.report)
        self.source = self.tmp / "native.jsonl"
        self.pane = {"pane_id": PANE, "agent_status": "idle", "agent_session": identity("codex"),
                     "terminal_id": "terminal-1", "revision": 1, "scroll": {"offset_from_bottom": 0}}
        self.visible = "• " + self.marker

    def write_source(self, rows):
        self.source.write_text(encode(rows))

    def test_completed_native_sources_preserve_authored_context(self):
        bad = ["- " + self.marker, "• " + self.marker, "> " + self.marker, "    " + self.marker,
               "     " + self.marker, "`" + self.marker + "`", "```\n" + self.marker,
               "~~~\n" + self.marker, "````\n```\n" + self.marker,
               "```\n```not-a-close\n" + self.marker, self.marker + ".old",
               self.marker.replace("report ", "report\n"), self.marker + "\nMore content",
               "> quoted example\n" + self.marker, "- authored example\n" + self.marker,
               "-\tauthored example\n" + self.marker, "1.\tauthored example\n" + self.marker]
        for kind, factory in (("codex", codex_rows), ("grok", grok_rows)):
            with self.subTest(kind=kind):
                self.assertTrue(delivery.bare_final(delivery.source_final(encode(factory(self.marker)), kind, SESSION), str(self.report)))
            for text in bad:
                with self.subTest(kind=kind, text=text):
                    self.assertFalse(delivery.bare_final(delivery.source_final(encode(factory(text)), kind, SESSION), str(self.report)))
        self.assertTrue(delivery.bare_final("```\nold example\n```\n" + self.marker, str(self.report)))
        self.assertTrue(delivery.bare_final("- earlier list\n\n" + self.marker, str(self.report)))
        for fenced in ("- removed", "> quoted example", "1. numbered example"):
            self.assertTrue(delivery.bare_final("```\n" + fenced + "\n```\n" + self.marker, str(self.report)))

    def test_latest_completion_and_identity_are_required(self):
        rows = codex_rows(self.marker)
        for candidate in (rows[:-1], rows + [{"type": "event_msg", "payload": {"type": "task_started", "turn_id": "new"}}],
                          rows + [{"type": "event_msg", "payload": {"type": "error"}}]):
            self.assertIsNone(delivery.source_final(encode(candidate), "codex", SESSION))
        self.assertIsNone(delivery.source_final(encode(rows), "codex", "different-session"))
        for payload in ({"id": "another-session"}, []):
            self.assertIsNone(delivery.source_final(encode(rows[:1] + [{"type": "session_meta", "payload": payload}] + rows[1:]), "codex", SESSION))
        rows[-1]["payload"]["turn_id"] = "other-turn"
        self.assertIsNone(delivery.source_final(encode(rows), "codex", SESSION))
        for body in ("not json", '{"type":"session_meta","payload":[]}\n', '{}\n'):
            self.assertIsNone(delivery.source_final(body, "codex", SESSION))
        rows = grok_rows(self.marker)
        for candidate in (rows[:-1], rows + [grok_row({"sessionUpdate": "user_message_chunk"})],
                          rows + [grok_row({"sessionUpdate": "error"})]):
            self.assertIsNone(delivery.source_final(encode(candidate), "grok", SESSION))
        for field, value in (("stop_reason", "refusal"), ("prompt_id", "different-prompt")):
            altered = copy.deepcopy(rows)
            altered[-1]["params"]["update"][field] = value
            self.assertIsNone(delivery.source_final(encode(altered), "grok", SESSION))
        self.assertIsNone(delivery.source_final(encode(rows), "grok", "other-session"))

    def test_grok_native_chunks_join_only_within_the_final_stream(self):
        rows = grok_rows("REPORT: ")
        rows.insert(-1, grok_row({"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": str(self.report)}},
                               {"promptId": "prompt-1", "streamStartMs": 42}))
        self.assertEqual(delivery.source_final(encode(rows), "grok", SESSION), self.marker)
        rows[-2]["params"]["_meta"]["streamStartMs"] = 43
        self.assertEqual(delivery.source_final(encode(rows), "grok", SESSION), str(self.report))

    def test_native_display_allowlist_never_joins_rows(self):
        self.assertEqual(delivery.decorated_row(self.visible, "codex", str(self.report)), self.visible)
        for suffix in ("", "          11:47 PM"):
            row = "     " + self.marker + suffix
            self.assertEqual(delivery.decorated_row(row, "grok", str(self.report)), row)
        for kind in ("codex", "grok", "unknown"):
            for row in ("- " + self.marker, "> " + self.marker, "    " + self.marker,
                        "      " + self.marker, "• " + self.marker + " extra", "     " + self.marker + " .old",
                        "     " + self.marker + "  32:75 PM", "• REPORT: \n" + str(self.report)):
                self.assertIsNone(delivery.decorated_row(row, kind, str(self.report)))

    def fake_client(self, after=None, visible=None):
        test = self

        class Client:
            def __init__(self):
                self.reads = 0

            def agent_get(self, agent):
                return dict(test.pane)

            def pane_get(self, pane_id):
                self.reads += 1
                return copy.deepcopy(after if self.reads > 1 and after is not None else test.pane)

            def pane_read(self, pane_id, lines):
                return (test.visible if visible is None else visible) + "\n"

        return Client()

    def test_probe_requires_stable_native_pane_source_and_file(self):
        self.write_source(codex_rows(self.marker))
        with patch.object(delivery, "source_path", return_value=self.source):
            self.assertTrue(delivery.probe(self.fake_client(), "worker", PANE, str(self.report), self.visible, 40)["found"])
            for field, value in (("terminal_id", "replacement"), ("revision", 2), ("agent_status", "working"),
                                 ("scroll", {"offset_from_bottom": 1}), ("agent_session", identity("grok"))):
                after = {**self.pane, field: value}
                self.assertFalse(delivery.probe(self.fake_client(after), "worker", PANE, str(self.report), self.visible, 40)["found"])
            self.assertFalse(delivery.probe(self.fake_client(visible="new output"), "worker", PANE, str(self.report), self.visible, 40)["found"])
            self.report.unlink()
            self.assertFalse(delivery.probe(self.fake_client(), "worker", PANE, str(self.report), self.visible, 40)["found"])

    def test_source_resolution_never_selects_the_newest_other_session(self):
        root = self.tmp / "sessions" / "2026" / "09" / "01"
        root.mkdir(parents=True)
        matching = root / ("rollout-fixed-" + SESSION + ".jsonl")
        matching.write_text(encode(codex_rows(self.marker)))
        (root / "rollout-newer-other-session.jsonl").write_text("{}\n")
        with patch.dict(os.environ, {"CODEX_HOME": str(self.tmp)}):
            self.assertEqual(delivery.source_path(identity("codex")), matching)
            self.assertIsNone(delivery.source_path({**identity("codex"), "value": "missing"}))
            (root / ("rollout-duplicate-" + SESSION + ".jsonl")).write_text("{}\n")
            self.assertIsNone(delivery.source_path(identity("codex")))

    def test_probe_rechecks_source_when_new_turn_has_not_repainted(self):
        self.write_source(codex_rows(self.marker))
        client = self.fake_client()

        def append_turn(pane_id, lines):
            self.source.write_text(encode(codex_rows(self.marker) + [
                {"type": "event_msg", "payload": {"type": "task_started", "turn_id": "new-turn"}}]))
            return self.visible

        with patch.object(delivery, "source_path", return_value=self.source), patch.object(client, "pane_read", side_effect=append_turn):
            self.assertFalse(delivery.probe(client, "worker", PANE, str(self.report), self.visible, 40)["found"])

    def test_public_probe_reports_first_read_and_reread_failures(self):
        source_bytes = encode(codex_rows(self.marker)).encode("utf-8")
        errors = (PermissionError(13, "Permission denied", str(self.source)),
                  OSError(5, "Input/output error", str(self.source)),
                  UnicodeDecodeError("utf-8", b"\xff", 0, 1, "invalid start byte"))
        args = ["probe-report", "--agent", "worker", "--pane", PANE, "--report", str(self.report), "--lines", "40"]
        for read_number in (1, 2):
            for error in errors:
                outcomes = [source_bytes] * (read_number - 1) + [error]
                output, diagnostic = io.StringIO(), io.StringIO()
                with self.subTest(read_number=read_number, error=type(error).__name__), \
                        patch.object(delivery, "source_path", return_value=self.source), \
                        patch.object(Path, "read_bytes", side_effect=outcomes), patch.object(sys, "stdin", io.StringIO(self.visible)):
                    code = cli.main(args, stdout=output, stderr=diagnostic, client=self.fake_client())
                    self.assertEqual(code, 1)
                    self.assertEqual(output.getvalue(), "")
                    message = json.loads(diagnostic.getvalue())["message"]
                    self.assertIn(str(self.source), message)
                    self.assertIn(str(error), message)
                    self.assertIn("Restore readable UTF-8 transcript bytes and file permissions", message)

    def test_absent_native_source_stays_unconfirmed_on_either_read(self):
        source_bytes = encode(codex_rows(self.marker)).encode("utf-8")
        for outcomes in ([FileNotFoundError()], [source_bytes, FileNotFoundError()]):
            with patch.object(delivery, "source_path", return_value=self.source), patch.object(Path, "read_bytes", side_effect=outcomes):
                result = delivery.probe(self.fake_client(), "worker", PANE, str(self.report), self.visible, 40)
                self.assertFalse(result["found"])
                self.assertEqual(result["reason"], "native_source_unavailable")
        with patch.object(delivery, "source_path", side_effect=FileNotFoundError()):
            self.assertFalse(delivery.probe(self.fake_client(), "worker", PANE, str(self.report), self.visible, 40)["found"])

    def test_native_directory_permission_error_is_not_a_missing_source(self):
        root = self.tmp / "sessions"
        error = PermissionError(13, "Permission denied", str(root))
        with patch.object(delivery, "source_root", return_value=root), patch("os.scandir", side_effect=error):
            with self.assertRaises(UsageError) as caught:
                delivery.probe(self.fake_client(), "worker", PANE, str(self.report), self.visible, 40)
        self.assertIn(str(root), str(caught.exception))
        self.assertIn("Permission denied", str(caught.exception))
        self.assertIn("Restore readable session directories and search permissions", str(caught.exception))

    def watcher_fixture(self):
        fake = self.tmp / "herdr"
        config = self.tmp / "fake.json"
        fake.write_text("#!/usr/bin/env python3\nimport json, os, sys\nfrom pathlib import Path\n"
                        "def main():\n    d=json.loads(Path(os.environ['FAKE_CONFIG']).read_text())\n"
                        "    command=sys.argv[1:3]\n"
                        "    if command==['agent','get']: print(json.dumps({'result':{'agent':d['pane']}}))\n"
                        "    elif command==['pane','get']: print(json.dumps({'result':{'pane':d['pane']}}))\n"
                        "    elif command==['pane','read']:\n"
                        "        if d.get('break_source'):\n"
                        "            counter=Path(os.environ['FAKE_CONFIG']+'.reads')\n"
                        "            count=int(counter.read_text())+1 if counter.exists() else 1\n"
                        "            counter.write_text(str(count))\n"
                        "            if count==2:\n"
                        "                source=Path(d['break_source'])\n"
                        "                source.unlink()\n"
                        "                source.mkdir()\n"
                        "        print(d['visible'])\n"
                        "    elif command==['pane','wait-output']: print('{}')\n"
                        "    else: sys.exit(2)\n"
                        "if __name__=='__main__': main()\n")
        fake.chmod(0o755)
        env = {**os.environ, "HERDR_ENV": "1", "HERDR_BIN": str(fake), "FAKE_CONFIG": str(config),
               "HOME": str(self.tmp), "CODEX_HOME": str(self.tmp / ".codex"), "FOREMAN_WAIT_BUDGET_SEC": "0"}
        return env, config

    def test_public_watcher_propagates_unreadable_source_as_tool_failure(self):
        env, config = self.watcher_fixture()
        source = self.tmp / (".codex/sessions/2026/09/01/rollout-fixed-" + SESSION + ".jsonl")
        source.parent.mkdir(parents=True)
        for failure in ("decode", "reread"):
            source.write_bytes(b"\xff" if failure == "decode" else encode(codex_rows(self.marker)).encode("utf-8"))
            pane = {"pane": self.pane, "visible": self.visible}
            if failure == "reread":
                pane["break_source"] = str(source)
            config.write_text(json.dumps(pane))
            result = subprocess.run(["bash", str(ROOT / "wait-report.sh"), "worker", str(self.report)],
                                    env=env, capture_output=True, text=True, check=False)
            with self.subTest(failure=failure):
                self.assertEqual(result.returncode, 2, result.stderr)
                self.assertEqual(result.stdout, "")
                self.assertIn(str(source), result.stderr)
                self.assertIn("invalid start byte" if failure == "decode" else "Is a directory", result.stderr)
                self.assertIn("Restore readable UTF-8 transcript bytes and file permissions", result.stderr)

    def test_public_watcher_uses_real_native_source_fixtures(self):
        env, config = self.watcher_fixture()
        for kind, factory, relative in (("codex", codex_rows, ".codex/sessions/2026/09/01/rollout-fixed-" + SESSION + ".jsonl"),
                                        ("grok", grok_rows, ".grok/sessions/project/" + SESSION + "/updates.jsonl")):
            source = self.tmp / relative
            source.parent.mkdir(parents=True)
            source.write_text(encode(factory(self.marker)))
            prefix = "• " if kind == "codex" else "     "
            for source_text, visible, expected in ((self.marker, prefix + self.marker, True),
                    ("```\n- removed\n```\n" + self.marker, prefix + self.marker, True),
                    ("```\n> quoted example\n```\n" + self.marker, prefix + self.marker, True),
                    ("- " + self.marker, prefix + self.marker, False), ("    " + self.marker, prefix + self.marker, False),
                    ("```\n" + self.marker, prefix + self.marker, False),
                    ("> quoted example\n" + self.marker, prefix + self.marker, False),
                    ("- authored example\n" + self.marker, prefix + self.marker, False),
                    ("-\tauthored example\n" + self.marker, prefix + self.marker, False),
                    (self.marker, prefix + "REPORT: \n" + str(self.report), False),
                    (self.marker, prefix + self.marker + ".old", False)):
                source.write_text(encode(factory(source_text)))
                config.write_text(json.dumps({"pane": {**self.pane, "agent_session": identity(kind)}, "visible": visible}))
                result = subprocess.run(["bash", str(ROOT / "wait-report.sh"), "worker", str(self.report)],
                                        env=env, capture_output=True, text=True, check=False)
                with self.subTest(kind=kind, source=source_text, visible=visible):
                    self.assertEqual(result.returncode, 0 if expected else 1, result.stderr)
                    self.assertEqual(json.loads(result.stdout)["found"], expected)

    def recovery_fixture(self):
        document = state.empty_state()
        state.add_assignment(document, AT, "judge", "worker", task="task-361", judge_mode="adjudication", context_session={"pane_id": PANE, **identity("codex")})
        brief = self.tmp / "brief.md"
        brief.write_text("Judge the original dispute.\n" + self.marker + "\n")
        common = self.tmp / "common.md"
        common.write_text("Shared round requirements.\n")
        # A version-1 dispatch result: it predates both the composition
        # metadata and the judge mode that recovery store 12 added (#478).
        assignment = {key: value for key, value in document["assignments"][0].items()
                      if key not in {"requirements", "reviewer_scope", "judge_mode"}}
        dispatch = {"schema_version": 1, "id": "dispatch-361", "at": AT, "fingerprint": "fingerprint-361",
                    "task": "task-361", "role": "judge", "agent": "worker", "fix_round": None, "status": "applied",
                    "assignment_index": 0, "brief": str(brief), "common": str(common),
                    "result": {**assignment, "schema_version": 1, "pane_id": PANE}, "report": None}
        document["recovery"]["dispatches"].append(dispatch)
        data = {"id": "recovery-361", "dispatch": dispatch["id"], "report": str(self.report)}
        source = codex_rows(self.marker)
        source.insert(2, {"type": "response_item", "payload": {"type": "message", "role": "user", "content": [
            {"type": "input_text", "text": assignment_text("judge", str(common), str(brief))}]}})
        artifacts = {"wait_receipt": json.dumps({"agent": "worker", "report_path": str(self.report), "state": "done",
                                                "found": False, "reason": "report file present, worker done on 2 consecutive reads, marker unconfirmed"}),
                     "pane": json.dumps({"result": {"pane": self.pane}}), "visible": self.visible,
                     "source": encode(source)}
        for key, body in artifacts.items():
            path = self.tmp / (key + ".txt")
            path.write_text(body)
            data[key] = str(path)
        return document, data

    def test_owner_cli_appends_recovery_preserving_original_records(self):
        document, data = self.recovery_fixture()
        document["assignments"][0]["context_session"] = None
        document["recovery"]["dispatches"][0]["result"]["context_session"] = None
        prior = copy.deepcopy(document)
        evidence_before = {key: Path(data[key]).read_bytes() for key in ("report", "wait_receipt", "source", "pane", "visible")}
        ledger_path, record = self.tmp / "state.json", self.tmp / "record.json"
        state.save_state(ledger_path, document)
        record.write_text(json.dumps(data))
        runner = FakeRunner()
        output, errors = io.StringIO(), io.StringIO()
        args = ["recover-report", "--state", str(ledger_path), "--record", str(record), "--now", AT]
        result = cli.main(args, stdout=output, stderr=errors, client=HerdrClient(runner=runner))
        self.assertEqual(result, 0, errors.getvalue())
        saved = json.loads(ledger_path.read_text())
        self.assertEqual(saved["assignments"], prior["assignments"])
        self.assertEqual(saved["recovery"]["dispatches"], prior["recovery"]["dispatches"])
        self.assertEqual(saved["recovery"]["delivery_recoveries"][0]["grants_review_approval"], False)
        self.assertEqual(runner.calls, [])
        for key, body in evidence_before.items():
            self.assertEqual(Path(data[key]).read_bytes(), body)
        before_replay = ledger_path.read_bytes()
        self.assertEqual(cli.main(args, stdout=io.StringIO(), stderr=io.StringIO()), 0)
        self.assertEqual(ledger_path.read_bytes(), before_replay)
        self.report.write_text("Changed after original recovery\n")
        self.assertEqual(cli.main(args, stdout=io.StringIO(), stderr=io.StringIO()), 1)
        self.assertEqual(ledger_path.read_bytes(), before_replay)

    def test_the_prompt_hash_is_checked_against_the_receipted_brief(self):
        # coding-policy#565: the brief's receipt and the prompt-hash check read
        # one set of bytes. A brief swapped back to the dispatched bytes
        # between the two reads would otherwise record a receipt for bytes the
        # hash never checked.
        from foreman.assign import tiered_prompt
        document, data = self.recovery_fixture()
        dispatch = document["recovery"]["dispatches"][0]
        tier = {"model": "fixture-model", "effort": "high"}
        prompt, prompt_hash = tiered_prompt(assignment_text("judge", dispatch["common"], dispatch["brief"]),
                                            tier, dispatch["common"], dispatch["brief"])
        dispatch["result"]["tier"] = {**tier, "prompt_hash": prompt_hash}
        rows = [json.loads(line) for line in Path(data["source"]).read_text().splitlines()]
        rows[2]["payload"]["content"][0]["text"] = prompt
        Path(data["source"]).write_text(encode(rows))
        delivery.recover(copy.deepcopy(document["recovery"]), copy.deepcopy(document["assignments"]), data, AT)
        brief = Path(dispatch["brief"])
        dispatched = brief.read_bytes()
        brief.write_bytes(b"Reworded after the dispatch.\n" + self.marker.encode("utf-8") + b"\n")
        real_receipt = recovery.receipt

        def receipt_then_restore(path):
            result = real_receipt(path)
            if path == dispatch["brief"]:
                brief.write_bytes(dispatched)
            return result
        with patch.object(recovery, "receipt", side_effect=receipt_then_restore):
            with self.assertRaisesRegex(UsageError, "differ from the dispatch's recorded prompt hash"):
                delivery.recover(document["recovery"], document["assignments"], data, AT)
        self.assertEqual(document["recovery"]["delivery_recoveries"], [])

    def metadata_recovery_fixture(self):
        document, data = self.recovery_fixture()
        rows = [json.loads(line) for line in Path(data["source"]).read_text().splitlines()]
        rows[2]["payload"]["internal_chat_message_metadata_passthrough"] = {
            "turn_id": "turn-1", "content_item_kinds": ["user.text"]}
        environment = {"type": "response_item", "payload": {"type": "message", "role": "user",
            "content": [{"type": "input_text", "text": "<environment_context>runtime</environment_context>"}],
            "internal_chat_message_metadata_passthrough": {
                "turn_id": "turn-1", "content_item_kinds": ["environments.environment_context"]}}}
        rows[3:3] = [{"type": "compacted", "payload": {}},
                     {"type": "turn_context", "payload": {"turn_id": "turn-1"}}, environment]
        return document, data, rows

    def test_owner_cli_retains_verified_prompt_across_same_turn_environment(self):
        document, data, rows = self.metadata_recovery_fixture()
        Path(data["source"]).write_text(encode(rows))
        original = copy.deepcopy(document)
        artifacts = {key: Path(data[key]).read_bytes()
                     for key in ("report", "wait_receipt", "source", "pane", "visible")}
        ledger_path, record = self.tmp / "state.json", self.tmp / "record.json"
        state.save_state(ledger_path, document)
        record.write_text(json.dumps(data))
        runner = FakeRunner()
        output, errors = io.StringIO(), io.StringIO()
        result = cli.main(["recover-report", "--state", str(ledger_path), "--record", str(record), "--now", AT],
                          stdout=output, stderr=errors, client=HerdrClient(runner=runner))
        self.assertEqual(result, 0, errors.getvalue())
        saved = json.loads(ledger_path.read_text())
        receipt = json.loads(output.getvalue())
        self.assertTrue(receipt["found"])
        self.assertFalse(receipt["grants_review_approval"])
        self.assertEqual(saved["recovery"]["delivery_recoveries"], [receipt])
        self.assertEqual(len(saved["recovery"]["events"]), len(original["recovery"]["events"]) + 1)
        saved["recovery"]["delivery_recoveries"] = original["recovery"]["delivery_recoveries"]
        saved["recovery"]["events"] = original["recovery"]["events"]
        self.assertEqual(saved, original)
        self.assertEqual(runner.calls, [])
        for key, body in artifacts.items():
            self.assertEqual(Path(data[key]).read_bytes(), body)

    def test_owner_cli_requires_new_start_to_retain_context_after_termination(self):
        for terminal in ("task_complete", "turn_aborted", "error"):
            for new_start in (False, True):
                with self.subTest(terminal=terminal, new_start=new_start):
                    document, data, rows = self.metadata_recovery_fixture()
                    ended = rows[:3]
                    if terminal == "task_complete":
                        ended.append(copy.deepcopy(rows[-2]))
                    ended.append({"type": "event_msg", "payload": {"type": terminal,
                        "turn_id": "turn-1", "last_agent_message": self.marker}})
                    following = copy.deepcopy(rows[2:])
                    if new_start:
                        ended.append({"type": "event_msg", "payload": {
                            "type": "task_started", "turn_id": "turn-2"}})
                        for row in following:
                            payload = row["payload"]
                            if "turn_id" in payload:
                                payload["turn_id"] = "turn-2"
                            if "internal_chat_message_metadata_passthrough" in payload:
                                payload["internal_chat_message_metadata_passthrough"]["turn_id"] = "turn-2"
                    Path(data["source"]).write_text(encode(ended + following))
                    ledger_path, record = self.tmp / "state.json", self.tmp / "record.json"
                    state.save_state(ledger_path, document)
                    record.write_text(json.dumps(data))
                    before = ledger_path.read_bytes()
                    artifacts = {key: Path(data[key]).read_bytes()
                                 for key in ("report", "wait_receipt", "source", "pane", "visible")}
                    output, errors, runner = io.StringIO(), io.StringIO(), FakeRunner()
                    result = cli.main(["recover-report", "--state", str(ledger_path),
                                       "--record", str(record), "--now", AT], stdout=output,
                                      stderr=errors, client=HerdrClient(runner=runner))
                    self.assertEqual(result, 0 if new_start else 1, errors.getvalue())
                    self.assertEqual(runner.calls, [])
                    for key, body in artifacts.items():
                        self.assertEqual(Path(data[key]).read_bytes(), body)
                    if new_start:
                        saved, receipt = json.loads(ledger_path.read_bytes()), json.loads(output.getvalue())
                        self.assertTrue(receipt["found"])
                        self.assertFalse(receipt["grants_review_approval"])
                        self.assertEqual(saved["recovery"]["delivery_recoveries"], [receipt])
                        self.assertEqual(saved["recovery"]["events"][:-1], document["recovery"]["events"])
                        self.assertEqual(saved["recovery"]["events"][-1]["kind"], "report_delivery_recovered")
                        for key in ("delivery_recoveries", "events"):
                            saved["recovery"][key] = document["recovery"][key]
                        self.assertEqual(saved, document)
                    else:
                        self.assertEqual(output.getvalue(), "")
                        self.assertIn("do not prove", errors.getvalue())
                        self.assertEqual(ledger_path.read_bytes(), before)

    def assert_owner_recovery(self, document, data, rows, accepted):
        Path(data["source"]).write_text(encode(rows))
        ledger_path, record = self.tmp / "state.json", self.tmp / "record.json"
        state.save_state(ledger_path, document)
        record.write_text(json.dumps(data))
        before = ledger_path.read_bytes()
        dispatch = document["recovery"]["dispatches"][0]
        paths = [Path(data[key]) for key in ("report", "wait_receipt", "source", "pane", "visible")]
        paths += [record, Path(dispatch["brief"]), Path(dispatch["common"])]
        artifacts = {path: path.read_bytes() for path in paths}
        result = subprocess.run(["bash", str(ROOT / "foreman.sh"), "recover-report",
                                 "--state", str(ledger_path), "--record", str(record), "--now", AT],
                                capture_output=True, text=True, check=False)
        self.assertEqual(result.returncode, 0 if accepted else 1, result.stderr)
        self.assertEqual({path: path.read_bytes() for path in paths}, artifacts)
        if not accepted:
            self.assertEqual(result.stdout, "")
            self.assertIn("do not prove", result.stderr)
            self.assertEqual(ledger_path.read_bytes(), before)
            return
        saved, receipt = json.loads(ledger_path.read_bytes()), json.loads(result.stdout)
        self.assertTrue(receipt["found"])
        self.assertFalse(receipt["grants_review_approval"])
        self.assertEqual(receipt["dispatch"], data["dispatch"])
        self.assertEqual(saved["recovery"]["delivery_recoveries"], [receipt])
        self.assertEqual(saved["recovery"]["events"], document["recovery"]["events"] + [{
            "schema_version": 1, "sequence": len(document["recovery"]["events"]) + 1,
            "at": AT, "kind": "report_delivery_recovered", "task": dispatch["task"],
            "details": {"recovery": data["id"], "dispatch": data["dispatch"]}}])
        for key in ("delivery_recoveries", "events"):
            saved["recovery"][key] = document["recovery"][key]
        self.assertEqual(saved, document)

    def test_owner_closes_modern_turn_across_unmarked_responses(self):
        metadata = "internal_chat_message_metadata_passthrough"
        for terminal in ("task_complete", "turn_aborted", "error"):
            for variation in ("user-tail", "assistant-tail", "unmarked-before-terminal",
                              "fresh-modern", "fresh-legacy", "wholly-legacy"):
                with self.subTest(terminal=terminal, variation=variation):
                    document, data, rows = self.metadata_recovery_fixture()
                    user = copy.deepcopy(rows[2])
                    del user["payload"][metadata]
                    ended = copy.deepcopy(rows[:3])
                    if variation == "wholly-legacy":
                        del ended[2]["payload"][metadata]
                    if variation == "unmarked-before-terminal":
                        ended.append(copy.deepcopy(user))
                    if terminal == "task_complete":
                        ended.append(copy.deepcopy(rows[-2]))
                    ended.append({"type": "event_msg", "payload": {
                        "type": terminal, "turn_id": "turn-1", "last_agent_message": self.marker}})
                    tail = [user] + copy.deepcopy(rows[-2:])
                    if variation == "assistant-tail":
                        del tail[0]
                    fresh = variation in ("fresh-modern", "fresh-legacy")
                    if fresh:
                        ended.append({"type": "event_msg", "payload": {
                            "type": "task_started", "turn_id": "turn-2"}})
                        tail[-1]["payload"]["turn_id"] = "turn-2"
                        if variation == "fresh-modern":
                            tail[0]["payload"][metadata] = {
                                "turn_id": "turn-2", "content_item_kinds": ["user.text"]}
                    self.assert_owner_recovery(document, data, ended + tail,
                                               fresh or variation == "wholly-legacy")

    def test_owner_reconciles_duplicate_and_contradictory_user_events(self):
        for modern in (False, True):
            for placement in ("before-response", "before-context", "after-context", "no-context"):
                if not modern and placement in ("before-context", "after-context"):
                    continue
                for matching in (False, True):
                    if placement == "before-response" and not matching:
                        continue
                    with self.subTest(modern=modern, placement=placement, matching=matching):
                        document, data, rows = self.metadata_recovery_fixture()
                        prompt = rows[2]["payload"]["content"][0]["text"]
                        if not modern or placement == "no-context":
                            rows = rows[:3] + rows[-2:]
                        if not modern:
                            del rows[2]["payload"]["internal_chat_message_metadata_passthrough"]
                        event = {"type": "event_msg", "payload": {
                            "type": "user_message", "message": prompt if matching else "Different assignment",
                            "images": [], "audio": [], "local_images": [], "text_elements": []}}
                        index = 2 if placement == "before-response" else 3 if placement == "before-context" else len(rows) - 2
                        rows.insert(index, event)
                        self.assert_owner_recovery(document, data, rows, matching)

    def test_owner_user_event_cannot_establish_or_rehabilitate_prompt_proof(self):
        for variation in ("event-only", "invalid-metadata", "missing-start", "legacy-context", "fresh-missing"):
            with self.subTest(variation=variation):
                document, data, rows = self.metadata_recovery_fixture()
                rows.insert(-2, {"type": "event_msg", "payload": {
                    "type": "user_message", "message": rows[2]["payload"]["content"][0]["text"]}})
                if variation == "event-only":
                    del rows[2]
                elif variation == "invalid-metadata":
                    rows[2]["payload"]["internal_chat_message_metadata_passthrough"] = None
                elif variation == "missing-start":
                    del rows[1]
                elif variation == "legacy-context":
                    del rows[2]["payload"]["internal_chat_message_metadata_passthrough"]
                else:
                    rows[3:-3] = [{"type": "event_msg", "payload": {
                        "type": "task_started", "turn_id": "turn-2"}}]
                    rows[-1]["payload"]["turn_id"] = "turn-2"
                self.assert_owner_recovery(document, data, rows, False)

    def test_codex_prior_turn_metadata_cannot_poison_or_supply_current_prompt(self):
        for valid_prior in (False, True):
            for missing_current in (False, True):
                with self.subTest(valid_prior=valid_prior, missing_current=missing_current):
                    document, data, rows = self.metadata_recovery_fixture()
                    prior = copy.deepcopy(rows[1:3])
                    prior[0]["payload"]["turn_id"] = "prior-turn"
                    metadata = prior[1]["payload"]["internal_chat_message_metadata_passthrough"]
                    metadata["turn_id"] = "prior-turn"
                    if not valid_prior:
                        metadata["content_item_kinds"] = ["agents_md.instructions", "environments.environment_context"]
                    if missing_current:
                        del rows[2]
                    rows[1:1] = prior
                    self.assert_owner_recovery(document, data, rows, not missing_current)

    def test_codex_later_user_text_replaces_prompt_even_when_it_looks_like_context(self):
        for native in (False, True):
            for text in ("New assignment", "<environment_context>user instruction</environment_context>"):
                document, data, rows = self.metadata_recovery_fixture()
                later = copy.deepcopy(rows[2])
                later["payload"]["content"][0]["text"] = text
                if not native:
                    del later["payload"]["internal_chat_message_metadata_passthrough"]
                rows.insert(-2, later)
                Path(data["source"]).write_text(encode(rows))
                with self.subTest(native=native, text=text):
                    self.assertEqual(delivery.source_prompt(encode(rows), "codex", SESSION), text)
                    before = copy.deepcopy(document)
                    with self.assertRaises(UsageError):
                        delivery.recover(document["recovery"], document["assignments"], data, AT)
                    self.assertEqual(document, before)

    def test_codex_environment_requires_unambiguous_metadata_and_verified_prompt_turn(self):
        metadata_key = "internal_chat_message_metadata_passthrough"
        bad_metadata = [None, [], "metadata", {}, {"turn_id": "turn-1"}]
        bad_metadata += [{"turn_id": "turn-1", "content_item_kinds": value} for value in (
            None, [], "environments.environment_context", {}, [None], [{}],
            ["unknown"], ["environments.environment_context", "user.text"],
            ["environments.environment_context", "unknown"],
            ["environments.environment_context", "environments.environment_context"])]
        bad_metadata += [{"turn_id": value, "content_item_kinds": ["environments.environment_context"]}
                         for value in (None, "", [], {}, "other-turn")]
        bad_metadata.append({"content_item_kinds": ["environments.environment_context"]})
        for index in (2, 5):
            for metadata in bad_metadata:
                document, data, rows = self.metadata_recovery_fixture()
                rows[index]["payload"][metadata_key] = metadata
                # Even matching text cannot authenticate unknown/malformed kinds.
                rows[5]["payload"]["content"] = copy.deepcopy(rows[2]["payload"]["content"])
                Path(data["source"]).write_text(encode(rows))
                before = copy.deepcopy(document)
                with self.subTest(index=index, metadata=metadata), self.assertRaises(UsageError):
                    delivery.recover(document["recovery"], document["assignments"], data, AT)
                self.assertEqual(document, before)

    def test_codex_environment_never_relaxes_other_recovery_proofs(self):
        for variation in ("legacy-prompt", "no-prompt", "new-turn", "missing-start", "context-turn",
                          "prompt-turn", "completion-turn", "session", "later-user-before-context",
                          "bad-content", "changed-prompt", "brief-path", "report-path", "no-completion",
                          "quoted-final", "wrapped-final"):
            document, data, rows = self.metadata_recovery_fixture()
            if variation == "legacy-prompt":
                del rows[2]["payload"]["internal_chat_message_metadata_passthrough"]
            elif variation == "no-prompt":
                del rows[2]
            elif variation == "new-turn":
                rows.insert(3, {"type": "event_msg", "payload": {"type": "task_started", "turn_id": "turn-2"}})
            elif variation == "missing-start":
                del rows[1]
            elif variation == "context-turn":
                rows[4]["payload"]["turn_id"] = "other-turn"
            elif variation == "prompt-turn":
                rows[2]["payload"]["internal_chat_message_metadata_passthrough"]["turn_id"] = "other-turn"
            elif variation == "completion-turn":
                rows[-1]["payload"]["turn_id"] = "other-turn"
            elif variation == "session":
                rows[0]["payload"]["id"] = "other-session"
            elif variation == "later-user-before-context":
                later = copy.deepcopy(rows[2])
                later["payload"]["content"][0]["text"] = "Later assignment"
                rows.insert(3, later)
            elif variation == "bad-content":
                rows[5]["payload"]["content"] = [{"type": "image", "text": "runtime"}]
            elif variation == "changed-prompt":
                rows[2]["payload"]["content"][0]["text"] += " changed"
            elif variation == "brief-path":
                document["recovery"]["dispatches"][0]["brief"] = str(self.report)
            elif variation == "report-path":
                data["report"] = document["recovery"]["dispatches"][0]["brief"]
            elif variation == "no-completion":
                rows.pop()
            else:
                final = "> " + self.marker if variation == "quoted-final" else "```\n" + self.marker
                rows[-2]["payload"]["content"][0]["text"] = final
                rows[-1]["payload"]["last_agent_message"] = final
            Path(data["source"]).write_text(encode(rows))
            before = copy.deepcopy(document)
            with self.subTest(variation=variation), self.assertRaises(UsageError):
                delivery.recover(document["recovery"], document["assignments"], data, AT)
            self.assertEqual(document, before)

    def test_recovery_accepts_actual_public_apply_output_with_null_reviewer_session(self):
        case = owner_fixture.RecoveryCommandTests()
        case.setUp()
        self.addCleanup(case.doCleanups)
        case.register()
        case.briefs["reviewer"].write_text("Review the original change.\n" + self.marker + "\n")
        client = case._client({"grok": "idle"})
        code, output, errors = case.invoke(case.apply_args("reviewer"), client)
        self.assertEqual(code, 0, errors)
        applied = json.loads(output)["applied"][0]
        original = case.saved()
        self.assertIsNone(original["assignments"][-1]["context_session"])
        self.assertNotIn("schema_version", applied)
        self.assertEqual(original["recovery"]["dispatches"][-1]["result"]["schema_version"], 2)
        prompt = next(call[4] for call in case.runner.calls if call[1:3] == ["agent", "prompt"])
        rows = grok_rows(self.marker)
        rows[0]["params"]["update"]["content"]["text"] = prompt
        pane = {**self.pane, "pane_id": applied["pane_id"], "agent_session": identity("grok")}
        artifacts = {"source": encode(rows), "pane": json.dumps({"result": {"pane": pane}}),
                     "visible": "     " + self.marker,
                     "wait_receipt": json.dumps({"agent": "grok", "report_path": str(self.report), "state": "idle",
                                                 "found": False, "reason": "report file present, worker idle on 2 consecutive reads, marker unconfirmed"})}
        data = {"id": "actual-apply-recovery", "dispatch": applied["dispatch_id"], "report": str(self.report)}
        for key, value in artifacts.items():
            path = self.tmp / ("actual-" + key + ".txt")
            path.write_text(value)
            data[key] = str(path)
        before_calls = list(case.runner.calls)
        code, _output, errors = case.owner("recover-report", data, client)
        self.assertEqual(code, 0, errors)
        recovered = case.saved()
        self.assertEqual(recovered["assignments"], original["assignments"])
        self.assertEqual(recovered["recovery"]["dispatches"], original["recovery"]["dispatches"])
        self.assertEqual(case.runner.calls, before_calls)

    def test_recovery_rejects_unbound_negative_receipts_and_authored_examples(self):
        for variation in ("authored", "session", "refusal", "brief", "pending", "no-prompt", "bad-pane", "path-proof"):
            document, data = self.recovery_fixture()
            if variation == "authored":
                Path(data["source"]).write_text(encode(codex_rows("- " + self.marker)))
            elif variation == "session":
                document["assignments"][0]["context_session"]["value"] = "different-session"
            elif variation == "refusal":
                negative = json.loads(Path(data["wait_receipt"]).read_text())
                negative["reason"] = "terminal_provider_refusal"
                Path(data["wait_receipt"]).write_text(json.dumps(negative))
            elif variation == "brief":
                Path(document["recovery"]["dispatches"][0]["brief"]).write_text("REPORT: /another/report.md\n")
            elif variation == "pending":
                document["recovery"]["dispatches"][0]["status"] = "sending"
            elif variation == "bad-pane":
                Path(data["pane"]).write_text('{"result": []}')
            elif variation == "path-proof":
                document["assignments"][0]["context_session"]["kind"] = "path"
                document["assignments"][0]["context_session"]["value"] = "/original/native/session.jsonl"
            else:
                document["assignments"][0]["context_session"] = None
                Path(data["source"]).write_text(encode(codex_rows(self.marker)))
            before = copy.deepcopy(document)
            with self.subTest(variation=variation), self.assertRaises(UsageError):
                delivery.recover(document["recovery"], document["assignments"], data, AT)
            self.assertEqual(document, before)

    def test_malformed_native_identity_stays_unconfirmed(self):
        for value in ({}, [], None, 42):
            self.assertIsNone(delivery.native_identity({"agent_session": {**identity("codex"), "agent": value}}))

    def test_store_upgrade_preserves_negative_and_dispatch_history(self):
        document, _data = self.recovery_fixture()
        store = document["recovery"]
        store["schema_version"] = 2
        del store["delivery_recoveries"]
        del store["role_clearances"]
        del store["refusal_authorizations"]
        del store["diagnoses"]
        del store["legacy_ruling_recoveries"]
        del store["approaches"]
        original = copy.deepcopy(store)
        self.assertTrue(recovery.migrate_store(store))
        self.assertEqual(store, {**original, "schema_version": recovery.RECOVERY_STORE_VERSION, "role_clearances": [], "delivery_recoveries": [], "refusal_authorizations": [], "diagnoses": [], "legacy_ruling_recoveries": [], "approaches": []})
        self.assertFalse(recovery.migrate_store(store))


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