From abb7fa021650b6cbb96babbd8c7b0ae3939dcc66 Mon Sep 17 00:00:00 2001 From: Don Garrett Date: Wed, 23 Sep 2026 22:47:29 +0000 Subject: [PATCH] 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) --- nisse-dev/src/nisse_dev/agent.py | 38 ++- nisse-dev/src/nisse_dev/cancel.py | 38 +++ nisse-dev/src/nisse_dev/control.py | 161 ++++++++++-- nisse-dev/src/nisse_dev/delegation.py | 33 ++- nisse-dev/src/nisse_dev/executor.py | 16 +- nisse-dev/src/nisse_dev/planner.py | 13 +- nisse-dev/src/nisse_dev/state.py | 2 + nisse-dev/src/nisse_dev/worker.py | 138 ++++++++++ nisse-dev/tests/test_agent_build.py | 7 +- nisse-dev/tests/test_delegation.py | 41 +++ nisse-dev/tests/test_stop.py | 356 ++++++++++++++++++++++++++ nisse-dev/tests/test_worker.py | 83 ++++++ 12 files changed, 887 insertions(+), 39 deletions(-) create mode 100644 nisse-dev/src/nisse_dev/cancel.py create mode 100644 nisse-dev/src/nisse_dev/worker.py create mode 100644 nisse-dev/tests/test_stop.py create mode 100644 nisse-dev/tests/test_worker.py diff --git a/nisse-dev/src/nisse_dev/agent.py b/nisse-dev/src/nisse_dev/agent.py index ac22e38..2510e4f 100644 --- a/nisse-dev/src/nisse_dev/agent.py +++ b/nisse-dev/src/nisse_dev/agent.py @@ -12,6 +12,7 @@ from collections.abc import Callable from dataclasses import dataclass from nisse_dev.approvals import PlanBoard +from nisse_dev.cancel import Cancel from nisse_dev.catalog import ModelCatalog, select_client from nisse_dev.config import AgentConfig from nisse_dev.control import Orchestrator @@ -28,6 +29,12 @@ from nisse_dev.poller import CommentPoller from nisse_dev.publish import MilestonePublisher from nisse_dev.state import StateStore from nisse_dev.tools import Workspace +from nisse_dev.worker import SerialWorker + +# How long shutdown waits for running work before leaving the store to the +# durability layer (A2/A3). A milestone can run far longer than a service +# manager allows for a stop, and a hard exit mid-work is already safe. +SHUTDOWN_WAIT_S = 10.0 TOOL_DRIVING_ROLES = ("planner", "executor", "delegate") @@ -67,14 +74,15 @@ class Agent: orchestrator: Orchestrator gitea: GiteaClient account: str + worker: SerialWorker _stopped: threading.Event def start(self) -> None: """Boot in the order A3 requires: reconcile against external truth before accepting a single operator command.""" - self.orchestrator.start() + self.worker.call(self.orchestrator.start) self.transport.start() - state = self.store.get_state() + state = self.worker.call(self.store.get_state) where = state.state if state else "IDLE" self.transport.send_message( f"nisse-dev up as {self.account} — project {self.config.project}, " @@ -87,11 +95,16 @@ class Agent: def stop(self) -> None: """Shut down outermost-first: stop accepting work, then stop talking, - then close the state store last so nothing writes to a closed file.""" + then close the state store last so nothing writes to a closed file. + The store belongs to the worker, so the worker closes it.""" self.orchestrator.stop_watching() self.transport.close() self.gitea.close() - self.store.close() + try: + self.worker.call(self.store.close, timeout=SHUTDOWN_WAIT_S) + except TimeoutError: + pass # still working; exiting mid-work is what A2/A3 are for + self.worker.stop(timeout=SHUTDOWN_WAIT_S) self._stopped.set() @@ -116,7 +129,15 @@ def build_agent( config: AgentConfig, *, notify: Callable[[str], None] | None = None ) -> Agent: oplog = OpLog("nisse-dev") - store = StateStore(config.state_db, oplog=oplog) + worker = SerialWorker( + on_error=lambda exc: oplog.emit( + "worker.job_failed", cause="unhandled exception", error=repr(exc) + ) + ) + worker.start() + # Opened on the worker, which is the only thread allowed to use it. + store = worker.call(lambda: StateStore(config.state_db, oplog=oplog)) + cancel = Cancel() # Before the clone, before the token checks: a typo'd role name should # cost a second, not a repository. catalog = discover_models(config, oplog) @@ -148,6 +169,7 @@ def build_agent( git_repo, model=delegate_model, backing_model=catalog.facts(delegate_model).backing, + cancel=cancel, ) executor = LLMExecutor( normalized, @@ -157,6 +179,7 @@ def build_agent( expert=expert, delegate=delegation, notify=announce, + cancel=cancel, ) publisher = MilestonePublisher( store, @@ -183,7 +206,7 @@ def build_agent( orchestrator = Orchestrator( store, transport, - LLMPlanningAgent(normalized, store, config.roles), + LLMPlanningAgent(normalized, store, config.roles, cancel=cancel), operators=config.operators, roles=config.roles, executor=executor, @@ -193,6 +216,8 @@ def build_agent( delegate=delegation, poller_factory=poller_factory, catalog=catalog, + worker=worker, + cancel=cancel, ) return Agent( config=config, @@ -201,5 +226,6 @@ def build_agent( orchestrator=orchestrator, gitea=gitea, account=account, + worker=worker, _stopped=threading.Event(), ) diff --git a/nisse-dev/src/nisse_dev/cancel.py b/nisse-dev/src/nisse_dev/cancel.py new file mode 100644 index 0000000..a66c818 --- /dev/null +++ b/nisse-dev/src/nisse_dev/cancel.py @@ -0,0 +1,38 @@ +"""The operator's kill switch, as the working loops see it (spec A4). + +`stop` has to reach work that is already running, from a thread that is not +running it. So it is a flag, set by whoever hears the command and checked by +the loops at each tool boundary. Setting it touches nothing durable: the +record of a stop is written by the worker when the stopped work returns. +""" + +import threading + +from nisse_dev.canonical import ToolCall, ToolResult +from nisse_dev.state import StateStore + +STOPPED_RESULT = "not run: the operator stopped the agent" + + +class Cancel: + def __init__(self) -> None: + self._event = threading.Event() + + def set(self) -> None: + self._event.set() + + def clear(self) -> None: + self._event.clear() + + @property + def requested(self) -> bool: + return self._event.is_set() + + +def skip_calls(store: StateStore, calls: list[ToolCall]) -> None: + """Record the calls a stop left unrun. Every tool call needs a result for + the transcript to render again, and the result says what happened.""" + for call in calls: + store.append_event( + ToolResult(call_id=call.call_id, result=STOPPED_RESULT, success=False) + ) diff --git a/nisse-dev/src/nisse_dev/control.py b/nisse-dev/src/nisse_dev/control.py index d0142d6..e20cee9 100644 --- a/nisse-dev/src/nisse_dev/control.py +++ b/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 diff --git a/nisse-dev/src/nisse_dev/delegation.py b/nisse-dev/src/nisse_dev/delegation.py index dce0477..9cb2560 100644 --- a/nisse-dev/src/nisse_dev/delegation.py +++ b/nisse-dev/src/nisse_dev/delegation.py @@ -28,6 +28,7 @@ from collections.abc import Callable from dataclasses import dataclass, field from typing import Any, Literal +from nisse_dev.cancel import Cancel, skip_calls from nisse_dev.canonical import ( AssistantText, DelegationCompleted, @@ -129,7 +130,7 @@ class Handback: @dataclass class DelegationResult: - status: Literal["handback", "capped", "failed", "interrupted"] + status: Literal["handback", "capped", "failed", "interrupted", "stopped"] note: str handback: Handback | None = None branch: str | None = None @@ -150,9 +151,9 @@ class DelegationResult: @property def usable(self) -> bool: - """Only a successful handback may be built upon. A cap breach or a - failure leaves a half-edited tree, which A13 says must gate rather - than be resumed.""" + """Only a successful handback may be built upon. A cap breach, a + failure or an operator's stop leaves a half-edited tree, which A13 + says must gate rather than be resumed.""" return self.status == "handback" and self.handback is not None @@ -168,6 +169,7 @@ class Delegation: backing_model: str = "", caps: Caps | None = None, now: Callable[[], float] = time.monotonic, + cancel: Cancel | None = None, ) -> None: self._client = client self._store = store @@ -177,6 +179,10 @@ class Delegation: self._backing_model = backing_model self._caps = caps or Caps() self._now = now + # A stopped delegation is discarded like a failed one, not preserved + # for resuming as an interrupted one is: the operator stopped it for + # a reason, and the next step is to talk about that reason. + self._cancel = cancel or Cancel() def run( self, *, brief: str, branch: str, role: str = "delegate" @@ -316,6 +322,14 @@ class Delegation: spent = 0 for turn in range(self._caps.turns): + if self._cancel.requested: + return DelegationResult( + "stopped", + f"stopped by the operator before turn {turn + 1}", + branch=branch, + artifacts=sorted(touched), + tokens=spent, + ) elapsed = self._now() - started_at if elapsed >= self._caps.wall_clock_s: return DelegationResult( @@ -386,7 +400,16 @@ class Delegation: tokens=spent, ) - for call in calls: + for index, call in enumerate(calls): + if self._cancel.requested: + skip_calls(self._store, calls[index:]) + return DelegationResult( + "stopped", + f"stopped by the operator during turn {turn + 1}", + branch=branch, + artifacts=sorted(touched), + tokens=spent, + ) if call.tool == "hand_back": handback = _parse_handback(call.arguments) if handback is not None and handback.succeeded and not did_work: diff --git a/nisse-dev/src/nisse_dev/executor.py b/nisse-dev/src/nisse_dev/executor.py index e188fa7..958d5cf 100644 --- a/nisse-dev/src/nisse_dev/executor.py +++ b/nisse-dev/src/nisse_dev/executor.py @@ -14,6 +14,7 @@ from collections.abc import Callable from dataclasses import dataclass from typing import Any, Literal +from nisse_dev.cancel import Cancel, skip_calls from nisse_dev.canonical import ToolCall, ToolResult from nisse_dev.delegation import Delegation from nisse_dev.expert import ASK_EXPERT_TOOL, BudgetExhausted, ExpertDesk @@ -45,7 +46,7 @@ _NUDGE = ( @dataclass(frozen=True) class ExecutorOutcome: - status: Literal["completed", "capped", "escalate"] + status: Literal["completed", "capped", "escalate", "stopped"] note: str @@ -62,6 +63,7 @@ class LLMExecutor: expert: ExpertDesk | None = None, delegate: Delegation | None = None, notify: Callable[[str], None] | None = None, + cancel: Cancel | None = None, ) -> None: self._client = client self._store = store @@ -75,6 +77,7 @@ class LLMExecutor: # without this the agent is silent for the whole milestone and a # working run looks exactly like a hung one. self._notify = notify + self._cancel = cancel or Cancel() def _report(self, turn: int, calls: list[ToolCall]) -> None: if self._notify is None: @@ -113,6 +116,8 @@ class LLMExecutor: ) if result.usable: return None + if result.status == "stopped": + return ExecutorOutcome("stopped", f"during a delegation: {result.note}") # A13: a cap breach or failure leaves a half-edited tree, so the local # loop must not resume on top of it. return ExecutorOutcome( @@ -151,6 +156,10 @@ class LLMExecutor: challenged = False for turn in range(self._max_turns): + # A4: `stop` takes effect at the next tool boundary. The start of + # a turn is one, and so is the gap between two tool calls below. + if self._cancel.requested: + return ExecutorOutcome("stopped", f"before turn {turn + 1}") try: produced = self._client.complete( self._store.read_events(), @@ -184,7 +193,10 @@ class LLMExecutor: ) self._report(turn, calls) - for call in calls: + for index, call in enumerate(calls): + if self._cancel.requested: + skip_calls(self._store, calls[index:]) + return ExecutorOutcome("stopped", f"during turn {turn + 1}") if call.tool in MUTATING_TOOLS: did_work = True if call.tool == "ask_expert": diff --git a/nisse-dev/src/nisse_dev/planner.py b/nisse-dev/src/nisse_dev/planner.py index f0b4e86..3f7a8fe 100644 --- a/nisse-dev/src/nisse_dev/planner.py +++ b/nisse-dev/src/nisse_dev/planner.py @@ -11,8 +11,9 @@ crash mid-planning leaves a truthful transcript. from typing import Any +from nisse_dev.cancel import Cancel from nisse_dev.canonical import AssistantText, ToolCall, ToolResult -from nisse_dev.control import AskHuman, PlanningError, SubmitPlan +from nisse_dev.control import AskHuman, PlanningError, PlanningStopped, SubmitPlan from nisse_dev.llm import BudgetExhausted, LLMClient, LLMError, RoleMap from nisse_dev.state import StateStore @@ -84,11 +85,15 @@ class LLMPlanningAgent: roles: RoleMap, *, system_prompt: str = SYSTEM_PROMPT, + cancel: Cancel | None = None, ) -> None: self._client = client self._store = store self._roles = roles self._system_prompt = system_prompt + # Checked before each turn only. A turn is one model call, and a turn + # that completes ends at a gate or a question, which is a halt anyway. + self._cancel = cancel or Cancel() def plan( self, @@ -105,9 +110,11 @@ class LLMPlanningAgent: f"{self._system_prompt}\n\n" f"Current task: draft the {gate} plan for project {project!r}." ) + self._check_stop() action, first_error = self._try_turn(model, instruction) if action is not None: return action + self._check_stop() action, second_error = self._try_turn(model, f"{instruction}\n\n{_NUDGE}") if action is None: detail = second_error or first_error @@ -117,6 +124,10 @@ class LLMPlanningAgent: ) return action + def _check_stop(self) -> None: + if self._cancel.requested: + raise PlanningStopped("stopped by the operator") + def _try_turn( self, model: str, system: str ) -> tuple[SubmitPlan | AskHuman | None, str | None]: diff --git a/nisse-dev/src/nisse_dev/state.py b/nisse-dev/src/nisse_dev/state.py index 2175835..284ac4e 100644 --- a/nisse-dev/src/nisse_dev/state.py +++ b/nisse-dev/src/nisse_dev/state.py @@ -136,6 +136,7 @@ class StateStore: self._conn.execute("PRAGMA synchronous=FULL") self._conn.execute("PRAGMA busy_timeout=5000") self._target_version = schema_version + self.closed = False self._init_or_migrate(migrations) def _init_or_migrate(self, migrations: Mapping[int, Migration]) -> None: @@ -335,6 +336,7 @@ class StateStore: def close(self) -> None: self._conn.close() + self.closed = True def __enter__(self) -> Self: return self diff --git a/nisse-dev/src/nisse_dev/worker.py b/nisse-dev/src/nisse_dev/worker.py new file mode 100644 index 0000000..346c716 --- /dev/null +++ b/nisse-dev/src/nisse_dev/worker.py @@ -0,0 +1,138 @@ +"""The one thread that runs orchestrator work and owns the state store. + +Planning turns, milestones and review responses can each run for minutes. +Running them on the thread that delivered the message blocked that thread for +as long: the Matrix sync loop read nothing new until the work finished, so an +operator's `stop` (A4) would only have arrived after the work it was meant to +stop. Every handler now runs here instead, and transports only hand work over. + +Owning the store is the other half. SQLite connections are bound to the thread +that opened them, and the store keeps that check on: it is opened on this +thread, used only from it, and any future code that touches it from elsewhere +fails loudly instead of racing. + +Jobs run one at a time, in order, so state changes never interleave — the +guarantee the orchestrator's lock used to give. +""" + +import sys +import threading +import traceback +from collections import deque +from collections.abc import Callable +from concurrent.futures import Future +from queue import Queue +from typing import Protocol, TypeVar + +T = TypeVar("T") + + +class Worker(Protocol): + @property + def busy(self) -> bool: ... + + def submit(self, job: Callable[[], object]) -> None: ... + + def call(self, job: Callable[[], T], timeout: float | None = None) -> T: ... + + +class InlineWorker: + """Runs jobs on the caller's thread, for tests and single-threaded use. + + A job submitted from inside a running job waits until that one finishes, + exactly as it would queue behind it on the real worker. Without that, a + test that says `stop` from inside a fake milestone would run the stop + handling in the middle of the milestone, which production never does. + """ + + def __init__(self) -> None: + self._pending: deque[Callable[[], object]] = deque() + self._running = False + + @property + def busy(self) -> bool: + return self._running + + def submit(self, job: Callable[[], object]) -> None: + self._pending.append(job) + if self._running: + return + self._running = True + try: + while self._pending: + self._pending.popleft()() + finally: + self._running = False + + def call(self, job: Callable[[], T], timeout: float | None = None) -> T: + if self._running: + return job() # already on the worker, as a nested call would be + box: list[T] = [] + self.submit(lambda: box.append(job())) + return box[0] + + +_STOP = object() + + +def _report(exc: BaseException) -> None: + traceback.print_exception(exc, file=sys.stderr) + + +class SerialWorker: + def __init__( + self, + *, + on_error: Callable[[BaseException], None] = _report, + name: str = "nisse-worker", + ) -> None: + self._queue: Queue[object] = Queue() + self._on_error = on_error + self._busy = False + self._thread = threading.Thread(target=self._run, name=name, daemon=True) + + @property + def busy(self) -> bool: + return self._busy + + def start(self) -> None: + self._thread.start() + + def submit(self, job: Callable[[], object]) -> None: + self._queue.put((job, None)) + + def call(self, job: Callable[[], T], timeout: float | None = None) -> T: + """Run a job here and wait for its result. Raises TimeoutError if it + has not finished in time; the job itself still runs.""" + if threading.current_thread() is self._thread: + return job() + future: Future[T] = Future() + self._queue.put((job, future)) + return future.result(timeout) + + def stop(self, timeout: float | None = None) -> None: + self._queue.put(_STOP) + self._thread.join(timeout) + + def _run(self) -> None: + while True: + item = self._queue.get() + if item is _STOP: + return + job, future = item # type: ignore[misc] + self._busy = True + try: + result = job() + except Exception as exc: # noqa: BLE001 -- see below + # One failed job must not take the agent's only worker with + # it. A caller that waits gets the exception; otherwise it is + # reported and the next job runs. + if future is not None: + future.set_exception(exc) + else: + self._on_error(exc) + else: + if future is not None: + future.set_result(result) + finally: + self._busy = False diff --git a/nisse-dev/tests/test_agent_build.py b/nisse-dev/tests/test_agent_build.py index 2ee391d..d30797f 100644 --- a/nisse-dev/tests/test_agent_build.py +++ b/nisse-dev/tests/test_agent_build.py @@ -3,7 +3,6 @@ delegation harness and the comment poller — were fully built, fully tested, and reachable from nothing but their own tests. Unit tests that construct a component directly cannot catch that; only assembly can.""" -import sqlite3 from pathlib import Path from typing import Any @@ -163,8 +162,7 @@ def test_stop_closes_everything_and_releases_the_wait( assert agent.transport.closed agent.wait() # returns immediately rather than hanging - with pytest.raises(sqlite3.ProgrammingError): - agent.store.get_state() # the store really is closed + assert agent.store.closed def test_poller_factory_ignores_the_agents_own_comments( @@ -268,7 +266,8 @@ def test_the_resolved_model_is_pinned_at_task_start( tmp_path: Path, origin: Path, monkeypatch ) -> None: agent = build_offline(tmp_path, origin, monkeypatch) - agent.orchestrator._begin_project("greet") + # The store belongs to the worker thread, so the work has to run there. + agent.worker.call(lambda: agent.orchestrator._begin_project("greet")) pinned = agent.orchestrator._data["models"] assert pinned["planner"] == { diff --git a/nisse-dev/tests/test_delegation.py b/nisse-dev/tests/test_delegation.py index 8a7fa55..4a55390 100644 --- a/nisse-dev/tests/test_delegation.py +++ b/nisse-dev/tests/test_delegation.py @@ -4,6 +4,7 @@ from typing import Any import pytest +from nisse_dev.cancel import STOPPED_RESULT, Cancel from nisse_dev.canonical import ( AssistantText, CanonicalEvent, @@ -134,6 +135,46 @@ def test_abort_discards_the_branch_and_its_work( assert not (repo.root / "half-done.py").exists() +def test_stop_discards_the_branch_and_settles_the_intent( + store: StateStore, ws: Workspace, repo: GitRepo +) -> None: + # A4 stop is treated like a failure, not like an outage: the operator + # stopped it for a reason, so nothing is kept for resuming. + cancel = Cancel() + client = ScriptedClient( + [ + [call("write_file", path="half-done.py", content="oops")], + [call("list_dir", path=".")], + ] + ) + original = client.complete + + def complete_then_stop(*args: Any, **kwargs: Any) -> list[CanonicalEvent]: + produced = original(*args, **kwargs) + if len(client.calls) == 2: + cancel.set() # the operator says stop during turn 2's model call + return produced + + client.complete = complete_then_stop # type: ignore[method-assign] + result = build(client, store, ws, repo, cancel=cancel).run( + brief="b", branch="deleg-stop" + ) + + assert result.status == "stopped" + assert not result.usable + assert repo.current_branch() == "main" + assert not repo.branch_exists("deleg-stop") + assert not (repo.root / "half-done.py").exists() + assert store.pending_intents() == [] + # The unrun call still has a result, so the transcript renders. + skipped = [ + e.event + for e in store.read_events() + if isinstance(e.event, ToolResult) and e.event.result == STOPPED_RESULT + ] + assert len(skipped) == 1 + + def test_turn_cap_is_hard(store: StateStore, ws: Workspace, repo: GitRepo) -> None: forever = [[call("list_dir", path=".")] for _ in range(20)] client = ScriptedClient(forever) diff --git a/nisse-dev/tests/test_stop.py b/nisse-dev/tests/test_stop.py new file mode 100644 index 0000000..23cdb19 --- /dev/null +++ b/nisse-dev/tests/test_stop.py @@ -0,0 +1,356 @@ +"""The operator's kill switch, `stop` (spec A4).""" + +import io +import threading +from pathlib import Path +from typing import Any + +from nisse_dev.cancel import STOPPED_RESULT, Cancel +from nisse_dev.canonical import AssistantText, ToolCall, ToolResult, UserMessage +from nisse_dev.control import ( + AWAIT_ARCH_SIGNOFF, + AWAIT_INPUT, + AWAIT_MILESTONE_SIGNOFF, + Orchestrator, + PlanningStopped, + StubAgent, + SubmitPlan, +) +from nisse_dev.delegation import DelegationResult +from nisse_dev.executor import ExecutorOutcome, LLMExecutor +from nisse_dev.llm import RoleMap +from nisse_dev.oplog import OpLog +from nisse_dev.planner import LLMPlanningAgent +from nisse_dev.state import StateStore +from nisse_dev.tools import ToolOutcome +from nisse_dev.transport import InMemoryTransport +from nisse_dev.worker import SerialWorker + +ROLES = RoleMap( + executor="llama3.2", + planner="claude-haiku-4-5", + delegate="claude-haiku-4-5", + consultant="claude-haiku-4-5", +) + + +def new_store(path: Path) -> StateStore: + return StateStore(path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())) + + +def user_messages(store: StateStore) -> list[str]: + return [ + e.event.text for e in store.read_events() if isinstance(e.event, UserMessage) + ] + + +# --- the orchestrator ------------------------------------------------------ + + +class OperatorDuringWork: + """An executor that hears from the operator mid-milestone, then does what + the real loop does at its next tool boundary.""" + + def __init__(self, cancel: Cancel, says: str) -> None: + self.cancel = cancel + self.says = says + self.transport: InMemoryTransport | None = None + + def run(self, milestone: int) -> ExecutorOutcome: + assert self.transport is not None + self.transport.receive("don", self.says) + if self.cancel.requested: + return ExecutorOutcome("stopped", "during turn 3") + return ExecutorOutcome("completed", "done") + + +def at_start_of_milestone( + tmp_path: Path, says: str +) -> tuple[StateStore, InMemoryTransport, Cancel]: + store = new_store(tmp_path) + transport = InMemoryTransport() + cancel = Cancel() + executor = OperatorDuringWork(cancel, says) + executor.transport = transport + orch = Orchestrator( + store, + transport, + StubAgent(), + operators={"don"}, + executor=executor, + cancel=cancel, + ) + orch.start() + for text in ("use demo", "approve", "approve"): + transport.receive("don", text) + return store, transport, cancel + + +def test_stop_halts_a_running_milestone_and_asks_why(tmp_path: Path) -> None: + store, transport, cancel = at_start_of_milestone(tmp_path, "stop") + transport.receive("don", "start milestone 1") + + assert store.get_state().state == AWAIT_INPUT + prompt = transport.outbound[-1] + assert prompt.kind == "question" + assert "stopped milestone 1 (during turn 3)" in prompt.text + assert "Why did you stop it?" in prompt.text + assert user_messages(store)[-1] == "stop" + assert not cancel.requested # the next piece of work is not pre-stopped + + +def test_the_reason_is_recorded_and_the_plan_stays_approved(tmp_path: Path) -> None: + store, transport, _ = at_start_of_milestone(tmp_path, "stop") + transport.receive("don", "start milestone 1") + transport.receive("don", "it was editing the wrong module") + + assert store.get_state().state == AWAIT_MILESTONE_SIGNOFF + # The next run renders this, so the model hears why it was stopped. + assert user_messages(store)[-1] == "it was editing the wrong module" + assert "start milestone 1" in transport.outbound[-1].text + + +def test_stop_with_nothing_running_says_so_and_changes_nothing( + tmp_path: Path, +) -> None: + store = new_store(tmp_path) + transport = InMemoryTransport() + orch = Orchestrator(store, transport, StubAgent(), operators={"don"}) + orch.start() + transport.receive("don", "use demo") + transport.receive("don", "stop") + + # At a gate, "stop" is the command, not rejection feedback. + assert store.get_state().state == AWAIT_ARCH_SIGNOFF + assert orch._data["revs"]["arch"] == 1 + assert ( + transport.outbound[-1].text == f"nothing to stop (state={AWAIT_ARCH_SIGNOFF})" + ) + + +def test_stop_from_a_non_operator_is_ignored(tmp_path: Path) -> None: + _, transport, cancel = at_start_of_milestone(tmp_path, "approve") + before = len(transport.outbound) + transport.receive("mallory", "stop") + assert len(transport.outbound) == before + assert not cancel.requested + + +def test_other_messages_during_work_are_answered_not_acted_on( + tmp_path: Path, +) -> None: + store, transport, _ = at_start_of_milestone(tmp_path, "approve") + transport.receive("don", "start milestone 1") + + texts = [m.text for m in transport.outbound] + assert "busy (EXECUTING); say 'stop' to halt" in texts + # Recorded, as every operator message is, but not applied to anything. + assert user_messages(store)[-1] == "approve" + + +class StopsBeforePlanning: + def __init__(self, cancel: Cancel) -> None: + self.cancel = cancel + self.transport: InMemoryTransport | None = None + self.answers: list[str | None] = [] + + def plan( + self, + gate: str, + project: str, + feedback: str | None = None, + answer: str | None = None, + ) -> SubmitPlan: + self.answers.append(answer) + if len(self.answers) == 1: + assert self.transport is not None + self.transport.receive("don", "stop") + if self.cancel.requested: + raise PlanningStopped("stopped by the operator") + return SubmitPlan(f"{gate} plan, after: {answer}") + + +def test_stop_during_planning_resumes_planning_with_the_reason( + tmp_path: Path, +) -> None: + store = new_store(tmp_path) + transport = InMemoryTransport() + cancel = Cancel() + planner = StopsBeforePlanning(cancel) + planner.transport = transport + orch = Orchestrator(store, transport, planner, operators={"don"}, cancel=cancel) + orch.start() + transport.receive("don", "use demo") + + assert store.get_state().state == AWAIT_INPUT + assert "stopped planning arch" in transport.outbound[-1].text + + transport.receive("don", "keep it to one module") + assert store.get_state().state == AWAIT_ARCH_SIGNOFF + assert planner.answers[-1] == "keep it to one module" + + +# --- on real threads ------------------------------------------------------- + + +class BlocksUntilStopped: + def __init__(self, cancel: Cancel) -> None: + self.cancel = cancel + self.running = threading.Event() + + def run(self, milestone: int) -> ExecutorOutcome: + self.running.set() + for _ in range(500): + if self.cancel.requested: + return ExecutorOutcome("stopped", "at a tool boundary") + threading.Event().wait(0.01) + return ExecutorOutcome("completed", "never stopped") + + +def test_stop_reaches_work_running_on_the_worker(tmp_path: Path) -> None: + # The bug this design fixes: the work used to run on the transport's own + # thread, so no message, stop included, was read until it finished. + worker = SerialWorker() + worker.start() + store = worker.call(lambda: new_store(tmp_path)) + transport = InMemoryTransport() + cancel = Cancel() + executor = BlocksUntilStopped(cancel) + orch = Orchestrator( + store, + transport, + StubAgent(), + operators={"don"}, + executor=executor, + worker=worker, + cancel=cancel, + ) + worker.call(orch.start) + for text in ("use demo", "approve", "approve", "start milestone 1"): + transport.receive("don", text) + assert executor.running.wait(5) + + transport.receive("don", "status") # delivered while the milestone runs + transport.receive("don", "stop") + worker.call(lambda: None) # wait for the stop to be recorded + + texts = [m.text for m in transport.outbound] + assert "busy (EXECUTING); say 'stop' to halt" in texts + assert worker.call(store.get_state).state == AWAIT_INPUT + assert "stopped milestone 1 (at a tool boundary)" in texts[-1] + worker.call(store.close) + worker.stop(timeout=5) + + +# --- the loops ------------------------------------------------------------- + + +class ScriptedClient: + def __init__(self, turns: list[list[Any]], on_call: Any = None) -> None: + self.turns = turns + self.on_call = on_call + self.calls = 0 + + def complete(self, events: Any, tools: Any, model: str, system: Any = None): + self.calls += 1 + if self.on_call: + self.on_call() + return self.turns.pop(0) if self.turns else [AssistantText("done", model)] + + +class StopsAfterFirstCall: + """A workspace whose first tool call is where the operator says stop.""" + + def __init__(self, cancel: Cancel) -> None: + self.cancel = cancel + self.ran: list[str] = [] + + def describe(self) -> str: + return "workspace" + + def dispatch(self, tool: str, arguments: dict[str, Any]) -> ToolOutcome: + self.ran.append(tool) + self.cancel.set() + return ToolOutcome("ok", success=True) + + +def test_executor_stops_before_a_turn_without_calling_the_model( + tmp_path: Path, +) -> None: + cancel = Cancel() + cancel.set() + client = ScriptedClient([]) + executor = LLMExecutor( + client, new_store(tmp_path), ROLES, StopsAfterFirstCall(cancel), cancel=cancel + ) + outcome = executor.run(1) + assert outcome == ExecutorOutcome("stopped", "before turn 1") + assert client.calls == 0 + + +def test_executor_stops_between_tool_calls_and_records_the_skipped_ones( + tmp_path: Path, +) -> None: + cancel = Cancel() + store = new_store(tmp_path) + ws = StopsAfterFirstCall(cancel) + calls = [ + ToolCall(call_id="a", tool="run_command", arguments={"command": "make"}), + ToolCall(call_id="b", tool="write_file", arguments={"path": "x"}), + ] + executor = LLMExecutor(ScriptedClient([calls]), store, ROLES, ws, cancel=cancel) + outcome = executor.run(1) + + assert outcome == ExecutorOutcome("stopped", "during turn 1") + assert ws.ran == ["run_command"] + results = { + e.event.call_id: e.event.result + for e in store.read_events() + if isinstance(e.event, ToolResult) + } + # Every call has a result, so the transcript still renders. + assert results == {"a": "ok", "b": STOPPED_RESULT} + + +class StoppedDelegate: + def run(self, *, brief: str, branch: str) -> DelegationResult: + return DelegationResult("stopped", "stopped by the operator during turn 2") + + +def test_a_stopped_delegation_stops_the_milestone(tmp_path: Path) -> None: + failing = [ + [ToolCall(call_id=f"c{n}", tool="run_command", arguments={"command": "x"})] + for n in range(2) + ] + + class Failing: + def describe(self) -> str: + return "workspace" + + def dispatch(self, tool: str, arguments: dict[str, Any]) -> ToolOutcome: + return ToolOutcome("exit 1", success=False) + + executor = LLMExecutor( + ScriptedClient(failing), + new_store(tmp_path), + ROLES, + Failing(), + delegate=StoppedDelegate(), + ) + outcome = executor.run(1) + assert outcome.status == "stopped" + assert "during a delegation" in outcome.note + + +def test_planner_does_not_start_a_turn_once_stopped(tmp_path: Path) -> None: + cancel = Cancel() + cancel.set() + client = ScriptedClient([]) + planner = LLMPlanningAgent(client, new_store(tmp_path), ROLES, cancel=cancel) + try: + planner.plan("arch", "demo") + except PlanningStopped: + pass + else: + raise AssertionError("planning ran after stop") + assert client.calls == 0 diff --git a/nisse-dev/tests/test_worker.py b/nisse-dev/tests/test_worker.py new file mode 100644 index 0000000..4b0a643 --- /dev/null +++ b/nisse-dev/tests/test_worker.py @@ -0,0 +1,83 @@ +import io +import threading +from pathlib import Path + +import pytest + +from nisse_dev.oplog import OpLog +from nisse_dev.state import StateStore +from nisse_dev.worker import InlineWorker, SerialWorker + + +@pytest.fixture +def worker() -> SerialWorker: + w = SerialWorker() + w.start() + yield w + w.stop(timeout=5) + + +def test_jobs_run_in_order_on_the_worker_thread(worker: SerialWorker) -> None: + seen: list[tuple[int, str]] = [] + for n in range(3): + worker.submit(lambda n=n: seen.append((n, threading.current_thread().name))) + worker.call(lambda: None) # drains everything queued before it + + assert [n for n, _ in seen] == [0, 1, 2] + assert {name for _, name in seen} == {"nisse-worker"} + + +def test_call_returns_the_result_and_raises_the_error(worker: SerialWorker) -> None: + assert worker.call(lambda: 42) == 42 + + def boom() -> None: + raise ValueError("bad job") + + with pytest.raises(ValueError, match="bad job"): + worker.call(boom) + + +def test_a_failed_job_is_reported_and_the_worker_carries_on() -> None: + errors: list[BaseException] = [] + worker = SerialWorker(on_error=errors.append) + worker.start() + + def boom() -> None: + raise RuntimeError("lost job") + + worker.submit(boom) + assert worker.call(lambda: "still here") == "still here" + assert [str(e) for e in errors] == ["lost job"] + worker.stop(timeout=5) + + +def test_call_from_the_worker_itself_runs_inline(worker: SerialWorker) -> None: + # Queueing behind itself would deadlock. + assert worker.call(lambda: worker.call(lambda: "nested")) == "nested" + + +def test_the_store_is_usable_only_from_the_thread_that_owns_it( + worker: SerialWorker, tmp_path: Path +) -> None: + store = worker.call( + lambda: StateStore(tmp_path / "state.db", oplog=OpLog("t", io.StringIO())) + ) + assert worker.call(store.get_state) is None + # The tripwire the worker design keeps on: any other thread fails loudly. + with pytest.raises(Exception, match="same thread"): + store.get_state() + worker.call(store.close) + + +def test_inline_worker_defers_a_job_submitted_from_inside_a_job() -> None: + worker = InlineWorker() + order: list[str] = [] + + def outer() -> None: + order.append("outer start") + worker.submit(lambda: order.append("inner")) + order.append("outer end") + + worker.submit(outer) + assert order == ["outer start", "outer end", "inner"] + assert not worker.busy