"""Marks — ONE primitive for operator judgment attached to an artifact.

    An ask is the session asking the operator. An annotation is the operator
    telling the session. A vote is the operator pointing at the good ones.

All three are the same thing, and before this module they were three mechanisms:
asks had two JSON sidecars per question and a walk-the-booth read path,
annotations had nothing, and votes had nothing — so the operator picked winners
out of a 270-image set and told the session IN CHAT. `golden-candidates`,
`sindra-finalists` and the `pancake-*` ladders are all that loop, running
through conversation because the session that posted the set had no way to read
the judgment it asked for.

    MARK
      target : the booth, or one item in it (an Item.rel — U1's identity, reused)
      shape  : pick  — one of N options the session declared in advance
               note  — free text the operator volunteered
               flag  — this one
      writer : the operator, in the browser
      reader : the session — `booth marks <booth> [--wait]`

One storage model (`.marks.json`), one read path (`marks_for`), one place
openness is computed (`open_marks`), one rendering slot (beside the artifact).

STDLIB ONLY, like links.py and asks.py: `scripts/booth` imports this under the
system python3 with no venv. See docs/contracts/u2_marks.contract.md, INV-5.

WHY ONE FILE PER BOOTH, and not a sidecar per mark (operator decision,
2026-09-21): U4 makes "does this booth still owe an answer?" a hot question —
the sweep asks it per booth per tick and the index asks it per card per page
load — so it has to be one read, not a walk of a booth that may hold 270 files.
And note the writer roles: a session writes pick declarations, the operator
writes judgments. That is two roles on one file, so the flock below is
load-bearing. It is NOT links.md's problem, though — links.md is an O_APPEND
content-hash log because 17 handles write it concurrently and locking its common
path would serialize them. A booth's marks see one session and one operator, so
locking the common path costs nothing. Same lock, deliberately not the same shape.
"""

from __future__ import annotations

import fcntl
import json
import os
import stat as statmod
from dataclasses import asdict, dataclass, field
from datetime import datetime
from pathlib import Path
from typing import IO, Literal, Sequence

from booth.asks import (
    ANSWER_SUFFIX,
    ASK_SUFFIX,
    AskError,
    NOTES_MAX,
    ask_stem,
    build_answer,
    is_ask_file,
    normalize_ask,
    valid_stem,
)

class MarksCorrupt(RuntimeError):
    """The mark file exists but cannot be parsed, and a WRITE was attempted.

    The read path is deliberately lenient — `marks_for` returns [] so a review
    page still loads. The write path must not inherit that leniency: reading a
    damaged file as "no marks" and then atomically replacing it destroys every
    judgment in the booth from one click, silently. Shipped in v0.2.0 and found
    by a cross-frontier contract panel, not by the suite.

    A page that renders without an annotation is recoverable. A file that
    overwrote the operator's judgment is not.
    """


# A booth's whole judgment lives in one document, so this is generous — a
# 270-item booth flagged throughout, with notes, is far under it. What it rules
# out is the case that is not marks at all: an unbounded read raises MemoryError
# and a deeply nested one raises RecursionError out of `json.loads`, neither of
# which is an OSError or a ValueError, and `list_booths` calls the reader once
# per booth on every index load. Bounded by `stat`, before the bytes are read.
MARKS_MAX_BYTES = 4 * 1024 * 1024

MARKS_FILE = ".marks.json"
MARKS_LOCK = ".marks.lock"
SCHEMA_VERSION = 1

PICK = "pick"
NOTE = "note"
FLAG = "flag"
SHAPES = (PICK, NOTE, FLAG)

TEXT_MAX = NOTES_MAX  # a note is the same kind of text as an ask's notes field

# A target is an Item.rel — a booth-relative POSIX path — or None for the booth
# itself. No second addressing scheme: U1 established `rel` as item identity and
# a mark that invented its own would need a translation layer nobody wants.
_TARGET_MAX = 1024


@dataclass(frozen=True)
class Mark:
    """One piece of operator judgment, with every fact any surface needs.

    Wide and flat on purpose. A nested shape-specific bag would make every
    template navigate it, and the three shapes share more than they differ.
    """

    id: str
    shape: str
    target: str | None
    created: str
    # --- pick: the session's declaration, normalized on READ ---
    declaration: dict | None = None
    prompt: str | None = None
    title: str = ""
    multi: bool = False
    questions: list[dict] = field(default_factory=list)
    options: list[dict] = field(default_factory=list)
    notes_enabled: bool = True
    notes_label: str = "notes"
    # --- the operator's judgment ---
    answer: dict | None = None
    text: str = ""
    flagged: bool = False
    by: str = ""
    error: str | None = None

    @property
    def is_open(self) -> bool:
        """Whether this mark still owes the session an answer. Delegates to the
        module predicate so there is exactly one of them (INV-2)."""
        return _is_open(self)


