From d79e399933210e568f86b5108f50c0ce8628adf5 Mon Sep 17 00:00:00 2001 From: Don Garrett Date: Tue, 29 Sep 2026 18:51:59 +0000 Subject: [PATCH 1/2] 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) Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE --- nisse-dev/tests/test_quint_counterexamples.py | 327 +++++++++++++ specs/README.md | 38 ++ specs/orchestrator.qnt | 461 ++++++++++++++++++ specs/orchestrator_scenarios.qnt | 108 ++++ 4 files changed, 934 insertions(+) create mode 100644 nisse-dev/tests/test_quint_counterexamples.py create mode 100644 specs/README.md create mode 100644 specs/orchestrator.qnt create mode 100644 specs/orchestrator_scenarios.qnt diff --git a/nisse-dev/tests/test_quint_counterexamples.py b/nisse-dev/tests/test_quint_counterexamples.py new file mode 100644 index 0000000..5373cbf --- /dev/null +++ b/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) diff --git a/specs/README.md b/specs/README.md new file mode 100644 index 0000000..9d53979 --- /dev/null +++ b/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. diff --git a/specs/orchestrator.qnt b/specs/orchestrator.qnt new file mode 100644 index 0000000..ca38283 --- /dev/null +++ b/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= +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, + } +} diff --git a/specs/orchestrator_scenarios.qnt b/specs/orchestrator_scenarios.qnt new file mode 100644 index 0000000..49fc6a7 --- /dev/null +++ b/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)) +} -- 2.54.0 From 9f38e7a4671b31f66695a61d8d52f770f8348fb2 Mon Sep 17 00:00:00 2001 From: Don Garrett Date: Tue, 29 Sep 2026 18:52:06 +0000 Subject: [PATCH 2/2] Ignore Apalache's output directory Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE --- .gitignore | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.gitignore b/.gitignore index 46559d1..71304fb 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,6 @@ __pycache__/ .ruff_cache/ dist/ nisse-dev.yaml + +# Apalache output (quint verify) +_apalache-out/ -- 2.54.0