Rename open to stalled and autopilot to auto-confirm; add pushed state; declare idempotency; propose maturity and projects model (RBT-WP-0008)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: sonnet
Assistant-Process: 295952@bnt-lap001
Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477

tests/test_run2.py

@@ -196,25 +196,25 @@ def test_irreversible_step_needs_the_typed_phrase(tmp_path):
 
 def test_preauthorization_only_for_irreversible_steps(tmp_path):
     with pytest.raises(R.RunError, match="only irreversible"):
-        new(tmp_path, mode="autopilot", preauthorize=["light-lamp"])
+        new(tmp_path, mode="auto-confirm", preauthorize=["light-lamp"])
 
 
-# ---- autopilot ------------------------------------------------------------------------------------------------
+# ---- auto-confirm ------------------------------------------------------------------------------------------------
 
 def drive(r, ticks=60, now=None, operator=None):
     log = []
     for _ in range(ticks):
-        out = R.autopilot_tick(r, now)
+        out = R.auto_confirm_tick(r, now)
         log.append(out)
         if out["action"] in ("finished", "stopped", "awaiting-operator", "waiting"):
             break
         if out["action"] == "started" and operator:
-            operator(out["item"])                   # the operator runs the step's command while autopilot confirms
+            operator(out["item"])                   # the operator runs the step's command while auto-confirm confirms
     return log
 
 
-def test_autopilot_confirms_when_preconditions_are_met_and_waits_for_the_operator_on_manual_only(tmp_path):
-    r = new(tmp_path, mode="autopilot")
+def test_auto_confirm_confirms_when_preconditions_are_met_and_waits_for_the_operator_on_manual_only(tmp_path):
+    r = new(tmp_path, mode="auto-confirm")
 
     def operator(sid):
         cmd = R.command_for(r, sid)
@@ -232,37 +232,37 @@ def test_autopilot_confirms_when_preconditions_are_met_and_waits_for_the_operato
     R.start(r, "close-log", phrase=PHRASE)                                            # the operator takes over the last step
     subprocess.run(R.command_for(r, "close-log"), shell=True, check=True, capture_output=True)
     R.verify(r, "close-log", confirm=True); R.complete(r, "close-log")
-    assert R.autopilot_tick(r)["action"] == "completed"                               # gate night-on-record
-    assert R.autopilot_tick(r)["action"] == "finished"
+    assert R.auto_confirm_tick(r)["action"] == "completed"                               # gate night-on-record
+    assert R.auto_confirm_tick(r)["action"] == "finished"
     R1.finalize(r, "succeeded")
     assert r.data["outcome"] == "succeeded" and not r.data["deviations"]
 
 
-def test_autopilot_timeout_stop(tmp_path):
+def test_auto_confirm_timeout_stop(tmp_path):
     def m(d):
         d["timeout"] = {"seconds": 30, "on_timeout": "stop"}
-    r = new(tmp_path, m, mode="autopilot")
+    r = new(tmp_path, m, mode="auto-confirm")
     log = drive(r)                                     # write-logbook's file check cannot pass: nobody ran its command
     assert log[-1]["action"] == "waiting" and log[-1]["item"] == "write-logbook"
     late = datetime.now(timezone.utc) + timedelta(seconds=31)
-    out = R.autopilot_tick(r, late)
+    out = R.auto_confirm_tick(r, late)
     assert out["action"] == "stopped" and "timed out after 30s" in out["detail"]
-    assert R.autopilot_tick(r)["action"] == "stopped"                                # stays stopped
+    assert R.auto_confirm_tick(r)["action"] == "stopped"                                # stays stopped
     assert r.data["deviations"] == []                                                # stopping is not a deviation
 
 
-def test_autopilot_push_through_records_deviation_and_moves_on(tmp_path):
+def test_auto_confirm_push_through_records_deviation_and_moves_on(tmp_path):
     def m(d):
         d["timeout"] = {"seconds": 30, "on_timeout": "push-through"}
-    r = new(tmp_path, m, mode="autopilot")
+    r = new(tmp_path, m, mode="auto-confirm")
     drive(r)
     late = datetime.now(timezone.utc) + timedelta(seconds=31)
-    out = R.autopilot_tick(r, late)
+    out = R.auto_confirm_tick(r, late)
     assert out["action"] == "push-through" and out["item"] == "write-logbook"
-    assert r.data["steps"]["write-logbook"]["state"] == "done" and r.data["steps"]["write-logbook"]["pushed_through"]
+    assert r.data["steps"]["write-logbook"]["state"] == "pushed"                     # its own state: not done, not failed
     dev = r.data["deviations"][0]
-    assert dev["kind"] == "push-through" and dev["mode"] == "autopilot" and dev["risk_acknowledged"] is False
-    nxt = R.autopilot_tick(r, late)                                                    # continues to the manual-only step
+    assert dev["kind"] == "push-through" and dev["mode"] == "auto-confirm" and dev["risk_acknowledged"] is False
+    nxt = R.auto_confirm_tick(r, late)                                                    # continues to the manual-only step
     assert nxt["item"] == "close-log" and nxt["action"] == "awaiting-operator"
     with pytest.raises(R1.RunError, match="deviations"):
         for k in r.data["steps"]:
@@ -270,37 +270,37 @@ def test_autopilot_push_through_records_deviation_and_moves_on(tmp_path):
         R1.exercise_update(r)
 
 
