Add the operator kill switch, stop (A4)

`stop` could not have worked by adding the command alone. Planning
turns, milestones and review responses ran synchronously on the thread
that delivered the message, so the Matrix sync loop read nothing until
the work finished. The state store was also opened on the main thread
and used from the transport and poller threads, which SQLite refuses
with its default same-thread check; the in-memory test transport never
showed it.

A single worker thread now runs every handler and owns the store:

- The worker opens the store, so the same-thread check stays on and
  fails loudly if anything touches the store from elsewhere
- Matrix messages are handed to the worker; pollers hand over and wait,
  so their cursor never passes an unhandled comment
- A message that arrives during long work gets an immediate "busy; say
  'stop' to halt" reply and is recorded, not applied later in a state
  the operator was not answering
- Tests keep running inline through InlineWorker, which defers a job
  submitted from inside another as the real worker would queue it

`stop` sets a shared Cancel flag straight from the transport's thread.
The executor and delegation loops check it before each model call and
between tool calls; the planner checks it before each turn. Calls left
unrun get a result saying so, so the transcript still renders. A
stopped delegation is discarded like a failed one. The orchestrator
moves to AWAIT_INPUT and asks why the work was stopped. The reply is
recorded for the model, and a stopped planning turn resumes with it as
the answer. `stop` with nothing running says so and changes nothing; at
a gate it is no longer taken as rejection feedback.

Closes #40

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

nisse-dev/src/nisse_dev/control.py

@@ -6,14 +6,20 @@ Gating is structural, not advisory: the agent's only actions are ``SubmitPlan``
 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.
+
+All of it runs on one worker (nisse_dev.worker), never on the thread that
+delivered a message. The one exception is `stop` (A4), which has to reach
+work that is already running: the transport's thread sets the cancel flag
+directly, and the worker records the stop once the work has returned.
 """
 
-import threading
 import uuid
-from collections.abc import Iterable, Mapping
+from collections.abc import Iterable, Iterator, Mapping
+from contextlib import contextmanager
 from dataclasses import dataclass
 from typing import Any, Protocol
 
+from nisse_dev.cancel import Cancel
 from nisse_dev.canonical import GateOpened, PlanSubmitted, UserMessage
 from nisse_dev.gates import Approved, Gate, Ignored, PlanRevision, Rejected
 from nisse_dev.git import GitError
@@ -22,6 +28,7 @@ from nisse_dev.llm import RoleMap
 from nisse_dev.reconcile import Reconciler, reconcile_on_boot
 from nisse_dev.state import StateStore
 from nisse_dev.transport import InboundMessage, Transport
+from nisse_dev.worker import InlineWorker, Worker
 
 IDLE = "IDLE"
 PLANNING_ARCH = "PLANNING_ARCH"
@@ -59,6 +66,10 @@ class PlanningError(RuntimeError):
     plan or retrying forever."""
 
 
+class PlanningStopped(RuntimeError):
+    """The operator said `stop` (A4) before a planning turn could run."""
+
+
 @dataclass(frozen=True)
 class SubmitPlan:
     content: str
@@ -170,6 +181,8 @@ class Orchestrator:
         delegate: Any = None,
         poller_factory: Any = None,
         catalog: Any = None,
+        worker: Worker | None = None,
+        cancel: Cancel | None = None,
     ) -> None:
         self._store = store
         self._transport = transport
@@ -190,9 +203,20 @@ class Orchestrator:
         # the factory owns how to build one.
         self._poller_factory = poller_factory
         self._pollers: dict[str, Any] = {}
-        # Transport and the comment poller deliver on separate threads;
-        # state transitions must not interleave.
-        self._lock = threading.RLock()
+        # Transport and the comment poller deliver on their own threads, and
+        # every handler runs on the worker, one at a time, so state
+        # transitions never interleave.
+        self._worker = worker or InlineWorker()
+        # Shared with the executor, delegation and planner, which check it
+        # at each tool boundary.
+        self._cancel = cancel or Cancel()
+        # Read from the transport's thread, so plain values only: whether
+        # long work is running, and the state name for a "busy" reply.
+        self._working = False
+        self._state_name = IDLE
+        # Set by whichever path halted because of a stop, so the stop's own
+        # job knows whether it stopped anything.
+        self._halted_by_stop = False
         self._data = _fresh_data()
         self._project: str | None = None
         transport.on_message(self._handle)
