fix: distinguish Pi turn outcomes in completion notifications

nebula_app/i18n/en-US.json

@@ -1,4 +1,7 @@
 {
+  "notifications": {
+    "turn_failed": "This turn failed."
+  },
   "language": {
     "self_name": "English",
     "system": "Follow system",

nebula_app/i18n/zh-CN.json

@@ -1,4 +1,7 @@
 {
+  "notifications": {
+    "turn_failed": "本轮失败。"
+  },
   "language": {
     "self_name": "简体中文",
     "system": "跟随系统",

nebula_app/src/ai_hook.rs

@@ -332,6 +332,7 @@ pub struct AiHookEvent {
     /// AI CLI identity, used as the toast title.
     pub source: String,
     pub kind: AiHookKind,
+    pub outcome: Option<AiTurnOutcome>,
     /// Human text when the event carries one (claude's notification message,
     /// codex's last assistant message).
     pub message: Option<String>,
@@ -369,6 +370,14 @@ pub struct AiHookEvent {
     pub attention: Option<AttentionContext>,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AiTurnOutcome {
+    Success,
+    Failed,
+    Cancelled,
+    Unknown,
+}
+
 impl AiHookEvent {
     pub fn capabilities(&self) -> AiHookCapabilities {
         capabilities_for(&self.source)
@@ -767,6 +776,16 @@ fn parse_envelope(bytes: &[u8]) -> Option<AiHookEvent> {
         answer,
         answer_cwd,
         pane,
+        outcome: if source == "pi" && kind == AiHookKind::TurnDone {
+            payload.get("outcome").map(|value| match value.as_str() {
+                Some("success") => AiTurnOutcome::Success,
+                Some("failed") => AiTurnOutcome::Failed,
+                Some("cancelled") => AiTurnOutcome::Cancelled,
+                _ => AiTurnOutcome::Unknown,
+            })
+        } else {
+            None
+        },
         source,
         kind,
         message,

nebula_app/src/ai_hook/win.rs

@@ -607,14 +607,14 @@ export const PebrelNotify = async ({ $, directory, worktree }) => {
 }
 "#;
 
-// Pi 官方扩展 API 在 agent_start/agent_end 提供稳定的回合边界。扩展只做
-// fire-and-forget 转发，且 NEBULA_HOOK_EXE 不存在时完全静默，因此全局安装
-// 不会影响从其他终端启动的 Pi。
+// agent_end 只结束一次尝试；Pi 0.80.4+ 的 agent_settled 才覆盖自动重试。
+// 旧版仍使用原有结束边界。无 hook 环境变量时，扩展不启动子进程。
 const PI_EXTENSION_TS: &str = r#"// Pebrel ↔ Pi bridge — AUTO-GENERATED by Pebrel, do not edit.
 import { spawn } from "node:child_process";
 import { openSync, readSync, closeSync } from "node:fs";
 import { randomUUID } from "node:crypto";
 import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
+import { VERSION } from "@earendil-works/pi-coding-agent";
 
 // Only provider IDs and the first JSONL metadata line are recovery identities.
 // A timestamped basename or a process ID cannot be handed to pi --session.
@@ -644,6 +644,9 @@ function sessionFor(ctx: any): { session_id?: string; session_file?: string } {
 
 export default function (pi: ExtensionAPI) {
   let active = false;
+  let outcome = "unknown";
+  const [major, minor, patch] = VERSION.split(".").map(Number);
+  const hasSettled = major > 0 || minor > 80 || (minor === 80 && patch >= 4);
   let sequence = 0;
   const sequenceEpoch = BigInt(Date.now()) * 1000000n;
   const bridge_instance = randomUUID();
@@ -656,6 +659,7 @@ export default function (pi: ExtensionAPI) {
       const event_id = `${bridge_instance}:${bridge_sequence}`;
       spawn(hook, ["pi", JSON.stringify({
         kind,
+        ...(kind === "done" ? { outcome } : {}),
         ...identity,
         bridge_instance,
         bridge_sequence,
@@ -669,18 +673,35 @@ export default function (pi: ExtensionAPI) {
     } catch (_) {}
   };
 
-  pi.on("agent_start", async (_event, ctx) => {
-    active = true;
-    send("prompt", ctx);
-  });
-  pi.on("agent_end", async (_event, ctx) => {
+  const finish = (ctx: any) => {
     if (!active) return;
     active = false;
     send("done", ctx);
+  };
+  pi.on("agent_start", async (_event, ctx) => {
+    active = true;
+    outcome = "unknown";
+    send("prompt", ctx);
   });
-  try { pi.on("session_start", async (_event, ctx) => send("session-start", ctx)); } catch (_) {}
+  pi.on("agent_end", async (event, ctx) => {
+    if (!active) return;
+    const last = event.messages?.findLast((message: any) => message.role === "assistant");
+    outcome = last?.stopReason === "stop" ? "success"
+      : last?.stopReason === "error" ? "failed"
+      : last?.stopReason === "aborted" ? "cancelled" : "unknown";
+    if (!hasSettled) finish(ctx);
+  });
+  if (hasSettled) pi.on("agent_settled", async (_event, ctx) => finish(ctx));
+  try { pi.on("session_start", async (_event, ctx) => {
+    active = false;
+    outcome = "unknown";
+    send("session-start", ctx);
+  }); } catch (_) {}
   try { pi.on("tool_result", async (_event, ctx) => send("tool-complete", ctx)); } catch (_) {}
-  try { pi.on("session_shutdown", async (_event, ctx) => send("session-end", ctx)); } catch (_) {}
+  try { pi.on("session_shutdown", async (_event, ctx) => {
+    active = false;
+    send("session-end", ctx);
+  }); } catch (_) {}
 }
 "#;
 

nebula_app/src/gpui_shell/terminal/view/runtime.rs

@@ -949,6 +949,14 @@ impl TerminalView {
                         self.agent_status_rule =
                             Some(format!("hook.background_tasks.active={active}"));
                     },
+                    AiHookKind::TurnDone
+                        if event.outcome.is_some_and(|outcome| {
+                            outcome != crate::ai_hook::AiTurnOutcome::Success
+                        }) =>
+                    {
+                        self.agent_status = AgentStatus::Idle;
+                        self.agent_turn_active = false;
+                    },
                     AiHookKind::TurnDone | AiHookKind::NeedsAttention => {
                         let screen_asks =
                             event.kind == AiHookKind::TurnDone && self.screen_tail_asks();
@@ -979,14 +987,10 @@ impl TerminalView {
                 }));
             }
         } else if from_primary_agent
-            && event.kind == AiHookKind::TurnDone
-            && matches!(self.agent_status, AgentStatus::Done | AgentStatus::Blocked)
+            && let Some(notification) = event
+                .turn_notification(super::ui_language(), self.agent_status == AgentStatus::Blocked)
         {
-            cx.emit(TerminalViewEvent::Notification(crate::notify::Notification::AiTurn {
-                program: event.source.clone(),
-                message: event.message.clone(),
-                attention: self.agent_status == AgentStatus::Blocked,
-            }));
+            cx.emit(TerminalViewEvent::Notification(notification));
         }
         cx.emit(TerminalViewEvent::TitleChanged);
         cx.notify();

nebula_app/src/gpui_shell/terminal/view/startup_tests.rs

@@ -136,6 +136,74 @@ fn review_regression_cold_resume_survives_initial_prompt_and_clears_on_exit(
     }));
 }
 
+#[gpui::test]
+fn pi_cancelled_turn_becomes_idle_instead_of_completed(cx: &mut TestAppContext) {
+    let (view, window, _) = open(cx);
+    view.update(window, |view, cx| {
+        for payload in [
+            r#"{"kind":"prompt","session_id":"pi-test"}"#,
+            r#"{"kind":"done","outcome":"cancelled","session_id":"pi-test"}"#,
+        ] {
+            let wire = format!("nebula-hook/1 source=pi\n{payload}");
+            let event = crate::ai_hook::parse_remote_envelope(wire.as_bytes(), Some(view.pane_id))
+                .expect("Pi hook");
+            assert!(view.handle_ai_hook(&event, cx));
+        }
+        assert_eq!(view.agent_status, crate::ai_agents::AgentStatus::Idle);
+        assert!(!view.agent_turn_active);
+    });
+}
+
+#[gpui::test]
+fn pi_outcomes_emit_only_the_matching_notification(cx: &mut TestAppContext) {
+    use std::cell::RefCell;
+    use std::rc::Rc;
+
+    let (view, window, _) = open(cx);
+    let notifications = Rc::new(RefCell::new(Vec::new()));
+    let observed = notifications.clone();
+    let _subscription = view.update(window, |_, cx| {
+        cx.subscribe(&view, move |_, _, event, _| {
+            if let TerminalViewEvent::Notification(notification) = event {
+                observed.borrow_mut().push(notification.clone());
+            }
+        })
+    });
+    for (outcome, status, expected_count) in [
+        ("cancelled", crate::ai_agents::AgentStatus::Idle, 0),
+        ("unknown", crate::ai_agents::AgentStatus::Idle, 0),
+        ("failed", crate::ai_agents::AgentStatus::Idle, 1),
+        ("success", crate::ai_agents::AgentStatus::Done, 2),
+    ] {
+        view.update(window, |view, cx| {
+            for payload in [
+                serde_json::json!({"kind": "prompt", "session_id": "pi-test"}),
+                serde_json::json!({"kind": "done", "outcome": outcome, "session_id": "pi-test"}),
+            ] {
+                let wire = format!("nebula-hook/1 source=pi\n{payload}");
+                let event =
+                    crate::ai_hook::parse_remote_envelope(wire.as_bytes(), Some(view.pane_id))
+                        .expect("Pi hook");
+                assert!(view.handle_ai_hook(&event, cx));
+            }
+            assert_eq!(view.agent_status, status);
+            if outcome != "success" {
+                assert!(!view.agent_turn_active);
+            }
+        });
+        assert_eq!(notifications.borrow().len(), expected_count, "{outcome}");
+    }
+    let notifications = notifications.borrow();
+    let crate::notify::Notification::AiTurn { message, attention, .. } = &notifications[0] else {
+        panic!("expected a failure notification");
+    };
+    assert_eq!(
+        message.as_deref(),
+        Some(super::ui_language().text(crate::i18n::Message::NotificationsTurnFailed))
+    );
+    assert!(!attention);
+}
+
 #[gpui::test]
 fn failed_cold_resume_keeps_target_for_retry(cx: &mut TestAppContext) {
     let (view, window, receiver) = open(cx);

nebula_app/src/notify.rs

@@ -106,6 +106,31 @@ pub enum Notification {
     AiTurn { program: String, message: Option<String>, attention: bool },
 }
 
+impl crate::ai_hook::AiHookEvent {
+    pub(crate) fn turn_notification(
+        &self,
+        language: crate::display::UiLanguage,
+        attention: bool,
+    ) -> Option<Notification> {
+        use crate::ai_hook::{AiHookKind, AiTurnOutcome};
+        if self.kind != AiHookKind::TurnDone || self.active_background_tasks() > 0 {
+            return None;
+        }
+        let message = match self.outcome {
+            Some(AiTurnOutcome::Cancelled | AiTurnOutcome::Unknown) => return None,
+            Some(AiTurnOutcome::Failed) => {
+                Some(language.text(crate::i18n::Message::NotificationsTurnFailed).to_owned())
+            },
+            Some(AiTurnOutcome::Success) | None => self.message.clone(),
+        };
+        Some(Notification::AiTurn {
+            program: self.source.clone(),
+            message,
+            attention: attention && self.outcome != Some(AiTurnOutcome::Failed),
+        })
+    }
+}
+
 /// Longest notification body any shell may render.
 ///
 /// A toast/banner is a glance layer, not a reader. Codex's turn-complete
@@ -309,6 +334,31 @@ fn spawn_toast(title: String, body: String, activation: Option<ToastActivation>)
 mod delivery_tests {
     use super::*;
 
+    #[test]
+    fn pi_outcomes_do_not_turn_failure_or_cancellation_into_success() {
+        for (outcome, expected) in [
+            (Some("failed"), Some("本轮失败。")),
+            (Some("cancelled"), None),
+            (Some("unknown"), None),
+            (Some("future"), None),
+        ] {
+            let payload = serde_json::json!({"kind": "done", "outcome": outcome});
+            let wire = format!("nebula-hook/1 source=pi\n{payload}");
+            let event = crate::ai_hook::parse_remote_envelope(wire.as_bytes(), Some(1)).unwrap();
+            let notification = event.turn_notification(crate::display::UiLanguage::ZhCn, false);
+            assert_eq!(notification.map(|value| value.toast_text().1).as_deref(), expected);
+        }
+        for outcome in [None, Some("success")] {
+            let mut payload = serde_json::json!({"kind": "done"});
+            if let Some(outcome) = outcome {
+                payload["outcome"] = outcome.into();
+            }
+            let wire = format!("nebula-hook/1 source=pi\n{payload}");
+            let event = crate::ai_hook::parse_remote_envelope(wire.as_bytes(), Some(1)).unwrap();
+            assert!(event.turn_notification(crate::display::UiLanguage::EnUs, false).is_some());
+        }
+    }
+
     #[test]
     fn every_registered_ai_is_recognized_for_bell_text_and_command_events() {
         for agent in crate::ai_agents::AgentKind::ALL {

scripts/tests/test_pi_bridge.mjs

@@ -15,9 +15,13 @@ globalThis.__pebrelBridgeSpawn = (_hook, args) => {
   events.push(JSON.parse(args[1]));
   return { unref() {} };
 };
+globalThis.__pebrelPiVersion = '0.85.1';
 const executable = stripTypeScriptTypes(embedded.replace(
   'import { spawn } from "node:child_process";',
   'const spawn = globalThis.__pebrelBridgeSpawn;',
+).replace(
+  'import { VERSION } from "@earendil-works/pi-coding-agent";',
+  'const VERSION = globalThis.__pebrelPiVersion;',
 ) + '\nexport { sessionFor };');
 const bridge = await import('data:text/javascript;base64,' + Buffer.from(executable).toString('base64'));
 mkdirSync(new URL('tmp/', root), { recursive: true });
@@ -50,6 +54,104 @@ test('stale or corrupt file cannot replace direct native identity', () => {
   }
 });
 
+test('a retryable failed attempt must not report successful completion', async () => {
+  const previous = process.env.PEBREL_HOOK_EXE;
+  process.env.PEBREL_HOOK_EXE = 'test-hook';
+  const firstEvent = events.length;
+  try {
+    const callbacks = new Map();
+    bridge.default({ on: (kind, callback) => callbacks.set(kind, callback) });
+    const ctx = { ...context('retrying'), isIdle: () => false };
+    await callbacks.get('agent_start')({}, ctx);
+    await callbacks.get('agent_end')({ messages: [
+      { role: 'assistant', stopReason: 'error', errorMessage: 'synthetic retryable failure' },
+    ] }, ctx);
+    assert.deepEqual(events.slice(firstEvent).map(event => event.kind), ['prompt']);
+  } finally {
+    events.splice(firstEvent);
+    if (previous === undefined) delete process.env.PEBREL_HOOK_EXE;
+    else process.env.PEBREL_HOOK_EXE = previous;
+  }
+});
+
+for (const [reason, outcome] of [['stop', 'success'], ['error', 'failed'], ['aborted', 'cancelled'], ['future', 'unknown']]) {
+  test(`settled ${reason} reports ${outcome} once, without leaking error text`, async () => {
+    const previous = process.env.PEBREL_HOOK_EXE;
+    process.env.PEBREL_HOOK_EXE = 'test-hook';
+    const firstEvent = events.length;
+    try {
+      const callbacks = new Map();
+      bridge.default({ on: (kind, callback) => callbacks.set(kind, callback) });
+      const ctx = context('settled');
+      await callbacks.get('agent_start')({}, ctx);
+      await callbacks.get('agent_end')({ messages: [
+        { role: 'assistant', stopReason: reason, errorMessage: 'SECRET-provider-request' },
+      ] }, ctx);
+      assert.deepEqual(events.slice(firstEvent).map(event => event.kind), ['prompt']);
+      assert.equal(typeof callbacks.get('agent_settled'), 'function');
+      await callbacks.get('agent_settled')({}, ctx);
+      await callbacks.get('agent_settled')({}, ctx);
+      assert.deepEqual(events.slice(firstEvent).map(event => event.kind), ['prompt', 'done']);
+      assert.equal(events.at(-1).outcome, outcome);
+      assert.ok(!JSON.stringify(events.slice(firstEvent)).includes('SECRET'));
+    } finally {
+      events.splice(firstEvent);
+      if (previous === undefined) delete process.env.PEBREL_HOOK_EXE;
+      else process.env.PEBREL_HOOK_EXE = previous;
+    }
+  });
+}
+
+test('retry success replaces the failed attempt; switching sessions drops unfinished results', async () => {
+  const previous = process.env.PEBREL_HOOK_EXE;
+  process.env.PEBREL_HOOK_EXE = 'test-hook';
+  const firstEvent = events.length;
+  try {
+    const callbacks = new Map();
+    bridge.default({ on: (kind, callback) => callbacks.set(kind, callback) });
+    const ctx = context('retry-success');
+    await callbacks.get('agent_start')({}, ctx);
+    await callbacks.get('agent_end')({ messages: [{ role: 'assistant', stopReason: 'error' }] }, ctx);
+    await callbacks.get('agent_start')({}, ctx);
+    await callbacks.get('agent_end')({ messages: [{ role: 'assistant', stopReason: 'stop' }] }, ctx);
+    await callbacks.get('agent_settled')({}, ctx);
+    assert.equal(events.slice(firstEvent).filter(event => event.kind === 'done').length, 1);
+    assert.equal(events.at(-1).outcome, 'success');
+    await callbacks.get('agent_start')({}, ctx);
+    await callbacks.get('agent_end')({ messages: [{ role: 'assistant', stopReason: 'error' }] }, ctx);
+    await callbacks.get('session_start')({}, context('new-session'));
+    await callbacks.get('agent_settled')({}, context('new-session'));
+    assert.equal(events.at(-1).kind, 'session-start');
+  } finally {
+    events.splice(firstEvent);
+    if (previous === undefined) delete process.env.PEBREL_HOOK_EXE;
+    else process.env.PEBREL_HOOK_EXE = previous;
+  }
+});
+
+test('pre-settled Pi versions retain agent_end fallback without registering unsupported events', async () => {
+  const legacy = await import('data:text/javascript;base64,' + Buffer.from(executable.replace(
+    'const VERSION = globalThis.__pebrelPiVersion;', 'const VERSION = "0.80.3";',
+  )).toString('base64'));
+  const previous = process.env.PEBREL_HOOK_EXE;
+  process.env.PEBREL_HOOK_EXE = 'test-hook';
+  const firstEvent = events.length;
+  try {
+    const callbacks = new Map();
+    legacy.default({ on: (kind, callback) => callbacks.set(kind, callback) });
+    assert.equal(callbacks.has('agent_settled'), false);
+    const ctx = context('legacy');
+    await callbacks.get('agent_start')({}, ctx);
+    await callbacks.get('agent_end')({ messages: [{ role: 'assistant', stopReason: 'stop' }] }, ctx);
+    assert.deepEqual(events.slice(firstEvent).map(event => event.kind), ['prompt', 'done']);
+    assert.equal(events.at(-1).outcome, 'success');
+  } finally {
+    events.splice(firstEvent);
+    if (previous === undefined) delete process.env.PEBREL_HOOK_EXE;
+    else process.env.PEBREL_HOOK_EXE = previous;
+  }
+});
+
 test('session switch retains one process ordering stream and reports identity before turns', async () => {
   const previous = process.env.PEBREL_HOOK_EXE;
   process.env.PEBREL_HOOK_EXE = 'test-hook';
@@ -59,7 +161,8 @@ test('session switch retains one process ordering stream and reports identity be
     await callbacks.get('session_start')({}, context('first'));
     await callbacks.get('session_start')({}, context('second'));
     await callbacks.get('agent_start')({}, context('second'));
-    await callbacks.get('agent_end')({}, context('second'));
+    await callbacks.get('agent_end')({ messages: [{ role: 'assistant', stopReason: 'stop' }] }, context('second'));
+    await callbacks.get('agent_settled')({}, context('second'));
     assert.deepEqual(events.map(event => event.session_id), ['first', 'second', 'second', 'second']);
     assert.deepEqual(events.map(event => event.kind), ['session-start', 'session-start', 'prompt', 'done']);
     assert.equal(new Set(events.map(event => event.bridge_instance)).size, 1);