def now_stamp() -> str:
    """ONE stamp format across every writer in this module.

    MICROSECONDS, matching `import_legacy_asks`. They diverged when the
    importer was moved to sub-second precision to stop same-second sidecars
    re-sorting — and the divergence opened a fresh ordering bug in the other
    direction, because `-` (0x2D) sorts before `.` (0x2E): a whole-second stamp
    lands ahead of ANY fractional stamp in the same second, so a later mark came
    out before an earlier import. Marks sort on `(created, id)`; one format is
    what makes that rule statable.
    """
    return datetime.now().astimezone().isoformat(timespec="microseconds")


def _clean_text(text) -> str:
    return (text or "").replace("\r\n", "\n").strip()[:TEXT_MAX]


def flag_id(target: str) -> str:
    """A flag's id is derived from its target, which is what makes flagging an
    UPSERT: one item has at most one flag state, so there is nothing to
    accumulate. Unflagging removes the mark rather than storing `false` — an
    absent flag and a false flag are the same judgment, and two representations
    of one state is how `.forever` became a problem."""
    return f"flag:{target}"


def _note_id(existing: set[str]) -> str:
    """A note gets a generated id because an item may carry several."""
    n = 1
    while f"note-{n}" in existing:
        n += 1
    return f"note-{n}"


def _valid_target(target) -> bool:
    if target is None:
        return True
    if not isinstance(target, str) or not target or len(target) > _TARGET_MAX:
        return False
    # A target names a file inside the booth. Absolute paths and traversal are
    # not "unlikely", they are the first thing a fuzzer tries.
    if target.startswith("/") or ".." in Path(target).parts:
        return False
    return True


# ---- storage ----------------------------------------------------------------


def _read_raw(booth: Path) -> list[dict]:
    """The stored mark entries, or [] for missing/corrupt.

    A booth with no marks and a booth whose mark file is truncated both render
    as "no marks", and neither is a 500 — the same posture `read_blurred` takes,
    for the same reason: a review surface that will not load is worse than one
    that has lost an annotation.
    """
    path = Path(booth) / MARKS_FILE
    try:
        st = path.stat()
        # Regular-file first, then size. `st_size` is 0 for a FIFO and 0 for a
        # symlink to /dev/zero, so both pass a byte cap and then `read_text`
        # either blocks with no EOF or allocates until the kernel intervenes.
        # This loop runs over EVERY booth on every index load.
        if not statmod.S_ISREG(st.st_mode) or st.st_size > MARKS_MAX_BYTES:
            return []
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError, UnicodeDecodeError, RecursionError, MemoryError):
        return []
    if not isinstance(raw, dict):
        return []
    entries = raw.get("marks")
    if not isinstance(entries, list):
        return []
    return [e for e in entries if isinstance(e, dict) and isinstance(e.get("id"), str)]


def _fingerprint(entries: list[dict]) -> str:
    """A stable serialization used ONLY to decide whether a write is a no-op."""
    return json.dumps(entries, sort_keys=True, ensure_ascii=False)


