vh/booth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(marks): one primitive for operator judgment, so the loop stops running through chat

Browse Source 
Five mechanisms existed to get one question next to one artifact. Three of
them were the same thing wearing different clothes, and the third of the three
had no code at all: the operator picked winners out of a 270-image set and
told the session in conversation. `sindra-finalists` is 86 items, every one
captioned, with the selection encoded in the booth's NAME.

A MARK is operator judgment attached to a target — the booth, or one item in
it, addressed by the `rel` U1 established as item identity. Three shapes:

  pick — one of N options a session declared in advance   (was: an ask)
  note — free text the operator volunteered               (had nothing)
  flag — this one                                         (had nothing)

One file per booth, one read path, one place openness is computed, one slot
beside the artifact. The storage shape is the operator's call (2026-09-21) and
follows from U4: "does this booth still owe an answer?" gets asked per booth
per sweep tick and per card per index render, so it has to be one read and not
a walk of a booth holding 270 files. Marks are also not links.md — that is an
O_APPEND content-hash log because 17 handles write it concurrently, whereas a
booth's marks see one session and one operator, so locking the common path
costs nothing.

The 2026-09-09 pick semantics are preserved by NOT rewriting them: partial
answers legal, a blank question lands in `unanswered`, `complete` false until
every question has a pick, the only refusal a submission carrying nothing.
`write_answer` split into the pure `build_answer` plus the storage that went
away with the sidecar; `normalize_ask` untouched.

Three findings worth naming, because each was caught by a gate rather than by
reading the diff again:

  * The seam review found `inline.place` indexes asks by SUBSCRIPT — the only
    consumer in the service that does — so a frozen dataclass breaks it, and
    `inline.py` had been missing from the contract's scope entirely.
  * A retargeted test found a regression in the legacy importer: a malformed
    sidecar that renders "broken" today would have silently vanished on
    migration. It now imports carrying its reason.
  * A partially-answered pick counted as CLOSED on the index while the panel
    beside it rendered it "partial" — the two disagreed about one booth. Open
    is the reading U4 needs, and it is declared rather than smuggled in.

`GET /b/<n>/marks.json` is new and load-bearing: sessions on other hosts polled
`<stem>.answer.json` over HTTP, so removing the sidecar without it would have
taken that capability away. `/b/<n>/asks` 308s to `/marks`. Legacy sidecars are
imported, never deleted — four are live and unanswered.

Also records the operator's deterministic-order directive as a cross-cutting v1
invariant, in ROADMAP.md with the per-collection rule table and as CLAUDE.md
invariant 6. The Booth's job is comparison; an order that moves between renders
does not crash, it misfiles the judgment.

