diff --git a/CLAUDE.md b/CLAUDE.md index d66e690..6418e64 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -12,6 +12,9 @@ policy — are usually settled there. - **Ask when anything is unclear.** Do not assume the operator always knows exactly what they want; surfacing ambiguity beats guessing. +- **Plan first.** Before implementing a build-order step or any nontrivial + change, present a short implementation plan (scope, layout, API sketch, + deferrals) and get operator sign-off. Code lands as a Gitea PR afterwards. - **Script repeated tasks.** If a job takes the same multiple commands more than once, save it as a shell script (e.g. in `~/bin`) instead of retyping it. - **Strongly prefer automated tests over manual testing.** Every new feature or diff --git a/nisse-dev/src/nisse_dev/control.py b/nisse-dev/src/nisse_dev/control.py new file mode 100644 index 0000000..b9b093d --- /dev/null +++ b/nisse-dev/src/nisse_dev/control.py @@ -0,0 +1,309 @@ +"""Control loop and state machine (spec A4, B1 subset). + +Gating is structural, not advisory: the agent's only actions are ``SubmitPlan`` +(posts the plan for approval and ends the agent's turn) and ``AskHuman`` +(pause, ask, resume on answer). There is deliberately no proceed/start action — +advancement past a gate happens only here, when an operator approval or command +arrives over transport. Every position change is checkpointed through the state +store before the loop continues, so a restart lands mid-gate, not mid-air. +""" + +import uuid +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from typing import Any, Protocol + +from nisse_dev.gates import Approved, Gate, Ignored, PlanRevision, Rejected +from nisse_dev.reconcile import Reconciler, reconcile_on_boot +from nisse_dev.state import StateStore +from nisse_dev.transport import InboundMessage, Transport + +IDLE = "IDLE" +PLANNING_ARCH = "PLANNING_ARCH" +AWAIT_ARCH_SIGNOFF = "AWAIT_ARCH_SIGNOFF" +MILESTONE_PLANNING = "MILESTONE_PLANNING" +AWAIT_MILESTONE_SIGNOFF = "AWAIT_MILESTONE_SIGNOFF" +EXECUTING = "EXECUTING" +AWAIT_INPUT = "AWAIT_INPUT" + +TRANSITIONS = { + (IDLE, PLANNING_ARCH), + (PLANNING_ARCH, AWAIT_ARCH_SIGNOFF), + (AWAIT_ARCH_SIGNOFF, PLANNING_ARCH), # rejected: replan with feedback + (AWAIT_ARCH_SIGNOFF, MILESTONE_PLANNING), + (MILESTONE_PLANNING, AWAIT_MILESTONE_SIGNOFF), + (AWAIT_MILESTONE_SIGNOFF, MILESTONE_PLANNING), + (AWAIT_MILESTONE_SIGNOFF, EXECUTING), +} + + +class TransitionError(RuntimeError): + pass + + +@dataclass(frozen=True) +class SubmitPlan: + content: str + + +@dataclass(frozen=True) +class AskHuman: + question: str + + +class PlanningAgent(Protocol): + def plan( + self, + gate: str, + project: str, + feedback: str | None = None, + answer: str | None = None, + ) -> SubmitPlan | AskHuman: ... + + +class StubAgent: + """Echoes canned plans so the approval loop is exercisable with no LLM.""" + + def plan( + self, + gate: str, + project: str, + feedback: str | None = None, + answer: str | None = None, + ) -> SubmitPlan | AskHuman: + notes = "" + if feedback: + notes += f" (revised after feedback: {feedback})" + if answer: + notes += f" (operator answered: {answer})" + return SubmitPlan(f"[stub] {gate} plan for {project}{notes}") + + +def _fresh_data() -> dict[str, Any]: + return { + "revs": {}, + "gate": None, + "question": None, + "milestone": 1, + "task_id": None, + } + + +class Orchestrator: + def __init__( + self, + store: StateStore, + transport: Transport, + agent: PlanningAgent, + operators: Iterable[str], + ) -> None: + self._store = store + self._transport = transport + self._agent = agent + self._operators = set(operators) + self._data = _fresh_data() + self._project: str | None = None + transport.on_message(self._handle) + + def start(self, reconcilers: Mapping[str, Reconciler] | None = None) -> None: + report = reconcile_on_boot(self._store, reconcilers or {}) + state = self._store.get_state() + if state is None: + self._transition(IDLE, cause="boot: fresh state") + self._transport.send_message("nisse-dev ready (idle)") + return + self._data = {**_fresh_data(), **state.data} + self._project = state.project + self._store.oplog.task_id = self._data.get("task_id") + gate = self._data.get("gate") + if report.action == "await_input": + self._transition(AWAIT_INPUT, cause="boot: unresolved intents") + self._transport.ask_question( + "restart found unresolved work: " + + "; ".join(report.notes) + + " — how should I proceed?" + ) + elif state.state == IDLE or ( + state.state in (AWAIT_ARCH_SIGNOFF, AWAIT_MILESTONE_SIGNOFF) and gate + ): + # A gate is a safe place to have been interrupted: nothing was in + # flight, the plan and revision are durable. Announce and resume. + if gate: + self._transport.send_message( + f"restarted while awaiting sign-off on {gate['name']} " + f"rev-{gate['rev']}; the gate is still open" + ) + else: + # Interrupted mid-turn (planning, executing, unanswered question): + # never blind-resume (A3.6). + self._transition(AWAIT_INPUT, cause=f"boot: restarted in {state.state}") + self._transport.ask_question( + f"restarted mid-{state.state}; continue or re-plan?" + ) + + def _handle(self, msg: InboundMessage) -> None: + if msg.sender not in self._operators: + return + state = self._store.get_state() + current = state.state if state else IDLE + text = msg.text.strip() + if text == "status": + self._send_status(current) + elif current == IDLE: + if text.startswith("use "): + self._begin_project(text[4:].strip()) + else: + self._transport.send_message("no active project; say 'use '") + elif current in (AWAIT_ARCH_SIGNOFF, AWAIT_MILESTONE_SIGNOFF): + self._handle_gate_reply(msg, text) + elif current == AWAIT_INPUT: + self._handle_answer(text) + else: + self._transport.send_message(f"busy ({current}); 'status' for details") + + def _send_status(self, current: str) -> None: + gate = self._data.get("gate") + gate_note = ( + f", awaiting sign-off on {gate['name']} rev-{gate['rev']}" if gate else "" + ) + self._transport.send_message( + f"state={current} project={self._project}{gate_note}" + ) + + def _begin_project(self, project: str) -> None: + task_id = f"{project}-{uuid.uuid4().hex[:8]}" + self._data = {**_fresh_data(), "task_id": task_id} + self._project = project + self._store.oplog.task_id = task_id + self._transition(PLANNING_ARCH, cause=f"operator command: use {project}") + self._run_planning_turn("arch") + + def _run_planning_turn( + self, gate_name: str, feedback: str | None = None, answer: str | None = None + ) -> None: + assert self._project is not None + action = self._agent.plan( + gate_name, self._project, feedback=feedback, answer=answer + ) + if isinstance(action, AskHuman): + self._data["question"] = {"text": action.question, "gate": gate_name} + self._transition( + AWAIT_INPUT, cause=f"agent question during {gate_name} planning" + ) + self._transport.ask_question(action.question) + return + rev = self._data["revs"].get(gate_name, 0) + 1 + self._data["revs"][gate_name] = rev + self._data["gate"] = { + "name": gate_name, + "rev": rev, + "plan": action.content, + "approved": False, + } + await_state = ( + AWAIT_ARCH_SIGNOFF if gate_name == "arch" else AWAIT_MILESTONE_SIGNOFF + ) + self._transition(await_state, cause=f"plan submitted ({gate_name} rev-{rev})") + self._transport.send_approval_request( + f"{action.content}\n\nreply 'approve' to approve rev-{rev}; " + "any other reply is rejection feedback" + ) + self._store.oplog.emit( + "gate.open", + cause=f"plan submitted (rev-{rev})", + gate=gate_name, + rev=f"rev-{rev}", + channel="transport", + ) + + def _handle_gate_reply(self, msg: InboundMessage, text: str) -> None: + gate_data = self._data["gate"] + name, rev = gate_data["name"], gate_data["rev"] + milestone = self._data["milestone"] + if gate_data["approved"]: + if text == f"start milestone {milestone}": + self._transition( + EXECUTING, cause=f"operator command: start milestone {milestone}" + ) + self._transport.send_message(f"[stub] executing milestone {milestone}") + else: + self._transport.send_message( + f"plan approved; say 'start milestone {milestone}' to begin" + ) + return + revision = PlanRevision(gate=name, rev=rev, content=gate_data["plan"]) + result = Gate(revision, self._operators).evaluate( + msg.sender, text, channel="transport" + ) + if isinstance(result, Ignored): + if result.notice: + self._transport.send_message(result.notice) + return + if isinstance(result, Rejected): + self._store.oplog.emit( + "gate.reject", + cause=f"rejected by {result.approver}", + gate=name, + rev=f"rev-{rev}", + channel=result.channel, + ) + self._data["gate"] = None + back = PLANNING_ARCH if name == "arch" else MILESTONE_PLANNING + self._transition(back, cause=f"plan rejected ({name} rev-{rev})") + self._run_planning_turn(name, feedback=result.feedback) + return + assert isinstance(result, Approved) + self._store.oplog.emit( + "gate.approve", + cause=f"approved by {result.approver}", + gate=name, + rev=f"rev-{rev}", + channel=result.channel, + ) + if name == "arch": + self._data["gate"] = None + self._transition( + MILESTONE_PLANNING, cause=f"arch plan approved (rev-{rev})" + ) + self._transport.send_message( + f"architecture approved; planning milestone {milestone}" + ) + self._run_planning_turn(f"milestone-{milestone}") + else: + gate_data["approved"] = True + self._transition( + AWAIT_MILESTONE_SIGNOFF, + cause=f"milestone plan approved (rev-{rev})", + ) + self._transport.send_message( + f"milestone plan approved; say 'start milestone {milestone}' to begin" + ) + + def _handle_answer(self, text: str) -> None: + question = self._data.get("question") + if question is None: + self._transport.send_message( + "awaiting operator resolution; no pending question to answer" + ) + return + self._data["question"] = None + gate_name = question["gate"] + back = PLANNING_ARCH if gate_name == "arch" else MILESTONE_PLANNING + self._transition(back, cause="operator answered question") + self._run_planning_turn(gate_name, answer=text) + + def _transition(self, new_state: str, *, cause: str) -> None: + state = self._store.get_state() + current = state.state if state else None + allowed = ( + current is None + or current == new_state + or new_state == AWAIT_INPUT + or current == AWAIT_INPUT + or (current, new_state) in TRANSITIONS + ) + if not allowed: + raise TransitionError(f"illegal transition {current} -> {new_state}") + self._store.transition( + new_state, cause=cause, project=self._project, data=self._data + ) diff --git a/nisse-dev/src/nisse_dev/gates.py b/nisse-dev/src/nisse_dev/gates.py new file mode 100644 index 0000000..eb66853 --- /dev/null +++ b/nisse-dev/src/nisse_dev/gates.py @@ -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) diff --git a/nisse-dev/src/nisse_dev/transport.py b/nisse-dev/src/nisse_dev/transport.py new file mode 100644 index 0000000..c52a99b --- /dev/null +++ b/nisse-dev/src/nisse_dev/transport.py @@ -0,0 +1,56 @@ +"""Transport interface (spec A1): all human-to-agent conversation. + +The orchestrator depends only on this interface; no platform SDK is imported +anywhere outside a concrete implementation. MatrixTransport arrives as its own +class + config with no orchestrator changes. +""" + +from collections.abc import Callable +from dataclasses import dataclass +from typing import Protocol + + +@dataclass(frozen=True) +class InboundMessage: + sender: str + text: str + + +@dataclass(frozen=True) +class OutboundMessage: + kind: str # "message" | "approval_request" | "question" + text: str + + +class Transport(Protocol): + def send_message(self, text: str) -> None: ... + + def send_approval_request(self, text: str) -> None: ... + + def ask_question(self, text: str) -> None: ... + + def on_message(self, callback: Callable[[InboundMessage], None]) -> None: ... + + +class InMemoryTransport: + """Records outbound traffic and lets tests inject inbound messages.""" + + def __init__(self) -> None: + self.outbound: list[OutboundMessage] = [] + self._callback: Callable[[InboundMessage], None] | None = None + + def send_message(self, text: str) -> None: + self.outbound.append(OutboundMessage("message", text)) + + def send_approval_request(self, text: str) -> None: + self.outbound.append(OutboundMessage("approval_request", text)) + + def ask_question(self, text: str) -> None: + self.outbound.append(OutboundMessage("question", text)) + + def on_message(self, callback: Callable[[InboundMessage], None]) -> None: + self._callback = callback + + def receive(self, sender: str, text: str) -> None: + assert self._callback is not None, "no message handler registered" + self._callback(InboundMessage(sender, text)) diff --git a/nisse-dev/tests/test_control.py b/nisse-dev/tests/test_control.py new file mode 100644 index 0000000..131ec19 --- /dev/null +++ b/nisse-dev/tests/test_control.py @@ -0,0 +1,155 @@ +import io +import json +from pathlib import Path + +import pytest + +from nisse_dev.control import ( + AWAIT_ARCH_SIGNOFF, + AWAIT_INPUT, + AWAIT_MILESTONE_SIGNOFF, + EXECUTING, + IDLE, + AskHuman, + Orchestrator, + StubAgent, + SubmitPlan, +) +from nisse_dev.oplog import OpLog +from nisse_dev.state import StateStore +from nisse_dev.transport import InMemoryTransport + + +class Harness: + def __init__(self, tmp_path: Path, agent: object | None = None) -> None: + self.stream = io.StringIO() + self.store = StateStore( + tmp_path / "state.db", oplog=OpLog("nisse-dev", self.stream) + ) + self.transport = InMemoryTransport() + self.orch = Orchestrator( + self.store, self.transport, agent or StubAgent(), operators={"don"} + ) + self.orch.start() + + @property + def state(self) -> str: + state = self.store.get_state() + assert state is not None + return state.state + + def events(self, kind: str) -> list[dict]: + return [ + e + for e in map(json.loads, self.stream.getvalue().splitlines()) + if e["event"] == kind + ] + + def outbound_texts(self) -> list[str]: + return [m.text for m in self.transport.outbound] + + +@pytest.fixture +def h(tmp_path: Path) -> Harness: + return Harness(tmp_path) + + +def test_boots_idle_and_greets(h: Harness) -> None: + assert h.state == IDLE + assert "ready" in h.outbound_texts()[0] + + +def test_use_project_submits_arch_plan_and_opens_gate(h: Harness) -> None: + h.transport.receive("don", "use demo") + assert h.state == AWAIT_ARCH_SIGNOFF + request = h.transport.outbound[-1] + assert request.kind == "approval_request" + assert "rev-1" in request.text + [opened] = h.events("gate.open") + assert opened["gate"] == "arch" + assert opened["rev"] == "rev-1" + assert opened["task_id"] is not None + + +def test_full_happy_path_to_executing(h: Harness) -> None: + h.transport.receive("don", "use demo") + h.transport.receive("don", "approve") + assert h.state == AWAIT_MILESTONE_SIGNOFF + h.transport.receive("don", "approve") + assert h.state == AWAIT_MILESTONE_SIGNOFF # approved, awaiting start + h.transport.receive("don", "start milestone 1") + assert h.state == EXECUTING + approvals = h.events("gate.approve") + assert [e["gate"] for e in approvals] == ["arch", "milestone-1"] + + +def test_reject_replans_with_feedback_and_bumps_revision(h: Harness) -> None: + h.transport.receive("don", "use demo") + h.transport.receive("don", "too vague, add tests") + assert h.state == AWAIT_ARCH_SIGNOFF + request = h.transport.outbound[-1] + assert "rev-2" in request.text + assert "too vague, add tests" in request.text + [rejected] = h.events("gate.reject") + assert rejected["rev"] == "rev-1" + + +def test_stale_approval_is_refused_then_current_works(h: Harness) -> None: + h.transport.receive("don", "use demo") + h.transport.receive("don", "needs work") # now at rev-2 + h.transport.receive("don", "approve rev-1") + assert h.state == AWAIT_ARCH_SIGNOFF + assert "not applied" in h.outbound_texts()[-1] + assert h.events("gate.approve") == [] + h.transport.receive("don", "approve rev-2") + assert h.state == AWAIT_MILESTONE_SIGNOFF + + +def test_non_operator_messages_are_ignored(h: Harness) -> None: + h.transport.receive("don", "use demo") + before = len(h.transport.outbound) + h.transport.receive("mallory", "approve") + assert h.state == AWAIT_ARCH_SIGNOFF + assert len(h.transport.outbound) == before + + +def test_start_milestone_requires_approval_first(h: Harness) -> None: + h.transport.receive("don", "use demo") + h.transport.receive("don", "approve") + h.transport.receive("don", "start milestone 1") + assert h.state == AWAIT_MILESTONE_SIGNOFF # it was rejection feedback + h.transport.receive("don", "approve rev-2") + h.transport.receive("don", "start milestone 1") + assert h.state == EXECUTING + + +def test_status_reports_state_and_gate(h: Harness) -> None: + h.transport.receive("don", "use demo") + h.transport.receive("don", "status") + status = h.outbound_texts()[-1] + assert "AWAIT_ARCH_SIGNOFF" in status + assert "demo" in status + assert "rev-1" in status + + +class QuestionThenPlan: + def plan( + self, + gate: str, + project: str, + feedback: str | None = None, + answer: str | None = None, + ) -> SubmitPlan | AskHuman: + if answer is None: + return AskHuman("monorepo or polyrepo?") + return SubmitPlan(f"[stub] {gate} plan using {answer}") + + +def test_ask_human_pauses_and_answer_resumes(tmp_path: Path) -> None: + h = Harness(tmp_path, agent=QuestionThenPlan()) + h.transport.receive("don", "use demo") + assert h.state == AWAIT_INPUT + assert h.transport.outbound[-1].kind == "question" + h.transport.receive("don", "monorepo") + assert h.state == AWAIT_ARCH_SIGNOFF + assert "monorepo" in h.transport.outbound[-1].text diff --git a/nisse-dev/tests/test_gates.py b/nisse-dev/tests/test_gates.py new file mode 100644 index 0000000..f76ff16 --- /dev/null +++ b/nisse-dev/tests/test_gates.py @@ -0,0 +1,48 @@ +from nisse_dev.gates import Approved, Gate, Ignored, PlanRevision, Rejected + +REVISION = PlanRevision(gate="arch", rev=2, content="the plan") +OPERATORS = {"don"} + + +def gate() -> Gate: + return Gate(REVISION, OPERATORS) + + +def test_bare_approve_binds_to_current_revision() -> None: + result = gate().evaluate("don", "approve", channel="transport") + assert isinstance(result, Approved) + assert result.revision.tag == "rev-2" + assert result.approver == "don" + + +def test_tagged_approve_matching_current_revision() -> None: + result = gate().evaluate("don", "approve rev-2", channel="gitea") + assert isinstance(result, Approved) + assert result.channel == "gitea" + + +def test_stale_revision_tag_is_ignored_with_notice() -> None: + result = gate().evaluate("don", "approve rev-1", channel="transport") + assert isinstance(result, Ignored) + assert result.notice is not None + assert "rev-1" in result.notice + assert "rev-2" in result.notice + + +def test_any_other_reply_is_rejection_with_feedback() -> None: + result = gate().evaluate("don", "too vague, split milestone 1", "transport") + assert isinstance(result, Rejected) + assert result.feedback == "too vague, split milestone 1" + + +def test_unknown_sender_is_silently_ignored() -> None: + result = gate().evaluate("mallory", "approve", channel="transport") + assert isinstance(result, Ignored) + assert result.notice is None + + +def test_approve_with_trailing_words_is_rejection() -> None: + # "approve, but rename the module" is feedback, not approval (A4: the + # typed word approve, nothing else). + result = gate().evaluate("don", "approve, but rename the module", "transport") + assert isinstance(result, Rejected) diff --git a/tests/integration/test_gate_restart.py b/tests/integration/test_gate_restart.py new file mode 100644 index 0000000..16b2a02 --- /dev/null +++ b/tests/integration/test_gate_restart.py @@ -0,0 +1,65 @@ +"""Integration: a restart mid-gate resumes the open gate with its revision +history intact; unresolved intents gate the whole boot to AWAIT_INPUT +(spec A2/A3/A4 together).""" + +import io +from pathlib import Path + +import pytest + +from nisse_dev.control import ( + AWAIT_ARCH_SIGNOFF, + AWAIT_INPUT, + AWAIT_MILESTONE_SIGNOFF, + Orchestrator, + StubAgent, +) +from nisse_dev.oplog import OpLog +from nisse_dev.state import StateStore +from nisse_dev.transport import InMemoryTransport + +pytestmark = pytest.mark.integration + + +def boot(db: Path) -> tuple[Orchestrator, StateStore, InMemoryTransport]: + store = StateStore(db, oplog=OpLog("nisse-dev", io.StringIO())) + transport = InMemoryTransport() + orch = Orchestrator(store, transport, StubAgent(), operators={"don"}) + orch.start() + return orch, store, transport + + +def test_restart_mid_gate_resumes_with_revisions_intact(tmp_path: Path) -> None: + db = tmp_path / "state.db" + _, store, transport = boot(db) + transport.receive("don", "use demo") + transport.receive("don", "needs work") # arch gate now at rev-2 + task_id = store.oplog.task_id + store.close() + + _, store2, transport2 = boot(db) + state = store2.get_state() + assert state is not None + assert state.state == AWAIT_ARCH_SIGNOFF + assert store2.oplog.task_id == task_id # correlation survives restart + notices = [m.text for m in transport2.outbound] + assert any("rev-2" in n and "still open" in n for n in notices) + + transport2.receive("don", "approve rev-1") # stale across the restart too + assert store2.get_state().state == AWAIT_ARCH_SIGNOFF + transport2.receive("don", "approve rev-2") + assert store2.get_state().state == AWAIT_MILESTONE_SIGNOFF + + +def test_restart_with_unresolved_intent_gates_boot(tmp_path: Path) -> None: + db = tmp_path / "state.db" + _, store, transport = boot(db) + transport.receive("don", "use demo") + store.begin_intent("push_branch", {"branch": "m1"}, cause="test") + store.close() + + _, store2, transport2 = boot(db) + assert store2.get_state().state == AWAIT_INPUT + question = transport2.outbound[-1] + assert question.kind == "question" + assert "unresolved work" in question.text