booth/tests/test_lifetime.py

"""U4 — derived lifetime.

A booth's lifetime stops being a boolean somebody remembered to press and
becomes a fact derived from the booth's own state:

    KEPT       `.forever` present                          never swept
    HELD       an open pick, or marks we cannot read        never swept
    EPHEMERAL  otherwise                                    swept past the TTL

See `docs/contracts/u4_derived_lifetime.contract.md`. The invariant ids in the
test names are that contract's.
"""

import json
import os
import pathlib
import re
import stat
import subprocess
import time

import pytest
from fastapi.testclient import TestClient

from booth.app import (
    KEEP_MARKER,
    is_kept,
    VIEW_MARKER,
    booth_age_seconds,
    create_app,
    hold_reason,
    is_expired,
    list_booths,
    record_view,
    sweep_once,
    zip_booth,
)
from booth.items import booth_items
from booth.marks import (answer_pick, declare_pick, delete_mark, hold_read, marks_for, open_marks,
                         read_error, set_flag, write_note)


def _touch(path, when=None):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(b"x")
    if when is not None:
        os.utime(path, (when, when))


def _stale(path, seconds=10_000):
    """Age a booth and everything in it well past any test TTL.

    `follow_symlinks=False`, because a booth may hold a dangling one and
    `os.utime` follows by default — which is the helper raising on a fixture the
    tests deliberately build."""
    t = time.time() - seconds
    for p in sorted(path.rglob("*"), reverse=True):
        try:
            os.utime(p, (t, t), follow_symlinks=False)
        except (OSError, NotImplementedError):
            pass
    os.utime(path, (t, t))


def _pick(booth, mark_id="q1", target=None):
    """Declare an unanswered pick — the thing that holds a booth."""
    return declare_pick(booth, mark_id,
                        {"prompt": "which one?", "options": ["a", "b"]}, target=target)


@pytest.fixture
def client(tmp_path):
    app = create_app(tmp_path, ttl_hours=24, start_sweeper=False)
    return TestClient(app), tmp_path


# ---- record_view / VIEW_MARKER ---------------------------------------------


def test_record_view_writes_the_marker(tmp_path):
    booth = tmp_path / "b"
    _touch(booth / "a.png")

    record_view(booth)

    assert (booth / VIEW_MARKER).exists()


def test_a_view_resets_the_clock_through_the_existing_age_rule(tmp_path):
    """INV-1. No new arithmetic: `.viewed` is a dotfile and NOT a `.lock`
    dotfile, so `_newest_mtime` already counts it. `booth_age_seconds` is
    untouched by this unit."""
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    _stale(booth)
    assert booth_age_seconds(booth) > 3600

    record_view(booth)

    assert booth_age_seconds(booth) < 60


def test_the_view_marker_is_not_special_cased_in_the_age_rule(tmp_path):
    """INV-1's ACTUAL falsifier, and the reason the test above is not enough.

    Cross-frontier review, 2026-09-22: an implementation that special-cases
    `.viewed` inside `_newest_mtime` — the new arithmetic INV-1 forbids —
    passes a "the clock moved" assertion perfectly well. What it cannot pass is
    equivalence: a booth holding `.viewed` and a booth holding any other
    non-lock dotfile of the SAME mtime must report the same age, because the
    rule is about the tree, not about this file.
    """
    when = time.time() - 5_000
    viewed = tmp_path / "viewed"
    _touch(viewed / "a.png", when=when)
    _touch(viewed / VIEW_MARKER, when=when)
    other = tmp_path / "other"
    _touch(other / "a.png", when=when)
    _touch(other / ".anything-else", when=when)
    for b in (viewed, other):
        os.utime(b, (when, when))

    now = time.time()
    assert abs(booth_age_seconds(viewed, now=now)
               - booth_age_seconds(other, now=now)) < 0.001

    # ...and a `.lock` dotfile is the one that must NOT count, or the exemption
    # the view rule rides on has stopped being an exemption.
    locked = tmp_path / "locked"
    _touch(locked / "a.png", when=when)
    _touch(locked / ".marks.lock")
    os.utime(locked, (when, when))
    assert booth_age_seconds(locked, now=now) > 3600


def test_a_viewed_booth_is_not_swept(tmp_path):
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    _stale(booth)

    record_view(booth)

    assert sweep_once(tmp_path, ttl_seconds=3600) == []
    assert booth.exists()


def test_the_view_marker_is_not_an_item(tmp_path):
    """A dotfile, so it costs nothing in counts, galleries or zips — the same
    reason `.forever`, `.marks.json` and `.booth.json` cost nothing."""
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    record_view(booth)

    assert [it.rel for it in booth_items(booth)] == ["a.png"]
    assert list_booths(tmp_path, ttl_seconds=3600)[0]["count"] == 1
    assert VIEW_MARKER not in zip_booth(booth).decode("latin-1")


