"""Marks: one primitive for operator judgment attached to an artifact.

`pick` (the session asks), `note` (the operator tells), `flag` (the operator
points at the good ones) — three shapes, one storage model, one read path.

See docs/contracts/u2_marks.contract.md.
"""
import ast
import json
import pathlib
import sys

import pytest

from booth.asks import ASK_SUFFIX, AskError, build_answer, normalize_ask
from booth.marks import (
    MARKS_FILE,
    Mark,
    answer_pick,
    as_dict,
    declare_pick,
    marks_for,
    open_marks,
)


def _single():
    return {"prompt": "Which render wins?", "options": ["A — baseline", "B — async"]}


def _multi():
    return {
        "title": "R18 batch review",
        "questions": [
            {"key": "q1", "prompt": "Render 1?", "options": ["keep", "drop"]},
            {"key": "q2", "prompt": "Render 2?", "options": ["keep", "drop"]},
        ],
    }


# ---- slice 1: storage, the record, the read path ----------------------------


def test_pick_round_trips_through_marks_json(tmp_path):
    """A declared pick lands in ONE file and reads back normalized."""
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "winner", _single())

    assert (booth / MARKS_FILE).is_file()
    marks = marks_for(booth)
    assert len(marks) == 1
    m = marks[0]
    assert isinstance(m, Mark)
    assert (m.id, m.shape, m.target) == ("winner", "pick", None)
    assert m.answer is None
    assert m.prompt == "Which render wins?"
    assert m.multi is False
    # normalize_ask emits `notes` (bool); the record carries it as notes_enabled (SR-2)
    assert m.notes_enabled is True
    assert m.notes_label == "notes"
    assert [o["id"] for o in m.options] == ["A — baseline", "B — async"]
    assert len(m.questions) == 1 and m.questions[0]["key"] is None


def test_marks_json_is_a_dotfile_so_it_is_not_an_item(tmp_path):
    """The dotfile skip in booth_items already covers it. Asserted, not assumed."""
    from booth.items import booth_items

    booth = tmp_path / "b"
    booth.mkdir()
    (booth / "a.png").write_bytes(b"\x89PNG")
    declare_pick(booth, "winner", _single())

    assert MARKS_FILE.startswith(".")
    assert [i.rel for i in booth_items(booth)] == ["a.png"]


def test_corrupt_marks_json_renders_as_empty(tmp_path):
    """A booth with no marks and a booth with a broken mark file both render as
    'no marks'. Neither is a 500."""
    booth = tmp_path / "b"
    booth.mkdir()
    (booth / MARKS_FILE).write_text("{not json at all")
    assert marks_for(booth) == []


def test_missing_booth_and_missing_file_are_empty(tmp_path):
    assert marks_for(tmp_path / "nope") == []
    (tmp_path / "b").mkdir()
    assert marks_for(tmp_path / "b") == []


def test_broken_pick_declaration_surfaces_error(tmp_path):
    """A declaration that cannot be rendered comes back with `error` set, so the
    page can SAY so rather than silently hiding a question the session believes
    it posted. Preserved from list_asks."""
    booth = tmp_path / "b"
    booth.mkdir()
    (booth / MARKS_FILE).write_text(json.dumps({
        "version": 1,
        "marks": [{"id": "bad", "shape": "pick", "target": None,
                   "created": "2026-09-21T10:00:00-07:00",
                   "declaration": {"prompt": "no options"}, "answer": None}],
    }))
    m = marks_for(booth)[0]
    assert m.error is not None
    assert m.id == "bad"


def test_declare_pick_validates_before_writing(tmp_path):
    """A session cannot land a question the renderer would refuse."""
    booth = tmp_path / "b"
    booth.mkdir()
    with pytest.raises(AskError):
        declare_pick(booth, "bad", {"prompt": "only one", "options": ["just me"]})
    assert not (booth / MARKS_FILE).exists()


def test_declare_pick_rejects_a_bad_id(tmp_path):
    booth = tmp_path / "b"
    booth.mkdir()
    with pytest.raises(AskError):
        declare_pick(booth, "../escape", _single())


def test_redeclaring_a_pick_clears_its_answer(tmp_path):
    """The question changed, so the old judgment is not an answer to it."""
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "winner", _single())
    answer_pick(booth, "winner", "A — baseline")
    assert marks_for(booth)[0].answer is not None

    declare_pick(booth, "winner", {"prompt": "Which one now?", "options": ["X", "Y"]})
    marks = marks_for(booth)
    assert len(marks) == 1
    assert marks[0].answer is None
    assert marks[0].prompt == "Which one now?"


def test_marks_are_oldest_first_by_created(tmp_path):
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "first", _single())
    declare_pick(booth, "second", _single())
    assert [m.id for m in marks_for(booth)] == ["first", "second"]


# ---- openness: THE one predicate (INV-2) ------------------------------------


def test_open_marks_counts_an_unanswered_pick(tmp_path):
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "winner", _single())
    assert [m.id for m in open_marks(marks_for(booth))] == ["winner"]


def test_open_marks_drops_a_complete_answer(tmp_path):
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "winner", _single())
    answer_pick(booth, "winner", "A — baseline")
    assert open_marks(marks_for(booth)) == []


def test_a_partial_answer_counts_as_open(tmp_path):
    """SR-7, the declared behaviour change. Today the index badge tests
    `answer is None` and so reports a half-answered four-question pick as
    closed, while the panel renders it `◐ partial`. A booth that still owes an
    answer is open — which is also what makes U4's pin rule correct."""
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "batch", _multi())
    answer_pick(booth, "batch", {"q1": "keep"})  # q2 left alone

    m = marks_for(booth)[0]
    assert m.answer is not None
    assert m.answer["complete"] is False
    assert m.answer["unanswered"] == ["q2"]
    assert [x.id for x in open_marks(marks_for(booth))] == ["batch"]


# ---- the pick semantics, preserved by NOT rewriting them (INV-4) ------------


def test_partial_answer_is_recorded_not_refused(tmp_path):
    """Operator ruling 2026-09-09, as a regression test."""
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "batch", _multi())
    m = answer_pick(booth, "batch", {"q1": "keep", "q2": ""})
    assert m.answer["answers"]["q1"]["label"] == "keep"
    assert "q2" not in m.answer["answers"]
    assert m.answer["unanswered"] == ["q2"]
    assert m.answer["complete"] is False


def test_nothing_to_record_is_still_refused(tmp_path):
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "batch", _multi())
    with pytest.raises(AskError):
        answer_pick(booth, "batch", {"q1": "", "q2": ""})