-def test_autopilot_gate_push_through_and_stop(tmp_path):
+def test_auto_confirm_gate_push_through_and_stop(tmp_path):
     def m(d):
         d["stages"][0]["gates"][0]["checks"].append({"done_when": "always fails", "command": "false", "expect": "exit-0"})
         d["stages"][0]["gates"][0]["timeout"] = {"seconds": 10, "on_timeout": "push-through"}
-    r = new(tmp_path, m, mode="autopilot")
+    r = new(tmp_path, m, mode="auto-confirm")
     drive(r)
     assert r.data["gates"]["lamp-check"]["state"] == "failed"        # all checks evaluated, one failing
-    out = R.autopilot_tick(r, datetime.now(timezone.utc) + timedelta(seconds=11))
-    assert out["action"] == "push-through" and r.data["gates"]["lamp-check"]["state"] == "waived"
+    out = R.auto_confirm_tick(r, datetime.now(timezone.utc) + timedelta(seconds=11))
+    assert out["action"] == "push-through" and r.data["gates"]["lamp-check"]["state"] == "pushed"   # not "waived": nobody approved it
     assert r.data["deviations"][0]["item"] == "lamp-check" and r.data["deviations"][0]["kind"] == "push-through"
 
 
-def test_autopilot_never_confirms_irreversible_steps_unless_preauthorized(tmp_path):
+def test_auto_confirm_never_confirms_irreversible_steps_unless_preauthorized(tmp_path):
     def m(d):
         for st in d["stages"]:
             for s in st.get("steps", []):
-                s.pop("autopilot", None)                                             # allow automation of close-log
-    r = new(tmp_path, m, mode="autopilot")
+                s.pop("auto_confirm", None)                                             # allow automation of close-log
+    r = new(tmp_path, m, mode="auto-confirm")
     drive(r, operator=lambda sid: (subprocess.run(R.command_for(r, sid), shell=True, check=True, capture_output=True) if R.command_for(r, sid) else None))
-    last = R.autopilot_tick(r)
+    last = R.auto_confirm_tick(r)
     assert last["action"] == "awaiting-operator" and "not pre-authorized" in last["detail"]
-    r2 = new(tmp_path, m, mode="autopilot", preauthorize=["close-log"])
+    r2 = new(tmp_path, m, mode="auto-confirm", preauthorize=["close-log"])
     drive(r2, operator=lambda sid: (subprocess.run(R.command_for(r2, sid), shell=True, check=True, capture_output=True) if R.command_for(r2, sid) else None))
     assert r2.data["steps"]["close-log"]["state"] in ("running", "done")
     assert r2.data["steps"]["close-log"]["authorized_by"] == "tester"
 
 
-def test_manual_mode_rejects_autopilot_and_autopilot_rejects_overrides(tmp_path):
-    with pytest.raises(R.RunError, match="autopilot run"):
-        R.autopilot_tick(new(tmp_path))
-    r = new(tmp_path, mode="autopilot")
+def test_manual_mode_rejects_auto_confirm_and_auto_confirm_rejects_overrides(tmp_path):
+    with pytest.raises(R.RunError, match="auto-confirm run"):
+        R.auto_confirm_tick(new(tmp_path))
+    r = new(tmp_path, mode="auto-confirm")
     with pytest.raises(R.RunError, match="manual mode"):
         R.start(r, "warn", alt="sound-horn", override_reason="x", acknowledge_risk=True)
 
@@ -314,3 +314,32 @@ def test_finished_v02_run_is_read_only_and_scores_count_it(tmp_path):
         R.set_datapoint(r, "weather", "clear")
     assert R1.scorecard(r.pack)["abandoned"] == 1
     assert R1.load(r.pack, r.data["run_id"]).data["spec"] == "runbook-run/v0.2"      # reloads through the shared loader
+
+
+def test_pushed_items_are_a_distinct_state_that_lets_the_run_continue_but_never_succeed(tmp_path):
+    def m(d):
+        d["timeout"] = {"seconds": 5, "on_timeout": "push-through"}
+        light = next(s for st in d["stages"] for s in st.get("steps", []) if s["id"] == "light-lamp")
+        light["verify"] = {"done_when": "never true here", "command": "false", "expect": "exit-0"}
+        d["stages"][0]["gates"][0]["timeout"] = {"seconds": 5, "on_timeout": "push-through"}   # the gate's own timeout wins
+    r = new(tmp_path, m, mode="auto-confirm")
+    late = datetime.now(timezone.utc) + timedelta(seconds=6)
+    seen = []
+    for _ in range(40):
+        out = R.auto_confirm_tick(r, late)
+        seen.append(out["action"])
+        if out["action"] in ("stopped", "awaiting-operator", "finished"):
+            break
+        late = late + timedelta(seconds=6)
+    assert r.data["steps"]["light-lamp"]["state"] == "pushed"
+    dp = r.data["datapoints"]["lamp_lit"]
+    assert dp["state"] == "pushed" and "value" not in dp and "error" not in dp        # not set, not err
+    assert "lamp_lit" not in r.data["params"]                                          # never usable as a value
+    assert R.is_complete(r, "light-lamp")                                              # dependents may proceed
+    assert r.data["gates"]["lamp-check"]["state"] == "pushed"                          # datapoints_ok needs 'ok', so the gate was pushed too
+    kinds = [d["kind"] for d in r.data["deviations"]]
+    assert kinds.count("push-through") >= 2 and "override" not in kinds
+    with pytest.raises(R1.RunError, match="succeeds only when every step is done"):
+        R1.finalize(r, "succeeded")
+    R1.finalize(r, "failed", "pushed through to collect problems")                     # a clean-completion failure, finishable as failed
+    assert r.data["outcome"] == "failed"