新增独立 Agent App Server

为 agent-cli 增加 stdio JSON-RPC 2.0 服务入口与运行控制协议

补充 Host 实时流监听和流式取消入口

新增 App Server 子进程闭环测试及协议文档

更新 README、架构、TODO 与共享决策记录

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 执行循环，

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",

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 使用同一编译器，

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

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<Option<Request>, 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<T: DeserializeOwned>(params: Value) -> Result<T, RpcError> {
+    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<Vec<Message>>,
+    pub stream: Option<bool>,
+}
+
+#[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<bool>,
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields, rename_all = "camelCase")]
+pub(super) struct ApprovalResume {
+    pub approval_id: String,
+    pub stream: Option<bool>,
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields, rename_all = "camelCase")]
+pub(super) struct Resolve {
+    pub approval_id: String,
+    pub decision: String,
+    pub reason: Option<String>,
+}
+
+impl Resolve {
+    pub fn decision(&self) -> Result<ApprovalDecision, RpcError> {
+        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::<Resolve>(params).unwrap().decision().is_err());
+        }
+        assert!(decode::<Start>(json!({"task":"x","apiKey":"secret"})).is_err());
+    }
+}

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<dyn std::error::Error>> {
     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::<Vec<_>>();
+        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<AgentHost, Box<dyn std::error::Error>> {
-    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<AgentHost, Box<dyn std::error::Error>> {
+    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<AgentHost, Box<dyn std::error::Error>> {
     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 <run_id>  # 内部 worker

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<ChildStdin>,
+    lines: Receiver<String>,
+}
+
+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<std::process::ExitStatus>;
+}
+
+impl ChildWaitTimeout for Child {
+    fn wait_timeout(&mut self, timeout: Duration) -> std::io::Result<std::process::ExitStatus> {
+        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));
+        }
+    }
+}

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<HostRunOutput, HostError> {
+        self.run_existing_with_cancellation_mode(run_id, cancellation, true)
+    }
+
     /// 执行已有 run，并允许宿主在 step 边界注入共享取消标记。
     pub fn run_existing_with_cancellation(
         &self,

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<F> 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<dyn HostStreamListener>,
+}
+
+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<Arc<dyn DurableEventListener>>,
+    stream_listener: Option<Arc<dyn HostStreamListener>>,
     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<dyn HostStreamListener>) -> Self {
+        self.stream_listener = Some(listener);
+        self
+    }
+
+    /// `HostStreamListener` 的闭包便捷入口。
+    pub fn with_stream_callback<F>(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 {

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::<String>(),
+        "你好"
+    );
+    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);
+}

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/末条消息内容去重的推断。

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，不是“等待运行完成”。

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；