[pull] main from xintaofei:main

src-tauri/src/acp/opencode_plugins.rs

@@ -3,7 +3,6 @@ use std::fs;
 use std::path::{Path, PathBuf};
 
 use serde::Serialize;
-use tokio::io::AsyncBufReadExt;
 
 use crate::web::event_bridge::{emit_event, EventEmitter};
 
@@ -507,16 +506,21 @@ pub async fn install_missing_plugins(
     let emitter_clone = emitter.clone();
     let task_id_clone = task_id.clone();
 
+    // `collect_lines_lossy` (not `Lines`/`next_line()`) matters here: bun/npm
+    // can emit OEM-codepage bytes (e.g. GBK on a zh-CN Windows) for localized
+    // OS-level error text, which `next_line()` chokes on and silently drops —
+    // truncating the live install log at the first non-UTF-8 byte. The
+    // collected return value is unused (the failure message below is built from
+    // the exit code, not captured stderr), so it is discarded.
     let stdout_handle = tokio::spawn({
         let emitter = emitter_clone.clone();
         let task_id = task_id_clone.clone();
         async move {
             if let Some(stdout) = stdout {
-                let reader = tokio::io::BufReader::new(stdout);
-                let mut lines = reader.lines();
-                while let Ok(Some(line)) = lines.next_line().await {
-                    emit_plugin_event(&emitter, &task_id, PluginInstallEventKind::Log, &line);
-                }
+                crate::process::collect_lines_lossy(tokio::io::BufReader::new(stdout), |line| {
+                    emit_plugin_event(&emitter, &task_id, PluginInstallEventKind::Log, line);
+                })
+                .await;
             }
         }
     });
@@ -526,11 +530,10 @@ pub async fn install_missing_plugins(
         let task_id = task_id_clone;
         async move {
             if let Some(stderr) = stderr {
-                let reader = tokio::io::BufReader::new(stderr);
-                let mut lines = reader.lines();
-                while let Ok(Some(line)) = lines.next_line().await {
-                    emit_plugin_event(&emitter, &task_id, PluginInstallEventKind::Log, &line);
-                }
+                crate::process::collect_lines_lossy(tokio::io::BufReader::new(stderr), |line| {
+                    emit_plugin_event(&emitter, &task_id, PluginInstallEventKind::Log, line);
+                })
+                .await;
             }
         }
     });

src-tauri/src/commands/acp.rs

