diff --git a/nisse-dev/src/nisse_dev/control.py b/nisse-dev/src/nisse_dev/control.py index 8e21c99..18eeaf1 100644 --- a/nisse-dev/src/nisse_dev/control.py +++ b/nisse-dev/src/nisse_dev/control.py @@ -13,8 +13,10 @@ 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 sys +import traceback import uuid -from collections.abc import Iterable, Iterator, Mapping +from collections.abc import Callable, Iterable, Iterator, Mapping from contextlib import contextmanager from dataclasses import dataclass from typing import Any, Protocol @@ -56,6 +58,11 @@ TRANSITIONS = { } +# States that are only right while work is running. Once a job ends the agent +# must be somewhere it can wait for the operator, or every message would be +# answered "busy" until a restart. +_WORKING = (PLANNING_ARCH, MILESTONE_PLANNING, EXECUTING) + _BOOT_CHOICES = ( "reply 'continue' to pick up where the restart cut in, " "or 're-plan' to plan the current step again" @@ -310,7 +317,38 @@ class Orchestrator: ) self._worker.submit(lambda: self._record(msg)) return - self._worker.submit(lambda: self._handle_now(msg)) + self._worker.submit( + lambda: self._settled("your message", lambda: self._handle_now(msg)) + ) + + def _settled(self, what: str, job: Callable[[], None]) -> None: + """Run one job and make sure it ends in a state that can wait for the + operator, however it ended. An exception is reported rather than left + to the worker, which would only print it; a job left in a working + state is moved to AWAIT_INPUT with the choice a restart offers.""" + error: Exception | None = None + try: + job() + except Exception as exc: # noqa: BLE001 -- reported below + error = exc + traceback.print_exception(exc, file=sys.stderr) + state = self._store.get_state() + current = state.state if state else IDLE + reason = f"{type(error).__name__}: {error}" if error else "no outcome" + try: + if current in _WORKING: + self._await_input( + cause=f"stranded in {current}: {reason}", + prompt=f"{what} ended mid-{current} ({reason}); {_BOOT_CHOICES}", + # The same choice as after a restart, answered the same way. + pending={"kind": "boot", "interrupted": current}, + ) + elif error is not None: + self._transport.send_message( + f"could not handle {what} ({reason}); still {current}" + ) + except Exception as exc: # noqa: BLE001 -- nothing left to tell + traceback.print_exception(exc, file=sys.stderr) def _record(self, msg: InboundMessage) -> None: self._store.append_event( @@ -375,6 +413,22 @@ class Orchestrator: else: self._transport.send_message("no active project; say 'use '") elif current in (AWAIT_ARCH_SIGNOFF, AWAIT_MILESTONE_SIGNOFF): + if not self._data.get("gate"): + # A sign-off state with no plan to sign off: replies could + # never be applied, so plan again from the operator's choice. + self._await_input( + cause=f"{current} without an open gate", + prompt=f"the plan awaiting sign-off is missing; {_BOOT_CHOICES}", + pending={ + "kind": "boot", + "interrupted": ( + PLANNING_ARCH + if current == AWAIT_ARCH_SIGNOFF + else MILESTONE_PLANNING + ), + }, + ) + return self._handle_gate_reply(msg, text) elif current == PR_OPEN: self._handle_pr_open(text) @@ -568,7 +622,9 @@ class Orchestrator: def _execute_milestone(self, milestone: int) -> None: if self._executor is None: + # The stub completes at once, so it still leaves EXECUTING. self._transport.send_message(f"[stub] executing milestone {milestone}") + self._publish_milestone(milestone) return self._transport.send_message(f"executing milestone {milestone}") with self._long_work(): @@ -626,7 +682,11 @@ class Orchestrator: thread, which waits for the worker so its cursor never moves past a comment that has not been handled. """ - self._worker.call(lambda: self._handle_plan_comment_now(comment)) + self._worker.call( + lambda: self._settled( + "a plan comment", lambda: self._handle_plan_comment_now(comment) + ) + ) def _handle_plan_comment_now(self, comment: Any) -> None: if comment.author not in self._gitea_operators: @@ -652,7 +712,11 @@ class Orchestrator: code, so it is operator-scoped like any other (A4, and the same reasoning C4 applies to tickets). """ - self._worker.call(lambda: self._handle_review_comment_now(comment)) + self._worker.call( + lambda: self._settled( + "a review comment", lambda: self._handle_review_comment_now(comment) + ) + ) def _handle_review_comment_now(self, comment: Any) -> None: if comment.author not in self._gitea_operators: @@ -723,6 +787,12 @@ class Orchestrator: def _publish_milestone(self, milestone: int) -> None: if self._publisher is None: + # No PR to wait on, but PR_OPEN's commands still lead on. + self._transition(PR_OPEN, cause=f"[stub] milestone {milestone} done") + self._transport.send_message( + f"[stub] milestone {milestone} done; no publisher, so no PR — " + "say 'next milestone' or 'project complete'" + ) return gate = self._data.get("gate") or {} try: diff --git a/nisse-dev/tests/test_control.py b/nisse-dev/tests/test_control.py index 6dc5665..2c08c2c 100644 --- a/nisse-dev/tests/test_control.py +++ b/nisse-dev/tests/test_control.py @@ -12,6 +12,7 @@ from nisse_dev.control import ( EXECUTING, IDLE, PLANNING_ARCH, + PR_OPEN, AskHuman, Orchestrator, PlanningError, @@ -52,6 +53,10 @@ class Harness: return [m.text for m in self.transport.outbound] +def executed(h: Harness) -> bool: + return any(e["to"] == EXECUTING for e in h.events("state.transition")) + + @pytest.fixture def h(tmp_path: Path) -> Harness: return Harness(tmp_path) @@ -74,14 +79,16 @@ def test_use_project_submits_arch_plan_and_opens_gate(h: Harness) -> None: assert opened["task_id"] is not None -def test_full_happy_path_to_executing(h: Harness) -> None: +def test_full_happy_path_through_execution(h: Harness) -> None: h.transport.receive("don", "use demo") h.transport.receive("don", "approve") assert h.state == AWAIT_MILESTONE_SIGNOFF h.transport.receive("don", "approve") assert h.state == AWAIT_MILESTONE_SIGNOFF # approved, awaiting start h.transport.receive("don", "start milestone 1") - assert h.state == EXECUTING + assert executed(h) + # The stub completes at once and waits in PR_OPEN, never in EXECUTING. + assert h.state == PR_OPEN approvals = h.events("gate.approve") assert [e["gate"] for e in approvals] == ["arch", "milestone-1"] @@ -123,7 +130,7 @@ def test_start_milestone_requires_approval_first(h: Harness) -> None: assert h.state == AWAIT_MILESTONE_SIGNOFF # it was rejection feedback h.transport.receive("don", "approve rev-2") h.transport.receive("don", "start milestone 1") - assert h.state == EXECUTING + assert executed(h) def test_a_message_is_matched_as_stripped_but_recorded_as_sent( @@ -231,15 +238,16 @@ def test_restart_mid_execution_continue_returns_to_the_approved_plan( h.transport.receive("don", "use demo") h.transport.receive("don", "approve") h.transport.receive("don", "approve") - h.transport.receive("don", "start milestone 1") - assert h.state == EXECUTING + # A crash mid-milestone: the store still says EXECUTING. + state = h.store.get_state() + h.store.transition(EXECUTING, cause="test", project="demo", data=state.data) h = Harness(tmp_path) assert h.state == AWAIT_INPUT h.transport.receive("don", "continue") assert h.state == AWAIT_MILESTONE_SIGNOFF assert "start milestone 1" in h.outbound_texts()[-1] h.transport.receive("don", "start milestone 1") - assert h.state == EXECUTING + assert executed(h) def test_restart_question_without_its_interrupted_state_still_replans( diff --git a/nisse-dev/tests/test_no_stranding.py b/nisse-dev/tests/test_no_stranding.py new file mode 100644 index 0000000..58a097b --- /dev/null +++ b/nisse-dev/tests/test_no_stranding.py @@ -0,0 +1,237 @@ +"""The orchestrator's resting-state invariant, checked under injected faults. + +Working states (planning, executing) are only right while work runs. After +any message is handled the agent must be somewhere it can wait for the +operator, and some reply must lead on from there. The first live run found +four ways to break this one at a time; this drives a whole project with a +fault at each step instead. +""" + +import io +from pathlib import Path +from typing import Any + +import pytest + +from nisse_dev.control import ( + AWAIT_ARCH_SIGNOFF, + AWAIT_INPUT, + AWAIT_MILESTONE_SIGNOFF, + EXECUTING, + IDLE, + MILESTONE_PLANNING, + PLANNING_ARCH, + PR_OPEN, + AskHuman, + Orchestrator, + PlanningError, + SubmitPlan, +) +from nisse_dev.executor import ExecutorOutcome +from nisse_dev.gitea import GiteaError, PullRequest +from nisse_dev.oplog import OpLog +from nisse_dev.publish import PublishResult +from nisse_dev.state import StateStore +from nisse_dev.transport import InMemoryTransport + +WORKING = {PLANNING_ARCH, MILESTONE_PLANNING, EXECUTING} + + +class Faults: + """One fault, fired once, at the nth call to one site.""" + + def __init__(self, site: str, nth: int, fault: Any) -> None: + self._site, self._nth, self._fault = site, nth, fault + self._calls: dict[str, int] = {} + self.fired = False + + def check(self, site: str) -> Any: + """Raise or return the fault if this call is the one; else None.""" + self._calls[site] = self._calls.get(site, 0) + 1 + if self.fired or site != self._site or self._calls[site] != self._nth: + return None + self.fired = True + if isinstance(self._fault, Exception): + raise self._fault + return self._fault + + +class Planner: + def __init__(self, faults: Faults) -> None: + self._faults = faults + + def plan( + self, + gate: str, + project: str, + feedback: str | None = None, + answer: str | None = None, + ) -> SubmitPlan | AskHuman: + return self._faults.check("planner") or SubmitPlan(f"{gate} plan") + + +class Executor: + def __init__(self, faults: Faults) -> None: + self._faults = faults + + def run(self, milestone: int) -> ExecutorOutcome: + return self._faults.check("executor") or ExecutorOutcome("completed", "done") + + +class Publisher: + def __init__(self, faults: Faults) -> None: + self._faults = faults + self._opened = 0 + + def reconcilers(self) -> dict[str, Any]: + return {} + + def publish(self, **kwargs: Any) -> PublishResult: + self._faults.check("publisher") + self._opened += 1 + pr = PullRequest(number=self._opened, state="open", head="h", url="u") + return PublishResult(pr, f"opened PR #{self._opened}") + + +def next_reply(store: StateStore, used_project: bool) -> str: + """What an operator following the prompts would say next.""" + state = store.get_state() + assert state is not None + data = state.data + match state.state: + case "IDLE": + return "status" if used_project else "use demo" + case "AWAIT_ARCH_SIGNOFF": + return "approve" + case "AWAIT_MILESTONE_SIGNOFF": + if data["gate"]["approved"]: + return f"start milestone {data['milestone']}" + return "approve" + case "PR_OPEN": + return "next milestone" if data["milestone"] < 2 else "project complete" + case "AWAIT_INPUT": + return "continue" + raise AssertionError(f"stranded in {state.state}") + + +FAULTS = [ + ("planner", 1, RuntimeError("bug in planning")), + ("planner", 1, PlanningError("planner returned 400")), + ("planner", 1, AskHuman("which language?")), + ("planner", 2, RuntimeError("bug in planning")), + ("planner", 3, OSError("disk full")), + ("executor", 1, RuntimeError("bug in the executor")), + ("executor", 1, ExecutorOutcome("escalate", "stuck")), + ("executor", 1, ExecutorOutcome("capped", "max turns")), + ("executor", 2, ExecutorOutcome("stopped", "operator")), + ("publisher", 1, RuntimeError("bug in publishing")), + ("publisher", 1, GiteaError("gitea is down")), + ("publisher", 2, RuntimeError("bug in publishing")), +] + + +@pytest.mark.parametrize( + ("site", "nth", "fault"), FAULTS, ids=[f"{s}{n}-{f!r}" for s, n, f in FAULTS] +) +def test_a_fault_anywhere_never_strands_the_agent( + tmp_path: Path, site: str, nth: int, fault: Any +) -> None: + faults = Faults(site, nth, fault) + store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())) + transport = InMemoryTransport() + Orchestrator( + store, + transport, + Planner(faults), + operators={"don"}, + executor=Executor(faults), + publisher=Publisher(faults), + ).start() + + used_project = False + for _ in range(40): + reply = next_reply(store, used_project) + used_project = used_project or reply.startswith("use ") + transport.receive("don", reply) + state = store.get_state() + assert state is not None + assert state.state not in WORKING, ( + f"after {reply!r}: stranded in {state.state}; " + f"last said: {transport.outbound[-1].text!r}" + ) + if reply == "project complete": + break + else: + raise AssertionError("never reached project complete") + assert faults.fired + assert store.get_state().state == IDLE + + +def test_a_signoff_state_without_its_gate_offers_a_way_out(tmp_path: Path) -> None: + faults = Faults("none", 0, None) + store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())) + transport = InMemoryTransport() + orch = Orchestrator(store, transport, Planner(faults), operators={"don"}) + orch.start() + transport.receive("don", "use demo") + state = store.get_state() + store.transition( + AWAIT_ARCH_SIGNOFF, + cause="test", + project="demo", + data={**state.data, "gate": None}, + ) + orch._data["gate"] = None # as a restart would load it + transport.receive("don", "approve") + assert store.get_state().state == AWAIT_INPUT + transport.receive("don", "continue") + assert store.get_state().state == AWAIT_ARCH_SIGNOFF + + +class RaisingReviewExecutor: + def run(self, milestone: int) -> ExecutorOutcome: + raise RuntimeError("bug while addressing review") + + +def test_a_failing_review_response_keeps_pr_open_and_says_so( + tmp_path: Path, +) -> None: + faults = Faults("none", 0, None) + store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())) + transport = InMemoryTransport() + orch = Orchestrator( + store, + transport, + Planner(faults), + operators={"don"}, + gitea_operators={"don"}, + executor=Executor(faults), + publisher=Publisher(faults), + ) + orch.start() + for reply in ("use demo", "approve", "approve", "start milestone 1"): + transport.receive("don", reply) + assert store.get_state().state == PR_OPEN + orch._executor = RaisingReviewExecutor() + + class Comment: + id, author, body = 1, "don", "add a test" + + orch.handle_review_comment(Comment()) + assert store.get_state().state == PR_OPEN + assert "could not handle a review comment" in transport.outbound[-1].text + transport.receive("don", "project complete") + assert store.get_state().state == IDLE + + +def test_the_stub_configuration_reaches_every_state(tmp_path: Path) -> None: + store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())) + transport = InMemoryTransport() + Orchestrator( + store, transport, Planner(Faults("none", 0, None)), operators={"don"} + ).start() + for reply in ("use demo", "approve", "approve", "start milestone 1"): + transport.receive("don", reply) + assert store.get_state().state == PR_OPEN + transport.receive("don", "next milestone") + assert store.get_state().state == AWAIT_MILESTONE_SIGNOFF diff --git a/tests/integration/test_milestone_live.py b/tests/integration/test_milestone_live.py index a10e63c..18337b1 100644 --- a/tests/integration/test_milestone_live.py +++ b/tests/integration/test_milestone_live.py @@ -16,6 +16,7 @@ from nisse_dev.control import ( AWAIT_INPUT, AWAIT_MILESTONE_SIGNOFF, EXECUTING, + PR_OPEN, Orchestrator, ) from nisse_dev.executor import LLMExecutor @@ -45,7 +46,9 @@ def _drive_to_execution( if state.state == AWAIT_INPUT: transport.receive("don", "Use your best judgement and proceed.") elif state.state == AWAIT_MILESTONE_SIGNOFF and gate.get("approved"): + # The milestone runs to completion inside this command. transport.receive("don", "start milestone 1") + return elif state.state in (AWAIT_ARCH_SIGNOFF, AWAIT_MILESTONE_SIGNOFF): transport.receive("don", "approve") else: @@ -85,7 +88,9 @@ def test_plan_approve_execute_produces_a_real_file( state = store.get_state() assert state is not None - assert state.state in (EXECUTING, AWAIT_INPUT) + # Never EXECUTING at rest: with no publisher a finished milestone waits in + # PR_OPEN, and one that stopped short waits for the operator. + assert state.state in (PR_OPEN, AWAIT_INPUT) created = list(workspace.root.rglob("*.py")) assert created, f"executor created no Python file; said: {transport.outbound[-1]}"