Model the orchestrator in Quint, and reproduce its counterexamples

specs/orchestrator.qnt models control.py: states, gate, pending
questions, comment cursors, publish intents and the worker queue, with
operator messages, Gitea comments, work outcomes (exceptions included)
and a crash at any point interleaving freely. Eight properties.

Seven counterexamples, each written out as a trace in
specs/orchestrator_scenarios.qnt and reproduced against the real code in
nisse-dev/tests/test_quint_counterexamples.py (xfail, strict, until
fixed):

1-4 an approval or feedback lands on a plan the operator had not seen:
    a double 'approve', 'approve rev-1' meant for the architecture,
    approving on both channels, and 'approve' plus an afterthought;
5   a crash after the PR opens but before PR_OPEN publishes again and
    cannot get back to PR_OPEN;
6   a crash while answering a review comment loses it silently;
7   'project complete' leaves the plan issue open.

no_stranding, await_input_has_a_question and execution_needs_approval
held over 200,000 random traces of 40 steps, and under Apalache's
exhaustive check to depth 8, which does find counterexample 1 at that
depth. No behaviour changes here.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE

nisse-dev/tests/test_quint_counterexamples.py

@@ -0,0 +1,327 @@
+"""The counterexamples found by specs/orchestrator.qnt, replayed against the
+real orchestrator. Each is marked xfail until its fix lands, and names the
+scenario in specs/orchestrator_scenarios.qnt it reproduces.
+
+The model is a hand-written reading of control.py; reproducing each trace
+here is what shows the flaw is in the design, not in the model.
+"""
+
+import io
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from nisse_dev.control import (
+    AWAIT_INPUT,
+    AWAIT_MILESTONE_SIGNOFF,
+    EXECUTING,
+    IDLE,
+    MILESTONE_PLANNING,
+    PR_OPEN,
+    Orchestrator,
+    StubAgent,
+)
+from nisse_dev.executor import ExecutorOutcome
+from nisse_dev.gitea import Comment, GiteaError, PullRequest
+from nisse_dev.oplog import OpLog
+from nisse_dev.poller import CommentPoller
+from nisse_dev.publish import PublishResult
+from nisse_dev.state import StateStore
+from nisse_dev.transport import InMemoryTransport
+from nisse_dev.worker import SerialWorker
+
+AGENT = "nisse-bot"
+
+
+class QueueingWorker(SerialWorker):
+    """Holds submitted jobs until released, so a test can queue two
+    messages before either runs, as a real worker does under load."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.held: list[Any] = []
+        self.holding = True
+
+    def submit(self, job: Any) -> None:
+        if self.holding:
+            self.held.append(job)
+        else:
+            job()
+
+    def call(self, job: Any, timeout: float | None = None) -> Any:
+        return job()
+
+    @property
+    def busy(self) -> bool:
+        return False
+
+    def release(self) -> None:
+        self.holding = False
+        jobs, self.held = self.held, []
+        for job in jobs:
+            job()
+
+
+class FakeGitea:
+    def __init__(self) -> None:
+        self.threads: dict[int, list[Comment]] = {}
+        self.prs: set[str] = set()
+        self._next = 100
+
+    def add(self, index: int, author: str, body: str) -> int:
+        self._next += 1
+        self.threads.setdefault(index, []).append(
+            Comment(id=self._next, author=author, body=body, created_at="t")
+        )
+        return self._next
+
+    def list_comments(self, repo: str, index: int, since: str | None = None) -> list:
+        return list(self.threads.get(index, []))
+
+
+class Board:
+    def __init__(self, gitea: FakeGitea) -> None:
+        self._gitea = gitea
+
+    def open_project_issue(self, project: str, task_id: str) -> int:
+        return 10
+
+    def post_plan(self, issue: int, gate: str, rev: int, plan: str) -> Any:
+        class Posted:
+            comment_id = self._gitea.add(issue, AGENT, f"{gate} rev-{rev}")
+
+        return Posted()
+
+    def echo(self, issue: int, text: str) -> None:
+        pass
+
+    def close(self, issue: int, note: str) -> None:
+        pass
+
+
+class Executor:
+    def __init__(self) -> None:
+        self.outcome = ExecutorOutcome("completed", "done")
+
+    def run(self, milestone: int) -> ExecutorOutcome:
+        return self.outcome
+
+
+class Publisher:
+    """Opens one PR per branch, as Gitea does; a second attempt fails."""
+
+    def __init__(self, gitea: FakeGitea, store: StateStore) -> None:
+        self._gitea = gitea
+        self._store = store
+        self.crash_after_opening = False
+        self.attempts = 0
+
+    def reconcilers(self) -> dict[str, Any]:
+        return {"open_pr": lambda intent: intent.payload["branch"] in self._gitea.prs}
+
+    def publish(self, **kwargs: Any) -> PublishResult:
+        branch = f"{kwargs['task_id']}/milestone-{kwargs['milestone']}"
+        self.attempts += 1
+        intent = self._store.begin_intent("open_pr", {"branch": branch}, cause="t")
+        if branch in self._gitea.prs:
+            raise GiteaError("pull request already exists for these targets")
+        self._gitea.prs.add(branch)
+        if self.crash_after_opening:
+            raise SystemExit("crash")  # the process dies here
+        self._store.commit_intent(intent, cause="opened")
+        return PublishResult(
+            PullRequest(number=20, state="open", head=branch, url="u"), "opened PR #20"
+        )
+
+    def push_followup(self, **kwargs: Any) -> str | None:
+        return "abc12345"
+
+    def comment(self, index: int, body: str) -> None:
+        pass
+
+
+class Agent:
+    def __init__(self, tmp_path: Path, gitea: FakeGitea, *, worker: Any = None) -> None:
+        self.store = StateStore(
+            tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())
+        )
+        self.transport = InMemoryTransport()
+        self.gitea = gitea
+        self.executor = Executor()
+        self.publisher = Publisher(gitea, self.store)
+        self.pollers: dict[int, CommentPoller] = {}
+
+        def factory(handler: Any) -> CommentPoller:
+            poller = CommentPoller(gitea, "o/r", handler, ignore_authors=[AGENT])
+
+            def start(index: int) -> None:
+                self.pollers[index] = poller
+
+            poller.start = start  # type: ignore[method-assign]
+            poller.stop = lambda: None  # type: ignore[method-assign]
+            return poller
+
+        self.orch = Orchestrator(
+            self.store,
+            self.transport,
+            StubAgent(),
+            operators={"don"},
+            gitea_operators={"don"},
+            board=Board(gitea),
+            executor=self.executor,
+            publisher=self.publisher,
+            poller_factory=factory,
+            **({"worker": worker} if worker else {}),
+        )
+        self.orch.start()
+
+    @property
+    def state(self) -> str:
+        return self.store.get_state().state
+
+    def gate(self) -> dict:
+        return self.store.get_state().data.get("gate") or {}
+
+    def say(self, *texts: str) -> None:
+        for text in texts:
+            self.transport.receive("don", text)
+
+
+XFAIL = pytest.mark.xfail(strict=True, reason="counterexample; fix pending")
+
+
+@XFAIL
+def test_double_approve_approves_an_unseen_plan(tmp_path: Path) -> None:
+    """doubleApproveApprovesAnUnseenPlanTest"""
+    worker = QueueingWorker()
+    agent = Agent(tmp_path, FakeGitea(), worker=worker)
+    worker.holding = False
+    agent.say("use demo")  # arch rev-1 is up
+    worker.holding = True
+    agent.say("approve", "approve")  # both queued, looking at arch rev-1
+    worker.release()
+    # The second approval must not approve the milestone plan, never seen.
+    assert not agent.gate().get("approved")
+
+
+@XFAIL
+def test_named_revision_approves_the_wrong_gate(tmp_path: Path) -> None:
+    """namedRevisionApprovesTheWrongGateTest"""
+    worker = QueueingWorker()
+    agent = Agent(tmp_path, FakeGitea(), worker=worker)
+    worker.holding = False
+    agent.say("use demo")
+    worker.holding = True
+    agent.say("approve", "approve rev-1")  # "rev-1" meant the architecture
+    worker.release()
+    assert not agent.gate().get("approved")
+
+
+@XFAIL
+def test_approval_on_both_channels_approves_the_next_plan(tmp_path: Path) -> None:
+    """approvalOnBothChannelsApprovesTheNextPlanTest"""
+    gitea = FakeGitea()
+    worker = QueueingWorker()
+    agent = Agent(tmp_path, gitea, worker=worker)
+    worker.holding = False
+    agent.say("use demo")
+    worker.holding = True
+    agent.say("approve")  # Matrix: queued
+    gitea.add(10, "don", "approve")  # and on the issue, about the same plan
+    # The poller delivers it; it waits behind the Matrix approval.
+    delivered = agent.pollers[10]
+    worker.held.append(lambda: delivered.poll_once(10))
+    worker.release()
+    assert agent.state == AWAIT_MILESTONE_SIGNOFF
+    assert not agent.gate().get("approved")
+
+
+@XFAIL
+def test_an_afterthought_rejects_the_next_plan(tmp_path: Path) -> None:
+    """anAfterthoughtRejectsTheNextPlanTest"""
+    worker = QueueingWorker()
+    agent = Agent(tmp_path, FakeGitea(), worker=worker)
+    worker.holding = False
+    agent.say("use demo")
+    worker.holding = True
+    agent.say("approve", "also, keep it in one file")
+    worker.release()
+    # The note was about the architecture; the milestone plan was not
+    # rejected by it.
+    assert agent.state == AWAIT_MILESTONE_SIGNOFF
+    assert agent.gate().get("rev") == 1
+
+
+@XFAIL
+def test_crash_after_opening_the_pr_publishes_again(tmp_path: Path) -> None:
+    """crashAfterOpeningThePrPublishesAgainTest"""
+    gitea = FakeGitea()
+    agent = Agent(tmp_path, gitea)
+    agent.say("use demo", "approve", "approve")
+    agent.publisher.crash_after_opening = True
+    with pytest.raises(SystemExit):
+        agent.say("start milestone 1")
+    assert agent.state == EXECUTING  # the PR exists; PR_OPEN was never recorded
+
+    agent = Agent(tmp_path, gitea)  # restart: the intent is verified
+    agent.say("continue", "start milestone 1")
+    # There is a PR for this milestone: the agent must end up watching it,
+    # not trying to open another.
+    assert agent.publisher.attempts == 0
+    assert agent.state == PR_OPEN
+
+
+@XFAIL
+def test_crash_during_a_review_loses_the_comment(tmp_path: Path) -> None:
+    """crashDuringAReviewLosesTheCommentTest"""
+    gitea = FakeGitea()
+    agent = Agent(tmp_path, gitea)
+    agent.say("use demo", "approve", "approve", "start milestone 1")
+    assert agent.state == PR_OPEN
+    gitea.add(20, "don", "please add a test")
+
+    class Crash(BaseException):
+        pass
+
+    def dies(milestone: int) -> ExecutorOutcome:
+        raise Crash()  # the process dies mid-review
+
+    agent.executor.run = dies  # type: ignore[method-assign]
+    with pytest.raises(Crash):
+        agent.pollers[20].poll_once(20)
+
+    agent = Agent(tmp_path, gitea)  # restart
+    assert agent.state == PR_OPEN
+    # The comment was taken but never answered: after the restart it must
+    # be handled again, or the operator told it was dropped.
+    delivered = agent.pollers[20].poll_once(20)
+    told = any(
+        "please add a test" in m.text or "review" in m.text
+        for m in agent.transport.outbound[1:]
+    )
+    assert delivered or told
+
+
+@XFAIL
+def test_project_complete_leaves_the_plan_issue_open(tmp_path: Path) -> None:
+    """projectCompleteLeavesThePlanIssueOpenTest"""
+    closed: list[int] = []
+    gitea = FakeGitea()
+    agent = Agent(tmp_path, gitea)
+    agent.orch._board.close = lambda issue, note: closed.append(issue)  # type: ignore[attr-defined]
+    agent.say("use demo", "approve", "approve", "start milestone 1", "project complete")
+    assert agent.state == IDLE
+    assert closed == [10]
+
+
+def test_the_harness_reaches_pr_open_without_faults(tmp_path: Path) -> None:
+    """Guard for the harness itself: with nothing queued or crashed, the
+    same flow is correct."""
+    agent = Agent(tmp_path, FakeGitea())
+    agent.say("use demo", "approve")
+    assert agent.state == AWAIT_MILESTONE_SIGNOFF
+    agent.say("approve", "start milestone 1")
+    assert agent.state == PR_OPEN
+    assert agent.publisher.attempts == 1
+    _ = (AWAIT_INPUT, MILESTONE_PLANNING)

specs/README.md

@@ -0,0 +1,38 @@
+# Formal models
+
+[Quint](https://quint-lang.org) models of designs where interleavings matter
+more than any single code path. A model is a hand-written reading of the code:
+it finds flaws the design and the code share, not places where the code
+strays from the design. So every counterexample is also reproduced as a failing
+test against the real code before anything is changed.
+
+## `orchestrator.qnt`: nisse-dev's control loop
+
+Models `nisse-dev/src/nisse_dev/control.py`: the states, the gate, pending
+questions, Gitea comment cursors, publish intents, and the worker queue.
+Operator messages, Gitea comments, planner/executor/publisher outcomes
+(exceptions included) and **a crash at any point** all interleave freely.
+Each pure function is named after the method it mirrors.
+
+The properties are at the end of the file (`all_properties` is their
+conjunction). Counterexamples are written out as short traces in
+`orchestrator_scenarios.qnt`, and replayed against the real code in
+`nisse-dev/tests/test_quint_counterexamples.py`.
+
+```sh
+npm install -g @informalsystems/quint     # 0.33 or later
+
+# Random simulation, one property at a time:
+quint run specs/orchestrator.qnt --invariant=no_stranding \
+    --max-steps=40 --max-samples=200000
+
+# The counterexample traces (each expects its property to break):
+quint test specs/orchestrator_scenarios.qnt
+```
+
+Exhaustive bounded checking (`quint verify`) uses Apalache and needs Java 17.
+
+**Keeping it true.** When `control.py` changes how a state is entered or
+left, change the matching function here in the same PR. When a
+counterexample's fix lands, flip its scenario's `expect` to the property
+holding, and drop the `xfail` from its Python reproduction.

specs/orchestrator.qnt

@@ -0,0 +1,461 @@
+// A model of nisse-dev's orchestrator (nisse-dev/src/nisse_dev/control.py),
+// written to look for counterexamples: interleavings of operator messages,
+// Gitea comments, work outcomes and crashes that break a property the
+// design relies on.
+//
+// Everything lives in one record, `s`, and each handler is a pure function
+// named after the method it mirrors. The model keeps what matters for the
+// properties below and drops the rest (texts, transcripts, models, tools).
+//
+// Durable state (the state store) survives a crash; volatile state (the
+// worker queue, work in flight, pollers, the cancel flag) does not. Gitea is
+// the outside world and survives everything.
+//
+// Run:  quint run specs/orchestrator.qnt --invariant=all_properties
+//       quint run specs/orchestrator.qnt --invariant=<one property>
+module orchestrator {
+
+  // ---- bounds --------------------------------------------------------------
+
+  pure val MAX_MS = 2        // milestones per task
+  pure val MAX_TASKS = 2     // tasks per run ('use' / 'abandon' / 'complete')
+  pure val MAX_QUEUE = 3     // jobs waiting on the worker
+  pure val MAX_IDS = 8       // Gitea comments per run
+
+  // ---- vocabulary ----------------------------------------------------------
+
+  pure val WORKING = Set("PLANNING_ARCH", "MILESTONE_PLANNING", "EXECUTING")
+  pure val SIGNOFF = Set("AWAIT_ARCH_SIGNOFF", "AWAIT_MILESTONE_SIGNOFF")
+  // Work that sets _working, so a message then gets "busy" (control._handle).
+  pure val LONG_WORK = Set("plan", "exec", "review")
+
+  // What the operator had in front of them: the plan last put up for sign-off.
+  type Plan = { g: str, rev: int }
+  pure val NO_PLAN: Plan = { g: "", rev: 0 }
+
+  type Gate = { on: bool, g: str, rev: int, approved: bool }
+  pure val NO_GATE: Gate = { on: false, g: "", rev: 0, approved: false }
+
+  type Pending = { kind: str, g: str, interrupted: str }
+  pure val NO_PENDING: Pending = { kind: "", g: "", interrupted: "" }
+
+  // A message or comment waiting on the worker. `saw` is what the operator
+  // had seen when they sent it; the agent never knows it, the properties do.
+  type Job = { src: str, kind: str, saw: Plan, cid: int }
+
+  type PlanComment = { id: int, tid: int, kind: str, saw: Plan }
+  type PrComment = { id: int, key: int }
+
+  type Watch = { on: bool, floor: int }
+  pure val OFF: Watch = { on: false, floor: 0 }
+
+  type Sys = {
+    // -- durable: the state store (control._data and the state) --
+    st: str,
+    tid: int,                 // task counter; a task's branches are its own
+    task: bool,               // a project is in use
+    gate: Gate,
+    revs: str -> int,
+    ms: int,
+    pending: Pending,
+    prOn: bool, prKey: int, prLast: int,          // _data["pr"]
+    boardOn: bool, boardPosted: bool, boardLast: int,  // _data["board"]
+    intents: Set[int],        // pending open_pr intents, by PR key
+    // -- Gitea --
+    planThread: List[PlanComment],
+    prThread: List[PrComment],
+    nextId: int,
+    giteaPr: int -> str,      // PR key -> "open" / "closed"
+    openIssues: Set[int],     // plan issues still open, by task
+    // -- volatile: lost in a crash --
+    queue: List[Job],
+    work: str,                // none / plan / exec / publish / review
+    workGate: str,
+    workCid: int,
+    pubStage: int,
+    cancel: bool,
+    planWatch: Watch,
+    prWatch: Watch,
+    // -- the operator's view, and history the properties read --
+    shown: Plan,
+    taken: Set[int],          // review comments the agent took up
+    answered: Set[int],       // ... and answered (pushed, asked, or reported)
+    badApproval: bool,
+    staleFeedback: bool,
+    dupPublish: bool,
+  }
+
+  // ---- helpers ------------------------------------------------------------
+
+  pure def msGate(ms: int): str = if (ms == 1) "ms1" else "ms2"
+  pure def prKeyOf(s: Sys): int = s.tid * 10 + s.ms
+  pure def prState(s: Sys, key: int): str =
+    if (s.giteaPr.keys().contains(key)) s.giteaPr.get(key) else "none"
+  pure def planIds(s: Sys): int =
+    s.planThread.foldl(0, (acc, c) => if (c.tid == s.tid and c.id > acc) c.id else acc)
+  pure def prIds(s: Sys, key: int): int =
+    s.prThread.foldl(0, (acc, c) => if (c.key == key and c.id > acc) c.id else acc)
+  pure def max(a: int, b: int): int = if (a > b) a else b
+  pure def queued(s: Sys, src: str): bool = s.queue.foldl(false, (acc, j) => acc or j.src == src)
+
+  // _await_input
+  pure def awaitInput(s: Sys, p: Pending): Sys = { ...s, st: "AWAIT_INPUT", pending: p }
+
+  // _settled: a job that leaves a working state behind with nothing running
+  // moves to AWAIT_INPUT with the restart choice.
+  pure def settle(s: Sys): Sys =
+    if (s.work == "none" and WORKING.contains(s.st))
+      awaitInput(s, { kind: "boot", g: "", interrupted: s.st })
+    else s
+
+  pure def endWork(s: Sys): Sys = { ...s, work: "none", workGate: "", workCid: 0, cancel: false }
+
+  // _run_planning_turn, first half: the turn is now in flight.
+  pure def startPlanning(s: Sys, g: str): Sys =
+    { ...s, st: if (g == "arch") "PLANNING_ARCH" else "MILESTONE_PLANNING",
+      work: "plan", workGate: g }
+
+  // _begin_project
+  pure def beginProject(s: Sys, giteaUp: bool): Sys =
+    val t = s.tid + 1
+    startPlanning(
+      { ...s, tid: t, task: true, gate: NO_GATE,
+        revs: Map("arch" -> 0, "ms1" -> 0, "ms2" -> 0), ms: 1, pending: NO_PENDING,
+        prOn: false, prKey: 0, prLast: 0,
+        boardOn: giteaUp, boardPosted: false, boardLast: 0,
+        openIssues: if (giteaUp) s.openIssues.union(Set(t)) else s.openIssues,
+        planWatch: OFF, prWatch: OFF },
+      "arch")
+
+  // _run_planning_turn, second half: the planner submitted a plan.
+  // _post_plan_to_board then watches the issue from here (a fresh watch
+  // skips what is already there).
+  pure def planSubmitted(s: Sys, giteaUp: bool): Sys =
+    val g = s.workGate
+    val rev = s.revs.get(g) + 1
+    val posted = s.boardOn and giteaUp
+    val t = endWork({ ...s,
+      revs: s.revs.set(g, rev),
+      gate: { on: true, g: g, rev: rev, approved: false },
+      st: if (g == "arch") "AWAIT_ARCH_SIGNOFF" else "AWAIT_MILESTONE_SIGNOFF",
+      shown: { g: g, rev: rev } })
+    if (posted)
+      { ...t, boardPosted: true, boardLast: planIds(t), planWatch: { on: true, floor: planIds(t) } }
+    else t
+
+  // _execute_milestone: start
+  pure def startExec(s: Sys): Sys = { ...s, st: "EXECUTING", work: "exec" }
+
+  // _apply_gate_reply, for Matrix and Gitea alike.
+  pure def applyGate(s: Sys, j: Job): Sys =
+    val current = { g: s.gate.g, rev: s.gate.rev }
+    if (s.gate.approved)
+      if (j.kind == "start") startExec(s) else s
+    else if (j.kind == "approve" or (j.kind == "approveRev" and j.saw.rev == s.gate.rev))
+      val s2 = { ...s, badApproval: s.badApproval or j.saw != current }
+      if (s.gate.g == "arch")
+        startPlanning({ ...s2, gate: NO_GATE }, msGate(s.ms))
+      else
+        { ...s2, gate: { ...s.gate, approved: true }, st: "AWAIT_MILESTONE_SIGNOFF" }
+    else if (j.kind == "approveRev")
+      s  // names another revision: ignored with a notice
+    else
+      // Anything else is rejection feedback, and the gate is planned again.
+      startPlanning({ ...s, staleFeedback: s.staleFeedback or j.saw != current, gate: NO_GATE },
+                    s.gate.g)
+
+  // _return_to_pr_open / _watch_pr
+  pure def returnToPrOpen(s: Sys): Sys =
+    { ...s, st: "PR_OPEN", prWatch: if (s.prOn) { on: true, floor: s.prLast } else OFF }
+
+  // _abandon: IDLE first, then close the issue and the PR, abort intents.
+  pure def abandon(s: Sys): Sys =
+    { ...s, st: "IDLE", task: false, gate: NO_GATE, pending: NO_PENDING,
+      prOn: false, boardOn: false, boardPosted: false, intents: Set(),
+      openIssues: s.openIssues.exclude(Set(s.tid)),
+      giteaPr: if (s.prOn) s.giteaPr.put(s.prKey, "closed") else s.giteaPr,
+      planWatch: OFF, prWatch: OFF }
+
+  // _handle_pr_open: 'project complete'
+  pure def complete(s: Sys): Sys =
+    { ...s, st: "IDLE", task: false, gate: NO_GATE, pending: NO_PENDING,
+      prOn: false, boardOn: false, boardPosted: false, planWatch: OFF, prWatch: OFF }
+
+  // _resume_after_restart
+  pure def resume(s: Sys, p: Pending, j: Job): Sys =
+    if (j.kind != "continue" and j.kind != "replan")
+      { ...s, pending: p }  // asked again
+    else if (p.interrupted == "PR_OPEN")
+      returnToPrOpen(s)
+    else if (j.kind == "continue" and p.interrupted == "EXECUTING" and s.gate.approved)
+      { ...s, st: "AWAIT_MILESTONE_SIGNOFF" }
+    else
+      val g =
+        if (p.interrupted == "PLANNING_ARCH") "arch"
+        else if (p.interrupted == "MILESTONE_PLANNING" or p.interrupted == "EXECUTING") msGate(s.ms)
+        else if (s.gate.g != "") s.gate.g
+        else if (s.revs.get("arch") == 0) "arch"
+        else msGate(s.ms)
+      startPlanning({ ...s, gate: NO_GATE }, g)
+
+  // _handle_answer
+  pure def answer(s: Sys, j: Job, giteaUp: bool): Sys =
+    val p = s.pending
+    val s0 = { ...s, pending: NO_PENDING }
+    if (p.kind == "")
+      if (j.kind == "use") beginProject({ ...s, planWatch: OFF, prWatch: OFF }, giteaUp) else s
+    else if (p.kind == "question" or p.kind == "planning_failed") startPlanning(s0, p.g)
+    else if (p.kind == "review_stopped") returnToPrOpen(s0)
+    else if (p.kind == "milestone_stopped") { ...s0, st: "AWAIT_MILESTONE_SIGNOFF" }
+    else if (p.kind == "boot") resume(s0, p, j)
+    else s0
+
+  // _handle_now, for a Matrix message the worker has reached.
+  pure def handleMatrix(s: Sys, j: Job, giteaUp: bool): Sys =
+    if (j.kind == "abandon")
+      if (s.st == "IDLE") s else abandon(s)
+    else if (s.st == "IDLE")
+      if (j.kind == "use") beginProject(s, giteaUp) else s
+    else if (SIGNOFF.contains(s.st))
+      if (not(s.gate.on))
+        awaitInput(s, { kind: "boot", g: "",
+          interrupted: if (s.st == "AWAIT_ARCH_SIGNOFF") "PLANNING_ARCH" else "MILESTONE_PLANNING" })
+      else applyGate(s, j)
+    else if (s.st == "PR_OPEN")
+      if (j.kind == "next" and s.ms < MAX_MS)
+        startPlanning({ ...s, ms: s.ms + 1, gate: NO_GATE, prOn: false, prWatch: OFF }, msGate(s.ms + 1))
+      else if (j.kind == "complete") complete(s)
+      else s
+    else if (s.st == "AWAIT_INPUT") answer(s, j, giteaUp)
+    else s  // busy
+
+  // _handle_plan_comment_now
+  pure def handlePlanComment(s: Sys, j: Job): Sys =
+    if (not(SIGNOFF.contains(s.st))) s  // not taken
+    else applyGate({ ...s, boardLast: max(s.boardLast, j.cid) }, j)
+
+  // _handle_review_comment_now, first half: the executor starts on it.
+  pure def handleReviewComment(s: Sys, j: Job): Sys =
+    if (s.st != "PR_OPEN") s  // not taken
+    else { ...s, prLast: max(s.prLast, j.cid), taken: s.taken.union(Set(j.cid)),
+           work: "review", workCid: j.cid }
+
+  pure def handleJob(s: Sys, j: Job, giteaUp: bool): Sys =
+    if (j.src == "matrix") handleMatrix(s, j, giteaUp)
+    else if (j.src == "plan") handlePlanComment(s, j)
+    else handleReviewComment(s, j)
+
+  // Orchestrator.start, after a crash: reconcile intents against Gitea,
+  // then resume, re-ask, or ask the restart question.
+  pure def boot(s: Sys, giteaUp: bool): Sys =
+    val unresolved = s.intents.size() > 0 and not(giteaUp)
+    val s0 = { ...s, intents: if (giteaUp) Set() else s.intents }
+    if (unresolved)
+      awaitInput(s0, { kind: "boot", g: "", interrupted: s.st })
+    else if (s.st == "AWAIT_INPUT" and s.pending.kind != "")
+      s0
+    else if (s.st == "PR_OPEN")
+      { ...s0, prWatch: if (s.prOn) { on: true, floor: s.prLast } else OFF }
+    else if (s.st == "IDLE" or (SIGNOFF.contains(s.st) and s.gate.on))
+      if (s.gate.on and s.boardPosted) { ...s0, planWatch: { on: true, floor: s.boardLast } } else s0
+    else
+      awaitInput(s0, { kind: "boot", g: "", interrupted: s.st })
+
+  // ---- the system ---------------------------------------------------------
+
+  var s: Sys
+
+  action init = s' = {
+    st: "IDLE", tid: 0, task: false, gate: NO_GATE,
+    revs: Map("arch" -> 0, "ms1" -> 0, "ms2" -> 0), ms: 1, pending: NO_PENDING,
+    prOn: false, prKey: 0, prLast: 0,
+    boardOn: false, boardPosted: false, boardLast: 0,
+    intents: Set(),
+    planThread: List(), prThread: List(), nextId: 1,
+    giteaPr: Map(), openIssues: Set(),
+    queue: List(), work: "none", workGate: "", workCid: 0, pubStage: 0, cancel: false,
+    planWatch: OFF, prWatch: OFF,
+    shown: NO_PLAN, taken: Set(), answered: Set(),
+    badApproval: false, staleFeedback: false, dupPublish: false,
+  }
+
+  // Each action has a deterministic form taking its choices as parameters,
+  // so scenarios (orchestrator_scenarios.qnt) can replay a trace exactly;
+  // `step` picks the parameters at random.
+
+  pure val MATRIX_KINDS = Set("use", "approve", "approveRev", "text", "start", "next",
+                              "complete", "abandon", "continue", "replan", "stop")
+  pure val COMMENT_KINDS = Set("approve", "approveRev", "text", "start")
+
+  // The operator sends a Matrix message (control._handle, on the
+  // transport's thread). While long work runs it only gets "busy"; `stop`
+  // reaches the cancel flag whenever the worker is busy.
+  action send(kind: str): bool = all {
+    s.tid < MAX_TASKS or kind != "use",
+    if (kind == "stop")
+      s' = { ...s, cancel: s.cancel or s.work != "none" }
+    else if (LONG_WORK.contains(s.work))
+      s' = s
+    else all {
+      s.queue.length() < MAX_QUEUE,
+      s' = { ...s, queue: s.queue.append({ src: "matrix", kind: kind, saw: s.shown, cid: 0 }) },
+    },
+  }
+
+  // The operator comments on the plan issue.
+  action commentPlan(kind: str): bool = all {
+    s.task, s.openIssues.contains(s.tid), s.nextId <= MAX_IDS,
+    s' = { ...s, nextId: s.nextId + 1,
+           planThread: s.planThread.append({ id: s.nextId, tid: s.tid, kind: kind, saw: s.shown }) },
+  }
+
+  // The operator comments on the open PR.
+  action commentPr: bool = all {
+    s.prOn, s.nextId <= MAX_IDS, prState(s, s.prKey) == "open",
+    s' = { ...s, nextId: s.nextId + 1, prThread: s.prThread.append({ id: s.nextId, key: s.prKey }) },
+  }
+
+  // A poller delivers the next new comment (CommentPoller.poll_once). It
+  // waits on the worker, so one job per poller at a time.
+  action pollPlan: bool = {
+    val c = s.planThread.foldl({ id: 0, tid: 0, kind: "", saw: NO_PLAN },
+      (acc, x) => if (x.tid == s.tid and x.id > s.planWatch.floor and (acc.id == 0 or x.id < acc.id)) x else acc)
+    all {
+      s.planWatch.on, c.id > 0, not(queued(s, "plan")), s.queue.length() < MAX_QUEUE,
+      s' = { ...s, planWatch: { on: true, floor: c.id },
+             queue: s.queue.append({ src: "plan", kind: c.kind, saw: c.saw, cid: c.id }) },
+    }
+  }
+
+  action pollPr: bool = {
+    val c = s.prThread.foldl({ id: 0, key: 0 },
+      (acc, x) => if (x.key == s.prKey and x.id > s.prWatch.floor and (acc.id == 0 or x.id < acc.id)) x else acc)
+    all {
+      s.prWatch.on, c.id > 0, not(queued(s, "pr")), s.queue.length() < MAX_QUEUE,
+      s' = { ...s, prWatch: { on: true, floor: c.id },
+             queue: s.queue.append({ src: "pr", kind: "review", saw: NO_PLAN, cid: c.id }) },
+    }
+  }
+
+  // The worker takes the next job, one at a time, and settles it.
+  action runJob(giteaUp: bool): bool = all {
+    s.work == "none", s.queue.length() > 0,
+    s' = settle(handleJob({ ...s, queue: s.queue.tail() }, s.queue.head(), giteaUp)),
+  }
+
+  // A planning turn ends.
+  action finishPlanning(outcome: str, giteaUp: bool): bool = all {
+    s.work == "plan", outcome != "stopped" or s.cancel,
+    s' =
+      if (outcome == "submit") planSubmitted(s, giteaUp)
+      else if (outcome == "ask" or outcome == "stopped")
+        awaitInput(endWork(s), { kind: "question", g: s.workGate, interrupted: "" })
+      else if (outcome == "error")
+        awaitInput(endWork(s), { kind: "planning_failed", g: s.workGate, interrupted: "" })
+      else settle(endWork(s)),
+  }
+
+  // The executor finishes a milestone.
+  action finishExec(outcome: str): bool = all {
+    s.work == "exec", outcome != "stopped" or s.cancel,
+    s' =
+      if (outcome == "completed") { ...s, work: "publish", pubStage: 0, cancel: false }
+      else if (outcome == "exception") settle(endWork(s))
+      else awaitInput(endWork(s), { kind: "milestone_stopped", g: "", interrupted: "" }),
+  }
+
+  // Publishing, one step at a time, so a crash can land between them
+  // (MilestonePublisher.publish and control._publish_milestone):
+  // 0 commit, begin_intent; 1 push, open the PR; 2 commit_intent;
+  // 3 record the PR, watch it, PR_OPEN.
+  action publishStep(outcome: str): bool = {
+    val key = prKeyOf(s)
+    all {
+      s.work == "publish",
+      s' =
+        if (s.pubStage == 0)
+          if (outcome == "nochanges" or outcome == "fail")
+            awaitInput(endWork(s), { kind: "milestone_stopped", g: "", interrupted: "" })
+          else if (outcome == "exception") settle(endWork(s))
+          else { ...s, intents: s.intents.union(Set(key)), pubStage: 1 }
+        else if (s.pubStage == 1)
+          if (outcome == "exception") settle(endWork(s))
+          else if (outcome == "fail" or outcome == "nochanges")
+            awaitInput(endWork(s), { kind: "milestone_stopped", g: "", interrupted: "" })
+          else if (prState(s, key) == "open")
+            // Gitea refuses a second PR for the branch: a GiteaError.
+            awaitInput(endWork({ ...s, dupPublish: true }),
+                       { kind: "milestone_stopped", g: "", interrupted: "" })
+          else { ...s, giteaPr: s.giteaPr.put(key, "open"), pubStage: 2 }
+        else if (s.pubStage == 2)
+          { ...s, intents: s.intents.exclude(Set(key)), pubStage: 3 }
+        else
+          endWork({ ...s, prOn: true, prKey: key, prLast: prIds(s, key),
+                    prWatch: { on: true, floor: prIds(s, key) }, planWatch: OFF,
+                    st: "PR_OPEN" }),
+    }
+  }
+
+  // The executor finishes responding to a review comment.
+  action finishReview(outcome: str): bool = all {
+    s.work == "review", outcome != "stopped" or s.cancel,
+    s' =
+      val t = endWork({ ...s, answered: s.answered.union(Set(s.workCid)) })
+      if (outcome == "stopped" or outcome == "escalate")
+        awaitInput(t, { kind: "review_stopped", g: "", interrupted: "" })
+      else t,
+  }
+
+  // The process dies and starts again: volatile state is gone.
+  action crash(giteaUp: bool): bool =
+    s' = boot({ ...s, queue: List(), work: "none", workGate: "", workCid: 0, pubStage: 0,
+                cancel: false, planWatch: OFF, prWatch: OFF }, giteaUp)
+
+  action step = {
+    nondet kind = MATRIX_KINDS.oneOf()
+    nondet ckind = COMMENT_KINDS.oneOf()
+    nondet up = Set(true, false).oneOf()
+    nondet planOutcome = Set("submit", "ask", "error", "stopped", "exception").oneOf()
+    nondet workOutcome = Set("completed", "stopped", "escalate", "exception").oneOf()
+    nondet pubOutcome = Set("ok", "nochanges", "fail", "exception").oneOf()
+    any {
+      send(kind), commentPlan(ckind), commentPr, pollPlan, pollPr, runJob(up),
+      finishPlanning(planOutcome, up), finishExec(workOutcome), publishStep(pubOutcome),
+      finishReview(workOutcome), crash(up),
+    }
+  }
+
+  // ---- properties ---------------------------------------------------------
+
+  // #66: a working state only while its work is running.
+  val no_stranding = WORKING.contains(s.st) implies s.work != "none"
+
+  // Waiting on the operator always means a question to answer.
+  val await_input_has_a_question = s.st == "AWAIT_INPUT" implies s.pending.kind != ""
+
+  // A milestone runs only on an approved plan for that milestone.
+  val execution_needs_approval =
+    s.st == "EXECUTING" implies (s.gate.on and s.gate.approved and s.gate.g == msGate(s.ms))
+
+  // An approval only ever approves the plan the operator was looking at.
+  val approval_only_of_what_was_seen = not(s.badApproval)
+
+  // Rejection feedback only ever lands on the plan it was written about.
+  val feedback_only_on_what_was_seen = not(s.staleFeedback)
+
+  // Never a second attempt to open a PR for a milestone that has one.
+  val one_pr_per_milestone = not(s.dupPublish)
+
+  // A review comment the agent took up is answered, or is being worked on.
+  val review_comments_answered =
+    s.taken.forall(id => s.answered.contains(id) or (s.work == "review" and s.workCid == id))
+
+  // With no task, no plan issue is left open.
+  val no_open_issue_without_a_task = not(s.task) implies s.openIssues.size() == 0
+
+  val all_properties = and {
+    no_stranding, await_input_has_a_question, execution_needs_approval,
+    approval_only_of_what_was_seen, feedback_only_on_what_was_seen,
+    one_pr_per_milestone, review_comments_answered, no_open_issue_without_a_task,
+  }
+}

specs/orchestrator_scenarios.qnt

@@ -0,0 +1,108 @@
+// Counterexamples found in orchestrator.qnt, written out as short traces.
+// Each ends by expecting the property it breaks to be false; once the design
+// is fixed, flip the expectation and keep the trace as a regression.
+//
+// Run:  quint test specs/orchestrator_scenarios.qnt   (runs every *Test)
+module orchestrator_scenarios {
+  import orchestrator.* from "./orchestrator"
+
+  // Architecture plan rev-1 is up for sign-off, posted to Gitea.
+  action toArchSignoff = init
+    .then(send("use")).then(runJob(true))
+    .then(finishPlanning("submit", true))
+
+  // The milestone plan is approved; the next word starts it.
+  action toMilestoneApproved = toArchSignoff
+    .then(send("approve")).then(runJob(true))
+    .then(finishPlanning("submit", true))
+    .then(send("approve")).then(runJob(true))
+
+  action toPrOpen = toMilestoneApproved
+    .then(send("start")).then(runJob(true))
+    .then(finishExec("completed"))
+    .then(publishStep("ok")).then(publishStep("ok"))
+    .then(publishStep("ok")).then(publishStep("ok"))
+
+  // 1. The operator sends "approve" twice, say a double tap or an impatient
+  //    repeat, while looking at the architecture plan. Both are queued
+  //    before either runs. The first approves the architecture; planning
+  //    the milestone runs; the second then approves the milestone plan,
+  //    which the operator has not seen.
+  run doubleApproveApprovesAnUnseenPlanTest =
+    toArchSignoff
+      .then(send("approve")).then(send("approve"))
+      .then(runJob(true))                    // approves arch rev-1
+      .then(finishPlanning("submit", true))  // milestone-1 rev-1 is up
+      .then(runJob(true))                    // the second "approve"
+      .expect(s.gate.g == "ms1" and s.gate.approved and not(approval_only_of_what_was_seen))
+
+  // 2. The same with the revision named: "approve rev-1" meant for the
+  //    architecture's rev-1 also matches the milestone's rev-1, because a
+  //    revision tag carries no gate name.
+  run namedRevisionApprovesTheWrongGateTest =
+    toArchSignoff
+      .then(send("approve")).then(send("approveRev"))
+      .then(runJob(true))
+      .then(finishPlanning("submit", true))
+      .then(runJob(true))
+      .expect(s.gate.approved and not(approval_only_of_what_was_seen))
+
+  // 3. Approval on both channels: the operator approves in Matrix and on
+  //    the Gitea issue. The Gitea approval is delivered and queued behind
+  //    the Matrix one, which approves the architecture and posts the
+  //    milestone plan. Re-attaching the plan watch does not recall a job
+  //    already queued, so the Gitea approval approves the milestone plan.
+  run approvalOnBothChannelsApprovesTheNextPlanTest =
+    toArchSignoff
+      .then(send("approve")).then(commentPlan("approve")).then(pollPlan)
+      .then(runJob(true))
+      .then(finishPlanning("submit", true))
+      .then(runJob(true))
+      .expect(s.gate.g == "ms1" and s.gate.approved and not(approval_only_of_what_was_seen))
+
+  // 4. An afterthought becomes a rejection: "approve", then "also, keep it
+  //    in one file", sent together. The note lands on the milestone plan
+  //    as rejection feedback and it is planned again.
+  run anAfterthoughtRejectsTheNextPlanTest =
+    toArchSignoff
+      .then(send("approve")).then(send("text"))
+      .then(runJob(true))
+      .then(finishPlanning("submit", true))
+      .then(runJob(true))
+      .expect(s.st == "MILESTONE_PLANNING" and not(feedback_only_on_what_was_seen))
+
+  // 5. A crash after the PR is opened but before PR_OPEN is recorded.
+  //    Boot finds the intent's PR and marks it done, but the state still
+  //    says EXECUTING, so it asks the restart question. 'continue' returns
+  //    to the approved plan, and starting it publishes again: Gitea refuses
+  //    a second PR for the branch, and there is no path back to PR_OPEN.
+  run crashAfterOpeningThePrPublishesAgainTest =
+    toMilestoneApproved
+      .then(send("start")).then(runJob(true))
+      .then(finishExec("completed"))
+      .then(publishStep("ok"))   // begin_intent
+      .then(publishStep("ok"))   // push, open the PR
+      .then(crash(true))         // intent verified and committed; state EXECUTING
+      .then(send("continue")).then(runJob(true))
+      .then(send("start")).then(runJob(true))
+      .then(finishExec("completed"))
+      .then(publishStep("ok")).then(publishStep("ok"))
+      .expect(s.st == "AWAIT_INPUT" and not(one_pr_per_milestone))
+
+  // 6. A crash while the executor answers a review comment. The comment's
+  //    cursor was saved before the work started (at most once), and a
+  //    restart in PR_OPEN resumes quietly, so the comment is never answered
+  //    and nobody is told.
+  run crashDuringAReviewLosesTheCommentTest =
+    toPrOpen
+      .then(commentPr).then(pollPr).then(runJob(true))  // the executor starts on it
+      .then(crash(true))
+      .expect(s.st == "PR_OPEN" and not(review_comments_answered))
+
+  // 7. A completed task leaves its plan issue open: 'project complete'
+  //    closes nothing, unlike 'abandon'.
+  run projectCompleteLeavesThePlanIssueOpenTest =
+    toPrOpen
+      .then(send("complete")).then(runJob(true))
+      .expect(s.st == "IDLE" and not(no_open_issue_without_a_task))
+}