def test_record_view_never_raises_on_a_read_only_booth(tmp_path):
    """INV-5. A read-only mount, a booth we do not own, a full disk: the cost
    is the timestamp, never the page."""
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    os.chmod(booth, stat.S_IRUSR | stat.S_IXUSR)
    try:
        record_view(booth)                       # must not raise
        assert not (booth / VIEW_MARKER).exists()
    finally:
        os.chmod(booth, stat.S_IRWXU)


def test_a_read_only_booth_still_serves_all_three_view_pages(client):
    """INV-5's ACTUAL falsifier — the one the unit test above does not reach.

    Cross-frontier review, 2026-09-22: `record_view` swallowing OSError proves
    nothing about the ROUTES. An implementation that keeps the swallow and then
    touches the marker a second time outside the guard 500s every one of these
    and still passes the unit test. The invariant is about the pages.
    """
    c, data = client
    booth = data / "b"
    _touch(booth / "a.png")
    os.chmod(booth, stat.S_IRUSR | stat.S_IXUSR)
    try:
        assert c.get("/b/b/").status_code == 200
        assert c.get("/b/b/view?f=a.png").status_code == 200
        assert c.get("/b/b/marks").status_code == 200
        assert not (booth / VIEW_MARKER).exists(), "nothing was written"
    finally:
        os.chmod(booth, stat.S_IRWXU)


def test_record_view_never_raises_on_a_missing_booth(tmp_path):
    record_view(tmp_path / "gone")               # must not raise


# ---- which routes count as a view ------------------------------------------


def test_the_booth_page_records_a_view(client):
    c, data = client
    _touch(data / "b" / "a.png")

    assert c.get("/b/b/").status_code == 200
    assert (data / "b" / VIEW_MARKER).exists()


def test_the_zip_download_records_a_view(client):
    """Same route, same deliberate act — and the marker is written before the
    early return, not after it."""
    c, data = client
    _touch(data / "b" / "a.png")

    assert c.get("/b/b/?download=1").status_code == 200
    assert (data / "b" / VIEW_MARKER).exists()


def test_the_zoom_page_records_a_view(client):
    c, data = client
    _touch(data / "b" / "a.png")

    assert c.get("/b/b/view?f=a.png").status_code == 200
    assert (data / "b" / VIEW_MARKER).exists()


def test_the_marks_page_records_a_view(client):
    c, data = client
    _touch(data / "b" / "a.png")

    assert c.get("/b/b/marks").status_code == 200
    assert (data / "b" / VIEW_MARKER).exists()


def test_the_legacy_asks_url_records_a_view_through_the_redirect(client):
    """`/asks` is a 308 to `/marks`; it needs no call of its own, and adding
    one would double-count."""
    c, data = client
    _touch(data / "b" / "a.png")

    assert c.get("/b/b/asks").status_code == 200   # followed
    assert (data / "b" / VIEW_MARKER).exists()


def test_a_verbatim_booth_records_a_view(client):
    """The index.html branch returns before the gallery is built, so the
    marker has to be written above that fork."""
    c, data = client
    (data / "b").mkdir()
    (data / "b" / "index.html").write_text("<html><body>hi</body></html>")

    assert c.get("/b/b/").status_code == 200
    assert (data / "b" / VIEW_MARKER).exists()


@pytest.mark.parametrize("path", ["/", "/healthz"])
def test_browsing_the_index_is_not_a_view(client, path):
    """INV-7 / the IA's rule: a view is a DELIBERATE act, so it cannot be
    triggered by browsing the index."""
    c, data = client
    _touch(data / "b" / "a.png")

    _stale(data / "b")
    before = booth_age_seconds(data / "b")

    c.get(path)

    assert not (data / "b" / VIEW_MARKER).exists()
    assert booth_age_seconds(data / "b") >= before - 1


def test_polling_marks_json_is_not_a_view(client):
    """INV-7. An agent must not be able to hold its own booth open by polling
    for the answer it is waiting on."""
    c, data = client
    _touch(data / "b" / "a.png")

    _stale(data / "b")
    before = booth_age_seconds(data / "b")

    for _ in range(3):
        assert c.get("/b/b/marks.json").status_code == 200

    assert not (data / "b" / VIEW_MARKER).exists()
    # The falsifier is the AGE, not the marker: a handler that wrote any other
    # non-dot file would hold the booth open just as effectively.
    assert booth_age_seconds(data / "b") >= before - 1


def test_fetching_a_file_is_not_a_view(client):
    """INV-7. Asset GETs are issued BY the page, and a hotlinked image would
    otherwise keep a booth alive forever."""
    c, data = client
    _touch(data / "b" / "a.png")

    _stale(data / "b")
    before = booth_age_seconds(data / "b")

    assert c.get("/b/b/a.png").status_code == 200

    assert not (data / "b" / VIEW_MARKER).exists()
    assert booth_age_seconds(data / "b") >= before - 1


