Merge pull request #260 from Kuddev/fix/feedback-hooks-20260923

fix(windows): invoke Codex hooks with PowerShell and bound helper lifetime

architecture/notes/nebula_hook/2026-09-23-bounded-forwarding-lifetime.md

@@ -0,0 +1,63 @@
+# Bound the lifetime of one-shot Hook forwarding
+
+## Status
+
+Implemented; pending review.
+
+## Context
+
+[Issue #258](https://github.com/Kuddev/pebrel/issues/258) reports an update that
+first failed because `pebrel-hook.exe` remained in use after the application
+closed, followed by a successful same-version repair incorrectly marked failed.
+The helper must finish even when its caller or notification receiver stalls.
+
+## Evidence
+
+`nebula_hook/src/main.rs` drained stdin through `read_to_end` and wrote the
+notification through synchronous `write_all`. Neither had a deadline. A caller
+retaining stdin can prevent EOF; a receiver that stops reading can block a write.
+Both conditions are reproduced by actual child processes in
+[`lifetime.rs`](../../../nebula_hook/tests/lifetime.rs).
+
+## Decision
+
+Run the existing forwarding operation on one process-owned worker. The main
+thread waits up to two seconds, then returns normally. Returning from this
+one-shot process releases its worker and OS handles without leaving a resident
+thread. This bounds stdin, oversize-input draining and notification I/O together.
+No new dependency, background application service or platform thread is added.
+
+The user's chained notifier is launched independently after forwarding/waiting;
+it is neither joined nor terminated. Cursor's submission-allow response remains
+on the main path, including timeout. All provider-visible exits remain successful.
+
+## Rejected alternatives
+
+- Timing only stdin leaves a blocked pipe write alive.
+- Joining the worker after timeout defeats the bound.
+- Terminating every process named `pebrel-hook.exe` can affect another installation
+  or invocation. This change retires only the helper's own process.
+- Skipping stdin entirely breaks providers which stream their payload into it.
+
+## Consequences
+
+A stalled or unusually slow invocation can lose its best-effort notification.
+The forwarding wait is bounded; this is not a universal process startup latency
+guarantee. Already running binaries from an older installation do not acquire
+the new deadline and may still require the user to close the owning CLI/restart.
+
+## Validation
+
+`cargo test --locked -p nebula_hook`: seven unit tests and five actual-process
+tests passed on Windows, covering retained stdin, a connected non-reading named
+pipe, normal completion, Cursor continuation and the original chained notifier.
+Non-Windows integration paths still require their native CI run.
+
+## Supersedes
+
+None.
+
+## Revisit when
+
+Providers require a longer documented delivery window or the transport becomes
+cancellable without a per-invocation worker.

nebula_app/src/ai_hook/win/codex_hooks.rs

@@ -111,8 +111,10 @@ pub(super) fn ensure_codex_hooks() -> bool {
 }
 
 fn groups(helper: &str, mode: CodexHookMode) -> Value {
-    // Codex 的 Windows runner 使用 COMSPEC /C；可执行文件路径自身仍需引号。
-    let command = format!("\"{}\" codex {}", helper.replace('"', "\\\""), mode.argument());
+    // Codex runs Windows hooks through PowerShell -Command. A quoted path is
+    // a string expression until invoked with &, and single quotes keep $, `
+    // and other path characters literal. PowerShell escapes ' by doubling it.
+    let command = format!("& '{}' codex {}", helper.replace('\'', "''"), mode.argument());
     installation::codex_groups(&command, Some(&command), mode)
 }
 
@@ -260,6 +262,78 @@ fn remove(directory: &Path) -> io::Result<bool> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn upgrades_cmd_style_owned_hooks_without_changing_user_entries() {
+        let root = tempfile::tempdir().unwrap();
+        let helper = "C:/Program Files/Pebrel/runtime/pebrel-hook.exe";
+        let old_command = format!("\"{helper}\" codex --hooks=full");
+        let previous =
+            installation::codex_groups(&old_command, Some(&old_command), CodexHookMode::Full);
+        let foreign = json!({"hooks":[{"type":"command","command":"user-hook"}]});
+        let mut hooks = json!({"hooks": previous});
+        hooks["hooks"]["Stop"].as_array_mut().unwrap().push(foreign.clone());
+        std::fs::write(root.path().join("hooks.json"), hooks.to_string()).unwrap();
+        std::fs::write(
+            root.path().join(MARKER),
+            json!({"groups": previous, "enabled_feature": true}).to_string(),
+        )
+        .unwrap();
+        let config = "notify = ['user-notifier']\n[features]\nhooks = true\n";
+        std::fs::write(root.path().join("config.toml"), config).unwrap();
+
+        assert!(!current_for_mode(root.path(), helper, Some(CodexHookMode::Full)).unwrap());
+        assert!(install(root.path(), helper, CodexHookMode::Full).unwrap());
+        assert!(current_for_mode(root.path(), helper, Some(CodexHookMode::Full)).unwrap());
+        assert!(!install(root.path(), helper, CodexHookMode::Full).unwrap());
+        let hooks: Value =
+            serde_json::from_str(&read(&root.path().join("hooks.json")).unwrap().unwrap()).unwrap();
+        let mut expected = json!({"hooks": groups(helper, CodexHookMode::Full)});
+        expected["hooks"]["Stop"].as_array_mut().unwrap().insert(0, foreign);
+        assert_eq!(hooks, expected);
+        assert_eq!(read(&root.path().join("config.toml")).unwrap().unwrap(), config);
+    }
+
+    #[test]
+    fn powershell_executes_generated_hooks_with_literal_paths_and_stdin() {
+        use std::io::Write as _;
+        use std::os::windows::process::CommandExt as _;
+
+        let root = tempfile::tempdir().unwrap();
+        let helper = root.path().join("Pebrel's $data `hook.ps1");
+        std::fs::write(
+            &helper,
+            "Write-Output ($args -join '|')\r\nWrite-Output ([Console]::In.ReadLine())\r\n",
+        )
+        .unwrap();
+        let powershell = Path::new(&std::env::var_os("SystemRoot").unwrap())
+            .join("System32/WindowsPowerShell/v1.0/powershell.exe");
+        for mode in [CodexHookMode::Turns, CodexHookMode::Full] {
+            let groups = groups(&helper.to_string_lossy().replace('\\', "/"), mode);
+            let command = groups["Stop"][0]["hooks"][0]["command"].as_str().unwrap();
+            let mut child = Command::new(&powershell)
+                .args([
+                    "-NoProfile",
+                    "-NonInteractive",
+                    "-ExecutionPolicy",
+                    "Bypass",
+                    "-Command",
+                    command,
+                ])
+                .creation_flags(0x0800_0000)
+                .stdin(Stdio::piped())
+                .stdout(Stdio::piped())
+                .stderr(Stdio::piped())
+                .spawn()
+                .unwrap();
+            child.stdin.take().unwrap().write_all(b"{\"session_id\":\"fixture\"}\n").unwrap();
+            let output = child.wait_with_output().unwrap();
+            assert!(output.status.success(), "{}", String::from_utf8_lossy(&output.stderr));
+            let stdout = String::from_utf8(output.stdout).unwrap();
+            assert!(stdout.contains(&format!("codex|{}", mode.argument())), "{stdout}");
+            assert!(stdout.contains(r#"{"session_id":"fixture"}"#), "{stdout}");
+        }
+    }
+
     #[test]
     fn install_upgrade_remove_preserves_notify_and_user_hooks() {
         let root = tempfile::tempdir().unwrap();

nebula_hook/src/main.rs

@@ -12,14 +12,15 @@
 //!    surfaces an error banner, 2 even blocks the turn), and kimi's `Stop` is
 //!    likewise a blockable event. Every path — including panic — must exit 0,
 //!    fast. Claude and kimi also write the payload to our stdin, so those modes
-//!    always drain stdin even when the message goes nowhere: an unread pipe
-//!    could surface as a hook write error.
+//!    drain stdin within a bounded invocation even when the message goes nowhere.
+//!    A caller retaining stdin or a stalled receiver must not leave this helper
+//!    running indefinitely and holding the installed executable open.
 //! 2. SCOPED: the hook config is global (settings.json / kimi's config.toml),
 //!    but the effect must be Nebula-only. The scope guard is the environment:
 //!    NEBULA_NOTIFY_PIPE only exists for processes spawned inside Nebula.
-//!    Anywhere else this is an invisible ~10 ms no-op.
-//! 3. FAST: pure std, no JSON handling (Nebula parses), one pipe write.
-//!    Keeps the whole claude→toast chain under ~50 ms.
+//!    Anywhere else it forwards nothing and exits without affecting the caller.
+//! 3. BOUNDED: pure std, no JSON handling (Pebrel parses), one pipe write.
+//!    Forwarding has a deadline; startup and notification latency depend on the host.
 //!
 //! Usage (installed by `nebula setup-ai` / Nebula's boot self-heal):
 //! ```text
@@ -42,6 +43,7 @@
 use std::io::{Read, Write};
 
 const MAX_PAYLOAD_BYTES: usize = 1 << 20;
+const FORWARD_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2);
 
 /// 其他 agent 的 hook runner 独有的环境变量。
 ///
@@ -167,7 +169,25 @@ fn log_outcome(source: &str, pane: &str, bytes: usize, outcome: &Outcome) {
 fn main() {
     // Constraint 1: never leak a failure to the calling CLI.
     let args: Vec<String> = std::env::args().skip(1).collect();
-    let _ = std::panic::catch_unwind(|| run(&args));
+    let forwarding_args = args.clone();
+    let (done, finished) = std::sync::mpsc::sync_channel(1);
+    // Bound the entire forwarding operation, including stdin drain and pipe
+    // writes. A read-only timeout would still leave a blocked writer alive.
+    // This one-shot process never joins a stuck worker: returning from main
+    // retires all of its threads and handles. The provider sees exit code 0.
+    if std::thread::Builder::new()
+        .name("hook-forward".into())
+        .spawn(move || {
+            let _ = std::panic::catch_unwind(|| run(&forwarding_args));
+            let _ = done.send(());
+        })
+        .is_ok()
+    {
+        let _ = finished.recv_timeout(FORWARD_TIMEOUT);
+    }
+    // A user-owned notifier is independent of our best-effort transport. Never
+    // wait for it and do not suppress it when the Pebrel pipe is unavailable.
+    chain_notifier(&args);
     // Cursor 的提交前 Hook 有响应合同；传输失败也不能阻止用户提交。
     if args.first().is_some_and(|source| source == "cursor")
         && native_event(&args) == Some("prompt")
@@ -307,7 +327,9 @@ fn run(args: &[String]) {
         }
     }
     log_outcome(source, &pane, payload.len(), &outcome);
+}
 
+fn chain_notifier(args: &[String]) {
     // Chain mode: keep a pre-existing codex notifier working. Runs even
     // outside Nebula — the original program must keep firing everywhere.
     let strs: Vec<&str> = args.iter().map(String::as_str).collect();

nebula_hook/tests/lifetime.rs

@@ -0,0 +1,142 @@
+//! Real-process lifetime regressions: a provider may retain its stdin handle,
+//! and a pipe receiver may stop draining. Neither can pin the installed helper.
+
+use std::io::Write as _;
+use std::process::{Child, Command, Output, Stdio};
+use std::time::{Duration, Instant};
+
+struct Invocation(Option<Child>);
+
+impl Invocation {
+    fn spawn(args: &[&str], pipe: Option<&str>) -> Self {
+        let mut command = Command::new(env!("CARGO_BIN_EXE_pebrel-hook"));
+        command.args(args).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::piped());
+        for suffix in ["NOTIFY_PIPE", "REMOTE_HOOK_TOKEN", "HOOK_LOG"] {
+            command.env_remove(format!("PEBREL_{suffix}"));
+            command.env_remove(format!("NEBULA_{suffix}"));
+        }
+        if let Some(pipe) = pipe {
+            command.env("PEBREL_NOTIFY_PIPE", pipe);
+        }
+        Self(Some(command.spawn().unwrap()))
+    }
+
+    fn finish(mut self) -> Output {
+        let deadline = Instant::now() + Duration::from_secs(6);
+        let child = self.0.as_mut().unwrap();
+        loop {
+            if child.try_wait().unwrap().is_some() {
+                let output = self.0.take().unwrap().wait_with_output().unwrap();
+                assert!(output.status.success(), "{}", String::from_utf8_lossy(&output.stderr));
+                return output;
+            }
+            assert!(Instant::now() < deadline, "helper retained a blocked invocation");
+            std::thread::sleep(Duration::from_millis(10));
+        }
+    }
+}
+
+impl Drop for Invocation {
+    fn drop(&mut self) {
+        if let Some(child) = &mut self.0 {
+            let _ = child.kill();
+            let _ = child.wait();
+        }
+    }
+}
+
+#[test]
+fn stdin_kept_open_cannot_pin_a_native_hook_executable() {
+    for args in [&["codex", "--hooks=full"][..], &["claude"][..]] {
+        let mut invocation = Invocation::spawn(args, None);
+        let mut input = invocation.0.as_mut().unwrap().stdin.take().unwrap();
+        input.write_all(b"{\"hook_event_name\":\"Stop\"}").unwrap();
+        // Hold stdin open across finish: EOF must not be necessary for exit.
+        let output = invocation.finish();
+        assert!(output.stdout.is_empty());
+        drop(input);
+    }
+}
+
+#[test]
+fn cursor_still_allows_submission_when_stdin_never_closes() {
+    let mut invocation = Invocation::spawn(&["cursor", "--event", "prompt"], None);
+    let input = invocation.0.as_mut().unwrap().stdin.take().unwrap();
+    let output = invocation.finish();
+    assert_eq!(output.stdout, b"{\"continue\":true}\n");
+    drop(input);
+}
+
+#[test]
+fn ordinary_completed_payload_exits_silently() {
+    let mut invocation = Invocation::spawn(&["codex", "--hooks=full"], None);
+    invocation.0.as_mut().unwrap().stdin.take().unwrap().write_all(b"{}").unwrap();
+    let output = invocation.finish();
+    assert!(output.stdout.is_empty() && output.stderr.is_empty());
+}
+
+#[test]
+fn legacy_notify_still_invokes_the_users_chained_program() {
+    let executable = std::env::current_exe().unwrap();
+    let invocation = Invocation::spawn(
+        &["codex", "--chain", executable.to_str().unwrap(), "--list", "--format=terse"],
+        None,
+    );
+    let output = invocation.finish();
+    assert!(
+        String::from_utf8(output.stdout)
+            .unwrap()
+            .contains("ordinary_completed_payload_exits_silently")
+    );
+}
+
+#[cfg(windows)]
+#[test]
+fn a_receiver_that_never_reads_cannot_pin_the_helper() {
+    use std::ffi::c_void;
+
+    #[link(name = "kernel32")]
+    unsafe extern "system" {
+        fn CreateNamedPipeW(
+            name: *const u16,
+            mode: u32,
+            pipe_mode: u32,
+            instances: u32,
+            output: u32,
+            input: u32,
+            timeout: u32,
+            security: *const c_void,
+        ) -> *mut c_void;
+        fn CloseHandle(handle: *mut c_void) -> i32;
+        fn GetNamedPipeClientProcessId(handle: *mut c_void, pid: *mut u32) -> i32;
+    }
+    struct Pipe(*mut c_void);
+    impl Drop for Pipe {
+        fn drop(&mut self) {
+            unsafe {
+                CloseHandle(self.0);
+            }
+        }
+    }
+    let name = format!(r"\\.\pipe\pebrel-hook-lifetime-test-{}", std::process::id());
+    let wide: Vec<u16> = name.encode_utf16().chain(Some(0)).collect();
+    // The server owns an inbound byte pipe and deliberately never drains it.
+    let handle =
+        unsafe { CreateNamedPipeW(wide.as_ptr(), 1, 0, 1, 4096, 4096, 0, std::ptr::null()) };
+    assert_ne!(handle as isize, -1, "{}", std::io::Error::last_os_error());
+    let _pipe = Pipe(handle);
+    let payload = "x".repeat(16 * 1024);
+    let invocation = Invocation::spawn(&["codex", &payload], Some(&name));
+    let deadline = Instant::now() + Duration::from_secs(6);
+    loop {
+        let mut pid = 0;
+        if unsafe { GetNamedPipeClientProcessId(handle, &mut pid) } != 0 {
+            assert_eq!(pid, invocation.0.as_ref().unwrap().id());
+            break;
+        }
+        assert!(Instant::now() < deadline, "helper never connected to the blocked receiver");
+        std::thread::sleep(Duration::from_millis(10));
+    }
+    let output = invocation.finish();
+    assert!(output.stdout.is_empty() && output.stderr.is_empty());
+}