From c64d98a18f72839092fa82ae808bc4ce34779423 Mon Sep 17 00:00:00 2001 From: kdletters Date: Wed, 9 Sep 2026 20:12:02 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8B=86=E5=88=86=20Host=20=E6=89=A7=E8=A1=8C?= =?UTF-8?q?=E5=85=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将 run、run_existing、streaming 与 cancellation 入口移入 execution.rs 保持 AgentHost 公开 API、Runtime/SQLite 合同和 Engine 接口不变 同步 README、架构说明与 TODO，保留跨职责终态 glue 在根模块 --- rust/README.md | 6 +- rust/crates/agent-host/src/execution.rs | 233 ++++++++++++++++++ rust/crates/agent-host/src/lib.rs | 220 +---------------- .../【任务】Agent内核落地TODO-2026-09-01.md | 5 +- .../【架构】独立Agent运行时-2026-09-01.md | 5 + 5 files changed, 246 insertions(+), 223 deletions(-) create mode 100644 rust/crates/agent-host/src/execution.rs diff --git a/rust/README.md b/rust/README.md index 89f4d2fe3..551349222 100644 --- a/rust/README.md +++ b/rust/README.md @@ -12,9 +12,9 @@ Engine 输出与 Runtime 消息，并从空快照重放事件；Fake CLI smoke Host 的 durable 控制面已进一步下沉到 `agent-runtime-sqlite::RuntimeService`：取消、无主失败收口、 审批决议和 Provider/Tool 对账由 Runtime 负责，Host 只保留薄委托；Engine 执行、checkpoint/trace、 worker 和外部工具桥仍留在 Host。该拆分不改公开 Host API、SQLite schema 或 Cargo.lock。 -Host 内部桥接按职责位于私有 `tools.rs`、`mcp.rs`、`context.rs` 和 `external.rs`，根模块显式 -re-export 稳定类型；`checkpoint.rs` 承接 checkpoint listener/trace，`codex.rs` 承接 Codex -server-request handler，执行循环仍在根编排层。 +Host 内部桥接按职责位于私有 `tools.rs`、`mcp.rs`、`context.rs`、`external.rs`、`checkpoint.rs`、 +`codex.rs` 和 `execution.rs`；根模块显式 re-export 稳定类型。`execution.rs` 只承接 run 入口、 +worker/lease 领取和执行恢复编排，checkpoint/trace 与具体适配器仍由各自模块负责。 公开许可证、registry、自动 webhook、跨主机调度及全量 Codex schema 不属于本期完成门槛， 以权威计划「原始范围复核」为准，不采用下方历史增量中的扩大范围表述。 diff --git a/rust/crates/agent-host/src/execution.rs b/rust/crates/agent-host/src/execution.rs new file mode 100644 index 000000000..cad2113c6 --- /dev/null +++ b/rust/crates/agent-host/src/execution.rs @@ -0,0 +1,233 @@ +//! AgentHost 的执行入口。 +//! +//! 这里仅组织 run 的生命周期和 worker/lease 入口；持久化投影、控制面和 +//! 外部适配器仍由父模块及其专用模块负责。 + +use super::{ + AgentHost, HostError, HostRunOutput, WORKER_LEASE_DURATION, approval_resume_from_record, + host_error_from_runtime, +}; +use agent_runtime_core::Message; +use agent_runtime_engine::{Cancellation, EngineError}; +use agent_runtime_sqlite::WorkerLease; + +impl AgentHost { + /// 执行一个新任务，并将 Engine 事件写入 SQLite。 + pub fn run(&self, task: impl Into) -> Result { + let handle = self.prepare_run(task)?; + self.run_existing_with_cancellation(&handle.run_id, Cancellation::new()) + } + + /// 运行一组已经构造好的初始消息；与 `run_prompt` 相比，这个入口允许 + /// 调用方自行决定用于展示/持久化的任务摘要。 + pub fn run_with_messages( + &self, + task: impl Into, + messages: Vec, + ) -> Result { + let handle = self.prepare_run_with_messages(task, messages)?; + self.run_existing_with_cancellation(&handle.run_id, Cancellation::new()) + } + + /// 使用 Provider 的真实流式端口运行一次新任务；同步 `run` API 保持兼容。 + pub fn run_streaming(&self, task: impl Into) -> Result { + let handle = self.prepare_run(task)?; + self.run_existing_streaming(&handle.run_id) + } + + pub fn run_with_messages_streaming( + &self, + task: impl Into, + messages: Vec, + ) -> Result { + let handle = self.prepare_run_with_messages(task, messages)?; + self.run_existing_streaming(&handle.run_id) + } + + /// 由后台 worker 使用已持久化的 run 身份执行任务。 + pub fn run_existing(&self, run_id: &str) -> Result { + self.run_existing_with_cancellation(run_id, Cancellation::new()) + } + + pub fn run_existing_streaming(&self, run_id: &str) -> Result { + self.run_existing_with_cancellation_mode(run_id, Cancellation::new(), true) + } + + /// 执行已有 run，并允许宿主在 step 边界注入共享取消标记。 + pub fn run_existing_with_cancellation( + &self, + run_id: &str, + cancellation: Cancellation, + ) -> Result { + self.run_existing_with_cancellation_mode(run_id, cancellation, false) + } + + fn run_existing_with_cancellation_mode( + &self, + run_id: &str, + cancellation: Cancellation, + streaming: bool, + ) -> Result { + let record = self + .runtime + .get_run(run_id)? + .ok_or_else(|| HostError::Config(format!("找不到指定 run: {run_id}")))?; + // reconciling 是不可执行的 recovery gate。`cancel` 会保留 + // cancel_requested 标记以阻止迟到的 approval resolve；因此必须在 + // 通用取消分支前先拒绝它，否则该标记会把 reconciling 误当成 queued + // 并被 finish_unclaimed_cancelled 终态化。 + if record.status == "reconciling" { + return Err(HostError::Config(format!( + "run 处于 reconciling，完成外部调用对账后才能继续: {run_id}" + ))); + } + if record.cancel_requested || record.status == "cancel_requested" { + // running/cancel_requested 可能来自已经退出的进程，即使旧的 + // lease 行已经被清掉也不能证明外部调用没有发生。先走 Runtime + // 的 stale probe，把它放进 reconciliation gate；只有 queued + //（尚未触发 Engine）才允许直接收束为 cancelled。 + if matches!(record.status.as_str(), "running" | "cancel_requested") { + if let Some(recovered) = self.runtime.reconcile_expired_run_if_stale(run_id)? { + return Err(HostError::Config(format!( + "取消 run 的 lease 已失效，已自动进入 reconciling（{}）；先完成外部调用对账: {run_id}", + recovered.status + ))); + } + return Err(HostError::Config(format!( + "run 的取消收口仍由持有 lease 的 worker 负责: {run_id}" + ))); + } + // 一个 queued run 尚未启动 Engine；没有 lease 时可以按显式 + // cancel 命令直接完成终态。异常残留 lease 仍拒绝越权收口。 + if self.runtime.get_run_lease(run_id)?.is_some() { + return Err(HostError::Config(format!( + "run 的取消收口仍由持有 lease 的 worker 负责: {run_id}" + ))); + } + self.finish_unclaimed_cancelled(record)?; + return Err(EngineError::Cancelled.into()); + } + if record.status == "queued" { + // Provider identity is part of the durable run contract. Validate it + // before claiming the lease so a reopened Host cannot silently use a + // different instance/protocol. + self.validate_persisted_provider_target(&record)?; + // CLI background metadata adds `providerKind` as an explicit marker; + // validate its persisted model before any lease or external call. + self.validate_persisted_provider_metadata(&record)?; + let lease = WorkerLease::new(run_id); + let (claimed, lease_record) = + self.runtime + .claim_run_with_lease(run_id, &lease, WORKER_LEASE_DURATION)?; + let checkpoint = match self.runtime.read_checkpoint_with_lease(run_id, &lease) { + Ok(checkpoint) => checkpoint, + Err(error) => { + // claim 成功后读取 checkpoint 失败也不能把 lease 留成 + // 无主 running；此时尚未启动外部调用，直接收束为 failed。 + return Err(self.fail_claimed_setup( + &claimed, + &lease, + host_error_from_runtime(error), + )); + } + }; + let approval_resume = match checkpoint.as_ref().map(|value| value.phase.as_str()) { + None | Some("safe") => None, + Some("awaiting_approval") => { + let Some(call_id) = checkpoint + .as_ref() + .and_then(|value| value.tool_call_id.as_deref()) + else { + let _ = self.runtime.release(run_id, &lease); + return Err(HostError::Config(format!( + "awaiting_approval checkpoint 缺少 tool_call_id: {run_id}" + ))); + }; + let approval = match self.runtime.get_approval_for_run_call(run_id, call_id) { + Ok(approval) => approval, + Err(error) => { + let _ = self.runtime.release(run_id, &lease); + return Err(host_error_from_runtime(error)); + } + }; + let Some(approval) = approval else { + let _ = self.runtime.release(run_id, &lease); + return Err(HostError::Config(format!( + "run 缺少 durable approval，请等待控制端 resolve: {run_id}" + ))); + }; + match approval.status.as_str() { + "allowed" | "denied" => { + match approval_resume_from_record( + &approval, + checkpoint + .as_ref() + .and_then(|value| value.provider_request_id.as_deref()), + ) { + Ok(resume) => Some(resume), + Err(error) => { + // The approval may have expired or its + // persisted binding may be corrupt. Do + // not leave the freshly claimed run in a + // worker-owned running state on this + // pre-Engine validation failure. + let _ = self.runtime.release(run_id, &lease); + return Err(error); + } + } + } + "pending" => { + let _ = self.runtime.release(run_id, &lease); + return Err(HostError::Config(format!( + "approval 尚未 resolve，不能启动 run: {}", + approval.id + ))); + } + status => { + let _ = self.runtime.release(run_id, &lease); + return Err(HostError::Config(format!( + "approval 状态 {} 不允许恢复: {}", + status, approval.id + ))); + } + } + } + Some(phase) => { + let _ = self.runtime.release(run_id, &lease); + return Err(HostError::Config(format!( + "run 有未对账的 {} checkpoint，不能启动；先完成外部调用对账: {run_id}", + phase + ))); + } + }; + return self.run_claimed_with_lease( + claimed, + lease, + lease_record.attempt, + checkpoint, + approval_resume, + cancellation, + streaming, + ); + } + if matches!(record.status.as_str(), "running" | "cancel_requested") { + // A restarted process may still leave a durable running row after + // its worker lease expires. Reconcile that stale row at the + // execution boundary so callers do not need a separate startup + // scanner; the Runtime gate never replays unknown side effects. + if let Some(recovered) = self.runtime.reconcile_expired_run_if_stale(run_id)? { + return Err(HostError::Config(format!( + "run lease 已过期，已自动进入 reconciling（{}）；先完成外部调用对账: {run_id}", + recovered.status + ))); + } + return Err(HostError::Config(format!( + "run 已被其它 worker 领取；lease 尚未过期，不能启动第二个 worker: {run_id}" + ))); + } + Err(HostError::Config(format!( + "run 当前状态 {} 不可执行: {run_id}", + record.status + ))) + } +} diff --git a/rust/crates/agent-host/src/lib.rs b/rust/crates/agent-host/src/lib.rs index 643ae4a35..9882fb684 100644 --- a/rust/crates/agent-host/src/lib.rs +++ b/rust/crates/agent-host/src/lib.rs @@ -42,6 +42,7 @@ use thiserror::Error; mod checkpoint; mod codex; mod context; +mod execution; mod external; mod mcp; mod tools; @@ -1678,225 +1679,6 @@ impl AgentHost { .map_err(host_error_from_runtime) } - /// 执行一个新任务，并将 Engine 事件写入 SQLite。 - pub fn run(&self, task: impl Into) -> Result { - let handle = self.prepare_run(task)?; - self.run_existing_with_cancellation(&handle.run_id, Cancellation::new()) - } - - /// 运行一组已经构造好的初始消息；与 `run_prompt` 相比，这个入口允许 - /// 调用方自行决定用于展示/持久化的任务摘要。 - pub fn run_with_messages( - &self, - task: impl Into, - messages: Vec, - ) -> Result { - let handle = self.prepare_run_with_messages(task, messages)?; - self.run_existing_with_cancellation(&handle.run_id, Cancellation::new()) - } - - /// 使用 Provider 的真实流式端口运行一次新任务；同步 `run` API 保持兼容。 - pub fn run_streaming(&self, task: impl Into) -> Result { - let handle = self.prepare_run(task)?; - self.run_existing_streaming(&handle.run_id) - } - - pub fn run_with_messages_streaming( - &self, - task: impl Into, - messages: Vec, - ) -> Result { - let handle = self.prepare_run_with_messages(task, messages)?; - self.run_existing_streaming(&handle.run_id) - } - - /// 由后台 worker 使用已持久化的 run 身份执行任务。 - pub fn run_existing(&self, run_id: &str) -> Result { - self.run_existing_with_cancellation(run_id, Cancellation::new()) - } - - pub fn run_existing_streaming(&self, run_id: &str) -> Result { - self.run_existing_with_cancellation_mode(run_id, Cancellation::new(), true) - } - - /// 执行已有 run，并允许宿主在 step 边界注入共享取消标记。 - pub fn run_existing_with_cancellation( - &self, - run_id: &str, - cancellation: Cancellation, - ) -> Result { - self.run_existing_with_cancellation_mode(run_id, cancellation, false) - } - - fn run_existing_with_cancellation_mode( - &self, - run_id: &str, - cancellation: Cancellation, - streaming: bool, - ) -> Result { - let record = self - .runtime - .get_run(run_id)? - .ok_or_else(|| HostError::Config(format!("找不到指定 run: {run_id}")))?; - // reconciling 是不可执行的 recovery gate。`cancel` 会保留 - // cancel_requested 标记以阻止迟到的 approval resolve；因此必须在 - // 通用取消分支前先拒绝它，否则该标记会把 reconciling 误当成 queued - // 并被 finish_unclaimed_cancelled 终态化。 - if record.status == "reconciling" { - return Err(HostError::Config(format!( - "run 处于 reconciling，完成外部调用对账后才能继续: {run_id}" - ))); - } - if record.cancel_requested || record.status == "cancel_requested" { - // running/cancel_requested 可能来自已经退出的进程，即使旧的 - // lease 行已经被清掉也不能证明外部调用没有发生。先走 Runtime - // 的 stale probe，把它放进 reconciliation gate；只有 queued - //（尚未触发 Engine）才允许直接收束为 cancelled。 - if matches!(record.status.as_str(), "running" | "cancel_requested") { - if let Some(recovered) = self.runtime.reconcile_expired_run_if_stale(run_id)? { - return Err(HostError::Config(format!( - "取消 run 的 lease 已失效，已自动进入 reconciling（{}）；先完成外部调用对账: {run_id}", - recovered.status - ))); - } - return Err(HostError::Config(format!( - "run 的取消收口仍由持有 lease 的 worker 负责: {run_id}" - ))); - } - // 一个 queued run 尚未启动 Engine；没有 lease 时可以按显式 - // cancel 命令直接完成终态。异常残留 lease 仍拒绝越权收口。 - if self.runtime.get_run_lease(run_id)?.is_some() { - return Err(HostError::Config(format!( - "run 的取消收口仍由持有 lease 的 worker 负责: {run_id}" - ))); - } - self.finish_unclaimed_cancelled(record)?; - return Err(EngineError::Cancelled.into()); - } - if record.status == "queued" { - // Provider identity is part of the durable run contract. Validate it - // before claiming the lease so a reopened Host cannot silently use a - // different instance/protocol. - self.validate_persisted_provider_target(&record)?; - // CLI background metadata adds `providerKind` as an explicit marker; - // validate its persisted model before any lease or external call. - self.validate_persisted_provider_metadata(&record)?; - let lease = WorkerLease::new(run_id); - let (claimed, lease_record) = - self.runtime - .claim_run_with_lease(run_id, &lease, WORKER_LEASE_DURATION)?; - let checkpoint = match self.runtime.read_checkpoint_with_lease(run_id, &lease) { - Ok(checkpoint) => checkpoint, - Err(error) => { - // claim 成功后读取 checkpoint 失败也不能把 lease 留成 - // 无主 running；此时尚未启动外部调用，直接收束为 failed。 - return Err(self.fail_claimed_setup( - &claimed, - &lease, - host_error_from_runtime(error), - )); - } - }; - let approval_resume = match checkpoint.as_ref().map(|value| value.phase.as_str()) { - None | Some("safe") => None, - Some("awaiting_approval") => { - let Some(call_id) = checkpoint - .as_ref() - .and_then(|value| value.tool_call_id.as_deref()) - else { - let _ = self.runtime.release(run_id, &lease); - return Err(HostError::Config(format!( - "awaiting_approval checkpoint 缺少 tool_call_id: {run_id}" - ))); - }; - let approval = match self.runtime.get_approval_for_run_call(run_id, call_id) { - Ok(approval) => approval, - Err(error) => { - let _ = self.runtime.release(run_id, &lease); - return Err(host_error_from_runtime(error)); - } - }; - let Some(approval) = approval else { - let _ = self.runtime.release(run_id, &lease); - return Err(HostError::Config(format!( - "run 缺少 durable approval，请等待控制端 resolve: {run_id}" - ))); - }; - match approval.status.as_str() { - "allowed" | "denied" => { - match approval_resume_from_record( - &approval, - checkpoint - .as_ref() - .and_then(|value| value.provider_request_id.as_deref()), - ) { - Ok(resume) => Some(resume), - Err(error) => { - // The approval may have expired or its - // persisted binding may be corrupt. Do - // not leave the freshly claimed run in a - // worker-owned running state on this - // pre-Engine validation failure. - let _ = self.runtime.release(run_id, &lease); - return Err(error); - } - } - } - "pending" => { - let _ = self.runtime.release(run_id, &lease); - return Err(HostError::Config(format!( - "approval 尚未 resolve，不能启动 run: {}", - approval.id - ))); - } - status => { - let _ = self.runtime.release(run_id, &lease); - return Err(HostError::Config(format!( - "approval 状态 {} 不允许恢复: {}", - status, approval.id - ))); - } - } - } - Some(phase) => { - let _ = self.runtime.release(run_id, &lease); - return Err(HostError::Config(format!( - "run 有未对账的 {} checkpoint，不能启动；先完成外部调用对账: {run_id}", - phase - ))); - } - }; - return self.run_claimed_with_lease( - claimed, - lease, - lease_record.attempt, - checkpoint, - approval_resume, - cancellation, - streaming, - ); - } - if matches!(record.status.as_str(), "running" | "cancel_requested") { - // A restarted process may still leave a durable running row after - // its worker lease expires. Reconcile that stale row at the - // execution boundary so callers do not need a separate startup - // scanner; the Runtime gate never replays unknown side effects. - if let Some(recovered) = self.runtime.reconcile_expired_run_if_stale(run_id)? { - return Err(HostError::Config(format!( - "run lease 已过期，已自动进入 reconciling（{}）；先完成外部调用对账: {run_id}", - recovered.status - ))); - } - return Err(HostError::Config(format!( - "run 已被其它 worker 领取；lease 尚未过期，不能启动第二个 worker: {run_id}" - ))); - } - Err(HostError::Config(format!( - "run 当前状态 {} 不可执行: {run_id}", - record.status - ))) - } - fn finish_unclaimed_cancelled(&self, record: RunRecord) -> Result<(), HostError> { self.runtime .finish_unclaimed_cancelled_if_safe(&record.id) diff --git a/rust/docs/【任务】Agent内核落地TODO-2026-09-01.md b/rust/docs/【任务】Agent内核落地TODO-2026-09-01.md index de823e5d5..9c4077d0c 100644 --- a/rust/docs/【任务】Agent内核落地TODO-2026-09-01.md +++ b/rust/docs/【任务】Agent内核落地TODO-2026-09-01.md @@ -27,7 +27,10 @@ 通用 reducer helper 仍在根模块供执行与恢复共享。 - [x] `codex.rs` 已承接两个 Codex server-request handler 与 durable tool-call helper，根路径 公开 re-export 保持兼容。 -- [ ] Engine execution 仍在根编排层，后续按风险分批拆分。 +- [x] `execution.rs` 已承接 run/run_existing、streaming、cancellation 入口及 worker/lease + 执行恢复编排；根路径 API 保持兼容，未改变 Runtime、SQLite schema 或 Engine 接口。 +- [ ] `finish_cancelled`、Engine worker 主体和跨模块终态 glue 仍在根模块；后续只在能保持边界 + 和回归证据时继续拆分，不为降低行数引入新的平行装配层。 ### 当前范围与消息一致性验收（2026-09-06） diff --git a/rust/docs/【架构】独立Agent运行时-2026-09-01.md b/rust/docs/【架构】独立Agent运行时-2026-09-01.md index 6c0ac9096..ab4d77996 100644 --- a/rust/docs/【架构】独立Agent运行时-2026-09-01.md +++ b/rust/docs/【架构】独立Agent运行时-2026-09-01.md @@ -56,6 +56,11 @@ Codex server-request handler 和 durable tool-call helper 现在位于私有 `co re-export 两个公开 handler 类型。它们同时绑定 ToolRouter、ApprovalPolicy、Runtime 记录和 版本化 wire，因此不下沉到 Runtime 或通用进程 adapter。 +Host 的 run 入口和 worker/lease 执行恢复编排位于私有 `execution.rs`。该模块只组织 +`run`/`run_existing`/streaming/cancellation 入口，并调用根模块保留的 checkpoint、Engine、 +控制面和终态 helper；它不引入新的公开 API，也不持有 SQLite 之外的状态。这样根模块逐步 +收敛为装配 facade 与跨职责 glue，后续仍可按边界继续拆分，但不以机械搬迁改变行为。 + ## 当前实现顺序 1. Core 契约和纯 reducer；