feat(acp): integrate runtime events and remote control

docs/architecture/product-architecture.md

@@ -31,6 +31,9 @@ Cargo feature、第三方依赖 owner、测试目标和本地/CI 验证分工见
 
 本文件只约束稳定边界，不记录单次 PR 进度，也不把未来可能支持的生态能力提前声明为公开接口。
 
+ACP Client 与 Runtime 事件投递、远程控制及 HarmonyOS 演进的分阶段设计见
+[ACP Client、Runtime 事件总线与移动端支持设计](acp-client-runtime-event-integration-design.md)。
+
 ## 1. 架构目标
 
 BitFun 同时面向桌面 GUI、TUI/CLI、Web、ACP、Server、Remote、SDK 和插件生态。架构目标是降低后端实现高频变更对稳定接口的影响，同时保持插件生态和 OpenCode-compatible 能力可以按受控路径扩展。

src/apps/cli/src/account.rs

@@ -503,6 +503,7 @@ impl CliAccountRoutingHost {
                 let response_json = serde_json::to_string(&response).unwrap_or_else(|error| {
                     serde_json::to_string(&RemoteResponse::Error {
                         message: format!("failed to serialize RPC response: {error}"),
+                        code: None,
                     })
                     .unwrap_or_else(|_| {
                         r#"{"resp":"error","message":"serialize failed"}"#.to_string()

src/apps/cli/src/modes/exec/lifecycle.rs

@@ -1289,10 +1289,16 @@ impl ExecMode {
         match event {
             AgenticEvent::ModelRoundStarted {
                 turn_id: event_turn_id,
-                model_config_id,
+                identity:
+                    bitfun_events::ModelRoundIdentity::Native {
+                        model_config_id, ..
+                    },
                 ..
+            } if event_turn_id == turn_id => {
+                self.record_resolved_model_config_id(session_id, model_config_id)
+                    .await;
             }
-            | AgenticEvent::ModelRoundCompleted {
+            AgenticEvent::ModelRoundCompleted {
                 turn_id: event_turn_id,
                 model_config_id,
                 ..

src/apps/desktop/src/api/app_state.rs

@@ -77,6 +77,7 @@ pub struct AppState {
     pub agent_registry: Arc<agents::AgentRegistry>,
     pub mcp_service: Option<Arc<mcp::MCPService>>,
     pub acp_client_service: Option<Arc<bitfun_acp::AcpClientService>>,
+    pub acp_event_publisher: Arc<crate::runtime::AcpEventPublisher>,
     pub token_usage_service: Arc<token_usage::TokenUsageService>,
     pub miniapp_manager: Arc<MiniAppManager>,
     pub js_worker_pool: Option<Arc<JsWorkerPool>>,
@@ -101,6 +102,7 @@ pub struct AppState {
 impl AppState {
     pub async fn new_async(
         token_usage_service: Arc<token_usage::TokenUsageService>,
+        acp_event_publisher: Arc<crate::runtime::AcpEventPublisher>,
     ) -> BitFunResult<Self> {
         let start_time = std::time::Instant::now();
 
@@ -344,6 +346,7 @@ impl AppState {
             agent_registry,
             mcp_service,
             acp_client_service,
+            acp_event_publisher,
             token_usage_service,
             miniapp_manager,
             js_worker_pool,

src/apps/desktop/src/api/event_coalescer.rs

@@ -766,6 +766,67 @@ mod tests {
         assert_eq!(coalescer.buffered_chars(), 0);
     }
 
+    #[test]
+    fn acp_shaped_text_chunks_with_empty_attempt_fields_share_one_key() {
+        // ACP stream mapping leaves attempt_id/attempt_index as None. The
+        // coalescer must normalize both to the same "none" token so one round
+        // stays one stream. If this fallback ever becomes a uuid or round_id,
+        // ACP streaming text silently fragments into one event per chunk.
+        let mut coalescer = TextChunkCoalescer::new();
+        assert_eq!(resolve_attempt_token(&None, None), "none");
+        assert!(coalescer
+            .push(text_chunk(
+                "acp-session",
+                "turn-1",
+                "round-1",
+                None,
+                None,
+                "Hel"
+            ))
+            .is_empty());
+        assert!(coalescer
+            .push(text_chunk(
+                "acp-session",
+                "turn-1",
+                "round-1",
+                None,
+                None,
+                "lo "
+            ))
+            .is_empty());
+        assert!(coalescer
+            .push(text_chunk(
+                "acp-session",
+                "turn-1",
+                "round-1",
+                None,
+                None,
+                "ACP"
+            ))
+            .is_empty());
+
+        let events = coalescer.flush();
+        assert_eq!(events.len(), 1);
+        match &events[0] {
+            AgenticEvent::TextChunk {
+                session_id,
+                turn_id,
+                round_id,
+                attempt_id,
+                attempt_index,
+                text,
+            } => {
+                assert_eq!(session_id, "acp-session");
+                assert_eq!(turn_id, "turn-1");
+                assert_eq!(round_id, "round-1");
+                assert!(attempt_id.is_none());
+                assert!(attempt_index.is_none());
+                assert_eq!(text, "Hello ACP");
+            }
+            other => panic!("expected merged ACP TextChunk, got {other:?}"),
+        }
+    }
+
     #[test]
     fn rate_ema_resets_after_long_idle_gap() {
         // A window longer than the reset threshold must replace the estimate

src/apps/desktop/src/api/remote_workspace_policy.rs

@@ -894,6 +894,10 @@ pub const REMOTE_WORKSPACE_COMMAND_POLICIES: &[(&str, RemoteWorkspacePolicy)] =
         RemoteWorkspacePolicy::LegacyUnaudited,
     ),
     ("install_update", RemoteWorkspacePolicy::WorkspaceAgnostic),
+    (
+        "list_acp_pending_permissions",
+        RemoteWorkspacePolicy::WorkspaceAgnostic,
+    ),
     (
         "list_agent_companion_pets",
         RemoteWorkspacePolicy::LegacyUnaudited,

src/apps/desktop/src/lib.rs

@@ -724,13 +724,24 @@ pub async fn run() {
     );
 
     let step_started = Instant::now();
-    let app_state = match AppState::new_async(token_usage_service).await {
+    let acp_event_publisher = crate::runtime::AcpEventPublisher::start(event_queue.clone());
+    let app_state = match AppState::new_async(token_usage_service, acp_event_publisher).await {
         Ok(state) => state,
         Err(e) => {
             log::error!("Failed to initialize AppState: {}", e);
             return;
         }
     };
+    if let Some(service) = app_state.acp_client_service.clone() {
+        let mailbox = bitfun_services_integrations::remote_connect::install_acp_permission_mailbox(
+            std::sync::Arc::new(
+                bitfun_services_integrations::remote_connect::AcpPermissionMailbox::default(),
+            ),
+        );
+        service.set_permission_observer(std::sync::Arc::new(
+            runtime::DesktopAcpPermissionObserver::new(mailbox),
+        ));
+    }
     startup_timings.record_elapsed("initialize_app_state", step_started);
     startup_trace.record_elapsed_step("native_pre_tauri", "initialize_app_state", step_started);
 
@@ -769,6 +780,24 @@ pub async fn run() {
         "initialize_desktop_agent_runtime",
         step_started,
     );
+    let acp_projection_writer =
+        runtime::install_desktop_acp_writer(runtime::AcpDurableProjectionWriter::new(
+            event_queue.clone(),
+            desktop_runtime.session_application().clone(),
+        ));
+    event_router.subscribe_internal(
+        "acp_durable_projection".to_string(),
+        acp_projection_writer.clone(),
+    );
+    if let Some(service) = app_state.acp_client_service.clone() {
+        bitfun_core::service::remote_connect::set_remote_acp_control_host(std::sync::Arc::new(
+            runtime::DesktopRemoteAcpControlHost::new(
+                service,
+                app_state.acp_event_publisher.clone(),
+                acp_projection_writer,
+            ),
+        ));
+    }
 
     let coordinator_state = CoordinatorState {
         coordinator: coordinator.clone(),
@@ -1682,6 +1711,7 @@ pub async fn run() {
             load_acp_json_config,
             save_acp_json_config,
             submit_acp_permission_response,
+            list_acp_pending_permissions,
             create_acp_flow_session,
             start_acp_dialog_turn,
             cancel_acp_dialog_turn,
@@ -2307,6 +2337,7 @@ pub(crate) fn perform_process_exit_cleanup() -> bool {
     if let Some(search_service) = get_global_workspace_search_service() {
         search_service.shutdown_blocking();
     }
+    runtime::flush_desktop_acp_writer_blocking();
     bitfun_core::util::process_manager::cleanup_all_processes();
     api::remote_connect_api::cleanup_on_exit();
     true
@@ -2547,6 +2578,8 @@ fn start_event_loop_with_transport(
     session_event_journal: Arc<SessionEventJournal>,
 ) {
     tokio::spawn(async move {
+        // Live WebView delivery stays origin-unaware. Origin is a subscriber
+        // routing fact on the envelope; journal/backfill do not persist it yet.
         event_loop_driver(event_queue, event_router, |event| {
             let transport = transport.clone();
             let session_event_journal = session_event_journal.clone();