Files
app/acp/cards.py
T

112 lines
5.0 KiB
Python

"""!card — the one write verb a room seat has.
A seat ends a turn with:
!card <title>
<body: task + DONE, until the next !command or end of message>
The runner (not the seat) files it: an issue by the seat's bot identity, placed in
Signal on the board, milestone of the room issue if any, no assignee. The room gets a
receipt linking the card; the card body links the room. Retries cannot duplicate:
the request id is (room issue, comment id, ordinal) and is checked against the
ledger (`<!-- card:<id> -->` in the new issue body) before filing.
Moving to Ready, assigning, closing, merging: chair only. Seats talk and file.
"""
from __future__ import annotations
import re
from typing import Any
from factory.gitea import REPO, Gitea, GiteaError
CARD_RE = re.compile(r"(?ms)^!card[ \t]+(?P<title>[^\n]+)\n(?P<body>.*?)(?=^![a-z]|\Z)")
SECOND_RE = re.compile(r"^!second #(\d+)$")
def seconds_in(text: str) -> list[int]:
"""A second is a message that is exactly `!second #N` (trailing newlines tolerated).
It becomes an execution signature (`!go #N`), so it is never extracted from a
longer message: no prose, no quotes, no code blocks, open or closed."""
m = SECOND_RE.match((text or "").rstrip("\r\n"))
return [int(m.group(1))] if m else []
SIGNAL_COLUMN, READY_COLUMN = 1, 2 # Factory board columns
FILED_RE = re.compile(r"(?m)^filed by `([a-z0-9_.-]+)` from room #(\d+)")
def parse_cards(text: str) -> list[tuple[str, str]]:
return [(m.group("title").strip(), m.group("body").strip()) for m in CARD_RE.finditer(text)]
def already_filed(g: Gitea, request_id: str) -> int | None:
rows = g.get(f"/repos/{REPO}/issues", state="all", type="issues", q=request_id, limit=5) or []
for r in rows:
if f"<!-- card:{request_id} -->" in (r.get("body") or ""):
return int(r["number"])
return None
def file_cards(seat: Gitea, board_seat: Gitea, room: int, comment_id: int, text: str, project: int = 1) -> list[dict[str, Any]]:
"""File every !card in `text`, posted by `seat`, into Signal. Returns receipts."""
out: list[dict[str, Any]] = []
try:
room_issue = seat.issue(room)
except GiteaError:
room_issue = {}
milestone = (room_issue.get("milestone") or {}).get("id")
for i, (title, body) in enumerate(parse_cards(text)):
rid = f"{room}-{comment_id}-{i}"
n = already_filed(seat, rid)
if n is None:
payload: dict[str, Any] = {
"title": title[:120],
"body": f"<!-- card:{rid} -->\nfiled by `{seat.seat_name}` from room #{room}\n\n{body}",
}
if milestone:
payload["milestone"] = milestone
d = seat.post(f"/repos/{REPO}/issues", payload)
n = int(d["number"])
try:
board_seat.post(f"/repos/{REPO}/projects/{project}/columns/{SIGNAL_COLUMN}/issues/{d['id']}")
except GiteaError as e:
out.append({"card": n, "board": f"not placed: {e}"})
seat.receipt(room, f"filed #{n} → Signal: {title[:80]}", REPO)
out.append({"card": n, "title": title, "request": rid})
return out
def second_cards(seat: Gitea, board_seat: Gitea, room: int, text: str, project: int = 1) -> list[dict[str, Any]]:
"""`!second #N` by a seat that did not file #N moves it Signal -> Ready. One seat cannot
second its own card; the chair can always pull a card back. Returns receipts."""
out: list[dict[str, Any]] = []
for n in seconds_in(text):
try:
card = seat.issue(n)
except GiteaError as e:
out.append({"second": n, "refused": f"no such card ({e.status})"}); continue
f = FILED_RE.search(card.get("body") or "")
if not f:
seat.receipt(room, f"!second #{n} refused: not a room-filed card", REPO)
out.append({"second": n, "refused": "not room-filed"}); continue
filer = f.group(1)
if filer == seat.seat_name:
seat.receipt(room, f"!second #{n} refused: a seat cannot second its own card", REPO)
out.append({"second": n, "refused": "own card"}); continue
if card.get("state") != "open":
out.append({"second": n, "refused": "closed"}); continue
# Sign first, then move: the board may pull a Ready card at any moment, and an
# unsigned Ready card would bounce to Signal (finishline.gate).
try:
seat.comment(n, f"!go #{n}", REPO) # the seconding seat's signature; gate reads only this exact form
except GiteaError as e:
out.append({"second": n, "refused": f"sign: {e}"}); continue
try:
board_seat.post(f"/repos/{REPO}/projects/{project}/issues/{card['id']}/move", {"column_id": READY_COLUMN, "position": 0})
except GiteaError as e:
out.append({"second": n, "refused": f"board: {e}"}); continue
seat.receipt(room, f"seconded #{n} (filed by `{filer}`) → Ready", REPO)
seat.receipt(n, f"seconded in room #{room} → Ready", REPO)
out.append({"second": n, "to": "Ready", "filer": filer})
return out