@@ -677,7 +677,7 @@ async fn run_npm_streaming(
     task_id: &str,
     emitter: &EventEmitter,
 ) -> Result<(bool, String), AcpError> {
-    use tokio::io::{AsyncBufReadExt, BufReader};
+    use tokio::io::BufReader;
 
     let mut cmd = crate::process::tokio_command("npm");
     for arg in args {
@@ -695,16 +695,20 @@ async fn run_npm_streaming(
     let emitter_clone = emitter.clone();
     let task_id_owned = task_id.to_string();
 
+    // `collect_lines_lossy` (not `Lines`/`next_line()`) matters here: npm can
+    // emit OEM-codepage bytes (e.g. GBK on a zh-CN Windows) for localized
+    // OS-level error text, which `next_line()` chokes on and silently drops —
+    // truncating both the live install log and the stderr this function
+    // returns for the caller's error message.
     let stdout_handle = tokio::spawn({
         let emitter = emitter_clone.clone();
         let task_id = task_id_owned.clone();
         async move {
             if let Some(out) = stdout {
-                let reader = BufReader::new(out);
-                let mut lines = reader.lines();
-                while let Ok(Some(line)) = lines.next_line().await {
-                    emit_agent_install_event(&emitter, &task_id, AgentInstallEventKind::Log, &line);
-                }
+                crate::process::collect_lines_lossy(BufReader::new(out), |line| {
+                    emit_agent_install_event(&emitter, &task_id, AgentInstallEventKind::Log, line);
+                })
+                .await;
             }
         }
     });
@@ -713,19 +717,20 @@ async fn run_npm_streaming(
         let emitter = emitter_clone;
         let task_id = task_id_owned;
         async move {
-            let mut collected = String::new();
-            if let Some(err) = stderr {
-                let reader = BufReader::new(err);
-                let mut lines = reader.lines();
-                while let Ok(Some(line)) = lines.next_line().await {
-                    emit_agent_install_event(&emitter, &task_id, AgentInstallEventKind::Log, &line);
-                    if !collected.is_empty() {
-                        collected.push('\n');
-                    }
-                    collected.push_str(&line);
+            match stderr {
+                Some(err) => {
+                    crate::process::collect_lines_lossy(BufReader::new(err), |line| {
+                        emit_agent_install_event(
+                            &emitter,
+                            &task_id,
+                            AgentInstallEventKind::Log,
+                            line,
+                        );
+                    })
+                    .await
                 }
+                None => String::new(),
             }
-            collected
         }
     });
 
@@ -8168,12 +8173,32 @@ pub async fn acp_install_uv_tool(
 pub(crate) async fn acp_detect_agent_local_version_core(
     agent_type: AgentType,
     conn: &sea_orm::DatabaseConnection,
+    emitter: &EventEmitter,
 ) -> Result<Option<String>, AcpError> {
+    // Snapshot the stored version before probing so we can tell whether this
+    // detection actually moves it. The settings page re-runs this for every
+    // agent on each open; emitting unconditionally would fan a reload storm out
+    // to every `useAcpAgents()` consumer, so we only notify on a real change.
+    let previous = agent_setting_service::get_by_agent_type(conn, agent_type)
+        .await
+        .ok()
+        .flatten()
+        .and_then(|m| m.installed_version);
+
     let detected = detect_local_version(agent_type).await;
     if let Some(version) = detected.clone() {
         let _ =
             agent_setting_service::set_installed_version(conn, agent_type, Some(version.clone()))
                 .await;
+        // Heal the composer's install status. The input box reads
+        // `installed_version` straight from this row and shows "not installed"
+        // while it's null (`acp_list_agents_core` never probes npm for it). When
+        // a live probe discovers a version the DB never recorded — an agent
+        // installed outside codeg, or by a build predating version tracking —
+        // wake `useAcpAgents()` so the composer stops claiming it's missing.
+        if previous.as_deref() != Some(version.as_str()) {
+            emit_acp_agents_updated(emitter, "local_version_detected", Some(agent_type));
+        }
         return Ok(Some(version));
     }
 
@@ -8190,24 +8215,26 @@ pub(crate) async fn acp_detect_agent_local_version_core(
         registry::AgentDistribution::Binary { .. }
     ) {
         let _ = agent_setting_service::set_installed_version(conn, agent_type, None).await;
+        // Mirror the heal in the clearing direction: a binary that vanished from
+        // disk must flip the composer back to "not installed".
+        if previous.is_some() {
+            emit_acp_agents_updated(emitter, "local_version_cleared", Some(agent_type));
+        }
         return Ok(None);
     }
 
-    let fallback = agent_setting_service::get_by_agent_type(conn, agent_type)
-        .await
-        .ok()
-        .flatten()
-        .and_then(|m| m.installed_version);
-    Ok(fallback)
+    Ok(previous)
 }
 
 #[cfg(feature = "tauri-runtime")]
 #[cfg_attr(feature = "tauri-runtime", tauri::command)]
 pub async fn acp_detect_agent_local_version(
     agent_type: AgentType,
     db: State<'_, AppDatabase>,
+    app: tauri::AppHandle,
 ) -> Result<Option<String>, AcpError> {
-    acp_detect_agent_local_version_core(agent_type, &db.conn).await
+    let emitter = EventEmitter::Tauri(app);
+    acp_detect_agent_local_version_core(agent_type, &db.conn, &emitter).await
 }
 
 pub(crate) async fn acp_prepare_npx_agent_core(

src-tauri/src/commands/office_tools.rs

@@ -29,7 +29,7 @@ use crate::commands::experts::{
 use crate::app_error::AppCommandError;
 use crate::commands::folders::resolve_tree_path;
 use crate::models::agent::AgentType;
-use crate::process::tokio_command;
+use crate::process::{collect_lines_lossy, tokio_command};
 use crate::web::event_bridge::EventEmitter;
 
 // ─── Error type ─────────────────────────────────────────────────────────
@@ -680,58 +680,6 @@ fn bounded_tail(s: &str, max: usize) -> String {
     format!("…{}", &s[start..])
 }
 
-/// Read `reader` line-by-line as UTF-8-*lossy* text, invoking `on_line` for each
-/// line (trailing newline trimmed) and returning the accumulated text.
-///
-/// Unlike a `Lines`/`next_line()` loop — which returns `Err(InvalidData)` and so
-/// aborts the whole stream on the first non-UTF-8 byte — this preserves a
-/// non-UTF-8 line lossily. PowerShell emits OEM-codepage bytes (e.g. GBK on a
-/// zh-CN Windows) for non-ASCII installer/error text, so without this a single
-/// localized line would truncate both the live log and the failure-diagnostic
-/// tail. A genuine read error records a short note and stops — `break`, never
-/// `continue`, so a persistent error can't spin.
-async fn collect_lines_lossy<R, F>(mut reader: R, mut on_line: F) -> String
-where
-    R: tokio::io::AsyncBufRead + Unpin,
-    F: FnMut(&str),
-{
-    use tokio::io::AsyncBufReadExt;
-
-    let mut buf = Vec::new();
-    let mut collected = String::new();
-    loop {
-        buf.clear();
-        match reader.read_until(b'\n', &mut buf).await {
-            Ok(0) => break, // EOF
-            Ok(_) => {
-                // Match `Lines` semantics: strip a trailing '\n' then one '\r'.
-                if buf.last() == Some(&b'\n') {
-                    buf.pop();
-                    if buf.last() == Some(&b'\r') {
-                        buf.pop();
-                    }
-                }
-                let line = String::from_utf8_lossy(&buf);
-                on_line(line.as_ref());
-                if !collected.is_empty() {
-                    collected.push('\n');
-                }
-                collected.push_str(line.as_ref());
-            }
-            Err(e) => {
-                let note = format!("<install reader error: {e}>");
-                on_line(&note);
-                if !collected.is_empty() {
-                    collected.push('\n');
-                }
-                collected.push_str(&note);
-                break;
-            }
-        }
-    }
-    collected
-}
-
 /// Stream `child`'s stdout+stderr line-by-line as OfficeCLI install Log events,
 /// bounded by `timeout`. Returns the exit status (`None` on timeout) plus the
 /// collected stdout/stderr tails for failure diagnostics.
@@ -1768,52 +1716,4 @@ mod tests {
         assert!(tail.chars().skip(1).all(|c| c == 'あ'));
     }
 
-    #[tokio::test]
-    async fn collect_lines_lossy_preserves_lines_around_invalid_utf8() {
-        use std::io::Cursor;
-        // A non-UTF-8 segment (0xFF 0xFE — invalid start bytes, like GBK output
-        // on a non-English Windows) sits between two valid lines. The old
-        // `next_line()` loop would abort here and drop "third"; this must not.
-        let data = b"first\n\xff\xfe garbage\nthird\n".to_vec();
-        let mut seen: Vec<String> = Vec::new();
-        let collected =
-            collect_lines_lossy(Cursor::new(data), |l| seen.push(l.to_string())).await;
-
-        assert_eq!(seen.len(), 3, "all three lines emitted: {seen:?}");
-        assert_eq!(seen[0], "first");
-        assert_eq!(seen[2], "third");
-        assert!(
-            seen[1].contains('\u{fffd}'),
-            "invalid bytes preserved lossily, not dropped: {:?}",
-            seen[1]
-        );
-        assert!(collected.contains("first") && collected.contains("third"));
-        assert!(collected.contains('\u{fffd}'));
-    }
-
-    #[tokio::test]
-    async fn collect_lines_lossy_handles_crlf_and_partial_last_line() {
-        use std::io::Cursor;
-        // CRLF endings trimmed like `Lines`; a final line with no trailing
-        // newline is still emitted (then EOF stops the loop).
-        let data = b"a\r\nb\r\nno-newline".to_vec();
-        let mut seen: Vec<String> = Vec::new();
-        let collected =
-            collect_lines_lossy(Cursor::new(data), |l| seen.push(l.to_string())).await;
-
-        assert_eq!(seen, vec!["a", "b", "no-newline"]);
-        assert_eq!(collected, "a\nb\nno-newline");
-    }
-
-    #[tokio::test]
-    async fn collect_lines_lossy_empty_input_yields_nothing() {
-        use std::io::Cursor;
-        let mut seen: Vec<String> = Vec::new();
-        let collected =
-            collect_lines_lossy(Cursor::new(Vec::<u8>::new()), |l| seen.push(l.to_string()))
-                .await;
-
-        assert!(seen.is_empty());
-        assert!(collected.is_empty());
-    }
 }