From c709b3d9b269a51f3f5c90ddd8ee40a6d7332ea7 Mon Sep 17 00:00:00 2001 From: kdletters Date: Thu, 10 Sep 2026 00:05:02 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E7=8B=AC=E7=AB=8B=20Agent=20?= =?UTF-8?q?App=20Server?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为 agent-cli 增加 stdio JSON-RPC 2.0 服务入口与运行控制协议 补充 Host 实时流监听和流式取消入口 新增 App Server 子进程闭环测试及协议文档 更新 README、架构、TODO 与共享决策记录 --- .../shared-memory/decision-log.md | 11 + rust/Cargo.lock | 1 + rust/README.md | 6 + rust/crates/agent-cli/Cargo.toml | 1 + rust/crates/agent-cli/src/app_server.rs | 507 ++++++++++++++++++ .../agent-cli/src/app_server/protocol.rs | 210 ++++++++ rust/crates/agent-cli/src/main.rs | 51 +- rust/crates/agent-cli/tests/app_server.rs | 327 +++++++++++ rust/crates/agent-host/src/execution.rs | 9 + rust/crates/agent-host/src/lib.rs | 58 +- rust/crates/agent-host/tests/live_stream.rs | 54 ++ .../【任务】Agent内核落地TODO-2026-09-01.md | 10 + ...议】Agent应用服务标准输入输出-2026-09-09.md | 65 +++ .../【架构】独立Agent运行时-2026-09-01.md | 5 + 14 files changed, 1310 insertions(+), 5 deletions(-) create mode 100644 rust/crates/agent-cli/src/app_server.rs create mode 100644 rust/crates/agent-cli/src/app_server/protocol.rs create mode 100644 rust/crates/agent-cli/tests/app_server.rs create mode 100644 rust/crates/agent-host/tests/live_stream.rs create mode 100644 rust/docs/【协议】Agent应用服务标准输入输出-2026-09-09.md diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index fd146cd2c..4a5a822b8 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -9007,6 +9007,17 @@ CI 上 `background_agent_runtime_recovers_stale_running_before_pending_task` 在 Engine worker 的内部 helper 暂不迁移，避免为降低文件行数扩大职责边界。 - 验证：Host 单测与 Clippy 定向通过；完整 workspace、Rustdoc、编码和 diff 门禁在提交前复跑。 +## 2026-09-09 独立 Agent App Server + +- 决策：`agent-cli` 新增 `app-server --stdio`，使用 Agent 自有 JSON-RPC 2.0 JSONL 协议， + 不兼容 Codex App Server wire，也不依赖 AGC。服务复用 Host 的 durable Runtime，支持 + initialize、run 查询/启动/取消/恢复、approval 控制和 shutdown。 +- 边界：单连接只允许一个活动 worker；主线程独占 stdout，worker 通过有界队列发送实时 + stream、已提交 durable event 和完成通知。Host 新增纯观察 `HostStreamListener`，不改变 + durable listener 或 SQLite schema。AGC 后续作为外部子进程客户端连接，不共享 Rust 类型。 +- 验证：CLI App Server 黑盒子进程测试 4/4、Host 80/80 与实时流 2/2 通过；workspace + lock/check、Clippy、fmt、编码和 diff 门禁通过。 + ## 2026-09-09 AGC 改用独立 Agent 内核的替换边界 - 决策：后续目标是用独立 `rust/` workspace 的 Kernel/Engine 替换 AGC 的 Codex 执行循环， diff --git a/rust/Cargo.lock b/rust/Cargo.lock index a85280a5d..b1b5abf4b 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -22,6 +22,7 @@ dependencies = [ "agent-codex", "agent-host", "agent-mcp", + "agent-provider-fake", "agent-provider-openai", "agent-runtime-core", "agent-runtime-engine", diff --git a/rust/README.md b/rust/README.md index 551349222..acc01a0c3 100644 --- a/rust/README.md +++ b/rust/README.md @@ -20,6 +20,12 @@ worker/lease 领取和执行恢复编排，checkpoint/trace 与具体适配器 当前仓库先保证一个最小闭环：中立运行时契约、已提交边界内可重放的事件状态、模型与工具循环、SQLite durable Runtime，以及可替换的 MCP/Skill/Provider 适配器。Codex 外部 backend 和 DAG 编排基础已经作为独立库提供；编排器目前包含有界的配额、消息去重、节点隔离/修复和可选的任务图/协调器原子快照，但真实 Codex wire、HTTP 服务和持久化的完整多 Agent 调度仍不进入内核依赖。 +`agent app-server --stdio` 是独立 Agent 自己的 JSON-RPC 2.0 JSONL 服务入口，不是 Codex +App Server 兼容层。它复用 `agent-host` 的 run/lease/checkpoint/approval 控制，在一个连接 +内按“先接受响应、后异步通知”运行单个 worker，支持查询、取消、审批恢复和 durable 事件； +具体协议见 `rust/docs/【协议】Agent应用服务标准输入输出-2026-09-09.md`。AGC 可以把它当作 +外部 Agent 子进程连接，双方不共享 Rust 类型或 SQLite。 + 独立 workspace 的工具链由 [`rust-toolchain.toml`](./rust-toolchain.toml) 固定为 Rust 1.96，并声明 `rustfmt` 与 `clippy` 组件；独立 CI runner 需要在执行 workflow 前预装同一工具链和组件。这样 `cargo check`、测试、格式化与 Clippy 使用同一编译器， diff --git a/rust/crates/agent-cli/Cargo.toml b/rust/crates/agent-cli/Cargo.toml index e8d243def..3c01848ac 100644 --- a/rust/crates/agent-cli/Cargo.toml +++ b/rust/crates/agent-cli/Cargo.toml @@ -15,6 +15,7 @@ agent-app.workspace = true agent-codex.workspace = true agent-host.workspace = true agent-mcp.workspace = true +agent-provider-fake.workspace = true agent-provider-openai.workspace = true agent-runtime-core.workspace = true agent-runtime-engine.workspace = true diff --git a/rust/crates/agent-cli/src/app_server.rs b/rust/crates/agent-cli/src/app_server.rs new file mode 100644 index 000000000..faf91c59a --- /dev/null +++ b/rust/crates/agent-cli/src/app_server.rs @@ -0,0 +1,507 @@ +//! 独立 stdio App Server。协议循环只做分发，运行与持久化仍由 Host 负责。 +//! +//! stdin reader 和执行 worker 向同一个有界队列发消息，只有主循环写 stdout， +//! 因此运行中仍能处理查询/取消，且响应和通知不会交错成损坏的 JSON 行。 + +mod protocol; + +use std::io::{self, BufRead, Read, Write}; +use std::sync::{Arc, mpsc}; +use std::thread::{self, JoinHandle}; +use std::time::{Duration, Instant}; + +use agent_app::{AgentTomlConfig, queued_run_metadata}; +use agent_host::{AgentHost, HostError, HostRunHandle, HostRunOutput}; +use agent_runtime_engine::{Cancellation, EngineError, EngineEvent, EngineStreamEvent}; +use serde_json::{Value, json}; + +use protocol::{Request, RpcError, decode, nonempty}; + +const MAX_FRAME_BYTES: usize = 1024 * 1024; +const QUEUE_CAPACITY: usize = 64; +const SHUTDOWN_GRACE: Duration = Duration::from_secs(1); + +enum Incoming { + Line(Vec), + Eof, + InputFailed, + Oversized, + Notification { + method: &'static str, + params: Value, + }, + Finished { + run_id: String, + result: Result, + }, +} + +struct Worker { + run_id: String, + cancellation: Cancellation, + join: JoinHandle<()>, +} + +struct Server { + config: Arc, + control: Arc, + sender: mpsc::SyncSender, + active: Option, + initialized: bool, + shutdown: bool, +} + +pub(super) fn run(config: AgentTomlConfig) -> Result<(), Box> { + // 只先打开控制面。Provider/MCP/Skill 在每个 worker 中装配，连接失败不会 + // 卡住 initialize，也不会消耗下次 Fake 请求需要的新脚本。 + let control = AgentHost::open(config.db_path()).map_err(|_| "App Server 数据库初始化失败")?; + let (sender, receiver) = mpsc::sync_channel(QUEUE_CAPACITY); + let input_sender = sender.clone(); + thread::Builder::new() + .name("agent-app-server-input".into()) + .spawn(move || { + let stdin = io::stdin(); + read_input(stdin.lock(), input_sender); + })?; + let mut server = Server { + config: Arc::new(config), + control: Arc::new(control), + sender, + active: None, + initialized: false, + shutdown: false, + }; + let stdout = io::stdout(); + let mut writer = stdout.lock(); + let result = server.serve(&receiver, &mut writer); + // 不论显式 shutdown、EOF 还是输出断开，都经过相同的取消/回收出口。 + server.stop_worker(&receiver, &mut writer, result.is_ok()); + result.map_err(Into::into) +} + +fn read_input(mut reader: impl BufRead, sender: mpsc::SyncSender) { + loop { + let mut line = Vec::new(); + let incoming = match reader + .by_ref() + .take(MAX_FRAME_BYTES as u64 + 1) + .read_until(b'\n', &mut line) + { + Ok(0) => Incoming::Eof, + Ok(_) if line.len() > MAX_FRAME_BYTES => Incoming::Oversized, + Ok(_) => Incoming::Line(line), + Err(_) => Incoming::InputFailed, + }; + let ended = !matches!(incoming, Incoming::Line(_)); + if sender.send(incoming).is_err() || ended { + break; + } + } +} + +fn write_packet(writer: &mut dyn Write, mut packet: Value) -> io::Result<()> { + // 查询、通知和结束结果共用脱敏边界，不能通过嵌套审批 request 绕过。 + super::redact_approval_tokens(&mut packet); + serde_json::to_writer(&mut *writer, &packet)?; + writer.write_all(b"\n")?; + writer.flush() +} + +fn notification(method: &str, params: Value) -> Value { + json!({"jsonrpc":"2.0", "method":method, "params":params}) +} + +fn identity(handle: &HostRunHandle) -> Value { + json!({"runId":handle.run_id,"sessionId":handle.session_id,"runtimeId":handle.runtime_id}) +} + +impl Server { + fn serve( + &mut self, + receiver: &mpsc::Receiver, + writer: &mut dyn Write, + ) -> io::Result<()> { + while !self.shutdown { + let incoming = receiver + .recv() + .map_err(|_| io::Error::other("服务队列已关闭"))?; + match incoming { + Incoming::Line(line) => match protocol::parse(&line) { + Ok(Some(request)) => { + let id = request.id.clone(); + let response = match self.dispatch(request) { + Ok(result) => json!({"jsonrpc":"2.0","id":id,"result":result}), + Err(error) => error.packet(id), + }; + // dispatch 只入队 worker 消息；主循环先写接受响应，再消费它们。 + write_packet(writer, response)?; + } + Ok(None) => {} + Err(error) => write_packet(writer, error)?, + }, + Incoming::Eof => break, + Incoming::InputFailed => return Err(io::Error::other("读取协议输入失败")), + Incoming::Oversized => { + write_packet( + writer, + RpcError::new(-32600, "协议输入超过 1 MiB").packet(Value::Null), + )?; + break; + } + other => self.worker_event(other, writer)?, + } + } + Ok(()) + } + + fn dispatch(&mut self, request: Request) -> Result { + if request.method == "initialize" { + if self.initialized { + return Err(RpcError::new(-32600, "连接已经 initialize")); + } + let params: protocol::Initialize = decode(request.params)?; + if params.protocol_version != protocol::PROTOCOL_VERSION { + return Err(RpcError::params()); + } + self.initialized = true; + return Ok(json!({ + "protocolVersion": protocol::PROTOCOL_VERSION, + "serverInfo":{"name":"agent-runtime","version":env!("CARGO_PKG_VERSION")}, + "capabilities":{"methods":protocol::METHODS,"streaming":true,"maxConcurrentRuns":1} + })); + } + if !self.initialized { + return Err(RpcError::new(-32002, "请先 initialize")); + } + match request.method.as_str() { + "run/start" => { + let params: protocol::Start = decode(request.params)?; + nonempty(¶ms.task)?; + self.ensure_idle()?; + let messages = if let Some(messages) = params.messages { + if messages.is_empty() { + return Err(RpcError::params()); + } + for message in &messages { + message.validate().map_err(|_| RpcError::params())?; + } + messages + } else { + super::prompt_messages_with_config(¶ms.task, &self.config) + .map_err(|_| RpcError::params())? + }; + let provider = self.config.provider(); + let handle = self + .control + .prepare_run_with_messages_and_metadata( + params.task, + messages, + queued_run_metadata(&self.config, &provider), + ) + .map_err(|_| RpcError::host())?; + self.spawn_worker(&handle.run_id, params.stream)?; + Ok(identity(&handle)) + } + "run/get" => { + let params: protocol::RunId = decode(request.params)?; + nonempty(¶ms.run_id)?; + let record = self + .control + .get_run(¶ms.run_id) + .map_err(|_| RpcError::host())? + .ok_or_else(RpcError::missing)?; + Ok(json!(record)) + } + "run/events" => { + let params: protocol::Events = decode(request.params)?; + self.ensure_run(¶ms.run_id)?; + if params.after_revision < 0 { + return Err(RpcError::params()); + } + Ok(json!( + self.control + .list_events(¶ms.run_id, params.after_revision) + .map_err(|_| RpcError::host())? + )) + } + "run/cancel" => { + let params: protocol::RunId = decode(request.params)?; + self.ensure_run(¶ms.run_id)?; + // 持久化取消先提交；当前 worker 的共享标记随后立即生效，不等待轮询。 + let record = self + .control + .cancel(¶ms.run_id) + .map_err(|_| RpcError::host())?; + if let Some(worker) = &self.active + && worker.run_id == params.run_id + { + worker.cancellation.cancel(); + } + Ok(json!(record)) + } + "run/resume" => { + let params: protocol::Resume = decode(request.params)?; + self.ensure_idle()?; + self.ensure_run(¶ms.run_id)?; + let record = self + .control + .get_run(¶ms.run_id) + .map_err(|_| RpcError::host())? + .ok_or_else(RpcError::missing)?; + if record.status != "queued" { + return Err(RpcError::host()); + } + let handle = self.run_identity(¶ms.run_id)?; + self.spawn_worker(¶ms.run_id, params.stream)?; + Ok(handle) + } + "approval/list" => { + let params: protocol::RunId = decode(request.params)?; + self.ensure_run(¶ms.run_id)?; + Ok(json!( + self.control + .list_approvals(¶ms.run_id) + .map_err(|_| RpcError::host())? + )) + } + "approval/resolve" => { + let params: protocol::Resolve = decode(request.params)?; + nonempty(¶ms.approval_id)?; + let decision = params.decision()?; + self.control + .get_approval(¶ms.approval_id) + .map_err(|_| RpcError::host())? + .ok_or_else(RpcError::missing)?; + Ok(json!( + self.control + .resolve_approval(¶ms.approval_id, decision) + .map_err(|_| RpcError::host())? + )) + } + "approval/resume" => { + let params: protocol::ApprovalResume = decode(request.params)?; + nonempty(¶ms.approval_id)?; + self.ensure_idle()?; + self.control + .get_approval(¶ms.approval_id) + .map_err(|_| RpcError::host())? + .ok_or_else(RpcError::missing)?; + let record = self + .control + .resume_approval(¶ms.approval_id) + .map_err(|_| RpcError::host())?; + let handle = self.run_identity(&record.id)?; + self.spawn_worker(&record.id, params.stream)?; + Ok(handle) + } + "shutdown" => { + let _: protocol::Empty = decode(request.params)?; + self.shutdown = true; + Ok(json!({"shutdown":true})) + } + _ => Err(RpcError::new(-32601, "未知 method")), + } + } + + fn ensure_run(&self, run_id: &str) -> Result<(), RpcError> { + nonempty(run_id)?; + self.control + .get_run(run_id) + .map_err(|_| RpcError::host())? + .ok_or_else(RpcError::missing) + .map(|_| ()) + } + + fn ensure_idle(&self) -> Result<(), RpcError> { + if self.active.is_some() { + Err(RpcError::new(-32001, "已有活动 worker")) + } else { + Ok(()) + } + } + + fn run_identity(&self, run_id: &str) -> Result { + let record = self + .control + .get_run(run_id) + .map_err(|_| RpcError::host())? + .ok_or_else(RpcError::missing)?; + let runtime_id = self + .control + .runtime() + .runtime_id_for_run(run_id) + .map_err(|_| RpcError::host())? + .ok_or_else(RpcError::missing)?; + Ok(json!({"runId":run_id,"sessionId":record.session_id,"runtimeId":runtime_id})) + } + + fn spawn_worker(&mut self, run_id: &str, streaming: Option) -> Result<(), RpcError> { + let cancellation = Cancellation::new(); + let worker_cancel = cancellation.clone(); + let control = self.control.clone(); + let config = self.config.clone(); + let sender = self.sender.clone(); + let worker_run_id = run_id.to_owned(); + let streaming = streaming.unwrap_or_else(|| self.config.streaming()); + let join = thread::Builder::new() + .name("agent-app-server-worker".into()) + .spawn(move || { + let result = execute_run( + &control, + &config, + &worker_run_id, + worker_cancel, + streaming, + &sender, + ); + let _ = sender.send(Incoming::Finished { + run_id: worker_run_id, + result, + }); + }) + .map_err(|_| { + let _ = self + .control + .fail_unclaimed_run(run_id, "App Server worker 启动失败"); + RpcError::host() + })?; + self.active = Some(Worker { + run_id: run_id.to_owned(), + cancellation, + join, + }); + Ok(()) + } + + fn worker_event(&mut self, event: Incoming, writer: &mut dyn Write) -> io::Result<()> { + match event { + Incoming::Notification { method, params } => { + write_packet(writer, notification(method, params)) + } + Incoming::Finished { run_id, result } => { + if let Some(worker) = self.active.take() { + let _ = worker.join.join(); + } + let packet = match result { + Ok(result) => { + notification("run/completed", json!({"runId":run_id,"result":result})) + } + Err(error) => { + let status = self + .control + .get_run(&run_id) + .ok() + .flatten() + .map(|record| record.status) + .unwrap_or_else(|| "unknown".into()); + if status == "cancelled" { + notification("run/cancelled", json!({"runId":run_id,"status":status})) + } else if matches!( + error, + HostError::Engine(EngineError::ApprovalRequired { .. }) + ) { + let approvals = self + .control + .list_approvals(&run_id) + .map_err(|_| io::Error::other("读取审批状态失败"))?; + notification( + "run/paused", + json!({"runId":run_id,"approvals":approvals}), + ) + } else { + notification( + "run/failed", + json!({"runId":run_id,"status":status, + "error":{"code":-32000,"message":"运行未完成，请检查持久化状态"}}), + ) + } + } + }; + write_packet(writer, packet) + } + _ => Ok(()), + } + } + + fn stop_worker( + &mut self, + receiver: &mpsc::Receiver, + writer: &mut dyn Write, + mut writable: bool, + ) { + let Some(worker) = &self.active else { + return; + }; + worker.cancellation.cancel(); + let _ = self.control.cancel(&worker.run_id); + let deadline = Instant::now() + SHUTDOWN_GRACE; + while self.active.is_some() { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + break; + } + match receiver.recv_timeout(remaining) { + Ok(Incoming::Finished { run_id, result }) => { + if writable { + let _ = self.worker_event(Incoming::Finished { run_id, result }, writer); + } else if let Some(worker) = self.active.take() { + let _ = worker.join.join(); + } + } + Ok(Incoming::Notification { method, params }) if writable => { + writable = write_packet(writer, notification(method, params)).is_ok(); + } + Ok(_) => {} + Err(_) => break, + } + } + // Rust 同步调用不能安全强杀。超出宽限后不阻塞退出、不伪造终态； + // SQLite 保留 cancel_requested/in-flight，下一进程必须显式对账。 + self.active.take(); + } +} + +fn execute_run( + control: &AgentHost, + config: &AgentTomlConfig, + run_id: &str, + cancellation: Cancellation, + streaming: bool, + sender: &mpsc::SyncSender, +) -> Result { + if cancellation.is_cancelled() { + control.cancel(run_id)?; + return Err(EngineError::Cancelled.into()); + } + let host = AgentHost::with_runtime(control.runtime().clone())?; + let fake_call_id = (config.provider() == "fake").then(|| format!("echo-{run_id}")); + let host = match super::configure_host_with_fake_call_id(host, config, fake_call_id.as_deref()) + { + Ok(host) => host, + Err(_) => { + // 配置错误原文可能带 URL 或私有路径，持久化和 wire 都只保存固定分类。 + control.fail_unclaimed_run(run_id, "App Server worker 配置初始化失败")?; + return Err(HostError::Config("worker 配置初始化失败".into())); + } + }; + let stream_sender = sender.clone(); + let event_sender = sender.clone(); + let host = host + .with_stream_listener(Arc::new(move |run_id: &str, event: &EngineStreamEvent| { + let _ = stream_sender.send(Incoming::Notification { + method: "run/stream", + params: json!({"runId":run_id,"event":event}), + }); + })) + .with_durable_event_callback(move |run_id: &str, revision: i64, event: &EngineEvent| { + let _ = event_sender.send(Incoming::Notification { + method: "run/event", + params: json!({"runId":run_id,"revision":revision,"event":event}), + }); + }); + if streaming { + host.run_existing_streaming_with_cancellation(run_id, cancellation) + } else { + host.run_existing_with_cancellation(run_id, cancellation) + } +} diff --git a/rust/crates/agent-cli/src/app_server/protocol.rs b/rust/crates/agent-cli/src/app_server/protocol.rs new file mode 100644 index 000000000..07725e6de --- /dev/null +++ b/rust/crates/agent-cli/src/app_server/protocol.rs @@ -0,0 +1,210 @@ +//! App Server 自有协议的窄输入边界；不复用 Codex DTO。 + +use agent_runtime_core::{ApprovalDecision, Message}; +use serde::{Deserialize, de::DeserializeOwned}; +use serde_json::{Value, json}; + +pub(super) const PROTOCOL_VERSION: u32 = 1; +pub(super) const METHODS: &[&str] = &[ + "initialize", + "run/start", + "run/get", + "run/events", + "run/cancel", + "run/resume", + "approval/list", + "approval/resolve", + "approval/resume", + "shutdown", +]; + +#[derive(Debug)] +pub(super) struct RpcError { + pub code: i64, + pub message: &'static str, +} + +impl RpcError { + pub fn new(code: i64, message: &'static str) -> Self { + Self { code, message } + } + pub fn params() -> Self { + Self::new(-32602, "请求参数无效") + } + pub fn host() -> Self { + Self::new(-32000, "Host 操作失败，请检查运行状态") + } + pub fn missing() -> Self { + Self::new(-32004, "记录不存在") + } + pub fn packet(&self, id: Value) -> Value { + json!({"jsonrpc":"2.0", "id":id, "error":{"code":self.code,"message":self.message}}) + } +} + +pub(super) struct Request { + pub id: Value, + pub method: String, + pub params: Value, +} + +/// 解析错误只返回固定文案，不把原始行、未知字段或凭据写回 stdout。 +pub(super) fn parse(line: &[u8]) -> Result, Value> { + let value: Value = serde_json::from_slice(line) + .map_err(|_| RpcError::new(-32700, "JSON 解析失败").packet(Value::Null))?; + let invalid = || RpcError::new(-32600, "JSON-RPC 请求无效").packet(Value::Null); + let object = value.as_object().ok_or_else(invalid)?; + if object.get("jsonrpc").and_then(Value::as_str) != Some("2.0") + || object + .keys() + .any(|key| !matches!(key.as_str(), "jsonrpc" | "id" | "method" | "params")) + { + return Err(invalid()); + } + let method = object + .get("method") + .and_then(Value::as_str) + .filter(|method| !method.is_empty()) + .ok_or_else(invalid)?; + let Some(id) = object.get("id") else { + // 通知不执行控制命令，包括无 id 的 run/start/shutdown。 + return Ok(None); + }; + if !id.is_string() && !id.is_i64() && !id.is_u64() { + return Err(invalid()); + } + let params = object.get("params").cloned().unwrap_or_else(|| json!({})); + if !params.is_object() { + return Err(RpcError::params().packet(id.clone())); + } + Ok(Some(Request { + id: id.clone(), + method: method.to_owned(), + params, + })) +} + +pub(super) fn decode(params: Value) -> Result { + serde_json::from_value(params).map_err(|_| RpcError::params()) +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub(super) struct Initialize { + pub protocol_version: u32, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(super) struct Empty {} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub(super) struct Start { + pub task: String, + pub messages: Option>, + pub stream: Option, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub(super) struct RunId { + pub run_id: String, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub(super) struct Events { + pub run_id: String, + #[serde(default)] + pub after_revision: i64, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub(super) struct Resume { + pub run_id: String, + pub stream: Option, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub(super) struct ApprovalResume { + pub approval_id: String, + pub stream: Option, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub(super) struct Resolve { + pub approval_id: String, + pub decision: String, + pub reason: Option, +} + +impl Resolve { + pub fn decision(&self) -> Result { + match (self.decision.as_str(), &self.reason) { + ("allow", None) => Ok(ApprovalDecision::Allow), + ("deny", Some(reason)) if !reason.trim().is_empty() => Ok(ApprovalDecision::Deny { + reason: reason.clone(), + }), + _ => Err(RpcError::params()), + } + } +} + +pub(super) fn nonempty(value: &str) -> Result<(), RpcError> { + if value.trim().is_empty() { + Err(RpcError::params()) + } else { + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rejects_invalid_envelopes_without_echoing_input() { + for input in [ + r#"not-json-secret"#, + "[]", + "null", + r#"{"jsonrpc":"1.0","id":1,"method":"initialize"}"#, + r#"{"jsonrpc":"2.0","id":null,"method":"run/start"}"#, + r#"{"jsonrpc":"2.0","id":true,"method":"run/start"}"#, + r#"{"jsonrpc":"2.0","id":1,"method":"run/start","result":"secret"}"#, + ] { + let error = parse(input.as_bytes()).err().expect("invalid envelope"); + assert!(!error.to_string().contains("secret")); + } + } + + #[test] + fn ignores_notifications_and_preserves_string_ids() { + assert!( + parse(br#"{"jsonrpc":"2.0","method":"run/start","params":{"task":"x"}}"#) + .unwrap() + .is_none() + ); + let request = parse(br#"{"jsonrpc":"2.0","id":"abc","method":"initialize"}"#) + .unwrap() + .unwrap(); + assert_eq!(request.id, "abc"); + assert_eq!(request.params, json!({})); + } + + #[test] + fn validates_approval_decisions_before_host_mutations() { + for params in [ + json!({"approvalId":"a","decision":"deny"}), + json!({"approvalId":"a","decision":"allow","reason":"unexpected"}), + json!({"approvalId":"a","decision":"deny","reason":" "}), + ] { + assert!(decode::(params).unwrap().decision().is_err()); + } + assert!(decode::(json!({"task":"x","apiKey":"secret"})).is_err()); + } +} diff --git a/rust/crates/agent-cli/src/main.rs b/rust/crates/agent-cli/src/main.rs index b5e69b8e7..8b502e7ab 100644 --- a/rust/crates/agent-cli/src/main.rs +++ b/rust/crates/agent-cli/src/main.rs @@ -22,6 +22,8 @@ use agent_skills::SkillLoader; use serde::Serialize; use serde_json::{Value, json}; +mod app_server; + /// `reconcile` 保留旧的单 run 入口，并用显式 `--stale` 选择一次有界扫描。 /// 默认值与 Runtime 的硬上限一致；扫描本身仍由 Host/Runtime 原子执行。 const DEFAULT_STALE_RECONCILE_LIMIT: usize = 256; @@ -69,6 +71,20 @@ fn main() { fn run() -> Result<(), Box> { let mut args = env::args().skip(1); let command = args.next().unwrap_or_else(|| "run".to_owned()); + // 服务入口独立校验参数，不能把未知选项误当任务，也不读取 stdin 作为 prompt。 + if command == "app-server" { + let options = args.collect::>(); + if options == ["--help"] || options == ["-h"] { + println!( + "用法: agent app-server --stdio\nJSON-RPC 2.0 JSONL；先 initialize，再 run/start。" + ); + return Ok(()); + } + if options != ["--stdio"] { + return Err("用法: agent app-server --stdio".into()); + } + return app_server::run(AgentTomlConfig::load()?); + } let config = AgentTomlConfig::load()?; let db = config.db_path(); @@ -402,9 +418,39 @@ fn open_configured_host( db: &Path, config: &AgentTomlConfig, ) -> Result> { - let mut host = AgentHost::open(db)?; + configure_host(AgentHost::open(db)?, config) +} + +/// CLI worker 和 App Server 共用装配过程；后者复用 control Host 的 Runtime， +/// 不为每次请求创建另一套存储，也不重复使用已消费完的 Fake Provider 脚本。 +fn configure_host( + host: AgentHost, + config: &AgentTomlConfig, +) -> Result> { + configure_host_with_fake_call_id(host, config, None) +} + +fn configure_host_with_fake_call_id( + mut host: AgentHost, + config: &AgentTomlConfig, + fake_call_id: Option<&str>, +) -> Result> { match config.provider().as_str() { - "fake" => host = host.with_fake_provider(), + "fake" => { + host = if let Some(call_id) = fake_call_id { + host.with_provider( + Arc::new(agent_provider_fake::FakeProvider::tool_then_text( + call_id, + "echo", + json!({"text": "hello from fake provider"}), + "fake provider complete", + )), + "fake", + ) + } else { + host.with_fake_provider() + } + } "openai" => { let model = config.model(); let model = if model == "fake" { @@ -1191,6 +1237,7 @@ fn mcp_list_from_client( fn print_help() { println!( r#"用法: + agent app-server --stdio # 独立 JSON-RPC 服务；无需 Codex agent run [--stream|--no-stream] [--jsonl] [任务] agent run --background [--jsonl] [任务] agent worker # 内部 worker diff --git a/rust/crates/agent-cli/tests/app_server.rs b/rust/crates/agent-cli/tests/app_server.rs new file mode 100644 index 000000000..86eac3ade --- /dev/null +++ b/rust/crates/agent-cli/tests/app_server.rs @@ -0,0 +1,327 @@ +//! `agent app-server --stdio` 的黑盒协议测试。 +//! +//! 测试只通过子进程的 stdin/stdout 交互，不读取宿主环境中的凭据，也不依赖 +//! AGC 的实现细节；这样可以把协议回归和 CLI 的装配/生命周期一起验收。 + +use std::fs; +use std::io::{BufRead, BufReader, Write}; +use std::path::{Path, PathBuf}; +use std::process::{Child, ChildStdin, Command, Stdio}; +use std::sync::mpsc::{self, Receiver}; +use std::thread; +use std::time::Duration; + +use serde_json::{Value, json}; + +const IO_TIMEOUT: Duration = Duration::from_secs(10); + +struct TempDir { + path: PathBuf, +} + +impl TempDir { + fn new() -> Self { + let base = std::env::var_os("TMPDIR") + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from("/home/dsk/data/tmp")); + fs::create_dir_all(&base).expect("create test temp base"); + for n in 0..1000u32 { + let path = base.join(format!("agent-app-server-{n}-{}", std::process::id())); + if fs::create_dir(&path).is_ok() { + return Self { path }; + } + } + panic!("unable to allocate test temp directory"); + } + + fn path(&self) -> &Path { + &self.path + } +} + +impl Drop for TempDir { + fn drop(&mut self) { + // 测试只删除自己生成的精确目录；目录内是该测试进程独占的数据库和配置。 + let _ = fs::remove_dir_all(&self.path); + } +} + +struct AppServer { + child: Child, + stdin: Option, + lines: Receiver, +} + +impl AppServer { + fn start(temp: &TempDir) -> Self { + let config = temp.path().join("empty.toml"); + fs::write(&config, "").expect("write empty config"); + let db = temp.path().join("agent.db"); + let home = temp.path().join("home"); + fs::create_dir_all(&home).expect("create isolated home"); + let mut child = Command::new(env!("CARGO_BIN_EXE_agent")) + .args(["app-server", "--stdio"]) + .current_dir(temp.path()) + .env_clear() + // `env_clear` prevents credential leakage, while PATH keeps the + // child runtime's normal process environment usable. + .env("PATH", std::env::var_os("PATH").unwrap_or_default()) + .env("AGENT_PROVIDER", "fake") + .env("AGENT_DB", &db) + .env("AGENT_CONFIG", &config) + .env("TMPDIR", temp.path()) + .env("HOME", &home) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn agent app-server"); + let stdin = child.stdin.take().expect("app-server stdin"); + let stdout = child.stdout.take().expect("app-server stdout"); + let (tx, rx) = mpsc::channel(); + thread::spawn(move || { + let reader = BufReader::new(stdout); + for line in reader.lines() { + match line { + Ok(line) => { + if tx.send(line).is_err() { + break; + } + } + Err(_) => break, + } + } + }); + Self { + child, + stdin: Some(stdin), + lines: rx, + } + } + + fn send(&mut self, value: Value) { + let stdin = self.stdin.as_mut().expect("app-server stdin is open"); + writeln!(stdin, "{value}").expect("write app-server request"); + stdin.flush().expect("flush app-server request"); + } + + fn close_stdin(&mut self) { + // Drop the pipe explicitly to exercise EOF cooperative shutdown. + self.stdin.take(); + } + + fn next(&self) -> Value { + let line = self + .lines + .recv_timeout(IO_TIMEOUT) + .expect("timed out waiting for app-server output"); + serde_json::from_str(&line) + .unwrap_or_else(|error| panic!("invalid JSONL output: {line}: {error}")) + } + + fn response(&self, id: i64) -> Value { + loop { + let message = self.next(); + if message.get("id") == Some(&json!(id)) { + return message; + } + } + } + + fn initialize(&mut self) { + self.send( + json!({"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":1}}), + ); + let response = self.response(1); + assert_eq!(response["result"]["protocolVersion"], 1); + } + + fn run_to_completion(&mut self, id: i64, task: &str, stream: bool) -> String { + self.send(json!({"jsonrpc":"2.0","id":id,"method":"run/start","params":{"task":task,"stream":stream}})); + let accepted = self.response(id); + let run_id = accepted["result"]["runId"] + .as_str() + .expect("runId") + .to_owned(); + for _ in 0..64 { + let message = self.next(); + if message["method"] == "run/completed" && message["params"]["runId"] == run_id { + assert!(message["params"]["result"]["output"].is_object()); + return run_id; + } + } + panic!("run/completed notification not received"); + } +} + +impl Drop for AppServer { + fn drop(&mut self) { + if let Some(stdin) = self.stdin.as_mut() { + let _ = stdin.flush(); + } + if self.child.try_wait().ok().flatten().is_none() { + let _ = self.child.kill(); + } + let _ = self.child.wait(); + } +} + +#[test] +fn protocol_requires_initialize_and_rejects_invalid_requests() { + let temp = TempDir::new(); + let mut server = AppServer::start(&temp); + server.send(json!({"jsonrpc":"2.0","id":1,"method":"run/get","params":{"runId":"missing"}})); + let response = server.response(1); + assert_eq!(response["error"]["code"], -32002); + + server.send( + json!({"jsonrpc":"2.0","id":2,"method":"initialize","params":{"protocolVersion":99}}), + ); + assert_eq!(server.response(2)["error"]["code"], -32602); + server.initialize(); + + server.send(json!({"jsonrpc":"2.0","id":6,"method":"run/get","params":{"runId":"missing"}})); + assert_eq!(server.response(6)["error"]["code"], -32004); + + server + .send(json!({"jsonrpc":"2.0","id":3,"method":"initialize","params":{"protocolVersion":1}})); + assert_eq!(server.response(3)["error"]["code"], -32600); + server.send(json!({"jsonrpc":"2.0","id":4,"method":"unknown","params":{}})); + assert_eq!(server.response(4)["error"]["code"], -32601); + server.send(json!({"jsonrpc":"2.0","id":5,"method":"run/start","params":{"task":""}})); + assert_eq!(server.response(5)["error"]["code"], -32602); + server.send(json!({"jsonrpc":"2.0","id":7,"method":"run/start","params":{"task":42}})); + assert_eq!(server.response(7)["error"]["code"], -32602); +} + +#[test] +fn run_stream_tool_and_persisted_queries_form_one_closed_loop() { + let temp = TempDir::new(); + let mut server = AppServer::start(&temp); + server.initialize(); + server.send(json!({"jsonrpc":"2.0","id":2,"method":"run/start","params":{"task":"hello","stream":true}})); + let accepted = server.response(2); + let run_id = accepted["result"]["runId"] + .as_str() + .expect("runId") + .to_owned(); + assert!(accepted["result"]["sessionId"].is_string()); + assert!(accepted["result"]["runtimeId"].is_string()); + + let mut completed = None; + let mut saw_stream = false; + let mut saw_tool = false; + for _ in 0..32 { + let message = server.next(); + match message["method"].as_str() { + Some("run/stream") => { + saw_stream = true; + } + Some("run/event") => { + // Engine 的实时流只转发 Provider 事件；工具闭环属于 durable + // run/event，事件标签按协议保持 snake_case。 + saw_tool |= message["params"]["event"]["type"] + .as_str() + .is_some_and(|kind| kind.contains("tool")); + } + Some("run/completed") => { + if message["params"]["runId"] == run_id { + completed = Some(message); + break; + } + } + Some("run/failed") | Some("run/cancelled") | Some("run/paused") + if message["params"]["runId"] == run_id => + { + panic!("run did not complete: {message}"); + } + _ => {} + } + } + let completed = completed.expect("run/completed notification"); + assert!( + saw_stream, + "stream=true must produce run/stream notifications" + ); + assert!(saw_tool, "fake provider tool loop must be observable"); + assert!(completed["params"]["result"]["output"].is_object()); + + // A long-lived app-server connection must be reusable. The implementation + // reassembles the deterministic fake provider for each accepted run. + let second_run = server.run_to_completion(6, "second run", false); + assert_ne!(second_run, run_id); + + server.send(json!({"jsonrpc":"2.0","id":3,"method":"run/get","params":{"runId":run_id}})); + let record = server.response(3)["result"].clone(); + assert!( + record["status"] + .as_str() + .is_some_and(|status| status == "completed" || status == "succeeded") + ); + server.send(json!({"jsonrpc":"2.0","id":4,"method":"run/events","params":{"runId":run_id,"afterRevision":0}})); + assert!(server.response(4)["result"].is_array()); + + server.send(json!({"jsonrpc":"2.0","id":5,"method":"shutdown","params":{}})); + assert_eq!(server.response(5)["result"]["shutdown"], true); + + // Durable records remain queryable after a process restart. + let mut restarted = AppServer::start(&temp); + restarted.initialize(); + restarted.send(json!({"jsonrpc":"2.0","id":2,"method":"run/get","params":{"runId":run_id}})); + assert!(restarted.response(2)["result"]["id"].is_string()); + restarted.send(json!({"jsonrpc":"2.0","id":3,"method":"run/events","params":{"runId":run_id,"afterRevision":0}})); + assert!( + restarted.response(3)["result"] + .as_array() + .is_some_and(|events| !events.is_empty()) + ); + restarted.send(json!({"jsonrpc":"2.0","id":4,"method":"shutdown","params":{}})); + assert_eq!(restarted.response(4)["result"]["shutdown"], true); +} + +#[test] +fn shutdown_is_acknowledged_and_process_exits_without_orphan() { + let temp = TempDir::new(); + let mut server = AppServer::start(&temp); + server.initialize(); + server.send(json!({"jsonrpc":"2.0","id":2,"method":"shutdown","params":{}})); + assert_eq!(server.response(2)["result"]["shutdown"], true); + assert!( + server.child.wait_timeout(IO_TIMEOUT).is_ok(), + "shutdown must terminate process" + ); +} + +#[test] +fn stdin_eof_terminates_server_without_orphan_process() { + let temp = TempDir::new(); + let mut server = AppServer::start(&temp); + server.initialize(); + server.close_stdin(); + assert!( + server.child.wait_timeout(IO_TIMEOUT).is_ok(), + "EOF must terminate process" + ); +} + +trait ChildWaitTimeout { + fn wait_timeout(&mut self, timeout: Duration) -> std::io::Result; +} + +impl ChildWaitTimeout for Child { + fn wait_timeout(&mut self, timeout: Duration) -> std::io::Result { + let start = std::time::Instant::now(); + loop { + if let Some(status) = self.try_wait()? { + return Ok(status); + } + if start.elapsed() >= timeout { + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "child still running", + )); + } + thread::sleep(Duration::from_millis(20)); + } + } +} diff --git a/rust/crates/agent-host/src/execution.rs b/rust/crates/agent-host/src/execution.rs index cad2113c6..1157ae543 100644 --- a/rust/crates/agent-host/src/execution.rs +++ b/rust/crates/agent-host/src/execution.rs @@ -53,6 +53,15 @@ impl AgentHost { self.run_existing_with_cancellation_mode(run_id, Cancellation::new(), true) } + /// 运行已有 run 的实时流模式，并允许调用方在执行期间协作取消。 + pub fn run_existing_streaming_with_cancellation( + &self, + run_id: &str, + cancellation: Cancellation, + ) -> Result { + self.run_existing_with_cancellation_mode(run_id, cancellation, true) + } + /// 执行已有 run，并允许宿主在 step 边界注入共享取消标记。 pub fn run_existing_with_cancellation( &self, diff --git a/rust/crates/agent-host/src/lib.rs b/rust/crates/agent-host/src/lib.rs index 9882fb684..af0da338d 100644 --- a/rust/crates/agent-host/src/lib.rs +++ b/rust/crates/agent-host/src/lib.rs @@ -25,8 +25,8 @@ use agent_runtime_core::{ }; use agent_runtime_engine::{ AgentEngine, AgentInput, AgentOutput, AllowList, ApprovalResume, Cancellation, - ContextCompressor, EchoProvider, EngineError, EngineEvent, EventListener, - OwnedProviderContextCompressor, + ContextCompressor, EchoProvider, EngineError, EngineEvent, EngineStreamEvent, EventListener, + OwnedProviderContextCompressor, StreamEventListener, }; use agent_runtime_sqlite::{ ApprovalRecord, CheckpointRecord, EventRecord, ExternalSessionRecord, @@ -257,6 +257,32 @@ where } } +/// Host 实时流监听器。它只观察 Provider 的增量事件，不参与 durable +/// 事务；run_id 由 Host 注入，保证并发运行时事件归属明确。 +pub trait HostStreamListener: Send + Sync { + fn on_stream_event(&self, run_id: &str, event: &EngineStreamEvent); +} + +impl HostStreamListener for F +where + F: Fn(&str, &EngineStreamEvent) + Send + Sync, +{ + fn on_stream_event(&self, run_id: &str, event: &EngineStreamEvent) { + self(run_id, event); + } +} + +struct RunStreamForwarder { + run_id: String, + listener: Arc, +} + +impl StreamEventListener for RunStreamForwarder { + fn on_stream_event(&self, event: &EngineStreamEvent) { + self.listener.on_stream_event(&self.run_id, event); + } +} + /// 默认回显工具，用于 CLI 离线自检；真实宿主可以注册自己的实现。 #[derive(Clone, Debug, Default)] pub struct EchoTool; @@ -292,6 +318,7 @@ pub struct AgentHost { /// summary providers are independent and must survive provider swaps. provider_compressor_auto: bool, durable_event_listener: Option>, + stream_listener: Option>, model: String, /// Built-in CLI provider kind used to fence queued metadata. Generic /// `with_provider` injections leave this unset for compatibility. @@ -363,6 +390,7 @@ impl AgentHost { context_compressor: None, provider_compressor_auto: false, durable_event_listener: None, + stream_listener: None, model: "fake".to_owned(), provider_kind: None, }) @@ -902,6 +930,20 @@ impl AgentHost { self.with_durable_event_listener(Arc::new(callback)) } + /// 注入实时 Provider 流回调；回调只观察流事件，不参与 durable 事务。 + pub fn with_stream_listener(mut self, listener: Arc) -> Self { + self.stream_listener = Some(listener); + self + } + + /// `HostStreamListener` 的闭包便捷入口。 + pub fn with_stream_callback(self, callback: F) -> Self + where + F: Fn(&str, &EngineStreamEvent) + Send + Sync + 'static, + { + self.with_stream_listener(Arc::new(callback)) + } + /// 注入已经由调用方显式读取的 MCP resource/prompt 内容；内容保持不 /// 可信，不会自动改变工具审批策略。 pub fn with_mcp_context(self, source: McpContextSource) -> Self { @@ -2369,9 +2411,19 @@ impl AgentHost { } else { input }; - let output = engine + let stream_forwarder = self + .stream_listener + .as_ref() + .map(|listener| RunStreamForwarder { + run_id: record.id.clone(), + listener: listener.clone(), + }); + let mut output = engine .with_listener(&collected) .with_checkpoint_listener(&checkpoints); + if streaming && let Some(forwarder) = stream_forwarder.as_ref() { + output = output.with_stream_listener(forwarder); + } let output = if streaming { output.run_streaming(input) } else { diff --git a/rust/crates/agent-host/tests/live_stream.rs b/rust/crates/agent-host/tests/live_stream.rs new file mode 100644 index 000000000..1cd1d6a40 --- /dev/null +++ b/rust/crates/agent-host/tests/live_stream.rs @@ -0,0 +1,54 @@ +use std::sync::{Arc, Mutex}; + +use agent_host::AgentHost; +use agent_provider_fake::{FakeProvider, FakeStep}; +use agent_runtime_core::ProviderStreamEvent; +use agent_runtime_engine::{Cancellation, EngineStreamEvent}; + +#[test] +fn stream_callback_receives_delta_with_run_id_before_return() { + let seen = Arc::new(Mutex::new(Vec::<(String, String)>::new())); + let seen_clone = seen.clone(); + let host = AgentHost::in_memory() + .unwrap() + .with_provider( + Arc::new(FakeProvider::new([FakeStep::stream_text(["你", "好"])])), + "fake", + ) + .with_stream_callback(move |run_id: &str, event: &EngineStreamEvent| { + if let ProviderStreamEvent::TextDelta { delta, .. } = event.event() { + seen_clone + .lock() + .unwrap() + .push((run_id.to_owned(), delta.clone())); + } + }); + + let output = host.run_streaming("问候").unwrap(); + let events = seen.lock().unwrap().clone(); + assert!(!events.is_empty()); + assert!(events.iter().all(|(run_id, _)| run_id == &output.run_id)); + assert_eq!( + events + .iter() + .map(|(_, delta)| delta.as_str()) + .collect::(), + "你好" + ); + assert_eq!(output.output.text, "你好"); +} + +#[test] +fn cancelled_stream_does_not_start_provider_or_tool() { + let provider = Arc::new(FakeProvider::new([FakeStep::stream_text(["不会执行"])])); + let cancellation = Cancellation::new(); + cancellation.cancel(); + let host = AgentHost::in_memory() + .unwrap() + .with_provider(provider.clone(), "fake"); + let handle = host.prepare_run("取消").unwrap(); + + let result = host.run_existing_streaming_with_cancellation(&handle.run_id, cancellation); + assert!(result.is_err()); + assert_eq!(provider.remaining_steps(), 1); +} diff --git a/rust/docs/【任务】Agent内核落地TODO-2026-09-01.md b/rust/docs/【任务】Agent内核落地TODO-2026-09-01.md index 9c4077d0c..b7b92ad8c 100644 --- a/rust/docs/【任务】Agent内核落地TODO-2026-09-01.md +++ b/rust/docs/【任务】Agent内核落地TODO-2026-09-01.md @@ -32,6 +32,16 @@ - [ ] `finish_cancelled`、Engine worker 主体和跨模块终态 glue 仍在根模块；后续只在能保持边界 和回归证据时继续拆分，不为降低行数引入新的平行装配层。 +### 独立 App Server（2026-09-09） + +- [x] `agent app-server --stdio` 提供自有 JSON-RPC 2.0 JSONL 协议；复用 Host Runtime，不依赖 + Codex App Server 或 AGC 客户端。 +- [x] 支持 initialize、run start/get/events/cancel/resume、approval list/resolve/resume 和 + shutdown；查询/取消可在 worker 执行时处理，stdout 仅输出协议 JSON。 +- [x] 黑盒子进程测试覆盖初始化门禁、错误码、stream/tool/durable event、同连接多 run、重启 + 查询、shutdown 和 EOF；4/4 通过。 +- [ ] 尚未接入 AGC；HTTP/WebSocket、并发多 worker、Codex wire 兼容和跨进程自动重连不属于本轮。 + ### 当前范围与消息一致性验收（2026-09-06） - [x] 复现并修复正常完成仍重复保存 assistant/tool-call 的缺陷，删除按最终 phase/末条消息内容去重的推断。 diff --git a/rust/docs/【协议】Agent应用服务标准输入输出-2026-09-09.md b/rust/docs/【协议】Agent应用服务标准输入输出-2026-09-09.md new file mode 100644 index 000000000..a8f312676 --- /dev/null +++ b/rust/docs/【协议】Agent应用服务标准输入输出-2026-09-09.md @@ -0,0 +1,65 @@ +# Agent App Server stdio 协议 + +## 范围 + +`agent app-server --stdio` 是新 Agent 自己的常驻服务入口。只用 stdin/stdout JSONL +通信，不依赖 AGC 或 Codex，不提供 Codex wire 兼容。复用 CLI 的 `AGENT_CONFIG`、 +`AGENT_DB`、Provider、Prompt、MCP 和 Skill 配置；密钥只从启动环境/既有配置引用读取， +不通过协议设置、打印或复制。stdout 只输出 JSON，诊断走 stderr。 + +每行一个 JSON-RPC 2.0 对象，不支持 batch。请求 id 只能是字符串或整数；客户端自行 +保证同连接未完成请求 id 唯一。输入上限 1 MiB，超限返回错误后断开。所有命令需要 id； +无 id 的合法通知被忽略，不触发副作用。首个有效命令必须是 initialize。 + +## 请求 + +| method | params | result | +| --- | --- | --- | +| initialize | `{ "protocolVersion": 1 }` | serverInfo、protocolVersion、capabilities | +| run/start | `{ "task": "你好", "stream": true }`；可选 `messages` 为 Core 消息数组 | `{runId, sessionId, runtimeId}`，只表示 durable queued 接受 | +| run/get | `{ "runId": "…" }` | 持久化 RunRecord，未知 ID 报错 | +| run/events | `{ "runId": "…", "afterRevision": 0 }` | 已落盘审计 EventRecord 数组 | +| run/cancel | `{ "runId": "…" }` | Host cancel 返回的 RunRecord，不承诺立刻 cancelled | +| run/resume | `{ "runId": "…", "stream": true }` | 接受 queued run；不自行对账、重排队或重放 unknown | +| approval/list | `{ "runId": "…" }` | 脱敏审批记录数组 | +| approval/resolve | `{ "approvalId": "…", "decision": "allow" }` 或 `deny` 加非空 `reason` | 脱敏审批记录；只记录决议 | +| approval/resume | `{ "approvalId": "…", "stream": true }` | Host 显式恢复同一 run 后返回运行身份 | +| shutdown | `{}` | `{ "shutdown": true }`，取消本进程活动 run 并退出 | + +stream 缺省沿用 CLI 配置。task 非空；messages 提供时必须是非空、合法 Core 消息数组， +客户端提供完整历史；未提供时复用 CLI Prompt 配置。每次 run/start 使用 Host 新建的 +session/run/runtime，不伪造 Codex thread/resume。结果中记录/消息沿用现有 Host 序列化格式。 + +## 通知与生命周期 + +- `run/stream`：`{runId, event}`，event 为实时 EngineStreamEvent（step + Provider 事件）； + 它是暂态流，不宣称已落盘，也不支持断线 token 重放。 +- `run/event`：`{runId, revision, event}`，仅在 run-level 审计事件提交后发出。 +- `run/completed`：`{runId, result}`，result 为 HostRunOutput；Host 已完成持久化收束。 +- `run/paused`：`{runId, approvals}`，持久化审批等待；审批 token 不回传。 +- `run/cancelled`：`{runId, status}`，只用于 Host 已确认 cancelled 的 run。 +- `run/failed`：`{runId, status, error}`，保留真实状态，unknown/reconciling 不冒充失败收束。 + +接受响应先于该次 worker 通知；每次执行尝试只发一个结束/暂停通知。客户端必须持续读取 +stdout。单连接只运行一个活动 worker，忙时新 start/resume 返回 `-32001`；查询、取消、 +审批决议可在模型调用中处理。所有运行仍由 Host SQLite/lease 协调，不另建状态机。 + +EOF、协议致命错误、stdout 断开或 shutdown 都对本进程活动 run 发出 cooperative cancel， +最多等 1 秒。同步 Provider/工具不能被安全强杀，超时退出保留真实 durable 状态；重启后由 +已有 CLI reconciliation 流程核对未知副作用，不自动恢复执行。正常退出会回收已结束 worker。 + +错误码：`-32700` JSON 解析失败；`-32600` envelope/重复 initialize 无效；`-32601` 未知 +method；`-32602` 参数或协议版本错误；`-32002` 尚未 initialize；`-32001` worker 忙； +`-32004` 记录不存在；`-32000` Host 操作失败（固定摘要，不回显原始上游错误或配置）。 + +## 最小交互 + +启动：`cargo run --locked -p agent-cli -- app-server --stdio`。在同一打开的 stdin 中依次写入： + +```jsonl +{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":1}} +{"jsonrpc":"2.0","id":2,"method":"run/start","params":{"task":"你好","stream":true}} +``` + +客户端读到 run/completed 等通知后再发 shutdown。直接管道写完并关闭 stdin 表示断开连接， +会取消尚未完成的 run，不是“等待运行完成”。 diff --git a/rust/docs/【架构】独立Agent运行时-2026-09-01.md b/rust/docs/【架构】独立Agent运行时-2026-09-01.md index ab4d77996..570bc0bf8 100644 --- a/rust/docs/【架构】独立Agent运行时-2026-09-01.md +++ b/rust/docs/【架构】独立Agent运行时-2026-09-01.md @@ -61,6 +61,11 @@ Host 的 run 入口和 worker/lease 执行恢复编排位于私有 `execution.rs 控制面和终态 helper；它不引入新的公开 API，也不持有 SQLite 之外的状态。这样根模块逐步 收敛为装配 facade 与跨职责 glue，后续仍可按边界继续拆分，但不以机械搬迁改变行为。 +`agent-cli` 另外提供独立的 `app-server --stdio` 入口。它使用自有 JSON-RPC JSONL 协议， +通过线程和有界消息队列复用 Host 的 durable run，不连接 Codex App Server，也不把 AGC +客户端作为依赖。服务端只允许一个活动 worker；查询、取消、审批和 shutdown 与模型执行 +共享同一 Host Runtime，协议和错误边界见 `rust/docs/【协议】Agent应用服务标准输入输出-2026-09-09.md`。 + ## 当前实现顺序 1. Core 契约和纯 reducer；