def test_an_offered_but_invalid_option_is_an_error(tmp_path):
    """A broken form, not a skipped question."""
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "winner", _single())
    with pytest.raises(AskError):
        answer_pick(booth, "winner", "C — never offered")


def test_answering_an_unknown_pick_raises(tmp_path):
    """SR-4: load_ask used to own this AskError. Extracting build_answer moved
    the path to the caller, and a stale form POST must be a 400, not a silent
    no-op."""
    booth = tmp_path / "b"
    booth.mkdir()
    with pytest.raises(AskError):
        answer_pick(booth, "ghost", "A — baseline")


def test_re_answering_overwrites(tmp_path):
    """The mark is the CURRENT judgment, not a log."""
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "winner", _single())
    answer_pick(booth, "winner", "A — baseline")
    m = answer_pick(booth, "winner", "B — async")
    assert m.answer["label"] == "B — async"
    assert len(marks_for(booth)) == 1


def test_build_answer_matches_the_old_sidecar_document(tmp_path):
    """INV-4. build_answer is write_answer's logic with the I/O removed; the
    document it produces is the one the sidecar carried."""
    ask = normalize_ask(_single(), "winner")
    doc = build_answer(ask, "B — async", notes="less banding", who="10.0.0.1")
    assert doc["stem"] == "winner"
    assert doc["choice"] == "B — async"
    assert doc["choice_index"] == 1
    assert doc["label"] == "B — async"
    assert doc["notes"] == "less banding"
    assert doc["complete"] is True
    assert doc["unanswered"] == []
    assert doc["answered_by"] == "10.0.0.1"
    assert "answered_at" in doc


def test_build_answer_sources_stem_from_the_ask(tmp_path):
    """SR-3: the extraction takes a pre-loaded ask and reads `stem` off it."""
    ask = normalize_ask(_multi(), "batch")
    doc = build_answer(ask, {"q1": "keep", "q2": "drop"})
    assert doc["stem"] == "batch"
    assert doc["complete"] is True


# ---- the JSON boundary (SR-8) -----------------------------------------------


def test_as_dict_round_trips_through_json(tmp_path):
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "winner", _single())
    m = marks_for(booth)[0]
    out = json.loads(json.dumps(as_dict(m)))
    assert out["id"] == "winner"
    assert out["shape"] == "pick"
    assert out["answer"] is None


# ---- the stdlib-only invariant (INV-5) --------------------------------------


@pytest.mark.parametrize("module", ["marks", "asks", "links", "manifest"])
def test_stdlib_only(module):
    """INV-5. scripts/booth imports these under the system python3 with NO venv,
    through a `python3 -c` heredoc that no AST extractor can see — so nothing
    but this test stands between a casual third-party import and `booth ask`
    breaking on every fleet host."""
    # `manifest` also carries a stricter copy in tests/test_manifest.py, which
    # additionally forbids importing `booth.*` — a cross-import between two
    # stdlib-only modules is a second way for this invariant to break.
    src = pathlib.Path(__file__).parent.parent / "booth" / f"{module}.py"
    tree = ast.parse(src.read_text())
    roots = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            roots.update(a.name.split(".")[0] for a in node.names)
        elif isinstance(node, ast.ImportFrom):
            # `node.level > 0` is a RELATIVE import (`from . import marks`),
            # which has no `module` root to inspect and used to slip through
            # this walk entirely. It cannot reach outside the package, so it is
            # stdlib-safe by construction — but it is recorded rather than
            # ignored, because `manifest.py` additionally forbids importing a
            # sibling and its own test needs to see one.
            if node.level:
                roots.add("booth")
            elif node.module:
                roots.add(node.module.split(".")[0])
    outside = {r for r in roots if r != "booth" and r not in sys.stdlib_module_names}
    assert not outside, f"booth/{module}.py imports non-stdlib: {sorted(outside)}"


# ---- slice 2: flag — the shape with no existing code ------------------------


def test_flag_is_an_upsert_and_unflag_removes(tmp_path):
    """One item has at most one flag state, so flagging twice is idempotent and
    clearing REMOVES the mark rather than storing `value: false` — an absent
    flag and a false flag are the same judgment."""
    from booth.marks import set_flag

    booth = tmp_path / "b"
    booth.mkdir()
    m = set_flag(booth, "DSC03389.jpg", True, who="10.0.0.1")
    assert m is not None and m.shape == "flag" and m.flagged is True
    assert m.target == "DSC03389.jpg"

    set_flag(booth, "DSC03389.jpg", True)  # again
    assert len(marks_for(booth)) == 1

    assert set_flag(booth, "DSC03389.jpg", False) is None
    assert marks_for(booth) == []


def test_unflagging_something_unflagged_is_not_an_error(tmp_path):
    from booth.marks import set_flag

    booth = tmp_path / "b"
    booth.mkdir()
    assert set_flag(booth, "nope.png", False) is None


def test_flags_on_different_items_coexist(tmp_path):
    from booth.marks import marks_for_target, set_flag

    booth = tmp_path / "b"
    booth.mkdir()
    set_flag(booth, "a.png", True)
    set_flag(booth, "b.png", True)
    marks = marks_for(booth)
    assert len(marks) == 2
    assert [m.id for m in marks_for_target(marks, "a.png")] == ["flag:a.png"]


def test_a_flag_never_counts_as_open(tmp_path):
    """Only a pick can owe an answer. A flag is born resolved."""
    from booth.marks import set_flag

    booth = tmp_path / "b"
    booth.mkdir()
    set_flag(booth, "a.png", True)
    assert open_marks(marks_for(booth)) == []


def test_a_flag_target_cannot_escape_the_booth(tmp_path):
    from booth.marks import set_flag

    booth = tmp_path / "b"
    booth.mkdir()
    for bad in ("../outside.png", "/etc/passwd", ""):
        with pytest.raises(AskError):
            set_flag(booth, bad, True)


# ---- slice 3: note ----------------------------------------------------------


def test_note_allows_several_per_target(tmp_path):
    """An item may carry several notes, so each gets a generated id."""
    from booth.marks import marks_for_target, write_note

    booth = tmp_path / "b"
    booth.mkdir()
    write_note(booth, "a.png", "banding in the gradient")
    write_note(booth, "a.png", "and the highlight clips")
    marks = marks_for(booth)
    assert len(marks) == 2
    assert {m.id for m in marks} == {"note-1", "note-2"}
    assert [m.text for m in marks_for_target(marks, "a.png")] == [
        "banding in the gradient", "and the highlight clips"]