# ---- is_held ----------------------------------------------------------------


def test_hold_reason_does_no_io_of_its_own(tmp_path):
    """INV-3. A predicate that reached for the filesystem could answer one way
    for the card and another for the sweeper; one that cannot, cannot.

    Discriminated rather than asserted: the marks handed in belong to a booth
    that does not exist on disk. Anything that went looking would raise or
    contradict itself.
    """
    nowhere = tmp_path / "does-not-exist"
    booth = tmp_path / "b"
    booth.mkdir()
    _pick(booth)
    real = marks_for(booth)
    assert not nowhere.exists()

    assert hold_reason(real, None) == "open"
    assert hold_reason([], None) is None
    assert hold_reason([], "damaged") == "unreadable"
    # ...and the same inputs give the same answer regardless of any booth.
    assert hold_reason(real, None) == "open"


def test_hold_read_answers_both_questions_from_one_read(tmp_path):
    """INV-3's other half, and Hulda's finding on 2026-09-22.

    `marks_for` then `read_error` is TWO reads, and two reads of one file are
    not one read of one state: a write landing between them can yield
    `([], None)` — no marks, no error — a pair that described the booth at no
    instant and is exactly the pair that deletes. `hold_read` is one read.

    It must also agree with `marks_for` on a clean file, or the badge beside
    the hold would be counting something else.
    """
    booth = tmp_path / "b"
    booth.mkdir()
    _pick(booth)

    marks, error = hold_read(booth)
    assert error is None
    assert [m.id for m in marks] == [m.id for m in marks_for(booth)]

    (booth / ".marks.json").write_text("not json at all")
    marks, error = hold_read(booth)
    assert marks == []
    assert error is not None, "one read reports the damage AND the emptiness"


def test_an_open_pick_holds(tmp_path):
    booth = tmp_path / "b"
    booth.mkdir()
    _pick(booth)

    assert hold_reason(marks_for(booth), None) == "open"


def test_an_answered_pick_does_not_hold(tmp_path):
    booth = tmp_path / "b"
    booth.mkdir()
    _pick(booth)
    answer_pick(booth, "q1", "a")

    assert hold_reason(marks_for(booth), None) is None


def test_a_partially_answered_pick_STILL_holds(tmp_path):
    """Operator-settled 2026-09-21, and the reason `open_marks` exists rather
    than an `answer is None` test: a lifetime rule that released a booth on the
    first radio click would sweep a review in flight."""
    booth = tmp_path / "b"
    booth.mkdir()
    declare_pick(booth, "batch", {"questions": [
        {"key": "one", "prompt": "a?", "options": ["x", "y"]},
        {"key": "two", "prompt": "b?", "options": ["x", "y"]},
    ]})
    answer_pick(booth, "batch", {"one": "x"})

    assert hold_reason(marks_for(booth), None) == "open"


def test_a_note_does_not_hold(tmp_path):
    """A note is the operator's OUTPUT, not an owed answer."""
    booth = tmp_path / "b"
    booth.mkdir()
    write_note(booth, None, "looks good")

    assert hold_reason(marks_for(booth), None) is None


def test_a_flag_does_not_hold(tmp_path):
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    set_flag(booth, "a.png", True)

    assert hold_reason(marks_for(booth), None) is None


def test_a_booth_with_no_marks_does_not_hold(tmp_path):
    booth = tmp_path / "b"
    _touch(booth / "a.png")

    assert hold_reason(marks_for(booth), read_error(booth)) is None


def test_unreadable_marks_hold(tmp_path):
    """INV-6. `marks_for` is lenient because a review page that will not render
    is worse than one missing an annotation. The same leniency on the DELETE
    path would wipe the booth whose judgment we just failed to read."""
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    (booth / ".marks.json").write_text("{ this is not json")

    assert marks_for(booth) == [], "the read stays lenient"
    assert hold_reason(marks_for(booth), read_error(booth)) == "unreadable", "the reaper does not"


# ---- the sweeper ------------------------------------------------------------


def test_a_held_booth_survives_the_sweep(tmp_path):
    """The core of the unit: a booth the operator still owes an answer to
    cannot be swept, however old."""
    doomed = tmp_path / "doomed"
    _touch(doomed / "a.png")
    _stale(doomed)

    held = tmp_path / "held"
    _touch(held / "a.png")
    _pick(held)
    _stale(held)

    wiped = sweep_once(tmp_path, ttl_seconds=3600)

    assert wiped == ["doomed"]
    assert held.exists()