def _read_raw_strict(booth: Path, *, blank_is_corrupt: bool = False) -> list[dict]:
    """Like `_read_raw`, but RAISES `MarksCorrupt` on a file it cannot parse.

    Absent, empty and valid-but-empty are all "no marks yet" and are fine — the
    distinction that matters is bytes-present-but-unreadable, because that is the
    case where writing would destroy something.

    `blank_is_corrupt` is the DELETE path's reading of a present-but-whitespace
    file, and only the delete path's: this 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. The write path keeps the lenient
    reading — a blank file is safe to overwrite, which is the question
    `_Locked` is asking. A VALID document with an empty `marks` list is not
    blank and never holds: that is what deleting the last mark leaves behind,
    and it must stay sweepable.
    """
    path = Path(booth) / MARKS_FILE
    try:
        st = path.stat()
    except FileNotFoundError:
        return []
    except OSError as exc:
        raise MarksCorrupt(f"{path} cannot be read: {exc}") from exc
    # The strict half has to refuse everything the lenient half tolerates, or a
    # file that reads as "no marks" gets replaced by a write that believed it.
    if not statmod.S_ISREG(st.st_mode):
        raise MarksCorrupt(f"{path} is not a regular file")
    if st.st_size > MARKS_MAX_BYTES:
        raise MarksCorrupt(
            f"{path} is too large to be a marks document ({st.st_size} bytes)")
    try:
        text = path.read_text(encoding="utf-8")
    except FileNotFoundError:
        return []
    except (OSError, UnicodeDecodeError, MemoryError) as exc:
        raise MarksCorrupt(f"{path} cannot be read: {exc}") from exc
    if not text.strip():
        if blank_is_corrupt:
            raise MarksCorrupt(f"{path} is present but holds no marks document")
        return []
    try:
        raw = json.loads(text)
    except (ValueError, RecursionError, MemoryError) as exc:
        raise MarksCorrupt(
            f"{path} is not valid JSON: {type(exc).__name__}") from exc
    if not isinstance(raw, dict) or not isinstance(raw.get("marks"), list):
        raise MarksCorrupt(f"{path} is not a marks document")
    entries = [e for e in raw["marks"] if isinstance(e, dict) and isinstance(e.get("id"), str)]
    if len(entries) != len(raw["marks"]):
        raise MarksCorrupt(f"{path} holds entries this version cannot read")
    return entries


def read_error(booth: Path) -> str | None:
    """Why this booth's marks cannot be read, or None if they can.

    `marks_for` is lenient on purpose — a review page that will not load is
    worse than one missing an annotation — and that leniency turns an
    unreadable file into "no marks". For a BROWSER that is the right trade. For
    the CLI it is not: a session that asked a question and is told "no such
    pick" will conclude the question was never posted, when in fact the file
    holding it is damaged. A machine consumer can act on the difference, so it
    gets to ask.
    """
    try:
        _read_raw_strict(booth)
    except MarksCorrupt as exc:
        return str(exc)
    return None


def _write_raw(booth: Path, entries: list[dict]) -> None:
    """Atomic replace, so a reader never sees a half-written document and a
    crash mid-write cannot truncate the file into a shorter — and therefore
    quieter — set of marks."""
    path = Path(booth) / MARKS_FILE
    doc = {"version": SCHEMA_VERSION, "marks": entries}
    body = json.dumps(doc, ensure_ascii=False, indent=2) + "\n"
    # The read bound is on the STORED bytes and `indent=2` grows them, so a
    # document that fits in memory can land over the limit on disk and then read
    # back as no marks at all. Refuse loudly instead: a write that fails is
    # recoverable, a file that silently empties is not.
    if len(body.encode("utf-8")) > MARKS_MAX_BYTES:
        raise MarksCorrupt(
            f"{path} would be larger than this version can read back "
            f"({len(body.encode('utf-8'))} bytes)")
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(body, encoding="utf-8")
    os.replace(tmp, path)


