Keep a pending question across restarts, and make the restart question answerable #57

Merged
dongar merged 1 commits from fix-await-input-restart into main 2026-09-29 15:54:02 +00:00
2 changed files with 161 additions and 7 deletions

No files matched your search

+67 -7
View File
@@ -56,6 +56,12 @@ TRANSITIONS = {
}
_BOOT_CHOICES = (
"reply 'continue' to pick up where the restart cut in, "
"or 're-plan' to plan the current step again"
)
class TransitionError(RuntimeError):
pass
@@ -251,9 +257,18 @@ class Orchestrator:
prompt=(
"restart found unresolved work: "
+ "; ".join(report.notes)
+ " — how should I proceed?"
+ f" — {_BOOT_CHOICES}"
),
pending={"kind": "boot"},
pending={"kind": "boot", "interrupted": state.state},
)
elif state.state == AWAIT_INPUT and self._data.get("pending"):
# Waiting on the operator is a committed state (A3 step 8):
# nothing was in flight, and the question and what answering it
# resumes are durable. Ask it again rather than replacing it.
pending = self._data["pending"]
self._transport.ask_question(
"restarted while waiting on you: "
+ pending.get("prompt", "my last question still stands")
)
elif state.state == IDLE or (
state.state in (AWAIT_ARCH_SIGNOFF, AWAIT_MILESTONE_SIGNOFF) and gate
@@ -270,8 +285,8 @@ class Orchestrator:
# never blind-resume (A3.6).
self._await_input(
cause=f"boot: restarted in {state.state}",
prompt=f"restarted mid-{state.state}; continue or re-plan?",
pending={"kind": "boot"},
prompt=f"restarted mid-{state.state}; {_BOOT_CHOICES}",
pending={"kind": "boot", "interrupted": state.state},
)
def _handle(self, msg: InboundMessage) -> None:
@@ -783,15 +798,21 @@ class Orchestrator:
self._unwatch(kind)
def _await_input(self, *, cause: str, prompt: str, pending: dict[str, Any]) -> None:
self._data["pending"] = pending
# The prompt is kept with what it resumes, so a restart can ask the
# same question again.
self._data["pending"] = {**pending, "prompt": prompt}
self._transition(AWAIT_INPUT, cause=cause)
self._transport.ask_question(prompt)
def _handle_answer(self, text: str) -> None:
pending = self._data.get("pending")
if pending is None:
# Reached only if state was forced externally; say what works
# rather than dead-ending the operator.
# Reached only if state was forced externally or a pending item
# was lost; starting over must still work from here.
if text.startswith("use "):
self.stop_watching()
self._begin_project(text[4:].strip())
return
self._transport.send_message(
"waiting for direction, but I have no pending item to resume; "
"try 'status', or 'use <project>' to start over"
@@ -804,6 +825,8 @@ class Orchestrator:
back = PLANNING_ARCH if gate_name == "arch" else MILESTONE_PLANNING
self._transition(back, cause="operator answered; resuming planning")
self._run_planning_turn(gate_name, answer=text)
case "boot":
self._resume_after_restart(pending, text)
case "milestone_stopped":
milestone = pending["milestone"]
self._transition(
@@ -819,6 +842,43 @@ class Orchestrator:
"noted. say 'status' for where things stand"
)
def _resume_after_restart(self, pending: dict[str, Any], text: str) -> None:
"""Answer to the question a restart asks (A3 step 7). Never a blind
resume: 'continue' re-runs the interrupted planning turn, or returns
an interrupted milestone to its approved plan to be started again."""
choice = text.strip().rstrip(".!").lower()
if choice not in ("continue", "re-plan", "replan"):
self._data["pending"] = pending
self._transport.ask_question(_BOOT_CHOICES)
return
milestone = int(self._data.get("milestone", 1))
gate = self._data.get("gate") or {}
interrupted = pending.get("interrupted")
if choice == "continue" and interrupted == EXECUTING and gate.get("approved"):
self._transition(
AWAIT_MILESTONE_SIGNOFF, cause="operator: continue after restart"
)
self._transport.send_message(
f"the milestone {milestone} plan is still approved; say "
f"'start milestone {milestone}' to run it"
)
return
if interrupted == PLANNING_ARCH:
gate_name = "arch"
elif interrupted in (MILESTONE_PLANNING, EXECUTING):
gate_name = f"milestone-{milestone}"
else:
# Not recorded (a question asked before the interrupted state
# was): judge by how far planning got.
planned = self._data.get("revs", {})
gate_name = gate.get("name") or (
"arch" if "arch" not in planned else f"milestone-{milestone}"
)
self._data["gate"] = None
back = PLANNING_ARCH if gate_name == "arch" else MILESTONE_PLANNING
self._transition(back, cause=f"operator: {choice} after restart")
self._run_planning_turn(gate_name)
def _transition(self, new_state: str, *, cause: str) -> None:
state = self._store.get_state()
current = state.state if state else None
+94
View File
@@ -10,8 +10,10 @@ from nisse_dev.control import (
AWAIT_MILESTONE_SIGNOFF,
EXECUTING,
IDLE,
PLANNING_ARCH,
AskHuman,
Orchestrator,
PlanningError,
StubAgent,
SubmitPlan,
)
@@ -153,3 +155,95 @@ def test_ask_human_pauses_and_answer_resumes(tmp_path: Path) -> None:
h.transport.receive("don", "monorepo")
assert h.state == AWAIT_ARCH_SIGNOFF
assert "monorepo" in h.transport.outbound[-1].text
# --- restarts (A3 steps 7 and 8) --------------------------------------------
class FailsOnce:
"""A planner whose first turn fails, as Opus 5.5's did on its first live
run when asked for forced tool use."""
def __init__(self) -> None:
self.calls = 0
def plan(
self,
gate: str,
project: str,
feedback: str | None = None,
answer: str | None = None,
) -> SubmitPlan | AskHuman:
self.calls += 1
if self.calls == 1:
raise PlanningError("planner returned 400")
return SubmitPlan(f"[stub] {gate} plan for {project}")
def test_restart_while_awaiting_input_asks_the_same_question(
tmp_path: Path,
) -> None:
agent = FailsOnce()
h = Harness(tmp_path, agent=agent)
h.transport.receive("don", "use demo")
assert h.state == AWAIT_INPUT
h = Harness(tmp_path, agent=agent) # restart on the same state
question = h.transport.outbound[-1]
assert question.kind == "question"
assert "restarted while waiting on you" in question.text
assert "planning for arch failed" in question.text
h.transport.receive("don", "Please retry.")
assert h.state == AWAIT_ARCH_SIGNOFF
def test_restart_mid_planning_offers_continue_and_it_replans(
tmp_path: Path,
) -> None:
h = Harness(tmp_path)
h.store.transition(PLANNING_ARCH, cause="test", project="demo", data={})
h = Harness(tmp_path)
assert h.state == AWAIT_INPUT
assert "'continue'" in h.outbound_texts()[-1]
h.transport.receive("don", "Please retry.") # neither choice: asked again
assert h.state == AWAIT_INPUT
assert h.transport.outbound[-1].kind == "question"
h.transport.receive("don", "continue")
assert h.state == AWAIT_ARCH_SIGNOFF
def test_restart_mid_execution_continue_returns_to_the_approved_plan(
h: Harness, tmp_path: Path
) -> None:
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
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
def test_restart_question_without_its_interrupted_state_still_replans(
tmp_path: Path,
) -> None:
# What the first live run left behind: a restart question recorded
# before the question kept its prompt or the state it interrupted.
h = Harness(tmp_path)
h.store.transition(
AWAIT_INPUT, cause="test", project="demo", data={"pending": {"kind": "boot"}}
)
h = Harness(tmp_path)
assert "my last question still stands" in h.outbound_texts()[-1]
h.transport.receive("don", "re-plan")
assert h.state == AWAIT_ARCH_SIGNOFF
def test_use_starts_over_when_nothing_is_pending(h: Harness) -> None:
h.store.transition(AWAIT_INPUT, cause="forced", project="demo", data={})
h.transport.receive("don", "use demo")
assert h.state == AWAIT_ARCH_SIGNOFF