def test_a_booth_with_unreadable_marks_survives_the_sweep(tmp_path):
    """INV-6, end to end."""
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    (booth / ".marks.json").write_text("{ truncated")
    _stale(booth)
    # A doomed sibling, so "spare everything" is not a passing implementation.
    doomed = tmp_path / "doomed"
    _touch(doomed / "a.png")
    _stale(doomed)

    assert sweep_once(tmp_path, ttl_seconds=3600) == ["doomed"]
    assert booth.exists()


def test_answering_the_last_pick_releases_the_booth(tmp_path):
    """The hold's exit. Answering IS a write, so the booth's clock resets too —
    it rejoins the sweep on a fresh clock, exactly as a released board does."""
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    _pick(booth)
    _stale(booth)

    assert sweep_once(tmp_path, ttl_seconds=3600) == []

    answer_pick(booth, "q1", "a")
    _stale(booth)

    assert sweep_once(tmp_path, ttl_seconds=3600) == ["b"]


def test_held_booth_is_still_reported_expired_by_age(tmp_path):
    """INV-2. `is_expired` stays a pure age question — expiry arithmetic and
    reaper policy are kept apart so they cannot drift into each other. This is
    the same split `is_kept` already has."""
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    _pick(booth)
    _stale(booth)

    assert is_expired(booth, ttl_seconds=3600)
    assert sweep_once(tmp_path, ttl_seconds=3600) == []


def test_a_held_booth_is_still_deletable(client):
    """INV-2. `sweep_once` is the ONLY caller that honours a hold, exactly as
    it is the only caller that honours the keep sentinel. A hold is protection
    from the timer, never from the operator."""
    c, data = client
    booth = data / "b"
    _touch(booth / "a.png")
    _pick(booth)

    c.post("/b/b/delete", follow_redirects=False)
    assert not booth.exists()


def test_a_held_booth_is_still_deletable_via_the_api(client):
    c, data = client
    booth = data / "b"
    _touch(booth / "a.png")
    _pick(booth)

    assert c.delete("/b/b").status_code == 200
    assert not booth.exists()


# ---- the index card ---------------------------------------------------------


def test_list_booths_carries_held_and_the_marks_error(tmp_path):
    _touch(tmp_path / "plain" / "a.png")

    held = tmp_path / "held"
    _touch(held / "a.png")
    _pick(held)

    broken = tmp_path / "broken"
    _touch(broken / "a.png")
    (broken / ".marks.json").write_text("nope")

    by = {b["name"]: b for b in list_booths(tmp_path, ttl_seconds=3600)}

    assert by["plain"]["hold"] is None
    assert by["held"]["hold"] == "open"
    assert by["broken"]["hold"] == "unreadable"


def test_the_card_says_held_instead_of_a_countdown(client):
    """INV-4. A booth that is not counting down always says why. An invisible
    rule that silently stopped the clock would be strictly worse than the
    boolean it replaces — `.forever` was at least visible as a lane."""
    c, data = client
    booth = data / "held"
    _touch(booth / "a.png")
    _pick(booth)

    html = c.get("/").text

    assert "held until answered" in html
    assert "expires in" not in html


def test_the_card_says_so_when_the_marks_are_unreadable(client):
    """The one hold nothing will release on its own, so it must not read as an
    ordinary held booth."""
    c, data = client
    booth = data / "broken"
    _touch(booth / "a.png")
    (booth / ".marks.json").write_text("{{{")

    html = c.get("/").text

    assert "marks unreadable" in html
    assert "expires in" not in html


def test_an_ordinary_booth_still_counts_down(client):
    c, data = client
    _touch(data / "plain" / "a.png")

    html = c.get("/").text

    assert "expires in" in html
    assert "held until answered" not in html


def test_the_booth_header_says_held_too(client):
    """The index card and the booth header are two surfaces on one fact, and a
    booth URL handed to the operator lands on the SECOND one."""
    c, data = client
    booth = data / "held"
    _touch(booth / "a.png")
    _pick(booth)

    html = c.get("/b/held/").text

    assert "held until answered" in html
    assert "expires in" not in html


def test_kept_beats_held_in_the_display(client):
    """A kept booth is exempt either way, and showing two reasons for one
    exemption is the two-representations-of-one-state trap."""
    c, data = client
    booth = data / "both"
    _touch(booth / "a.png")
    _touch(booth / KEEP_MARKER)
    _pick(booth)

    html = c.get("/").text

    assert "held until answered" not in html
    assert "expires in" not in html


# ---- release is activity, stated ------------------------------------------


def test_releasing_a_board_RECORDS_A_VIEW(client):
    """A released board survives another full TTL because RELEASING IT IS
    SOMEBODY TOUCHING IT — a rule — and no longer because unlinking a file
    happened to bump the directory's mtime, which is not one.

    The behaviour is unchanged. Its reason is now stated, which is what U4 owed
    the TTL doctrine it inherited.
    """
    c, data = client
    booth = data / "links"
    _touch(booth / "a.png")
    _touch(booth / KEEP_MARKER)

    c.post("/b/links/unkeep", follow_redirects=False)

    assert not (booth / KEEP_MARKER).exists()
    assert (booth / VIEW_MARKER).exists(), "release is activity, on purpose"


