refactor(runtime): unify embedded event subscriptions

docs/architecture/agent-runtime-deployment-design.md

@@ -30,14 +30,15 @@ flowchart LR
 
 | 范围 | 当前状态 |
 |---|---|
-| Embedded GUI/Headless CLI/ACP/SDK Host | 保持现状；本设计没有改变其依赖或生命周期 |
-| Embedded interactive TUI | CLI crate 私有的 `CliAgentRuntimeClient` 统一暴露 Session、Turn、Permission 和事件访问；前三者使用 Rust Runtime SDK（当前 preview），事件继续复用既有 CLI event source；仍在当前 CLI 进程内运行 |
+| Embedded Desktop GUI | 继续使用现有 Desktop 事件投影和 Tauri adapter；本设计没有改变其依赖或生命周期 |
+| Embedded TUI/Headless CLI/Peer Host | Session、Turn、Permission 和事件订阅统一通过同一个 Rust Runtime SDK（当前 preview）；CLI crate 只保留第一方 adapter 和各形态自己的展示/断流策略 |
+| ACP/SDK Host | 使用同一个 Runtime 事件入口的 session-scoped 订阅；各自协议和进程生命周期保持独立 |
 | Runtime ownership | 已有可选的 Embedded 共享锁 / Shared 独占锁原语；尚未接入产品入口 |
 | Shared local IPC | 已有未发布、仅 crate 内可见的 discovery、实例锁、严格握手、Health 和 cleanup 基础；尚无生产 consumer |
 | Shared Session/Turn/Tool/Permission | 尚未设计为稳定 wire，也没有产品 consumer |
 | Shared GUI/TUI/Remote | 尚未交付，没有 `--shared` 或隐藏 Host 命令 |
 
-因此当前新增的是基础设施，不是用户可用的 Shared Runtime 产品。TUI 的私有 client 只收敛第一方调用边界，不代表事件已经迁入 Rust Runtime SDK（当前 preview），也不代表已经存在 Shared consumer。
+因此当前完成的是 Embedded 入口的调用边界收敛，不是用户可用的 Shared Runtime 产品。具体 `EventQueue` 仍由 Core 产品装配，Runtime SDK 只提供同进程订阅入口；没有 Shared event wire、事件重放或 Shared consumer。
 
 ## 2. 最少名词
 
@@ -70,6 +71,26 @@ flowchart TB
 
 复用的是 Runtime API、权威事实和 owner；不复用 renderer、CLI 参数、SDK wire、远程认证或平台窗口生命周期。任何新能力必须先进入既有 Runtime owner，再由需要它的 adapter 映射，禁止在 Shared 路径复制业务实现。
 
+### 3.1 Embedded 事件交付
+
+```mermaid
+flowchart LR
+  Queue["EventQueue"] --> Owner["Core product event queue owner"]
+  Owner -->|"injects read-only AgentEventSource"| Runtime["Agent Runtime API"]
+  Runtime --> TUI["TUI adapter"]
+  Runtime --> Exec["Headless adapter"]
+  Runtime --> Peer["Peer fanout adapter"]
+  Runtime --> ACP["ACP adapter"]
+  Runtime --> SDK["SDK Host adapter"]
+```
+
+- Core product assembly 创建事件 source，并维持旧消费队列的排空 task；第一方产品入口不再获得第二个订阅 API。
+- TUI、Headless CLI 和 Peer Host 只从 `AgentRuntime` 订阅，不能直接持有 Core-specific event source。
+- `bitfun-core` 的旧 event-source/builder API 仅保留为 deprecated 源码兼容 facade；它们委托给同一个 Core owner，不形成第二套运行时或第一方调用路径。
+- 各 adapter 继续拥有自己的失败投影：TUI 标记当前视图不可信，Headless CLI 返回非成功终态，Peer Host 中断其拥有的 turns，ACP 取消 turn 并返回协议错误，SDK Host 终结 Query 并提供 `RestartHost` recovery。
+- 有界 receiver 的 `Lagged` 或 `Closed` 是显式失败；当前没有 cursor/replay 合同，禁止伪装成透明恢复。
+- 这条链路仍全部位于当前 Embedded 进程，不增加 SDK Host、IPC 或后台进程依赖。
+
 ## 4. 当前基础架构
 
 ### 4.1 Runtime ownership
