fix: amplifier panes no longer stuck blue on provider errors

crates/freshell-activity/src/amplifier/mod.rs

@@ -9,8 +9,8 @@
 //!   by the caller's inotify events or force-read failsafes).
 //! * [`tracker`] — `server/coding-cli/amplifier-activity-tracker.ts` (the
 //!   terminal-keyed state machine: PTY Enter is only PROVISIONALLY busy;
-//!   `prompt:submit` confirms; `prompt:complete`/`session:end` is the single
-//!   turn boundary).
+//!   `prompt:submit` confirms; `prompt:complete` / `session:end` / root
+//!   `orchestrator:complete` are the turn-end boundaries).
 
 pub mod reducer;
 pub mod tailer;

crates/freshell-activity/src/amplifier/reducer.rs

@@ -6,8 +6,11 @@
 //!
 //! Contract facts encoded (plan §2 of the legacy durability plan):
 //! * `prompt:submit` is the ONLY input that (re)enters busy (E2/E5).
-//! * `prompt:complete` is the single turn boundary (E2/E3).
-//! * `session:end` while busy ends the turn (E7); while idle it is ignored.
+//! * Turn-end boundary SET (2026-08-10 amendment): `prompt:complete` (E2/E3),
+//!   `session:end` while busy (E7), and `orchestrator:complete` with a null
+//!   `data.parent_id` (provider-error turns never write `prompt:complete`).
+//!   The FIRST boundary record ends the turn; later ones land at idle and
+//!   are ignored.
 //! * `session:resume` never implies a phase change (E7).
 //! * Transitions key on event TYPE only; timestamps are carried through for
 //!   `at` fields but never used to order or gate transitions (E3).
@@ -208,6 +211,40 @@ pub fn reduce_amplifier_event(
             let at = record.ts.clone();
             (next, vec![ReducerEffect::TurnCompleted { at }])
         }
+        "orchestrator:complete" => {
+            // Turn-end boundary (2026-08-10 stuck-busy fix; parity with
+            // server/coding-cli/amplifier-events-reducer.ts,
+            // docs/plans/2026-08-10-amplifier-stuck-busy.md). On
+            // provider-error turns the CLI writes `provider:error` then
+            // `orchestrator:complete` and NEVER `prompt:complete` (27/27
+            // observed error turns), so without this arm a pane stays busy
+            // forever. This is a real CLI-written turn-end record, not a
+            // fabricated completion (the deadman policy stands).
+            //
+            // Exactly-once: on healthy turns this record precedes
+            // `prompt:complete` (724/724 observed; structural in the CLI
+            // source); the later `prompt:complete` lands at idle and the
+            // phase guard swallows it. Known accepted tradeoff: rare
+            // (~0.06%, 3/4,718 census turns) stray mid-turn
+            // `orchestrator:complete` records (status success or error — no
+            // status gate can block them) now end the turn early — one
+            // early completion beats an eternally stuck pane. Transitions
+            // still key on event TYPE only (E3): no status gate.
+            //
+            // Sub-agent guard: sub-agent sessions write their own
+            // events.jsonl and root-file records always carry
+            // `data.parent_id` null — a non-null parent_id can only be a
+            // sub-agent record and must never end the root turn.
+            if record.parent_id.is_some() {
+                return (next, Vec::new());
+            }
+            if next.phase != LifecyclePhase::Busy {
+                return (next, Vec::new());
+            }
+            next.phase = LifecyclePhase::Idle;
+            let at = record.ts.clone();
+            (next, vec![ReducerEffect::TurnCompleted { at }])
+        }
         "session:config" => {
             let Some(cwd) = record.config_cwd.clone() else {
                 return (next, Vec::new());
@@ -219,8 +256,9 @@ pub fn reduce_amplifier_event(
             )
         }
         // session:resume never implies busy; everything else (session:start,
-        // execution:*, llm:*, tool:*, orchestrator:*, ...) never changes phase
-        // — post-complete background naming events are covered here (E2).
+        // execution:*, llm:*, tool:*, and orchestrator:* other than
+        // orchestrator:complete, ...) never changes phase — post-complete
+        // background naming events are covered here (E2).
         _ => (next, Vec::new()),
     }
 }
@@ -277,10 +315,87 @@ mod tests {
         assert!(effects.is_empty());
     }
 
