Merge pull request 'Add control loop, gating primitives, and transport interface (A4/B1)' (#6) from control-loop-gating into main

Reviewed-on: dongar/nisse#6
Reviewed-by: Don Garrett <dgarrett@acm.org>

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

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 <project>'")
+        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
+        )

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)

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))

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

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)

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