def test_keeping_a_board_does_not_need_to_record_a_view(client):
    """Keep exempts it outright, so there is nothing for a clock reset to buy —
    and writing the marker anyway would put a second mechanism on one job."""
    c, data = client
    booth = data / "b"
    _touch(booth / "a.png")

    c.post("/b/b/keep", follow_redirects=False)

    assert (booth / KEEP_MARKER).exists()
    assert not (booth / VIEW_MARKER).exists()


def test_the_marks_page_carries_the_lifetime_line(client):
    """INV-4's third home, and the one that matters most. A verbatim booth's own
    index.html is served untouched, so it has no Booth-rendered header — and a
    report that ASKS something is the archetype of a held booth. Without this
    the booths most likely to be held would be the ones that never say so."""
    c, data = client
    booth = data / "report"
    booth.mkdir()
    (booth / "index.html").write_text("<html><body>the report</body></html>")
    _pick(booth)

    html = c.get("/b/report/marks").text

    assert "held until answered" in html


@pytest.mark.parametrize("setup,expect_held", [
    ("open_pick", True),
    ("answered_pick", False),
    ("partial_pick", True),
    ("note_only", False),
    ("corrupt", True),
    ("nothing", False),
])
def test_the_card_and_the_sweeper_never_disagree(tmp_path, setup, expect_held):
    """INV-3, in its falsifiable form. Two surfaces reading one truth is how the
    zoom view lost its captions; here the stakes are a deletion, so the check is
    behavioural rather than by inspection — what the index reports as held and
    what the sweeper refuses to take must agree on every shape of booth."""
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    if setup == "open_pick":
        _pick(booth)
    elif setup == "answered_pick":
        _pick(booth)
        answer_pick(booth, "q1", "a")
    elif setup == "partial_pick":
        declare_pick(booth, "batch", {"questions": [
            {"key": "one", "prompt": "a?", "options": ["x", "y"]},
            {"key": "two", "prompt": "b?", "options": ["x", "y"]},
        ]})
        answer_pick(booth, "batch", {"one": "x"})
    elif setup == "note_only":
        write_note(booth, None, "fine")
    elif setup == "corrupt":
        (booth / ".marks.json").write_text("}{")
    _stale(booth)

    card = list_booths(tmp_path, ttl_seconds=3600)[0]
    survived = sweep_once(tmp_path, ttl_seconds=3600) == []

    assert (card["hold"] is not None) is expect_held
    assert survived is expect_held, "the card and the reaper must not disagree"


# ---- INV-4, discriminated per surface ---------------------------------------
#
# The first cut of these tests all rendered `GET /` and checked a substring,
# which meant every one of them passed under a change that dropped the line
# from the booth header, from the marks page, or from the board branch. Four
# of four cross-frontier arms found the board branch; the rest of this block
# is the same lesson applied to the surfaces they did not have to find.


def _board(data, name="links", rows=1):
    b = data / name
    b.mkdir(parents=True, exist_ok=True)
    (b / "links.md").write_text("".join(
        f"- [row {i}](http://x{i}/) <sub>· nobody · 2026-08-23 10:00</sub>\n"
        for i in range(rows)))
    return b


def test_a_BOARD_booths_header_still_states_its_lifetime(client):
    """The drift all four arms found on 2026-09-22.

    The booth header's sub-line forks on `{% if board %}`, and the lifetime
    macro sat only in the `{% else %}` — so a booth carrying `links.md`
    rendered a link count and NOTHING about its lifetime. INV-4 says the
    templates have no path that renders neither, and that was a path.

    Reachable, not theoretical: the release button on the kept card drops the
    sentinel, and a hand-made `links.md` booth never had one. The standing
    board being kept by construction is what hid it.
    """
    c, data = client
    _board(data, rows=2)

    html = c.get("/b/links/").text

    assert "2 links" in html
    assert "expires in" in html, "a non-kept board must say when it goes"


def test_a_held_BOARD_booth_says_held_in_its_header(client):
    c, data = client
    b = _board(data, rows=1)
    _pick(b)

    html = c.get("/b/links/").text

    assert "held until answered" in html
    assert "expires in" not in html


def test_kept_beats_held_in_the_HEADER_not_just_the_index(client):
    """The first version of this test only rendered `/`, where kept cards took
    a hardcoded `kept` string and never reached the macro — so swapping the
    macro's precedence to held-before-kept passed it. This renders the surface
    that actually calls the macro with `kept=True`."""
    c, data = client
    booth = data / "both"
    _touch(booth / "a.png")
    _touch(booth / KEEP_MARKER)
    _pick(booth)

    html = c.get("/b/both/").text

    assert "kept" in html
    assert "held until answered" not in html
    assert "expires in" not in html


