Merge pull request 'Never leave the agent stranded in a working state' (#66) from no-stranded-states into main

Reviewed-on: dongar/nisse#66
Reviewed-by: Don Garrett <dgarrett@acm.org>

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 <project>'")
         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:

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(

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

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]}"