From 3ba935367ad72d0464fca054800d2ccbbc8521c1 Mon Sep 17 00:00:00 2001 From: kdletters Date: Wed, 9 Sep 2026 19:49:29 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8B=86=E5=88=86=20Host=20=E6=A3=80=E6=9F=A5?= =?UTF-8?q?=E7=82=B9=E4=B8=8E=E4=BA=8B=E4=BB=B6=E6=8A=95=E5=BD=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将 DurableCheckpoints、RuntimeTraceProgress 和 runtime trace 投影移入 checkpoint 模块 保持 Engine 执行、Runtime reducer 和公开 Host API 边界不变 补齐拆分后的 workspace、Clippy、Rustdoc、编码与差异验证 --- rust/README.md | 3 +- rust/crates/agent-host/src/checkpoint.rs | 636 ++++++++++++++++++ rust/crates/agent-host/src/lib.rs | 633 +---------------- .../【任务】Agent内核落地TODO-2026-09-01.md | 4 +- ...通用Agent内核与单Agent程序建设计划-2026-09-02.md | 6 +- 5 files changed, 655 insertions(+), 627 deletions(-) create mode 100644 rust/crates/agent-host/src/checkpoint.rs diff --git a/rust/README.md b/rust/README.md index 3ff3724b6..18b1f2d88 100644 --- a/rust/README.md +++ b/rust/README.md @@ -13,7 +13,8 @@ Host 的 durable 控制面已进一步下沉到 `agent-runtime-sqlite::RuntimeSe 审批决议和 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 和 Codex server-request handler 仍在根编排层。 +re-export 稳定类型；`checkpoint.rs` 承接 checkpoint listener/trace，执行循环和 Codex +server-request handler 仍在根编排层。 公开许可证、registry、自动 webhook、跨主机调度及全量 Codex schema 不属于本期完成门槛， 以权威计划「原始范围复核」为准，不采用下方历史增量中的扩大范围表述。 diff --git a/rust/crates/agent-host/src/checkpoint.rs b/rust/crates/agent-host/src/checkpoint.rs new file mode 100644 index 000000000..7c22b9f0e --- /dev/null +++ b/rust/crates/agent-host/src/checkpoint.rs @@ -0,0 +1,636 @@ +//! Checkpoint and runtime trace projection for AgentHost. +//! +//! This module owns the Engine checkpoint listener and deterministic trace +//! projection. Runtime reducer primitives remain in the parent module because +//! setup and cancellation also use them. + +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; + +use super::{ + CollectedEvents, HostError, commit_runtime_event, host_error_from_runtime, reduce_runtime_event, +}; +use agent_runtime_core::{ + ContentPart, Message, MessageRole, RuntimeEvent, RuntimeSnapshot, SystemClock, ToolCall, +}; +use agent_runtime_engine::{CheckpointListener, EngineCheckpoint, EngineEvent}; +use agent_runtime_sqlite::{ + CheckpointRecord, NewCheckpoint, NewToolCall, RuntimeService, WorkerLease, +}; + +/// 把 Engine 的边界检查点写进当前 run 的 fenced lease。 +/// +/// 回调发生在 Provider/工具调用前后，若写入失败就让 Engine 停止；这样 +/// 宿主不会在没有 durable 游标的情况下继续触发外部副作用。 +pub(super) struct DurableCheckpoints<'a> { + pub(super) runtime: RuntimeService, + pub(super) runtime_id: String, + pub(super) run_id: String, + pub(super) session_id: String, + pub(super) lease: WorkerLease, + pub(super) attempt: i64, + pub(super) pending_tool_calls: Arc>>, + pub(super) observed: &'a CollectedEvents, + pub(super) trace_progress: Mutex, + /// Checkpoints are delivered synchronously, but the listener itself is + /// shared behind `&self`; keep the compaction edge idempotent so a retry + /// cannot append a second Core event for the same compression window. + pub(super) compaction_open: Mutex, +} + +/// 两个位置都属于本次 Engine 尝试的观察序列，不按内容对历史消息去重。 +/// checkpoint 可以先投影消息，工具结果索引则仍需后续 trace 消费；压缩前 +/// 消费完成的事件位置也要保留，不能在上下文替换后重放旧结果。 +#[derive(Default)] +pub(super) struct RuntimeTraceProgress { + pub(super) events_through: usize, + pub(super) messages_through: usize, +} + +impl CheckpointListener for DurableCheckpoints<'_> { + fn on_checkpoint(&self, checkpoint: &EngineCheckpoint) -> Result<(), String> { + let input = self.new_checkpoint(checkpoint)?; + if checkpoint.phase == agent_runtime_engine::CheckpointPhase::AwaitingApproval + && let Some(call) = self.pending_tool_call(checkpoint.tool_call_id.as_deref())? + { + return self.save_approval_checkpoint(input, call); + } + if checkpoint.phase == agent_runtime_engine::CheckpointPhase::Compacting { + return self.save_compaction_started(input); + } + if self.compaction_is_open()? { + return self.save_compaction_completed(input); + } + self.runtime + .save_checkpoint_with_lease(input, &self.lease) + .map(|_| ()) + .map_err(|error| error.to_string()) + } +} + +impl DurableCheckpoints<'_> { + fn mark_messages_projected(&self) -> Result<(), String> { + let event_count = self + .observed + .events + .lock() + .map_err(|_| "Engine 事件收集器锁已损坏".to_owned())? + .len(); + self.trace_progress + .lock() + .map_err(|_| "Runtime 投影游标锁已损坏".to_owned())? + .messages_through = event_count; + Ok(()) + } + + pub(super) fn persist_trace( + &self, + snapshot: &mut RuntimeSnapshot, + events: &[EngineEvent], + ) -> Result<(), HostError> { + let mut progress = self + .trace_progress + .lock() + .map_err(|_| HostError::Config("Runtime 投影游标锁已损坏".to_owned()))?; + persist_runtime_trace( + &self.runtime, + snapshot, + &self.runtime_id, + &self.run_id, + &self.lease, + events, + &mut progress, + ) + } + + fn pending_tool_call(&self, call_id: Option<&str>) -> Result, String> { + let Some(call_id) = call_id else { + return Ok(None); + }; + self.pending_tool_calls + .lock() + .map(|pending| pending.get(call_id).cloned()) + .map_err(|_| "工具调用观察器锁已损坏".to_owned()) + } + + fn clear_pending_tool_call(&self, call_id: &str) -> Result<(), String> { + self.pending_tool_calls + .lock() + .map(|mut pending| { + pending.remove(call_id); + }) + .map_err(|_| "工具调用观察器锁已损坏".to_owned()) + } + + /// The first AwaitingApproval checkpoint is the last durable boundary + /// before ApprovalPolicy (and therefore before a possible side effect). + /// Persist the requested tool row, runtime ToolCallRequested event and + /// checkpoint together so a crash cannot leave an approval cursor with no + /// durable call identity. Later checkpoints for an already materialized + /// call keep the ordinary fenced checkpoint path. + fn save_approval_checkpoint(&self, input: NewCheckpoint, call: ToolCall) -> Result<(), String> { + if input.tool_call_id.as_deref() != Some(call.id()) { + return Err("工具调用观察到的 call id 与 checkpoint 不一致".to_owned()); + } + let snapshot = self + .runtime + .load_runtime_snapshot(&self.runtime_id) + .map_err(|error| error.to_string())? + .ok_or_else(|| format!("runtime 中找不到 run: {}", self.run_id))?; + if snapshot.run(&self.run_id).is_some_and(|run| { + run.tool_calls() + .iter() + .any(|existing| existing.id() == call.id()) + }) { + // Resume paths may already have materialized this call in an + // earlier attempt. Do not manufacture a duplicate reducer event. + self.runtime + .save_checkpoint_with_lease(input, &self.lease) + .map(|_| ()) + .map_err(|error| error.to_string())?; + self.clear_pending_tool_call(call.id())?; + return Ok(()); + } + let checkpoint_messages = serde_json::from_value::>(input.messages.clone()) + .map_err(|error| format!("审批 checkpoint 消息无法解码: {error}"))?; + let run = snapshot + .run(&self.run_id) + .ok_or_else(|| format!("runtime 中找不到 run: {}", self.run_id))?; + if run.messages().len() > checkpoint_messages.len() + || run + .messages() + .iter() + .zip(&checkpoint_messages) + .any(|(current, checkpoint)| current != checkpoint) + { + return Err("runtime 消息不是审批 checkpoint 的完整前缀".to_owned()); + } + + // The checkpoint already contains the assistant message carrying the + // model's tool-call batch. Materialize any missing message prefix in + // this same transaction before the first ToolCallRequested event; + // later calls remain only in the checkpoint until their own approval. + let mut next = snapshot.clone(); + let mut runtime_events = Vec::new(); + for message in &checkpoint_messages[run.messages().len()..] { + let event = RuntimeEvent::message_appended( + self.runtime_id.clone(), + next.revision() + 1, + SystemClock.now_millis(), + self.run_id.clone(), + message, + ) + .map_err(|error| error.to_string())?; + next = reduce_runtime_event(&next, &event).map_err(|error| error.to_string())?; + runtime_events.push(event); + } + let event = RuntimeEvent::tool_call_requested( + self.runtime_id.clone(), + next.revision() + 1, + SystemClock.now_millis(), + self.run_id.clone(), + &call, + ) + .map_err(|error| error.to_string())?; + next = reduce_runtime_event(&next, &event).map_err(|error| error.to_string())?; + runtime_events.push(event); + self.runtime + .create_tool_call_with_checkpoint_runtime_and_lease( + NewToolCall { + id: call.id().to_owned(), + session_id: self.session_id.clone(), + run_id: self.run_id.clone(), + tool_name: call.name().to_owned(), + arguments: call.arguments().clone(), + status: "requested".to_owned(), + }, + input, + &self.lease, + &self.runtime_id, + Some(snapshot.revision()), + &next, + &runtime_events, + ) + .map_err(|error| error.to_string())?; + // 只有联合提交成功，观察序列中的消息才算已经投影。最终 phase + // 会继续变成 tool/provider-in-flight 或 safe，不能拿它作去重凭据。 + self.mark_messages_projected()?; + self.clear_pending_tool_call(call.id())?; + Ok(()) + } + + fn new_checkpoint(&self, checkpoint: &EngineCheckpoint) -> Result { + Ok(NewCheckpoint { + run_id: self.run_id.clone(), + phase: checkpoint.phase.as_str().to_owned(), + step: i64::try_from(checkpoint.step) + .map_err(|_| "checkpoint step 超出 SQLite INTEGER 范围".to_owned())?, + next_step: i64::try_from(checkpoint.next_step) + .map_err(|_| "checkpoint next_step 超出 SQLite INTEGER 范围".to_owned())?, + messages: serde_json::to_value(&checkpoint.messages) + .map_err(|error| format!("检查点消息无法编码: {error}"))?, + provider_request_id: checkpoint.provider_request_id.clone(), + tool_call_id: checkpoint.tool_call_id.clone(), + attempt: self.attempt, + }) + } + + fn compaction_is_open(&self) -> Result { + self.compaction_open + .lock() + .map(|open| *open) + .map_err(|_| "compaction listener 锁已损坏".to_owned()) + } + + fn set_compaction_open(&self, value: bool) -> Result<(), String> { + let mut open = self + .compaction_open + .lock() + .map_err(|_| "compaction listener 锁已损坏".to_owned())?; + *open = value; + Ok(()) + } + + /// Persist the opening edge and its checkpoint together. This prevents a + /// stale running snapshot from being mistaken for a safe cursor after a + /// crash immediately before compression starts. + fn save_compaction_started(&self, input: NewCheckpoint) -> Result<(), String> { + if self.compaction_is_open()? { + return self + .runtime + .save_checkpoint_with_lease(input, &self.lease) + .map(|_| ()) + .map_err(|error| error.to_string()); + } + let runtime_id = self + .runtime + .runtime_id_for_run(&self.run_id) + .map_err(|error| error.to_string())? + .ok_or_else(|| format!("run 缺少 runtime 身份: {}", self.run_id))?; + let mut snapshot = self + .runtime + .load_runtime_snapshot(&runtime_id) + .map_err(|error| error.to_string())? + .ok_or_else(|| format!("runtime 中找不到 run: {}", self.run_id))?; + // 压缩将替换消息及派生工具索引。先完成旧上下文的事件/工具行， + // 再进入 Compacting；否则收尾 trace 会把已压缩的旧消息重新追加。 + let observed = self + .observed + .snapshot() + .map_err(|error| error.to_string())?; + self.persist_trace(&mut snapshot, &observed) + .map_err(|error| error.to_string())?; + if snapshot + .run(&self.run_id) + .is_some_and(|run| run.status() == agent_runtime_core::RunStatus::Compacting) + { + self.set_compaction_open(true)?; + self.runtime + .save_checkpoint_with_lease(input, &self.lease) + .map(|_| ()) + .map_err(|error| error.to_string())?; + return Ok(()); + } + let event = RuntimeEvent::compaction_started( + snapshot.runtime_id().to_owned(), + snapshot.revision() + 1, + SystemClock.now_millis(), + self.run_id.clone(), + ) + .map_err(|error| error.to_string())?; + let next = reduce_runtime_event(&snapshot, &event).map_err(|error| error.to_string())?; + self.runtime + .save_checkpoint_with_runtime_and_lease( + input, + &self.lease, + snapshot.runtime_id(), + Some(snapshot.revision()), + &next, + std::slice::from_ref(&event), + ) + .map_err(|error| error.to_string())?; + self.set_compaction_open(true)?; + Ok(()) + } + + /// Atomically rewrite the Core message context, close the compaction + /// state, and save the first post-compression checkpoint. The checkpoint + /// is the source of truth for the exact compressed history; the Engine + /// event remains observational and is deliberately not re-applied later. + fn save_compaction_completed(&self, input: NewCheckpoint) -> Result<(), String> { + let runtime_id = self + .runtime + .runtime_id_for_run(&self.run_id) + .map_err(|error| error.to_string())? + .ok_or_else(|| format!("run 缺少 runtime 身份: {}", self.run_id))?; + let snapshot = self + .runtime + .load_runtime_snapshot(&runtime_id) + .map_err(|error| error.to_string())? + .ok_or_else(|| format!("runtime 中找不到 run: {}", self.run_id))?; + let Some(run) = snapshot.run(&self.run_id) else { + return Err(format!("runtime 中找不到 run: {}", self.run_id)); + }; + if run.status() != agent_runtime_core::RunStatus::Compacting { + // A retry after a successfully committed atomic edge only needs + // the ordinary checkpoint write; never append duplicate events. + self.set_compaction_open(false)?; + return self + .runtime + .save_checkpoint_with_lease(input, &self.lease) + .map(|_| ()) + .map_err(|error| error.to_string()); + } + let messages = serde_json::from_value::>(input.messages.clone()) + .map_err(|error| format!("压缩 checkpoint 消息无效: {error}"))?; + let context_event = RuntimeEvent::context_compacted( + snapshot.runtime_id().to_owned(), + snapshot.revision() + 1, + SystemClock.now_millis(), + self.run_id.clone(), + usize::try_from(input.step).map_err(|_| "压缩 step 无效".to_owned())?, + &messages, + ) + .map_err(|error| error.to_string())?; + let compacted = + reduce_runtime_event(&snapshot, &context_event).map_err(|error| error.to_string())?; + let completed_event = RuntimeEvent::compaction_completed( + compacted.runtime_id().to_owned(), + compacted.revision() + 1, + SystemClock.now_millis(), + self.run_id.clone(), + ) + .map_err(|error| error.to_string())?; + let next = reduce_runtime_event(&compacted, &completed_event) + .map_err(|error| error.to_string())?; + let events = [context_event, completed_event]; + self.runtime + .save_checkpoint_with_runtime_and_lease( + input, + &self.lease, + snapshot.runtime_id(), + Some(snapshot.revision()), + &next, + &events, + ) + .map_err(|error| error.to_string())?; + self.set_compaction_open(false)?; + Ok(()) + } +} + +/// 把 Engine 的观察事件压缩成 Core 可重放的生命周期事件。 +/// Step/model 事件本身仍保存在 run-level 审计表，不强行扩张 Core 枚举。 +pub(super) fn persist_runtime_trace( + runtime: &RuntimeService, + snapshot: &mut RuntimeSnapshot, + runtime_id: &str, + run_id: &str, + lease: &WorkerLease, + events: &[EngineEvent], + progress: &mut RuntimeTraceProgress, +) -> Result<(), HostError> { + let session_id = runtime + .get_run(run_id) + .map_err(host_error_from_runtime)? + .ok_or_else(|| HostError::Config(format!("找不到工具调用所属 run: {run_id}")))? + .session_id; + for (event_index, event) in events.iter().enumerate().skip(progress.events_through) { + match event { + EngineEvent::ModelCompleted { response, .. } => { + let mut parts = response.content().to_vec(); + for call in response.tool_calls() { + parts.push( + ContentPart::tool_call(call.id(), call.name(), call.arguments().clone()) + .map_err(|error| HostError::Config(error.to_string()))?, + ); + } + if !parts.is_empty() && event_index >= progress.messages_through { + let message = Message::new(MessageRole::Assistant, parts); + append_runtime_message(runtime, snapshot, runtime_id, run_id, message)?; + } + } + EngineEvent::ToolRequested { call, .. } => { + // The checkpoint listener closes the pre-approval boundary by + // atomically materializing this call when possible. The + // Engine trace is replayed after the run returns, so avoid a + // second reducer event for that already committed identity. + if runtime + .get_tool_call(call.id()) + .map_err(host_error_from_runtime)? + .is_some_and(|record| record.run_id == run_id) + && snapshot.run(run_id).is_some_and(|run| { + run.tool_calls().iter().any(|item| item.id() == call.id()) + }) + { + progress.events_through = event_index + 1; + continue; + } + let runtime_event = RuntimeEvent::tool_call_requested( + runtime_id, + snapshot.revision() + 1, + SystemClock.now_millis(), + run_id, + call, + ) + .map_err(|error| HostError::Config(error.to_string()))?; + let next = reduce_runtime_event(snapshot, &runtime_event)?; + runtime + .create_tool_call_with_runtime_and_lease( + NewToolCall { + id: call.id().to_owned(), + session_id: session_id.clone(), + run_id: run_id.to_owned(), + tool_name: call.name().to_owned(), + arguments: call.arguments().clone(), + status: "requested".to_owned(), + }, + Some(lease), + runtime_id, + Some(snapshot.revision()), + &next, + std::slice::from_ref(&runtime_event), + ) + .map_err(host_error_from_runtime)?; + *snapshot = next; + } + EngineEvent::ToolCompleted { result, .. } => { + let call = snapshot + .run(run_id) + .and_then(|run| { + run.tool_calls() + .iter() + .find(|call| call.id() == result.call_id()) + }) + .ok_or_else(|| { + HostError::Config(format!( + "tool result 缺少对应调用记录: {}", + result.call_id() + )) + })?; + let runtime_event = RuntimeEvent::tool_result( + runtime_id, + snapshot.revision() + 1, + SystemClock.now_millis(), + run_id, + result, + result.is_error(), + ) + .map_err(|error| HostError::Config(error.to_string()))?; + let tool_message = Message::new( + MessageRole::Tool, + vec![ + ContentPart::tool_result( + result.call_id(), + result.output().clone(), + result.is_error(), + ) + .map_err(|error| HostError::Config(error.to_string()))?, + ], + ); + let mut next = reduce_runtime_event(snapshot, &runtime_event)?; + let mut runtime_events = vec![runtime_event]; + // checkpoint 已投影的消息无需重复追加，但 ToolResult 事件 + // 与工具行仍要落盘；两个位置不能合并成一个“已处理”标记。 + if event_index >= progress.messages_through { + let message_event = RuntimeEvent::message_appended( + runtime_id, + next.revision() + 1, + SystemClock.now_millis(), + run_id, + &tool_message, + ) + .map_err(|error| HostError::Config(error.to_string()))?; + next = reduce_runtime_event(&next, &message_event)?; + runtime_events.push(message_event); + } + let input = NewToolCall { + id: call.id().to_owned(), + session_id: session_id.clone(), + run_id: run_id.to_owned(), + tool_name: call.name().to_owned(), + arguments: call.arguments().clone(), + status: "requested".to_owned(), + }; + let status = if result.is_error() { + "error" + } else { + "completed" + }; + let checkpoint = runtime + .read_checkpoint(run_id) + .map_err(host_error_from_runtime)?; + // Engine persists one final `safe` checkpoint after the whole + // tool batch. While replaying the trace, that checkpoint can + // already contain results for later calls even though this + // Core snapshot only contains the current result. Do not + // feed that future cursor into the joint transaction for an + // intermediate call; the final ToolCompleted is the only + // point where the checkpoint and Core snapshot describe the + // same prefix. The ordinary tool/runtime transaction keeps + // the intermediate Core projection atomic without rewriting + // the batch's already-durable safe boundary. + if let Some(checkpoint) = + checkpoint.filter(|_| is_final_tool_completion(events, event_index)) + { + // The Engine has already persisted this post-result cursor. + // Reusing the exact durable value lets the result/runtime + // transition and cursor update share one fenced transaction. + runtime + .complete_tool_call_with_checkpoint_runtime_and_lease( + input, + checkpoint_input_from_record(checkpoint), + lease, + runtime_id, + Some(snapshot.revision()), + &next, + &runtime_events, + status, + result.output().clone(), + ) + .map_err(host_error_from_runtime)?; + } else { + // Intermediate batch results, and legacy runs without a + // checkpoint, use the narrower tool/runtime transaction. + // The durable batch checkpoint is left untouched until + // the final result is projected. + runtime + .complete_tool_call_with_runtime_and_lease( + input, + Some(lease), + runtime_id, + Some(snapshot.revision()), + &next, + &runtime_events, + status, + result.output().clone(), + ) + .map_err(host_error_from_runtime)?; + } + *snapshot = next; + } + // The checkpoint listener projects the compaction lifecycle and + // the exact context rewrite atomically with the next durable + // checkpoint. Keep these observations in the run-level audit + // stream but do not append a second Core event here. + EngineEvent::CompactionStarted { .. } + | EngineEvent::ContextCompacted { .. } + | EngineEvent::CompactionCompleted { .. } => {} + EngineEvent::ApprovalDenied { call_id, .. } => { + // Engine 已把拒绝作为失败 ToolResult 回填；这里无需重复追加事件。 + let _ = call_id; + } + // Finished 的 Core 事件只由 run/session/checkpoint 终态事务提交。 + EngineEvent::Finished { .. } | EngineEvent::StepStarted { .. } => {} + } + // 前一条提交完成才推进；压缩前 flush 或收尾失败后都从已提交位置 + // 继续，不让一次部分失败重复消费更早的工具结果。 + progress.events_through = event_index + 1; + } + Ok(()) +} + +/// The Engine writes the batch's `safe` checkpoint only after its final tool +/// result. During Host trace replay, earlier `ToolCompleted` events must not +/// reuse that future checkpoint in the checkpoint/runtime joint transaction. +/// Keep this decision local to the trace so no second public persistence API +/// is needed. +pub(super) fn is_final_tool_completion(events: &[EngineEvent], event_index: usize) -> bool { + !events[event_index + 1..] + .iter() + .any(|event| matches!(event, EngineEvent::ToolCompleted { .. })) +} + +/// Convert the durable checkpoint projection back into the owned command used +/// by the joint tool/checkpoint transaction. The conversion is lossless; the +/// timestamp is intentionally omitted because the adapter assigns it on write. +pub(super) fn checkpoint_input_from_record(record: CheckpointRecord) -> NewCheckpoint { + NewCheckpoint { + run_id: record.run_id, + phase: record.phase, + step: record.step, + next_step: record.next_step, + messages: record.messages, + provider_request_id: record.provider_request_id, + tool_call_id: record.tool_call_id, + attempt: record.attempt, + } +} + +pub(super) fn append_runtime_message( + runtime: &RuntimeService, + snapshot: &mut RuntimeSnapshot, + runtime_id: &str, + run_id: &str, + message: Message, +) -> Result<(), HostError> { + let runtime_event = RuntimeEvent::message_appended( + runtime_id, + snapshot.revision() + 1, + SystemClock.now_millis(), + run_id, + &message, + ) + .map_err(|error| HostError::Config(error.to_string()))?; + commit_runtime_event(runtime, snapshot, runtime_event) +} diff --git a/rust/crates/agent-host/src/lib.rs b/rust/crates/agent-host/src/lib.rs index df97da551..1831966b3 100644 --- a/rust/crates/agent-host/src/lib.rs +++ b/rust/crates/agent-host/src/lib.rs @@ -32,24 +32,26 @@ use agent_runtime_core::{ }; use agent_runtime_engine::{ AgentEngine, AgentInput, AgentOutput, AllowList, ApprovalResume, Cancellation, - CheckpointListener, ContextCompressor, EchoProvider, EngineCheckpoint, EngineError, - EngineEvent, EventListener, OwnedProviderContextCompressor, validate_tool_arguments, + ContextCompressor, EchoProvider, EngineError, EngineEvent, EventListener, + OwnedProviderContextCompressor, validate_tool_arguments, }; use agent_runtime_sqlite::{ ApprovalRecord, CheckpointRecord, EventRecord, ExternalSessionRecord, - MAX_EXTERNAL_SESSION_SCAN_LIMIT, NewApproval, NewCheckpoint, NewEvent, NewExternalSession, - NewToolCall, RunRecord, RuntimeService, RuntimeServiceError, SessionRecord, SqliteStore, - StorageError, ToolCallRecord, WorkerLease, + MAX_EXTERNAL_SESSION_SCAN_LIMIT, NewApproval, NewEvent, NewExternalSession, NewToolCall, + RunRecord, RuntimeService, RuntimeServiceError, SessionRecord, SqliteStore, StorageError, + ToolCallRecord, WorkerLease, }; use agent_skills::{ActivatedSkill, SkillLoader}; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use thiserror::Error; +mod checkpoint; mod context; mod external; mod mcp; mod tools; +use checkpoint::{DurableCheckpoints, RuntimeTraceProgress, checkpoint_input_from_record}; use context::SkillActivationContextSource; pub use context::SkillContextSource; pub use external::{ @@ -261,367 +263,6 @@ where } } -/// 把 Engine 的边界检查点写进当前 run 的 fenced lease。 -/// -/// 回调发生在 Provider/工具调用前后，若写入失败就让 Engine 停止；这样 -/// 宿主不会在没有 durable 游标的情况下继续触发外部副作用。 -struct DurableCheckpoints<'a> { - runtime: RuntimeService, - runtime_id: String, - run_id: String, - session_id: String, - lease: WorkerLease, - attempt: i64, - pending_tool_calls: Arc>>, - observed: &'a CollectedEvents, - trace_progress: Mutex, - /// Checkpoints are delivered synchronously, but the listener itself is - /// shared behind `&self`; keep the compaction edge idempotent so a retry - /// cannot append a second Core event for the same compression window. - compaction_open: Mutex, -} - -/// 两个位置都属于本次 Engine 尝试的观察序列，不按内容对历史消息去重。 -/// checkpoint 可以先投影消息，工具结果索引则仍需后续 trace 消费；压缩前 -/// 消费完成的事件位置也要保留，不能在上下文替换后重放旧结果。 -#[derive(Default)] -struct RuntimeTraceProgress { - events_through: usize, - messages_through: usize, -} - -impl CheckpointListener for DurableCheckpoints<'_> { - fn on_checkpoint(&self, checkpoint: &EngineCheckpoint) -> Result<(), String> { - let input = self.new_checkpoint(checkpoint)?; - if checkpoint.phase == agent_runtime_engine::CheckpointPhase::AwaitingApproval - && let Some(call) = self.pending_tool_call(checkpoint.tool_call_id.as_deref())? - { - return self.save_approval_checkpoint(input, call); - } - if checkpoint.phase == agent_runtime_engine::CheckpointPhase::Compacting { - return self.save_compaction_started(input); - } - if self.compaction_is_open()? { - return self.save_compaction_completed(input); - } - self.runtime - .save_checkpoint_with_lease(input, &self.lease) - .map(|_| ()) - .map_err(|error| error.to_string()) - } -} - -impl DurableCheckpoints<'_> { - fn mark_messages_projected(&self) -> Result<(), String> { - let event_count = self - .observed - .events - .lock() - .map_err(|_| "Engine 事件收集器锁已损坏".to_owned())? - .len(); - self.trace_progress - .lock() - .map_err(|_| "Runtime 投影游标锁已损坏".to_owned())? - .messages_through = event_count; - Ok(()) - } - - fn persist_trace( - &self, - snapshot: &mut RuntimeSnapshot, - events: &[EngineEvent], - ) -> Result<(), HostError> { - let mut progress = self - .trace_progress - .lock() - .map_err(|_| HostError::Config("Runtime 投影游标锁已损坏".to_owned()))?; - persist_runtime_trace( - &self.runtime, - snapshot, - &self.runtime_id, - &self.run_id, - &self.lease, - events, - &mut progress, - ) - } - - fn pending_tool_call(&self, call_id: Option<&str>) -> Result, String> { - let Some(call_id) = call_id else { - return Ok(None); - }; - self.pending_tool_calls - .lock() - .map(|pending| pending.get(call_id).cloned()) - .map_err(|_| "工具调用观察器锁已损坏".to_owned()) - } - - fn clear_pending_tool_call(&self, call_id: &str) -> Result<(), String> { - self.pending_tool_calls - .lock() - .map(|mut pending| { - pending.remove(call_id); - }) - .map_err(|_| "工具调用观察器锁已损坏".to_owned()) - } - - /// The first AwaitingApproval checkpoint is the last durable boundary - /// before ApprovalPolicy (and therefore before a possible side effect). - /// Persist the requested tool row, runtime ToolCallRequested event and - /// checkpoint together so a crash cannot leave an approval cursor with no - /// durable call identity. Later checkpoints for an already materialized - /// call keep the ordinary fenced checkpoint path. - fn save_approval_checkpoint(&self, input: NewCheckpoint, call: ToolCall) -> Result<(), String> { - if input.tool_call_id.as_deref() != Some(call.id()) { - return Err("工具调用观察到的 call id 与 checkpoint 不一致".to_owned()); - } - let snapshot = self - .runtime - .load_runtime_snapshot(&self.runtime_id) - .map_err(|error| error.to_string())? - .ok_or_else(|| format!("runtime 中找不到 run: {}", self.run_id))?; - if snapshot.run(&self.run_id).is_some_and(|run| { - run.tool_calls() - .iter() - .any(|existing| existing.id() == call.id()) - }) { - // Resume paths may already have materialized this call in an - // earlier attempt. Do not manufacture a duplicate reducer event. - self.runtime - .save_checkpoint_with_lease(input, &self.lease) - .map(|_| ()) - .map_err(|error| error.to_string())?; - self.clear_pending_tool_call(call.id())?; - return Ok(()); - } - let checkpoint_messages = serde_json::from_value::>(input.messages.clone()) - .map_err(|error| format!("审批 checkpoint 消息无法解码: {error}"))?; - let run = snapshot - .run(&self.run_id) - .ok_or_else(|| format!("runtime 中找不到 run: {}", self.run_id))?; - if run.messages().len() > checkpoint_messages.len() - || run - .messages() - .iter() - .zip(&checkpoint_messages) - .any(|(current, checkpoint)| current != checkpoint) - { - return Err("runtime 消息不是审批 checkpoint 的完整前缀".to_owned()); - } - - // The checkpoint already contains the assistant message carrying the - // model's tool-call batch. Materialize any missing message prefix in - // this same transaction before the first ToolCallRequested event; - // later calls remain only in the checkpoint until their own approval. - let mut next = snapshot.clone(); - let mut runtime_events = Vec::new(); - for message in &checkpoint_messages[run.messages().len()..] { - let event = RuntimeEvent::message_appended( - self.runtime_id.clone(), - next.revision() + 1, - SystemClock.now_millis(), - self.run_id.clone(), - message, - ) - .map_err(|error| error.to_string())?; - next = reduce_runtime_event(&next, &event).map_err(|error| error.to_string())?; - runtime_events.push(event); - } - let event = RuntimeEvent::tool_call_requested( - self.runtime_id.clone(), - next.revision() + 1, - SystemClock.now_millis(), - self.run_id.clone(), - &call, - ) - .map_err(|error| error.to_string())?; - next = reduce_runtime_event(&next, &event).map_err(|error| error.to_string())?; - runtime_events.push(event); - self.runtime - .create_tool_call_with_checkpoint_runtime_and_lease( - NewToolCall { - id: call.id().to_owned(), - session_id: self.session_id.clone(), - run_id: self.run_id.clone(), - tool_name: call.name().to_owned(), - arguments: call.arguments().clone(), - status: "requested".to_owned(), - }, - input, - &self.lease, - &self.runtime_id, - Some(snapshot.revision()), - &next, - &runtime_events, - ) - .map_err(|error| error.to_string())?; - // 只有联合提交成功，观察序列中的消息才算已经投影。最终 phase - // 会继续变成 tool/provider-in-flight 或 safe，不能拿它作去重凭据。 - self.mark_messages_projected()?; - self.clear_pending_tool_call(call.id())?; - Ok(()) - } - - fn new_checkpoint(&self, checkpoint: &EngineCheckpoint) -> Result { - Ok(NewCheckpoint { - run_id: self.run_id.clone(), - phase: checkpoint.phase.as_str().to_owned(), - step: i64::try_from(checkpoint.step) - .map_err(|_| "checkpoint step 超出 SQLite INTEGER 范围".to_owned())?, - next_step: i64::try_from(checkpoint.next_step) - .map_err(|_| "checkpoint next_step 超出 SQLite INTEGER 范围".to_owned())?, - messages: serde_json::to_value(&checkpoint.messages) - .map_err(|error| format!("检查点消息无法编码: {error}"))?, - provider_request_id: checkpoint.provider_request_id.clone(), - tool_call_id: checkpoint.tool_call_id.clone(), - attempt: self.attempt, - }) - } - - fn compaction_is_open(&self) -> Result { - self.compaction_open - .lock() - .map(|open| *open) - .map_err(|_| "compaction listener 锁已损坏".to_owned()) - } - - fn set_compaction_open(&self, value: bool) -> Result<(), String> { - let mut open = self - .compaction_open - .lock() - .map_err(|_| "compaction listener 锁已损坏".to_owned())?; - *open = value; - Ok(()) - } - - /// Persist the opening edge and its checkpoint together. This prevents a - /// stale running snapshot from being mistaken for a safe cursor after a - /// crash immediately before compression starts. - fn save_compaction_started(&self, input: NewCheckpoint) -> Result<(), String> { - if self.compaction_is_open()? { - return self - .runtime - .save_checkpoint_with_lease(input, &self.lease) - .map(|_| ()) - .map_err(|error| error.to_string()); - } - let runtime_id = self - .runtime - .runtime_id_for_run(&self.run_id) - .map_err(|error| error.to_string())? - .ok_or_else(|| format!("run 缺少 runtime 身份: {}", self.run_id))?; - let mut snapshot = self - .runtime - .load_runtime_snapshot(&runtime_id) - .map_err(|error| error.to_string())? - .ok_or_else(|| format!("runtime 中找不到 run: {}", self.run_id))?; - // 压缩将替换消息及派生工具索引。先完成旧上下文的事件/工具行， - // 再进入 Compacting；否则收尾 trace 会把已压缩的旧消息重新追加。 - let observed = self - .observed - .snapshot() - .map_err(|error| error.to_string())?; - self.persist_trace(&mut snapshot, &observed) - .map_err(|error| error.to_string())?; - if snapshot - .run(&self.run_id) - .is_some_and(|run| run.status() == agent_runtime_core::RunStatus::Compacting) - { - self.set_compaction_open(true)?; - self.runtime - .save_checkpoint_with_lease(input, &self.lease) - .map(|_| ()) - .map_err(|error| error.to_string())?; - return Ok(()); - } - let event = RuntimeEvent::compaction_started( - snapshot.runtime_id().to_owned(), - snapshot.revision() + 1, - SystemClock.now_millis(), - self.run_id.clone(), - ) - .map_err(|error| error.to_string())?; - let next = reduce_runtime_event(&snapshot, &event).map_err(|error| error.to_string())?; - self.runtime - .save_checkpoint_with_runtime_and_lease( - input, - &self.lease, - snapshot.runtime_id(), - Some(snapshot.revision()), - &next, - std::slice::from_ref(&event), - ) - .map_err(|error| error.to_string())?; - self.set_compaction_open(true)?; - Ok(()) - } - - /// Atomically rewrite the Core message context, close the compaction - /// state, and save the first post-compression checkpoint. The checkpoint - /// is the source of truth for the exact compressed history; the Engine - /// event remains observational and is deliberately not re-applied later. - fn save_compaction_completed(&self, input: NewCheckpoint) -> Result<(), String> { - let runtime_id = self - .runtime - .runtime_id_for_run(&self.run_id) - .map_err(|error| error.to_string())? - .ok_or_else(|| format!("run 缺少 runtime 身份: {}", self.run_id))?; - let snapshot = self - .runtime - .load_runtime_snapshot(&runtime_id) - .map_err(|error| error.to_string())? - .ok_or_else(|| format!("runtime 中找不到 run: {}", self.run_id))?; - let Some(run) = snapshot.run(&self.run_id) else { - return Err(format!("runtime 中找不到 run: {}", self.run_id)); - }; - if run.status() != agent_runtime_core::RunStatus::Compacting { - // A retry after a successfully committed atomic edge only needs - // the ordinary checkpoint write; never append duplicate events. - self.set_compaction_open(false)?; - return self - .runtime - .save_checkpoint_with_lease(input, &self.lease) - .map(|_| ()) - .map_err(|error| error.to_string()); - } - let messages = serde_json::from_value::>(input.messages.clone()) - .map_err(|error| format!("压缩 checkpoint 消息无效: {error}"))?; - let context_event = RuntimeEvent::context_compacted( - snapshot.runtime_id().to_owned(), - snapshot.revision() + 1, - SystemClock.now_millis(), - self.run_id.clone(), - usize::try_from(input.step).map_err(|_| "压缩 step 无效".to_owned())?, - &messages, - ) - .map_err(|error| error.to_string())?; - let compacted = - reduce_runtime_event(&snapshot, &context_event).map_err(|error| error.to_string())?; - let completed_event = RuntimeEvent::compaction_completed( - compacted.runtime_id().to_owned(), - compacted.revision() + 1, - SystemClock.now_millis(), - self.run_id.clone(), - ) - .map_err(|error| error.to_string())?; - let next = reduce_runtime_event(&compacted, &completed_event) - .map_err(|error| error.to_string())?; - let events = [context_event, completed_event]; - self.runtime - .save_checkpoint_with_runtime_and_lease( - input, - &self.lease, - snapshot.runtime_id(), - Some(snapshot.revision()), - &next, - &events, - ) - .map_err(|error| error.to_string())?; - self.set_compaction_open(false)?; - Ok(()) - } -} - /// Host 对 Codex App Server server-request 的中立接线。 /// /// 这个 handler 只处理中立的 `item/tool/call` 请求：先把参数解码为 Core @@ -4740,228 +4381,6 @@ fn restore_approval_checkpoint( Ok(()) } -/// 把 Engine 的观察事件压缩成 Core 可重放的生命周期事件。 -/// Step/model 事件本身仍保存在 run-level 审计表，不强行扩张 Core 枚举。 -fn persist_runtime_trace( - runtime: &RuntimeService, - snapshot: &mut RuntimeSnapshot, - runtime_id: &str, - run_id: &str, - lease: &WorkerLease, - events: &[EngineEvent], - progress: &mut RuntimeTraceProgress, -) -> Result<(), HostError> { - let session_id = runtime - .get_run(run_id) - .map_err(host_error_from_runtime)? - .ok_or_else(|| HostError::Config(format!("找不到工具调用所属 run: {run_id}")))? - .session_id; - for (event_index, event) in events.iter().enumerate().skip(progress.events_through) { - match event { - EngineEvent::ModelCompleted { response, .. } => { - let mut parts = response.content().to_vec(); - for call in response.tool_calls() { - parts.push( - ContentPart::tool_call(call.id(), call.name(), call.arguments().clone()) - .map_err(|error| HostError::Config(error.to_string()))?, - ); - } - if !parts.is_empty() && event_index >= progress.messages_through { - let message = Message::new(MessageRole::Assistant, parts); - append_runtime_message(runtime, snapshot, runtime_id, run_id, message)?; - } - } - EngineEvent::ToolRequested { call, .. } => { - // The checkpoint listener closes the pre-approval boundary by - // atomically materializing this call when possible. The - // Engine trace is replayed after the run returns, so avoid a - // second reducer event for that already committed identity. - if runtime - .get_tool_call(call.id()) - .map_err(host_error_from_runtime)? - .is_some_and(|record| record.run_id == run_id) - && snapshot.run(run_id).is_some_and(|run| { - run.tool_calls().iter().any(|item| item.id() == call.id()) - }) - { - progress.events_through = event_index + 1; - continue; - } - let runtime_event = RuntimeEvent::tool_call_requested( - runtime_id, - snapshot.revision() + 1, - SystemClock.now_millis(), - run_id, - call, - ) - .map_err(|error| HostError::Config(error.to_string()))?; - let next = reduce_runtime_event(snapshot, &runtime_event)?; - runtime - .create_tool_call_with_runtime_and_lease( - NewToolCall { - id: call.id().to_owned(), - session_id: session_id.clone(), - run_id: run_id.to_owned(), - tool_name: call.name().to_owned(), - arguments: call.arguments().clone(), - status: "requested".to_owned(), - }, - Some(lease), - runtime_id, - Some(snapshot.revision()), - &next, - std::slice::from_ref(&runtime_event), - ) - .map_err(host_error_from_runtime)?; - *snapshot = next; - } - EngineEvent::ToolCompleted { result, .. } => { - let call = snapshot - .run(run_id) - .and_then(|run| { - run.tool_calls() - .iter() - .find(|call| call.id() == result.call_id()) - }) - .ok_or_else(|| { - HostError::Config(format!( - "tool result 缺少对应调用记录: {}", - result.call_id() - )) - })?; - let runtime_event = RuntimeEvent::tool_result( - runtime_id, - snapshot.revision() + 1, - SystemClock.now_millis(), - run_id, - result, - result.is_error(), - ) - .map_err(|error| HostError::Config(error.to_string()))?; - let tool_message = Message::new( - MessageRole::Tool, - vec![ - ContentPart::tool_result( - result.call_id(), - result.output().clone(), - result.is_error(), - ) - .map_err(|error| HostError::Config(error.to_string()))?, - ], - ); - let mut next = reduce_runtime_event(snapshot, &runtime_event)?; - let mut runtime_events = vec![runtime_event]; - // checkpoint 已投影的消息无需重复追加，但 ToolResult 事件 - // 与工具行仍要落盘；两个位置不能合并成一个“已处理”标记。 - if event_index >= progress.messages_through { - let message_event = RuntimeEvent::message_appended( - runtime_id, - next.revision() + 1, - SystemClock.now_millis(), - run_id, - &tool_message, - ) - .map_err(|error| HostError::Config(error.to_string()))?; - next = reduce_runtime_event(&next, &message_event)?; - runtime_events.push(message_event); - } - let input = NewToolCall { - id: call.id().to_owned(), - session_id: session_id.clone(), - run_id: run_id.to_owned(), - tool_name: call.name().to_owned(), - arguments: call.arguments().clone(), - status: "requested".to_owned(), - }; - let status = if result.is_error() { - "error" - } else { - "completed" - }; - let checkpoint = runtime - .read_checkpoint(run_id) - .map_err(host_error_from_runtime)?; - // Engine persists one final `safe` checkpoint after the whole - // tool batch. While replaying the trace, that checkpoint can - // already contain results for later calls even though this - // Core snapshot only contains the current result. Do not - // feed that future cursor into the joint transaction for an - // intermediate call; the final ToolCompleted is the only - // point where the checkpoint and Core snapshot describe the - // same prefix. The ordinary tool/runtime transaction keeps - // the intermediate Core projection atomic without rewriting - // the batch's already-durable safe boundary. - if let Some(checkpoint) = - checkpoint.filter(|_| is_final_tool_completion(events, event_index)) - { - // The Engine has already persisted this post-result cursor. - // Reusing the exact durable value lets the result/runtime - // transition and cursor update share one fenced transaction. - runtime - .complete_tool_call_with_checkpoint_runtime_and_lease( - input, - checkpoint_input_from_record(checkpoint), - lease, - runtime_id, - Some(snapshot.revision()), - &next, - &runtime_events, - status, - result.output().clone(), - ) - .map_err(host_error_from_runtime)?; - } else { - // Intermediate batch results, and legacy runs without a - // checkpoint, use the narrower tool/runtime transaction. - // The durable batch checkpoint is left untouched until - // the final result is projected. - runtime - .complete_tool_call_with_runtime_and_lease( - input, - Some(lease), - runtime_id, - Some(snapshot.revision()), - &next, - &runtime_events, - status, - result.output().clone(), - ) - .map_err(host_error_from_runtime)?; - } - *snapshot = next; - } - // The checkpoint listener projects the compaction lifecycle and - // the exact context rewrite atomically with the next durable - // checkpoint. Keep these observations in the run-level audit - // stream but do not append a second Core event here. - EngineEvent::CompactionStarted { .. } - | EngineEvent::ContextCompacted { .. } - | EngineEvent::CompactionCompleted { .. } => {} - EngineEvent::ApprovalDenied { call_id, .. } => { - // Engine 已把拒绝作为失败 ToolResult 回填；这里无需重复追加事件。 - let _ = call_id; - } - // Finished 的 Core 事件只由 run/session/checkpoint 终态事务提交。 - EngineEvent::Finished { .. } | EngineEvent::StepStarted { .. } => {} - } - // 前一条提交完成才推进；压缩前 flush 或收尾失败后都从已提交位置 - // 继续，不让一次部分失败重复消费更早的工具结果。 - progress.events_through = event_index + 1; - } - Ok(()) -} - -/// The Engine writes the batch's `safe` checkpoint only after its final tool -/// result. During Host trace replay, earlier `ToolCompleted` events must not -/// reuse that future checkpoint in the checkpoint/runtime joint transaction. -/// Keep this decision local to the trace so no second public persistence API -/// is needed. -fn is_final_tool_completion(events: &[EngineEvent], event_index: usize) -> bool { - !events[event_index + 1..] - .iter() - .any(|event| matches!(event, EngineEvent::ToolCompleted { .. })) -} - /// 为“没有工具调用的成功终态”预构造 Runtime 事件批次。 /// /// Provider 返回后，`provider_in_flight` checkpoint 已经是 durable 游标； @@ -5017,40 +4436,6 @@ fn build_terminal_runtime_batch( Ok((next, runtime_events)) } -/// Convert the durable checkpoint projection back into the owned command used -/// by the joint tool/checkpoint transaction. The conversion is lossless; the -/// timestamp is intentionally omitted because the adapter assigns it on write. -fn checkpoint_input_from_record(record: CheckpointRecord) -> NewCheckpoint { - NewCheckpoint { - run_id: record.run_id, - phase: record.phase, - step: record.step, - next_step: record.next_step, - messages: record.messages, - provider_request_id: record.provider_request_id, - tool_call_id: record.tool_call_id, - attempt: record.attempt, - } -} - -fn append_runtime_message( - runtime: &RuntimeService, - snapshot: &mut RuntimeSnapshot, - runtime_id: &str, - run_id: &str, - message: Message, -) -> Result<(), HostError> { - let runtime_event = RuntimeEvent::message_appended( - runtime_id, - snapshot.revision() + 1, - SystemClock.now_millis(), - run_id, - &message, - ) - .map_err(|error| HostError::Config(error.to_string()))?; - commit_runtime_event(runtime, snapshot, runtime_event) -} - /// Resolver metadata 复用 queued metadata 的大小和 secret-key 规则，避免 /// 对账状态更新把凭据写入 durable external-session 行。 fn validate_external_reconciliation_metadata(metadata: &Value) -> Result<(), HostError> { @@ -5081,6 +4466,7 @@ fn unique_suffix() -> u128 { #[cfg(test)] mod tests { + use super::checkpoint::{is_final_tool_completion, persist_runtime_trace}; use super::mcp::mcp_error_as_tool_error; use super::*; use agent_codex::{ @@ -5096,7 +4482,8 @@ mod tests { ProviderError, ProviderErrorKind, ProviderInstanceId, ProviderProtocolId, ProviderRequest, ProviderResponse, SkillActivation, SkillDefinition, ToolErrorKind, ToolSource, }; - use agent_runtime_engine::CompressionRequest; + use agent_runtime_engine::{CheckpointListener, CompressionRequest, EngineCheckpoint}; + use agent_runtime_sqlite::NewCheckpoint; use serde_json::Value; use std::sync::Barrier; use std::sync::mpsc::{self, Receiver, Sender}; diff --git a/rust/docs/【任务】Agent内核落地TODO-2026-09-01.md b/rust/docs/【任务】Agent内核落地TODO-2026-09-01.md index 04be75864..02b1c5fde 100644 --- a/rust/docs/【任务】Agent内核落地TODO-2026-09-01.md +++ b/rust/docs/【任务】Agent内核落地TODO-2026-09-01.md @@ -23,7 +23,9 @@ - [x] 拆分后 Host 80/80、Runtime control 4/4、workspace 双特性和 Clippy 通过。 - [x] `external.rs` 已承接 ExternalBackendToolExecutor 与 Codex session metadata sink；公开类型和 生命周期合同保持不变。 -- [ ] Engine execution、checkpoint/trace 与 Codex/External backend 仍在根编排层，后续按风险分批拆分。 +- [x] `checkpoint.rs` 已承接 DurableCheckpoints、checkpoint listener 和 runtime trace 投影； + 通用 reducer helper 仍在根模块供执行与恢复共享。 +- [ ] Engine execution 与 Codex server-request handler 仍在根编排层，后续按风险分批拆分。 ### 当前范围与消息一致性验收（2026-09-06） diff --git a/rust/docs/【计划】独立通用Agent内核与单Agent程序建设计划-2026-09-02.md b/rust/docs/【计划】独立通用Agent内核与单Agent程序建设计划-2026-09-02.md index 797c3adde..2b6108baa 100644 --- a/rust/docs/【计划】独立通用Agent内核与单Agent程序建设计划-2026-09-02.md +++ b/rust/docs/【计划】独立通用Agent内核与单Agent程序建设计划-2026-09-02.md @@ -302,8 +302,10 @@ system/developer 内容固定保留，外部内容标记为不可信。先做确 - [x] `external.rs` 已承接通用 ExternalBackendToolExecutor、Codex session metadata sink 及 生命周期 helper；根模块只保留 re-export 和装配调用。Host 80/80、Runtime control 4/4、 workspace 双特性、两套 Clippy、Rustdoc、fmt、编码、依赖和 diff 门禁通过。 -- [ ] `execution`、`checkpoint`、`external` 仍保留在根文件，后续拆分必须继续维持 Engine glue - 不进入 Runtime 的边界。 +- [x] `checkpoint.rs` 已承接 DurableCheckpoints、RuntimeTraceProgress、checkpoint listener 和 + runtime trace 投影；根模块保留通用 reducer helper 及执行编排，公开 API 不变。 +- [ ] `execution` 与 Codex server-request handler 仍保留在根文件，后续拆分必须继续维持 + Engine glue 不进入 Runtime 的边界。 ### 原始范围复核（2026-09-06）