def test_the_booth_header_alone_carries_the_hold(client):
    """Rendered without touching the index, so dropping the header's line
    cannot be masked by the index card still having one."""
    c, data = client
    booth = data / "held"
    _touch(booth / "a.png")
    _pick(booth)

    assert "held until answered" in c.get("/b/held/").text


def test_a_kept_board_still_surfaces_unreadable_marks(client):
    """A kept booth is exempt either way, so the hold is moot — but damaged
    judgment is not an exemption, it is a thing somebody has to fix, and the
    durable boards are where losing the operator's marks costs most. The kept
    card used to render a hardcoded `kept` and say nothing.

    Cross-frontier paraphrase panel, 2026-09-22: `held` and `marks_error` are
    raw facts on the card; only the DISPLAY has a precedence."""
    c, data = client
    booth = data / "board"
    _touch(booth / "a.png")
    _touch(booth / KEEP_MARKER)
    (booth / ".marks.json").write_text("<<<broken>>>")

    index = c.get("/").text
    assert "marks unreadable" in index
    assert "kept" in index

    assert "marks unreadable" in c.get("/b/board/").text


def test_a_404_zoom_url_is_not_a_view(client):
    """`record_view` sits BELOW the 404s in the zoom route, so a crawler
    walking dead zoom URLs cannot hold a booth open — the same reasoning that
    keeps `marks.json` off the view list. Regin and Kimi both raised it."""
    c, data = client
    booth = data / "b"
    _touch(booth / "a.png")

    assert c.get("/b/b/view?f=no-such-file.png").status_code == 404

    assert not (booth / VIEW_MARKER).exists()


# ---- INV-2, through the CLI the sessions actually call ----------------------


def test_booth_rm_deletes_a_held_booth(tmp_path):
    """INV-2 via `scripts/booth`, not just the HTTP routes. A hold is
    protection from the TIMER; nothing about it may reach a deliberate delete,
    and `booth rm` is the one 17 handles type."""
    script = pathlib.Path(__file__).parent.parent / "scripts" / "booth"
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    _pick(booth)
    assert hold_reason(*hold_read(booth)) == "open"

    r = subprocess.run([str(script), "rm", "b"], capture_output=True, text=True,
                       env={**os.environ, "BOOTH_DATA_DIR": str(tmp_path),
                            "BOOTH_URL": "http://booth.invalid"}, timeout=30)

    assert r.returncode == 0, r.stderr
    assert not booth.exists()


# ---- what the bug-hunt panel found (2026-09-22) ----------------------------
#
# Four independent paths into one shape: a read that FAILED still resolving to
# "no hold", and therefore to a delete. No single arm found all four. The
# invariant declared to the panel was "a deletion decision must never be made
# from a read that failed"; these are the tests that make it true.


def test_a_pick_that_hydrated_BROKEN_still_holds(tmp_path):
    """The round's strongest finding (Hulda, solo).

    `.marks.json` parses, but one mark fails normalization, so `_hydrate_safe`
    hands back a Mark carrying `error`. `_is_open` returns False for an errored
    pick ON PURPOSE — it can never be answered, which is what the CLI's exit
    code 4 means — so the booth read as `not held` and SWEPT, while the panel
    beside it rendered the broken mark in full. The fail-safe was built for
    file-level damage and missed entry-level.

    A mark we cannot read is judgment we cannot see. Deleting the booth it
    belongs to is the one thing we must not do with it.
    """
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    (booth / ".marks.json").write_text(json.dumps({
        "version": 1,
        "marks": [{"id": "q1", "shape": "pick", "created": "2026-09-22T00:00:00.000000+00:00",
                   "doc": "this should be a dict, not a string"}],
    }))

    marks = marks_for(booth)
    assert len(marks) == 1 and marks[0].error is not None, "the panel still renders it"
    assert open_marks(marks) == [], "and it is NOT open — a broken pick cannot be answered"

    assert hold_reason(*hold_read(booth)) == "unreadable", "but it still holds"

    _stale(booth)
    doomed = tmp_path / "doomed"
    _touch(doomed / "a.png")
    _stale(doomed)
    assert sweep_once(tmp_path, ttl_seconds=3600) == ["doomed"]
    assert booth.exists()


def test_a_present_but_blank_marks_file_holds(tmp_path):
    """Gróa, solo. `_read_raw_strict` early-returns for whitespace-only content
    — "absent, empty and valid-but-empty are all no marks yet", which is right
    for the WRITE path it was written for. On the delete path it is not: our
    writer never produces a blank marks document, so a blank one that exists
    is something that went wrong, and rmtree is not the response to that."""
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    (booth / ".marks.json").write_text("   \n\n  ")
    _stale(booth)

    assert sweep_once(tmp_path, ttl_seconds=3600) == []
    assert booth.exists()


