"""Approval gating primitives (architecture L3). Approvals bind to a plan revision, never to "the current gate": every revision carries a monotonic ``rev-N`` tag, an approval naming a stale revision is answered with a notice and never applied to a newer one, and the first valid approval wins. Approval is always the typed word ``approve`` (optionally ``approve rev-N``); any other reply is a rejection with feedback. The approver is authenticated against a per-channel operator allowlist. Revisions are numbered across a whole work item, not per gate, so ``rev-N`` names one plan. A reply that does not name a revision binds to the one that was up when it arrived (its ``about``): if that is not the current revision, it is ambiguous and not applied, neither as an approval nor as feedback. That is how a second "approve", or an afterthought, sent while one plan was up can never land on the next. The channel seam: any approval source (Matrix transport, Gitea comment poller, Vikunja comments) feeds operator replies to ``Gate.evaluate`` with its channel name. """ import re from collections.abc import Set as AbstractSet from dataclasses import dataclass @dataclass(frozen=True) class PlanRevision: gate: str rev: int content: str @property def tag(self) -> str: return f"rev-{self.rev}" @dataclass(frozen=True) class Approved: revision: PlanRevision approver: str channel: str @dataclass(frozen=True) class Rejected: feedback: str approver: str channel: str @dataclass(frozen=True) class Ignored: notice: str | None # posted back to the channel when not None _APPROVE_RE = re.compile(r"^approve(?:\s+(rev-\d+))?$") def command_form(text: str) -> str: """How a message reads as a command: case, surrounding whitespace, runs of inner whitespace and trailing '.' or '!' do not matter, so a phone's "Approve." is still an approval. Only for matching; what the operator wrote is what gets recorded and passed on.""" return " ".join(text.strip().rstrip(".!").split()).lower() class Gate: def __init__(self, revision: PlanRevision, operators: AbstractSet[str]) -> None: self._revision = revision self._operators = operators def evaluate( self, sender: str, text: str, channel: str, about: int | None = None ) -> Approved | Rejected | Ignored: """`about` is the revision that was up when the reply arrived; None only for records kept before replies were stamped.""" if sender not in self._operators: return Ignored(notice=None) text = text.strip() match = _APPROVE_RE.match(command_form(text)) named = match.group(1) if match else None current = self._revision if named is None and about is not None and about != current.rev: was = f"rev-{about}" if about else "no plan" if match is None: return Ignored( notice=( f"noted, but it arrived while {was} was up, so it is not " f"taken as feedback on {current.gate} {current.tag}" ) ) return Ignored( notice=( f"that approval arrived while {was} was up; the current plan is " f"{current.gate} {current.tag}, not applied. Reply again to " "approve it" ) ) if match is None: return Rejected(feedback=text, approver=sender, channel=channel) if named is not None and named != self._revision.tag: return Ignored( notice=( f"approval names {named}, but the current plan is " f"{self._revision.tag}; not applied" ) ) return Approved(revision=self._revision, approver=sender, channel=channel)