fix(update): restore complete streaming progress

Restore the pre-refactor update stage events and human-readable download progress, preserve progressPercent through the admin SSE wire, and wait for the stream subscription before starting replacement.

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

backend/crates/gateway-api/src/admin/system.rs

@@ -206,6 +206,7 @@ struct SystemUpdateEventView {
     step: Option<String>,
     message: String,
     terminal: bool,
+    progress_percent: Option<u8>,
     at: String,
 }
 
@@ -408,6 +409,7 @@ impl From<SystemUpdateEvent> for SystemUpdateEventView {
             step: event.step,
             message: event.message,
             terminal: event.terminal,
+            progress_percent: event.progress_percent,
             at: event.occurred_at.to_rfc3339(),
         }
     }
@@ -421,6 +423,7 @@ impl SystemUpdateEventView {
             "step": self.step,
             "message": self.message,
             "terminal": self.terminal,
+            "progressPercent": self.progress_percent,
             "at": self.at,
         })
     }

backend/crates/gateway-api/tests/admin/mod.rs

@@ -84,6 +84,10 @@ pub(super) struct AdminTestFixture {
 
 impl AdminTestFixture {
     pub async fn new() -> Self {
+        Self::with_system(Arc::new(UnusedSystem)).await
+    }
+
+    pub async fn with_system(system: Arc<dyn SystemOperations>) -> Self {
         let api_key = Arc::new(Mutex::new(None));
         let auth = Arc::new(MemoryAuthStore::new(api_key.clone()));
         let settings = Arc::new(MemorySettingsStore::new(api_key));
@@ -113,7 +117,7 @@ impl AdminTestFixture {
             providers,
             Arc::new(NoopSnapshot),
             Arc::new(NoopProbe),
-            Arc::new(UnusedSystem),
+            system,
         )
         .await
         .expect("initialize test admin services");

backend/crates/gateway-api/tests/admin/system.rs

@@ -1,4 +1,24 @@
-use gateway_api::admin::system::{UpdateDetailQuery, UpdateRequest};
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use axum::{
+    Router,
+    body::{Body, to_bytes},
+    http::{Request, header},
+};
+use chrono::Utc;
+use futures::stream;
+use gateway_admin::{
+    model::system::{
+        SystemOperationAccepted, SystemUpdateDetail, SystemUpdateEvent, SystemUpdateEventLevel,
+        SystemUpdateStatus, SystemVersion,
+    },
+    ports::system::{SystemOperationError, SystemOperations, SystemUpdateEventStream},
+};
+use gateway_api::admin::system::{self, UpdateDetailQuery, UpdateRequest};
+use tower::ServiceExt as _;
+
+use super::{AdminTestFixture, AdminTestState, unavailable_system};
 
 #[test]
 fn update_detail_query_should_default_to_no_refresh_and_reject_unknown_fields() {
@@ -16,3 +36,80 @@ fn update_request_should_preserve_target_version_for_domain_validation() {
         serde_json::from_str::<UpdateRequest>(r#"{"targetVersion":"v0.2.0","extra":1}"#).is_err()
     );
 }
+
+#[tokio::test]
+async fn update_event_stream_should_preserve_download_progress_percent() {
+    let fixture = AdminTestFixture::with_system(Arc::new(ProgressSystem)).await;
+    fixture.auth.insert_session("valid-session");
+    let response = app(fixture.state())
+        .oneshot(
+            Request::builder()
+                .uri("/api/admin/system/update/events")
+                .header(header::COOKIE, "cpr_admin_session=valid-session")
+                .header("x-request-id", "req_system_update_events")
+                .body(Body::empty())
+                .expect("update event request"),
+        )
+        .await
+        .expect("update event response");
+    let body = to_bytes(response.into_body(), 1024 * 1024)
+        .await
+        .expect("update event body");
+    let body = String::from_utf8(body.to_vec()).expect("UTF-8 event stream");
+
+    assert!(
+        body.contains(r#""progressPercent":40"#),
+        "unexpected update event stream: {body}"
+    );
+}
+
+fn app(state: AdminTestState) -> Router {
+    system::router::<AdminTestState>().with_state(state)
+}
+
+struct ProgressSystem;
+
+#[async_trait]
+impl SystemOperations for ProgressSystem {
+    async fn version(&self) -> Result<SystemVersion, SystemOperationError> {
+        Err(unavailable_system())
+    }
+
+    async fn update_detail(&self, _: bool) -> Result<SystemUpdateDetail, SystemOperationError> {
+        Err(unavailable_system())
+    }
+
+    fn update_events(&self) -> SystemUpdateEventStream {
+        Box::pin(stream::once(async {
+            SystemUpdateEvent {
+                id: "update-progress-1".to_owned(),
+                operation_id: Some("update-1".to_owned()),
+                level: SystemUpdateEventLevel::Info,
+                step: Some("download".to_owned()),
+                message: "已下载 4.0 MiB / 10.0 MiB (40%)".to_owned(),
+                terminal: true,
+                progress_percent: Some(40),
+                occurred_at: Utc::now(),
+            }
+        }))
+    }
+
+    async fn perform_update(
+        &self,
+        _: Option<String>,
+    ) -> Result<SystemOperationAccepted, SystemOperationError> {
+        Err(unavailable_system())
+    }
+
+    async fn update_status(&self) -> Result<SystemUpdateStatus, SystemOperationError> {
+        Err(unavailable_system())
+    }
+
+    async fn rollback(&self) -> Result<SystemOperationAccepted, SystemOperationError> {
+        Err(unavailable_system())
+    }
+
+    async fn restart(&self) -> Result<SystemOperationAccepted, SystemOperationError> {
+        Err(unavailable_system())
+    }
+}

backend/crates/gateway-host/src/system_update/download.rs

@@ -49,7 +49,7 @@ pub(crate) async fn download_file(
         .map_err(|error| super::internal(format!("failed to create download file: {error}")))?;
     let mut stream = response.bytes_stream();
     let mut downloaded = 0_u64;
-    let mut last_percent = 0_u8;
+    let mut next_progress = 10_u8;
     while let Some(chunk) = stream.next().await {
         let chunk = chunk.map_err(|error| upstream(format!("download stream failed: {error}")))?;
         downloaded = downloaded.saturating_add(u64::try_from(chunk.len()).unwrap_or(u64::MAX));
@@ -59,9 +59,17 @@ pub(crate) async fn download_file(
         file.write_all(&chunk)
             .map_err(|error| super::internal(format!("failed to write download: {error}")))?;
         let percent = download_percent(downloaded, expected_size);
-        if percent > last_percent {
-            last_percent = percent;
-            events.emit_progress(operation_id, "downloading release archive", percent);
+        if percent >= next_progress || downloaded == expected_size {
+            events.emit_progress(
+                operation_id,
+                &format!(
+                    "已下载 {} / {} ({percent}%)",
+                    format_bytes(downloaded),
+                    format_bytes(expected_size)
+                ),
+                percent,
+            );
+            next_progress = percent.saturating_add(10);
         }
     }
     file.sync_all()
@@ -155,3 +163,15 @@ fn download_percent(downloaded: u64, total: u64) -> u8 {
     )
     .unwrap_or(100)
 }
+
+pub(crate) fn format_bytes(bytes: u64) -> String {
+    const KIB: u64 = 1024;
+    const MIB: u64 = KIB * 1024;
+    if bytes >= MIB {
+        return format!("{:.1} MiB", bytes as f64 / MIB as f64);
+    }
+    if bytes >= KIB {
+        return format!("{:.1} KiB", bytes as f64 / KIB as f64);
+    }
+    format!("{bytes} B")
+}

backend/crates/gateway-host/src/system_update/mod.rs

@@ -232,8 +232,15 @@ impl ProcessSystemOperations {
             .as_deref()
             .ok_or_else(|| conflict("update repository is not configured"))?;
         self.events
-            .info(None, Some("release"), "fetching latest release");
-        let release = fetch_latest(&self.config.github_api_base, repository).await?;
+            .info(None, Some("release"), "正在获取最新 Release 信息");
+        let release = match fetch_latest(&self.config.github_api_base, repository).await {
+            Ok(release) => release,
+            Err(error) => {
+                self.events
+                    .error_terminal(None, Some("release"), error.to_string());
+                return Err(error);
+            }
+        };
         let detail = detail_from_release(&self.config, &release);
         if detail.latest_version != target {
             self.events.warning_terminal(
@@ -258,12 +265,17 @@ impl ProcessSystemOperations {
             Some(&target),
             &self.config.version,
         )?;
+        self.events.info(
+            Some(&operation_id),
+            Some("prepare"),
+            format!("准备更新到 v{target}"),
+        );
         let result = self.install_release(&release, &target, &operation_id).await;
         match &result {
             Ok(()) => self.events.success_terminal(
                 Some(&operation_id),
                 Some("done"),
-                "release files replaced",
+                "更新文件已替换，等待服务重启生效",
             ),
             Err(error) => {
                 self.events
@@ -294,7 +306,23 @@ impl ProcessSystemOperations {
         version: &str,
         operation_id: &str,
     ) -> Result<(), OperationError> {
+        self.events.info(
+            Some(operation_id),
+            Some("asset"),
+            "正在选择匹配当前平台的更新包",
+        );
         let archive = select_archive(release, version)?;
+        self.events.info(
+            Some(operation_id),
+            Some("asset"),
+            format!(
+                "已选择更新包 {} ({})",
+                archive.name,
+                download::format_bytes(archive.size)
+            ),
+        );
+        self.events
+            .info(Some(operation_id), Some("verify"), "正在校验更新资源");
         if archive.size == 0 || archive.size > MAX_DOWNLOAD_SIZE {
             return Err(invalid("release archive size is invalid"));
         }
@@ -306,12 +334,16 @@ impl ProcessSystemOperations {
         if checksum.size == 0 || checksum.size > MAX_CHECKSUM_SIZE {
             return Err(invalid("release checksum size is invalid"));
         }
+        self.events
+            .info(Some(operation_id), Some("prepare"), "正在创建临时更新目录");
         fs::create_dir_all(&self.config.update_temp_dir)
             .map_err(|error| internal(format!("failed to prepare update temp dir: {error}")))?;
         let temp_root = fs::canonicalize(&self.config.update_temp_dir)
             .map_err(|error| internal(format!("failed to resolve update temp dir: {error}")))?;
         let temp = UpdateTempDir::create(&temp_root)?;
         let archive_path = temp.path().join(&archive.name);
+        self.events
+            .info(Some(operation_id), Some("download"), "开始下载更新包");
         download_file(
             &archive.browser_download_url,
             &archive_path,
@@ -321,6 +353,10 @@ impl ProcessSystemOperations {
             &self.events,
         )
         .await?;
+        self.events
+            .success(Some(operation_id), Some("download"), "更新包下载完成");
+        self.events
+            .info(Some(operation_id), Some("checksum"), "正在校验 checksum");
         verify_checksum(
             &archive_path,
             &archive.name,
@@ -329,12 +365,23 @@ impl ProcessSystemOperations {
             &self.config.github_api_base,
         )
         .await?;
+        self.events
+            .success(Some(operation_id), Some("checksum"), "checksum 校验通过");
+        self.events
+            .info(Some(operation_id), Some("extract"), "正在解压更新包");
         let extracted = extract_release(&archive_path, temp.path())?;
+        self.events
+            .success(Some(operation_id), Some("extract"), "更新包解压完成");
+        self.events
+            .info(Some(operation_id), Some("replace"), "正在替换应用文件");
         replace_release_files(
             &self.config.executable_path()?,
             &self.config.web_dist_dir,
             extracted,
-        )
+        )?;
+        self.events
+            .success(Some(operation_id), Some("replace"), "应用文件替换完成");
+        Ok(())
     }
 }
 
@@ -515,6 +562,17 @@ impl UpdateEvents {
         );
     }
 
+    fn success(&self, operation_id: Option<&str>, step: Option<&str>, message: impl Into<String>) {
+        self.emit(
+            SystemUpdateEventLevel::Success,
+            operation_id,
+            step,
+            message,
+            None,
+            false,
+        );
+    }
+
     fn warning_terminal(
         &self,
         operation_id: Option<&str>,

backend/crates/gateway-host/tests/system_update/mod.rs

@@ -204,6 +204,68 @@ async fn update_events_should_open_authenticated_sse_stream() {
     assert!(stream.next().await.is_some());
 }
 
+#[tokio::test]
+async fn update_events_should_preserve_complete_release_stage_sequence() {
+    let server = MockServer::start().await;
+    let fixture = Fixture::new();
+    fixture
+        .mount_release(
+            &server,
+            TARGET_VERSION,
+            ArchiveKind::Safe,
+            ChecksumKind::Valid,
+        )
+        .await;
+    let service = fixture.service(&server);
+    let events = service.update_events();
+    service
+        .perform_update(Some(TARGET_VERSION.to_owned()))
+        .await
+        .expect("update");
+    let steps = events
+        .map(|event| event.step.expect("release update step"))
+        .collect::<Vec<_>>()
+        .await;
+
+    assert_eq!(
+        steps,
+        [
+            "release", "prepare", "asset", "asset", "verify", "prepare", "download", "download",
+            "download", "checksum", "checksum", "extract", "extract", "replace", "replace", "done",
+        ]
+    );
+}
+
+#[tokio::test]
+async fn update_events_should_report_download_bytes_and_percent() {
+    let server = MockServer::start().await;
+    let fixture = Fixture::new();
+    fixture
+        .mount_release(
+            &server,
+            TARGET_VERSION,
+            ArchiveKind::Safe,
+            ChecksumKind::Valid,
+        )
+        .await;
+    let service = fixture.service(&server);
+    let events = service.update_events();
+    service
+        .perform_update(Some(TARGET_VERSION.to_owned()))
+        .await
+        .expect("update");
+    let progress = events
+        .filter_map(
+            |event| async move { event.progress_percent.map(|value| (value, event.message)) },
+        )
+        .collect::<Vec<_>>()
+        .await;
+
+    assert!(progress.iter().any(|(percent, message)| {
+        *percent == 100 && message.starts_with("已下载 ") && message.ends_with(" (100%)")
+    }));
+}
+
 #[tokio::test]
 async fn update_should_fail_when_release_checksum_is_missing() {
     let server = MockServer::start().await;

frontend/src/stores/modules/system-update.ts

@@ -1,4 +1,4 @@
-import { useEventSource } from '@vueuse/core'
+import { until, useEventSource } from '@vueuse/core'
 import { delay } from 'es-toolkit'
 import { defineStore } from 'pinia'
 
@@ -14,6 +14,7 @@ import { ApiError } from '@/api/request'
 import { errorMessage } from '@/utils/async'
 
 const maxUpdateLogs = 200
+const updateEventReadyTimeoutMs = 3_000
 const restartReadyTimeoutMs = 60_000
 const restartProbeTimeoutMs = 2_000
 const restartReadyPollIntervalMs = 500
@@ -106,19 +107,30 @@ export const useSystemUpdateStore = defineStore('system-update', () => {
     }
   })
 
-  function connectUpdateEvents(force = false) {
+  async function connectUpdateEvents(force = false) {
     if (force) {
       disconnectUpdateEvents()
     }
     else if (updateEventSource.value) {
-      return
+      return updateEventStatus.value === 'OPEN'
     }
 
     updateStreamError.value = ''
     openUpdateEventSource()
     if (!updateEventSource.value) {
       updateStreamError.value = '当前浏览器不支持实时更新日志'
+      return false
     }
+
+    if (updateEventStatus.value !== 'OPEN') {
+      await until(updateEventStatus).toBe('OPEN', {
+        timeout: updateEventReadyTimeoutMs,
+      })
+    }
+    const connected = updateEventStatus.value === 'OPEN'
+    if (!connected)
+      updateStreamError.value = '更新日志连接超时'
+    return connected
   }
 
   function disconnectUpdateEvents() {
@@ -186,7 +198,7 @@ export const useSystemUpdateStore = defineStore('system-update', () => {
       return null
 
     clearUpdateLogs()
-    connectUpdateEvents(true)
+    await connectUpdateEvents(true)
     updating.value = true
     updateError.value = ''
     updateSuccess.value = false