def test_a_booth_whose_own_stat_fails_reads_as_FRESH_not_ancient(tmp_path):
    """Hulda, solo. `_newest_mtime` returned 0.0 when the booth's own stat
    failed, which made it maximally ancient and therefore the FIRST thing the
    sweeper takes — a permissions problem resolving to a deletion. Not knowing
    a booth's age is a reason to leave it alone."""
    gone = tmp_path / "vanished"
    assert booth_age_seconds(gone) < 60, "unknowable age is FRESH, not epoch-old"


def test_a_dangling_symlink_is_still_skipped_not_treated_as_unknowable(tmp_path):
    """The other half of that fix, and the reason it is not a blanket catch: a
    dangling symlink and a file removed mid-scan both raise FileNotFoundError,
    and neither is a thing with an mtime worth counting. Only OTHER stat
    failures mean 'there is something here we cannot read'."""
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    _stale(booth)
    (booth / "dangling").symlink_to(tmp_path / "nowhere")
    os.utime(booth, (time.time() - 10_000,) * 2)

    assert booth_age_seconds(booth) > 3600, "a dangling link does not make a booth fresh"


def test_is_kept_treats_an_unreadable_sentinel_as_KEPT(tmp_path):
    """Gróa, solo. `Path.exists()` maps ELOOP and EACCES to False, so a kept
    booth whose sentinel could not be stat'd became sweepable. A symlink
    sentinel counts too, dangling or not — somebody put it there to mean keep."""
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    (booth / KEEP_MARKER).symlink_to(tmp_path / "does-not-exist")
    _stale(booth)

    assert is_kept(booth) is True
    assert sweep_once(tmp_path, ttl_seconds=3600) == []


def test_record_view_refuses_to_follow_a_planted_symlink(tmp_path):
    """Three of four arms, independently. `Path.touch()` follows an existing
    symlink, so a booth carrying `.viewed -> /anywhere` turned every page view
    into an mtime write at an arbitrary path under the service uid — and any
    fleet session can write into a booth, because making a folder is the whole
    API. A symlink here now raises ELOOP into the swallow: view-recording
    quietly stops for that booth, which is the right way to lose this."""
    outside = tmp_path / "outside.txt"
    outside.write_text("do not touch me")
    old = time.time() - 50_000
    os.utime(outside, (old, old))

    booth = tmp_path / "b"
    _touch(booth / "a.png")
    (booth / VIEW_MARKER).symlink_to(outside)

    record_view(booth)                      # must not raise, must not follow

    assert abs(outside.stat().st_mtime - old) < 1, "the target's mtime is untouched"


def test_a_lock_file_is_not_a_viewable_item(tmp_path):
    """The panel pointed `?f=.marks.lock` at the zoom route and held a booth
    open with a file the service created itself. A dotfile is not an item, and
    a view of a thing that is not an item is not a view of the booth."""
    app = create_app(tmp_path, ttl_hours=24, start_sweeper=False)
    c = TestClient(app)
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    _touch(booth / ".marks.lock")

    c.get("/b/b/view?f=.marks.lock")

    assert not (booth / VIEW_MARKER).exists()


def test_releasing_an_ALREADY_released_booth_is_not_activity(client):
    """An unconditional `record_view` on unkeep made POSTing release at an
    already-released booth an endless TTL refresh — contradicting this route's
    own no-op promise and diverging from the CLI, which removes the sentinel
    without recording anything. Release is activity; releasing nothing is not."""
    c, data = client
    booth = data / "b"
    _touch(booth / "a.png")
    assert not (booth / KEEP_MARKER).exists()

    r = c.post("/b/b/unkeep", follow_redirects=False)

    assert r.status_code == 303, "still a no-op, still not a 500"
    assert not (booth / VIEW_MARKER).exists()


def test_a_forever_that_is_a_DIRECTORY_does_not_500_the_release(client):
    """Pre-existing, re-exposed: `unlink` on a directory raises
    IsADirectoryError straight through the route, which made the card's release
    button permanently dead for that booth."""
    c, data = client
    booth = data / "b"
    _touch(booth / "a.png")
    (booth / KEEP_MARKER).mkdir()

    r = c.post("/b/b/unkeep", follow_redirects=False)

    assert r.status_code == 303


def test_a_VALID_but_empty_marks_document_does_NOT_hold(tmp_path):
    """The other side of the blank-file fix, and the regression it would have
    been without this. `_write_raw` writes `{"version": 1, "marks": []}` when
    the last mark is deleted — a valid document with nothing in it. If THAT
    held, every booth the operator ever finished judging would be immortal.

    Blank bytes are damage; an empty list is an answer.
    """
    booth = tmp_path / "b"
    _touch(booth / "a.png")
    _pick(booth)
    delete_mark(booth, "q1")
    assert (booth / ".marks.json").read_text().strip(), "the file is still there"
    _stale(booth)

    assert hold_reason(*hold_read(booth)) is None
    assert sweep_once(tmp_path, ttl_seconds=3600) == ["b"]


