Never leave the agent stranded in a working state
CI / checks (push) Skipped
CI / checks (pull_request) Failing after 1m20s
CI / checks (push) Skipped
CI / checks (pull_request) Failing after 1m20s
The first live run found four dead ends one at a time: states where no operator message could move the agent. Most had one shape: work ended, by an exception or by a path that never transitioned, and left the agent in PLANNING_* or EXECUTING, where every message is answered "busy". The worker only prints an exception, so that would keep happening. - Every job the orchestrator runs for an operator message or a Gitea comment now goes through _settled(). If the job leaves a working state behind, whatever the reason, the agent moves to AWAIT_INPUT with the continue / re-plan choice a restart offers, and says why. An exception in a waiting state is reported and the state kept, since its commands still lead on. - A sign-off state whose gate record is missing asks to plan again instead of crashing on every reply. - The stub configurations (no executor, no publisher) complete the milestone and wait in PR_OPEN instead of resting in EXECUTING. test_no_stranding.py drives a whole two-milestone project with a fault injected at each planner, executor and publisher call, and asserts after every message that the agent is not in a working state and that the project still reaches 'project complete'. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
This commit is contained in:
1 parent
1ccea20a10
commit
92e5452161
3 files changed
+325
-10
No files matched your search
@@ -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(
|
||||
|
||||
@@ -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
|
||||
Reference in new issue
Block a user