From 377bb136ae2b57f876c40a1efbcad2a273fc69b3 Mon Sep 17 00:00:00 2001 From: limityan Date: Wed, 26 Aug 2026 18:37:23 +0800 Subject: [PATCH] feat(sdk): return authoritative turn settlement results --- .../agent-runtime-lifecycle-sequence.md | 3 +- .../agent-runtime-services-design.md | 7 +- src/apps/cli/src/agent/runtime_client.rs | 18 +- src/apps/cli/src/dispatch/worker.rs | 2 +- .../coordination/background_outcomes.rs | 2 + .../src/agentic/coordination/coordinator.rs | 221 +++++++++++++++++- .../src/agentic/coordination/scheduler.rs | 9 +- .../src/agentic/session/session_manager.rs | 172 +++++++++++++- .../src/agentic/session/transcript_render.rs | 27 +++ .../assembly/core/src/product_runtime.rs | 166 ++++++++++++- .../contracts/runtime-ports/src/agent_api.rs | 55 ++++- .../agent-runtime/examples/sdk_minimal.rs | 2 +- .../execution/agent-runtime/src/runtime.rs | 17 +- src/crates/execution/agent-runtime/src/sdk.rs | 27 +-- .../agent_session_contracts/sdk_smoke.rs | 2 +- .../session_operation_ports.rs | 20 +- .../app-server/src/server/handlers/session.rs | 2 +- .../app-server/tests/agent_kernel.rs | 8 +- src/crates/interfaces/sdk-host/src/host.rs | 134 +++++++---- .../sdk-host/tests/host_lifecycle.rs | 128 +++++++++- 20 files changed, 901 insertions(+), 121 deletions(-) diff --git a/docs/architecture/agent-runtime-lifecycle-sequence.md b/docs/architecture/agent-runtime-lifecycle-sequence.md index 08a981181c..e80844c7ed 100644 --- a/docs/architecture/agent-runtime-lifecycle-sequence.md +++ b/docs/architecture/agent-runtime-lifecycle-sequence.md @@ -239,7 +239,8 @@ 5. **结算同步点（[runtime.rs:1038-1051](../../src/crates/execution/agent-runtime/src/runtime.rs#L1038-L1051)）**： `wait_for_turn_settlement()` 是阻塞调用，调用方用它等 turn 进入 - `TurnOutcome::{Completed, Cancelled, Failed}`，之后才读 transcript / 算 usage。 + `TurnOutcome::{Completed, Cancelled, Failed}`，并取得 Runtime 确认的终态、最终回答和结束原因； + transcript 与 usage 仍按需通过各自的只读接口获取。 6. **post-call hooks（[post_call_hooks.rs:156-172](../../src/crates/execution/agent-runtime/src/post_call_hooks.rs#L156-L172)）**： 仅在「成功工具调用后」（`SuccessfulToolPostCall`）触发，当前唯一具体钩子是 diff --git a/docs/architecture/agent-runtime-services-design.md b/docs/architecture/agent-runtime-services-design.md index b4fc15976d..1acc2fdb84 100644 --- a/docs/architecture/agent-runtime-services-design.md +++ b/docs/architecture/agent-runtime-services-design.md @@ -68,7 +68,7 @@ Agent Runtime API 的逻辑归属与物理部署分离：相同归属模块可 私有 SDK Host 或目标机器 Runtime 中。任何 Rust 部署都只管理自己进程树内的服务与 Node/Bun Plugin Host；不能因为多个 GUI/TUI/Remote Client 连接就复制 Runtime 状态模块，或按 Client/Workspace 创建 Plugin Host。 -Rust Runtime SDK 以 `AGENT_RUNTIME_SDK_API_VERSION` 标记兼容边界。当前接口版本为 v9 preview： +Rust Runtime SDK 以 `AGENT_RUNTIME_SDK_API_VERSION` 标记兼容边界。当前接口版本为 v10 preview： 小版本更新允许增加可选 builder hook、有默认实现的端口方法或注册表查询能力，但不得向外部可用 Rust 结构体字面量（struct literal）构造的 DTO 直接增加字段，也不得改变既有端口语义、错误分类、session / turn 标识含义或 默认 feature 依赖。任何需要调用方改写现有嵌入代码的变更，必须提升接口版本并提供兼容迁移路径。 @@ -98,6 +98,9 @@ v8 增加 `TurnTokenUsage` 聚合事实。对应 SDK Host protocol v5 支持本 v9 删除从未接入真实执行路径的 Harness descriptor registry、builder 注入和查询接口。命名工作流由 Product Assembly 选择并在 `agent-workflows` / 现有产品 owner 中执行，SDK 调用方无需安装或注入另一套工作流框架。 +v10 让 Turn settlement 返回 Runtime 确认的终态、最终回答和结束原因。现有 Rust embedder 需要接收 +`AgentTurnSettlementResult`；CLI 和 App Server adapter 可在不扩展各自 wire contract 的情况下丢弃返回值。 + 只要外部调用方仍必须导入 `bitfun-core`、启用 `product-full`、持有具体服务管理器、读取产品命令 注册表、理解 ACP/内部端口或依赖全局可变状态，公开 SDK 发布边界就不成立。公开 SDK 的完整 术语、能力等价和版本要求以 [`agent-sdk-product-architecture.md`](agent-sdk-product-architecture.md) 为准。 @@ -441,7 +444,7 @@ impl AgentRuntime { 该 Rust 接口是内部产品入口复用的当前形态，不是公开 Python/TypeScript SDK 的目标 API。它必须只接收 已组装的类型化部件，不负责创建 文件系统、终端、MCP、AI 客户端、Remote 提供方或产品命令。 -当前 v9 preview 接口以 message / attachment / metadata、默认标准执行目标和活动 Turn 文本 steer 作为最小输入形态；若把 +当前 v10 preview 接口以 message / attachment / metadata、默认标准执行目标和活动 Turn 文本 steer 作为最小输入形态；若把 model-round cancellation token、结构化 AgentInput 或更复杂的事件游标纳入公开 SDK， 必须分别评审 Rust Runtime SDK、SDK Host protocol 和公开 SDK API 的版本，并保留旧路径兼容。 diff --git a/src/apps/cli/src/agent/runtime_client.rs b/src/apps/cli/src/agent/runtime_client.rs index a027c80e2f..b619d60112 100644 --- a/src/apps/cli/src/agent/runtime_client.rs +++ b/src/apps/cli/src/agent/runtime_client.rs @@ -1296,7 +1296,7 @@ impl CliAgentRuntimeClient { }; match &self.backend { CliAgentRuntimeBackend::Embedded(runtime) => { - runtime.wait_for_turn_settlement(request).await + runtime.wait_for_turn_settlement(request).await.map(|_| ()) } CliAgentRuntimeBackend::Shared(client) => { let result = client @@ -2762,10 +2762,10 @@ mod dual_backend_behavior_tests { AgentSessionWorkspaceRequest, AgentSubmissionPort, AgentSubmissionRequest, AgentSubmissionResult, AgentTurnCancellationPort, AgentTurnCancellationRequest, AgentTurnCancellationResult, AgentTurnSettlementPort, AgentTurnSettlementRequest, - AgenticEvent, DialogSubmitOutcome, PermissionReply, PermissionRequest, - PermissionRequestManager, PermissionRequestSource, PermissionRequestSourceKind, PortError, - PortErrorKind, PortResult, RuntimeError, SessionState, SessionTranscript, - SessionTranscriptReader, SessionTranscriptRequest, + AgentTurnSettlementResult, AgentTurnSettlementStatus, AgenticEvent, DialogSubmitOutcome, + PermissionReply, PermissionRequest, PermissionRequestManager, PermissionRequestSource, + PermissionRequestSourceKind, PortError, PortErrorKind, PortResult, RuntimeError, + SessionState, SessionTranscript, SessionTranscriptReader, SessionTranscriptRequest, }; use bitfun_agent_runtime_ipc::{ RuntimeInstanceIdentity, RuntimeIpcClient, RuntimeIpcClientError, RuntimeIpcErrorCode, @@ -3101,7 +3101,7 @@ mod dual_backend_behavior_tests { async fn wait_for_turn_settlement( &self, request: AgentTurnSettlementRequest, - ) -> PortResult<()> { + ) -> PortResult { match self .state .settlement_outcomes @@ -3111,7 +3111,11 @@ mod dual_backend_behavior_tests { .cloned() { Some(Some(error)) => Err(error), - _ => Ok(()), + _ => Ok(AgentTurnSettlementResult { + status: AgentTurnSettlementStatus::Completed, + final_response: Some("fixture result".to_string()), + finish_reason: Some("stop".to_string()), + }), } } } diff --git a/src/apps/cli/src/dispatch/worker.rs b/src/apps/cli/src/dispatch/worker.rs index 7fffb31e5c..a294fcefd3 100644 --- a/src/apps/cli/src/dispatch/worker.rs +++ b/src/apps/cli/src/dispatch/worker.rs @@ -400,7 +400,7 @@ async fn run_inner(store: &DispatchStore, job_id: &str) -> Result<()> { }) .await; let (terminal_state, terminal_error) = match settlement { - Ok(()) => (terminal_state, terminal_error), + Ok(_) => (terminal_state, terminal_error), Err(error) => ( DispatchJobState::Failed, Some(format!( diff --git a/src/crates/assembly/core/src/agentic/coordination/background_outcomes.rs b/src/crates/assembly/core/src/agentic/coordination/background_outcomes.rs index f90e67fabe..e218c8a82e 100644 --- a/src/crates/assembly/core/src/agentic/coordination/background_outcomes.rs +++ b/src/crates/assembly/core/src/agentic/coordination/background_outcomes.rs @@ -704,6 +704,8 @@ mod tests { "persisted child result".to_string(), &[], TurnStats::default(), + Some("complete".to_string()), + Some(true), ) .await .expect("complete child turn"); diff --git a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs index 04977940a7..382457857a 100644 --- a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs +++ b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs @@ -107,7 +107,8 @@ use bitfun_runtime_ports::{ agent_workspace_references_from_metadata, resolve_permission_mode, AgentMessageWorkspaceReferencesRequest, AgentSessionComposerUpdate, AgentSessionWorkspaceBinding, AgentThreadGoalDeliveryKind, AgentThreadGoalDeliveryRequest, - AgentWorkspaceReference, AgentWorkspaceReferenceKind, AgentWorkspaceReferenceSearchEntry, + AgentTurnSettlementResult, AgentTurnSettlementStatus, AgentWorkspaceReference, + AgentWorkspaceReferenceKind, AgentWorkspaceReferenceSearchEntry, AgentWorkspaceReferenceSearchRequest, AgentWorkspaceReferenceSearchResult, DelegationPolicy, PermissionDelegationContext, PermissionMode, PermissionModeLayers, PermissionRuntimeCeiling, RemoteExecPort, ResolvedPermissionMode, SessionStoragePathRequest, @@ -2939,6 +2940,8 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet final_response.clone(), &execution_result.new_messages, stats, + Some(execution_result.effective_finish_reason.clone()), + Some(execution_result.has_final_response), ) .await } @@ -2950,6 +2953,8 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet final_response.clone(), &execution_result.new_messages, stats, + Some(execution_result.effective_finish_reason.clone()), + Some(execution_result.has_final_response), ) .await } @@ -2979,6 +2984,24 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet } } + if persistence_succeeded { + let status = if execution_result.success && execution_result.has_final_response { + AgentTurnSettlementStatus::Completed + } else { + AgentTurnSettlementStatus::Failed + }; + session_manager.record_turn_settlement_result( + session_id, + turn_id, + AgentTurnSettlementResult { + status, + final_response: (status == AgentTurnSettlementStatus::Completed) + .then_some(final_response.clone()), + finish_reason: Some(execution_result.effective_finish_reason.clone()), + }, + ); + } + if recovery_generation.is_some() { if let Err(error) = event_queue .enqueue( @@ -3249,6 +3272,16 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet } } + session_manager.record_turn_settlement_result( + session_id, + turn_id, + AgentTurnSettlementResult { + status: AgentTurnSettlementStatus::Cancelled, + final_response: None, + finish_reason: Some("cancelled".to_string()), + }, + ); + match session_manager .update_session_state_for_turn_if_processing(session_id, turn_id, SessionState::Idle) .await @@ -3321,6 +3354,16 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet } }; + session_manager.record_turn_settlement_result( + session_id, + turn_id, + AgentTurnSettlementResult { + status: AgentTurnSettlementStatus::Cancelled, + final_response: None, + finish_reason: Some("interrupted".to_string()), + }, + ); + match session_manager .update_session_state_for_turn_if_processing(session_id, turn_id, SessionState::Idle) .await @@ -3493,6 +3536,16 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet } } + session_manager.record_turn_settlement_result( + session_id, + turn_id, + AgentTurnSettlementResult { + status: AgentTurnSettlementStatus::Failed, + final_response: None, + finish_reason: Some("failed".to_string()), + }, + ); + match session_manager .update_session_state_for_turn_if_processing( session_id, @@ -7287,7 +7340,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet "Dialog turn not found: {turn_id}" ))); } - return Err(BitFunError::Service(format!( + return Err(BitFunError::OutcomeUnknown(format!( "Turn settlement evidence is unavailable: session_id={session_id}, turn_id={turn_id}" ))); } @@ -13767,6 +13820,23 @@ impl ConversationCoordinator { } let duration_ms = started_at.elapsed().as_millis().min(u128::from(u64::MAX)) as u64; + let cancelled = cancellation_token.is_cancelled() + || results.iter().any(|result| { + result + .result + .result + .get("category") + .and_then(serde_json::Value::as_str) + == Some("cancelled") + }); + let success = results.iter().all(user_shell_tool_result_succeeded); + let finish_reason = if cancelled { + "cancelled" + } else if success { + "complete" + } else { + "tool_error" + }; if let Err(error) = session_manager .complete_dialog_turn( &session_id, @@ -13779,6 +13849,8 @@ impl ConversationCoordinator { total_tokens: 0, duration_ms, }, + Some(finish_reason.to_string()), + Some(false), ) .await { @@ -13796,15 +13868,6 @@ impl ConversationCoordinator { return; } - let cancelled = cancellation_token.is_cancelled() - || results.iter().any(|result| { - result - .result - .result - .get("category") - .and_then(serde_json::Value::as_str) - == Some("cancelled") - }); if cancelled { Self::persist_cancelled_dialog_turn( event_queue.as_ref(), @@ -13816,7 +13879,6 @@ impl ConversationCoordinator { ) .await; } else { - let success = results.iter().all(user_shell_tool_result_succeeded); let _ = session_manager .update_session_state_for_turn_if_processing( &session_id, @@ -14646,7 +14708,9 @@ mod tests { use crate::agentic::tools::{ToolPipeline, ToolStateManager}; use crate::agentic::TurnSkillAgentSnapshot; use crate::infrastructure::PathManager; + use crate::util::errors::BitFunError; use bitfun_agent_runtime::permission::PermissionRequestManager; + use bitfun_runtime_ports::AgentTurnSettlementStatus; use bitfun_runtime_services::test_support::FakeRuntimePort; use bitfun_services_core::permission_store::ProjectPermissionSqliteStore; @@ -16249,6 +16313,13 @@ mod tests { ) .await; + assert_eq!( + session_manager + .turn_settlement_result(&session.session_id, &turn_id) + .and_then(|result| result.final_response), + Some("complete response".to_string()) + ); + let events = coordinator.event_queue.dequeue_batch(10).await; assert!(events.iter().any(|envelope| matches!( &envelope.event, @@ -16263,6 +16334,130 @@ mod tests { .expect("clean up persisted test session"); } + #[tokio::test] + async fn transient_turns_keep_authoritative_terminal_results() { + let workspace = tempfile::tempdir().expect("workspace"); + let (coordinator, session_manager) = test_persistent_coordinator(); + let session = session_manager + .create_transient_session_with_id_and_details( + Some("transient-session".to_string()), + "Transient completion".to_string(), + "agentic".to_string(), + SessionConfig { + workspace_path: Some(workspace.path().to_string_lossy().into_owned()), + ..Default::default() + }, + None, + SessionKind::Standard, + ) + .await + .expect("create transient session"); + let turn_id = session_manager + .start_dialog_turn( + &session.session_id, + "agentic".to_string(), + "finish".to_string(), + Some("turn-transient-result".to_string()), + None, + None, + ) + .await + .expect("start turn"); + let intermediate = Message::assistant("intermediate tool-round text".to_string()) + .with_turn_id(turn_id.clone()) + .with_round_id("round-tool".to_string()); + let final_message = Message::assistant("authoritative final answer".to_string()) + .with_turn_id(turn_id.clone()) + .with_round_id("round-final".to_string()); + + ConversationCoordinator::persist_completed_dialog_turn( + coordinator.event_queue.as_ref(), + session_manager.as_ref(), + None, + &session.session_id, + &turn_id, + &ExecutionResult { + final_message: final_message.clone(), + total_rounds: 2, + success: true, + new_messages: vec![intermediate, final_message], + finish_reason: FinishReason::Complete, + total_tools: 1, + duration_ms: 1, + partial_recovery_reason: None, + effective_finish_reason: "complete".to_string(), + has_final_response: true, + }, + None, + ) + .await; + + let settlement = session_manager + .turn_settlement_result(&session.session_id, &turn_id) + .expect("transient turn settlement result"); + assert_eq!(settlement.status, AgentTurnSettlementStatus::Completed); + assert_eq!( + settlement.final_response.as_deref(), + Some("authoritative final answer") + ); + assert_eq!(settlement.finish_reason.as_deref(), Some("complete")); + + let cancelled_turn_id = session_manager + .start_dialog_turn( + &session.session_id, + "agentic".to_string(), + "cancel".to_string(), + Some("turn-transient-cancelled".to_string()), + None, + None, + ) + .await + .expect("start cancelled turn"); + ConversationCoordinator::persist_cancelled_dialog_turn( + coordinator.event_queue.as_ref(), + session_manager.as_ref(), + None, + &session.session_id, + &cancelled_turn_id, + true, + ) + .await; + let cancelled = session_manager + .turn_settlement_result(&session.session_id, &cancelled_turn_id) + .expect("cancelled turn settlement result"); + assert_eq!(cancelled.status, AgentTurnSettlementStatus::Cancelled); + assert_eq!(cancelled.final_response, None); + assert_eq!(cancelled.finish_reason.as_deref(), Some("cancelled")); + + let failed_turn_id = session_manager + .start_dialog_turn( + &session.session_id, + "agentic".to_string(), + "fail".to_string(), + Some("turn-transient-failed".to_string()), + None, + None, + ) + .await + .expect("start failed turn"); + ConversationCoordinator::persist_failed_dialog_turn( + coordinator.event_queue.as_ref(), + session_manager.as_ref(), + None, + &session.session_id, + &failed_turn_id, + &BitFunError::AIClient("provider failed".to_string()), + true, + ) + .await; + let failed = session_manager + .turn_settlement_result(&session.session_id, &failed_turn_id) + .expect("failed turn settlement result"); + assert_eq!(failed.status, AgentTurnSettlementStatus::Failed); + assert_eq!(failed.final_response, None); + assert_eq!(failed.finish_reason.as_deref(), Some("failed")); + } + async fn create_two_turn_session( session_manager: &SessionManager, workspace: &std::path::Path, @@ -16299,6 +16494,8 @@ mod tests { format!("reply to {prompt}"), &[], TurnStats::default(), + Some("complete".to_string()), + Some(true), ) .await .expect("complete persisted turn"); diff --git a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs b/src/crates/assembly/core/src/agentic/coordination/scheduler.rs index afdec65f49..e564d0e135 100644 --- a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs +++ b/src/crates/assembly/core/src/agentic/coordination/scheduler.rs @@ -4316,7 +4316,14 @@ mod tests { .await .expect_err("missing settlement evidence must not be treated as success"); - assert!(matches!(error, BitFunError::Service(_)), "{error}"); + assert!( + matches!( + &error, + BitFunError::OutcomeUnknown(message) + if message.contains(session_id) && message.contains(turn_id) + ), + "{error}" + ); } fn desktop_active_turn(turn_id: &str) -> ActiveDialogTurn { diff --git a/src/crates/assembly/core/src/agentic/session/session_manager.rs b/src/crates/assembly/core/src/agentic/session/session_manager.rs index 06a190d858..9b34621cd1 100644 --- a/src/crates/assembly/core/src/agentic/session/session_manager.rs +++ b/src/crates/assembly/core/src/agentic/session/session_manager.rs @@ -55,7 +55,9 @@ use crate::util::sanitize_plain_model_output; use crate::util::timing::elapsed_ms_u64; use bitfun_core_types::SessionExecutionTarget; pub use bitfun_runtime_ports::SessionViewRestoreTiming; -use bitfun_runtime_ports::{PermissionMode, SessionStoragePathRequest, SessionStorePort}; +use bitfun_runtime_ports::{ + AgentTurnSettlementResult, PermissionMode, SessionStoragePathRequest, SessionStorePort, +}; use bitfun_services_core::session::{ apply_session_lineage, collect_hidden_subagent_cascade as collect_hidden_subagent_cascade_ids, merge_session_custom_metadata as merge_session_custom_metadata_value, @@ -66,9 +68,9 @@ use dashmap::{mapref::entry::Entry, DashMap}; use log::{debug, error, info, warn}; use serde::{Deserialize, Serialize}; use serde_json::json; -use std::collections::HashSet; +use std::collections::{HashSet, VecDeque}; use std::path::{Path, PathBuf}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::Instant; use std::time::{Duration, SystemTime}; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; @@ -311,6 +313,12 @@ pub struct SessionManager { /// removed with that Session; they are never serialized into public config. transient_session_ids: Arc>, + /// Recent authoritative terminal results for live Turn settlement callers. + /// The bounded cache preserves the exact execution result; persisted Turns + /// remain the fallback, while transient Sessions depend on this copy. + turn_settlement_results: Arc>, + turn_settlement_result_order: Arc>>, + /// Exact admission accounting for loaded sessions. A permit is acquired /// before create/restore publishes runtime state and released on unload/delete/eviction. active_session_capacity: Arc, @@ -467,6 +475,7 @@ impl SessionManager { pub(crate) fn evict_loaded_session_for_test(&self, session_id: &str) { self.sessions.remove(session_id); self.transient_session_ids.remove(session_id); + self.clear_turn_settlement_results(session_id); self.release_active_session_reservation(session_id); self.release_session_write_lock(session_id); } @@ -2020,6 +2029,8 @@ impl SessionManager { sessions: Arc::new(DashMap::new()), active_turn_permission_modes: Arc::new(DashMap::new()), transient_session_ids: Arc::new(DashMap::new()), + turn_settlement_results: Arc::new(DashMap::new()), + turn_settlement_result_order: Arc::new(Mutex::new(VecDeque::new())), active_session_capacity: Arc::new(Semaphore::new(config.max_active_sessions)), active_session_permits: Arc::new(DashMap::new()), session_storage_path_index: Arc::new(DashMap::new()), @@ -2054,6 +2065,62 @@ impl SessionManager { self.persistence_manager.clone() } + pub(crate) fn record_turn_settlement_result( + &self, + session_id: &str, + turn_id: &str, + result: AgentTurnSettlementResult, + ) { + const MAX_RECENT_TURN_SETTLEMENT_RESULTS: usize = 1_024; + let key = (session_id.to_string(), turn_id.to_string()); + let mut order = self + .turn_settlement_result_order + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + order.retain(|existing| existing != &key); + self.turn_settlement_results.insert(key.clone(), result); + order.push_back(key); + while order.len() > MAX_RECENT_TURN_SETTLEMENT_RESULTS { + if let Some(oldest) = order.pop_front() { + self.turn_settlement_results.remove(&oldest); + } + } + } + + pub(crate) fn turn_settlement_result( + &self, + session_id: &str, + turn_id: &str, + ) -> Option { + self.turn_settlement_results + .get(&(session_id.to_string(), turn_id.to_string())) + .map(|entry| entry.value().clone()) + } + + fn clear_turn_settlement_results(&self, session_id: &str) { + let mut order = self + .turn_settlement_result_order + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + order.retain(|key| { + if key.0 != session_id { + return true; + } + self.turn_settlement_results.remove(key); + false + }); + } + + fn clear_turn_settlement_result(&self, session_id: &str, turn_id: &str) { + let key = (session_id.to_string(), turn_id.to_string()); + let mut order = self + .turn_settlement_result_order + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + order.retain(|existing| existing != &key); + self.turn_settlement_results.remove(&key); + } + pub async fn append_evidence_event( &self, event: EvidenceLedgerEvent, @@ -2470,6 +2537,8 @@ impl SessionManager { let sessions = self.sessions.clone(); let active_turn_permission_modes = self.active_turn_permission_modes.clone(); let transient_session_ids = self.transient_session_ids.clone(); + let turn_settlement_results = self.turn_settlement_results.clone(); + let turn_settlement_result_order = self.turn_settlement_result_order.clone(); let active_session_capacity = self.active_session_capacity.clone(); let active_session_permits = self.active_session_permits.clone(); let session_storage_path_index = self.session_storage_path_index.clone(); @@ -2505,6 +2574,8 @@ impl SessionManager { sessions, active_turn_permission_modes, transient_session_ids, + turn_settlement_results, + turn_settlement_result_order, active_session_capacity, active_session_permits, session_storage_path_index, @@ -4869,6 +4940,7 @@ impl SessionManager { .await?; self.sessions.remove(session_id); self.transient_session_ids.remove(session_id); + self.clear_turn_settlement_results(session_id); self.release_active_session_reservation(session_id); self.session_storage_path_index.remove(session_id); Ok(true) @@ -4913,6 +4985,7 @@ impl SessionManager { if self.sessions.remove(session_id).is_none() { return Ok(false); } + self.clear_turn_settlement_results(session_id); self.release_active_session_reservation(session_id); self.active_turn_permission_modes.remove(session_id); clear_session_runtime_stores( @@ -5087,6 +5160,7 @@ impl SessionManager { ); self.sessions.remove(session_id); self.transient_session_ids.remove(session_id); + self.clear_turn_settlement_results(session_id); self.release_active_session_reservation(session_id); debug!( "Session deletion stage completed: session_id={}, stage=in_memory_remove, duration_ms={}", @@ -6489,6 +6563,7 @@ impl SessionManager { None }; // RefMut guard released here -- DashMap shard lock is free. + self.clear_turn_settlement_results(session_id); if let Some(session) = session_snapshot { self.persistence_manager @@ -7619,6 +7694,8 @@ impl SessionManager { final_response: String, new_messages: &[Message], stats: TurnStats, + finish_reason: Option, + has_final_response: Option, ) -> BitFunResult<()> { if !self.should_persist_session_id(session_id) { debug!( @@ -7762,6 +7839,8 @@ impl SessionManager { } turn.status = TurnStatus::Completed; turn.recovery = None; + turn.finish_reason = finish_reason; + turn.has_final_response = has_final_response; turn.duration_ms = Some(stats.duration_ms); turn.end_time = Some(completion_timestamp); @@ -7922,6 +8001,8 @@ impl SessionManager { final_response: String, new_messages: &[Message], stats: TurnStats, + finish_reason: Option, + has_final_response: Option, ) -> BitFunResult<()> { let _mutation_guard = self.acquire_session_mutation(session_id).await?; let workspace_path = self @@ -8004,6 +8085,8 @@ impl SessionManager { turn.status = TurnStatus::Completed; turn.recovery_epoch = Some(execution_generation); turn.recovery = None; + turn.finish_reason = finish_reason; + turn.has_final_response = has_final_response; turn.duration_ms = Some(stats.duration_ms); turn.end_time = Some(completion_timestamp); @@ -8082,6 +8165,8 @@ impl SessionManager { Self::append_generation_rounds(&mut turn, turn_id, generation_messages, now); turn.status = TurnStatus::Error; turn.recovery = None; + turn.finish_reason = Some("failed".to_string()); + turn.has_final_response = Some(false); turn.end_time = Some(now); if recovered_generation { @@ -8651,6 +8736,7 @@ impl SessionManager { .insert(session_id.to_string(), updated_session); self.context_store .replace_context(session_id, messages.clone()); + self.clear_turn_settlement_result(session_id, turn_id); Ok(InterruptedTurnRecoveryPlan { session_id: session_id.to_string(), @@ -8931,6 +9017,8 @@ impl SessionManager { turn.model_rounds = model_rounds; turn.status = TurnStatus::Error; turn.error = Some(error.clone()); + turn.finish_reason = Some("failed".to_string()); + turn.has_final_response = Some(false); turn.duration_ms = Some(completion_timestamp.saturating_sub(turn.start_time)); turn.end_time = Some(completion_timestamp); @@ -8985,7 +9073,7 @@ impl SessionManager { &self, session_id: &str, ) -> BitFunResult>> { - if !self.config.enable_persistence { + if !self.should_persist_session_id(session_id) { return Ok(None); } let Some(workspace_path) = self.effective_session_storage_path(session_id).await else { @@ -9593,7 +9681,9 @@ mod tests { ReasoningCatalogBinding, ReasoningConfig, ReasoningPreset, ReasoningPresetAction, SessionExecutionTarget, }; - use bitfun_runtime_ports::SessionStoragePathRequest; + use bitfun_runtime_ports::{ + AgentTurnSettlementResult, AgentTurnSettlementStatus, SessionStoragePathRequest, + }; use bitfun_services_core::session::SessionBranchBoundary; use dashmap::{try_result::TryResult, DashMap}; use serde_json::json; @@ -9814,6 +9904,54 @@ mod tests { ) } + #[tokio::test] + async fn refreshed_turn_settlement_results_remain_bounded_and_clear_with_the_session() { + let workspace = TestWorkspace::new(); + let persistence_manager = Arc::new( + PersistenceManager::new(workspace.path_manager()).expect("persistence manager"), + ); + let manager = test_manager(persistence_manager); + let result = |status| AgentTurnSettlementResult { + status, + final_response: None, + finish_reason: None, + }; + + manager.record_turn_settlement_result( + "session-refresh", + "turn-refresh", + result(AgentTurnSettlementStatus::Cancelled), + ); + for index in 0..1_023 { + manager.record_turn_settlement_result( + "other-session", + &format!("turn-{index}"), + result(AgentTurnSettlementStatus::Completed), + ); + } + manager.record_turn_settlement_result( + "session-refresh", + "turn-refresh", + result(AgentTurnSettlementStatus::Completed), + ); + manager.record_turn_settlement_result( + "other-session", + "turn-overflow", + result(AgentTurnSettlementStatus::Completed), + ); + + assert_eq!( + manager + .turn_settlement_result("session-refresh", "turn-refresh") + .map(|result| result.status), + Some(AgentTurnSettlementStatus::Completed) + ); + manager.clear_turn_settlement_results("session-refresh"); + assert!(manager + .turn_settlement_result("session-refresh", "turn-refresh") + .is_none()); + } + #[tokio::test] async fn completion_replaces_a_projected_text_prefix_with_runtime_generation_content() { let workspace = TestWorkspace::new(); @@ -9888,6 +10026,8 @@ mod tests { total_tokens: 0, duration_ms: 1, }, + Some("complete".to_string()), + Some(true), ) .await .expect("completion should persist"); @@ -9898,6 +10038,8 @@ mod tests { .expect("turn should load") .expect("turn should exist"); assert_eq!(completed.status, TurnStatus::Completed); + assert_eq!(completed.finish_reason.as_deref(), Some("complete")); + assert_eq!(completed.has_final_response, Some(true)); assert_eq!(completed.model_rounds.len(), 1); assert_eq!(completed.model_rounds[0].id, "round-final"); assert_eq!( @@ -10543,6 +10685,8 @@ mod tests { total_tokens: 0, duration_ms: 1, }, + Some("complete".to_string()), + Some(true), ) .await .expect_err("injected recovered completion write must fail"); @@ -10636,6 +10780,15 @@ mod tests { .mark_dialog_turn_interrupted(&session.session_id, &turn_id) .await .expect("turn should become interrupted"); + manager.record_turn_settlement_result( + &session.session_id, + &turn_id, + AgentTurnSettlementResult { + status: AgentTurnSettlementStatus::Cancelled, + final_response: None, + finish_reason: Some("interrupted".to_string()), + }, + ); manager .update_session_state_for_turn_if_processing( &session.session_id, @@ -10647,6 +10800,9 @@ mod tests { let plan = reopen_interrupted_turn_for_test(&manager, &session.session_id, &turn_id, 0) .await .expect("turn should reopen"); + assert!(manager + .turn_settlement_result(&session.session_id, &turn_id) + .is_none()); manager .complete_recovered_dialog_turn( @@ -10663,6 +10819,8 @@ mod tests { total_tokens: 0, duration_ms: 1, }, + Some("complete".to_string()), + Some(true), ) .await .expect("recovered completion should persist"); @@ -10681,6 +10839,8 @@ mod tests { ); assert!(completed.recovery.is_none()); assert_eq!(completed.recovery_epoch, Some(plan.execution_generation)); + assert_eq!(completed.finish_reason.as_deref(), Some("complete")); + assert_eq!(completed.has_final_response, Some(true)); } #[tokio::test] @@ -14405,6 +14565,8 @@ mod tests { .expect("persistence should be enabled"); assert_eq!(turns.len(), 1); assert_eq!(turns[0].status, TurnStatus::Error); + assert_eq!(turns[0].finish_reason.as_deref(), Some("failed")); + assert_eq!(turns[0].has_final_response, Some(false)); assert_eq!( turns[0].error.as_deref(), Some("terminal persistence failed") diff --git a/src/crates/assembly/core/src/agentic/session/transcript_render.rs b/src/crates/assembly/core/src/agentic/session/transcript_render.rs index 428b56dc6d..fe337688f7 100644 --- a/src/crates/assembly/core/src/agentic/session/transcript_render.rs +++ b/src/crates/assembly/core/src/agentic/session/transcript_render.rs @@ -87,6 +87,20 @@ pub(crate) fn transcript_display_user_content(turn: &DialogTurnData) -> String { .unwrap_or_else(|| strip_prompt_markup(&turn.user_message.content)) } +/// Returns the last visible assistant response from the effective model +/// attempt. Intermediate tool rounds and private subagent output are excluded. +pub(crate) fn transcript_final_assistant_content(turn: &DialogTurnData) -> Option { + transcript_round_blocks(turn, &SessionTranscriptExportOptions::default()) + .into_iter() + .rev() + .find_map(|round| { + round.blocks.into_iter().find_map(|block| match block { + TranscriptRoundBlock::Assistant(content) => Some(content), + TranscriptRoundBlock::Thinking(_) | TranscriptRoundBlock::Tool(_) => None, + }) + }) +} + /// Final visible assistant prose for product search and other read-only views. /// Thinking, tool inputs/results, superseded attempts, and subagent items are /// excluded by the same transcript projection used for exports. @@ -643,5 +657,18 @@ mod search_projection_tests { assert_eq!(content, "final visible answer"); assert!(!content.contains("private")); assert!(!content.contains("subagent")); + + let mut prior_round = turn.model_rounds[0].clone(); + prior_round.id = "round_0".to_string(); + prior_round.round_index = 0; + prior_round.text_items = vec![text("intermediate", "intermediate answer", 0, false)]; + let mut final_turn = turn; + final_turn.model_rounds[0].round_index = 1; + final_turn.model_rounds.insert(0, prior_round); + + assert_eq!( + transcript_final_assistant_content(&final_turn).as_deref(), + Some("final visible answer") + ); } } diff --git a/src/crates/assembly/core/src/product_runtime.rs b/src/crates/assembly/core/src/product_runtime.rs index 364a07dc5b..1ea4d9d1b7 100644 --- a/src/crates/assembly/core/src/product_runtime.rs +++ b/src/crates/assembly/core/src/product_runtime.rs @@ -19,7 +19,8 @@ use bitfun_agent_runtime::sdk::{ AgentSessionLineageEntry, AgentSessionLineageInspection, AgentSessionLineagePort, AgentSessionLineageRequest, AgentSessionLineageSnapshot, AgentSessionLineageTranscriptRequest, AgentSessionUsagePort, AgentSessionUsageRequest, AgentTurnCancellationResult, - AgentTurnSettlementPort, AgentTurnSettlementRequest, SessionTranscript, + AgentTurnSettlementPort, AgentTurnSettlementRequest, AgentTurnSettlementResult, + AgentTurnSettlementStatus, SessionTranscript, }; use bitfun_core_types::{SESSION_PROVIDER_ACP, SESSION_PROVIDER_METADATA_KEY}; #[cfg(feature = "product-search")] @@ -55,6 +56,7 @@ use crate::agentic::events::EventQueue; use crate::agentic::keyed_lock::KeyedAsyncLockGuard; use crate::agentic::persistence::session_branch::SessionBranchRequest; use crate::agentic::persistence::{PersistenceManager, SessionMetadataPage}; +use crate::agentic::session::transcript_render::transcript_final_assistant_content; #[cfg(feature = "product-search")] use crate::agentic::session::transcript_render::{ transcript_display_assistant_content, transcript_display_user_content, @@ -2289,24 +2291,109 @@ impl AgentTurnSettlementPort for CoreSessionOperationsPort { async fn wait_for_turn_settlement( &self, request: AgentTurnSettlementRequest, - ) -> PortResult<()> { + ) -> PortResult { if request.wait_timeout_ms == 0 { return Err(PortError::new( PortErrorKind::InvalidRequest, "turn settlement timeout must be greater than zero", )); } - self.coordinator + let wait_error = self + .coordinator .wait_for_turn_settlement( &request.session_id, &request.turn_id, Duration::from_millis(request.wait_timeout_ms), ) .await - .map_err(runtime_port_error) + .err() + .map(runtime_port_error); + if let Some(error) = wait_error + .as_ref() + .filter(|error| error.kind != PortErrorKind::OutcomeUnknown) + { + return Err(error.clone()); + } + + let session_manager = self.coordinator.get_session_manager(); + let _mutation = session_manager + .acquire_session_mutation(&request.session_id) + .await + .map_err(runtime_port_error)?; + let turn_is_loaded = session_manager + .get_session(&request.session_id) + .is_some_and(|session| { + session + .dialog_turn_ids + .iter() + .any(|turn_id| turn_id == &request.turn_id) + }); + let persisted_turns = session_manager + .load_persisted_transcript_turns_locked(&request.session_id) + .await + .map_err(runtime_port_error)?; + let turn = persisted_turns + .as_deref() + .and_then(|turns| turns.iter().find(|turn| turn.turn_id == request.turn_id)); + + if let Some(turn) = turn.filter(|turn| turn.status != TurnStatus::InProgress) { + let persisted_result = persisted_turn_settlement_result(turn)?; + if let Some(cached_result) = session_manager + .turn_settlement_result(&request.session_id, &request.turn_id) + .filter(|cached| cached.status == persisted_result.status) + { + return Ok(cached_result); + } + return Ok(persisted_result); + } + if persisted_turns.is_none() && turn_is_loaded { + if let Some(cached_result) = + session_manager.turn_settlement_result(&request.session_id, &request.turn_id) + { + return Ok(cached_result); + } + } + if let Some(error) = wait_error { + return Err(error); + } + + Err(PortError::new( + PortErrorKind::OutcomeUnknown, + format!( + "Authoritative turn result is unavailable: session_id={}, turn_id={}", + request.session_id, request.turn_id + ), + )) } } +fn persisted_turn_settlement_result( + turn: &DialogTurnData, +) -> PortResult { + let final_response = transcript_final_assistant_content(turn); + let status = match turn.status { + TurnStatus::Completed if turn.has_final_response != Some(false) => { + AgentTurnSettlementStatus::Completed + } + TurnStatus::Completed | TurnStatus::Error => AgentTurnSettlementStatus::Failed, + TurnStatus::Cancelled => AgentTurnSettlementStatus::Cancelled, + TurnStatus::InProgress => { + return Err(PortError::new( + PortErrorKind::OutcomeUnknown, + format!("Dialog turn is still in progress: {}", turn.turn_id), + )); + } + }; + + Ok(AgentTurnSettlementResult { + final_response: (status == AgentTurnSettlementStatus::Completed) + .then_some(final_response) + .flatten(), + finish_reason: turn.finish_reason.clone(), + status, + }) +} + #[cfg(test)] mod tests { use std::path::{Path, PathBuf}; @@ -2314,7 +2401,10 @@ mod tests { use std::time::Duration; use crate::service::session::SessionTranscriptExportOptions; - use bitfun_agent_runtime::sdk::{AgentEventSource, AgentRuntime}; + use bitfun_agent_runtime::sdk::{ + AgentEventSource, AgentRuntime, AgentTurnSettlementPort, AgentTurnSettlementRequest, + AgentTurnSettlementResult, AgentTurnSettlementStatus, + }; use bitfun_runtime_ports::{ AgentContextReloadRequest, AgentContextReloadTarget, LocalWorkspaceSnapshotSessionRequest, LocalWorkspaceSnapshotTurnRequest, @@ -2350,8 +2440,8 @@ mod tests { use crate::agentic::tools::{ToolPipeline, ToolStateManager}; use crate::infrastructure::PathManager; use crate::service::session::{ - DialogTurnData, DialogTurnRecoveryData, DialogTurnRecoveryStatus, SessionMetadata, - TurnStatus, UserMessageData, + DialogTurnData, DialogTurnRecoveryData, DialogTurnRecoveryStatus, SessionKind, + SessionMetadata, TurnStatus, UserMessageData, }; use crate::service::session_usage::UsageTokenSource; use crate::service::snapshot::manager::clear_snapshot_manager_for_test; @@ -3273,7 +3363,7 @@ mod tests { max_active_sessions: 100, session_idle_timeout: Duration::from_secs(3600), auto_save_interval: Duration::from_secs(300), - enable_persistence: false, + enable_persistence: true, prompt_cache_policy: PromptCachePolicy::default(), }, )); @@ -3397,6 +3487,66 @@ mod tests { .await .expect("latest-turn fork should restore through the resolved storage path"); + let settlement = port + .wait_for_turn_settlement(AgentTurnSettlementRequest { + session_id: session_id.to_string(), + turn_id: visible_turn.turn_id.clone(), + wait_timeout_ms: 10, + }) + .await + .expect("a cold persisted turn should remain authoritative without a live tracker"); + assert_eq!(settlement.status, AgentTurnSettlementStatus::Completed); + + let transient = session_manager + .create_transient_session_with_id_and_details( + Some("session-transient-settlement".to_string()), + "Transient settlement".to_string(), + "agentic".to_string(), + crate::agentic::core::SessionConfig { + workspace_path: Some(workspace_root.to_string_lossy().into_owned()), + ..Default::default() + }, + None, + SessionKind::Standard, + ) + .await + .expect("transient session"); + let transient_turn_id = session_manager + .start_dialog_turn( + &transient.session_id, + "agentic".to_string(), + "finish transiently".to_string(), + Some("turn-transient-settlement".to_string()), + None, + None, + ) + .await + .expect("transient turn"); + session_manager + .reset_session_state_if_processing(&transient.session_id, &transient_turn_id); + session_manager.record_turn_settlement_result( + &transient.session_id, + &transient_turn_id, + AgentTurnSettlementResult { + status: AgentTurnSettlementStatus::Completed, + final_response: Some("transient final answer".to_string()), + finish_reason: Some("complete".to_string()), + }, + ); + + let transient_settlement = port + .wait_for_turn_settlement(AgentTurnSettlementRequest { + session_id: transient.session_id, + turn_id: transient_turn_id, + wait_timeout_ms: 10, + }) + .await + .expect("transient settlement should use the Runtime-owned result cache"); + assert_eq!( + transient_settlement.final_response.as_deref(), + Some("transient final answer") + ); + assert_ne!(result.session_id, session_id); assert_eq!(result.agent_type, "agentic"); assert_eq!( diff --git a/src/crates/contracts/runtime-ports/src/agent_api.rs b/src/crates/contracts/runtime-ports/src/agent_api.rs index 56369e0fe7..b00b6a60a3 100644 --- a/src/crates/contracts/runtime-ports/src/agent_api.rs +++ b/src/crates/contracts/runtime-ports/src/agent_api.rs @@ -405,6 +405,24 @@ pub struct AgentTurnSettlementRequest { pub wait_timeout_ms: u64, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AgentTurnSettlementStatus { + Completed, + Failed, + Cancelled, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct AgentTurnSettlementResult { + pub status: AgentTurnSettlementStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub final_response: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub finish_reason: Option, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct AgentSessionWorkspaceRequest { @@ -1809,8 +1827,10 @@ pub trait AgentSessionUsagePort: Send + Sync { #[async_trait::async_trait] pub trait AgentTurnSettlementPort: Send + Sync { - async fn wait_for_turn_settlement(&self, request: AgentTurnSettlementRequest) - -> PortResult<()>; + async fn wait_for_turn_settlement( + &self, + request: AgentTurnSettlementRequest, + ) -> PortResult; } #[async_trait::async_trait] @@ -2056,9 +2076,38 @@ pub trait SessionTranscriptReader: Send + Sync { #[cfg(test)] mod tests { - use super::*; use std::sync::Mutex; + use super::*; + + #[test] + fn turn_settlement_result_serializes_authoritative_terminal_facts() { + let result = AgentTurnSettlementResult { + status: AgentTurnSettlementStatus::Completed, + final_response: Some("final answer".to_string()), + finish_reason: Some("complete".to_string()), + }; + + assert_eq!( + serde_json::to_value(&result).expect("serialize turn settlement result"), + serde_json::json!({ + "status": "completed", + "finalResponse": "final answer", + "finishReason": "complete", + }) + ); + assert_eq!( + serde_json::from_value::(serde_json::json!({ + "status": "cancelled" + })) + .expect("deserialize turn settlement result"), + AgentTurnSettlementResult { + status: AgentTurnSettlementStatus::Cancelled, + final_response: None, + finish_reason: None, + } + ); + } #[test] fn session_revert_contract_preserves_authoritative_transcript_and_composer_intent() { let result = AgentSessionRevertResult { diff --git a/src/crates/execution/agent-runtime/examples/sdk_minimal.rs b/src/crates/execution/agent-runtime/examples/sdk_minimal.rs index 9bbb7da207..634fc9f380 100644 --- a/src/crates/execution/agent-runtime/examples/sdk_minimal.rs +++ b/src/crates/execution/agent-runtime/examples/sdk_minimal.rs @@ -49,7 +49,7 @@ impl AgentSubmissionPort for ExampleAgentProvider { #[tokio::main(flavor = "current_thread")] async fn main() -> Result<(), Box> { let compatibility = AgentRuntimeSdkCompatibility::current(); - assert_eq!(compatibility.api_version, 9); + assert_eq!(compatibility.api_version, 10); let provider = Arc::new(ExampleAgentProvider::default()); let events = AgentEventStream::new(); diff --git a/src/crates/execution/agent-runtime/src/runtime.rs b/src/crates/execution/agent-runtime/src/runtime.rs index 12e9165dd2..f14363d497 100644 --- a/src/crates/execution/agent-runtime/src/runtime.rs +++ b/src/crates/execution/agent-runtime/src/runtime.rs @@ -33,13 +33,14 @@ use bitfun_runtime_ports::{ AgentThreadGoalManagementPort, AgentThreadGoalUpdateStatusRequest, AgentTransientSessionDiscardRequest, AgentTurnCancellationPort, AgentTurnCancellationRequest, AgentTurnCancellationResult, AgentTurnInterruptionRequest, AgentTurnInterruptionResult, - AgentTurnSettlementPort, AgentTurnSettlementRequest, AgentUserAnswersRequest, - AgentUserShellCommandPort, AgentUserShellCommandRequest, AgentUserShellCommandResult, - AgentWorkspaceReference, AgentWorkspaceReferencePort, AgentWorkspaceReferenceSearchRequest, - AgentWorkspaceReferenceSearchResult, DialogSteerOutcome, DialogSubmitOutcome, - PermissionAuditRecord, PermissionGrant, PermissionGrantKey, PluginRuntimeBinding, PortError, - PortErrorKind, PortResult, RuntimeEventEnvelope, SessionTranscript, SessionTranscriptReader, - SessionTranscriptRequest, ThreadGoal, WorkspaceDiffSnapshot, + AgentTurnSettlementPort, AgentTurnSettlementRequest, AgentTurnSettlementResult, + AgentUserAnswersRequest, AgentUserShellCommandPort, AgentUserShellCommandRequest, + AgentUserShellCommandResult, AgentWorkspaceReference, AgentWorkspaceReferencePort, + AgentWorkspaceReferenceSearchRequest, AgentWorkspaceReferenceSearchResult, DialogSteerOutcome, + DialogSubmitOutcome, PermissionAuditRecord, PermissionGrant, PermissionGrantKey, + PluginRuntimeBinding, PortError, PortErrorKind, PortResult, RuntimeEventEnvelope, + SessionTranscript, SessionTranscriptReader, SessionTranscriptRequest, ThreadGoal, + WorkspaceDiffSnapshot, }; use bitfun_runtime_services::RuntimeServices; @@ -1449,7 +1450,7 @@ impl AgentRuntime { pub async fn wait_for_turn_settlement( &self, request: AgentTurnSettlementRequest, - ) -> Result<(), RuntimeError> { + ) -> Result { let port = self.turn_settlement.as_ref().ok_or_else(|| { RuntimeError::Port(PortError::new( PortErrorKind::NotAvailable, diff --git a/src/crates/execution/agent-runtime/src/sdk.rs b/src/crates/execution/agent-runtime/src/sdk.rs index d41b8ae0e1..d75eaee234 100644 --- a/src/crates/execution/agent-runtime/src/sdk.rs +++ b/src/crates/execution/agent-runtime/src/sdk.rs @@ -8,7 +8,7 @@ use std::sync::Arc; -pub const AGENT_RUNTIME_SDK_API_VERSION: u32 = 9; +pub const AGENT_RUNTIME_SDK_API_VERSION: u32 = 10; #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[non_exhaustive] @@ -92,17 +92,18 @@ pub use bitfun_runtime_ports::{ AgentThreadGoalManagementPort, AgentThreadGoalUpdateStatusRequest, AgentTransientSessionDiscardRequest, AgentTurnCancellationPort, AgentTurnCancellationRequest, AgentTurnCancellationResult, AgentTurnInterruptionRequest, AgentTurnInterruptionResult, - AgentTurnSettlementPort, AgentTurnSettlementRequest, AgentUserAnswersRequest, - AgentUserShellCommandPort, AgentUserShellCommandRequest, AgentUserShellCommandResult, - AgentWorkspaceReference, AgentWorkspaceReferenceKind, AgentWorkspaceReferencePort, - AgentWorkspaceReferenceSearchEntry, AgentWorkspaceReferenceSearchRequest, - AgentWorkspaceReferenceSearchResult, AgentWorkspaceReferenceSourceRange, ClockPort, - DialogSteerOutcome, DialogSubmissionPolicy, DialogSubmitOutcome, FileSystemPort, GitPort, - McpCatalogPort, NetworkPort, PermissionAuditRecord, PermissionDelegationContext, - PermissionGrant, PermissionGrantKey, PermissionReply, PermissionReplySource, PermissionRequest, - PermissionRequestEvent, PermissionRequestSource, PermissionRequestSourceKind, PortError, - PortErrorKind, PortResult, RemoteAssistantWorkspaceFacts, RemoteCapabilityPort, - RemoteConnectionPort, RemoteProjectionPort, RemoteRecentWorkspaceFacts, RemoteWorkspaceFacts, + AgentTurnSettlementPort, AgentTurnSettlementRequest, AgentTurnSettlementResult, + AgentTurnSettlementStatus, AgentUserAnswersRequest, AgentUserShellCommandPort, + AgentUserShellCommandRequest, AgentUserShellCommandResult, AgentWorkspaceReference, + AgentWorkspaceReferenceKind, AgentWorkspaceReferencePort, AgentWorkspaceReferenceSearchEntry, + AgentWorkspaceReferenceSearchRequest, AgentWorkspaceReferenceSearchResult, + AgentWorkspaceReferenceSourceRange, ClockPort, DialogSteerOutcome, DialogSubmissionPolicy, + DialogSubmitOutcome, FileSystemPort, GitPort, McpCatalogPort, NetworkPort, + PermissionAuditRecord, PermissionDelegationContext, PermissionGrant, PermissionGrantKey, + PermissionReply, PermissionReplySource, PermissionRequest, PermissionRequestEvent, + PermissionRequestSource, PermissionRequestSourceKind, PortError, PortErrorKind, PortResult, + RemoteAssistantWorkspaceFacts, RemoteCapabilityPort, RemoteConnectionPort, + RemoteProjectionPort, RemoteRecentWorkspaceFacts, RemoteWorkspaceFacts, RemoteWorkspaceFileRuntimeHost, RemoteWorkspaceKind, RemoteWorkspacePort, RemoteWorkspaceRuntimeHost, RemoteWorkspaceUpdate, RuntimeEventEnvelope, RuntimeEventSink, RuntimeEventType, RuntimeServiceCapability, RuntimeServicePort, SessionStorageKind, @@ -617,7 +618,7 @@ impl AgentRuntime { pub async fn wait_for_turn_settlement( &self, request: AgentTurnSettlementRequest, - ) -> Result<(), RuntimeError> { + ) -> Result { self.inner.wait_for_turn_settlement(request).await } diff --git a/src/crates/execution/agent-runtime/tests/agent_session_contracts/sdk_smoke.rs b/src/crates/execution/agent-runtime/tests/agent_session_contracts/sdk_smoke.rs index 5b19081d37..a5b681cb77 100644 --- a/src/crates/execution/agent-runtime/tests/agent_session_contracts/sdk_smoke.rs +++ b/src/crates/execution/agent-runtime/tests/agent_session_contracts/sdk_smoke.rs @@ -74,7 +74,7 @@ impl AgentModeCatalogPort for FakeModeCatalog { fn sdk_facade_exposes_versioned_preview_compatibility_contract() { let compatibility = AgentRuntimeSdkCompatibility::current(); - assert_eq!(compatibility.api_version, 9); + assert_eq!(compatibility.api_version, 10); assert_eq!(compatibility.crate_version, env!("CARGO_PKG_VERSION")); assert_eq!(compatibility.stability, AgentRuntimeSdkStability::Preview); } diff --git a/src/crates/execution/agent-runtime/tests/agent_session_contracts/session_operation_ports.rs b/src/crates/execution/agent-runtime/tests/agent_session_contracts/session_operation_ports.rs index 268937a38f..678f620828 100644 --- a/src/crates/execution/agent-runtime/tests/agent_session_contracts/session_operation_ports.rs +++ b/src/crates/execution/agent-runtime/tests/agent_session_contracts/session_operation_ports.rs @@ -4,8 +4,8 @@ use bitfun_agent_runtime::sdk::{ AgentRuntimeBuilder, AgentSessionForkAtTurnRequest, AgentSessionForkPort, AgentSessionForkRequest, AgentSessionForkResult, AgentSessionUsagePort, AgentSessionUsageRequest, AgentSubmissionPort, AgentSubmissionRequest, AgentSubmissionResult, - AgentTurnSettlementPort, AgentTurnSettlementRequest, PortErrorKind, PortResult, - SessionUsageReport, + AgentTurnSettlementPort, AgentTurnSettlementRequest, AgentTurnSettlementResult, + AgentTurnSettlementStatus, PortErrorKind, PortResult, SessionUsageReport, }; use bitfun_agent_runtime::sdk::{AgentSessionCreateRequest, AgentSessionCreateResult}; @@ -97,9 +97,13 @@ impl AgentTurnSettlementPort for RecordingSessionOperations { async fn wait_for_turn_settlement( &self, request: AgentTurnSettlementRequest, - ) -> PortResult<()> { + ) -> PortResult { self.settlement_requests.lock().unwrap().push(request); - Ok(()) + Ok(AgentTurnSettlementResult { + status: AgentTurnSettlementStatus::Completed, + final_response: Some("authoritative response".to_string()), + finish_reason: Some("complete".to_string()), + }) } } @@ -148,7 +152,7 @@ async fn runtime_delegates_narrow_session_operations_to_registered_ports() { .await .expect("generate usage"); assert_eq!(report.session_id, "session-1"); - runtime + let settlement = runtime .wait_for_turn_settlement(AgentTurnSettlementRequest { session_id: "session-1".to_string(), turn_id: "turn-1".to_string(), @@ -157,6 +161,12 @@ async fn runtime_delegates_narrow_session_operations_to_registered_ports() { .await .expect("wait for turn settlement"); + assert_eq!(settlement.status, AgentTurnSettlementStatus::Completed); + assert_eq!( + settlement.final_response.as_deref(), + Some("authoritative response") + ); + assert_eq!(settlement.finish_reason.as_deref(), Some("complete")); assert_eq!(operations.settlement_requests.lock().unwrap().len(), 1); } diff --git a/src/crates/interfaces/app-server/src/server/handlers/session.rs b/src/crates/interfaces/app-server/src/server/handlers/session.rs index 7c45345776..5b2b24aedd 100644 --- a/src/crates/interfaces/app-server/src/server/handlers/session.rs +++ b/src/crates/interfaces/app-server/src/server/handlers/session.rs @@ -354,7 +354,7 @@ pub(in crate::server) fn builder( .runtime() .wait_for_turn_settlement(request.0) .await - .map(|()| WaitForSettlementResponse {}) + .map(|_| WaitForSettlementResponse {}) .map_err(|error| { BitfunAppRuntime::session_runtime_error(&session_id, error) }), diff --git a/src/crates/interfaces/app-server/tests/agent_kernel.rs b/src/crates/interfaces/app-server/tests/agent_kernel.rs index c34bd11fe5..cf8a685c80 100644 --- a/src/crates/interfaces/app-server/tests/agent_kernel.rs +++ b/src/crates/interfaces/app-server/tests/agent_kernel.rs @@ -527,9 +527,13 @@ impl ports::AgentTurnSettlementPort for Phase2Provider { async fn wait_for_turn_settlement( &self, request: ports::AgentTurnSettlementRequest, - ) -> PortResult<()> { + ) -> PortResult { self.settlements.lock().unwrap().push(request); - Ok(()) + Ok(ports::AgentTurnSettlementResult { + status: ports::AgentTurnSettlementStatus::Completed, + final_response: Some("fixture result".to_string()), + finish_reason: Some("stop".to_string()), + }) } } diff --git a/src/crates/interfaces/sdk-host/src/host.rs b/src/crates/interfaces/sdk-host/src/host.rs index 513348eb58..afba40f55e 100644 --- a/src/crates/interfaces/sdk-host/src/host.rs +++ b/src/crates/interfaces/sdk-host/src/host.rs @@ -11,9 +11,10 @@ use bitfun_agent_runtime::sdk::{ AgentSessionCreateResult, AgentSessionDeleteRequest, AgentSessionModelUpdateRequest, AgentSessionReleaseRequest, AgentSessionRestoreRequest, AgentSessionWorkspaceRequest, AgentSubmissionSource, AgentTurnCancellationRequest, AgentTurnSettlementRequest, - DialogSubmissionPolicy, DialogSubmitOutcome, PermissionReply, PermissionReplySource, - PermissionRequest, PermissionRequestEvent, PermissionRequestSourceKind, PortError, - PortErrorKind, RuntimeError, TurnTokenUsage, AUTO_APPROVE_ASK_CONTEXT_KEY, + AgentTurnSettlementResult, AgentTurnSettlementStatus, DialogSubmissionPolicy, + DialogSubmitOutcome, PermissionReply, PermissionReplySource, PermissionRequest, + PermissionRequestEvent, PermissionRequestSourceKind, PortError, PortErrorKind, RuntimeError, + TurnTokenUsage, AUTO_APPROVE_ASK_CONTEXT_KEY, }; use bitfun_agent_runtime::user_questions::USER_INPUT_AVAILABLE_CONTEXT_KEY; use bitfun_core_types::ErrorCategory; @@ -235,7 +236,7 @@ struct QueryLease { session_id: String, turn_id: String, operation_id: String, - output: StdMutex, + event_budget: StdMutex, structured_output_requested: bool, usage: StdMutex>, terminal: AtomicBool, @@ -246,8 +247,7 @@ struct QueryLease { } #[derive(Default)] -struct QueryOutputBuffer { - text: String, +struct QueryEventBudget { wire_bytes: usize, structured_attempt: Option, } @@ -259,11 +259,10 @@ struct QueryOutputAttempt { attempt_index: Option, } -impl QueryOutputBuffer { - fn append(&mut self, text: &str, structured_attempt: Option) -> bool { +impl QueryEventBudget { + fn observe(&mut self, text: &str, structured_attempt: Option) -> bool { if let Some(attempt) = structured_attempt { if self.structured_attempt.as_ref() != Some(&attempt) { - self.text.clear(); self.wire_bytes = 0; self.structured_attempt = Some(attempt); } @@ -273,7 +272,6 @@ impl QueryOutputBuffer { if encoded_bytes > MAX_QUERY_OUTPUT_WIRE_BYTES.saturating_sub(self.wire_bytes) { return false; } - self.text.push_str(text); self.wire_bytes += encoded_bytes; true } @@ -1409,7 +1407,7 @@ impl SdkHostConnection { session_id: submitted_session_id.clone(), turn_id: turn_id.clone(), operation_id: operation_id.clone(), - output: StdMutex::new(QueryOutputBuffer::default()), + event_budget: StdMutex::new(QueryEventBudget::default()), structured_output_requested: params.output_schema.is_some(), usage: StdMutex::new(None), terminal: AtomicBool::new(false), @@ -1618,10 +1616,10 @@ impl SdkHostConnection { if let Some(projected) = project_query_event(&envelope.event) { if let QueryEvent::AssistantTextDelta { text } = &projected { let output_exceeded = { - let mut output = lease - .output + let mut event_budget = lease + .event_budget .lock() - .expect("SDK Host Query output lock poisoned"); + .expect("SDK Host Query event budget lock poisoned"); let structured_attempt = lease.structured_output_requested.then(|| { match &envelope.event { @@ -1640,7 +1638,7 @@ impl SdkHostConnection { ), } }); - !output.append(text, structured_attempt) + !event_budget.observe(text, structured_attempt) }; if output_exceeded { connection @@ -1692,7 +1690,9 @@ impl SdkHostConnection { } } if let Some((status, error)) = terminal { - connection.finish_query(&lease, status, error, true).await; + connection + .finish_query(&lease, status, error, true, false) + .await; return; } } @@ -2077,7 +2077,7 @@ impl SdkHostConnection { }; for query in queries { query.stop_forwarding.cancel(); - self.finish_query(&query, QueryTerminalStatus::Cancelled, None, true) + self.finish_query(&query, QueryTerminalStatus::Cancelled, None, true, false) .await; } self.send_success( @@ -2604,6 +2604,7 @@ impl SdkHostConnection { status: QueryTerminalStatus, error: Option, emit_result: bool, + preserve_host_failure: bool, ) { if !lease.finish_once() { return; @@ -2628,7 +2629,11 @@ impl SdkHostConnection { }), ) .await; - let settlement_confirmed = matches!(settlement, Ok(Ok(()))); + let settlement_result = match settlement { + Ok(Ok(result)) => Some(result), + Ok(Err(_)) | Err(_) => None, + }; + let settlement_confirmed = settlement_result.is_some(); { let mut state = self.inner.state.lock().await; state.queries.remove(&lease.query_id); @@ -2646,7 +2651,14 @@ impl SdkHostConnection { return; } if emit_result { - self.send_query_result(lease, status, error).await; + self.send_query_result( + lease, + status, + error, + settlement_result.expect("confirmed settlement result"), + preserve_host_failure, + ) + .await; } } @@ -2655,16 +2667,52 @@ impl SdkHostConnection { lease: &QueryLease, mut status: QueryTerminalStatus, mut error: Option, + settlement: AgentTurnSettlementResult, + preserve_host_failure: bool, ) -> bool { if !lease.emit_output { return true; } - let output_text = lease - .output - .lock() - .expect("SDK Host Query output lock poisoned") - .text - .clone(); + let mut output_text = settlement.final_response.unwrap_or_default(); + match settlement.status { + AgentTurnSettlementStatus::Completed if !preserve_host_failure => { + status = QueryTerminalStatus::Completed; + error = None; + } + AgentTurnSettlementStatus::Completed => {} + AgentTurnSettlementStatus::Failed => { + status = QueryTerminalStatus::Failed; + if error.is_none() { + error = Some(QueryResultError::new( + ErrorCode::Internal, + false, + None, + &lease.query_id, + format!( + "Query completed unsuccessfully: {}", + settlement.finish_reason.as_deref().unwrap_or("failed") + ), + )); + } + } + AgentTurnSettlementStatus::Cancelled => { + if !preserve_host_failure { + status = QueryTerminalStatus::Cancelled; + error = None; + } + } + } + if json_string_content_bytes(&output_text) > MAX_QUERY_OUTPUT_WIRE_BYTES { + output_text.clear(); + status = QueryTerminalStatus::Failed; + error = Some(QueryResultError::new( + ErrorCode::Overloaded, + false, + None, + &lease.query_id, + "SDK Host Query output exceeded the protocol size limit", + )); + } let structured = if lease.structured_output_requested && status == QueryTerminalStatus::Completed { match serde_json::from_str(&output_text) { @@ -2757,8 +2805,14 @@ impl SdkHostConnection { ); } } - self.finish_query(lease, QueryTerminalStatus::Failed, Some(error), emit_result) - .await; + self.finish_query( + lease, + QueryTerminalStatus::Failed, + Some(error), + emit_result, + true, + ) + .await; } async fn reject_permission_and_finish( @@ -3409,8 +3463,8 @@ fn runtime_error_kind(error: &RuntimeError) -> &'static str { #[cfg(test)] mod runtime_error_tests { use super::{ - is_local_image_path, runtime_error_facts, runtime_error_kind, QueryOutputAttempt, - QueryOutputBuffer, + is_local_image_path, runtime_error_facts, runtime_error_kind, QueryEventBudget, + QueryOutputAttempt, }; use crate::protocol::{ErrorCode, RecoveryAction}; use bitfun_agent_runtime::sdk::{PortError, PortErrorKind, RuntimeError}; @@ -3424,28 +3478,30 @@ mod runtime_error_tests { } #[test] - fn structured_output_keeps_only_the_last_model_attempt() { + fn structured_event_budget_keeps_only_the_last_model_attempt() { let attempt = |round_id: &str| QueryOutputAttempt { round_id: round_id.to_string(), attempt_id: Some(format!("{round_id}-attempt")), attempt_index: Some(0), }; - let mut output = QueryOutputBuffer::default(); + let mut budget = QueryEventBudget::default(); - assert!(output.append(r#"{"draft":true}"#, Some(attempt("round-1")))); - assert!(output.append(r#"{"final":true}"#, Some(attempt("round-2")))); + assert!(budget.observe(r#"{"draft":true}"#, Some(attempt("round-1")))); + let first_attempt_bytes = budget.wire_bytes; + assert!(budget.observe(r#"{"final":true}"#, Some(attempt("round-2")))); - assert_eq!(output.text, r#"{"final":true}"#); + assert_eq!(budget.wire_bytes, first_attempt_bytes); } #[test] - fn plain_output_still_aggregates_model_rounds() { - let mut output = QueryOutputBuffer::default(); + fn plain_event_budget_still_aggregates_model_rounds() { + let mut budget = QueryEventBudget::default(); - assert!(output.append("first", None)); - assert!(output.append("second", None)); + assert!(budget.observe("first", None)); + let first_bytes = budget.wire_bytes; + assert!(budget.observe("second", None)); - assert_eq!(output.text, "firstsecond"); + assert!(budget.wire_bytes > first_bytes); } #[test] diff --git a/src/crates/interfaces/sdk-host/tests/host_lifecycle.rs b/src/crates/interfaces/sdk-host/tests/host_lifecycle.rs index 3488e37820..0551bb5593 100644 --- a/src/crates/interfaces/sdk-host/tests/host_lifecycle.rs +++ b/src/crates/interfaces/sdk-host/tests/host_lifecycle.rs @@ -14,9 +14,10 @@ use bitfun_agent_runtime::sdk::{ AgentSessionWorkspaceBinding, AgentSessionWorkspaceRequest, AgentSubmissionPort, AgentSubmissionRequest, AgentSubmissionResult, AgentTransientSessionDiscardRequest, AgentTurnCancellationPort, AgentTurnCancellationRequest, AgentTurnCancellationResult, - AgentTurnSettlementPort, AgentTurnSettlementRequest, DialogSubmitOutcome, PermissionRequest, - PermissionRequestManager, PermissionRequestSource, PermissionRequestSourceKind, PortError, - PortErrorKind, PortResult, SessionState, + AgentTurnSettlementPort, AgentTurnSettlementRequest, AgentTurnSettlementResult, + AgentTurnSettlementStatus, DialogSubmitOutcome, PermissionRequest, PermissionRequestManager, + PermissionRequestSource, PermissionRequestSourceKind, PortError, PortErrorKind, PortResult, + SessionState, }; use bitfun_core_types::ErrorCategory; use bitfun_events::{AgenticEvent, ToolEventData, ToolEventIdentity}; @@ -53,6 +54,9 @@ struct FakeOwner { queue_dialog: bool, dialog_session_override: Option, output_text: Option, + settlement_output_text: Option, + settlement_status: Mutex>, + keep_completed_settlement_after_cancel: bool, emit_tool_events: bool, block_dialog_submit: bool, block_agent_resolution: bool, @@ -123,11 +127,30 @@ impl FakeOwner { Self { queue: Mutex::new(Some(queue)), emit_terminal: true, + settlement_output_text: Some(output_text.clone()), output_text: Some(output_text), ..Self::default() } } + fn with_stream_and_settlement_output( + queue: Arc, + output_text: String, + settlement_output_text: String, + ) -> Self { + Self { + queue: Mutex::new(Some(queue)), + emit_terminal: true, + output_text: Some(output_text), + settlement_output_text: Some(settlement_output_text), + ..Self::default() + } + } + + fn set_settlement_status(&self, status: AgentTurnSettlementStatus) { + *self.settlement_status.lock().unwrap() = Some(status); + } + fn with_tool_events(queue: Arc) -> Self { Self { queue: Mutex::new(Some(queue)), @@ -537,7 +560,7 @@ impl AgentTurnSettlementPort for FakeOwner { async fn wait_for_turn_settlement( &self, request: AgentTurnSettlementRequest, - ) -> PortResult<()> { + ) -> PortResult { self.settlement_requests.lock().unwrap().push(request); if self.fail_settlement { return Err(PortError::new( @@ -545,7 +568,28 @@ impl AgentTurnSettlementPort for FakeOwner { "turn settlement is unknown", )); } - Ok(()) + let status = self + .settlement_status + .lock() + .unwrap() + .unwrap_or(AgentTurnSettlementStatus::Completed); + Ok(AgentTurnSettlementResult { + status, + final_response: (status == AgentTurnSettlementStatus::Completed).then(|| { + self.settlement_output_text + .clone() + .or_else(|| self.output_text.clone()) + .unwrap_or_else(|| "fixture result".to_string()) + }), + finish_reason: Some( + match status { + AgentTurnSettlementStatus::Completed => "stop", + AgentTurnSettlementStatus::Failed => "failed", + AgentTurnSettlementStatus::Cancelled => "cancelled", + } + .to_string(), + ), + }) } } @@ -825,15 +869,17 @@ async fn host_with_query_limit( async fn host_with_output( output_text: &str, + settlement_output_text: &str, ) -> ( SdkHostConnection, Arc, mpsc::Receiver, ) { let queue = Arc::new(EventQueue::new(EventQueueConfig::default())); - let owner = Arc::new(FakeOwner::with_output( + let owner = Arc::new(FakeOwner::with_stream_and_settlement_output( queue.clone(), output_text.to_string(), + settlement_output_text.to_string(), )); let runtime = AgentRuntimeBuilder::new() .with_submission_port(owner.clone()) @@ -866,6 +912,9 @@ impl AgentTurnCancellationPort for FakeOwner { &self, request: AgentTurnCancellationRequest, ) -> PortResult { + if !self.keep_completed_settlement_after_cancel { + self.set_settlement_status(AgentTurnSettlementStatus::Cancelled); + } let cancel_index = { let mut requests = self.cancel_requests.lock().unwrap(); requests.push(request.clone()); @@ -1304,7 +1353,8 @@ async fn initialize_is_required_and_version_mismatch_fails_closed() { #[tokio::test] async fn query_streams_existing_events_and_one_terminal_result() { - let (host, _, mut output) = host().await; + let (host, _, mut output) = + host_with_output("intermediate tool-round text", "authoritative final answer").await; initialize(&host, &mut output).await; host.handle_request(request(serde_json::json!({ @@ -1331,20 +1381,27 @@ async fn query_streams_existing_events_and_one_terminal_result() { assert_ne!(operation_id, query_id); assert_eq!(event["params"]["operationId"], operation_id); assert_eq!(event["params"]["event"]["type"], "assistant_text_delta"); - assert_eq!(event["params"]["event"]["text"], "fixture result"); + assert_eq!( + event["params"]["event"]["text"], + "intermediate tool-round text" + ); let result = output.recv().await.unwrap(); assert_eq!(result["method"], "query/result"); assert_eq!(result["params"]["queryId"], query_id); assert_eq!(result["params"]["operationId"], operation_id); assert_eq!(result["params"]["status"], "completed"); - assert_eq!(result["params"]["output"]["text"], "fixture result"); + assert_eq!( + result["params"]["output"]["text"], + "authoritative final answer" + ); assert!(output.try_recv().is_err(), "terminal result must be unique"); } #[tokio::test] async fn query_passes_output_schema_to_runtime_and_returns_parsed_json() { - let (host, owner, mut output) = host_with_output(r#"{"summary":"ready"}"#).await; + let (host, owner, mut output) = + host_with_output(r#"{"summary":"ready"}"#, r#"{"summary":"ready"}"#).await; initialize(&host, &mut output).await; let schema = serde_json::json!({ "type": "object", @@ -1406,7 +1463,7 @@ async fn query_rejects_a_non_object_output_schema_before_submission() { #[tokio::test] async fn query_fails_when_structured_output_is_not_json() { - let (host, _, mut output) = host_with_output("not json").await; + let (host, _, mut output) = host_with_output("not json", "not json").await; initialize(&host, &mut output).await; host.handle_request(request(serde_json::json!({ @@ -1625,6 +1682,53 @@ async fn escaped_query_output_fails_before_exceeding_the_wire_budget() { assert_eq!(result["params"]["output"]["text"], ""); } +#[tokio::test] +async fn host_output_failure_wins_when_runtime_completed_before_cancellation() { + let queue = Arc::new(EventQueue::new(EventQueueConfig::default())); + let owner = Arc::new(FakeOwner { + queue: Mutex::new(Some(queue.clone())), + emit_terminal: true, + output_text: Some("\\".repeat(384 * 1024 + 1)), + settlement_output_text: Some("authoritative final answer".to_string()), + keep_completed_settlement_after_cancel: true, + ..FakeOwner::default() + }); + let runtime = AgentRuntimeBuilder::new() + .with_submission_port(owner.clone()) + .with_dialog_turn_port(owner.clone()) + .with_cancellation_port(owner.clone()) + .with_turn_settlement_port(owner.clone()) + .with_session_management_port(owner.clone()) + .with_session_close_port(owner.clone()) + .with_permission_request_manager(permission_manager()) + .with_event_source(AgentEventSource::new(queue)) + .build() + .unwrap(); + let (sender, mut output) = mpsc::channel(16); + let host = SdkHostConnection::new( + runtime, + "D:/workspace/project", + sender, + SdkHostConfig::default(), + fake_installer(), + ); + initialize(&host, &mut output).await; + + host.handle_request(request(serde_json::json!({ + "jsonrpc": "2.0", + "id": "query-output-race", + "method": "query/start", + "params": { "prompt": "produce excessive output" } + }))) + .await; + + assert_eq!(output.recv().await.unwrap()["id"], "query-output-race"); + let result = output.recv().await.unwrap(); + assert_eq!(result["method"], "query/result"); + assert_eq!(result["params"]["status"], "failed"); + assert_eq!(result["params"]["error"]["data"]["code"], "overloaded"); +} + #[tokio::test] async fn cancellation_after_terminal_result_is_idempotent_with_full_query_identity() { let (host, _, mut output) = host().await; @@ -2806,6 +2910,7 @@ async fn terminal_failure_is_typed_and_emitted_after_settlement() { .unwrap() .to_string(); let queue = owner.queue.lock().unwrap().clone().unwrap(); + owner.set_settlement_status(AgentTurnSettlementStatus::Failed); queue .enqueue( AgenticEvent::DialogTurnFailed { @@ -2936,6 +3041,7 @@ async fn provider_quota_and_billing_keep_distinct_wire_codes() { // Cloned out of the guard first: the guard must not survive the // `enqueue` await below. let queue = owner.queue.lock().unwrap().clone().unwrap(); + owner.set_settlement_status(AgentTurnSettlementStatus::Failed); queue .enqueue( AgenticEvent::DialogTurnFailed {