+    #[test]
+    fn orchestrator_complete_ends_a_busy_turn_on_the_provider_error_path() {
+        // The stuck-busy bug: provider:error then orchestrator:complete,
+        // nothing after — prompt:complete never arrives.
+        let state = create_reducer_state();
+        let (state, _) = reduce_amplifier_event(&state, &record("prompt:submit"));
+        let (state, effects) = reduce_amplifier_event(&state, &record("provider:error"));
+        assert_eq!(state.phase, LifecyclePhase::Busy);
+        assert!(effects.is_empty());
+        let (state, effects) = reduce_amplifier_event(&state, &record("orchestrator:complete"));
+        assert_eq!(state.phase, LifecyclePhase::Idle);
+        assert!(matches!(effects[0], ReducerEffect::TurnCompleted { .. }));
+    }
+
+    #[test]
+    fn orchestrator_complete_then_late_prompt_complete_completes_exactly_once() {
+        let state = create_reducer_state();
+        let (state, _) = reduce_amplifier_event(&state, &record("prompt:submit"));
+        let (state, effects) = reduce_amplifier_event(&state, &record("orchestrator:complete"));
+        assert_eq!(state.phase, LifecyclePhase::Idle);
+        assert!(matches!(effects[0], ReducerEffect::TurnCompleted { .. }));
+        let (state, effects) = reduce_amplifier_event(&state, &record("prompt:complete"));
+        assert_eq!(state.phase, LifecyclePhase::Idle);
+        assert!(effects.is_empty());
+    }
+
+    #[test]
+    fn prompt_complete_then_orchestrator_complete_completes_exactly_once() {
+        let state = create_reducer_state();
+        let (state, _) = reduce_amplifier_event(&state, &record("prompt:submit"));
+        let (state, effects) = reduce_amplifier_event(&state, &record("prompt:complete"));
+        assert!(matches!(effects[0], ReducerEffect::TurnCompleted { .. }));
+        let (state, effects) = reduce_amplifier_event(&state, &record("orchestrator:complete"));
+        assert_eq!(state.phase, LifecyclePhase::Idle);
+        assert!(effects.is_empty());
+    }
+
+    #[test]
+    fn orchestrator_complete_at_idle_is_a_no_op() {
+        let state = create_reducer_state();
+        let (state, effects) = reduce_amplifier_event(&state, &record("orchestrator:complete"));
+        assert_eq!(state.phase, LifecyclePhase::Idle);
+        assert!(effects.is_empty());
+    }
+
+    #[test]
+    fn subagent_orchestrator_complete_never_ends_the_root_turn() {
+        let state = create_reducer_state();
+        let (state, _) = reduce_amplifier_event(&state, &record("prompt:submit"));
+        let sub = ParsedRecord::from_json(&json!({
+            "ts": "2026-07-23T10:00:00.000Z",
+            "schema": { "name": "amplifier.log", "ver": "1.0.0" },
+            "event": "orchestrator:complete",
+            "session_id": "sess-1",
+            "data": { "parent_id": "0000000000000000-59ae93e4abde4aca_sub-agent" }
+        }))
+        .unwrap();
+        let (state, effects) = reduce_amplifier_event(&state, &sub);
+        assert_eq!(state.phase, LifecyclePhase::Busy);
+        assert!(effects.is_empty());
+    }
+
+    #[test]
+    fn other_orchestrator_events_never_end_a_busy_turn() {
+        let state = create_reducer_state();
+        let (state, _) = reduce_amplifier_event(&state, &record("prompt:submit"));
+        let (state, effects) =
+            reduce_amplifier_event(&state, &record("orchestrator:steering_injected"));
+        assert_eq!(state.phase, LifecyclePhase::Busy);
+        assert!(effects.is_empty());
+    }
+
     #[test]
     fn session_resume_and_noise_events_never_change_phase() {
         let state = create_reducer_state();
-        for event in ["session:resume", "session:start", "execution:start"] {
+        for event in [
+            "session:resume",
+            "session:start",
+            "execution:start",
+            "orchestrator:steering_injected",
+        ] {
             let (next, effects) = reduce_amplifier_event(&state, &record(event));
             assert_eq!(next.phase, LifecyclePhase::Idle, "{event}");
             assert!(effects.is_empty(), "{event}");

crates/freshell-activity/src/amplifier/tailer.rs

@@ -34,7 +34,10 @@ pub const READ_BATCH_MAX_BYTES: u64 = 16 * 1024 * 1024;
 
 /// Lifecycle event-name prefixes the reducer cares about; lines are checked
 /// with plain substring scans (both `"event":"x` and `"event": "x`).
-const EVENT_PREFIXES: [&str; 4] = ["session:", "prompt:", "execution:", "orchestrator:steering"];
+/// `orchestrator:` covers `orchestrator:complete` (a turn-end boundary since
+/// the 2026-08-10 stuck-busy fix — without this the reducer never sees it)
+/// plus `orchestrator:steering_injected` (~2 extra parses per turn).
+const EVENT_PREFIXES: [&str; 4] = ["session:", "prompt:", "execution:", "orchestrator:"];
 
 fn matches_prefilter(line: &str) -> bool {
     EVENT_PREFIXES.iter().any(|prefix| {
@@ -471,6 +474,47 @@ mod tests {
         }
     }
 
+    #[test]
+    fn prefilter_admits_orchestrator_complete() {
+        // orchestrator:complete is a turn-end boundary (2026-08-10 stuck-busy
+        // fix): if the prefilter drops it, the reducer arm is dead code.
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("events.jsonl");
+        std::fs::write(
+            &path,
+            [
+                line("session:start"),
+                line("prompt:submit"),
+                line("orchestrator:complete"),
+                line("prompt:complete"),
+            ]
+            .concat(),
+        )
+        .unwrap();
+
+        let mut tailer = AmplifierEventsTailer::new(&path);
+        tailer.attach(AttachAt::Start).unwrap();
+        match tailer.read() {
+            TailerReadOutcome::Ok {
+                records,
+                skipped_lines,
+                ..
+            } => {
+                assert_eq!(
+                    records.iter().map(|r| r.event.as_str()).collect::<Vec<_>>(),
+                    vec![
+                        "session:start",
+                        "prompt:submit",
+                        "orchestrator:complete",
+                        "prompt:complete"
+                    ]
+                );
+                assert_eq!(skipped_lines, 0);
+            }
+            other => panic!("expected ok, got {other:?}"),
+        }
+    }
+
     #[test]
     fn file_reset_degrades_never_guesses() {
         let dir = tempfile::tempdir().unwrap();

crates/freshell-activity/src/amplifier/tracker.rs

@@ -7,9 +7,10 @@
 //! * PTY Enter (`note_input` + submit-shaped) is only a PROVISIONAL busy with
 //!   a submit-grace reversion (one force-read retry, then a silent revert —
 //!   no turn.complete). A `prompt:submit` record (reducer `TurnBegan` effect
-//!   via [`AmplifierActivityTracker::apply_lifecycle`]) confirms busy;
-//!   `prompt:complete`/`session:end` (`TurnCompleted`) is the single turn
-//!   boundary and emits exactly one turn.complete via the ledger.
+//!   via [`AmplifierActivityTracker::apply_lifecycle`]) confirms busy; a
+//!   turn-end record — `prompt:complete` / `session:end` / root
+//!   `orchestrator:complete` (`TurnCompleted`) — ends the turn and emits
+//!   exactly one turn.complete via the ledger.
 //! * PTY output only refreshes liveness (feeds the deadman). The deadman
 //!   never fabricates a completion: it requests a force-read of the events
 //!   tail and STAYS busy.

crates/freshell-ws/src/activity.rs

@@ -3143,6 +3143,84 @@ mod tests {
         );
     }
 
+    /// Provider-error turn (2026-08-10 stuck-busy fix): the CLI writes
+    /// provider:error then orchestrator:complete and NEVER prompt:complete.
+    /// The lane must still complete the turn — this is the only Rust test
+    /// that exercises tailer prefilter + reducer + tracker together for the
+    /// error path (a reducer-only fix is invisible here).
+    #[tokio::test(flavor = "multi_thread")]
+    async fn amplifier_events_lane_completes_error_turn_on_orchestrator_complete() {
+        let dir = tempfile::tempdir().unwrap();
+        let events_path = dir.path().join("events.jsonl");
+        std::fs::write(
+            &events_path,
+            [
+                amplifier_line("session:start"),
+                amplifier_line("prompt:submit"),
+            ]
+            .concat(),
+        )
+        .unwrap();
+
+        let (hub, mut rx) = hub();
+        observer_send(
+            &hub,
+            ActivityEvent::Created {
+                terminal_id: "t1".into(),
+                mode: "amplifier".into(),
+                resume_session_id: None,
+                at: now_ms(),
+            },
+        );
+        observer_send(
+            &hub,
+            ActivityEvent::Input {
+                terminal_id: "t1".into(),
+                data: "\r".into(),
+                at: now_ms(),
+            },
+        );
+        let busy = next_frame_matching(&mut rx, "amplifier.activity.updated", 2_000, |v| {
+            v["upsert"][0]["phase"] == "busy"
+        })
+        .await
+        .expect("provisional busy upsert");
+        assert_eq!(busy["upsert"][0]["terminalId"], "t1");
+
+        hub.attach_amplifier_association("t1", "sess-1", &events_path);
+        next_frame_matching(&mut rx, "amplifier.activity.updated", 3_000, |v| {
+            v["upsert"][0]["sessionId"] == "sess-1"
+        })
+        .await
+        .expect("bind upsert");
+
+        // The turn dies on a provider error: append exactly what the real
+        // CLI writes (provider:error is prefilter noise; orchestrator:complete
+        // must end the turn).
+        let mut f = std::fs::OpenOptions::new()
+            .append(true)
+            .open(&events_path)
+            .unwrap();
+        f.write_all(
+            [
+                amplifier_line("provider:error"),
+                amplifier_line("orchestrator:complete"),
+            ]
+            .concat()
+            .as_bytes(),
+        )
+        .unwrap();
+        f.flush().unwrap();
+        drop(f);
+
+        let complete = next_frame_of_type(&mut rx, "terminal.turn.complete", 5_000)
+            .await
+            .expect("turn.complete driven by orchestrator:complete");
+        assert_eq!(complete["provider"], "amplifier");
+        assert_eq!(complete["sessionId"], "sess-1");
+        assert_eq!(complete["completionSeq"], 1);
+    }
+
     /// Steady-state zero-wake proof: idle tracked terminals arm NO timers and
     /// read NO files. (The 20-agents-idle scenario in miniature.)
     #[tokio::test(flavor = "multi_thread")]