class _Locked:
    """Exclusive flock held across the whole read-modify-write.

    The lock lives on a sidecar dotfile rather than on `.marks.json` itself,
    because the write path replaces that file — flock follows the inode, so
    locking a file you are about to os.replace protects nothing after the swap.
    Same reason `links.py` locks `.links.lock`.
    """

    def __init__(self, booth: Path):
        self.booth = Path(booth)
        self.entries: list[dict] = []
        self._lf: IO[str] | None = None
        self._before: str = ""
        self._made_lock = False

    def __enter__(self) -> "_Locked":
        self.booth.mkdir(parents=True, exist_ok=True)
        lock = self.booth / MARKS_LOCK
        # `touch(exist_ok=True)` on an EXISTING file bumps its mtime, and a
        # booth's TTL is measured from its newest mtime — so an unconditional
        # touch would keep a booth alive just for being read through a write
        # path. Create it only when it is not there.
        #
        # ONCE CREATED, THE LOCK FILE IS NEVER REMOVED (see __exit__).
        if not lock.exists():
            # Creating a directory entry bumps the DIRECTORY's mtime, which is
            # what `_newest_mtime` reads. An earlier version put the clock back
            # with `os.utime` — which closed the bug and opened a race: the
            # restore ran before the flock, so anything landing in the window
            # between the stat and the utime had its bump rolled backward. An
            # `rsync -a` batch is the case that bites, because it PRESERVES
            # source mtimes and so has only the directory's freshness to look
            # alive by. It could also raise OSError on a read-only directory
            # and take the route down with it.
            #
            # THE RESTORE STAYS, and the honest reason is that the alternative
            # was worse. Ignoring a booth directory's own mtime whenever the
            # booth holds anything would close the race outright — and would
            # also silently retire the documented behaviour that RELEASING a
            # kept board resets its clock, which the CLI header, the README and
            # a deliberate test all pin. That is a TTL doctrine change, not a
            # bug fix, and it does not belong in one.
            #
            # ⚠ RESIDUAL RACE, stated rather than papered over: between the stat
            # and the utime, another writer's directory-entry change can be
            # rolled backward. The case that bites is an `rsync -a` batch, which
            # preserves source mtimes and so has only the directory's freshness
            # to look alive by. The window is the two syscalls below and the
            # booth must also be one being written to at that instant.
            #
            # The concrete half IS fixed: a failing utime (read-only directory,
            # a booth whose owner we are not) used to escape and take the whole
            # route down with a 500. Not putting the clock back is a cost this
            # module can absorb; not answering the request is not.
            before = self.booth.stat()
            lock.touch()
            self._made_lock = True
            try:
                os.utime(self.booth, (before.st_atime, before.st_mtime))
            except OSError:
                pass
        self._lf = lock.open("r+")
        fcntl.flock(self._lf, fcntl.LOCK_EX)
        try:
            # STRICT here, lenient in marks_for — see MarksCorrupt.
            self.entries = _read_raw_strict(self.booth)
        except MarksCorrupt:
            fcntl.flock(self._lf, fcntl.LOCK_UN)
            self._lf.close()
            self._lf = None
            raise
        self._before = _fingerprint(self.entries)
        return self

    def __exit__(self, exc_type, exc, tb) -> Literal[False]:
        """Never suppresses. The annotation is `Literal[False]` rather than
        `bool` on purpose: a `bool` tells a type checker this manager MIGHT
        swallow an exception, and a swallowed write error would report success
        on a mark that never reached disk."""
        lf = self._lf
        assert lf is not None, "__exit__ without __enter__"
        try:
            # Write only if something actually changed. Marking IS activity and
            # SHOULD reset the booth's TTL — but a write that changes nothing is
            # not activity, and unflagging something that was never flagged
            # would otherwise keep a dead booth alive forever.
            if exc_type is None and _fingerprint(self.entries) != self._before:
                _write_raw(self.booth, self.entries)
            # THE LOCK FILE IS NEVER UNLINKED. It used to be, on the no-op path,
            # so a booth that had never been marked was left exactly as it was
            # found. That tidiness cost mutual exclusion outright: `flock` binds
            # to an INODE, so unlinking the lock while a second writer is blocked
            # on it leaves that writer holding an exclusive lock on a deleted
            # file, and the NEXT writer creates a fresh lock and takes it at
            # once. Two processes then run the read-modify-write concurrently,
            # the later `os.replace` drops the earlier one's mark, and both of
            # them obeyed the protocol. A zero-byte dotfile is the cheaper
            # thing to leave behind — `booth_items` skips it, the zip skips it,
            # and `_newest_mtime` exempts it so it cannot hold a booth open.
        finally:
            fcntl.flock(lf, fcntl.LOCK_UN)
            lf.close()
            self._lf = None
        return False

    def find(self, mark_id: str) -> dict | None:
        return next((e for e in self.entries if e.get("id") == mark_id), None)


# ---- read -------------------------------------------------------------------


def _entry_type_error(entry: dict) -> str | None:
    """The stored scalars this module refuses to guess at.

    `_clean_text` did `(text or "").replace(...)` and `marks_for` sorts on
    `(created, id)` — so a stored `text` that is a dict, or a `created` that is a
    number, raised AttributeError or TypeError 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 and `/healthz` does the same, so one hand-edited or
    foreign-written file took down the front page for every booth on the
    service. A wrong type is a broken mark, and this module already knows how to
    render one of those.
    """
    for name in ("created", "by", "text", "error"):
        value = entry.get(name)
        if value is not None and not isinstance(value, str):
            return f"{name} is {type(value).__name__}, not a string"
    return None


