diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2a1c7e869c..23795ca8a8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -52,8 +52,8 @@ jobs: cache-bin: false save-if: ${{ github.event_name != 'pull_request' }} - - name: Run CLI tests - run: cargo test --locked -p bitfun-cli + - name: Run CLI, ACP, and agent runtime tests + run: cargo test --locked -p bitfun-cli -p bitfun-acp -p bitfun-agent-runtime # ── Rust: build check ───────────────────────────────────────────── rust-build-check: diff --git a/docs/architecture/agent-runtime-services-design.md b/docs/architecture/agent-runtime-services-design.md index c981d18766..b0000ea795 100644 --- a/docs/architecture/agent-runtime-services-design.md +++ b/docs/architecture/agent-runtime-services-design.md @@ -9,8 +9,9 @@ CLI Agent 体验边界见 [`cli-product-line-design.md`](cli-product-line-design 本文中的接口片段只说明依赖方向和职责，不自动构成当前 API 或实施承诺。当前接口名称、字段和消费方以代码为准； 新增公共类型前必须有真实生产调用方、版本边界和验证路径。现有 Agent Runtime SDK 仍是 v1 preview，CLI、ACP、 -Desktop 仍保留 `bitfun-core/product-full` 兼容 owner。CLI 已消费独立的产品组装结果和 SDK 端口，但这不等于 -协调器、调度器、持久化或工具执行 owner 已迁移；ACP 与 Desktop 尚未完成对应入口切换。 +Desktop 仍保留 `bitfun-core/product-full` 兼容 owner。CLI 与 CLI 托管的 ACP server 已消费各自的产品组装结果和 +SDK 端口，但这不等于协调器、调度器、持久化或工具执行 owner 已迁移；ACP 的完整持久化历史、模型/模式、MCP、 +客户端路径与 Desktop 入口仍保留明确的兼容边界。 阅读路径：第 1 节确认 SDK、内核、产品特性、扩展接口和 crate 边界；第 2-3 节说明稳定接口、 运行时服务、内核、工具和工作流；第 4 节说明产品组装与扩展注册；第 5 节作为质量保护和 @@ -695,7 +696,8 @@ pub struct HarnessExecutionContext { `src/crates/assembly/core` 仍承担 `bitfun-core` 兼容组装。现有 `ProductAssembler` 是具体结构体， 通过 `assemble(ProductAssemblyInput)` 产生 `ProductRuntimeParts`，本文件不再为它定义第二套目标接口。 -当前 CLI 入口已使用类型化 `RuntimeServices` 构造 `ProductRuntimeParts`，并通过一个调用级上下文把 +当前 CLI 与 CLI 托管的 ACP server 已使用类型化 `RuntimeServices`，分别以 `DeliveryProfile::Cli` 和 +`DeliveryProfile::Acp` 构造 `ProductRuntimeParts`。CLI 通过一个调用级上下文把 Agent Runtime SDK、Harness、能力注册、调用级权限和 Agentic 事件广播交给 TUI、Exec、Session、Usage 与 交互模式下的 Peer Host。SDK 已承接会话创建/列举/删除/恢复、类型化转录读取、轮次提交/取消，以及 CLI/TUI 的工具确认、拒绝和用户问题回答；Peer Host 当前只消费同一 SDK 的工具确认与拒绝。SDK v1 尚未覆盖的固定 ID @@ -704,8 +706,10 @@ Agent Runtime SDK、Harness、能力注册、调用级权限和 Agentic 事件 Core 的 Network、Git 和 MCP Catalog 当前仍含兼容 marker，因此该诊断不等于对这些外部服务做实时探活。 该切换仍是 `product-full` 兼容组装，不是 owner 迁移。协调器、调度器、持久化、工具管线和 Agentic Event Queue -仍由 Core 唯一持有；CLI 不复制这些状态。ACP 与 Desktop 尚未接入组装结果，Server 仅提供健康检查、信息与 -ping 路由。未接入入口的 profile、枚举分支和单元测试仍不能证明对应产品形态可用。 +仍由 Core 唯一持有；CLI 与 ACP 不复制这些状态。ACP 服务端通过 SDK 处理会话创建/列举、轮次、取消、交互响应和事件订阅， +但完整持久化历史回放、模型/模式、模型目录/配置和 MCP 仍走单一 Core 兼容门面；ACP stdio、连接和协议投影仍在 +`interfaces/acp`。Desktop 尚未接入组装结果，Server 仅提供健康检查、信息与 ping 路由。未接入入口的 profile、 +枚举分支和单元测试仍不能证明对应产品形态可用。 职责： @@ -718,8 +722,8 @@ ping 路由。未接入入口的 profile、枚举分支和单元测试仍不能 | 阶段 | 约束 | |---|---| -| 当前 | CLI 与 Peer Host 消费真实 Runtime Parts 与 SDK，Core 兼容门面只承接 SDK v1 缺口；不扩张字段或再造描述符 | -| 迁移 | 迁移执行 owner 或继续接入 ACP、Desktop 前，必须分别证明行为等价；relay 的 Cargo 反向边已删除，room/device 状态、account/sync 存储、asset store 与 HTTP/WebSocket router 已下沉，但 embedded TCP bind、静态 fallback 和任务生命周期仍是 assembly 兼容债务 | +| 当前 | CLI、Peer Host 与 CLI 托管的 ACP server 消费真实 Runtime Parts / SDK，Core 兼容门面只承接已列明的 SDK v1 缺口；不扩张字段或再造描述符 | +| 迁移 | 迁移执行 owner、ACP 剩余兼容路径或 Desktop 入口前，必须分别证明行为等价；relay 的 Cargo 反向边已删除，room/device 状态、account/sync 存储、asset store 与 HTTP/WebSocket router 已下沉，但 embedded TCP bind、静态 fallback 和任务生命周期仍是 assembly 兼容债务 | | 完成 | 每个声称支持的 profile 都由生产入口消费组装结果，并有最小入口验证；无消费方的 profile 不对外宣称可用 | 产品定义、品牌资源和界面布局的长期边界以 @@ -731,6 +735,7 @@ ping 路由。未接入入口的 profile、枚举分支和单元测试仍不能 - 具体运行时服务通过 `RuntimeServicesBuilder` / provider registry 构造。 - CLI 只选择 `DeliveryProfile::Cli` 一次；必需服务缺失时组装失败，不回退到静态计划或另一 profile。 +- CLI 的 ACP stdio 入口只选择 `DeliveryProfile::Acp` 一次；组装或 SDK runtime 构造失败时在接受 stdio 请求前退出。 - CLI 的 `json` 输出为单结果文档，`stream-json` 直接复用现有 `AgenticEventEnvelope`；协议层不新增 `schema_version`、`sequence` 或平行事件 taxonomy。 - 能力计划选择工具提供方组计划和 Harness 描述符；当前不存在供任意模块注册所有对象的通用组装注册表。 @@ -778,8 +783,9 @@ ping 路由。未接入入口的 profile、枚举分支和单元测试仍不能 当前 Runtime SDK 已提供会话创建、列出、删除、恢复和类型化转录读取。`AgentSessionRestoreRequest/Result` 与 `AgentSessionRestorePort` 归 Agent Runtime SDK，以继续复用 Runtime owner 的完整 `SessionState`；类型化 -`SessionTranscript` 归 `runtime-ports`。两者都由 `assembly/core` 注入真实 persistence owner，CLI/TUI 是当前恢复与 -转录消费方。CLI/TUI 的工具确认、拒绝和用户问题回答，以及 Peer Host 的工具确认与拒绝，通过类型化 +`SessionTranscript` 归 `runtime-ports`。两者都由 `assembly/core` 注入真实持久化 owner，当前由 CLI/TUI +消费；ACP 为保证模型配置与完整历史来自同一次恢复，继续通过 Core 兼容门面读取协议回放所需的完整轮次，避免为单一协议 +扩张通用 transcript。CLI/TUI 的工具确认、拒绝和用户问题回答，以及 ACP 服务端 / Peer Host 的工具确认与拒绝，通过类型化 `AgentInteractionResponsePort` 回到 Core 的工具管线或用户输入 owner，不改变审批策略或交互所有权。 `CoreAgentRuntimeCompatibility` 仍承载未迁移的 固定 ID 会话创建、模型更新、持久化、分支、用量和快照等操作；不能据此把整个兼容门面一次性删除，也不能把这些 @@ -835,6 +841,11 @@ feature group、运行时服务要求、工具提供方组、Harness 描述符 `bitfun-acp` 保持集成归属。 +CLI 托管的 ACP 服务端使用 `DeliveryProfile::Acp` 组装一个 Agent Runtime，通过 SDK 处理会话创建/列举、轮次提交/取消、 +交互响应和只读 Agent 事件订阅。ACP 只把共享运行时事实映射成协议更新；标准输入输出、连接、权限 RPC 与通知生命周期 +不进入 SDK。完整持久化历史恢复、模型/模式、模型目录/配置、MCP 和 ACP 客户端路径仍是明确的 Core 兼容范围， +不据此扩张通用 runtime DTO。 + 继续拥有： - ACP protocol。 @@ -1021,13 +1032,19 @@ Product 测试： wire schema、Relay ACK/重放和重连协议未在该切换中扩张。 - CLI 通过 SDK 处理已覆盖的 session/turn/cancel 操作，并通过一个 Core 兼容门面处理 SDK v1 缺口； 该门面复用现有 owner，不建立第二套状态或事件 schema。 +- CLI 托管的 ACP 服务端已以 `DeliveryProfile::Acp` 构造真实 Runtime Parts；会话创建/列举、轮次、取消、工具确认/拒绝和 + Agent 事件订阅复用同一 SDK 语义，ACP stdio、连接与协议投影保持不变。Agentic Event Queue 仍是唯一事件 owner； + 全局有界 broadcast 继续服务 CLI/TUI，活动 ACP prompt 使用固定容量、仅接收本会话事件的临时通道，并在最后一个订阅者 + 释放时立即回收。CLI 宿主进程只保留一个旧消费队列排空任务，不增加每会话转发任务或第二套事件 schema。ACP 组装入口使用独立的 + 轮次提交适配器，在会话锁内拒绝忙碌会话的第二个 prompt；CLI/TUI、Desktop 和远程入口的既有排队策略不变。 仍需完成： - 把 embedded relay 的 TCP bind、静态 fallback 和任务生命周期移出 assembly；room/device 状态、account/sync 存储、asset store 与 HTTP/WebSocket router 已归属 `services/relay-service`，Cargo 反向边已删除并由通用边界检查保护。 - 继续缩小 CLI 的 Core 兼容门面；只有稳定端口、真实生产调用方和行为等价测试齐备时才迁移 owner。 -- 让 ACP、Desktop 依次接入产品组装，并为每条路径证明行为等价；ACP 生命周期和 Desktop 平台资源仍留在入口。 +- 继续按真实复用需求缩小 ACP 的完整持久化历史、模型/模式、MCP 与客户端兼容路径；Desktop 入口另行接入产品组装并 + 证明行为等价。ACP 生命周期和 Desktop 平台资源仍留在各自入口。 - 为 Agent Runtime SDK 增加至少一个非 `bitfun-core` 的真实嵌入方；预览 facade 和单元测试不等于外部可用 SDK。 - 仅在真实端到端切片中接入插件主机；外部插件先转换为类型化工具、Hook、事件、权限请求或诊断， 不把生态对象带入 Agent Runtime。 diff --git a/docs/architecture/cli-product-line-design.md b/docs/architecture/cli-product-line-design.md index f9922697cb..57ae1d453b 100644 --- a/docs/architecture/cli-product-line-design.md +++ b/docs/architecture/cli-product-line-design.md @@ -123,8 +123,9 @@ BitFun CLI 应成为可独立安装和发布的 Agent 产品，而不是 Desktop - 独立 CLI 测试与打包工作流；主 CI 的三平台 workspace check 同时覆盖 `bitfun-cli` 编译。 上述切换不等于运行时 owner 已迁移，也不表示 CLI-P0 全部完成。CLI crate 仍以 `bitfun-core/product-full` -承载协调器、调度器、持久化、工具管线和部分 SDK v1 缺口，但 Peer Host 不再自行构造这些 owner；ACP stdio -仍走原入口，插件命令仍以来源管理和静态预览为主。兼容门面只转发，不重新计算或写入同一事实。 +承载协调器、调度器、持久化、工具管线和部分 SDK v1 缺口，但 Peer Host 不再自行构造这些 owner；ACP 的 stdio、 +连接和协议投影仍由 `interfaces/acp` 持有，后端已切换至 `DeliveryProfile::Acp` 与组装后的 SDK runtime；插件命令 +仍以来源管理和静态预览为主。兼容门面只转发，不重新计算或写入同一事实。 目标态仍存在以下结构缺口： @@ -147,13 +148,14 @@ CLI-P0 的目标是建立后续功能补齐所需的稳定边界，不改变现 CLI-P0 不是一个统一重构 PR。静态 profile、真实 Runtime Services、Runtime Parts、调用级审批、共享事件源和 本地 Agent 纵向入口已接入；旧门面仅在后续 owner 迁移的行为等价成立后退出。配置解释、产品定制消费、TUI -进一步拆分、ACP 切换和 package smoke 仍需独立交付，不能由本次运行时切换代替。 +进一步拆分和 package smoke 仍需独立交付。CLI 托管的 ACP 服务端已独立切换到 ACP profile 与组装后的 SDK runtime； +真实模型、PTY 与权限失败等进程级验收仍需另行完成，不能由本次运行时切换代替。 其余工作独立立项，不能与 profile 迁移互相充当完成条件： | 切片 | 范围 | 退出条件 | |---|---|---| -| 调用级审批 | TUI 与 `exec` 已使用调用级策略且不写全局配置；ACP 需独立迁移 | Runtime-context `Allow always`、审批规划、`exec` 安全默认值和显式 `--auto` 有 focused test；真实模型/PTY 审批流与 ACP 另行验收 | +| 调用级审批 | TUI、`exec` 与 ACP 已使用各自调用级策略且不写全局配置 | Runtime-context `Allow always`、审批规划、`exec` 安全默认值和显式 `--auto` 有 focused test；真实模型/PTY 审批流与 ACP 仍需另行验收 | | 输出协议 | 保持通用 `text/json/stream-json` 心智，复用现有 Agentic envelope，不新建 CLI schema | 已覆盖结果/envelope 序列化、参数与前置 JSON 失败、失败完成、同会话跨 turn 隔离和 stream-json/Patch stdout 冲突；真实信号、模型权限失败与 Patch I/O 故障注入仍需进程级契约 | | 配置解释 | Canonical Config 层级、全局/项目持续来源、加载状态和兼容导入 dry-run | 不自动写入；冲突、未知字段、待确认能力和凭据引用可解释 | | 产品定制 | 消费最小产品定义、组装结果和已注册 TUI layout/theme ID | 第二个真实 CLI 产品复用后再提升公共字段 | @@ -585,7 +587,7 @@ CLI Agent 能力加强必须落在共享 Agent Runtime、Tool Runtime 或 Harnes | Action/Keymap | registry 唯一性、Slash/Palette/Help/dispatch 一致、配置键位真实输入、冲突来源和终端恢复 fallback | 通用 `cargo check --workspace` 负责三平台 CLI 编译保护；独立 CLI CI 运行 -`cargo test --locked -p bitfun-cli`。已落地的 focused 协议契约进入该测试；完整进程/PTY 矩阵与打包 smoke +`cargo test --locked -p bitfun-cli -p bitfun-acp -p bitfun-agent-runtime`。已落地的 focused 协议契约进入该测试；完整进程/PTY 矩阵与打包 smoke 仍按对应切片补入门禁，不能由序列化单测代替。 ### 10.2 阶段退出条件 diff --git a/docs/architecture/product-architecture.md b/docs/architecture/product-architecture.md index 7867074678..9e68558dfc 100644 --- a/docs/architecture/product-architecture.md +++ b/docs/architecture/product-architecture.md @@ -317,7 +317,7 @@ flowchart LR | HarmonyOS PC 原生 CLI/TUI | 未来平台目标，当前未实现 | 目标、问题和风险见平台规约；具体适配另立专题，HAP、手机 Remote App 与远端代执行均不替代 | | HarmonyOS PC GUI | 完整 HarmonyOS PC 支持的另一目标形态，当前未实现 | 与 CLI/TUI 共享稳定能力和 Runtime 语义，但独立设计宿主、界面与发布验证；Web、Remote 或现有 Tauri Desktop 均不能替代 | | HarmonyOS 手机 Remote App | `src/apps/mobile/harmonyos` 是 phone-only ArkTS 远程入口，不持有本地 Rust Agent Runtime | 保持当前能力并按移动端专题独立演进；本轮不提前设计移动 Runtime/TUI/GUI，也不能据此宣称 HarmonyOS PC 本地能力 | -| ACP | 生产入口仍直接依赖 `bitfun-core/product-full` | `DeliveryProfile::Acp` 尚未进入入口组装；不得把测试中的 profile 解释为生产隔离 | +| ACP | CLI 托管的服务端仍以 `bitfun-core/product-full` 作为兼容执行层 | 入口已选择 `DeliveryProfile::Acp` 并消费 Runtime Parts；组装层在入队前原子拒绝忙碌会话，不改变其他产品入口的排队行为；会话恢复、模型/模式、MCP、客户端与协议生命周期仍留在现有 Core/ACP 归属，不据此宣称完整解耦 | | Server / Remote | 当前生产路由没有插件状态消费闭环；Remote 插件执行未实现 | 不在本地替远端项目发现、准备或执行插件；未接入时返回明确不支持 | | Web / Mobile Web | 依赖现有后端入口，不持有插件执行单元 | 对应 profile 当前为空计划或未接入生产，不能据枚举值宣称独立产品能力 | | SDK | 仅有 preview 门面、空 profile 计划和测试替身 | 不牵引 `product-full`、具体服务管理器或插件 host ABI；未满足独立嵌入验证前不宣称可发布 | diff --git a/src/apps/cli/src/product_assembly.rs b/src/apps/cli/src/product_assembly.rs index 9dc5045fe5..eaedd61151 100644 --- a/src/apps/cli/src/product_assembly.rs +++ b/src/apps/cli/src/product_assembly.rs @@ -7,14 +7,27 @@ use bitfun_runtime_services::RuntimeServices; pub(crate) fn assemble_cli_runtime_parts( services: RuntimeServices, ) -> Result { - ProductAssembler::new().assemble(ProductAssemblyInput::new(DeliveryProfile::Cli, services)) + assemble_runtime_parts(DeliveryProfile::Cli, services) +} + +pub(crate) fn assemble_acp_runtime_parts( + services: RuntimeServices, +) -> Result { + assemble_runtime_parts(DeliveryProfile::Acp, services) +} + +fn assemble_runtime_parts( + profile: DeliveryProfile, + services: RuntimeServices, +) -> Result { + ProductAssembler::new().assemble(ProductAssemblyInput::new(profile, services)) } #[cfg(test)] mod tests { use std::sync::Arc; - use super::assemble_cli_runtime_parts; + use super::{assemble_acp_runtime_parts, assemble_cli_runtime_parts}; use crate::runtime::{ approval::{CliApprovalPolicy, CliPermissionService}, services::{CliClock, CliRuntimeEventSink, CliRuntimeServicesProvider}, @@ -87,4 +100,34 @@ mod tests { )); assert!(!parts.harness_registry().provider_ids().is_empty()); } + + #[test] + fn acp_product_assembly_uses_acp_profile_and_production_services() { + let workspace = tempfile::tempdir().expect("workspace"); + let services = CliRuntimeServicesProvider::new( + workspace.path(), + Arc::new(CliPermissionService::new(CliApprovalPolicy::Ask)), + Arc::new(CliRuntimeEventSink::new(8)), + Arc::new(CliClock), + ) + .expect("provider") + .build() + .expect("runtime services"); + + let parts = assemble_acp_runtime_parts(services).expect("ACP product runtime parts"); + + assert_eq!(parts.plan().profile(), DeliveryProfile::Acp); + assert!(parts.missing_service_requirements().is_empty()); + assert!(parts + .service_availability() + .iter() + .all(|entry| entry.status() == ProductServiceCapabilityStatus::Available)); + assert!(matches!( + parts.plugin_runtime().availability(), + PluginRuntimeAvailability::Disabled { + reason: PluginRuntimeUnavailableReason::UnsupportedProfile + } + )); + assert!(!parts.harness_registry().provider_ids().is_empty()); + } } diff --git a/src/apps/cli/src/root_handlers.rs b/src/apps/cli/src/root_handlers.rs index 260e233ac5..38950933dd 100644 --- a/src/apps/cli/src/root_handlers.rs +++ b/src/apps/cli/src/root_handlers.rs @@ -456,6 +456,9 @@ pub(crate) async fn serve_acp_stdio() -> Result<()> { .context("Failed to initialize agentic system")?; tracing::info!("Agentic system initialized"); - bitfun_acp::BitfunAcpRuntime::serve_stdio(agentic_system).await?; + let workspace_root = std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")); + let runtime = crate::runtime::AcpRuntimeContext::build(agentic_system, workspace_root)?; + let (agent_runtime, compatibility) = runtime.parts(); + bitfun_acp::BitfunAcpRuntime::serve_stdio(agent_runtime, compatibility).await?; Ok(()) } diff --git a/src/apps/cli/src/runtime/events.rs b/src/apps/cli/src/runtime/events.rs index f61e85ef15..77e65a396a 100644 --- a/src/apps/cli/src/runtime/events.rs +++ b/src/apps/cli/src/runtime/events.rs @@ -1,8 +1,7 @@ use std::sync::Arc; +use bitfun_agent_runtime::sdk::{AgentEventReceiver, AgentEventSource}; use bitfun_core::agentic::events::EventQueue; -use bitfun_events::AgenticEventEnvelope; -use tokio::sync::broadcast; struct EventQueueDrain { task: tokio::task::JoinHandle<()>, @@ -13,11 +12,7 @@ impl EventQueueDrain { let task = tokio::spawn(async move { loop { queue.wait_for_events().await; - loop { - if queue.dequeue_configured_batch().await.is_empty() { - break; - } - } + while !queue.dequeue_configured_batch().await.is_empty() {} } }); Self { task } @@ -32,20 +27,24 @@ impl Drop for EventQueueDrain { #[derive(Clone)] pub(crate) struct CliAgentEventSource { - queue: Arc, + source: AgentEventSource, _drain: Arc, } impl CliAgentEventSource { pub(crate) fn new(queue: Arc) -> Self { Self { - _drain: Arc::new(EventQueueDrain::start(queue.clone())), - queue, + source: AgentEventSource::new(queue.clone()), + _drain: Arc::new(EventQueueDrain::start(queue)), } } - pub(crate) fn subscribe(&self) -> broadcast::Receiver { - self.queue.subscribe() + pub(crate) fn subscribe(&self) -> AgentEventReceiver { + self.source.subscribe() + } + + pub(crate) fn runtime_source(&self) -> AgentEventSource { + self.source.clone() } } diff --git a/src/apps/cli/src/runtime/mod.rs b/src/apps/cli/src/runtime/mod.rs index 0268bff8aa..8a1f904cc6 100644 --- a/src/apps/cli/src/runtime/mod.rs +++ b/src/apps/cli/src/runtime/mod.rs @@ -10,7 +10,7 @@ use bitfun_core::product_runtime::{CoreAgentRuntimeCompatibility, CoreProductAge use bitfun_core::runtime_ports::PluginRuntimeAvailability; use bitfun_runtime_services::RuntimeServices; -use crate::product_assembly::assemble_cli_runtime_parts; +use crate::product_assembly::{assemble_acp_runtime_parts, assemble_cli_runtime_parts}; pub(crate) mod approval; pub(crate) mod events; @@ -90,6 +90,7 @@ 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( agentic_system.coordinator.clone(), scheduler.clone(), @@ -115,7 +116,7 @@ impl CliRuntimeContext { Ok(Self { workspace_root, - agent_events: CliAgentEventSource::new(agentic_system.event_queue.clone()), + agent_events, agent_runtime, compatibility, services, @@ -158,6 +159,57 @@ impl CliRuntimeContext { } } +#[derive(Clone)] +pub(crate) struct AcpRuntimeContext { + _agent_events: CliAgentEventSource, + agent_runtime: AgentRuntime, + compatibility: CoreAgentRuntimeCompatibility, +} + +impl AcpRuntimeContext { + pub(crate) fn build( + agentic_system: AgenticSystem, + workspace_root: impl AsRef, + ) -> Result { + let scheduler = ensure_dialog_scheduler(&agentic_system); + let runtime_events = Arc::new(CliRuntimeEventSink::new(RUNTIME_EVENT_BUFFER)); + let provider = CliRuntimeServicesProvider::new( + workspace_root, + Arc::new(CliPermissionService::new(CliApprovalPolicy::Ask)), + runtime_events, + Arc::new(CliClock), + )?; + let parts = assemble_acp_runtime_parts(provider.build()?) + .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_runtime = CoreProductAgentRuntime::build_acp( + agentic_system.coordinator.clone(), + scheduler.clone(), + agent_events.runtime_source(), + services, + harness_registry, + ) + .map_err(anyhow::Error::msg) + .context("Failed to build ACP Agent Runtime SDK")?; + let compatibility = CoreAgentRuntimeCompatibility::build( + agentic_system.coordinator, + scheduler, + agentic_system.token_usage_service, + ); + + Ok(Self { + _agent_events: agent_events, + agent_runtime, + compatibility, + }) + } + + pub(crate) fn parts(&self) -> (AgentRuntime, CoreAgentRuntimeCompatibility) { + (self.agent_runtime.clone(), self.compatibility.clone()) + } +} + fn ensure_dialog_scheduler(agentic_system: &AgenticSystem) -> Arc { if let Some(scheduler) = coordination::get_global_scheduler() { return scheduler; diff --git a/src/apps/cli/tests/acp_stdio_cli.rs b/src/apps/cli/tests/acp_stdio_cli.rs new file mode 100644 index 0000000000..b40b2c5b0b --- /dev/null +++ b/src/apps/cli/tests/acp_stdio_cli.rs @@ -0,0 +1,101 @@ +use std::process::Stdio; +use std::time::Duration; + +use serde_json::json; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; + +#[tokio::test] +async fn acp_stdio_initializes_the_production_assembled_runtime() { + let temp = tempfile::tempdir().expect("tempdir"); + let workspace = temp.path().join("workspace"); + let user_root = temp.path().join("user-root"); + let home_root = temp.path().join("home-root"); + let config_root = temp.path().join("host-config"); + std::fs::create_dir_all(&workspace).expect("create workspace"); + + let mut child = tokio::process::Command::new(env!("CARGO_BIN_EXE_bitfun-cli")) + .arg("acp") + .current_dir(&workspace) + .env_remove("BITFUN_USER_ROOT") + .env_remove("BITFUN_HOME") + .env("BITFUN_E2E_STORAGE_GUARD", "1") + .env("BITFUN_E2E_USER_ROOT", &user_root) + .env("BITFUN_E2E_HOME", &home_root) + .env("APPDATA", &config_root) + .env("XDG_CONFIG_HOME", &config_root) + .env("HOME", &home_root) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn() + .expect("start production ACP server"); + + let mut stdin = child.stdin.take().expect("ACP stdin"); + let mut stdout = BufReader::new(child.stdout.take().expect("ACP stdout")); + let mut stderr = child.stderr.take().expect("ACP stderr"); + let stderr_reader = tokio::spawn(async move { + let mut bytes = Vec::new(); + stderr + .read_to_end(&mut bytes) + .await + .expect("read ACP stderr"); + String::from_utf8_lossy(&bytes).into_owned() + }); + + let request = json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientCapabilities": {} + } + }); + let mut request = serde_json::to_vec(&request).expect("serialize initialize request"); + request.push(b'\n'); + stdin + .write_all(&request) + .await + .expect("write ACP initialize request"); + stdin.flush().await.expect("flush ACP initialize request"); + + let mut response_line = String::new(); + tokio::time::timeout( + Duration::from_secs(60), + stdout.read_line(&mut response_line), + ) + .await + .expect("production ACP initialize should not hang") + .expect("read ACP initialize response"); + let response: serde_json::Value = + serde_json::from_str(&response_line).expect("valid ACP JSON-RPC response"); + assert_eq!(response.get("id"), Some(&json!(1))); + assert_eq!(response.pointer("/result/protocolVersion"), Some(&json!(1))); + assert_eq!( + response.pointer("/result/agentInfo/name"), + Some(&json!("bitfun-acp")) + ); + + // ACP is a long-lived subprocess whose host owns termination. Close its + // request stream, then always reap it; the kill fallback covers protocol + // transport tasks that remain alive after the client finishes its check. + drop(stdin); + drop(stdout); + let status = match tokio::time::timeout(Duration::from_secs(1), child.wait()).await { + Ok(result) => Some(result.expect("wait for ACP server")), + Err(_) => { + child.kill().await.expect("stop hung ACP server"); + let _ = child.wait().await.expect("reap ACP server"); + None + } + }; + let stderr = stderr_reader.await.expect("join ACP stderr reader"); + + if let Some(status) = status { + assert!( + status.success(), + "ACP server exited unsuccessfully: {stderr}" + ); + } +} diff --git a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs b/src/crates/assembly/core/src/agentic/coordination/scheduler.rs index 6c933feea1..44e758a5c1 100644 --- a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs +++ b/src/crates/assembly/core/src/agentic/coordination/scheduler.rs @@ -787,6 +787,7 @@ impl DialogScheduler { delivery.session_id.clone(), resolved_turn_id.clone(), queued_turn, + false, ) .await; if result.is_err() { @@ -925,7 +926,7 @@ impl DialogScheduler { enqueued_at: SystemTime::now(), execution: QueuedTurnExecution::Standard, }; - self.submit_queued_turn(session_id, resolved_turn_id, queued_turn) + self.submit_queued_turn(session_id, resolved_turn_id, queued_turn, false) .await } @@ -979,8 +980,13 @@ impl DialogScheduler { }), }; - self.submit_queued_turn(session_id.clone(), resolved_turn_id.clone(), queued_turn) - .await?; + self.submit_queued_turn( + session_id.clone(), + resolved_turn_id.clone(), + queued_turn, + false, + ) + .await?; Ok(HiddenSubagentSubmitResult { receiver: result_rx, cancel_handle: HiddenSubagentQueueCancelHandle { @@ -1070,9 +1076,10 @@ impl DialogScheduler { session_id: String, resolved_turn_id: String, queued_turn: QueuedTurn, + reject_if_busy: bool, ) -> Result { let _operation_guard = self.lock_session_operation(&session_id).await; - self.submit_queued_turn_locked(session_id, resolved_turn_id, queued_turn) + self.submit_queued_turn_locked(session_id, resolved_turn_id, queued_turn, reject_if_busy) .await } @@ -1081,6 +1088,7 @@ impl DialogScheduler { session_id: String, resolved_turn_id: String, queued_turn: QueuedTurn, + reject_if_busy: bool, ) -> Result { if let Some(workspace_path) = queued_turn.workspace_path.as_deref() { let requested_storage_path = Self::resolve_session_restore_path( @@ -1110,6 +1118,16 @@ impl DialogScheduler { policy: queued_turn.policy, }); + if reject_if_busy + && matches!( + action, + DialogSubmitQueueAction::EnqueueThenStartNext + | DialogSubmitQueueAction::EnqueueForActiveTurn + ) + { + return Err("Session state does not allow starting new dialog: Processing".to_string()); + } + match action { DialogSubmitQueueAction::StartImmediately => { let tid = self.start_turn(&session_id, &queued_turn).await?; @@ -2279,11 +2297,19 @@ fn agent_dialog_turn_prepended_messages( .collect() } -#[async_trait::async_trait] -impl AgentDialogTurnPort for DialogScheduler { - async fn submit_dialog_turn( +impl DialogScheduler { + pub(crate) async fn submit_agent_dialog_turn_reject_if_busy( &self, request: AgentDialogTurnRequest, + ) -> PortResult { + self.submit_agent_dialog_turn_with_busy_policy(request, true) + .await + } + + async fn submit_agent_dialog_turn_with_busy_policy( + &self, + request: AgentDialogTurnRequest, + reject_if_busy: bool, ) -> PortResult { let image_contexts = agent_dialog_turn_image_contexts(&request.attachments)?; let prepended_messages = @@ -2293,27 +2319,48 @@ impl AgentDialogTurnPort for DialogScheduler { } else { Some(serde_json::Value::Object(request.metadata)) }; - - self.submit_with_prepended_messages( - request.session_id, - request.message, - request.original_message, - request.turn_id, - request.agent_type, - request.workspace_path, - request.remote_connection_id, - request.remote_ssh_host, - request.policy, - request.reply_route, - user_message_metadata, + let resolved_turn_id = request + .turn_id + .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()); + let queued_turn = QueuedTurn { + user_input: request.message, + original_user_input: request.original_message, prepended_messages, + turn_id: Some(resolved_turn_id.clone()), + agent_type: request.agent_type, + workspace_path: request.workspace_path, + remote_connection_id: request.remote_connection_id, + remote_ssh_host: request.remote_ssh_host, + policy: request.policy, + reply_route: request.reply_route, + user_message_metadata, image_contexts, + enqueued_at: SystemTime::now(), + execution: QueuedTurnExecution::Standard, + }; + + self.submit_queued_turn( + request.session_id, + resolved_turn_id, + queued_turn, + reject_if_busy, ) .await .map_err(|error| PortError::new(PortErrorKind::Backend, error)) } } +#[async_trait::async_trait] +impl AgentDialogTurnPort for DialogScheduler { + async fn submit_dialog_turn( + &self, + request: AgentDialogTurnRequest, + ) -> PortResult { + self.submit_agent_dialog_turn_with_busy_policy(request, false) + .await + } +} + #[async_trait::async_trait] impl AgentLifecycleDeliveryPort for DialogScheduler { async fn deliver_background_result( @@ -3051,6 +3098,66 @@ mod tests { ); } + #[tokio::test] + async fn reject_busy_dialog_port_does_not_enqueue_or_replace_the_active_turn() { + let (scheduler, session_manager, _, root) = test_scheduler(); + let session_id = "acp-session"; + let workspace = root.path().join("workspace"); + std::fs::create_dir_all(&workspace).expect("workspace"); + session_manager + .create_session_with_id( + Some(session_id.to_string()), + "ACP".to_string(), + "agentic".to_string(), + SessionConfig { + workspace_path: Some(workspace.to_string_lossy().to_string()), + ..Default::default() + }, + ) + .await + .expect("create ACP session"); + session_manager + .update_session_state( + session_id, + SessionState::Processing { + current_turn_id: "active-turn".to_string(), + phase: ProcessingPhase::Thinking, + }, + ) + .await + .expect("mark active turn"); + + let error = scheduler + .submit_agent_dialog_turn_reject_if_busy(AgentDialogTurnRequest { + session_id: session_id.to_string(), + message: "second prompt".to_string(), + original_message: None, + turn_id: Some("rejected-turn".to_string()), + agent_type: "agentic".to_string(), + workspace_path: None, + remote_connection_id: None, + remote_ssh_host: None, + policy: DialogSubmissionPolicy::for_source(DialogTriggerSource::Cli), + reply_route: None, + prepended_reminders: Vec::new(), + attachments: Vec::new(), + metadata: serde_json::Map::new(), + }) + .await + .expect_err("busy ACP prompt must be rejected"); + + assert_eq!(error.kind, PortErrorKind::Backend); + assert!(error.message.contains("Processing"), "{error}"); + assert_eq!(scheduler.queue_depth(session_id), 0); + assert!(matches!( + session_manager + .get_session(session_id) + .expect("session") + .state, + SessionState::Processing { current_turn_id, .. } if current_turn_id == "active-turn" + )); + } + #[test] fn background_result_injection_preserves_the_exact_task_id() { assert_eq!( diff --git a/src/crates/assembly/core/src/product_runtime.rs b/src/crates/assembly/core/src/product_runtime.rs index 48acdaa6d9..ccd18c29c8 100644 --- a/src/crates/assembly/core/src/product_runtime.rs +++ b/src/crates/assembly/core/src/product_runtime.rs @@ -9,7 +9,7 @@ mod runtime_services; use std::path::{Path, PathBuf}; use std::sync::Arc; -use bitfun_agent_runtime::sdk::AgentRuntime; +use bitfun_agent_runtime::sdk::{AgentEventSource, AgentRuntime}; use bitfun_harness::HarnessRegistry; use bitfun_runtime_ports::{SessionStoragePathRequest, SessionStorePort, SessionViewRestoreTiming}; use bitfun_runtime_services::RuntimeServices; @@ -89,6 +89,24 @@ impl CoreProductAgentRuntime { harness_registry, ) } + + /// Build the ACP surface with its protocol requirement that a session + /// rejects a second prompt while another turn is active. + pub fn build_acp( + coordinator: Arc, + scheduler: Arc, + event_source: AgentEventSource, + services: RuntimeServices, + harness_registry: HarnessRegistry, + ) -> Result { + CoreServiceAgentRuntime::acp_product_agent_runtime( + coordinator, + scheduler, + event_source, + services, + harness_registry, + ) + } } /// Core-owned compatibility boundary for product operations not yet exposed by @@ -140,6 +158,16 @@ impl CoreAgentRuntimeCompatibility { .await } + pub async fn update_session_agent_type( + &self, + session_id: &str, + agent_type: &str, + ) -> BitFunResult<()> { + self.coordinator + .update_session_agent_type(session_id, agent_type) + .await + } + pub async fn create_session_with_workspace( &self, session_id: Option, @@ -648,6 +676,7 @@ mod tests { } let _ = build; + let _ = CoreProductAgentRuntime::build_acp; } #[test] @@ -666,6 +695,7 @@ mod tests { let _ = CoreAgentRuntimeCompatibility::generate_session_usage_report; let _ = CoreAgentRuntimeCompatibility::list_persisted_sessions; let _ = CoreAgentRuntimeCompatibility::load_persisted_session_turns; + let _ = CoreAgentRuntimeCompatibility::update_session_agent_type; let _ = CoreAgentRuntimeCompatibility::is_turn_processing; } diff --git a/src/crates/assembly/core/src/service_agent_runtime.rs b/src/crates/assembly/core/src/service_agent_runtime.rs index ec98e6a702..c23aa58274 100644 --- a/src/crates/assembly/core/src/service_agent_runtime.rs +++ b/src/crates/assembly/core/src/service_agent_runtime.rs @@ -6,8 +6,8 @@ //! implementations until a reviewed port/provider migration proves equivalence. use bitfun_agent_runtime::sdk::{ - AgentInteractionResponsePort, AgentRuntime, AgentRuntimeBuilder, AgentSessionRestorePort, - RuntimeError, + AgentEventSource, AgentInteractionResponsePort, AgentRuntime, AgentRuntimeBuilder, + AgentSessionRestorePort, RuntimeError, }; use bitfun_runtime_ports::{ AgentDialogTurnPort, AgentDialogTurnRequest, AgentInputAttachment, AgentLifecycleDeliveryPort, @@ -426,6 +426,23 @@ impl ScheduledSessionManagementPort { } } +/// ACP accepts one prompt at a time per session. Keep that protocol-specific +/// admission rule in the product assembly instead of changing the shared +/// scheduler policy used by GUI, TUI, and remote-control surfaces. +struct RejectBusyAgentDialogTurnPort(Arc); + +#[async_trait::async_trait] +impl AgentDialogTurnPort for RejectBusyAgentDialogTurnPort { + async fn submit_dialog_turn( + &self, + request: AgentDialogTurnRequest, + ) -> bitfun_runtime_ports::PortResult { + self.0 + .submit_agent_dialog_turn_reject_if_busy(request) + .await + } +} + #[async_trait::async_trait] impl AgentSessionManagementPort for ScheduledSessionManagementPort { async fn list_sessions( @@ -893,6 +910,44 @@ impl CoreServiceAgentRuntime { scheduler: Arc, services: bitfun_runtime_services::RuntimeServices, harness_registry: bitfun_harness::HarnessRegistry, + ) -> Result { + let dialog_turn: Arc = scheduler.clone(); + Self::product_agent_runtime_with_dialog_turn( + coordinator, + scheduler, + dialog_turn, + None, + services, + harness_registry, + ) + } + + pub(crate) fn acp_product_agent_runtime( + coordinator: Arc, + scheduler: Arc, + event_source: AgentEventSource, + services: bitfun_runtime_services::RuntimeServices, + harness_registry: bitfun_harness::HarnessRegistry, + ) -> Result { + let dialog_turn: Arc = + Arc::new(RejectBusyAgentDialogTurnPort(scheduler.clone())); + Self::product_agent_runtime_with_dialog_turn( + coordinator, + scheduler, + dialog_turn, + Some(event_source), + services, + harness_registry, + ) + } + + fn product_agent_runtime_with_dialog_turn( + coordinator: Arc, + scheduler: Arc, + dialog_turn: Arc, + event_source: Option, + services: bitfun_runtime_services::RuntimeServices, + harness_registry: bitfun_harness::HarnessRegistry, ) -> Result { let submission: Arc = coordinator.clone(); let session_management = @@ -903,10 +958,9 @@ impl CoreServiceAgentRuntime { let thread_goal_management: Arc = coordinator.clone(); let interaction_response: Arc = coordinator; let cancellation: Arc = scheduler.clone(); - let dialog_turn: Arc = scheduler.clone(); let lifecycle_delivery: Arc = scheduler; - core_agent_runtime_builder( + let builder = core_agent_runtime_builder( submission, session_management, session_restore, @@ -916,11 +970,16 @@ impl CoreServiceAgentRuntime { interaction_response, ) .with_dialog_turn_port(dialog_turn) - .with_lifecycle_delivery_port(lifecycle_delivery) - .with_services(services) - .with_harness_registry(Arc::new(harness_registry)) - .build() - .map_err(|error| error.to_string()) + .with_lifecycle_delivery_port(lifecycle_delivery); + let builder = match event_source { + Some(event_source) => builder.with_event_source(event_source), + None => builder, + }; + builder + .with_services(services) + .with_harness_registry(Arc::new(harness_registry)) + .build() + .map_err(|error| error.to_string()) } pub(crate) fn global_agent_runtime_with_lifecycle_delivery() -> Result { diff --git a/src/crates/execution/agent-runtime/src/event_queue.rs b/src/crates/execution/agent-runtime/src/event_queue.rs index 72cbf3b28a..53b5ca667b 100644 --- a/src/crates/execution/agent-runtime/src/event_queue.rs +++ b/src/crates/execution/agent-runtime/src/event_queue.rs @@ -6,13 +6,83 @@ use bitfun_events::{ AgenticEvent, AgenticEventEnvelope as EventEnvelope, AgenticEventPriority as EventPriority, }; use log::{debug, trace, warn}; -use std::collections::BinaryHeap; -use std::sync::Arc; +use std::collections::{BinaryHeap, HashMap}; +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, RwLock as StdRwLock, Weak, +}; use tokio::sync::{broadcast, Mutex, Notify}; const MIN_EVENT_BROADCAST_BUFFER: usize = 1024; +// Session-scoped protocol consumers can pause while servicing an RPC. Keep a +// separate, fixed burst budget instead of inheriting the much larger global +// event queue capacity for every active session. +const SESSION_EVENT_BROADCAST_BUFFER: usize = MIN_EVENT_BROADCAST_BUFFER; const SLOW_EVENT_QUEUE_LATENCY_MS: u128 = 250; +struct SessionBroadcast { + sender: broadcast::Sender, +} + +type SessionBroadcastMap = HashMap>; + +/// Receiver for one session's bounded event stream. +/// +/// Dropping the last receiver removes the channel immediately so inactive +/// sessions retain neither their broadcast buffer nor an entry on the enqueue +/// path. +pub struct SessionEventReceiver { + receiver: Option>, + channel: Weak, + session_id: String, + channels: Weak>, + has_channels: Weak, +} + +impl std::fmt::Debug for SessionEventReceiver { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SessionEventReceiver") + .field("session_id", &self.session_id) + .finish_non_exhaustive() + } +} + +impl SessionEventReceiver { + pub async fn recv(&mut self) -> Result { + self.receiver + .as_mut() + .expect("session event receiver used after drop") + .recv() + .await + } +} + +impl Drop for SessionEventReceiver { + fn drop(&mut self) { + // Decrement the Tokio receiver count before deciding whether this + // channel still has an active consumer. + drop(self.receiver.take()); + + let (Some(channels), Some(channel), Some(has_channels)) = ( + self.channels.upgrade(), + self.channel.upgrade(), + self.has_channels.upgrade(), + ) else { + return; + }; + let mut channels = channels + .write() + .expect("session event channels lock poisoned"); + let remove = channels.get(&self.session_id).is_some_and(|current| { + Arc::ptr_eq(current, &channel) && current.sender.receiver_count() == 0 + }); + if remove { + channels.remove(&self.session_id); + } + has_channels.store(!channels.is_empty(), Ordering::Release); + } +} + /// Event queue configuration #[derive(Debug, Clone)] pub struct EventQueueConfig { @@ -53,6 +123,14 @@ pub struct EventQueue { /// Broadcast stream for non-consuming subscribers. broadcast_tx: broadcast::Sender, + /// Session-scoped streams for protocol consumers that require isolation + /// from traffic produced by other sessions. + session_broadcasts: Arc>, + + /// Avoid locking the session channel map on the common TUI/GUI path when + /// no protocol consumer is active. + has_session_broadcasts: Arc, + /// Configuration config: EventQueueConfig, @@ -71,6 +149,8 @@ impl EventQueue { queue: Arc::new(Mutex::new(BinaryHeap::new())), notify: Arc::new(Notify::new()), broadcast_tx, + session_broadcasts: Arc::new(StdRwLock::new(HashMap::new())), + has_session_broadcasts: Arc::new(AtomicBool::new(false)), config, stats: Arc::new(Mutex::new(QueueStats::default())), } @@ -102,7 +182,22 @@ impl EventQueue { // Broadcast delivery is authoritative for non-consuming runtime // subscribers and must not depend on capacity in the legacy dequeue - // buffer. + // buffer. Session-scoped subscribers receive only matching traffic so + // an unrelated session cannot consume their bounded backlog. + let session_channel = if self.has_session_broadcasts.load(Ordering::Acquire) { + envelope.event.session_id().and_then(|session_id| { + self.session_broadcasts + .read() + .expect("session event channels lock poisoned") + .get(session_id) + .cloned() + }) + } else { + None + }; + if let Some(channel) = session_channel { + let _ = channel.sender.send(envelope.clone()); + } let _ = self.broadcast_tx.send(envelope); { @@ -182,6 +277,61 @@ impl EventQueue { self.broadcast_tx.subscribe() } + /// Subscribe to one session before events enter the bounded receiver. + pub fn subscribe_session(&self, session_id: &str) -> SessionEventReceiver { + let existing = { + let channels = self + .session_broadcasts + .read() + .expect("session event channels lock poisoned"); + channels.get(session_id).map(|channel| { + // Subscribe while holding the read lock so the last previous + // receiver cannot remove this channel between lookup and the + // Tokio receiver count increment. + (channel.clone(), channel.sender.subscribe()) + }) + }; + if let Some((channel, receiver)) = existing { + return self.session_receiver(session_id, &channel, receiver); + } + + // Allocate outside the map lock. The second check below resolves a + // concurrent first subscriber without holding the global enqueue path + // while Tokio allocates the bounded channel. + let (sender, candidate_receiver) = broadcast::channel(SESSION_EVENT_BROADCAST_BUFFER); + let candidate = Arc::new(SessionBroadcast { sender }); + let (channel, receiver) = { + let mut channels = self + .session_broadcasts + .write() + .expect("session event channels lock poisoned"); + if let Some(channel) = channels.get(session_id).cloned() { + let receiver = channel.sender.subscribe(); + (channel, receiver) + } else { + channels.insert(session_id.to_string(), candidate.clone()); + self.has_session_broadcasts.store(true, Ordering::Release); + (candidate, candidate_receiver) + } + }; + self.session_receiver(session_id, &channel, receiver) + } + + fn session_receiver( + &self, + session_id: &str, + channel: &Arc, + receiver: broadcast::Receiver, + ) -> SessionEventReceiver { + SessionEventReceiver { + receiver: Some(receiver), + channel: Arc::downgrade(channel), + session_id: session_id.to_string(), + channels: Arc::downgrade(&self.session_broadcasts), + has_channels: Arc::downgrade(&self.has_session_broadcasts), + } + } + /// Clear all events for a session pub async fn clear_session(&self, session_id: &str) -> EventBusResult<()> { // Remove all events for this session from the queue @@ -354,4 +504,15 @@ mod tests { } assert_eq!(first_ids, second_ids); } + + #[test] + fn dropping_last_session_receiver_releases_its_channel() { + let queue = EventQueue::new(EventQueueConfig::default()); + let receiver = queue.subscribe_session("session"); + assert_eq!(queue.session_broadcasts.read().unwrap().len(), 1); + + drop(receiver); + + assert!(queue.session_broadcasts.read().unwrap().is_empty()); + } } diff --git a/src/crates/execution/agent-runtime/src/event_source.rs b/src/crates/execution/agent-runtime/src/event_source.rs new file mode 100644 index 0000000000..d62d472187 --- /dev/null +++ b/src/crates/execution/agent-runtime/src/event_source.rs @@ -0,0 +1,222 @@ +//! Shared read-only access to product agent events. + +use std::sync::Arc; + +use bitfun_events::AgenticEventEnvelope; +use tokio::sync::broadcast; + +use crate::event_queue::{EventQueue, SessionEventReceiver}; + +/// Cloneable source for subscribing to the runtime's existing agent event queue. +/// +/// This source is read-only: queue draining and task lifecycle remain owned by +/// the product host. Global receivers use the existing bounded broadcast; +/// session receivers use a scoped bounded channel owned by the same queue. +/// Neither path creates a forwarding task or a second event schema. +#[derive(Clone)] +pub struct AgentEventSource { + queue: Arc, +} + +impl std::fmt::Debug for AgentEventSource { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AgentEventSource").finish_non_exhaustive() + } +} + +impl AgentEventSource { + pub fn new(queue: Arc) -> Self { + Self { queue } + } + + pub fn subscribe(&self) -> AgentEventReceiver { + self.queue.subscribe() + } + + pub fn subscribe_session(&self, session_id: &str) -> AgentSessionEventReceiver { + self.queue.subscribe_session(session_id) + } +} + +/// Existing Tokio receiver types keep TUI `recv` and `try_recv` behavior +/// unchanged and avoid a second adapter layer. +pub type AgentEventReceiver = broadcast::Receiver; + +/// A bounded receiver that contains events for exactly one session and +/// releases its channel when the final subscriber is dropped. +pub type AgentSessionEventReceiver = SessionEventReceiver; + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use bitfun_events::AgenticEvent; + use tokio::sync::broadcast::error::RecvError; + + use crate::event_queue::{EventQueue, EventQueueConfig}; + + use super::AgentEventSource; + + #[test] + fn source_construction_does_not_require_a_tokio_runtime() { + let queue = Arc::new(EventQueue::new(EventQueueConfig::default())); + let source = AgentEventSource::new(queue); + let _receiver = source.subscribe(); + } + + #[tokio::test] + async fn clones_preserve_broadcast_order() { + let queue = Arc::new(EventQueue::new(EventQueueConfig { + max_queue_size: 32, + batch_size: 4, + })); + let source = AgentEventSource::new(queue.clone()); + let clone = source.clone(); + + let mut first = source.subscribe(); + let mut second = clone.subscribe(); + for index in 0..32 { + queue + .enqueue( + AgenticEvent::SessionStateChanged { + session_id: "session".to_string(), + new_state: index.to_string(), + }, + None, + ) + .await + .expect("event should enqueue"); + } + + for expected in 0..32 { + let first = first.recv().await.expect("first event"); + let second = second.recv().await.expect("second event"); + assert_eq!(first.id, second.id); + assert!(matches!( + first.event, + AgenticEvent::SessionStateChanged { ref new_state, .. } + if new_state == &expected.to_string() + )); + } + } + + #[tokio::test] + async fn source_retains_default_sized_burst() { + let queue = Arc::new(EventQueue::new(EventQueueConfig::default())); + let source = AgentEventSource::new(queue.clone()); + let mut receiver = source.subscribe(); + + for index in 0..2048 { + queue + .enqueue( + AgenticEvent::SessionStateChanged { + session_id: "session".to_string(), + new_state: index.to_string(), + }, + None, + ) + .await + .expect("event should enqueue"); + } + + for expected in 0..2048 { + let envelope = receiver.recv().await.expect("burst event"); + assert!(matches!( + envelope.event, + AgenticEvent::SessionStateChanged { ref new_state, .. } + if new_state == &expected.to_string() + )); + } + } + + #[tokio::test] + async fn receiver_reports_lag_and_closed_source() { + let queue = Arc::new(EventQueue::new(EventQueueConfig { + max_queue_size: 1, + batch_size: 1, + })); + let source = AgentEventSource::new(queue.clone()); + let mut receiver = source.subscribe(); + + for index in 0..1025 { + queue + .enqueue( + AgenticEvent::SessionStateChanged { + session_id: "session".to_string(), + new_state: index.to_string(), + }, + None, + ) + .await + .expect("event should enqueue"); + } + assert_eq!(receiver.recv().await, Err(RecvError::Lagged(1))); + + drop(source); + drop(queue); + while receiver.recv().await.is_ok() {} + assert_eq!(receiver.recv().await, Err(RecvError::Closed)); + } + + #[tokio::test] + async fn scoped_receiver_ignores_other_session_backlog() { + let queue = Arc::new(EventQueue::new(EventQueueConfig { + max_queue_size: 1, + batch_size: 1, + })); + let source = AgentEventSource::new(queue.clone()); + let mut receiver = source.subscribe_session("target-session"); + + for index in 0..1025 { + queue + .enqueue( + AgenticEvent::SessionStateChanged { + session_id: "other-session".to_string(), + new_state: index.to_string(), + }, + None, + ) + .await + .expect("unrelated event should enqueue"); + } + queue + .enqueue( + AgenticEvent::SessionStateChanged { + session_id: "target-session".to_string(), + new_state: "ready".to_string(), + }, + None, + ) + .await + .expect("target event should enqueue"); + + let envelope = receiver.recv().await.expect("target event"); + assert!(matches!( + envelope.event, + AgenticEvent::SessionStateChanged { ref session_id, ref new_state } + if session_id == "target-session" && new_state == "ready" + )); + } + + #[tokio::test] + async fn scoped_receiver_has_an_independent_bounded_budget() { + let queue = Arc::new(EventQueue::new(EventQueueConfig::default())); + let source = AgentEventSource::new(queue.clone()); + let mut receiver = source.subscribe_session("target-session"); + + for index in 0..1025 { + queue + .enqueue( + AgenticEvent::SessionStateChanged { + session_id: "target-session".to_string(), + new_state: index.to_string(), + }, + None, + ) + .await + .expect("target event should enqueue"); + } + + assert_eq!(receiver.recv().await, Err(RecvError::Lagged(1))); + } +} diff --git a/src/crates/execution/agent-runtime/src/lib.rs b/src/crates/execution/agent-runtime/src/lib.rs index ae0ef164a7..a462a8d7b7 100644 --- a/src/crates/execution/agent-runtime/src/lib.rs +++ b/src/crates/execution/agent-runtime/src/lib.rs @@ -14,6 +14,7 @@ pub mod dialog_turn; pub mod event_bus; pub mod event_queue; pub mod event_router; +pub mod event_source; pub mod events; pub mod evidence_ledger; pub mod file_read_state; diff --git a/src/crates/execution/agent-runtime/src/runtime.rs b/src/crates/execution/agent-runtime/src/runtime.rs index 080dc27163..01ef50157b 100644 --- a/src/crates/execution/agent-runtime/src/runtime.rs +++ b/src/crates/execution/agent-runtime/src/runtime.rs @@ -24,6 +24,7 @@ use bitfun_runtime_ports::{ }; use bitfun_runtime_services::RuntimeServices; +use crate::event_source::{AgentEventReceiver, AgentEventSource, AgentSessionEventReceiver}; use crate::post_call_hooks::RuntimeHookRegistry; #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] @@ -54,6 +55,8 @@ pub enum RuntimeError { MissingInteractionResponsePort, #[error("runtime event sink is not registered")] MissingEventSink, + #[error("agent event source is not registered")] + MissingEventSource, #[error(transparent)] Port(#[from] PortError), } @@ -190,6 +193,7 @@ pub struct AgentRuntime { interaction_response: Option>, services: Option, event_stream: Option, + event_source: Option, tool_registry: Option>, harness_registry: Option>, hook_registry: RuntimeHookRegistry, @@ -265,6 +269,10 @@ impl std::fmt::Debug for AgentRuntime { "event_stream", &self.event_stream.as_ref().map(|_| ""), ) + .field( + "event_source", + &self.event_source.as_ref().map(|_| ""), + ) .field( "tool_registry", &self.tool_registry.as_ref().map(|_| ""), @@ -312,6 +320,7 @@ pub struct AgentRuntimeBuilder { interaction_response: Option>, services: Option, event_stream: Option, + event_source: Option, tool_registry: Option>, harness_registry: Option>, hook_registry: RuntimeHookRegistry, @@ -394,6 +403,11 @@ impl AgentRuntimeBuilder { self } + pub fn with_event_source(mut self, source: AgentEventSource) -> Self { + self.event_source = Some(source); + self + } + pub fn with_tool_registry(mut self, registry: Arc) -> Self { self.tool_registry = Some(registry); self @@ -432,6 +446,7 @@ impl AgentRuntimeBuilder { interaction_response, services, event_stream, + event_source, tool_registry, harness_registry, hook_registry, @@ -455,6 +470,7 @@ impl AgentRuntimeBuilder { interaction_response, services, event_stream, + event_source, tool_registry, harness_registry, hook_registry, @@ -561,6 +577,23 @@ pub struct AgentRunHandle { } impl AgentRuntime { + pub fn subscribe_events(&self) -> Result { + self.event_source + .as_ref() + .map(AgentEventSource::subscribe) + .ok_or(RuntimeError::MissingEventSource) + } + + pub fn subscribe_session_events( + &self, + session_id: &str, + ) -> Result { + self.event_source + .as_ref() + .map(|source| source.subscribe_session(session_id)) + .ok_or(RuntimeError::MissingEventSource) + } + pub fn services(&self) -> Option<&RuntimeServices> { self.services.as_ref() } @@ -1631,6 +1664,30 @@ mod tests { assert_eq!(error, RuntimeError::MissingSessionRestorePort); } + #[test] + fn event_subscription_requires_configured_source() { + let ports = Arc::new(FakeAgentRuntimePorts::default()); + let runtime = AgentRuntimeBuilder::new() + .with_submission_port(ports) + .build() + .expect("runtime"); + + assert_eq!( + runtime.subscribe_events().unwrap_err(), + RuntimeError::MissingEventSource + ); + } + + #[test] + fn runtime_error_message_preserves_port_error_text() { + let error = RuntimeError::Port(PortError::new( + bitfun_runtime_ports::PortErrorKind::Backend, + "original backend message", + )); + + assert_eq!(error.into_message(), "original backend message"); + } + #[tokio::test] async fn transcript_read_requires_registered_reader() { let ports = Arc::new(FakeAgentRuntimePorts::default()); diff --git a/src/crates/execution/agent-runtime/src/sdk.rs b/src/crates/execution/agent-runtime/src/sdk.rs index 6f21e077c8..f2ffaec2b3 100644 --- a/src/crates/execution/agent-runtime/src/sdk.rs +++ b/src/crates/execution/agent-runtime/src/sdk.rs @@ -35,6 +35,7 @@ impl AgentRuntimeSdkCompatibility { } pub use crate::context_profile::{ContextProfile, ContextProfilePolicy, ModelCapabilityProfile}; +pub use crate::event_source::{AgentEventReceiver, AgentEventSource, AgentSessionEventReceiver}; pub use crate::post_call_hooks::{ RuntimeHookErrorPolicy, RuntimeHookKind, RuntimeHookPlan, RuntimeHookRegistry, RuntimeHookRegistryBuildError, @@ -61,10 +62,10 @@ pub use bitfun_runtime_ports::{ AgentSubmissionResult, AgentSubmissionSource, AgentThreadGoalCreateRequest, AgentThreadGoalDeliveryRequest, AgentThreadGoalGetRequest, AgentThreadGoalManagementPort, AgentThreadGoalUpdateStatusRequest, AgentTurnCancellationPort, AgentTurnCancellationRequest, - AgentTurnCancellationResult, ClockPort, DialogSubmitOutcome, FileSystemPort, GitPort, - McpCatalogPort, NetworkPort, PermissionDecision, PermissionPort, PermissionRequest, PortError, - PortResult, RemoteAssistantWorkspaceFacts, RemoteCapabilityPort, RemoteConnectionPort, - RemoteProjectionPort, RemoteRecentWorkspaceFacts, RemoteWorkspaceFacts, + AgentTurnCancellationResult, ClockPort, DialogSubmissionPolicy, DialogSubmitOutcome, + FileSystemPort, GitPort, McpCatalogPort, NetworkPort, PermissionDecision, PermissionPort, + PermissionRequest, PortError, PortResult, RemoteAssistantWorkspaceFacts, RemoteCapabilityPort, + RemoteConnectionPort, RemoteProjectionPort, RemoteRecentWorkspaceFacts, RemoteWorkspaceFacts, RemoteWorkspaceFileRuntimeHost, RemoteWorkspaceKind, RemoteWorkspacePort, RemoteWorkspaceRuntimeHost, RemoteWorkspaceUpdate, RuntimeEventEnvelope, RuntimeEventSink, RuntimeEventType, RuntimeServiceCapability, RuntimeServicePort, SessionStorageKind, @@ -170,6 +171,11 @@ impl AgentRuntimeBuilder { self } + pub fn with_event_source(mut self, source: AgentEventSource) -> Self { + self.inner = self.inner.with_event_source(source); + self + } + pub fn with_tool_registry(mut self, registry: Arc) -> Self { self.inner = self.inner.with_tool_registry(registry); self @@ -196,6 +202,17 @@ impl AgentRuntimeBuilder { } impl AgentRuntime { + pub fn subscribe_events(&self) -> Result { + self.inner.subscribe_events() + } + + pub fn subscribe_session_events( + &self, + session_id: &str, + ) -> Result { + self.inner.subscribe_session_events(session_id) + } + pub fn services(&self) -> Option<&RuntimeServices> { self.inner.services() } diff --git a/src/crates/interfaces/acp/AGENTS-CN.md b/src/crates/interfaces/acp/AGENTS-CN.md index 80467b6825..637fe64e57 100644 --- a/src/crates/interfaces/acp/AGENTS-CN.md +++ b/src/crates/interfaces/acp/AGENTS-CN.md @@ -6,14 +6,18 @@ `bitfun-acp` 负责基于已组装产品 runtime 的 Agent Client Protocol 入口与 ACP client 行为。ACP protocol / client 细节留在这里或应用入口 adapter 中；跨层只共享稳定 capability facts。 -当前生产代码仍直接依赖 `bitfun-core` 的 `product-full`，尚未消费 `DeliveryProfile::Acp` 或 -`ProductAssembler`。“基于已组装 runtime”是目标边界；迁移入口时必须保持现有 ACP 行为等价。 +CLI 托管的 ACP 服务端已通过 `ProductAssembler` 消费 `DeliveryProfile::Acp`，并使用 Agent Runtime SDK +完成会话创建/列举、轮次提交/取消、交互响应和 Agent 事件订阅。`bitfun-acp` 仍直接依赖 +`bitfun-core/product-full`，用于一次性恢复完整持久化历史、模型/模式配置、模型目录与配置读取、MCP 配置， +以及本 crate 的 ACP 客户端路径。在这些生产路径分别获得可移植替代并证明等价前，不得宣称整个 crate +已与 Core 解耦。 ## 护栏 - Remote ACP workspace 复用本地 ACP client 配置。修改 ACP client 行为时，必须保持 manager、remote shell probing、remote capability store 和 workspace menu availability 语义。 - ACP config persistence、remote probing、timeout policy 和 workspace surface selection 属于 ACP / app-surface 行为，不要移动到 `core-types`、`runtime-ports` 或 `agent-tools`。 - ACP external-agent tool 的命名、schema、validation、presentation 和 result shape 属于 `bitfun-agent-tools` 的 portable contract；ACP 应调用这些 helper，不要在本层重复定义。 +- ACP 标准输入输出、连接管理和协议通知投影留在本 crate。共享运行时事实可以经过 SDK 边界；ACP 协议请求、客户端选择和生命周期状态不得进入 SDK。 - 如果未来需要 contract，只表达观测事实：environment identity、capability facts、request / response DTO。 ## 验证 diff --git a/src/crates/interfaces/acp/AGENTS.md b/src/crates/interfaces/acp/AGENTS.md index 1fc5343b98..140a5bb436 100644 --- a/src/crates/interfaces/acp/AGENTS.md +++ b/src/crates/interfaces/acp/AGENTS.md @@ -8,10 +8,14 @@ Scope: this guide applies to `src/crates/interfaces/acp`. runtime. Keep ACP protocol/client details here or in app-surface adapters; share only stable capability facts through contract crates. -Current production code still depends directly on `bitfun-core` with -`product-full`; it does not yet consume `DeliveryProfile::Acp` or -`ProductAssembler`. Treat the assembled-runtime wording as the target boundary, -and preserve existing ACP behavior while migrating that entry explicitly. +The CLI-hosted ACP server consumes `DeliveryProfile::Acp` through +`ProductAssembler` and uses the Agent Runtime SDK for session creation/listing, +dialog submission/cancellation, interaction responses, and agent event +subscription. `bitfun-acp` still depends directly on `bitfun-core` with +`product-full` for single-pass full persisted-history restore, model/mode configuration, +model catalog/config reads, MCP provisioning, and the ACP client half of this +crate. Do not describe the crate as Core-independent until those production +paths have separately proven portable replacements. ## Guardrails @@ -24,6 +28,9 @@ and preserve existing ACP behavior while migrating that entry explicitly. - ACP external-agent tool naming, schema, validation, presentation, and result shape are portable contracts owned by `bitfun-agent-tools`; ACP should call those helpers instead of redefining them locally. +- Keep ACP stdio/connection ownership and protocol notification projection in + this crate. Shared runtime facts may cross the SDK boundary; ACP protocol + requests, client choices, and lifecycle state may not. - If a future contract is needed, make it observational: environment identity, capability facts, and request/response DTOs only. diff --git a/src/crates/interfaces/acp/Cargo.toml b/src/crates/interfaces/acp/Cargo.toml index 91afe13e25..a4eee58505 100644 --- a/src/crates/interfaces/acp/Cargo.toml +++ b/src/crates/interfaces/acp/Cargo.toml @@ -10,6 +10,7 @@ name = "bitfun_acp" [dependencies] bitfun-core = { path = "../../assembly/core", default-features = false, features = ["product-full"] } +bitfun-agent-runtime = { path = "../../execution/agent-runtime" } bitfun-agent-tools = { path = "../../execution/tool-contracts" } bitfun-events = { path = "../../contracts/events" } diff --git a/src/crates/interfaces/acp/src/runtime.rs b/src/crates/interfaces/acp/src/runtime.rs index 79276a81f3..c1893d0fc1 100644 --- a/src/crates/interfaces/acp/src/runtime.rs +++ b/src/crates/interfaces/acp/src/runtime.rs @@ -10,7 +10,8 @@ use agent_client_protocol::schema::{ }; use agent_client_protocol::{Client, ConnectionTo, Error, Result}; use async_trait::async_trait; -use bitfun_core::agentic::system::AgenticSystem; +use bitfun_agent_runtime::sdk::{AgentRuntime, RuntimeError}; +use bitfun_core::product_runtime::CoreAgentRuntimeCompatibility; use dashmap::DashMap; use crate::server::{AcpRuntime, AcpServer}; @@ -25,7 +26,8 @@ mod session; mod thinking; pub struct BitfunAcpRuntime { - pub(crate) agentic_system: AgenticSystem, + pub(crate) agent_runtime: AgentRuntime, + pub(crate) compatibility: CoreAgentRuntimeCompatibility, pub(crate) sessions: DashMap, pub(crate) connections: DashMap>, } @@ -42,16 +44,20 @@ pub(crate) struct AcpSessionState { } impl BitfunAcpRuntime { - pub fn new(agentic_system: AgenticSystem) -> Self { + pub fn new(agent_runtime: AgentRuntime, compatibility: CoreAgentRuntimeCompatibility) -> Self { Self { - agentic_system, + agent_runtime, + compatibility, sessions: DashMap::new(), connections: DashMap::new(), } } - pub async fn serve_stdio(agentic_system: AgenticSystem) -> Result<()> { - AcpServer::new(Arc::new(Self::new(agentic_system))) + pub async fn serve_stdio( + agent_runtime: AgentRuntime, + compatibility: CoreAgentRuntimeCompatibility, + ) -> Result<()> { + AcpServer::new(Arc::new(Self::new(agent_runtime, compatibility))) .serve_stdio() .await } @@ -59,6 +65,10 @@ impl BitfunAcpRuntime { pub(crate) fn internal_error(error: impl std::fmt::Display) -> Error { Error::internal_error().data(serde_json::json!(error.to_string())) } + + pub(crate) fn runtime_error(error: RuntimeError) -> Error { + Self::internal_error(error.into_message()) + } } #[async_trait] diff --git a/src/crates/interfaces/acp/src/runtime/content.rs b/src/crates/interfaces/acp/src/runtime/content.rs index 96b98ff380..3cc7c636be 100644 --- a/src/crates/interfaces/acp/src/runtime/content.rs +++ b/src/crates/interfaces/acp/src/runtime/content.rs @@ -2,18 +2,18 @@ use agent_client_protocol::schema::{ Annotations, BlobResourceContents, ContentBlock, EmbeddedResourceResource, ImageContent, ResourceLink, Role, TextResourceContents, }; -use bitfun_core::agentic::image_analysis::ImageContextData; +use bitfun_agent_runtime::sdk::AgentInputAttachment; pub(super) struct ParsedPrompt { pub(super) user_message: String, pub(super) original_user_message: Option, - pub(super) image_contexts: Vec, + pub(super) attachments: Vec, } pub(super) fn parse_prompt_blocks(session_id: &str, blocks: Vec) -> ParsedPrompt { let mut text_parts = Vec::new(); let mut original_text_parts = Vec::new(); - let mut image_contexts = Vec::new(); + let mut attachments = Vec::new(); for (index, block) in blocks.into_iter().enumerate() { match block { @@ -30,7 +30,7 @@ pub(super) fn parse_prompt_blocks(session_id: &str, blocks: Vec) - } if let Some(context) = image_to_context(session_id, index, image) { text_parts.push(format!("[Attached image: {}]", context.id)); - image_contexts.push(context); + attachments.push(context); } } ContentBlock::ResourceLink(link) => { @@ -52,7 +52,7 @@ pub(super) fn parse_prompt_blocks(session_id: &str, blocks: Vec) - blob_resource_to_image_context(session_id, index, &blob) { text_parts.push(format!("[Attached image resource: {}]", context.id)); - image_contexts.push(context); + attachments.push(context); } else { text_parts.push(blob_resource_text(&blob)); } @@ -88,7 +88,7 @@ pub(super) fn parse_prompt_blocks(session_id: &str, blocks: Vec) - ParsedPrompt { user_message, original_user_message, - image_contexts, + attachments, } } @@ -103,37 +103,33 @@ fn image_to_context( session_id: &str, index: usize, image: ImageContent, -) -> Option { +) -> Option { if image.data.trim().is_empty() { - return image.uri.clone().map(|uri| ImageContextData { - id: prompt_context_id(session_id, "image", index), - image_path: file_uri_to_path(&uri).or(Some(uri)), - data_url: None, - mime_type: image.mime_type, - metadata: Some(serde_json::json!({ - "source": "acp", - "uri": image.uri, - })), + return image.uri.clone().map(|uri| { + image_attachment( + prompt_context_id(session_id, "image", index), + file_uri_to_path(&uri).or(Some(uri)), + None, + image.mime_type, + serde_json::json!({ "source": "acp", "uri": image.uri }), + ) }); } - Some(ImageContextData { - id: prompt_context_id(session_id, "image", index), - image_path: None, - data_url: Some(format!("data:{};base64,{}", image.mime_type, image.data)), - mime_type: image.mime_type, - metadata: Some(serde_json::json!({ - "source": "acp", - "uri": image.uri, - })), - }) + Some(image_attachment( + prompt_context_id(session_id, "image", index), + None, + Some(format!("data:{};base64,{}", image.mime_type, image.data)), + image.mime_type, + serde_json::json!({ "source": "acp", "uri": image.uri }), + )) } fn blob_resource_to_image_context( session_id: &str, index: usize, blob: &BlobResourceContents, -) -> Option { +) -> Option { let mime_type = blob .mime_type .clone() @@ -142,16 +138,36 @@ fn blob_resource_to_image_context( return None; } - Some(ImageContextData { - id: prompt_context_id(session_id, "resource_image", index), - image_path: None, - data_url: Some(format!("data:{};base64,{}", mime_type, blob.blob)), + Some(image_attachment( + prompt_context_id(session_id, "resource_image", index), + None, + Some(format!("data:{};base64,{}", mime_type, blob.blob)), mime_type, - metadata: Some(serde_json::json!({ - "source": "acp_resource", - "uri": blob.uri, - })), - }) + serde_json::json!({ "source": "acp_resource", "uri": blob.uri }), + )) +} + +fn image_attachment( + id: String, + image_path: Option, + data_url: Option, + mime_type: String, + context_metadata: serde_json::Value, +) -> AgentInputAttachment { + let mut metadata = serde_json::Map::new(); + if let Some(image_path) = image_path { + metadata.insert("imagePath".to_string(), image_path.into()); + } + if let Some(data_url) = data_url { + metadata.insert("dataUrl".to_string(), data_url.into()); + } + metadata.insert("mimeType".to_string(), mime_type.into()); + metadata.insert("metadata".to_string(), context_metadata); + AgentInputAttachment { + kind: "remote_image".to_string(), + id, + metadata, + } } fn resource_link_text(link: &ResourceLink) -> String { @@ -233,3 +249,53 @@ fn prompt_context_id(session_id: &str, kind: &str, index: usize) -> String { fn file_uri_to_path(uri: &str) -> Option { uri.strip_prefix("file://").map(|path| path.to_string()) } + +#[cfg(test)] +mod tests { + use agent_client_protocol::schema::{ContentBlock, ImageContent, TextContent}; + + use super::parse_prompt_blocks; + + #[test] + fn image_data_becomes_portable_runtime_attachment() { + let parsed = parse_prompt_blocks( + "session/one", + vec![ + ContentBlock::Text(TextContent::new("describe")), + ContentBlock::Image(ImageContent::new("UExBSU4=", "image/png")), + ], + ); + + assert_eq!(parsed.original_user_message.as_deref(), Some("describe")); + assert!(parsed.user_message.contains("[Attached image:")); + assert_eq!(parsed.attachments.len(), 1); + let attachment = &parsed.attachments[0]; + assert_eq!(attachment.kind, "remote_image"); + assert_eq!(attachment.id, "acp_image_session_one_1"); + assert_eq!(attachment.metadata["mimeType"], "image/png"); + assert_eq!( + attachment.metadata["dataUrl"], + "data:image/png;base64,UExBSU4=" + ); + assert_eq!(attachment.metadata["metadata"]["source"], "acp"); + } + + #[test] + fn image_uri_preserves_path_mime_and_source_metadata() { + let parsed = parse_prompt_blocks( + "session", + vec![ContentBlock::Image( + ImageContent::new(String::new(), "image/jpeg").uri("file:///workspace/clip.jpg"), + )], + ); + + let attachment = &parsed.attachments[0]; + assert_eq!(attachment.metadata["imagePath"], "/workspace/clip.jpg"); + assert_eq!(attachment.metadata["mimeType"], "image/jpeg"); + assert_eq!( + attachment.metadata["metadata"]["uri"], + "file:///workspace/clip.jpg" + ); + assert!(attachment.metadata.get("dataUrl").is_none()); + } +} diff --git a/src/crates/interfaces/acp/src/runtime/model.rs b/src/crates/interfaces/acp/src/runtime/model.rs index 742f62aa10..9dd5cc5385 100644 --- a/src/crates/interfaces/acp/src/runtime/model.rs +++ b/src/crates/interfaces/acp/src/runtime/model.rs @@ -73,8 +73,7 @@ impl BitfunAcpRuntime { let normalized_model_id = normalize_model_selection(model_id).await?; - self.agentic_system - .coordinator + self.compatibility .update_session_model(&bitfun_session_id, &normalized_model_id) .await .map_err(Self::internal_error)?; diff --git a/src/crates/interfaces/acp/src/runtime/prompt.rs b/src/crates/interfaces/acp/src/runtime/prompt.rs index ac2912c671..680ab4b5d6 100644 --- a/src/crates/interfaces/acp/src/runtime/prompt.rs +++ b/src/crates/interfaces/acp/src/runtime/prompt.rs @@ -5,20 +5,23 @@ use agent_client_protocol::schema::{ SessionUpdate, StopReason, }; use agent_client_protocol::{Client, ConnectionTo, Error, Result}; -use bitfun_core::agentic::coordination::{DialogSubmissionPolicy, DialogTriggerSource}; -use bitfun_core::agentic::events::EventEnvelope; +use bitfun_agent_runtime::sdk::{ + AgentDialogTurnRequest, AgentSessionEventReceiver, AgentSubmissionSource, + AgentToolConfirmationRequest, AgentToolRejectionRequest, AgentTurnCancellationRequest, + DialogSubmissionPolicy, DialogSubmitOutcome, +}; use bitfun_events::AgenticEvent as CoreEvent; use log::warn; use serde_json::json; use tokio::sync::broadcast; -use super::content::parse_prompt_blocks; +use super::content::{parse_prompt_blocks, ParsedPrompt}; use super::events::{ permission_request, send_update, tool_event_updates, PERMISSION_ALLOW_ONCE, PERMISSION_REJECT_ONCE, }; use super::thinking::{InlineThinkRouter, InlineThinkSegment}; -use super::BitfunAcpRuntime; +use super::{AcpSessionState, BitfunAcpRuntime}; impl BitfunAcpRuntime { pub(super) async fn run_prompt(&self, request: PromptRequest) -> Result { @@ -36,47 +39,34 @@ impl BitfunAcpRuntime { let parsed_prompt = parse_prompt_blocks(&session_id, request.prompt); - if parsed_prompt.user_message.trim().is_empty() && parsed_prompt.image_contexts.is_empty() { + if parsed_prompt.user_message.trim().is_empty() && parsed_prompt.attachments.is_empty() { return Err(Error::invalid_params().data("empty prompt")); } - let mut event_rx = self.agentic_system.event_queue.subscribe(); - if parsed_prompt.image_contexts.is_empty() { - self.agentic_system - .coordinator - .start_dialog_turn( - acp_session.bitfun_session_id.clone(), - parsed_prompt.user_message, - parsed_prompt.original_user_message, - None, - acp_session.mode_id.clone(), - Some(acp_session.cwd.clone()), - None, - None, - DialogSubmissionPolicy::for_source(DialogTriggerSource::Cli), - Some(acp_user_message_metadata()), - ) - .await - .map_err(Self::internal_error)?; - } else { - self.agentic_system - .coordinator - .start_dialog_turn_with_image_contexts( - acp_session.bitfun_session_id.clone(), - parsed_prompt.user_message, - parsed_prompt.original_user_message, - parsed_prompt.image_contexts, - None, - acp_session.mode_id.clone(), - Some(acp_session.cwd.clone()), - None, - None, - DialogSubmissionPolicy::for_source(DialogTriggerSource::Cli), - Some(acp_user_message_metadata()), - ) - .await - .map_err(Self::internal_error)?; - } + let mut event_rx = self + .agent_runtime + .subscribe_session_events(&acp_session.bitfun_session_id) + .map_err(Self::runtime_error)?; + let outcome = self + .agent_runtime + .submit_dialog_turn(dialog_turn_request(&acp_session, parsed_prompt)) + .await + .map_err(Self::runtime_error)?; + let turn_id = match resolve_started_prompt_turn(outcome) { + Ok(turn_id) => turn_id, + Err(queued_turn_id) => { + self.agent_runtime + .cancel_turn(turn_cancellation_request( + &acp_session.bitfun_session_id, + Some(&queued_turn_id), + "acp_busy_rejected", + )) + .await + .map_err(Self::runtime_error)?; + return Err(Error::internal_error() + .data("Session state does not allow starting new dialog: Processing")); + } + }; let stop_reason = wait_for_prompt_completion( self, @@ -84,6 +74,7 @@ impl BitfunAcpRuntime { &connection, &acp_session.acp_session_id, &acp_session.bitfun_session_id, + &turn_id, ) .await?; @@ -98,29 +89,75 @@ impl BitfunAcpRuntime { .ok_or_else(|| Error::resource_not_found(Some(session_id.clone())))?; let acp_session = acp_session.clone(); - self.agentic_system - .coordinator - .cancel_active_turn_for_session( + self.agent_runtime + .cancel_turn(turn_cancellation_request( &acp_session.bitfun_session_id, - std::time::Duration::from_secs(5), - ) + None, + "acp_client_cancelled", + )) .await - .map_err(Self::internal_error)?; + .map_err(Self::runtime_error)?; Ok(()) } } -fn acp_user_message_metadata() -> serde_json::Value { +fn dialog_turn_request(session: &AcpSessionState, prompt: ParsedPrompt) -> AgentDialogTurnRequest { + AgentDialogTurnRequest { + session_id: session.bitfun_session_id.clone(), + message: prompt.user_message, + original_message: prompt.original_user_message, + turn_id: None, + agent_type: session.mode_id.clone(), + workspace_path: Some(session.cwd.clone()), + remote_connection_id: None, + remote_ssh_host: None, + policy: DialogSubmissionPolicy::for_source(AgentSubmissionSource::Cli), + reply_route: None, + prepended_reminders: Vec::new(), + attachments: prompt.attachments, + metadata: acp_user_message_metadata(), + } +} + +fn resolve_started_prompt_turn( + outcome: DialogSubmitOutcome, +) -> std::result::Result { + match outcome { + DialogSubmitOutcome::Started { turn_id, .. } => Ok(turn_id), + DialogSubmitOutcome::Queued { turn_id, .. } => Err(turn_id), + } +} + +fn turn_cancellation_request( + session_id: &str, + turn_id: Option<&str>, + reason: &str, +) -> AgentTurnCancellationRequest { + AgentTurnCancellationRequest { + session_id: session_id.to_string(), + turn_id: turn_id.map(ToOwned::to_owned), + source: Some(AgentSubmissionSource::Cli), + requester_session_id: None, + reason: Some(reason.to_string()), + wait_timeout_ms: Some(5_000), + } +} + +fn acp_user_message_metadata() -> serde_json::Map { json!({ "acp_transport": true }) + .as_object() + .cloned() + .expect("ACP metadata must be an object") } async fn wait_for_prompt_completion( runtime: &BitfunAcpRuntime, - event_rx: &mut broadcast::Receiver, + event_rx: &mut AgentSessionEventReceiver, connection: &ConnectionTo, acp_session_id: &str, bitfun_session_id: &str, + turn_id: &str, ) -> Result { let mut seen_tool_calls = HashSet::new(); let mut inline_think = InlineThinkRouter::new(); @@ -129,10 +166,26 @@ async fn wait_for_prompt_completion( let event = match event_rx.recv().await { Ok(envelope) => envelope.event, Err(broadcast::error::RecvError::Lagged(count)) => { - warn!("ACP event receiver lagged: skipped {} events", count); - continue; + let message = format!( + "agent event stream lagged; cancelled turn after skipping {count} events" + ); + cancel_turn_after_event_stream_failure( + runtime, + bitfun_session_id, + turn_id, + "acp_event_stream_lagged", + ) + .await; + return Err(Error::internal_error().data(message)); } Err(broadcast::error::RecvError::Closed) => { + cancel_turn_after_event_stream_failure( + runtime, + bitfun_session_id, + turn_id, + "acp_event_stream_closed", + ) + .await; return Err(Error::internal_error().data("event stream closed")); } }; @@ -140,6 +193,9 @@ async fn wait_for_prompt_completion( if event.session_id() != Some(bitfun_session_id) { continue; } + if !prompt_event_matches_turn(&event, turn_id) { + continue; + } match event { CoreEvent::TextChunk { text, .. } => { @@ -202,6 +258,40 @@ async fn wait_for_prompt_completion( } } +async fn cancel_turn_after_event_stream_failure( + runtime: &BitfunAcpRuntime, + session_id: &str, + turn_id: &str, + reason: &str, +) { + if let Err(error) = runtime + .agent_runtime + .cancel_turn(turn_cancellation_request(session_id, Some(turn_id), reason)) + .await + { + warn!( + "Failed to cancel ACP turn after event stream failure: session_id={}, turn_id={}, error={}", + session_id, + turn_id, + error.into_message() + ); + } +} + +fn prompt_event_matches_turn(event: &CoreEvent, expected_turn_id: &str) -> bool { + match event { + CoreEvent::DialogTurnStarted { turn_id, .. } + | CoreEvent::DialogTurnCompleted { turn_id, .. } + | CoreEvent::DialogTurnCancelled { turn_id, .. } + | CoreEvent::DialogTurnFailed { turn_id, .. } + | CoreEvent::TextChunk { turn_id, .. } + | CoreEvent::ThinkingChunk { turn_id, .. } + | CoreEvent::ToolEvent { turn_id, .. } => turn_id == expected_turn_id, + CoreEvent::SystemError { .. } => true, + _ => false, + } +} + fn send_inline_think_segments( connection: &ConnectionTo, acp_session_id: &str, @@ -236,9 +326,11 @@ async fn handle_permission_request( Err(error) => { let reason = format!("ACP permission request failed: {}", error); let _ = runtime - .agentic_system - .coordinator - .reject_tool(tool_id, reason.clone()) + .agent_runtime + .reject_tool(AgentToolRejectionRequest { + tool_id: tool_id.to_string(), + reason: reason.clone(), + }) .await; return Err(error); } @@ -249,29 +341,35 @@ async fn handle_permission_request( if selected.option_id.to_string() == PERMISSION_ALLOW_ONCE => { runtime - .agentic_system - .coordinator - .confirm_tool(tool_id, None) + .agent_runtime + .confirm_tool(AgentToolConfirmationRequest { + tool_id: tool_id.to_string(), + updated_input: None, + }) .await - .map_err(BitfunAcpRuntime::internal_error)?; + .map_err(BitfunAcpRuntime::runtime_error)?; } RequestPermissionOutcome::Selected(selected) if selected.option_id.to_string() == PERMISSION_REJECT_ONCE => { runtime - .agentic_system - .coordinator - .reject_tool(tool_id, "Rejected by ACP client".to_string()) + .agent_runtime + .reject_tool(AgentToolRejectionRequest { + tool_id: tool_id.to_string(), + reason: "Rejected by ACP client".to_string(), + }) .await - .map_err(BitfunAcpRuntime::internal_error)?; + .map_err(BitfunAcpRuntime::runtime_error)?; } RequestPermissionOutcome::Cancelled => { runtime - .agentic_system - .coordinator - .reject_tool(tool_id, "ACP permission request cancelled".to_string()) + .agent_runtime + .reject_tool(AgentToolRejectionRequest { + tool_id: tool_id.to_string(), + reason: "ACP permission request cancelled".to_string(), + }) .await - .map_err(BitfunAcpRuntime::internal_error)?; + .map_err(BitfunAcpRuntime::runtime_error)?; } RequestPermissionOutcome::Selected(selected) => { let reason = format!( @@ -279,21 +377,123 @@ async fn handle_permission_request( selected.option_id ); runtime - .agentic_system - .coordinator - .reject_tool(tool_id, reason) + .agent_runtime + .reject_tool(AgentToolRejectionRequest { + tool_id: tool_id.to_string(), + reason, + }) .await - .map_err(BitfunAcpRuntime::internal_error)?; + .map_err(BitfunAcpRuntime::runtime_error)?; } _ => { runtime - .agentic_system - .coordinator - .reject_tool(tool_id, "Unsupported ACP permission outcome".to_string()) + .agent_runtime + .reject_tool(AgentToolRejectionRequest { + tool_id: tool_id.to_string(), + reason: "Unsupported ACP permission outcome".to_string(), + }) .await - .map_err(BitfunAcpRuntime::internal_error)?; + .map_err(BitfunAcpRuntime::runtime_error)?; } } Ok(()) } + +#[cfg(test)] +mod tests { + use bitfun_agent_runtime::sdk::{ + AgentInputAttachment, AgentSubmissionSource, DialogSubmitOutcome, + }; + use bitfun_events::AgenticEvent; + + use super::{ + dialog_turn_request, prompt_event_matches_turn, resolve_started_prompt_turn, + turn_cancellation_request, AcpSessionState, ParsedPrompt, + }; + + fn session() -> AcpSessionState { + AcpSessionState { + acp_session_id: "acp-session".to_string(), + bitfun_session_id: "bitfun-session".to_string(), + cwd: "/workspace".to_string(), + mode_id: "agentic".to_string(), + model_id: "auto".to_string(), + mcp_server_ids: Vec::new(), + } + } + + #[test] + fn dialog_request_preserves_cli_confirmation_and_acp_metadata() { + let request = dialog_turn_request( + &session(), + ParsedPrompt { + user_message: "describe".to_string(), + original_user_message: Some("describe".to_string()), + attachments: vec![AgentInputAttachment::remote_image( + "image-1", + "clip.png", + "data:image/png;base64,QQ==", + )], + }, + ); + + assert_eq!(request.session_id, "bitfun-session"); + assert_eq!(request.workspace_path.as_deref(), Some("/workspace")); + assert_eq!(request.policy.trigger_source, AgentSubmissionSource::Cli); + assert!(request.policy.requires_tool_confirmation()); + assert_eq!(request.metadata["acp_transport"], true); + assert_eq!(request.attachments.len(), 1); + } + + #[test] + fn cancellation_request_keeps_bounded_wait_and_active_turn_semantics() { + let request = + turn_cancellation_request(&session().bitfun_session_id, None, "acp_client_cancelled"); + + assert_eq!(request.session_id, "bitfun-session"); + assert_eq!(request.turn_id, None); + assert_eq!(request.source, Some(AgentSubmissionSource::Cli)); + assert_eq!(request.wait_timeout_ms, Some(5_000)); + } + + #[test] + fn queued_prompt_is_rejected_with_its_exact_turn_identity() { + let queued_turn_id = resolve_started_prompt_turn(DialogSubmitOutcome::Queued { + session_id: "bitfun-session".to_string(), + turn_id: "turn-queued".to_string(), + }) + .expect_err("queued prompt must not be treated as started"); + let cancellation = + turn_cancellation_request("bitfun-session", Some(&queued_turn_id), "acp_busy_rejected"); + + assert_eq!(cancellation.turn_id.as_deref(), Some("turn-queued")); + assert_eq!(cancellation.reason.as_deref(), Some("acp_busy_rejected")); + } + + #[test] + fn prompt_events_are_scoped_to_the_submitted_turn() { + let current = AgenticEvent::TextChunk { + session_id: "bitfun-session".to_string(), + turn_id: "turn-current".to_string(), + round_id: "round".to_string(), + attempt_id: None, + attempt_index: None, + text: "current".to_string(), + }; + let other = AgenticEvent::DialogTurnCompleted { + session_id: "bitfun-session".to_string(), + turn_id: "turn-other".to_string(), + total_rounds: 1, + total_tools: 0, + duration_ms: 1, + partial_recovery_reason: None, + success: Some(true), + finish_reason: None, + has_final_response: Some(true), + }; + + assert!(prompt_event_matches_turn(¤t, "turn-current")); + assert!(!prompt_event_matches_turn(&other, "turn-current")); + } +} diff --git a/src/crates/interfaces/acp/src/runtime/session.rs b/src/crates/interfaces/acp/src/runtime/session.rs index 4e79071a48..affb4c7810 100644 --- a/src/crates/interfaces/acp/src/runtime/session.rs +++ b/src/crates/interfaces/acp/src/runtime/session.rs @@ -1,5 +1,4 @@ use std::path::Path; -use std::time::{SystemTime, UNIX_EPOCH}; use agent_client_protocol::schema::{ CurrentModeUpdate, ListSessionsRequest, ListSessionsResponse, LoadSessionRequest, @@ -7,8 +6,10 @@ use agent_client_protocol::schema::{ SessionMode, SessionModeState, SessionUpdate, SetSessionModeRequest, SetSessionModeResponse, }; use agent_client_protocol::{Client, ConnectionTo, Error, Result}; +use bitfun_agent_runtime::sdk::{ + AgentSessionCreateRequest, AgentSessionListRequest, SessionStoragePathRequest, +}; use bitfun_core::agentic::agents::get_agent_registry; -use bitfun_core::agentic::core::SessionConfig; use chrono::{DateTime, Utc}; use super::events::send_update; @@ -27,28 +28,27 @@ impl BitfunAcpRuntime { let cwd = request.cwd.to_string_lossy().to_string(); let mcp_servers = request.mcp_servers; let session = self - .agentic_system - .coordinator - .create_session( - format!( + .agent_runtime + .create_session(AgentSessionCreateRequest { + session_name: format!( "ACP Session - {}", chrono::Local::now().format("%Y-%m-%d %H:%M:%S") ), - "agentic".to_string(), - SessionConfig { - workspace_path: Some(cwd.clone()), - ..Default::default() - }, - ) + agent_type: "agentic".to_string(), + workspace_path: Some(cwd.clone()), + remote_connection_id: None, + remote_ssh_host: None, + metadata: serde_json::Map::new(), + }) .await - .map_err(Self::internal_error)?; + .map_err(Self::runtime_error)?; let acp_session = AcpSessionState { acp_session_id: session.session_id.clone(), bitfun_session_id: session.session_id.clone(), cwd, mode_id: session.agent_type.clone(), - model_id: normalize_session_model_id(session.config.model_id.as_deref()), + model_id: normalize_session_model_id(None), mcp_server_ids: self .provision_mcp_servers(&session.session_id, mcp_servers) .await?, @@ -79,15 +79,20 @@ impl BitfunAcpRuntime { let cwd = request.cwd.to_string_lossy().to_string(); let session_id = request.session_id.to_string(); let mcp_servers = request.mcp_servers; - // Restore the runtime session *and* the persisted turns in one pass. - // The turns are needed to stream the conversation history back to the - // client via `session/update` notifications, which the ACP `loadSession` - // contract requires of any agent advertising the `loadSession` - // capability. + // ACP history replay and model selection must come from one persisted + // snapshot. Keep this compatibility path until the runtime contract can + // return the rich turn data ACP actually projects. let (session, turns) = self - .agentic_system - .coordinator - .restore_session_with_turns(Path::new(&cwd), &session_id) + .compatibility + .restore_session_with_turns_for_workspace( + SessionStoragePathRequest { + workspace_path: Path::new(&cwd).to_path_buf(), + remote_connection_id: None, + remote_ssh_host: None, + }, + &session_id, + false, + ) .await .map_err(Self::internal_error)?; @@ -144,19 +149,22 @@ impl BitfunAcpRuntime { .and_then(|value| value.parse::().ok()); let mut summaries = self - .agentic_system - .coordinator - .list_sessions(&cwd) + .agent_runtime + .list_sessions(AgentSessionListRequest { + workspace_path: cwd.to_string_lossy().to_string(), + remote_connection_id: None, + remote_ssh_host: None, + }) .await - .map_err(Self::internal_error)?; - summaries.sort_by_key(|summary| std::cmp::Reverse(summary.last_activity_at)); + .map_err(Self::runtime_error)?; + summaries.sort_by_key(|summary| std::cmp::Reverse(summary.last_active_at_ms)); let limit = 100usize; let filtered = summaries .into_iter() .filter(|summary| { cursor - .map(|cursor| system_time_to_unix_ms(summary.last_activity_at) < cursor) + .map(|cursor| u128::from(summary.last_active_at_ms) < cursor) .unwrap_or(true) }) .collect::>(); @@ -170,14 +178,14 @@ impl BitfunAcpRuntime { Path::new(&cwd).to_path_buf(), ) .title(summary.session_name.clone()) - .updated_at(system_time_to_rfc3339(summary.last_activity_at)) + .updated_at(unix_ms_to_rfc3339(summary.last_active_at_ms)) }) .collect::>(); let next_cursor = if filtered.len() > limit { filtered .get(limit - 1) - .map(|summary| system_time_to_unix_ms(summary.last_activity_at).to_string()) + .map(|summary| summary.last_active_at_ms.to_string()) } else { None }; @@ -210,8 +218,7 @@ impl BitfunAcpRuntime { validate_mode_id(mode_id).await?; - self.agentic_system - .coordinator + self.compatibility .update_session_agent_type(&bitfun_session_id, mode_id) .await .map_err(Self::internal_error)?; @@ -273,12 +280,23 @@ async fn validate_mode_id(mode_id: &str) -> Result<()> { } } -fn system_time_to_unix_ms(time: SystemTime) -> u128 { - time.duration_since(UNIX_EPOCH) - .map(|duration| duration.as_millis()) - .unwrap_or_default() +fn unix_ms_to_rfc3339(time_ms: u64) -> String { + let time_ms = i64::try_from(time_ms).unwrap_or(i64::MAX); + DateTime::::from_timestamp_millis(time_ms) + .unwrap_or(DateTime::::UNIX_EPOCH) + .to_rfc3339() } -fn system_time_to_rfc3339(time: SystemTime) -> String { - DateTime::::from(time).to_rfc3339() +#[cfg(test)] +mod tests { + use super::unix_ms_to_rfc3339; + + #[test] + fn session_timestamps_remain_rfc3339_after_runtime_projection() { + assert_eq!(unix_ms_to_rfc3339(0), "1970-01-01T00:00:00+00:00"); + assert_eq!( + unix_ms_to_rfc3339(1_700_000_000_000), + "2023-11-14T22:13:20+00:00" + ); + } }