@@ -141,7 +162,8 @@ flowchart LR
 ```
 
 - CLI 不依赖 SDK Host，GUI/TUI 也不依赖公开 SDK package。
-- 交互式 TUI 的启动页和会话页复用一个 CLI 私有 Runtime client；Session、Turn 和 Permission 使用 Rust Runtime SDK（当前 preview），事件继续使用既有 CLI event source。该 client 只是第一方 adapter，不是公开 SDK 或第二套 Runtime。
+- 交互式 TUI 的启动页和会话页复用一个 CLI 私有 Runtime client；Session、Turn、Permission 和事件订阅都使用 Rust Runtime SDK（当前 preview）。该 client 只是第一方 adapter，不是公开 SDK、SDK Host client 或第二套 Runtime。
+- Headless CLI 和 Peer Host 使用同一 Runtime 订阅入口，但分别保留确定性退出与 Peer fanout 语义；共享订阅入口不等于共享 renderer 或产品生命周期。
 - TUI 不是 Server；未来是否连接 Shared deployment 是部署选择，不改变 TUI 的 renderer/键位职责。
 - Agent SDK Host 只服务外部 SDK 合同，不成为第一方 rich-client 的通用底座。
 - Headless CLI 默认继续 Embedded；CI 或测试可保持独立进程和独立 workspace，不承担后台实例成本。

src/apps/cli/src/agent/runtime_client.rs

@@ -10,7 +10,7 @@ use std::sync::{Arc, RwLock};
 use tokio::sync::Mutex;
 
 use bitfun_agent_runtime::sdk::{
-    AgentDialogTurnRequest, AgentLocalCommandTurnRecordRequest, AgentRuntime,
+    AgentDialogTurnRequest, AgentEventReceiver, AgentLocalCommandTurnRecordRequest, AgentRuntime,
     AgentSessionCreateRequest, AgentSessionDeleteRequest, AgentSessionForkRequest,
     AgentSessionForkResult, AgentSessionListRequest, AgentSessionModeUpdateRequest,
     AgentSessionModelUpdateRequest, AgentSessionRestoreRequest, AgentSessionUsageRequest,
@@ -23,7 +23,6 @@ use bitfun_agent_runtime::user_questions::USER_INPUT_AVAILABLE_CONTEXT_KEY;
 use bitfun_runtime_ports::{AgentSessionSummary, AgentSubmissionSource, DialogSubmissionPolicy};
 
 use crate::runtime::approval::CliApprovalPolicy;
-use crate::runtime::events::CliAgentEventSource;
 use crate::runtime::CliRuntimeContext;
 
 fn validated_session_summary(
@@ -99,7 +98,6 @@ fn session_mode_migration_notice(
 /// Stateless regarding agent_type; callers pass it per call.
 pub(crate) struct CliAgentRuntimeClient {
     runtime: AgentRuntime,
-    event_source: CliAgentEventSource,
     approval_policy: Arc<RwLock<CliApprovalPolicy>>,
     workspace_path: Arc<RwLock<Option<PathBuf>>>,
     /// Session ID — uses Mutex for interior mutability
@@ -112,16 +110,15 @@ impl CliAgentRuntimeClient {
     pub(crate) fn new(runtime: &CliRuntimeContext, workspace_path: Option<PathBuf>) -> Self {
         Self {
             runtime: runtime.agent_runtime().clone(),
-            event_source: runtime.agent_events().clone(),
             approval_policy: Arc::new(RwLock::new(runtime.approval_policy())),
             workspace_path: Arc::new(RwLock::new(workspace_path)),
             session_id: Arc::new(Mutex::new(None)),
             current_turn_id: Arc::new(Mutex::new(None)),
         }
     }
 
-    pub(crate) fn event_source(&self) -> &CliAgentEventSource {
-        &self.event_source
+    pub(crate) fn subscribe_events(&self) -> std::result::Result<AgentEventReceiver, RuntimeError> {
+        self.runtime.subscribe_events()
     }
 
     pub(crate) fn subscribe_permission_requests(
@@ -696,6 +693,18 @@ mod tests {
         assert!(!source.contains(&compatibility_update));
     }
 
+    #[test]
+    fn agent_events_use_the_runtime_sdk_without_a_core_event_source() {
+        let source = include_str!("runtime_client.rs").replace("\r\n", "\n");
+        let runtime_subscription = ["self.runtime", ".subscribe_events()"].concat();
+        let core_event_field = ["event_source", ": CliAgent", "EventSource"].concat();
+        let core_event_method = ["pub(crate) fn event", "_source("].concat();
+
+        assert!(source.contains(&runtime_subscription));
+        assert!(!source.contains(&core_event_field));
+        assert!(!source.contains(&core_event_method));
+    }
+
     fn session_summary(session_id: &str) -> AgentSessionSummary {
         AgentSessionSummary {
             session_id: session_id.to_string(),

src/apps/cli/src/modes/chat/run.rs

@@ -174,7 +174,10 @@ impl ChatMode {
             }
         }
 
-        let mut event_rx = self.agent.event_source().subscribe();
+        let mut event_rx = self
+            .agent
+            .subscribe_events()
+            .map_err(|error| anyhow::anyhow!(error.into_message()))?;
         let mut permission_rx = self.agent.subscribe_permission_requests().ok();
         if let Ok(pending) = self.agent.pending_permission_requests() {
             for request in pending.into_iter().filter(|request| {

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

@@ -617,7 +617,10 @@ impl ExecMode {
             }
         };
         tracing::info!(session_id = %session_id, "Session ready");
-        let mut event_rx = self.agent.event_source().subscribe();
+        let mut event_rx = self
+            .agent
+            .subscribe_events()
+            .map_err(|error| anyhow::anyhow!(error.into_message()))?;
 
         self.print_text(|| {
             eprintln!("Executing: {}", self.message);

src/apps/cli/src/peer_host/bootstrap.rs

@@ -30,12 +30,16 @@ pub(crate) async fn ensure_peer_host_ready(runtime: &CliRuntimeContext) -> Resul
     };
 
     let filesystem_service = Arc::new(FileSystemServiceFactory::create_default());
+    let agent_events = runtime
+        .agent_runtime()
+        .subscribe_events()
+        .map_err(|error| anyhow::anyhow!(error.into_message()))
+        .context("Peer Host agent event stream is unavailable")?;
 
     let state = PeerHostState {
         agent_runtime: runtime.agent_runtime().clone(),
         local_workspace_snapshot: runtime.local_workspace_snapshot().clone(),
         compatibility: runtime.compatibility().clone(),
-        agent_events: runtime.agent_events().clone(),
         turns: PeerTurnTracker::new(),
         workspace_service,
         filesystem_service,
@@ -46,7 +50,7 @@ pub(crate) async fn ensure_peer_host_ready(runtime: &CliRuntimeContext) -> Resul
         return Ok(());
     }
 
-    start_peer_event_fanout(state);
+    start_peer_event_fanout(state, agent_events);
     tracing::info!("CLI peer host services ready");
     Ok(())
 }

src/apps/cli/src/peer_host/fanout.rs

@@ -3,7 +3,7 @@
 use std::collections::HashSet;
 use std::sync::OnceLock;
 
-use bitfun_agent_runtime::sdk::PermissionRequestEvent;
+use bitfun_agent_runtime::sdk::{AgentEventReceiver, PermissionRequestEvent};
 use bitfun_agent_tools::effective_tool_invocation;
 use bitfun_core::service::remote_connect::encryption::encrypt_to_base64;
 use bitfun_core::service::remote_connect::remote_server::RemoteCommand;
@@ -85,9 +85,8 @@ fn peer_event_sender() -> &'static mpsc::Sender<QueuedPeerDeviceEvent> {
 }
 
 /// Subscribe to the invocation-scoped event source and forward only Peer-owned turns.
-pub(crate) fn start_peer_event_fanout(state: PeerHostState) {
+pub(crate) fn start_peer_event_fanout(state: PeerHostState, mut rx: AgentEventReceiver) {
     start_peer_permission_event_fanout(state.clone());
-    let mut rx = state.agent_events.subscribe();
     state.turns.mark_event_stream_ready();
     tokio::spawn(async move {
         loop {

src/apps/cli/src/peer_host/state.rs

@@ -10,8 +10,6 @@ use bitfun_core::service::filesystem::FileSystemService;
 use bitfun_core::service::workspace::WorkspaceService;
 use bitfun_runtime_ports::{AgentSubmissionSource, AgentTurnCancellationRequest};
 
-use crate::runtime::events::CliAgentEventSource;
-
 const MAX_TRACKED_PEER_TURNS: usize = i32::MAX as usize;
 const MAX_BACKGROUND_PEER_AUTHORIZATIONS: usize = i32::MAX as usize;
 const MAX_PENDING_PEER_TASK_CANCELLATIONS: usize = i32::MAX as usize;
@@ -903,7 +901,6 @@ pub(crate) struct PeerHostState {
     pub(crate) agent_runtime: AgentRuntime,
     pub(crate) local_workspace_snapshot: Arc<dyn bitfun_runtime_ports::LocalWorkspaceSnapshotPort>,
     pub(crate) compatibility: CoreAgentRuntimeCompatibility,
-    pub(crate) agent_events: CliAgentEventSource,
     pub(crate) turns: PeerTurnTracker,
     pub(crate) workspace_service: Arc<WorkspaceService>,
     pub(crate) filesystem_service: Arc<FileSystemService>,

src/apps/cli/src/runtime/mod.rs

@@ -7,7 +7,7 @@ use bitfun_core::agentic::system::AgenticSystem;
 use bitfun_core::product_assembly::{ProductAssemblyPlan, ProductServiceCapabilityAvailability};
 use bitfun_core::product_runtime::{
     build_local_runtime_services, ensure_product_dialog_scheduler, CoreAgentRuntimeCompatibility,
-    CoreLocalWorkspaceSnapshot, CoreProductAgentRuntime,
+    CoreLocalWorkspaceSnapshot, CoreProductAgentRuntime, CoreProductEventQueueOwner,
 };
 use bitfun_core::runtime_ports::PluginRuntimeAvailability;
 use bitfun_runtime_ports::LocalWorkspaceSnapshotPort;
@@ -16,10 +16,8 @@ use bitfun_runtime_services::RuntimeServices;
 use crate::product_assembly::{assemble_acp_runtime_parts, assemble_cli_runtime_parts};
 
 pub(crate) mod approval;
-pub(crate) mod events;
 
 use approval::CliApprovalPolicy;
-use events::CliAgentEventSource;
 
 const RUNTIME_EVENT_BUFFER: usize = 256;
 
@@ -55,7 +53,7 @@ pub(crate) struct CliRuntimeContext {
     agent_runtime: AgentRuntime,
     local_workspace_snapshot: Arc<dyn LocalWorkspaceSnapshotPort>,
     compatibility: CoreAgentRuntimeCompatibility,
-    agent_events: CliAgentEventSource,
+    _agent_event_queue_owner: CoreProductEventQueueOwner,
     services: RuntimeServices,
     product: CliProductRuntimeState,
     approval_policy: CliApprovalPolicy,
@@ -85,11 +83,13 @@ impl CliRuntimeContext {
                 .collect(),
         };
         let (services, harness_registry, _disabled_plugin_runtime) = parts.into_runtime_parts();
-        let agent_events = CliAgentEventSource::new(agentic_system.event_queue.clone());
-        let agent_runtime = CoreProductAgentRuntime::build(
+        let agent_event_queue_owner =
+            CoreProductEventQueueOwner::new(agentic_system.event_queue.clone());
+        let agent_runtime = CoreProductAgentRuntime::build_with_event_source(
             agentic_system.coordinator.clone(),
             scheduler.clone(),
             agentic_system.token_usage_service.clone(),
+            agent_event_queue_owner.runtime_source(),
             services.clone(),
             harness_registry,
         )
@@ -110,7 +110,7 @@ impl CliRuntimeContext {
 
         Ok(Self {
             workspace_root,
-            agent_events,
+            _agent_event_queue_owner: agent_event_queue_owner,
             agent_runtime,
             local_workspace_snapshot,
             compatibility,
@@ -136,10 +136,6 @@ impl CliRuntimeContext {
         &self.local_workspace_snapshot
     }
 
-    pub(crate) fn agent_events(&self) -> &CliAgentEventSource {
-        &self.agent_events
-    }
-
     pub(crate) fn services(&self) -> &RuntimeServices {
         &self.services
     }
@@ -155,9 +151,9 @@ impl CliRuntimeContext {
 
 #[derive(Clone)]
 pub(crate) struct AcpRuntimeContext {
-    _agent_events: CliAgentEventSource,
     agent_runtime: AgentRuntime,
     compatibility: CoreAgentRuntimeCompatibility,
+    _agent_event_queue_owner: CoreProductEventQueueOwner,
 }
 
 impl AcpRuntimeContext {
@@ -170,11 +166,12 @@ impl AcpRuntimeContext {
         let parts = assemble_acp_runtime_parts(services)
             .context("Failed to assemble ACP product runtime")?;
         let (services, harness_registry, _disabled_plugin_runtime) = parts.into_runtime_parts();
-        let agent_events = CliAgentEventSource::new(agentic_system.event_queue.clone());
+        let agent_event_queue_owner =
+            CoreProductEventQueueOwner::new(agentic_system.event_queue.clone());
         let agent_runtime = CoreProductAgentRuntime::build_acp(
             agentic_system.coordinator.clone(),
             scheduler.clone(),
-            agent_events.runtime_source(),
+            agent_event_queue_owner.runtime_source(),
             services,
             harness_registry,
         )
@@ -184,9 +181,9 @@ impl AcpRuntimeContext {
             CoreAgentRuntimeCompatibility::build(agentic_system.coordinator, scheduler);
 
         Ok(Self {
-            _agent_events: agent_events,
             agent_runtime,
             compatibility,
+            _agent_event_queue_owner: agent_event_queue_owner,
         })
     }
 

src/apps/sdk-host/src/runtime.rs

@@ -5,8 +5,8 @@ use bitfun_agent_runtime::sdk::AgentRuntime;
 use bitfun_core::agentic::system::{self, AgenticSystem};
 use bitfun_core::product_assembly::{DeliveryProfile, ProductAssembler, ProductAssemblyInput};
 use bitfun_core::product_runtime::{
-    build_local_runtime_services, ensure_product_dialog_scheduler, CoreProductAgentEventSource,
-    CoreProductAgentRuntime, CoreRuntimeServicesProvider,
+    build_local_runtime_services, ensure_product_dialog_scheduler, CoreProductAgentRuntime,
+    CoreProductEventQueueOwner, CoreRuntimeServicesProvider,
 };
 
 const RUNTIME_EVENT_BUFFER: usize = 256;
@@ -15,7 +15,7 @@ const DELIVERY_PROFILE: DeliveryProfile = DeliveryProfile::Sdk;
 pub(crate) struct SdkHostRuntime {
     workspace_root: PathBuf,
     agent_runtime: AgentRuntime,
-    _agent_events: CoreProductAgentEventSource,
+    _agent_event_queue_owner: CoreProductEventQueueOwner,
 }
 
 impl SdkHostRuntime {
@@ -40,12 +40,13 @@ impl SdkHostRuntime {
         bind_core_execution_ports(&agentic_system);
         let scheduler = ensure_product_dialog_scheduler(&agentic_system);
         let (services, harness_registry, _disabled_plugin_runtime) = parts.into_runtime_parts();
-        let agent_events = CoreProductAgentEventSource::new(agentic_system.event_queue.clone());
+        let agent_event_queue_owner =
+            CoreProductEventQueueOwner::new(agentic_system.event_queue.clone());
         let agent_runtime = CoreProductAgentRuntime::build_sdk_host(
             agentic_system.coordinator,
             scheduler,
             agentic_system.token_usage_service,
-            agent_events.runtime_source(),
+            agent_event_queue_owner.runtime_source(),
             services,
             harness_registry,
         )
@@ -55,7 +56,7 @@ impl SdkHostRuntime {
         Ok(Self {
             workspace_root,
             agent_runtime,
-            _agent_events: agent_events,
+            _agent_event_queue_owner: agent_event_queue_owner,
         })
     }