def _hydrate(entry: dict) -> Mark:
    """One stored entry -> one Mark, declarations normalized.

    A pick's declaration is stored RAW and normalized here, exactly as
    `write_ask` + `load_ask` did: validated at write, re-read at render, so a
    declaration that went bad on disk surfaces as `error` instead of being
    unrepresentable. A broken question the session believes it posted has to be
    visible — silently hiding it is the one outcome nobody can debug.
    """
    mid = entry["id"]
    shape = entry.get("shape") if entry.get("shape") in SHAPES else NOTE
    bad = _entry_type_error(entry)
    if bad is not None:
        # `created` is dropped rather than coerced, which sorts the entry to the
        # TOP of the booth's marks: a mark nobody can read is the one that wants
        # looking at, and burying it under 270 items' worth of notes is how it
        # stays unnoticed. Deterministic, and stated — `("", id)` against
        # `(created, id)`.
        return Mark(id=mid, shape=shape, target=None, created="",
                    error=f"unreadable mark: {bad}")
    target = entry.get("target")
    if not _valid_target(target):
        target = None
    base = {
        "id": mid,
        "shape": shape,
        "target": target,
        "created": entry.get("created") or "",
        "by": entry.get("by") or "",
    }

    if shape == PICK:
        decl = entry.get("declaration")
        answer = entry.get("answer") if isinstance(entry.get("answer"), dict) else None
        stored_error = entry.get("error")
        if isinstance(stored_error, str) and stored_error:
            # A reason recorded by whoever wrote the entry — the legacy importer
            # discovers these, and the reason has to survive to the page or a
            # question the session believes it posted disappears silently.
            return Mark(**base, declaration=decl if isinstance(decl, dict) else None,
                        answer=answer, error=stored_error)
        if not isinstance(decl, dict):
            return Mark(**base, declaration=None, answer=answer,
                        error="pick has no declaration")
        try:
            norm = normalize_ask(decl, mid)
        except AskError as exc:
            return Mark(**base, declaration=decl, answer=answer, error=str(exc))
        # THE ANSWER'S SHAPE IS VALIDATED HERE, at the ONE boundary every
        # surface crosses — not at the three render sites that happen to draw
        # it today, and not defensively in the template, which would hide that
        # anything is wrong.
        #
        # `{"answer": {"answers": [], "notes": ""}}` is well-formed JSON with a
        # wrong-shaped value. It passed `_entry_type_error`, passed the
        # `isinstance(answer, dict)` check above, and `marks_for` and
        # `hold_read` both reported the mark HEALTHY with no read error — and
        # then `_ask_inline.html` did `a.answer.answers.get(q.key)`, Jinja asked
        # a LIST for `.get`, and the gallery page and the marks page returned
        # 500. Measured at 42ea67f, so it predates U3; U3 guarded only its own
        # surface with `_safe_fragments` and left these two by scope.
        #
        # This is the v0.2.2 lesson finished rather than half-done. That outage
        # was a file that could not be PARSED and the reader was made lenient;
        # this one parses perfectly and breaks one layer further in, at render,
        # where no leniency exists. `read_error` was answering a narrower
        # question than every caller assumed.
        #
        # ONLY the multi case is checked, because only the multi case indexes:
        # a single-question pick's answer IS the record, with no `answers` key
        # to get wrong. Requiring one unconditionally would break every single
        # pick, which is the direction a too-eager guard fails in.
        if norm["multi"] and isinstance(answer, dict) and \
                not isinstance(answer.get("answers"), dict):
            return Mark(**base, declaration=decl, answer=None,
                        prompt=norm["prompt"], title=norm["title"],
                        multi=norm["multi"], questions=norm["questions"],
                        options=norm.get("options", []),
                        notes_enabled=norm["notes"], notes_label=norm["notes_label"],
                        error="this pick's answer is stored in a shape the page "
                              "cannot render; the answer was dropped and the "
                              "question is unanswered")
        return Mark(
            **base,
            declaration=decl,
            prompt=norm["prompt"],
            title=norm["title"],
            multi=norm["multi"],
            questions=norm["questions"],
            options=norm.get("options", []),
            notes_enabled=norm["notes"],      # normalize_ask emits it as `notes`
            notes_label=norm["notes_label"],
            answer=answer,
        )

    if shape == FLAG:
        return Mark(**base, flagged=True)

    return Mark(**base, text=_clean_text(entry.get("text")))


