fix: complete update log streaming

Co-authored-by: Codex <noreply@openai.com>

backend/src/admin/system/updater/mod.rs

@@ -110,6 +110,8 @@ struct SystemUpdateEvent {
     level: UpdateLogLevel,
     step: Option<String>,
     message: String,
+    #[serde(skip_serializing_if = "is_false")]
+    terminal: bool,
     at: String,
 }
 
@@ -157,18 +159,23 @@ pub(crate) async fn update_event_stream(
     let receiver = update_event_sender().subscribe();
     let shutdown = shutdown_subscription();
     let stream = futures::stream::unfold(
-        (receiver, shutdown),
-        |(mut receiver, mut shutdown)| async move {
+        (receiver, shutdown, false),
+        |(mut receiver, mut shutdown, close_after_send)| async move {
+            if close_after_send {
+                return None;
+            }
+
             loop {
                 tokio::select! {
                     _ = shutdown.recv() => return None,
                     received = receiver.recv() => {
                         match received {
                             Ok(message) => {
+                                let close_after_send = message.terminal;
                                 let id = message.id.clone();
                                 let data = serde_json::to_string(&message).unwrap_or_else(|_| "{}".to_string());
                                 let event = Event::default().event("update").id(id).data(data);
-                                return Some((Ok(event), (receiver, shutdown)));
+                                return Some((Ok(event), (receiver, shutdown, close_after_send)));
                             }
                             Err(broadcast::error::RecvError::Lagged(_)) => continue,
                             Err(broadcast::error::RecvError::Closed) => return None,
@@ -192,7 +199,7 @@ pub(crate) async fn perform_update(
         .map_err(|_| AdminError::conflict("System update already running"))?;
     let config = SystemUpdateConfig::from_env();
     if let Some(reason) = config.update_support_error() {
-        emit_update_event(
+        emit_terminal_update_event(
             UpdateLogLevel::Error,
             None,
             Some("preflight"),
@@ -217,7 +224,7 @@ pub(crate) async fn perform_update(
     )
     .await
     .map_err(|error| {
-        emit_update_event(UpdateLogLevel::Error, None, Some("release"), error.clone());
+        emit_terminal_update_event(UpdateLogLevel::Error, None, Some("release"), error.clone());
         AdminError::bad_gateway(error)
     })?;
     let info = update_info_from_release(&config, release.clone());
@@ -226,7 +233,7 @@ pub(crate) async fn perform_update(
             "远端最新版本已变更为 v{}，请重新检查并确认",
             info.latest_version
         );
-        emit_update_event(
+        emit_terminal_update_event(
             UpdateLogLevel::Warning,
             None,
             Some("release"),
@@ -235,7 +242,7 @@ pub(crate) async fn perform_update(
         return Err(AdminError::conflict(message));
     }
     if !info.has_update {
-        emit_update_event(
+        emit_terminal_update_event(
             UpdateLogLevel::Warning,
             None,
             Some("release"),
@@ -261,14 +268,14 @@ pub(crate) async fn perform_update(
     );
     let result = perform_release_update(&config, &release, &target_version, &operation_id).await;
     if let Err(error) = &result {
-        emit_update_event(
+        emit_terminal_update_event(
             UpdateLogLevel::Error,
             Some(&operation_id),
             Some("failed"),
             error.to_string(),
         );
     } else {
-        emit_update_event(
+        emit_terminal_update_event(
             UpdateLogLevel::Success,
             Some(&operation_id),
             Some("done"),
@@ -732,6 +739,25 @@ fn emit_update_event(
     operation_id: Option<&str>,
     step: Option<&str>,
     message: impl Into<String>,
+) {
+    emit_update_event_with_terminal(level, operation_id, step, message, false);
+}
+
+fn emit_terminal_update_event(
+    level: UpdateLogLevel,
+    operation_id: Option<&str>,
+    step: Option<&str>,
+    message: impl Into<String>,
+) {
+    emit_update_event_with_terminal(level, operation_id, step, message, true);
+}
+
+fn emit_update_event_with_terminal(
+    level: UpdateLogLevel,
+    operation_id: Option<&str>,
+    step: Option<&str>,
+    message: impl Into<String>,
+    terminal: bool,
 ) {
     let now = Utc::now();
     let event = SystemUpdateEvent {
@@ -744,11 +770,16 @@ fn emit_update_event(
         level,
         step: step.map(ToString::to_string),
         message: message.into(),
+        terminal,
         at: now.to_rfc3339(),
     };
     let _ = update_event_sender().send(event);
 }
 
+fn is_false(value: &bool) -> bool {
+    !*value
+}
+
 fn format_bytes(bytes: u64) -> String {
     const KIB: u64 = 1024;
     const MIB: u64 = KIB * 1024;

backend/tests/admin/system/mod.rs

@@ -860,6 +860,70 @@ async fn update_events_should_close_on_shutdown() {
     assert!(body.is_empty());
 }
 
+#[tokio::test]
+async fn update_events_should_close_after_terminal_update_log() {
+    let _guard = SYSTEM_ENV_LOCK.lock().await;
+    let repository = "zyycn/codex-proxy-rs-events-terminal-route";
+    let github = MockServer::start().await;
+    let deploy = tempfile::tempdir().unwrap();
+    let exe_path = deploy.path().join("codex-proxy-rs");
+    let web_dist = deploy.path().join("web/dist");
+    std::fs::create_dir_all(&web_dist).unwrap();
+    std::fs::write(&exe_path, "old-binary").unwrap();
+    std::fs::write(web_dist.join("index.html"), "old-web").unwrap();
+    mount_latest_release_with_archive(&github, repository, "0.4.0", "new-binary", "new-web").await;
+    set_system_update_env(repository, &github.uri());
+    set_system_update_paths(deploy.path(), &exe_path, &web_dist);
+    let (app, _dir) = admin_system_test_app("system-events-terminal.sqlite").await;
+
+    let sse_response = app
+        .clone()
+        .oneshot(
+            Request::builder()
+                .method("GET")
+                .uri("/api/admin/system/update-events")
+                .header("cookie", "cpr_admin_session=session_1")
+                .header("x-request-id", "req_system_update_events_terminal")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    let update_app = app.clone();
+    let update_task = tokio::spawn(async move {
+        update_app
+            .oneshot(
+                Request::builder()
+                    .method("POST")
+                    .uri("/api/admin/system/update")
+                    .header("cookie", "cpr_admin_session=session_1")
+                    .header("x-request-id", "req_system_update_events_terminal_update")
+                    .header(CONTENT_TYPE, "application/json")
+                    .body(confirmed_update_body("0.4.0"))
+                    .unwrap(),
+            )
+            .await
+            .unwrap()
+    });
+
+    let body = tokio::time::timeout(
+        Duration::from_secs(2),
+        to_bytes(sse_response.into_body(), 128 * 1024),
+    )
+    .await
+    .expect("update event stream should end after terminal update log")
+    .unwrap();
+    let update_response = update_task.await.unwrap();
+    let text = String::from_utf8(body.to_vec()).unwrap();
+
+    assert!(
+        update_response.status() == StatusCode::OK
+            && text.contains("\"step\":\"done\"")
+            && text.contains("\"terminal\":true")
+            && text.contains("更新文件已替换，等待服务重启生效")
+    );
+}
+
 async fn get_update_detail(app: &axum::Router, refresh: bool, request_id: &str) -> Value {
     let response = app
         .clone()

frontend/src/composables/useSystemUpdate.ts

@@ -19,6 +19,7 @@ export interface SystemUpdateLog {
   level: SystemUpdateLogLevel
   step?: string
   message: string
+  terminal?: boolean
   at: string
 }
 
@@ -84,8 +85,12 @@ function clearUpdateLogs() {
   updateStreamError.value = ''
 }
 
-function connectUpdateEvents() {
-  if (updateEventSource.value) return
+function connectUpdateEvents(options: { force?: boolean } = {}) {
+  if (options.force) {
+    disconnectUpdateEvents()
+  } else if (updateEventSource.value) {
+    return
+  }
 
   if (typeof EventSource === 'undefined') {
     updateStreamError.value = '当前浏览器不支持实时更新日志'
@@ -97,6 +102,8 @@ function connectUpdateEvents() {
   updateEventSource.value = source
 
   source.onopen = () => {
+    if (updateEventSource.value !== source) return
+
     updateStreaming.value = true
     updateStreamError.value = ''
   }
@@ -106,6 +113,9 @@ function connectUpdateEvents() {
       const log = JSON.parse((event as MessageEvent<string>).data)
       if (isUpdateLog(log)) {
         appendUpdateLog(log)
+        if (log.terminal) {
+          disconnectUpdateEvents(source)
+        }
       }
     } catch {
       updateStreamError.value = '更新日志解析失败'
@@ -113,6 +123,8 @@ function connectUpdateEvents() {
   })
 
   source.onerror = () => {
+    if (updateEventSource.value !== source) return
+
     updateStreaming.value = false
     updateStreamError.value = '更新日志连接中断'
     if (source.readyState === EventSource.CLOSED) {
@@ -121,8 +133,10 @@ function connectUpdateEvents() {
   }
 }
 
-function disconnectUpdateEvents() {
-  updateEventSource.value?.close()
+function disconnectUpdateEvents(source: EventSource | null = updateEventSource.value) {
+  source?.close()
+  if (source && updateEventSource.value !== source) return
+
   updateEventSource.value = null
   updateStreaming.value = false
 }
@@ -184,7 +198,7 @@ async function updateNow(targetVersion: string) {
   if (!canUpdate.value || !currentInfo || updating.value || !confirmedTargetVersion) return null
 
   clearUpdateLogs()
-  connectUpdateEvents()
+  connectUpdateEvents({ force: true })
   updating.value = true
   updateError.value = ''
   updateSuccess.value = false

frontend/src/layout/components/SystemUpdateModal.vue

@@ -45,7 +45,6 @@ const {
   updateNow,
   restartNow,
   clearRestartTimer,
-  connectUpdateEvents,
   disconnectUpdateEvents,
 } = useSystemUpdate()
 
@@ -54,7 +53,6 @@ const updateConfirmOpen = shallowRef(false)
 const updateConfirmInfo = shallowRef<SystemUpdateInfo | null>(null)
 const updateConfirmPreviousTarget = shallowRef('')
 const preparingUpdate = shallowRef(false)
-let updateLogPinTimer: number | undefined
 
 const statusView = computed(() => {
   if (updating.value) {
@@ -137,7 +135,6 @@ const renderedReleaseNotes = computed(() => renderMarkdown(updateInfo.value?.not
 const showUpdateProgress = computed(
   () => hasUpdate.value || updating.value || restarting.value || updateLogRows.value.length > 0,
 )
-const shouldConnectUpdateEvents = computed(() => (open.value && hasUpdate.value) || updating.value)
 
 const updateConfirmRows = computed(() => [
   {
@@ -212,20 +209,6 @@ function pinUpdateLogsToBottom() {
   })
 }
 
-function startUpdateLogPinning() {
-  if (updateLogPinTimer !== undefined) return
-
-  pinUpdateLogsToBottom()
-  updateLogPinTimer = window.setInterval(pinUpdateLogsToBottom, 250)
-}
-
-function stopUpdateLogPinning() {
-  if (updateLogPinTimer === undefined) return
-
-  window.clearInterval(updateLogPinTimer)
-  updateLogPinTimer = undefined
-}
-
 async function handleCheckUpdates(force = true) {
   try {
     const data = await checkUpdates(force)
@@ -304,45 +287,17 @@ watch(open, (visible) => {
   }
 })
 
-watch(
-  shouldConnectUpdateEvents,
-  (shouldConnect) => {
-    if (shouldConnect) {
-      connectUpdateEvents()
-    } else {
-      disconnectUpdateEvents()
-    }
-  },
-  { immediate: true },
-)
-
 watch(
   () => updateLogs.value.at(-1)?.id,
-  () => {
+  (logId, previousLogId) => {
+    if (!logId || logId === previousLogId) return
+
     pinUpdateLogsToBottom()
   },
-)
-
-watch(
-  () => updateLogRows.value.length,
-  () => {
-    pinUpdateLogsToBottom()
-  },
-)
-
-watch(
-  () => updating.value || restarting.value || updateStreaming.value,
-  (active) => {
-    if (active) {
-      startUpdateLogPinning()
-      return
-    }
-    stopUpdateLogPinning()
-  },
+  { flush: 'post' },
 )
 
 onUnmounted(() => {
-  stopUpdateLogPinning()
   clearRestartTimer()
   disconnectUpdateEvents()
 })