booth/booth/marks.py

"""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
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,
)

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:
    return datetime.now().astimezone().isoformat(timespec="seconds")


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.
    """
    try:
        raw = json.loads((Path(booth) / MARKS_FILE).read_text(encoding="utf-8"))
    except (OSError, ValueError, UnicodeDecodeError):
        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 _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}
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(doc, ensure_ascii=False, indent=2) + "\n", 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 including dotfiles — 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.
        if not lock.exists():
            lock.touch()
            self._made_lock = True
        self._lf = lock.open("r+")
        fcntl.flock(self._lf, fcntl.LOCK_EX)
        self.entries = _read_raw(self.booth)
        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)
            elif self._made_lock and not (self.booth / MARKS_FILE).exists():
                # Nothing was written and this booth had no marks before: do not
                # leave a lock file behind as the only trace of a no-op.
                (self.booth / MARKS_LOCK).unlink(missing_ok=True)
        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 _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
    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))
        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 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(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 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) -> Mark:
    """A session poses a pick.

    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.
    """
    if not valid_stem(mark_id):
        raise AskError("bad mark id: letters, digits, . _ - only")
    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": existing.get("target") if existing else None,
            "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) 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:
        have = {e.get("id") for e in lk.entries}
        for mtime, stem, decl, err in found:
            if stem in have:
                continue
            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):
                pass
            entry = {
                "id": stem,
                "shape": PICK,
                "target": None,
                "created": datetime.fromtimestamp(mtime).astimezone().isoformat(timespec="seconds"),
                "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.
    return [_hydrate(e) for e in created]