def _hydrate_safe(entry: dict) -> Mark:
    """`_hydrate`, with the promise that it cannot raise.

    `_entry_type_error` covers the shapes we know how to name; this is the
    backstop for the ones we do not, and it exists because of WHERE this runs.
    One unreadable mark must cost that mark, never the page — and on the index
    it is not even that booth's page, it is all of them.
    """
    try:
        return _hydrate(entry)
    except Exception as exc:                       # noqa: BLE001 - deliberate
        return Mark(id=str(entry.get("id", "")), shape=NOTE, target=None,
                    created="", error=f"unreadable mark: {exc}")


def marks_for(booth: Path) -> list[Mark]:
    """Every mark in a booth, oldest first, declarations normalized and answers
    folded in. ONE file read — which is the whole point of the storage shape."""
    entries = _read_raw(booth)
    marks = [_hydrate_safe(e) for e in entries]
    # (created, id) rather than created alone: two marks written in the same
    # second would otherwise order by however json listed them.
    marks.sort(key=lambda m: (m.created, m.id))
    return marks


def _is_open(mark: Mark) -> bool:
    """THE openness predicate. Nothing else may spell this out.

    A partially-answered pick is STILL OPEN. Today's index badge tests
    `answer is None` and so calls a half-answered four-question pick closed,
    while the panel beside it renders that same pick `◐ partial` — the two
    disagree about one booth. Open is the reading that makes U4 correct: a
    lifetime rule that unpinned a booth on the first radio click would sweep a
    review in flight.
    """
    if mark.shape != PICK or mark.error is not None:
        return False
    if mark.answer is None:
        return True
    return not mark.answer.get("complete", False)


def open_marks(marks: Sequence[Mark]) -> list[Mark]:
    """The marks still owed an answer. The index badge, the booth header, the
    panel filter and U4's pin rule all call this rather than re-deriving it."""
    return [m for m in marks if _is_open(m)]


def hold_read(booth: Path) -> tuple[list[Mark], str | None]:
    """ONE read of `.marks.json`, answering both questions the LIFETIME rule asks:
    what is still open, and whether the file could be read at all.

    U4 decides whether a booth may be SWEPT from those two facts. Asking them
    with two calls — `marks_for` then `read_error` — reads the file twice, and
    two reads of one file are not one read of one state: a write or a repair
    landing between them yields a pair that never described the booth at any
    instant. The losing pair is `([], None)` — no marks, no error — which is
    exactly the one that deletes. Cross-frontier review (2026-09-22) found it;
    that is why this exists rather than the obvious two calls.

    When the file reads clean the marks are byte-identical to `marks_for`'s:
    `_read_raw_strict` raises rather than dropping an entry, so a non-raising
    strict read returns the same entries the lenient read would, hydrated and
    sorted the same way. The caller can therefore use this ONE read for the
    display too, and fall back to `marks_for` only on the error path, where
    leniency is the point.
    """
    try:
        entries = _read_raw_strict(booth, blank_is_corrupt=True)
    except MarksCorrupt as exc:
        return [], str(exc)
    marks = [_hydrate_safe(e) for e in entries]
    marks.sort(key=lambda m: (m.created, m.id))
    return marks, None


def marks_for_target(marks: Sequence[Mark], rel: str | None) -> list[Mark]:
    """The marks attached to one item, or to the booth itself for None."""
    return [m for m in marks if m.target == rel]


def as_dict(mark: Mark) -> dict:
    """The JSON boundary — `booth marks` output. Python consumers take the
    dataclass, and so does Jinja (every template accesses marks by attribute);
    this exists so the CLI has one serialization instead of one per verb."""
    return asdict(mark)


# ---- write ------------------------------------------------------------------


def declare_pick(booth: Path, mark_id: str, doc: dict, target: str | None = None) -> Mark:
    """A session poses a pick, about the booth or about ONE item in it.

    Validated through `normalize_ask` BEFORE anything is written, so a session
    cannot land a question the renderer would refuse. Re-declaring an existing
    id replaces the declaration and CLEARS its answer: the question changed, so
    the old judgment is not an answer to it — and it may move the target, since
    a re-declaration is a new question.

    `target` is an `Item.rel`, or None for the booth. It exists because the
    2026-09-09 ruling is that a question belongs WITH the artifact it is about: a
    four-voice audition wants the radio group under that voice. The record and
    the renderer both supported it before this parameter did, which meant a
    session could not actually produce one.
    """
    if not valid_stem(mark_id):
        raise AskError("bad mark id: letters, digits, . _ - only")
    if not _valid_target(target):
        raise AskError("a pick's target must be a path inside the booth")
    normalize_ask(doc, mark_id)  # raises AskError; nothing written yet
    with _Locked(booth) as lk:
        existing = lk.find(mark_id)
        if existing is not None and existing.get("shape") != PICK:
            raise AskError(f"{mark_id!r} is already a {existing.get('shape')}")
        entry = {
            "id": mark_id,
            "shape": PICK,
            "target": target,
            "created": existing.get("created") if existing else now_stamp(),
            "declaration": doc,
            "answer": None,
        }
        if existing is None:
            lk.entries.append(entry)
        else:
            lk.entries[lk.entries.index(existing)] = entry
        return _hydrate(entry)


