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