# ---- pre-existing defects the bug-hunt panel surfaced ----------------------


def test_marks_json_says_so_when_the_file_is_damaged(tmp_path):
    """Gróa's strongest solo. `booth marks` exits 3 on an unreadable file so a
    caller can tell "not yet" from "broken"; the HTTP mirror — the ONLY reader a
    remote session has — reported the same damage as an empty success. One
    question, two surfaces, two answers, which is the thing U4 exists to stop.

    The status stays 200 on purpose: reads are lenient here, and a pinned
    status is a promise to remote clients this fix has no business breaking.
    `test_a_corrupt_marks_file_gives_the_browser_a_409_not_a_500` pins it.
    """
    app = create_app(tmp_path, ttl_hours=24, start_sweeper=False)
    c = TestClient(app)
    booth = tmp_path / "b"
    _touch(booth / "a.png")

    clean = c.get("/b/b/marks.json").json()
    assert "error" not in clean

    (booth / ".marks.json").write_text("{{{ not json")
    damaged = c.get("/b/b/marks.json")

    assert damaged.status_code == 200
    assert "cannot be read" in damaged.json()["error"]
    assert damaged.json()["open"] == []


def test_an_unreadable_links_file_does_not_take_down_the_booth_page(client):
    """Hulda, pre-existing. `is_file()` then an unguarded `read_text()` was a
    500 waiting on a mode change or an EIO. The board is one tile on that page,
    and a page that will not load is worse than one missing a tile — the same
    posture `read_blurred`, `marks_for` and `read_manifest` already take."""
    c, data = client
    booth = data / "links"
    _touch(booth / "a.png")
    (booth / "links.md").write_text("- [x](http://x/) <sub>· y · 2026-01-01 00:00</sub>\n")
    os.chmod(booth / "links.md", 0)
    try:
        assert c.get("/b/links/").status_code == 200
    finally:
        os.chmod(booth / "links.md", 0o644)


def test_the_index_order_has_a_tie_breaker(tmp_path):
    """CLAUDE.md invariant 6. Sorting on mtime alone left equal-mtime booths in
    whatever order `iterdir()` yielded, which is not a rule — and two booths
    landed by one `rsync` batch share an mtime exactly. The operator refers to
    cards positionally, so a sequence that moves between renders misfiles his
    judgment rather than crashing."""
    when = time.time() - 100
    for n in ("charlie", "alpha", "bravo"):
        _touch(tmp_path / n / "a.png", when=when)
        os.utime(tmp_path / n, (when, when))

    runs = {tuple(b["name"] for b in list_booths(tmp_path, ttl_seconds=86400))
            for _ in range(5)}

    assert len(runs) == 1, "the same state must render the same sequence"
    assert runs.pop() == ("charlie", "bravo", "alpha"), "newest first, then name"


def test_a_booth_name_cannot_reach_a_js_string_context(client):
    """Hulda, solo, pre-existing and live. The confirm dialogs interpolated the
    booth NAME into a JS string literal inside `onsubmit`. Jinja's autoescape is
    HTML-attribute escaping, not JS-string escaping: the browser decodes `&#39;`
    back to `'` before the JS parser sees it, so a booth named `'+alert(1)+'`
    executed on submit. Booth names are agent-authored — making a folder is the
    whole API — so that is a live path.
    """
    c, data = client
    # A canary spelled so that nothing in the repo's prose can collide with it —
    # the first version of this test matched the fix's OWN comment explaining
    # what it fixed, which is a passing test measuring the wrong thing.
    hostile = "'+xssCanary7+'"
    _touch(data / hostile / "a.png")
    _touch(data / "kept-one" / "a.png")
    _touch(data / "kept-one" / KEEP_MARKER)

    html = c.get("/").text

    # The ATTRIBUTE form, not the bare word — the replacement's own comment
    # explains what it replaced and says "onsubmit" while doing so.
    assert "onsubmit=" not in html, "no inline handler may carry a name at all"
    assert "onclick=" not in html

    # The name DOES appear as visible text, correctly escaped, and that is the
    # point: HTML-escaping is the right escaping for an HTML text node. What
    # must not happen is the name reaching a place where the JS parser reads it.
    scripts = re.findall(r"<script\b[^>]*>(.*?)</script>", html, re.S)
    assert scripts, "the page does ship script, so this check is not vacuous"
    for block in scripts:
        assert "xssCanary7" not in block.replace("&#39;", "'"), "the name is in a script"

    assert 'data-confirm="wipe"' in html, "the name travels as data, where escaping is escaping"
    assert ">&#39;+xssCanary7+&#39;<" in html, "and still renders as the name it is"