def test_booth_level_note_has_target_none(tmp_path):
    """The booth itself is a legal target."""
    from booth.marks import marks_for_target, write_note

    booth = tmp_path / "b"
    booth.mkdir()
    write_note(booth, None, "ship v3, not v4")
    m = marks_for(booth)[0]
    assert m.target is None
    assert marks_for_target(marks_for(booth), None) == [m]


def test_an_empty_note_is_refused(tmp_path):
    """Nothing to record — the same posture as an empty pick submission."""
    from booth.marks import write_note

    booth = tmp_path / "b"
    booth.mkdir()
    for blank in ("", "   ", "\n\n", None):
        with pytest.raises(AskError):
            write_note(booth, "a.png", blank)
    assert marks_for(booth) == []


def test_a_note_never_counts_as_open(tmp_path):
    from booth.marks import write_note

    booth = tmp_path / "b"
    booth.mkdir()
    write_note(booth, "a.png", "seen it")
    assert open_marks(marks_for(booth)) == []


def test_delete_mark_is_the_operators_undo(tmp_path):
    from booth.marks import delete_mark, write_note

    booth = tmp_path / "b"
    booth.mkdir()
    write_note(booth, "a.png", "wrong call")
    assert delete_mark(booth, "note-1") is True
    assert marks_for(booth) == []
    assert delete_mark(booth, "note-1") is False


# ---- the lock (INV-6) -------------------------------------------------------


def test_concurrent_writers_do_not_lose_a_mark(tmp_path):
    """INV-6. Two processes writing the same booth's marks at once: both land.
    Processes, not threads — the flock is what is under test, and a thread-level
    test would pass on the GIL alone."""
    import subprocess
    import sys as _sys

    booth = tmp_path / "b"
    booth.mkdir()
    repo = str(pathlib.Path(__file__).parent.parent)
    prog = (
        "import sys; sys.path.insert(0, %r);"
        "from booth.marks import set_flag;"
        "set_flag(%r, sys.argv[1], True)" % (repo, str(booth))
    )
    procs = [subprocess.Popen([_sys.executable, "-c", prog, f"item{i}.png"])
             for i in range(8)]
    for p in procs:
        assert p.wait() == 0

    got = {m.target for m in marks_for(booth)}
    assert got == {f"item{i}.png" for i in range(8)}, f"lost a mark: {sorted(got)}"


# ---- slice 4: the legacy sidecars — imported, NEVER deleted ----------------


def _sidecar(booth, stem, doc, mtime=None):
    booth.mkdir(parents=True, exist_ok=True)
    p = booth / f"{stem}{ASK_SUFFIX}"
    p.write_text(json.dumps(doc))
    if mtime is not None:
        import os as _os
        _os.utime(p, (mtime, mtime))
    return p


def test_import_reads_a_sidecar_into_marks(tmp_path):
    from booth.marks import import_legacy_asks

    booth = tmp_path / "b"
    _sidecar(booth, "winner", _single())
    got = import_legacy_asks(booth)

    assert [m.id for m in got] == ["winner"]
    m = marks_for(booth)[0]
    assert m.shape == "pick"
    assert m.prompt == "Which render wins?"
    assert m.answer is None


def test_import_is_idempotent_and_keeps_the_sidecar(tmp_path):
    """INV-7. The ROADMAP's non-goal is explicit: no migration that deletes
    anything. Four of these are live and unanswered right now."""
    from booth.marks import import_legacy_asks

    booth = tmp_path / "b"
    p = _sidecar(booth, "winner", _single())
    import_legacy_asks(booth)
    assert import_legacy_asks(booth) == []          # second run is a no-op
    assert len(marks_for(booth)) == 1
    assert p.is_file(), "the importer deleted the operator's sidecar"


def test_import_carries_an_existing_answer(tmp_path):
    from booth.asks import ANSWER_SUFFIX, build_answer, normalize_ask
    from booth.marks import import_legacy_asks

    booth = tmp_path / "b"
    _sidecar(booth, "winner", _single())
    # The legacy answer sidecar, written the way the retired `write_answer` did.
    doc = build_answer(normalize_ask(_single(), "winner"), "B — async", notes="less banding")
    (booth / f"winner{ANSWER_SUFFIX}").write_text(json.dumps(doc))
    import_legacy_asks(booth)

    m = marks_for(booth)[0]
    assert m.answer["label"] == "B — async"
    assert m.answer["notes"] == "less banding"
    assert open_marks(marks_for(booth)) == []


def test_import_does_not_clobber_a_newer_answer(tmp_path):
    """A stem already present as a mark is skipped outright."""
    from booth.marks import import_legacy_asks

    booth = tmp_path / "b"
    _sidecar(booth, "winner", _single())
    declare_pick(booth, "winner", _single())
    answer_pick(booth, "winner", "A — baseline")

    assert import_legacy_asks(booth) == []
    assert marks_for(booth)[0].answer["label"] == "A — baseline"


def test_import_preserves_mtime_ordering(tmp_path):
    """SR-5. list_asks ordered by file mtime; marks_for orders by the stored
    `created`. Seed it from the sidecar's mtime or the four live asks silently
    reorder on import."""
    from booth.marks import import_legacy_asks

    booth = tmp_path / "b"
    _sidecar(booth, "later", _single(), mtime=2_000_000_000)
    _sidecar(booth, "earlier", _single(), mtime=1_000_000_000)
    import_legacy_asks(booth)
    assert [m.id for m in marks_for(booth)] == ["earlier", "later"]


def test_import_surfaces_a_broken_sidecar_instead_of_dropping_it(tmp_path):
    """A question the session believes it posted stays visible."""
    from booth.marks import import_legacy_asks

    booth = tmp_path / "b"
    _sidecar(booth, "broke", {"prompt": "no options here"})
    import_legacy_asks(booth)
    m = marks_for(booth)[0]
    assert m.id == "broke"
    assert m.error is not None


def test_import_ignores_a_bad_stem(tmp_path):
    """A filename that could not have been written by `booth ask` is left alone
    rather than becoming a mark with an unusable id."""
    from booth.marks import import_legacy_asks

    booth = tmp_path / "b"
    booth.mkdir()
    (booth / f"..evil{ASK_SUFFIX}").write_text(json.dumps(_single()))
    assert import_legacy_asks(booth) == []


def test_import_on_a_booth_with_nothing_is_empty(tmp_path):
    from booth.marks import import_legacy_asks

    assert import_legacy_asks(tmp_path / "nope") == []
    (tmp_path / "b").mkdir()
    assert import_legacy_asks(tmp_path / "b") == []