@@ -202,7 +226,8 @@ class Orchestrator:
         # resumed or discarded by inspecting the branch and the log, and only a
         # genuinely unclean cut is left for the sweep below to gate on.
         if self._delegate is not None:
-            self._delegate.recover(self._transport.send_message)
+            with self._long_work():
+                self._delegate.recover(self._transport.send_message)
         # The publisher owns the external-truth query for its own intents, so
         # a configured publisher always contributes its reconciler (A3).
         registered: dict[str, Reconciler] = dict(reconcilers or {})
@@ -210,6 +235,8 @@ class Orchestrator:
             registered = {**self._publisher.reconcilers(), **registered}
         report = reconcile_on_boot(self._store, registered)
         state = self._store.get_state()
+        if state is not None:
+            self._state_name = state.state
         if state is None:
             self._transition(IDLE, cause="boot: fresh state")
             self._transport.send_message("nisse-dev ready (idle)")
@@ -248,12 +275,68 @@ class Orchestrator:
             )
 
     def _handle(self, msg: InboundMessage) -> None:
-        with self._lock:
-            self._handle_locked(msg)
-
-    def _handle_locked(self, msg: InboundMessage) -> None:
+        """Runs on the transport's thread, so it touches nothing durable and
+        hands everything to the worker."""
         if msg.sender not in self._operators:
             return
+        text = msg.text.strip()
+        if text == "stop":
+            # Straight to the flag, not behind the work in the queue: the
+            # work is what it has to reach (A4).
+            if self._worker.busy:
+                self._cancel.set()
+            self._worker.submit(lambda: self._finish_stop(msg))
+            return
+        if self._working:
+            # Queued, it would run after the work, in whatever state the work
+            # left behind, which is not the state the operator was answering.
+            self._transport.send_message(
+                f"busy ({self._state_name}); say 'stop' to halt"
+            )
+            self._worker.submit(lambda: self._record(msg))
+            return
+        self._worker.submit(lambda: self._handle_now(msg))
+
+    def _record(self, msg: InboundMessage) -> None:
+        self._store.append_event(
+            UserMessage(text=msg.text.strip(), source=f"transport:{msg.sender}")
+        )
+
+    def _finish_stop(self, msg: InboundMessage) -> None:
+        """The worker's half of `stop`. Whatever was running has returned by
+        now, and has already moved to AWAIT_INPUT if the stop reached it."""
+        self._record(msg)
+        halted = self._halted_by_stop
+        self._halted_by_stop = False
+        self._cancel.clear()
+        if not halted:
+            self._transport.send_message(f"nothing to stop (state={self._state_name})")
+
+    @contextmanager
+    def _long_work(self) -> Iterator[None]:
+        self._working = True
+        try:
+            yield
+        finally:
+            self._working = False
+
+    def _halt_for_stop(
+        self, *, what: str, note: str, next_step: str, pending: dict[str, Any]
+    ) -> None:
+        """A4: stopped work moves to AWAIT_INPUT, and what happens next is the
+        operator's call. The prompt asks why, because the model should hear
+        the reason before it tries again."""
+        self._halted_by_stop = True
+        self._await_input(
+            cause=f"operator: stop ({what})",
+            prompt=(
+                f"stopped {what} ({note}). Why did you stop it? Your reply is "
+                f"recorded for the model. {next_step}"
+            ),
+            pending=pending,
+        )
+
+    def _handle_now(self, msg: InboundMessage) -> None:
         # Every operator message enters the canonical transcript, so planning
         # turns see feedback and answers as conversation rather than as
         # out-of-band parameters (A10).