def answer_pick(booth: Path, mark_id: str, choice, notes: str = "", who: str = "",
                qnotes: dict | None = None) -> Mark:
    """Record the operator's pick. Every semantic belongs to
    `asks.build_answer`; this function owns storage and nothing else.

    Re-answering overwrites — the mark is the CURRENT judgment, not a log.

    The AskError for an id that names no live pick is raised HERE. It used to
    come from `load_ask` inside `write_answer`; extracting the answer-builder
    moved the path to this caller, and a stale form POST has to be a 400 rather
    than a silent no-op.
    """
    with _Locked(booth) as lk:
        entry = lk.find(mark_id)
        if entry is None or entry.get("shape") != PICK:
            raise AskError("no such pick")
        decl = entry.get("declaration")
        if not isinstance(decl, dict):
            raise AskError("pick has no declaration")
        ask = normalize_ask(decl, mark_id)  # raises AskError on a bad declaration
        entry["answer"] = build_answer(ask, choice, notes, who, qnotes)
        return _hydrate(entry)


def write_note(booth: Path, target: str | None, text: str, who: str = "") -> Mark:
    """Attach free text to an item, or to the booth itself.

    The operator telling the session — the direction that had no mechanism at
    all before this, which is why the loop ran through chat. Several notes per
    target are legal (a review makes more than one remark about one image), so
    each gets a generated id rather than upserting like a flag.

    Empty text after cleaning is refused, the same posture an empty pick
    submission takes: recording a mark that says nothing is strictly worse for
    the reading session than recording no mark.
    """
    if not _valid_target(target):
        raise AskError("a note's target must be a path inside the booth")
    body = _clean_text(text)
    if not body:
        raise AskError("nothing to record — the note is empty")
    with _Locked(booth) as lk:
        entry = {
            "id": _note_id({e.get("id") for e in lk.entries}),
            "shape": NOTE,
            "target": target,
            "created": now_stamp(),
            "text": body,
            "by": who or "",
        }
        lk.entries.append(entry)
        return _hydrate(entry)


def set_flag(booth: Path, target: str, on: bool, who: str = "") -> Mark | None:
    """Flag or unflag one item — the operator pointing at the good ones.

    This is the shape that makes a 270-image booth tractable, and the one that
    closes a loop currently running through conversation: `golden-candidates`,
    `sindra-finalists` and the `pancake-*` ladders are all the operator
    selecting winners and then telling the session by hand.

    UPSERT keyed by target (see `flag_id`): flagging twice is idempotent, and
    unflagging REMOVES the mark and returns None rather than storing a false.
    Unflagging something that was never flagged is not an error — it is the
    state the caller asked for.
    """
    if not _valid_target(target) or target is None:
        raise AskError("a flag's target must be a path inside the booth")
    mid = flag_id(target)
    with _Locked(booth) as lk:
        entry = lk.find(mid)
        if not on:
            if entry is not None:
                lk.entries.remove(entry)
            return None
        if entry is not None:
            return _hydrate(entry)  # already flagged; nothing to change
        entry = {
            "id": mid,
            "shape": FLAG,
            "target": target,
            "created": now_stamp(),
            "by": who or "",
        }
        lk.entries.append(entry)
        return _hydrate(entry)


def delete_mark(booth: Path, mark_id: str) -> bool:
    """Remove one mark by id — the operator's undo. True if it was there.

    Deleting a mark is the operator withdrawing a judgment, which is his to do.
    Nothing else in this module deletes anything.
    """
    with _Locked(booth) as lk:
        entry = lk.find(mark_id)
        if entry is None:
            return False
        lk.entries.remove(entry)
        return True


# ---- migration --------------------------------------------------------------