# ---- the deterministic-order invariant (operator directive, 2026-09-21) ------


def test_marks_order_is_deterministic_across_reads(tmp_path):
    """Every ordered collection the Booth renders needs a STATED rule, because
    the operator judges positionally — "the third one", "the one after the
    banded one" — and an order that moves between renders misfiles the judgment
    instead of crashing. Marks order by `(created, id)`."""
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "b-pick", _single())
    from booth.marks import set_flag, write_note
    write_note(booth, "z.png", "later note")
    set_flag(booth, "a.png", True)
    write_note(booth, None, "booth note")

    first = [m.id for m in marks_for(booth)]
    for _ in range(5):
        assert [m.id for m in marks_for(booth)] == first


def test_same_second_marks_tie_break_on_id(tmp_path):
    """The `created` stamp is second-resolution, so two marks written inside one
    second would order by whatever json listed them. The id is the tie-break."""
    booth = tmp_path / "b"
    booth.mkdir()
    stamp = "2026-09-21T12:00:00-07:00"
    (booth / MARKS_FILE).write_text(json.dumps({"version": 1, "marks": [
        {"id": "zeta", "shape": "note", "target": None, "created": stamp, "text": "z"},
        {"id": "alpha", "shape": "note", "target": None, "created": stamp, "text": "a"},
    ]}))
    assert [m.id for m in marks_for(booth)] == ["alpha", "zeta"]
    # and reversing the stored order changes nothing
    (booth / MARKS_FILE).write_text(json.dumps({"version": 1, "marks": [
        {"id": "alpha", "shape": "note", "target": None, "created": stamp, "text": "a"},
        {"id": "zeta", "shape": "note", "target": None, "created": stamp, "text": "z"},
    ]}))
    assert [m.id for m in marks_for(booth)] == ["alpha", "zeta"]


def test_open_marks_preserves_the_read_order(tmp_path):
    """The filtered view must not re-order — the panel and the badge have to
    agree with the list they came from."""
    booth = tmp_path / "b"
    booth.mkdir()
    for mid in ("c", "a", "b"):
        declare_pick(booth, mid, _single())
    marks = marks_for(booth)
    assert [m.id for m in open_marks(marks)] == [m.id for m in marks]


# ---- INV-3: the judgment travels, like the caption --------------------------


def _png(p):
    p.write_bytes(
        b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
        b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00"
        b"\x01\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82"
    )


@pytest.fixture
def client(tmp_path):
    from fastapi.testclient import TestClient

    from booth.app import create_app
    return TestClient(create_app(tmp_path, ttl_hours=24, start_sweeper=False)), tmp_path


def test_zoom_carries_the_marks(client):
    """U1's rule, extended from the caption to the judgment: full size is where
    the operator is actually deciding, so the flag state and the notes have to
    be there."""
    from booth.marks import set_flag, write_note

    c, data = client
    b = data / "b"
    b.mkdir()
    _png(b / "a.png")
    set_flag(b, "a.png", True)
    write_note(b, "a.png", "banding in the gradient")

    html = c.get("/b/b/view?f=a.png").text
    assert "banding in the gradient" in html
    assert "✔ flagged" in html
    assert 'action="/b/b/flag"' in html


def test_the_tile_carries_the_marks(client):
    from booth.marks import set_flag, write_note

    c, data = client
    b = data / "b"
    b.mkdir()
    _png(b / "a.png")
    _png(b / "b.png")
    set_flag(b, "a.png", True)
    write_note(b, "b.png", "this one is soft")

    html = c.get("/b/b/").text
    assert "✔ flagged" in html and "○ flag" in html      # a flagged one and an unflagged one
    assert "this one is soft" in html
    assert 'id="item-a.png"' in html


def test_flag_and_note_round_trip_through_the_browser(client):
    c, data = client
    b = data / "b"
    b.mkdir()
    _png(b / "a.png")

    assert c.post("/b/b/flag", data={"target": "a.png", "on": "1"},
                  follow_redirects=False).status_code == 303
    assert c.post("/b/b/note", data={"target": "a.png", "text": "soft"},
                  follow_redirects=False).status_code == 303
    doc = c.get("/b/b/marks.json").json()
    assert {m["shape"] for m in doc["marks"]} == {"flag", "note"}
    assert doc["open"] == []                              # neither owes an answer

    # and the operator can withdraw one
    note_id = next(m["id"] for m in doc["marks"] if m["shape"] == "note")
    assert c.post("/b/b/unmark", data={"mark": note_id},
                  follow_redirects=False).status_code == 303
    assert [m["shape"] for m in c.get("/b/b/marks.json").json()["marks"]] == ["flag"]


def test_a_no_op_write_does_not_touch_the_booth(tmp_path):
    """A booth's TTL is measured from its newest mtime, INCLUDING dotfiles — so
    writing `.marks.json` resets the clock. Marking IS activity and should; a
    write that changes nothing is not activity and must not. Unflagging
    something never flagged would otherwise keep a dead booth alive."""
    from booth.marks import delete_mark, set_flag

    booth = tmp_path / "b"
    booth.mkdir()
    assert set_flag(booth, "ghost.png", False) is None
    assert delete_mark(booth, "nothing") is False
    assert not (booth / MARKS_FILE).exists(), "a no-op write created the mark file"


def test_a_real_write_then_a_no_op_leaves_the_file_alone(tmp_path):
    import os
    from booth.marks import set_flag

    booth = tmp_path / "b"
    booth.mkdir()
    set_flag(booth, "a.png", True)
    path = booth / MARKS_FILE
    os.utime(path, (1_000_000_000, 1_000_000_000))
    before = path.stat().st_mtime

    set_flag(booth, "a.png", True)            # idempotent: already flagged
    assert path.stat().st_mtime == before, "an idempotent flag rewrote the file"


# ---- findings from the cross-frontier contract panel, 2026-09-22 -------------
#
# Heid panel (thread 01M33VSNFER4N1554G0Y0VC9C8). Four arms, artifact-only.


