From 416489e5b7ab45d998cfe8d486ba1520d8500b65 Mon Sep 17 00:00:00 2001 From: limityan Date: Mon, 27 Jul 2026 13:33:23 +0800 Subject: [PATCH] refactor(runtime): unify embedded event subscriptions --- .../agent-runtime-deployment-design.md | 30 ++++- src/apps/cli/src/agent/runtime_client.rs | 21 +++- src/apps/cli/src/modes/chat/run.rs | 5 +- src/apps/cli/src/modes/exec/lifecycle.rs | 5 +- src/apps/cli/src/peer_host/bootstrap.rs | 8 +- src/apps/cli/src/peer_host/fanout.rs | 5 +- src/apps/cli/src/peer_host/state.rs | 3 - src/apps/cli/src/runtime/events.rs | 1 - src/apps/cli/src/runtime/mod.rs | 27 ++-- src/apps/sdk-host/src/runtime.rs | 13 +- .../assembly/core/src/product_runtime.rs | 118 ++++++++++++++++-- .../core/src/service_agent_runtime.rs | 3 +- 12 files changed, 184 insertions(+), 55 deletions(-) delete mode 100644 src/apps/cli/src/runtime/events.rs diff --git a/docs/architecture/agent-runtime-deployment-design.md b/docs/architecture/agent-runtime-deployment-design.md index 495ffc98f0..a877218b19 100644 --- a/docs/architecture/agent-runtime-deployment-design.md +++ b/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，不承担后台实例成本。 diff --git a/src/apps/cli/src/agent/runtime_client.rs b/src/apps/cli/src/agent/runtime_client.rs index fc558a29e7..56c85a967e 100644 --- a/src/apps/cli/src/agent/runtime_client.rs +++ b/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>, workspace_path: Arc>>, /// Session ID — uses Mutex for interior mutability @@ -112,7 +110,6 @@ impl CliAgentRuntimeClient { pub(crate) fn new(runtime: &CliRuntimeContext, workspace_path: Option) -> 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)), @@ -120,8 +117,8 @@ impl CliAgentRuntimeClient { } } - pub(crate) fn event_source(&self) -> &CliAgentEventSource { - &self.event_source + pub(crate) fn subscribe_events(&self) -> std::result::Result { + 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(), diff --git a/src/apps/cli/src/modes/chat/run.rs b/src/apps/cli/src/modes/chat/run.rs index a59f3f1276..1ed667ecdc 100644 --- a/src/apps/cli/src/modes/chat/run.rs +++ b/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| { diff --git a/src/apps/cli/src/modes/exec/lifecycle.rs b/src/apps/cli/src/modes/exec/lifecycle.rs index 668f209fa7..28c65b8149 100644 --- a/src/apps/cli/src/modes/exec/lifecycle.rs +++ b/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); diff --git a/src/apps/cli/src/peer_host/bootstrap.rs b/src/apps/cli/src/peer_host/bootstrap.rs index 18b3f1fcb2..81856d809a 100644 --- a/src/apps/cli/src/peer_host/bootstrap.rs +++ b/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(()) } diff --git a/src/apps/cli/src/peer_host/fanout.rs b/src/apps/cli/src/peer_host/fanout.rs index 722cb73360..72a5c19925 100644 --- a/src/apps/cli/src/peer_host/fanout.rs +++ b/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 { } /// 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 { diff --git a/src/apps/cli/src/peer_host/state.rs b/src/apps/cli/src/peer_host/state.rs index 04c34c9d0d..c1f5fc721c 100644 --- a/src/apps/cli/src/peer_host/state.rs +++ b/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, pub(crate) compatibility: CoreAgentRuntimeCompatibility, - pub(crate) agent_events: CliAgentEventSource, pub(crate) turns: PeerTurnTracker, pub(crate) workspace_service: Arc, pub(crate) filesystem_service: Arc, diff --git a/src/apps/cli/src/runtime/events.rs b/src/apps/cli/src/runtime/events.rs deleted file mode 100644 index 5c413a4ae7..0000000000 --- a/src/apps/cli/src/runtime/events.rs +++ /dev/null @@ -1 +0,0 @@ -pub(crate) use bitfun_core::product_runtime::CoreProductAgentEventSource as CliAgentEventSource; diff --git a/src/apps/cli/src/runtime/mod.rs b/src/apps/cli/src/runtime/mod.rs index a2de5ded21..2ceec6f876 100644 --- a/src/apps/cli/src/runtime/mod.rs +++ b/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, 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, }) } diff --git a/src/apps/sdk-host/src/runtime.rs b/src/apps/sdk-host/src/runtime.rs index 29f0d459da..9a38ad5171 100644 --- a/src/apps/sdk-host/src/runtime.rs +++ b/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, }) } diff --git a/src/crates/assembly/core/src/product_runtime.rs b/src/crates/assembly/core/src/product_runtime.rs index 8ccb501d55..83baefca9f 100644 --- a/src/crates/assembly/core/src/product_runtime.rs +++ b/src/crates/assembly/core/src/product_runtime.rs @@ -71,14 +71,15 @@ impl Drop for ProductEventQueueDrain { } } -/// Shared product-host event source that keeps the legacy queue bounded. +/// Product-host owner that keeps the legacy event queue bounded while the +/// assembled Agent Runtime exposes the read-only subscription surface. #[derive(Clone)] -pub struct CoreProductAgentEventSource { +pub struct CoreProductEventQueueOwner { source: AgentEventSource, _drain: Arc, } -impl CoreProductAgentEventSource { +impl CoreProductEventQueueOwner { pub fn new(queue: Arc) -> Self { Self { source: AgentEventSource::new(queue.clone()), @@ -86,12 +87,36 @@ impl CoreProductAgentEventSource { } } + pub fn runtime_source(&self) -> AgentEventSource { + self.source.clone() + } +} + +/// Compatibility wrapper for hosts that still subscribe outside `AgentRuntime`. +/// +/// First-party hosts must retain [`CoreProductEventQueueOwner`] and subscribe +/// through `AgentRuntime`. This wrapper remains only to avoid silently breaking +/// existing `bitfun-core` consumers during the migration. +#[deprecated(note = "use CoreProductEventQueueOwner and subscribe through AgentRuntime instead")] +#[derive(Clone)] +pub struct CoreProductAgentEventSource { + owner: CoreProductEventQueueOwner, +} + +#[allow(deprecated)] +impl CoreProductAgentEventSource { + pub fn new(queue: Arc) -> Self { + Self { + owner: CoreProductEventQueueOwner::new(queue), + } + } + pub fn subscribe(&self) -> AgentEventReceiver { - self.source.subscribe() + self.owner.runtime_source().subscribe() } pub fn runtime_source(&self) -> AgentEventSource { - self.source.clone() + self.owner.runtime_source() } } @@ -344,12 +369,49 @@ impl CoreProductAgentRuntime { ) } + #[deprecated(note = "use build_with_event_source for first-party product runtimes")] pub fn build( coordinator: Arc, scheduler: Arc, token_usage_service: Arc, services: RuntimeServices, harness_registry: HarnessRegistry, + ) -> Result { + Self::build_with_optional_event_source( + coordinator, + scheduler, + token_usage_service, + None, + services, + harness_registry, + ) + } + + pub fn build_with_event_source( + coordinator: Arc, + scheduler: Arc, + token_usage_service: Arc, + event_source: AgentEventSource, + services: RuntimeServices, + harness_registry: HarnessRegistry, + ) -> Result { + Self::build_with_optional_event_source( + coordinator, + scheduler, + token_usage_service, + Some(event_source), + services, + harness_registry, + ) + } + + fn build_with_optional_event_source( + coordinator: Arc, + scheduler: Arc, + token_usage_service: Arc, + event_source: Option, + services: RuntimeServices, + harness_registry: HarnessRegistry, ) -> Result { let session_operations = Arc::new(CoreSessionOperationsPort::new( coordinator.clone(), @@ -358,6 +420,7 @@ impl CoreProductAgentRuntime { CoreServiceAgentRuntime::product_agent_runtime( coordinator, scheduler, + event_source, session_operations.clone(), session_operations.clone(), session_operations, @@ -1048,7 +1111,7 @@ mod tests { use std::sync::Arc; use std::time::Duration; - use bitfun_agent_runtime::sdk::AgentRuntime; + use bitfun_agent_runtime::sdk::{AgentEventSource, AgentRuntime}; use bitfun_harness::HarnessRegistry; use bitfun_runtime_ports::{ LocalWorkspaceSnapshotSessionRequest, LocalWorkspaceSnapshotTurnRequest, @@ -1056,11 +1119,13 @@ mod tests { use bitfun_runtime_services::RuntimeServices; use uuid::Uuid; + #[allow(deprecated)] + use super::CoreProductAgentEventSource; use super::{ generate_core_session_usage_report, latest_persisted_turn_id, runtime_port_error, validate_latest_turn_fork_scope, validate_persisted_session_id, - CoreAgentRuntimeCompatibility, CoreLocalWorkspaceSnapshot, CoreProductAgentEventSource, - CoreProductAgentRuntime, CoreSessionOperationsPort, + CoreAgentRuntimeCompatibility, CoreLocalWorkspaceSnapshot, CoreProductAgentRuntime, + CoreProductEventQueueOwner, CoreSessionOperationsPort, }; use crate::agentic::coordination::{ConversationCoordinator, DialogScheduler}; use crate::agentic::events::{EventQueue, EventQueueConfig, EventRouter}; @@ -1122,12 +1187,13 @@ mod tests { } #[tokio::test] - async fn product_event_source_broadcasts_while_draining_the_legacy_queue() { + async fn product_event_queue_owner_broadcasts_while_draining_the_legacy_queue() { let queue = Arc::new(EventQueue::new(EventQueueConfig { max_queue_size: 4, batch_size: 2, })); - let source = CoreProductAgentEventSource::new(queue.clone()); + let owner = CoreProductEventQueueOwner::new(queue.clone()); + let source = owner.runtime_source(); let mut first = source.subscribe(); let mut second = source.subscribe(); @@ -1179,7 +1245,8 @@ mod tests { #[test] fn product_agent_runtime_exposes_reviewed_full_and_narrow_builders() { - fn build( + #[allow(deprecated)] + fn legacy_build( coordinator: Arc, scheduler: Arc, token_usage_service: Arc, @@ -1195,7 +1262,34 @@ mod tests { ) } - let _ = build; + fn build_with_event_source( + coordinator: Arc, + scheduler: Arc, + token_usage_service: Arc, + event_source: AgentEventSource, + services: RuntimeServices, + harness_registry: HarnessRegistry, + ) -> Result { + CoreProductAgentRuntime::build_with_event_source( + coordinator, + scheduler, + token_usage_service, + event_source, + services, + harness_registry, + ) + } + + #[allow(deprecated)] + fn legacy_event_source_methods_are_source_compatible() { + let _ = CoreProductAgentEventSource::new; + let _ = CoreProductAgentEventSource::subscribe; + let _ = CoreProductAgentEventSource::runtime_source; + } + + let _ = legacy_build; + let _ = build_with_event_source; + legacy_event_source_methods_are_source_compatible(); let _ = CoreProductAgentRuntime::build_session_surface; let _ = CoreProductAgentRuntime::build_acp; let _ = CoreProductAgentRuntime::build_sdk_host; diff --git a/src/crates/assembly/core/src/service_agent_runtime.rs b/src/crates/assembly/core/src/service_agent_runtime.rs index 1ceaa7ada0..de83d22cd2 100644 --- a/src/crates/assembly/core/src/service_agent_runtime.rs +++ b/src/crates/assembly/core/src/service_agent_runtime.rs @@ -1077,6 +1077,7 @@ impl CoreServiceAgentRuntime { pub(crate) fn product_agent_runtime( coordinator: Arc, scheduler: Arc, + event_source: Option, session_fork: Arc, session_usage: Arc, turn_settlement: Arc, @@ -1088,7 +1089,7 @@ impl CoreServiceAgentRuntime { coordinator, scheduler, dialog_turn, - None, + event_source, Some(session_fork), Some(session_usage), Some(turn_settlement),