def import_legacy_asks(booth: Path) -> list[Mark]:
    """Read every `*.ask.json` / `*.answer.json` in a booth into `.marks.json`.

    EXPLICIT AND ONE-SHOT, not lazy. A read that writes would fire on every
    index page load for every booth, which is the wrong trade for the four
    sidecars that exist on the live service.

    IDEMPOTENT: an id already present as a mark is skipped outright, so a second
    run is a no-op and a judgment recorded since the first run is never
    clobbered by the older sidecar.

    NOTHING IS DELETED. The sidecars stay on disk — the ROADMAP's non-goal is
    explicit about it, and `booth_items` already excludes them from the tile
    list, so an imported-but-kept sidecar does not show up as a file.

    Returns the marks it created, oldest first. `created` is seeded from the
    sidecar's MTIME rather than from now(): `list_asks` ordered by mtime and
    `marks_for` orders by `created`, so seeding from now() would silently
    reshuffle a booth's questions at the moment of migration.
    """
    booth = Path(booth)
    if not booth.is_dir():
        return []

    found: list[tuple[float, str, dict | None, str | None]] = []
    for p in sorted(booth.iterdir()):
        if not p.is_file() or p.name.startswith(".") or not is_ask_file(p.name):
            continue
        stem = ask_stem(p.name)
        if not valid_stem(stem):
            # Could not have been written by `booth ask`. Left alone rather than
            # imported under an id nothing could address.
            continue
        # A sidecar that cannot be read is imported WITH ITS REASON rather than
        # skipped. `list_asks` surfaced these as `⚠ broken` on the page, and
        # dropping them on migration would turn a visible broken question into a
        # question that was never there — found by a retargeted test, which is
        # the whole argument for retargeting them instead of deleting them.
        try:
            mtime = p.stat().st_mtime
        except OSError:
            continue
        try:
            decl = json.loads(p.read_text(encoding="utf-8"))
        except (OSError, ValueError, UnicodeDecodeError,
                RecursionError, MemoryError) as exc:
            found.append((mtime, stem, None, f"unreadable ask: {exc}"))
            continue
        if not isinstance(decl, dict):
            found.append((mtime, stem, None, "ask must be a JSON object"))
            continue
        found.append((mtime, stem, decl, None))

    if not found:
        return []
    found.sort(key=lambda t: (t[0], t[1]))  # mtime, then name — never compares payloads

    created: list[dict] = []
    with _Locked(booth) as lk:
        by_id = {e.get("id"): e for e in lk.entries}
        for mtime, stem, decl, err in found:
            answer = None
            ap = booth / f"{stem}{ANSWER_SUFFIX}"
            try:
                loaded = json.loads(ap.read_text(encoding="utf-8"))
                if isinstance(loaded, dict):
                    answer = loaded
            except (OSError, ValueError, UnicodeDecodeError,
                    RecursionError, MemoryError):
                pass

            prior = by_id.get(stem)
            if prior is not None:
                # The stem is already a mark, so the DECLARATION is not imported
                # — that is the idempotence rule, and a mark declared since the
                # sidecar outranks it. But a legacy ANSWER must not be stranded:
                # if the existing mark is an unanswered pick and the sidecar
                # holds the operator's choice, adopt it. Ordinary reads are
                # forbidden from looking at sidecars, so a skip here would lose
                # that judgment permanently.
                if (answer is not None
                        and prior.get("shape") == PICK
                        and prior.get("answer") is None):
                    prior["answer"] = answer
                    created.append(prior)
                continue

            entry = {
                "id": stem,
                "shape": PICK,
                "target": None,
                # MICROSECONDS, not seconds. `found` is ordered by fractional
                # mtime and `marks_for` re-sorts on this string, so truncating
                # to the whole second threw away the only thing distinguishing
                # two sidecars written in the same second — and the `(created,
                # id)` tie-break then silently re-sorted them alphabetically,
                # reversing the order the importer had just established. The
                # ROADMAP states this import's order is `(mtime, name)`; an
                # order that is stated and not kept is worse than one never
                # claimed.
                "created": datetime.fromtimestamp(mtime).astimezone().isoformat(
                    timespec="microseconds"),
                "declaration": decl,
                "answer": answer,
            }
            if err:
                entry["error"] = err
            lk.entries.append(entry)
            created.append(entry)

    # Hydrated AFTER the lock so a broken declaration surfaces as `error` here
    # exactly as it does on a normal read, rather than through a second path.
    # `_hydrate_safe`, not `_hydrate`: this is the one path that reads entries
    # it did not write, and it was the one without the guard.
    return [_hydrate_safe(e) for e in created]