def test_a_write_over_a_corrupt_marks_file_refuses_instead_of_replacing(tmp_path):
    """DATA LOSS, shipped in v0.2.0. Found by Kimi (flag 2), converged with Hulda.

    `marks_for` is deliberately lenient — an unparseable file reads as "no marks"
    so a review page still loads. The write path inherited that leniency through
    the same reader, so the next flag toggle appended one entry to an empty list
    and atomically replaced the file: every judgment in that booth gone, from one
    click, silently.

    The read stays lenient and the WRITE goes strict. That asymmetry is the fix —
    a page that renders without an annotation is recoverable, a file that
    overwrote the operator's judgment is not, and this repo's standing rule is
    that nothing deletes his data.
    """
    from booth.marks import MarksCorrupt, set_flag, write_note

    booth = tmp_path / "b"
    booth.mkdir()
    write_note(booth, "a.png", "judgment one")
    write_note(booth, "b.png", "judgment two")
    raw = (booth / MARKS_FILE).read_text()
    (booth / MARKS_FILE).write_text(raw[: len(raw) // 2])   # truncated mid-write

    with pytest.raises(MarksCorrupt):
        set_flag(booth, "c.png", True)

    # The damaged bytes are still on disk — untouched, recoverable by hand.
    assert (booth / MARKS_FILE).read_text() == raw[: len(raw) // 2]
    # And the read path is still lenient, so the page renders rather than 500s.
    assert marks_for(booth) == []


def test_an_absent_or_empty_marks_file_is_not_corrupt(tmp_path):
    """The strict write path must not mistake "nothing yet" for "damaged"."""
    from booth.marks import set_flag

    booth = tmp_path / "b"
    booth.mkdir()
    assert set_flag(booth, "a.png", True) is not None      # no file at all
    (booth / MARKS_FILE).write_text("")
    assert set_flag(booth, "b.png", True) is not None      # zero bytes
    (booth / MARKS_FILE).write_text('{"version": 1, "marks": []}')
    assert set_flag(booth, "c.png", True) is not None      # valid but empty


def test_a_pick_can_target_one_item(tmp_path):
    """Found by Hulda (flag 1), converged with Regin.

    `Mark.target` carries an item rel, `marks_for_target` retrieves by it, and
    the panel template already renders "on <item>" for a pick — but
    `declare_pick` had no target parameter, so a session could not actually
    produce one. A question about ONE artifact is the 2026-09-09 ruling's whole
    point; the record supported it and the door was missing.
    """
    from booth.marks import marks_for_target

    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "which-crop", _single(), target="v3/DSC03389.jpg")
    m = marks_for(booth)[0]
    assert m.target == "v3/DSC03389.jpg"
    assert [x.id for x in marks_for_target(marks_for(booth), "v3/DSC03389.jpg")] == ["which-crop"]
    # and it still answers normally
    answer_pick(booth, "which-crop", "A — baseline")
    assert marks_for(booth)[0].answer["complete"] is True


def test_a_pick_target_cannot_escape_the_booth(tmp_path):
    booth = tmp_path / "b"
    booth.mkdir()
    for bad in ("../outside.png", "/etc/passwd"):
        with pytest.raises(AskError):
            declare_pick(booth, "p", _single(), target=bad)


def test_redeclaring_a_pick_may_move_its_target(tmp_path):
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "p", _single(), target="a.png")
    declare_pick(booth, "p", _single(), target="b.png")
    assert marks_for(booth)[0].target == "b.png"


def test_import_adopts_a_legacy_answer_for_an_already_declared_pick(tmp_path):
    """Found by Gróa (flag 10).

    The idempotence rule skipped any stem already present as a mark. If a
    session had re-declared that stem through marks (so the mark exists, still
    unanswered) while the operator's answer sat in the legacy sidecar, the import
    skipped and that answer was stranded on disk forever — with the read path
    forbidden from looking at sidecars. Adopting the answer preserves both rules:
    idempotent, and never clobbers a NEWER judgment.
    """
    from booth.asks import ANSWER_SUFFIX, build_answer, normalize_ask
    from booth.marks import import_legacy_asks

    booth = tmp_path / "b"
    _sidecar(booth, "winner", _single())
    doc = build_answer(normalize_ask(_single(), "winner"), "B — async", notes="from the sidecar")
    (booth / f"winner{ANSWER_SUFFIX}").write_text(json.dumps(doc))
    declare_pick(booth, "winner", _single())          # re-declared, unanswered
    assert marks_for(booth)[0].answer is None

    import_legacy_asks(booth)
    got = marks_for(booth)[0]
    assert got.answer is not None, "the legacy answer was stranded"
    assert got.answer["choice"] == "B — async"
    assert open_marks(marks_for(booth)) == []


def test_import_never_overwrites_an_answer_made_through_marks(tmp_path):
    """The other half of the same rule: a judgment recorded SINCE the sidecar
    outranks it, and adoption must not reach back over it."""
    from booth.asks import ANSWER_SUFFIX, build_answer, normalize_ask
    from booth.marks import import_legacy_asks

    booth = tmp_path / "b"
    _sidecar(booth, "winner", _single())
    old = build_answer(normalize_ask(_single(), "winner"), "A — baseline")
    (booth / f"winner{ANSWER_SUFFIX}").write_text(json.dumps(old))
    declare_pick(booth, "winner", _single())
    answer_pick(booth, "winner", "B — async")         # the operator changed his mind

    import_legacy_asks(booth)
    assert marks_for(booth)[0].answer["choice"] == "B — async"


def test_the_doc_view_carries_the_marks(client):
    """INV-3's third surface — flagged 4/4 by the panel as named in the rule but
    covered by no test, so shipping it unmarked would have passed."""
    from booth.marks import write_note

    c, data = client
    b = data / "b"
    b.mkdir()
    (b / "notes.md").write_text("# report\n\nprose here\n")
    write_note(b, "notes.md", "this section is wrong")

    html = c.get("/b/b/view?f=notes.md").text
    assert "prose here" in html
    assert "this section is wrong" in html


def test_a_corrupt_marks_file_gives_the_browser_a_409_not_a_500(client):
    """The request was fine and the service is fine — the state on disk is not,
    and the refusal is deliberate. A 500 would read as "the Booth is broken" and
    send the operator looking for something to restart."""
    c, data = client
    b = data / "b"
    b.mkdir()
    from booth.marks import write_note
    write_note(b, "a.png", "keep me")
    (b / MARKS_FILE).write_text("{truncated")

    r = c.post("/b/b/flag", data={"target": "a.png", "on": "1"}, follow_redirects=False)
    assert r.status_code == 409
    body = r.json()
    assert "cannot be read" in body["error"] and body["fix"]
    # the page still renders, so the operator can see the booth at all
    assert c.get("/b/b/").status_code == 200
    assert c.get("/b/b/marks.json").status_code == 200


# ---- findings from the cross-frontier BUG-HUNT panel, 2026-09-22 -------------
#
# Heid panel (thread 01M33XEC1H0298C0D968FWBN7A). Four arms, artifact-only,
# diff-scoped. The headline was 4/4 convergent and none of it had a guard: the
# panel's own mutation tables showed the lock lifecycle SURVIVED every existing
# test, because `test_a_no_op_write_does_not_touch_the_booth` asserts only that
# `.marks.json` is absent and never looks at the lock or at the clock the
# sweeper actually reads.


def test_the_lock_file_is_never_unlinked(tmp_path):
    """The lock must outlive the operation that created it.

    `flock` binds to an INODE, not to a path. Unlinking `.marks.lock` while a
    second writer is blocked on it leaves that writer holding an exclusive lock
    on a deleted inode — and the next writer along creates a FRESH lock file and
    takes it immediately. Two processes then run the read-modify-write
    concurrently and the later `os.replace` drops the earlier one's mark, with
    no error anywhere. Both of them obeyed the protocol.

    The cleanup existed to keep a no-op from leaving a lock file as its only
    trace. That is a tidiness goal, and it bought a lost-update race.
    """
    from booth.marks import MARKS_LOCK, set_flag

    booth = tmp_path / "b"
    booth.mkdir()
    assert set_flag(booth, "ghost.png", False) is None      # a no-op
    assert (booth / MARKS_LOCK).exists(), "the no-op path unlinked the lock file"


def test_a_no_op_does_not_reset_the_ttl_clock(tmp_path):
    """The property the no-op guard actually exists for, asserted against the
    clock the sweeper reads instead of against one file's absence.

    Creating or removing a directory entry bumps the DIRECTORY's mtime, and
    `_newest_mtime` seeds from exactly that. So `touch` + `unlink` of the lock
    reset the booth's age to zero while leaving no trace behind — the comment on
    the create-only guard reasons about the lock FILE's mtime and misses that
    the directory moved underneath it. Repeated, it kept a dead booth alive
    forever, which is the precise outcome the guard was written to prevent.
    """
    import os

    from booth.app import booth_age_seconds
    from booth.marks import delete_mark, set_flag

    booth = tmp_path / "b"
    booth.mkdir()
    old = 1_000_000_000
    os.utime(booth, (old, old))

    set_flag(booth, "ghost.png", False)     # no-op: never flagged
    delete_mark(booth, "nothing")           # no-op: no such mark

    age = booth_age_seconds(booth, now=old + 90_000)
    assert age > 86_400, f"a no-op reset the TTL clock (age fell to {age:.0f}s)"


def test_a_real_mark_still_resets_the_ttl_clock(tmp_path):
    """The other half of the same rule, so the fix cannot overshoot into
    'marking is never activity'. Marking IS activity and must reset the clock;
    only a write that changes nothing must not."""
    import os

    from booth.app import booth_age_seconds
    from booth.marks import set_flag

    booth = tmp_path / "b"
    booth.mkdir()
    old = 1_000_000_000
    os.utime(booth, (old, old))

    set_flag(booth, "a.png", True)          # a real mark

    assert booth_age_seconds(booth, now=old + 90_000) < 86_400


def test_a_non_string_note_text_does_not_crash_the_read(tmp_path):
    """`_clean_text` did `(text or "").replace(...)`, so a stored `text` that is
    valid JSON but not a string raised AttributeError out of the READ path.

    That is not a marks bug, it is an INDEX bug: `list_booths` reads every
    booth's marks on every page load, so one poisoned file took down `/` and
    `/healthz` for all 25 booths. The module's stated posture is that a mark it
    cannot read renders as broken, never as a 500.
    """
    booth = tmp_path / "b"
    booth.mkdir()
    (booth / MARKS_FILE).write_text(json.dumps({
        "version": 1,
        "marks": [{"id": "n1", "shape": "note", "text": 7,
                   "created": "2026-09-21T00:00:00+00:00"}],
    }))

    marks = marks_for(booth)
    assert len(marks) == 1
    assert marks[0].error, "a poisoned note read clean instead of reading broken"


def test_a_non_string_created_does_not_crash_the_sort(tmp_path):
    """`marks_for` sorts on `(created, id)`. A stored `created` of the wrong type
    made that comparison raise TypeError — same blast radius as the note above,
    reached through the sort rather than through hydration."""
    booth = tmp_path / "b"
    booth.mkdir()
    (booth / MARKS_FILE).write_text(json.dumps({
        "version": 1,
        "marks": [
            {"id": "a", "shape": "note", "text": "fine",
             "created": "2026-09-21T00:00:00+00:00"},
            {"id": "b", "shape": "note", "text": "also fine", "created": 17},
        ],
    }))

    marks = marks_for(booth)
    assert len(marks) == 2
    # An unreadable mark loses its `created` and so sorts FIRST — the stated
    # rule is `("", id)` against `(created, id)`. A mark nobody can read is the
    # one that wants looking at, and the alternative is it landing at an
    # arbitrary position in the middle of the panel.
    assert [m.id for m in marks] == ["b", "a"]
    assert marks[0].error and not marks[1].error


def test_legacy_import_order_survives_same_second_mtimes(tmp_path):
    """ROADMAP states the legacy import's order is `(mtime, name)`. It was
    stamping `created` at whole-second resolution, so two sidecars written in
    the same second lost the fractional part that distinguished them and
    `marks_for`'s `(created, id)` tie-break silently re-sorted them into
    alphabetical order — reversing the pair the importer had just ordered.

    Deterministic order is a v1 invariant precisely because the operator refers
    to things positionally. An order that is stated and not kept is worse than
    one that was never claimed.
    """
    import os

    from booth.marks import import_legacy_asks

    booth = tmp_path / "b"
    booth.mkdir()
    for stem in ("zeta", "alpha"):
        (booth / f"{stem}{ASK_SUFFIX}").write_text(json.dumps(_single()))
    # Same whole second, different fractions: `zeta` is OLDER and must come first.
    os.utime(booth / f"zeta{ASK_SUFFIX}", (1_700_000_000.10, 1_700_000_000.10))
    os.utime(booth / f"alpha{ASK_SUFFIX}", (1_700_000_000.90, 1_700_000_000.90))

    imported = [m.id for m in import_legacy_asks(booth)]
    assert imported == ["zeta", "alpha"], "the importer's own order is wrong"
    assert [m.id for m in marks_for(booth)] == imported, (
        "the read path re-sorted what the importer ordered"
    )


def test_the_index_survives_a_poisoned_marks_file(client):
    """The blast radius, asserted where it actually hurts.

    `list_booths` reads every booth's marks on every index load and `/healthz`
    does the same. One hand-edited or foreign-written `.marks.json` therefore
    took down the front page for all 25 booths — the single-booth failure the
    lenient reader exists to contain, escaping the booth it belongs to.
    """
    c, data = client
    good = data / "good"
    good.mkdir()
    _png(good / "a.png")
    bad = data / "bad"
    bad.mkdir()
    (bad / MARKS_FILE).write_text(json.dumps({
        "version": 1,
        "marks": [{"id": "n1", "shape": "note", "text": {"oops": True}, "created": 3}],
    }))

    assert c.get("/").status_code == 200
    assert c.get("/healthz").status_code == 200
    assert c.get("/b/bad/").status_code == 200


def test_answer_treats_a_non_string_notes_field_as_no_notes(client):
    """`booth_note` guards `text` with `isinstance(..., str)`; `booth_answer`
    passed `notes` straight to `_clean_notes`, which calls `.replace` on it. A
    multipart FILE part named `notes` is a str to nobody, so the route 500'd on
    hostile-but-legal input where its sibling handled the same class of value.

    Both routes now read the field the same way: a value that is not text is no
    value. The CHOICE is the judgment and it still lands — throwing the whole
    answer away over a junk optional field would be the wrong trade."""
    c, data = client
    b = data / "b"
    b.mkdir()
    declare_pick(b, "winner", _single())

    r = c.post(
        "/b/b/answer",
        data={"ask": "winner", "choice": "A — baseline"},
        files={"notes": ("n.txt", b"surprise", "text/plain")},
        follow_redirects=False,
    )
    assert r.status_code == 303
    mark = next(m for m in marks_for(b) if m.id == "winner")
    assert mark.answer["choice"] == "A — baseline"
    assert not mark.answer.get("notes")


def test_an_inline_doc_tile_offers_a_note_control(client):
    """Three item branches, two of them call `marknotes`. The doc branch got the
    flag button and not the note field, so the operator could point at a report
    and not write down why — on the one item kind whose whole purpose is prose.

    This is the exact failure the `blurtoggle` macro comment names ("patched two
    of three"), recurring on the macro that was written to prevent it.
    """
    c, data = client
    b = data / "b"
    b.mkdir()
    (b / "report.md").write_text("# report\n\nprose here\n")

    html = c.get("/b/b/").text
    assert 'value="report.md"' in html, "the doc tile has no mark controls at all"
    # `marknotes`' add-field, which only that macro emits. The booth-level panel
    # has its own note form, so the presence of /note on the page proves nothing.
    assert 'placeholder="a note on this item"' in html, (
        "an inline doc tile has no way to add a note"
    )


def test_the_marks_panel_survives_a_booth_that_also_has_a_link_board(client):
    """The board booth renders as a board instead of a gallery, which is right —
    but the suppression was unconditional, so a pick declared on a booth that
    happens to carry a `links.md` had no form to answer it and no way to say so."""
    c, data = client
    b = data / "b"
    b.mkdir()
    (b / "links.md").write_text("- [a thing](http://example.invalid) <sub>· who · when</sub>\n")
    declare_pick(b, "winner", _single())

    html = c.get("/b/b/").text
    assert "Which render wins?" in html, "a pick on a board booth was unanswerable"


def test_the_zoom_view_does_not_navigate_away_from_a_note_being_typed(client):
    """The viewer's arrow keys move between images and Escape goes back. The
    note textarea landed in the same page, and the handler is on `document`, so
    an arrow key meant for the caret threw away the draft instead of moving it.

    Asserted structurally: the handler must bail on events from an editable
    target. There is no browser in this suite, and a guard nobody can test is
    exactly how this shipped."""
    c, data = client
    b = data / "b"
    b.mkdir()
    _png(b / "a.png")

    js = c.get("/b/b/view?f=a.png").text
    assert "isEditable" in js, "the viewer's key handler has no editing guard"


@pytest.mark.parametrize("route", ["booth_answer", "booth_note", "booth_flag",
                                   "booth_unmark", "booth_import_asks"])
def test_mark_writes_do_not_block_the_event_loop(route):
    """Every mark write takes a blocking `flock` and does synchronous disk I/O.
    In an `async def` handler that runs ON the event loop, so a lock held by
    another process — the CLI mid-`marks-import`, a second browser tab — freezes
    every other request, including the index and `/healthz`.

    Structural, like `test_stdlib_only`, and for the same reason: the failure is
    a property of where the call runs, which no single-process response
    assertion can see. The rule is that an async mark-write handler hands the
    locked section to a worker thread and never calls the writer inline.
    """
    src = pathlib.Path(__file__).parent.parent / "booth" / "app.py"
    fn = next(
        n for n in ast.walk(ast.parse(src.read_text()))
        if isinstance(n, ast.AsyncFunctionDef) and n.name == route
    )
    writers = {"answer_pick", "write_note", "set_flag", "delete_mark",
               "import_legacy_asks"}
    for node in ast.walk(fn):
        if not isinstance(node, ast.Call):
            continue
        name = getattr(node.func, "id", None) or getattr(node.func, "attr", None)
        if name in writers:
            pytest.fail(f"{route} calls {name}() on the event loop; "
                        "dispatch it through run_in_threadpool")


def test_an_unreadable_mark_is_visible_on_the_page(client):
    """Surviving the poisoned file is half of it. A note whose stored `text` is
    unreadable hydrates with empty text, and the panel rendered that as an empty
    `<pre>` with a withdraw button beside it — which looks exactly like a note
    the operator wrote and then cleared.

    `_hydrate`'s own docstring forbids this for picks ("a broken question the
    session believes it posted has to be visible — silently hiding it is the one
    outcome nobody can debug"). It is the same argument for every shape."""
    c, data = client
    b = data / "b"
    b.mkdir()
    (b / MARKS_FILE).write_text(json.dumps({
        "version": 1,
        "marks": [{"id": "n1", "shape": "note", "text": {"oops": True},
                   "created": "2026-09-21T00:00:00+00:00"}],
    }))

    html = c.get("/b/b/").text
    assert "⚠ broken" in html, "an unreadable mark rendered as an empty note"
    assert "n1" in html


def test_a_marks_file_no_one_can_parse_does_not_take_down_the_index(tmp_path):
    """The v0.2.2 round adopted the RecursionError finding and closed only half
    of it. `_hydrate_safe` guards hydration; `json.loads` runs BEFORE that, in
    `_read_raw`, whose `except (OSError, ValueError, UnicodeDecodeError)` does
    not cover RecursionError or MemoryError.

    So a 400 KB file of nothing but brackets, in any one booth, still returned
    500 for `/` and `/healthz` across every booth on the service. Found by the
    U5 code-review panel against the sibling module and confirmed by running it.
    The read is bounded now and both classes are caught.
    """
    booth = tmp_path / "b"
    booth.mkdir()
    (booth / MARKS_FILE).write_text("[" * 200_000 + "]" * 200_000)

    assert marks_for(booth) == []


def test_a_marks_file_too_large_to_be_marks_is_refused_before_it_is_read(tmp_path):
    """Bounded by `stat`, not survived. A booth holds one marks document, and
    the index reads every booth's on every page load."""
    from booth.marks import MARKS_MAX_BYTES

    booth = tmp_path / "b"
    booth.mkdir()
    (booth / MARKS_FILE).write_text(" " * (MARKS_MAX_BYTES + 10))

    assert marks_for(booth) == []


def test_a_write_over_an_unparseable_marks_file_still_refuses(tmp_path):
    """The strict half of the asymmetry has to see the same failures the lenient
    half does, or a file that reads as "no marks" gets replaced by a write that
    believed it. Same two exception classes, same bound."""
    from booth.marks import MarksCorrupt, set_flag

    booth = tmp_path / "b"
    booth.mkdir()
    (booth / MARKS_FILE).write_text("[" * 200_000 + "]" * 200_000)

    with pytest.raises(MarksCorrupt):
        set_flag(booth, "a.png", True)


# ---- findings from the U5 diff-scoped BUG-HUNT panel, 2026-09-22 ------------


def test_the_marks_reader_never_blocks_on_a_file_that_is_not_a_file(tmp_path):
    """Same hole the size cap opened in the manifest, in the sibling it was
    copied from. `st_size` is 0 for a FIFO, so it passes the cap, and then
    `read_text` blocks with no EOF. `list_booths` reads every booth's marks on
    every `GET /` and `/healthz`."""
    import os
    import signal

    booth = tmp_path / "b"
    booth.mkdir()
    os.mkfifo(booth / MARKS_FILE)

    def _timeout(signum, frame):
        raise AssertionError("marks_for blocked on a FIFO and never returned")

    old = signal.signal(signal.SIGALRM, _timeout)
    signal.alarm(5)
    try:
        assert marks_for(booth) == []
    finally:
        signal.alarm(0)
        signal.signal(signal.SIGALRM, old)


def test_new_marks_and_imported_marks_share_one_stamp_format(tmp_path):
    """The v0.2.2 fix for the legacy-import ordering opened a NEW ordering bug,
    which is the shape worth remembering. `import_legacy_asks` moved to
    microsecond precision while `now_stamp` stayed at whole seconds, and `-` is
    0x2D against `.` at 0x2E — so `...T10:00:00-07:00` sorts BEFORE
    `...T10:00:00.500000-07:00`, putting a LATER mark ahead of an EARLIER
    import inside the same second.

    Deterministic order is a v1 invariant precisely because the operator refers
    to things positionally. One format, or the rule cannot be stated.
    """
    from booth.marks import now_stamp

    stamp = now_stamp()
    assert "." in stamp.split("T")[1], f"now_stamp is not sub-second: {stamp}"
    assert len(stamp.split(".")[1].split("+")[0].split("-")[0]) == 6


def test_the_importer_cannot_raise_out_of_a_poisoned_entry(tmp_path):
    """`marks_for` routes every entry through `_hydrate_safe`; the importer's
    return still went through the bare `_hydrate`, so the one path that reads
    entries it did not write was the one without the guard."""
    booth = tmp_path / "b"
    booth.mkdir()
    (booth / MARKS_FILE).write_text(json.dumps({
        "version": 1,
        "marks": [{"id": "n1", "shape": "note", "text": {"bad": True},
                   "created": "2026-09-21T00:00:00+00:00"}],
    }))
    (booth / f"q1{ASK_SUFFIX}").write_text(json.dumps(_single()))

    from booth.marks import import_legacy_asks
    out = import_legacy_asks(booth)          # must not raise
    assert isinstance(out, list)


def test_a_document_that_would_not_read_back_is_refused_at_the_write(tmp_path):
    """The read bound is on the STORED bytes and the write adds `indent=2`, so a
    document that fits in memory can land over the limit on disk and then read
    back as no marks at all — every mark in the booth gone, silently. Refuse
    loudly instead: a write that fails is recoverable.

    Asserted against `_write_raw` directly, because no single mark can get
    there: `_clean_text` caps a note at TEXT_MAX and a flag is a fixed shape.
    The reachable path is accumulation — `_note_id` puts no ceiling on how many
    notes one booth may carry — which is thousands of writes, not one. Testing
    it through `write_note` would need a fixture nobody could justify, and
    would be testing the cap rather than the guard.
    """
    from booth.marks import MARKS_MAX_BYTES, MarksCorrupt, _write_raw

    booth = tmp_path / "b"
    booth.mkdir()
    bulk = [{"id": f"note-{i}", "shape": "note", "text": "x" * 500,
             "created": "2026-09-21T00:00:00.000000+00:00"}
            for i in range(MARKS_MAX_BYTES // 400)]

    with pytest.raises(MarksCorrupt):
        _write_raw(booth, bulk)
    assert not (booth / MARKS_FILE).exists(), "a refused write still landed"


def test_a_clock_restore_that_fails_does_not_take_the_route_down(tmp_path):
    """The concrete half of the mtime-restore finding.

    `_Locked.__enter__` puts the booth directory's clock back after creating its
    lock, and `os.utime` can fail — a read-only directory, a booth whose owner
    we are not. It used to escape into the route and answer 500 for what is
    otherwise a perfectly good request. Not putting the clock back is a cost
    this module can absorb; not answering is not.

    The RACE half of that finding is documented in the code and deliberately not
    closed: the alternative fix would silently retire the documented behaviour
    that releasing a kept board resets its clock
    (`test_releasing_a_board_RESETS_its_ttl_clock` pins that on purpose), which
    is a TTL doctrine change rather than a bug fix.
    """
    import os

    from booth.marks import MARKS_LOCK, set_flag

    booth = tmp_path / "b"
    booth.mkdir()
    real_utime = os.utime

    def boom(path, *a, **kw):
        if str(path) == str(booth):
            raise PermissionError("read-only directory")
        return real_utime(path, *a, **kw)

    os.utime = boom
    try:
        assert set_flag(booth, "a.png", True) is not None
    finally:
        os.utime = real_utime

    assert (booth / MARKS_LOCK).exists()
    assert [m.target for m in marks_for(booth)] == ["a.png"]