@@ -327,9 +410,20 @@ class Orchestrator:
     ) -> None:
         assert self._project is not None
         try:
-            action = self._agent.plan(
-                gate_name, self._project, feedback=feedback, answer=answer
+            with self._long_work():
+                action = self._agent.plan(
+                    gate_name, self._project, feedback=feedback, answer=answer
+                )
+        except PlanningStopped as exc:
+            # Answering resumes planning with the answer, so the reason for
+            # the stop is the first thing the planner hears.
+            self._halt_for_stop(
+                what=f"planning {gate_name}",
+                note=str(exc),
+                next_step="Planning resumes with your reply.",
+                pending={"kind": "question", "gate": gate_name},
             )
+            return
         except PlanningError as exc:
             self._await_input(
                 cause=f"planning failed: {exc}",
@@ -456,7 +550,19 @@ class Orchestrator:
             self._transport.send_message(f"[stub] executing milestone {milestone}")
             return
         self._transport.send_message(f"executing milestone {milestone}")
-        outcome = self._executor.run(milestone)
+        with self._long_work():
+            outcome = self._executor.run(milestone)
+        if outcome.status == "stopped":
+            self._halt_for_stop(
+                what=f"milestone {milestone}",
+                note=outcome.note,
+                next_step=(
+                    f"The plan is still approved; say 'start milestone "
+                    f"{milestone}' to run it again."
+                ),
+                pending={"kind": "milestone_stopped", "milestone": milestone},
+            )
+            return
         if outcome.status == "completed":
             self._transport.send_message(f"milestone {milestone} done: {outcome.note}")
             self._publish_milestone(milestone)
@@ -495,12 +601,13 @@ class Orchestrator:
         """A comment on the project issue may satisfy an open plan gate (A4).
 
         Approvals bind to a revision here exactly as over transport, and the
-        first valid approval on either channel wins.
+        first valid approval on either channel wins. Called from the poller
+        thread, which waits for the worker so its cursor never moves past a
+        comment that has not been handled.
         """
-        with self._lock:
-            self._handle_plan_comment_locked(comment)
+        self._worker.call(lambda: self._handle_plan_comment_now(comment))
 
-    def _handle_plan_comment_locked(self, comment: Any) -> None:
+    def _handle_plan_comment_now(self, comment: Any) -> None:
         if comment.author not in self._gitea_operators:
             return
         state = self._store.get_state()
@@ -524,10 +631,9 @@ class Orchestrator:
         code, so it is operator-scoped like any other (A4, and the same
         reasoning C4 applies to tickets).
         """
-        with self._lock:
-            self._handle_review_comment_locked(comment)
+        self._worker.call(lambda: self._handle_review_comment_now(comment))
 
-    def _handle_review_comment_locked(self, comment: Any) -> None:
+    def _handle_review_comment_now(self, comment: Any) -> None:
         if comment.author not in self._gitea_operators:
             self._store.oplog.emit(
                 "gitea.comment_ignored",
@@ -553,7 +659,19 @@ class Orchestrator:
         self._store.append_event(
             UserMessage(text=comment.body, source=f"gitea:comment:{comment.id}")
         )
-        outcome = self._executor.run(milestone)
+        with self._long_work():
+            outcome = self._executor.run(milestone)
+        if outcome.status == "stopped":
+            self._halt_for_stop(
+                what=f"the response to review comment {comment.id}",
+                note=outcome.note,
+                next_step=(
+                    f"The milestone {milestone} plan is still approved; say "
+                    f"'start milestone {milestone}' to run it again."
+                ),
+                pending={"kind": "milestone_stopped", "milestone": milestone},
+            )
+            return
         if outcome.status != "completed":
             self._await_input(
                 cause=f"review response {outcome.status}",
@@ -716,3 +834,4 @@ class Orchestrator:
         self._store.transition(
             new_state, cause=cause, project=self._project, data=self._data
         )
+        self._state_name = new_state