Add control loop, gating primitives, and transport interface (A4/B1)
CI / checks (push) Successful in 9s
CI / checks (pull_request) Successful in 9s

Step-2 PR A: the approval loop end to end with a stub agent, no LLM
and no network.

- transport.py: Transport interface per A1 (send_message,
  send_approval_request, ask_question, on_message) with an
  InMemoryTransport; no platform SDK outside an implementation
- gates.py: PlanRevision with monotonic rev-N tags; Gate.evaluate
  enforces revision binding (stale tags refused with a notice),
  operator allowlisting, and approve-word-only semantics; any other
  reply is rejection feedback. Channel seam ready for Gitea/Vikunja
  approval sources
- control.py: B1 state machine subset with structural gating — the
  agent's only actions are SubmitPlan (ends its turn) and AskHuman;
  no proceed tool exists. Gate state, revision counters, task_id,
  and pending questions checkpoint through StateStore, and boot
  resumes an open gate or gates to AWAIT_INPUT, never blind-resumes
- gate.open/approve/reject A9 events complete the log taxonomy
- CLAUDE.md: record the plan-first working convention
- Tests: 15 unit (gates + orchestrator flows), 2 integration
  (mid-gate restart with revisions intact; unresolved intents gate
  the whole boot)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
dongarandClaude Fable 5 committed 2026-09-05 21:28:07 +00:00
1 parent 78122c407e
commit a251104a44
7 files changed
+711

No files matched your search

+75
View File
@@ -0,0 +1,75 @@
"""Approval gating primitives (spec A4).
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.
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+))?$")
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
) -> Approved | Rejected | Ignored:
if sender not in self._operators:
return Ignored(notice=None)
text = text.strip()
match = _APPROVE_RE.match(text)
if match is None:
return Rejected(feedback=text, approver=sender, channel=channel)
named = match.group(1)
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)