242 tests. No version bump — a release tier for this is the operator's call.
This commit is contained in:
vh
2026-09-21 23:38:27 -07:00
parent 9272c9872e
commit c7f9437a64
23 changed files with 2677 additions and 650 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_asks.py
+149 -90
View File
@@ -1,5 +1,12 @@
"""Asks: session poses a multiple-choice question; operator answers in the
browser; the answer lands as a sidecar the session reads."""
"""Picks: a session poses a multiple-choice question; the operator answers it in
the browser; the answer lands where the session reads it.
A `pick` is one shape of MARK (see tests/test_marks.py and booth/marks.py). What
stays here is what did not move: `normalize_ask`, the declaration validator, and
`build_answer`, which together carry the operator-settled 2026-09-09 semantics —
plus the page and route integration, retargeted from the two-sidecars-per-question
storage that marks replaced.
"""
import json
import pathlib
@@ -11,16 +18,31 @@ from booth.asks import (
ANSWER_SUFFIX,
ASK_SUFFIX,
AskError,
list_asks,
load_ask,
build_answer,
normalize_ask,
read_answer,
write_answer,
write_ask,
)
from booth.marks import (
answer_pick,
declare_pick,
import_legacy_asks,
marks_for,
open_marks,
)
def _ask(booth, stem="winner", **kw):
"""Declare a pick, the way a session does now."""
doc = {"prompt": "Which render wins?", "options": ["A — baseline", "B — async"]}
doc.update(kw)
booth.mkdir(parents=True, exist_ok=True)
declare_pick(booth, stem, doc)
return booth
def _sidecar(booth, stem="winner", **kw):
"""Write a LEGACY `<stem>.ask.json`. Only for the tests that are about the
legacy files themselves — they are still excluded from the item list, and
still importable."""
doc = {"prompt": "Which render wins?", "options": ["A — baseline", "B — async"]}
doc.update(kw)
booth.mkdir(parents=True, exist_ok=True)
@@ -76,72 +98,79 @@ def test_normalize_rejects(doc):
# ---- files ------------------------------------------------------------------
def test_load_ask_reports_bad_json(tmp_path):
(tmp_path / f"x{ASK_SUFFIX}").write_text("{not json")
def test_a_malformed_declaration_is_refused(tmp_path):
"""The unreadable-FILE case moved to the legacy importer, which surfaces it
as a broken mark rather than raising — see test_marks.py. What is left here
is the declaration itself being wrong, which is refused at the door."""
with pytest.raises(AskError):
load_ask(tmp_path, "x")
normalize_ask("not an object", "x")
with pytest.raises(AskError):
load_ask(tmp_path, "missing")
declare_pick(tmp_path, "x", {"prompt": "no options"})
def test_list_asks_folds_answer_and_surfaces_errors(tmp_path):
def test_marks_for_folds_answer_and_surfaces_errors(tmp_path):
_ask(tmp_path, "one")
_ask(tmp_path, "two")
(tmp_path / f"broken{ASK_SUFFIX}").write_text("[]")
(tmp_path / ".hidden.ask.json").write_text("{}") # dotfiles never listed
write_answer(tmp_path, "two", "B — async", "less banding", who="10.0.0.1")
answer_pick(tmp_path, "two", "B — async", "less banding", who="10.0.0.1")
asks = list_asks(tmp_path)
by = {a["stem"]: a for a in asks}
assert set(by) == {"one", "two", "broken"}
assert by["one"]["answer"] is None and by["one"]["error"] is None
assert by["two"]["answer"]["choice"] == "B — async"
assert by["two"]["answer"]["choice_index"] == 1
assert by["two"]["answer"]["notes"] == "less banding"
assert by["two"]["answer"]["answered_by"] == "10.0.0.1"
assert by["broken"]["error"] and by["broken"]["options"] == []
by = {m.id: m for m in marks_for(tmp_path)}
assert set(by) == {"one", "two"} # the broken SIDECAR is not a mark
assert by["one"].answer is None and by["one"].error is None
assert by["two"].answer["choice"] == "B — async"
assert by["two"].answer["choice_index"] == 1
assert by["two"].answer["notes"] == "less banding"
assert by["two"].answer["answered_by"] == "10.0.0.1"
# The broken legacy sidecar is not silently swallowed either — importing it
# lands a mark carrying the error, so a question the session believes it
# posted stays visible instead of vanishing.
import_legacy_asks(tmp_path)
broken = next(m for m in marks_for(tmp_path) if m.id == "broken")
assert broken.error and broken.options == []
def test_write_answer_validates_choice_and_is_atomic(tmp_path):
_ask(tmp_path)
def test_answer_validates_the_choice_that_was_made(tmp_path):
_sidecar(tmp_path)
with pytest.raises(AskError):
write_answer(tmp_path, "winner", "C — nope")
_shape("C — nope")
with pytest.raises(AskError):
write_answer(tmp_path, "nosuch", "A — baseline")
ans = write_answer(tmp_path, "winner", "A — baseline", " ok \r\n")
_shape("A — baseline", doc={"prompt": "p", "options": ["only"]})
ans = _shape("A — baseline", notes=" ok \r\n")
assert ans["notes"] == "ok"
assert ans["answered_at"]
assert read_answer(tmp_path, "winner") == ans
assert not (tmp_path / f"winner{ANSWER_SUFFIX}.tmp").exists()
# re-answer overwrites — the sidecar is the CURRENT answer, not a log
write_answer(tmp_path, "winner", "B — async")
assert read_answer(tmp_path, "winner")["choice_index"] == 1
assert _shape("B — async")["choice_index"] == 1
def test_write_answer_drops_notes_when_ask_disables_them(tmp_path):
_ask(tmp_path, notes=False)
assert write_answer(tmp_path, "winner", "A — baseline", "ignored")["notes"] == ""
def test_answer_drops_notes_when_the_pick_disables_them(tmp_path):
_sidecar(tmp_path, notes=False)
assert _shape("A — baseline", doc=dict(SINGLE, notes=False), notes="ignored")["notes"] == ""
def test_write_ask_roundtrip_and_stem_guard(tmp_path):
p = write_ask(tmp_path / "b", "pick", "Pick one", ["x", {"id": "y", "label": "Y"}], notes=False)
assert p.name == f"pick{ASK_SUFFIX}"
a = load_ask(tmp_path / "b", "pick")
assert [o["id"] for o in a["options"]] == ["x", "y"] and a["notes"] is False
def test_declare_pick_roundtrip_and_id_guard(tmp_path):
declare_pick(tmp_path / "b", "pick",
{"prompt": "Pick one", "options": ["x", {"id": "y", "label": "Y"}], "notes": False})
a = next(m for m in marks_for(tmp_path / "b") if m.id == "pick")
assert [o["id"] for o in a.options] == ["x", "y"] and a.notes_enabled is False
for bad in ("../x", ".hidden", "a/b", ""):
with pytest.raises(AskError):
write_ask(tmp_path / "b", bad, "p", ["a", "b"])
declare_pick(tmp_path / "b", bad, {"prompt": "p", "options": ["a", "b"]})
with pytest.raises(AskError):
write_ask(tmp_path / "b", "ok", "p", ["solo"])
declare_pick(tmp_path / "b", "ok", {"prompt": "p", "options": ["solo"]})
# ---- gallery + index integration -------------------------------------------
def test_gallery_hides_ask_and_answer_files(tmp_path):
b = _ask(tmp_path / "b")
def test_gallery_hides_legacy_ask_and_answer_files(tmp_path):
"""The legacy sidecars are never deleted (the ROADMAP says so), so they are
still on disk in live booths and must still not render as tiles."""
b = _sidecar(tmp_path / "b")
(b / "a.png").write_bytes(b"x")
write_answer(b, "winner", "A — baseline")
(b / f"winner{ANSWER_SUFFIX}").write_text(json.dumps({"stem": "winner"}))
names = {it["name"] for it in build_gallery(b)}
assert names == {"a.png"}
@@ -149,11 +178,11 @@ def test_gallery_hides_ask_and_answer_files(tmp_path):
def test_list_booths_counts_open_asks(tmp_path):
b = _ask(tmp_path / "b", "one")
_ask(b, "two")
write_answer(b, "two", "A — baseline")
answer_pick(b, "two", "A — baseline")
(tmp_path / "plain").mkdir()
by = {x["name"]: x for x in list_booths(tmp_path, 3600)}
assert by["b"]["asks_open"] == 1 and by["b"]["asks_total"] == 2
assert by["plain"]["asks_open"] == 0 and by["plain"]["asks_total"] == 0
assert by["b"]["marks_open"] == 1 and by["b"]["marks_total"] == 2
assert by["plain"]["marks_open"] == 0 and by["plain"]["marks_total"] == 0
assert by["b"]["count"] == 0 # ask/answer files are not "items"
@@ -169,23 +198,26 @@ def test_booth_page_renders_open_ask_as_form(client):
assert 'value="B — async"' in html
assert 'action="/b/b/answer"' in html
assert "<textarea" in html
assert "1 open ask" in html
assert "1 open" in html
def test_answer_route_writes_sidecar_and_page_shows_it(client):
def test_answer_route_records_the_pick_and_page_shows_it(client):
c, data = client
_ask(data / "b")
r = c.post("/b/b/answer", data={"ask": "winner", "choice": "B — async", "notes": "less banding"},
follow_redirects=False)
assert r.status_code == 303 and r.headers["location"] == "/b/b/#ask-winner"
ans = json.loads((data / "b" / f"winner{ANSWER_SUFFIX}").read_text())
assert r.status_code == 303 and r.headers["location"] == "/b/b/#mark-winner"
ans = _answer_of(data / "b", "winner")
assert ans["choice"] == "B — async" and ans["notes"] == "less banding"
assert ans["answered_by"] # TestClient's client addr
html = c.get("/b/b/").text
assert "answered" in html and "less banding" in html
assert "1 open ask" not in html
# the sidecar is fetchable over HTTP for remote sessions
assert c.get("/b/b/winner.answer.json").json()["choice"] == "B — async"
assert "1 open" not in html
# a session on ANOTHER host reads the judgment over HTTP — the capability the
# per-stem sidecar used to carry, now one request for the whole booth
doc = c.get("/b/b/marks.json").json()
assert doc["open"] == []
assert doc["marks"][0]["answer"]["choice"] == "B — async"
def test_answer_route_rejects_bad_choice_and_unknown_ask(client):
@@ -206,14 +238,16 @@ def test_answer_json_404s_until_answered(client):
def test_notes_field_hidden_when_disabled(client):
c, data = client
_ask(data / "b", notes=False)
assert "<textarea" not in c.get("/b/b/").text
html = c.get("/b/b/").text
assert 'name="notes"' not in html # the pick's free-text field
assert 'name="text"' in html # the add-a-note control stays
def test_index_card_shows_open_ask_badge(client):
def test_index_card_shows_open_mark_badge(client):
c, data = client
_ask(data / "b")
html = c.get("/").text
assert "1 ask" in html
assert "1 open" in html
# ---- multi-question asks ----------------------------------------------------
@@ -230,6 +264,32 @@ MULTI = {
def _multi(booth, stem="batch", **kw):
"""Declare a multi-question pick, the way a session does now."""
doc = json.loads(json.dumps(MULTI)); doc.update(kw)
booth.mkdir(parents=True, exist_ok=True)
declare_pick(booth, stem, doc)
return booth
SINGLE = {"prompt": "Which render wins?", "options": ["A — baseline", "B — async"]}
def _shape(choice, doc=None, stem="winner", **kw):
"""`build_answer` over a normalized declaration — the 2026-09-09 semantics
with no storage under them."""
return build_answer(normalize_ask(json.loads(json.dumps(doc or SINGLE)), stem), choice, **kw)
def _answer_of(booth, mark_id):
"""The recorded judgment for one pick — read back through marks, which is the
one read path now that `<stem>.answer.json` is gone."""
m = next(x for x in marks_for(booth) if x.id == mark_id)
return m.answer
def _multi_sidecar(booth, stem="batch", **kw):
"""The LEGACY multi sidecar — for the tests that are about the legacy files
themselves (still excluded from the item list, still importable)."""
doc = json.loads(json.dumps(MULTI)); doc.update(kw)
booth.mkdir(parents=True, exist_ok=True)
(booth / f"{stem}{ASK_SUFFIX}").write_text(json.dumps(doc))
@@ -264,52 +324,51 @@ def test_normalize_multi_rejects(doc):
normalize_ask(doc, "s")
def test_write_answer_multi_accepts_a_partial_answer(tmp_path):
def test_multi_accepts_a_partial_answer(tmp_path):
"""Blanks are legal (operator ruling 2026-09-09): refusing the whole
submission because one of four was skipped threw away the three that were
made."""
_multi(tmp_path)
a = write_answer(tmp_path, "batch", {"r1": "keep"}) # r2 not submitted at all
_multi_sidecar(tmp_path)
a = _shape({"r1": "keep"}, MULTI, "batch") # r2 not submitted at all
assert a["complete"] is False and a["unanswered"] == ["r2"]
assert list(a["answers"]) == ["r1"]
b = write_answer(tmp_path, "batch", {"r1": "keep", "r2": ""}) # r2 an empty radio group
b = _shape({"r1": "keep", "r2": ""}, MULTI, "batch") # r2 an empty radio group
assert b["unanswered"] == ["r2"] and b["complete"] is False
# a note without a pick is still worth keeping
c = write_answer(tmp_path, "batch", {"r1": "", "r2": "k"}, qnotes={"r1": "undecided"})
c = _shape({"r1": "", "r2": "k"}, MULTI, "batch", qnotes={"r1": "undecided"})
assert c["answers"]["r1"] == {"prompt": "Render 1?", "choice": None,
"choice_index": None, "label": "", "notes": "undecided"}
assert c["unanswered"] == ["r1"]
# nothing at all is refused: it would flip the ask to answered with no decision
with pytest.raises(AskError):
write_answer(tmp_path, "batch", {"r1": "", "r2": ""})
_shape({"r1": "", "r2": ""}, MULTI, "batch")
# ...but notes alone are a real submission
d = write_answer(tmp_path, "batch", {"r1": "", "r2": ""}, "ask me tomorrow")
d = _shape({"r1": "", "r2": ""}, MULTI, "batch", notes="ask me tomorrow")
assert d["complete"] is False and d["notes"] == "ask me tomorrow" and d["answers"] == {}
def test_single_ask_may_be_answered_with_notes_only(tmp_path):
_ask(tmp_path)
_sidecar(tmp_path)
with pytest.raises(AskError):
write_answer(tmp_path, "winner", "")
a = write_answer(tmp_path, "winner", "", "neither is right, rerun")
answer_pick(tmp_path, "winner", "")
a = _shape("", notes="neither is right, rerun")
assert a["choice"] is None and a["complete"] is False
assert a["notes"] == "neither is right, rerun"
def test_write_answer_multi_still_rejects_a_bad_option(tmp_path):
_multi(tmp_path)
def test_multi_still_rejects_a_bad_option(tmp_path):
_multi_sidecar(tmp_path)
with pytest.raises(AskError):
write_answer(tmp_path, "batch", {"r1": "keep", "r2": "nope"})
_shape({"r1": "keep", "r2": "nope"}, MULTI, "batch")
with pytest.raises(AskError):
write_answer(tmp_path, "batch", "keep") # wrong shape
ans = write_answer(tmp_path, "batch", {"r1": "drop", "r2": "k"}, "overall fine",
_shape("keep", MULTI, "batch") # wrong shape
ans = _shape({"r1": "drop", "r2": "k"}, MULTI, "batch", notes="overall fine",
qnotes={"r1": "banding", "r2": "ignored: notes off"})
assert list(ans["answers"]) == ["r1", "r2"]
assert ans["answers"]["r1"] == {"prompt": "Render 1?", "choice": "drop", "choice_index": 1,
"label": "drop", "notes": "banding"}
assert ans["answers"]["r2"]["choice"] == "k" and ans["answers"]["r2"]["notes"] == ""
assert ans["notes"] == "overall fine" and ans["title"] == "R18 batch review"
assert read_answer(tmp_path, "batch") == ans
def test_multi_page_and_route(client):
@@ -323,24 +382,24 @@ def test_multi_page_and_route(client):
# a partial submission is RECORDED, not refused
assert c.post("/b/b/answer", data={"ask": "batch", "choice.r1": "keep"},
follow_redirects=False).status_code == 303
part = json.loads((data / "b" / f"batch{ANSWER_SUFFIX}").read_text())
part = _answer_of(data / "b", "batch")
assert part["complete"] is False and part["unanswered"] == ["r2"]
assert "1/2" in c.get("/b/b/").text and "partial" in c.get("/b/b/").text
r = c.post("/b/b/answer", data={"ask": "batch", "choice.r1": "keep", "notes.r1": "crisp",
"choice.r2": "d", "notes": "ship r1"}, follow_redirects=False)
assert r.status_code == 303
ans = c.get("/b/b/batch.answer.json").json()
ans = _answer_of(data / "b", "batch")
assert ans["answers"]["r1"]["choice"] == "keep" and ans["answers"]["r1"]["notes"] == "crisp"
assert ans["answers"]["r2"]["choice"] == "d" and ans["notes"] == "ship r1"
html = c.get("/b/b/").text
assert "answered" in html and "crisp" in html and "ship r1" in html
def test_write_ask_accepts_full_doc(tmp_path):
write_ask(tmp_path / "b", "batch", doc=MULTI)
assert load_ask(tmp_path / "b", "batch")["multi"] is True
def test_declare_pick_accepts_a_full_multi_doc(tmp_path):
declare_pick(tmp_path / "b", "batch", MULTI)
assert next(m for m in marks_for(tmp_path / "b") if m.id == "batch").multi is True
with pytest.raises(AskError):
write_ask(tmp_path / "b", "bad", doc={"questions": []})
declare_pick(tmp_path / "b", "bad", {"questions": []})
# ---- verbatim-index booths ---------------------------------------------------
@@ -368,7 +427,7 @@ def test_verbatim_chip_disappears_once_answered(client):
c, data = client
b = _ask(data / "b")
(b / "index.html").write_text("<!doctype html><body>hi</body>")
write_answer(b, "winner", "A — baseline")
answer_pick(b, "winner", "A — baseline")
assert "booth-nav-asks" not in c.get("/b/b/").text
@@ -383,20 +442,20 @@ def test_asks_page_renders_forms_and_answers_back_to_itself(client):
c, data = client
b = _ask(data / "b")
(b / "index.html").write_text("<!doctype html><body>hi</body>")
page = c.get("/b/b/asks").text
page = c.get("/b/b/marks").text
assert "Which render wins?" in page and 'type="radio"' in page
assert 'name="back" value="asks"' in page
r = c.post("/b/b/answer", data={"ask": "winner", "choice": "B — async", "back": "asks"},
assert 'name="back" value="marks"' in page
r = c.post("/b/b/answer", data={"ask": "winner", "choice": "B — async", "back": "marks"},
follow_redirects=False)
assert r.headers["location"] == "/b/b/asks#ask-winner"
assert read_answer(b, "winner")["choice"] == "B — async"
assert "answered" in c.get("/b/b/asks").text
assert r.headers["location"] == "/b/b/marks#mark-winner"
assert _answer_of(b, "winner")["choice"] == "B — async"
assert "answered" in c.get("/b/b/marks").text
def test_asks_page_on_a_booth_with_none(client):
c, data = client
(data / "b").mkdir()
assert "no asks" in c.get("/b/b/asks").text
assert "no marks" in c.get("/b/b/marks").text
def test_asks_page_404s_for_unknown_booth(client):
@@ -415,7 +474,7 @@ def test_single_ask_keeps_its_title(tmp_path):
def test_asks_page_shows_a_single_ask_title(client):
c, data = client
_ask(data / "b", title="emmie — pick the anchor")
assert "emmie — pick the anchor" in c.get("/b/b/asks").text
assert "emmie — pick the anchor" in c.get("/b/b/marks").text
# ---- inline placement in a verbatim report -----------------------------------
@@ -460,7 +519,7 @@ def test_inline_form_submits_every_question_in_one_post(client):
r = c.post("/b/b/answer", data={"ask": "batch", "choice.r1": "keep", "notes.r1": "crisp",
"choice.r2": "d", "notes": "ship r1"}, follow_redirects=False)
assert r.status_code == 303
ans = read_answer(b, "batch")
ans = _answer_of(b, "batch")
assert ans["answers"]["r1"]["choice"] == "keep" and ans["answers"]["r2"]["choice"] == "d"
# and the recorded pick now shows inline, on the report itself
html = c.get("/b/b/").text
@@ -510,7 +569,7 @@ def test_radios_are_not_html_required_anywhere(client):
assert "required" not in c.get("/b/b/").text
(b / "index.html").write_text('<!doctype html><body><div data-booth-ask="batch"></div></body>')
assert "required" not in c.get("/b/b/").text
assert "required" not in c.get("/b/b/asks").text
assert "required" not in c.get("/b/b/marks").text
def test_partial_answer_renders_as_skipped_inline(client):
@@ -527,4 +586,4 @@ def test_empty_submission_is_refused_with_400(client):
c, data = client
b = _multi(data / "b")
assert c.post("/b/b/answer", data={"ask": "batch", "choice.r1": "", "choice.r2": ""}).status_code == 400
assert read_answer(b, "batch") is None # the ask stays OPEN, not falsely answered
assert _answer_of(b, "batch") is None # the pick stays OPEN, not falsely answered
tests/test_marks.py
+715
View File
@@ -0,0 +1,715 @@
"""Marks: one primitive for operator judgment attached to an artifact.
`pick` (the session asks), `note` (the operator tells), `flag` (the operator
points at the good ones) — three shapes, one storage model, one read path.
See docs/contracts/u2_marks.contract.md.
"""
import ast
import json
import pathlib
import sys
import pytest
from booth.asks import ASK_SUFFIX, AskError, build_answer, normalize_ask
from booth.marks import (
MARKS_FILE,
Mark,
answer_pick,
as_dict,
declare_pick,
marks_for,
open_marks,
)
def _single():
return {"prompt": "Which render wins?", "options": ["A — baseline", "B — async"]}
def _multi():
return {
"title": "R18 batch review",
"questions": [
{"key": "q1", "prompt": "Render 1?", "options": ["keep", "drop"]},
{"key": "q2", "prompt": "Render 2?", "options": ["keep", "drop"]},
],
}
# ---- slice 1: storage, the record, the read path ----------------------------
def test_pick_round_trips_through_marks_json(tmp_path):
"""A declared pick lands in ONE file and reads back normalized."""
booth = tmp_path / "b"
booth.mkdir()
declare_pick(booth, "winner", _single())
assert (booth / MARKS_FILE).is_file()
marks = marks_for(booth)
assert len(marks) == 1
m = marks[0]
assert isinstance(m, Mark)
assert (m.id, m.shape, m.target) == ("winner", "pick", None)
assert m.answer is None
assert m.prompt == "Which render wins?"
assert m.multi is False
# normalize_ask emits `notes` (bool); the record carries it as notes_enabled (SR-2)
assert m.notes_enabled is True
assert m.notes_label == "notes"
assert [o["id"] for o in m.options] == ["A — baseline", "B — async"]
assert len(m.questions) == 1 and m.questions[0]["key"] is None
def test_marks_json_is_a_dotfile_so_it_is_not_an_item(tmp_path):
"""The dotfile skip in booth_items already covers it. Asserted, not assumed."""
from booth.items import booth_items
booth = tmp_path / "b"
booth.mkdir()
(booth / "a.png").write_bytes(b"\x89PNG")
declare_pick(booth, "winner", _single())
assert MARKS_FILE.startswith(".")
assert [i.rel for i in booth_items(booth)] == ["a.png"]
def test_corrupt_marks_json_renders_as_empty(tmp_path):
"""A booth with no marks and a booth with a broken mark file both render as
'no marks'. Neither is a 500."""
booth = tmp_path / "b"
booth.mkdir()
(booth / MARKS_FILE).write_text("{not json at all")
assert marks_for(booth) == []
def test_missing_booth_and_missing_file_are_empty(tmp_path):
assert marks_for(tmp_path / "nope") == []
(tmp_path / "b").mkdir()
assert marks_for(tmp_path / "b") == []
def test_broken_pick_declaration_surfaces_error(tmp_path):
"""A declaration that cannot be rendered comes back with `error` set, so the
page can SAY so rather than silently hiding a question the session believes
it posted. Preserved from list_asks."""
booth = tmp_path / "b"
booth.mkdir()
(booth / MARKS_FILE).write_text(json.dumps({
"version": 1,
"marks": [{"id": "bad", "shape": "pick", "target": None,
"created": "2026-09-21T10:00:00-07:00",
"declaration": {"prompt": "no options"}, "answer": None}],
}))
m = marks_for(booth)[0]
assert m.error is not None
assert m.id == "bad"
def test_declare_pick_validates_before_writing(tmp_path):
"""A session cannot land a question the renderer would refuse."""
booth = tmp_path / "b"
booth.mkdir()
with pytest.raises(AskError):
declare_pick(booth, "bad", {"prompt": "only one", "options": ["just me"]})
assert not (booth / MARKS_FILE).exists()
def test_declare_pick_rejects_a_bad_id(tmp_path):
booth = tmp_path / "b"
booth.mkdir()
with pytest.raises(AskError):
declare_pick(booth, "../escape", _single())
def test_redeclaring_a_pick_clears_its_answer(tmp_path):
"""The question changed, so the old judgment is not an answer to it."""
booth = tmp_path / "b"
booth.mkdir()
declare_pick(booth, "winner", _single())
answer_pick(booth, "winner", "A — baseline")
assert marks_for(booth)[0].answer is not None
declare_pick(booth, "winner", {"prompt": "Which one now?", "options": ["X", "Y"]})
marks = marks_for(booth)
assert len(marks) == 1
assert marks[0].answer is None
assert marks[0].prompt == "Which one now?"
def test_marks_are_oldest_first_by_created(tmp_path):
booth = tmp_path / "b"
booth.mkdir()
declare_pick(booth, "first", _single())
declare_pick(booth, "second", _single())
assert [m.id for m in marks_for(booth)] == ["first", "second"]
# ---- openness: THE one predicate (INV-2) ------------------------------------
def test_open_marks_counts_an_unanswered_pick(tmp_path):
booth = tmp_path / "b"
booth.mkdir()
declare_pick(booth, "winner", _single())
assert [m.id for m in open_marks(marks_for(booth))] == ["winner"]
def test_open_marks_drops_a_complete_answer(tmp_path):
booth = tmp_path / "b"
booth.mkdir()
declare_pick(booth, "winner", _single())
answer_pick(booth, "winner", "A — baseline")
assert open_marks(marks_for(booth)) == []
def test_a_partial_answer_counts_as_open(tmp_path):
"""SR-7, the declared behaviour change. Today the index badge tests
`answer is None` and so reports a half-answered four-question pick as
closed, while the panel renders it `◐ partial`. A booth that still owes an
answer is open — which is also what makes U4's pin rule correct."""
booth = tmp_path / "b"
booth.mkdir()
declare_pick(booth, "batch", _multi())
answer_pick(booth, "batch", {"q1": "keep"}) # q2 left alone
m = marks_for(booth)[0]
assert m.answer is not None
assert m.answer["complete"] is False
assert m.answer["unanswered"] == ["q2"]
assert [x.id for x in open_marks(marks_for(booth))] == ["batch"]
# ---- the pick semantics, preserved by NOT rewriting them (INV-4) ------------
def test_partial_answer_is_recorded_not_refused(tmp_path):
"""Operator ruling 2026-09-09, as a regression test."""
booth = tmp_path / "b"
booth.mkdir()
declare_pick(booth, "batch", _multi())
m = answer_pick(booth, "batch", {"q1": "keep", "q2": ""})
assert m.answer["answers"]["q1"]["label"] == "keep"
assert "q2" not in m.answer["answers"]
assert m.answer["unanswered"] == ["q2"]
assert m.answer["complete"] is False
def test_nothing_to_record_is_still_refused(tmp_path):
booth = tmp_path / "b"
booth.mkdir()
declare_pick(booth, "batch", _multi())
with pytest.raises(AskError):
answer_pick(booth, "batch", {"q1": "", "q2": ""})
def test_an_offered_but_invalid_option_is_an_error(tmp_path):
"""A broken form, not a skipped question."""
booth = tmp_path / "b"
booth.mkdir()
declare_pick(booth, "winner", _single())
with pytest.raises(AskError):
answer_pick(booth, "winner", "C — never offered")
def test_answering_an_unknown_pick_raises(tmp_path):
"""SR-4: load_ask used to own this AskError. Extracting build_answer moved
the path to the caller, and a stale form POST must be a 400, not a silent
no-op."""
booth = tmp_path / "b"
booth.mkdir()
with pytest.raises(AskError):
answer_pick(booth, "ghost", "A — baseline")
def test_re_answering_overwrites(tmp_path):
"""The mark is the CURRENT judgment, not a log."""
booth = tmp_path / "b"
booth.mkdir()
declare_pick(booth, "winner", _single())
answer_pick(booth, "winner", "A — baseline")
m = answer_pick(booth, "winner", "B — async")
assert m.answer["label"] == "B — async"
assert len(marks_for(booth)) == 1
def test_build_answer_matches_the_old_sidecar_document(tmp_path):
"""INV-4. build_answer is write_answer's logic with the I/O removed; the
document it produces is the one the sidecar carried."""
ask = normalize_ask(_single(), "winner")
doc = build_answer(ask, "B — async", notes="less banding", who="10.0.0.1")
assert doc["stem"] == "winner"
assert doc["choice"] == "B — async"
assert doc["choice_index"] == 1
assert doc["label"] == "B — async"
assert doc["notes"] == "less banding"
assert doc["complete"] is True
assert doc["unanswered"] == []
assert doc["answered_by"] == "10.0.0.1"
assert "answered_at" in doc
def test_build_answer_sources_stem_from_the_ask(tmp_path):
"""SR-3: the extraction takes a pre-loaded ask and reads `stem` off it."""
ask = normalize_ask(_multi(), "batch")
doc = build_answer(ask, {"q1": "keep", "q2": "drop"})
assert doc["stem"] == "batch"
assert doc["complete"] is True
# ---- the JSON boundary (SR-8) -----------------------------------------------
def test_as_dict_round_trips_through_json(tmp_path):
booth = tmp_path / "b"
booth.mkdir()
declare_pick(booth, "winner", _single())
m = marks_for(booth)[0]
out = json.loads(json.dumps(as_dict(m)))
assert out["id"] == "winner"
assert out["shape"] == "pick"
assert out["answer"] is None
# ---- the stdlib-only invariant (INV-5) --------------------------------------
@pytest.mark.parametrize("module", ["marks", "asks", "links"])
def test_stdlib_only(module):
"""INV-5. scripts/booth imports these under the system python3 with NO venv,
through a `python3 -c` heredoc that no AST extractor can see — so nothing
but this test stands between a casual third-party import and `booth ask`
breaking on every fleet host."""
src = pathlib.Path(__file__).parent.parent / "booth" / f"{module}.py"
tree = ast.parse(src.read_text())
roots = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
roots.update(a.name.split(".")[0] for a in node.names)
elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module:
roots.add(node.module.split(".")[0])
outside = {r for r in roots if r != "booth" and r not in sys.stdlib_module_names}
assert not outside, f"booth/{module}.py imports non-stdlib: {sorted(outside)}"
# ---- slice 2: flag — the shape with no existing code ------------------------
def test_flag_is_an_upsert_and_unflag_removes(tmp_path):
"""One item has at most one flag state, so flagging twice is idempotent and
clearing REMOVES the mark rather than storing `value: false` — an absent
flag and a false flag are the same judgment."""
from booth.marks import set_flag
booth = tmp_path / "b"
booth.mkdir()
m = set_flag(booth, "DSC03389.jpg", True, who="10.0.0.1")
assert m is not None and m.shape == "flag" and m.flagged is True
assert m.target == "DSC03389.jpg"
set_flag(booth, "DSC03389.jpg", True) # again
assert len(marks_for(booth)) == 1
assert set_flag(booth, "DSC03389.jpg", False) is None
assert marks_for(booth) == []
def test_unflagging_something_unflagged_is_not_an_error(tmp_path):
from booth.marks import set_flag
booth = tmp_path / "b"
booth.mkdir()
assert set_flag(booth, "nope.png", False) is None
def test_flags_on_different_items_coexist(tmp_path):
from booth.marks import marks_for_target, set_flag
booth = tmp_path / "b"
booth.mkdir()
set_flag(booth, "a.png", True)
set_flag(booth, "b.png", True)
marks = marks_for(booth)
assert len(marks) == 2
assert [m.id for m in marks_for_target(marks, "a.png")] == ["flag:a.png"]
def test_a_flag_never_counts_as_open(tmp_path):
"""Only a pick can owe an answer. A flag is born resolved."""
from booth.marks import set_flag
booth = tmp_path / "b"
booth.mkdir()
set_flag(booth, "a.png", True)
assert open_marks(marks_for(booth)) == []
def test_a_flag_target_cannot_escape_the_booth(tmp_path):
from booth.marks import set_flag
booth = tmp_path / "b"
booth.mkdir()
for bad in ("../outside.png", "/etc/passwd", ""):
with pytest.raises(AskError):
set_flag(booth, bad, True)
# ---- slice 3: note ----------------------------------------------------------
def test_note_allows_several_per_target(tmp_path):
"""An item may carry several notes, so each gets a generated id."""
from booth.marks import marks_for_target, write_note
booth = tmp_path / "b"
booth.mkdir()
write_note(booth, "a.png", "banding in the gradient")
write_note(booth, "a.png", "and the highlight clips")
marks = marks_for(booth)
assert len(marks) == 2
assert {m.id for m in marks} == {"note-1", "note-2"}
assert [m.text for m in marks_for_target(marks, "a.png")] == [
"banding in the gradient", "and the highlight clips"]
def test_booth_level_note_has_target_none(tmp_path):
"""The booth itself is a legal target."""
from booth.marks import marks_for_target, write_note
booth = tmp_path / "b"
booth.mkdir()
write_note(booth, None, "ship v3, not v4")
m = marks_for(booth)[0]
assert m.target is None
assert marks_for_target(marks_for(booth), None) == [m]
def test_an_empty_note_is_refused(tmp_path):
"""Nothing to record — the same posture as an empty pick submission."""
from booth.marks import write_note
booth = tmp_path / "b"
booth.mkdir()
for blank in ("", " ", "\n\n", None):
with pytest.raises(AskError):
write_note(booth, "a.png", blank)
assert marks_for(booth) == []
def test_a_note_never_counts_as_open(tmp_path):
from booth.marks import write_note
booth = tmp_path / "b"
booth.mkdir()
write_note(booth, "a.png", "seen it")
assert open_marks(marks_for(booth)) == []
def test_delete_mark_is_the_operators_undo(tmp_path):
from booth.marks import delete_mark, write_note
booth = tmp_path / "b"
booth.mkdir()
write_note(booth, "a.png", "wrong call")
assert delete_mark(booth, "note-1") is True
assert marks_for(booth) == []
assert delete_mark(booth, "note-1") is False
# ---- the lock (INV-6) -------------------------------------------------------
def test_concurrent_writers_do_not_lose_a_mark(tmp_path):
"""INV-6. Two processes writing the same booth's marks at once: both land.
Processes, not threads — the flock is what is under test, and a thread-level
test would pass on the GIL alone."""
import subprocess
import sys as _sys
booth = tmp_path / "b"
booth.mkdir()
repo = str(pathlib.Path(__file__).parent.parent)
prog = (
"import sys; sys.path.insert(0, %r);"
"from booth.marks import set_flag;"
"set_flag(%r, sys.argv[1], True)" % (repo, str(booth))
)
procs = [subprocess.Popen([_sys.executable, "-c", prog, f"item{i}.png"])
for i in range(8)]
for p in procs:
assert p.wait() == 0
got = {m.target for m in marks_for(booth)}
assert got == {f"item{i}.png" for i in range(8)}, f"lost a mark: {sorted(got)}"
# ---- slice 4: the legacy sidecars — imported, NEVER deleted ----------------
def _sidecar(booth, stem, doc, mtime=None):
booth.mkdir(parents=True, exist_ok=True)
p = booth / f"{stem}{ASK_SUFFIX}"
p.write_text(json.dumps(doc))
if mtime is not None:
import os as _os
_os.utime(p, (mtime, mtime))
return p
def test_import_reads_a_sidecar_into_marks(tmp_path):
from booth.marks import import_legacy_asks
booth = tmp_path / "b"
_sidecar(booth, "winner", _single())
got = import_legacy_asks(booth)
assert [m.id for m in got] == ["winner"]
m = marks_for(booth)[0]
assert m.shape == "pick"
assert m.prompt == "Which render wins?"
assert m.answer is None
def test_import_is_idempotent_and_keeps_the_sidecar(tmp_path):
"""INV-7. The ROADMAP's non-goal is explicit: no migration that deletes
anything. Four of these are live and unanswered right now."""
from booth.marks import import_legacy_asks
booth = tmp_path / "b"
p = _sidecar(booth, "winner", _single())
import_legacy_asks(booth)
assert import_legacy_asks(booth) == [] # second run is a no-op
assert len(marks_for(booth)) == 1
assert p.is_file(), "the importer deleted the operator's sidecar"
def test_import_carries_an_existing_answer(tmp_path):
from booth.asks import ANSWER_SUFFIX, build_answer, normalize_ask
from booth.marks import import_legacy_asks
booth = tmp_path / "b"
_sidecar(booth, "winner", _single())
# The legacy answer sidecar, written the way the retired `write_answer` did.
doc = build_answer(normalize_ask(_single(), "winner"), "B — async", notes="less banding")
(booth / f"winner{ANSWER_SUFFIX}").write_text(json.dumps(doc))
import_legacy_asks(booth)
m = marks_for(booth)[0]
assert m.answer["label"] == "B — async"
assert m.answer["notes"] == "less banding"
assert open_marks(marks_for(booth)) == []
def test_import_does_not_clobber_a_newer_answer(tmp_path):
"""A stem already present as a mark is skipped outright."""
from booth.marks import import_legacy_asks
booth = tmp_path / "b"
_sidecar(booth, "winner", _single())
declare_pick(booth, "winner", _single())
answer_pick(booth, "winner", "A — baseline")
assert import_legacy_asks(booth) == []
assert marks_for(booth)[0].answer["label"] == "A — baseline"
def test_import_preserves_mtime_ordering(tmp_path):
"""SR-5. list_asks ordered by file mtime; marks_for orders by the stored
`created`. Seed it from the sidecar's mtime or the four live asks silently
reorder on import."""
from booth.marks import import_legacy_asks
booth = tmp_path / "b"
_sidecar(booth, "later", _single(), mtime=2_000_000_000)
_sidecar(booth, "earlier", _single(), mtime=1_000_000_000)
import_legacy_asks(booth)
assert [m.id for m in marks_for(booth)] == ["earlier", "later"]
def test_import_surfaces_a_broken_sidecar_instead_of_dropping_it(tmp_path):
"""A question the session believes it posted stays visible."""
from booth.marks import import_legacy_asks
booth = tmp_path / "b"
_sidecar(booth, "broke", {"prompt": "no options here"})
import_legacy_asks(booth)
m = marks_for(booth)[0]
assert m.id == "broke"
assert m.error is not None
def test_import_ignores_a_bad_stem(tmp_path):
"""A filename that could not have been written by `booth ask` is left alone
rather than becoming a mark with an unusable id."""
from booth.marks import import_legacy_asks
booth = tmp_path / "b"
booth.mkdir()
(booth / f"..evil{ASK_SUFFIX}").write_text(json.dumps(_single()))
assert import_legacy_asks(booth) == []
def test_import_on_a_booth_with_nothing_is_empty(tmp_path):
from booth.marks import import_legacy_asks
assert import_legacy_asks(tmp_path / "nope") == []
(tmp_path / "b").mkdir()
assert import_legacy_asks(tmp_path / "b") == []
# ---- the deterministic-order invariant (operator directive, 2026-09-21) ------
def test_marks_order_is_deterministic_across_reads(tmp_path):
"""Every ordered collection the Booth renders needs a STATED rule, because
the operator judges positionally — "the third one", "the one after the
banded one" — and an order that moves between renders misfiles the judgment
instead of crashing. Marks order by `(created, id)`."""
booth = tmp_path / "b"
booth.mkdir()
declare_pick(booth, "b-pick", _single())
from booth.marks import set_flag, write_note
write_note(booth, "z.png", "later note")
set_flag(booth, "a.png", True)
write_note(booth, None, "booth note")
first = [m.id for m in marks_for(booth)]
for _ in range(5):
assert [m.id for m in marks_for(booth)] == first
def test_same_second_marks_tie_break_on_id(tmp_path):
"""The `created` stamp is second-resolution, so two marks written inside one
second would order by whatever json listed them. The id is the tie-break."""
booth = tmp_path / "b"
booth.mkdir()
stamp = "2026-09-21T12:00:00-07:00"
(booth / MARKS_FILE).write_text(json.dumps({"version": 1, "marks": [
{"id": "zeta", "shape": "note", "target": None, "created": stamp, "text": "z"},
{"id": "alpha", "shape": "note", "target": None, "created": stamp, "text": "a"},
]}))
assert [m.id for m in marks_for(booth)] == ["alpha", "zeta"]
# and reversing the stored order changes nothing
(booth / MARKS_FILE).write_text(json.dumps({"version": 1, "marks": [
{"id": "alpha", "shape": "note", "target": None, "created": stamp, "text": "a"},
{"id": "zeta", "shape": "note", "target": None, "created": stamp, "text": "z"},
]}))
assert [m.id for m in marks_for(booth)] == ["alpha", "zeta"]
def test_open_marks_preserves_the_read_order(tmp_path):
"""The filtered view must not re-order — the panel and the badge have to
agree with the list they came from."""
booth = tmp_path / "b"
booth.mkdir()
for mid in ("c", "a", "b"):
declare_pick(booth, mid, _single())
marks = marks_for(booth)
assert [m.id for m in open_marks(marks)] == [m.id for m in marks]
# ---- INV-3: the judgment travels, like the caption --------------------------
def _png(p):
p.write_bytes(
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00"
b"\x01\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82"
)
@pytest.fixture
def client(tmp_path):
from fastapi.testclient import TestClient
from booth.app import create_app
return TestClient(create_app(tmp_path, ttl_hours=24, start_sweeper=False)), tmp_path
def test_zoom_carries_the_marks(client):
"""U1's rule, extended from the caption to the judgment: full size is where
the operator is actually deciding, so the flag state and the notes have to
be there."""
from booth.marks import set_flag, write_note
c, data = client
b = data / "b"
b.mkdir()
_png(b / "a.png")
set_flag(b, "a.png", True)
write_note(b, "a.png", "banding in the gradient")
html = c.get("/b/b/view?f=a.png").text
assert "banding in the gradient" in html
assert "✔ flagged" in html
assert 'action="/b/b/flag"' in html
def test_the_tile_carries_the_marks(client):
from booth.marks import set_flag, write_note
c, data = client
b = data / "b"
b.mkdir()
_png(b / "a.png")
_png(b / "b.png")
set_flag(b, "a.png", True)
write_note(b, "b.png", "this one is soft")
html = c.get("/b/b/").text
assert "✔ flagged" in html and "○ flag" in html # a flagged one and an unflagged one
assert "this one is soft" in html
assert 'id="item-a.png"' in html
def test_flag_and_note_round_trip_through_the_browser(client):
c, data = client
b = data / "b"
b.mkdir()
_png(b / "a.png")
assert c.post("/b/b/flag", data={"target": "a.png", "on": "1"},
follow_redirects=False).status_code == 303
assert c.post("/b/b/note", data={"target": "a.png", "text": "soft"},
follow_redirects=False).status_code == 303
doc = c.get("/b/b/marks.json").json()
assert {m["shape"] for m in doc["marks"]} == {"flag", "note"}
assert doc["open"] == [] # neither owes an answer
# and the operator can withdraw one
note_id = next(m["id"] for m in doc["marks"] if m["shape"] == "note")
assert c.post("/b/b/unmark", data={"mark": note_id},
follow_redirects=False).status_code == 303
assert [m["shape"] for m in c.get("/b/b/marks.json").json()["marks"]] == ["flag"]
def test_a_no_op_write_does_not_touch_the_booth(tmp_path):
"""A booth's TTL is measured from its newest mtime, INCLUDING dotfiles — so
writing `.marks.json` resets the clock. Marking IS activity and should; a
write that changes nothing is not activity and must not. Unflagging
something never flagged would otherwise keep a dead booth alive."""
from booth.marks import delete_mark, set_flag
booth = tmp_path / "b"
booth.mkdir()
assert set_flag(booth, "ghost.png", False) is None
assert delete_mark(booth, "nothing") is False
assert not (booth / MARKS_FILE).exists(), "a no-op write created the mark file"
def test_a_real_write_then_a_no_op_leaves_the_file_alone(tmp_path):
import os
from booth.marks import set_flag
booth = tmp_path / "b"
booth.mkdir()
set_flag(booth, "a.png", True)
path = booth / MARKS_FILE
os.utime(path, (1_000_000_000, 1_000_000_000))
before = path.stat().st_mtime
set_flag(booth, "a.png", True) # idempotent: already flagged
assert path.stat().st_mtime == before, "an idempotent flag rewrote the file"