1use crate::{Result, RuntimeError, ToolRegistry};
2use futures::stream::Stream;
3use reqwest::Client;
4use serde_json::{json, Value};
5use std::path::{Path, PathBuf};
6use std::pin::Pin;
7use std::sync::Arc;
8use std::sync::Mutex;
9use std::time::Duration;
10use tokio::sync::{mpsc, RwLock};
11use tokio_util::sync::CancellationToken;
12
13mod anthropic_quota;
14mod api;
15mod api_sync;
16pub mod attachments;
17mod auth;
18mod axel_context;
19#[cfg(test)]
20mod body_golden;
21pub mod budget;
22mod cache_diagnosis;
23pub mod capture_worker;
24pub mod chat_capture;
25pub(crate) mod cloud_invoke;
26pub mod compaction;
27pub mod context;
28pub mod continuation;
29pub mod google_gemini;
30pub mod google_vertex;
31pub(crate) mod helpers;
32mod http_phases;
33mod image_limits;
34mod image_retention;
35pub mod keepwarm;
36pub mod tool_keepwarm;
37#[cfg(test)]
41#[path = "../../tests/support/cache_sim.rs"]
42pub(crate) mod cache_sim;
43pub mod memory_context;
44pub mod memory_history;
45pub mod message_time;
46pub mod openai;
47pub mod relay;
48mod request;
49mod request_encoding;
50mod sse;
51mod sse_types;
52mod stream;
53pub mod subagent;
54pub mod subagent_watch;
55pub mod telemetry;
56mod read_dedupe;
57pub mod tool_spill;
58pub mod trace;
59pub mod worker_context;
60pub(crate) mod transport;
61mod types;
62
63use api::ApiMethods;
64use auth::AuthMethods;
65use helpers::HelperMethods;
66use stream::StreamMethods;
67use types::AuthState;
68pub use types::{AgentEvent, CancelCause, LlmEvent, SessionEvent, StreamEvent, TurnCompletion};
69pub use stream::activation_policy;
70
71pub enum BeforeToolCallDecision {
73 Continue { input: Value },
74 Block { reason: String },
75}
76
77pub async fn emit_before_tool_call(
81 hook_bus: &Arc<crate::extensions::hooks::HookBus>,
82 tool_name: &str,
83 runtime_tool_name: Option<&str>,
84 input: Value,
85 session_id: Option<&str>,
86) -> crate::extensions::hooks::events::HookResult {
87 let mut event = crate::extensions::hooks::events::HookEvent::before_tool_call(tool_name, input)
88 .with_session(session_id);
89 if let Some(runtime_tool_name) = runtime_tool_name {
90 event.tool_runtime_name = Some(runtime_tool_name.to_string());
91 }
92 hook_bus.emit(&event).await
93}
94
95pub use crate::tools::CONFIRM_ANSWER_ALLOW_ALL;
101
102const MAX_CONFIRM_MESSAGE_CHARS: usize = 1200;
106const MAX_CONFIRM_MESSAGE_LINES: usize = 12;
107
108fn cap_confirm_message(message: &str) -> String {
114 let message = redact_secret_text(&crate::session::sanitize_prompt_text(message));
115 let lines: Vec<&str> = message.lines().collect();
116 let over_lines = lines.len() > MAX_CONFIRM_MESSAGE_LINES;
117 let over_chars = message.chars().count() > MAX_CONFIRM_MESSAGE_CHARS;
118 if !over_lines && !over_chars {
119 return message;
120 }
121 let kept = lines[..lines.len().min(MAX_CONFIRM_MESSAGE_LINES)].join("\n");
122 let kept: String = kept.chars().take(MAX_CONFIRM_MESSAGE_CHARS).collect();
123 format!("{kept}\n…[message truncated by host]")
124}
125
126const MAX_REWRITE_PREVIEW_LINES: usize = 6;
129const MAX_REWRITE_PREVIEW_CHARS: usize = 400;
130
131fn redact_secret_text(text: &str) -> String {
140 crate::runtime::trace::export::redact_command_text(text)
141}
142
143fn redact_preview_value(value: &mut Value) {
146 crate::runtime::trace::export::redact_value(value);
147 match value {
148 Value::Object(map) => map.values_mut().for_each(redact_preview_value),
149 Value::Array(items) => items.iter_mut().for_each(redact_preview_value),
150 Value::String(s) => *s = redact_secret_text(s),
151 _ => {}
152 }
153}
154
155fn final_input_preview(input: &Value) -> String {
158 let mut input = input.clone();
159 redact_preview_value(&mut input);
160 let full = match input.get("command").and_then(Value::as_str) {
161 Some(cmd) => cmd.to_string(),
162 None => serde_json::to_string_pretty(&input).unwrap_or_else(|_| input.to_string()),
163 };
164 let lines: Vec<&str> = full.lines().collect();
165 let mut out: String = lines[..lines.len().min(MAX_REWRITE_PREVIEW_LINES)]
166 .iter()
167 .map(|l| format!(" {l}"))
168 .collect::<Vec<_>>()
169 .join("\n");
170 let mut cut = lines.len() > MAX_REWRITE_PREVIEW_LINES;
171 if out.chars().count() > MAX_REWRITE_PREVIEW_CHARS {
172 out = out.chars().take(MAX_REWRITE_PREVIEW_CHARS).collect();
173 cut = true;
174 }
175 if cut {
176 out.push_str("\n …[preview truncated]");
177 }
178 out
179}
180
181fn confirm_prompt_body(
187 message: &str,
188 sources: &[String],
189 tool: Option<&str>,
190 rewrite: Option<(&[String], &Value)>,
191) -> String {
192 let source = if sources.is_empty() {
193 "(unknown)".to_string()
194 } else {
195 sources.join(", ")
196 };
197 let rewritten = match rewrite {
198 Some((by, input)) => format!(
199 "Input rewritten by: {}\nInput that will run:\n{}\n",
200 if by.is_empty() { "(unknown)".to_string() } else { by.join(", ") },
201 final_input_preview(input)
202 ),
203 None => String::new(),
204 };
205 crate::session::sanitize_prompt_text(&format!(
209 "Extension: {source}\nTool: {}\n{rewritten}\nMessage from {source}:\n{}\n\nAllow this tool call?",
210 tool.unwrap_or("(unknown)"),
211 cap_confirm_message(message)
212 ))
213}
214
215fn approved(
218 modified_input: Option<Value>,
219) -> crate::extensions::hooks::events::HookResult {
220 match modified_input {
221 Some(input) => crate::extensions::hooks::events::HookResult::Modify { input },
222 None => crate::extensions::hooks::events::HookResult::Continue,
223 }
224}
225
226pub async fn resolve_before_tool_call_result(
235 hook_result: crate::extensions::hooks::events::HookResult,
236 secret_prompt: Option<&crate::tools::SecretPromptHandle>,
237 auto_approve_confirms: bool,
238 session_allow_all: Option<&crate::tools::SessionConfirmLatch>,
239) -> crate::extensions::hooks::events::HookResult {
240 match hook_result {
241 crate::extensions::hooks::events::HookResult::Confirm {
242 message,
243 sources,
244 generations,
245 tool,
246 modified_input,
247 modified_by,
248 } => {
249 let keys = crate::tools::latch_keys(&sources, &generations);
250 if auto_approve_confirms {
251 tracing::info!(message = %message, "confirm auto-approved (auto_approve_confirms=true)");
252 return approved(modified_input);
253 }
254 if session_allow_all.is_some_and(|latch| latch.allows_all(&keys)) {
255 tracing::info!(
256 message = %message,
257 extensions = ?sources,
258 "confirm auto-approved (extension allow-all latched)"
259 );
260 return approved(modified_input);
261 }
262
263 let Some(prompt) = secret_prompt else {
264 return crate::extensions::hooks::events::HookResult::Block {
265 reason: format!(
266 "Tool call requires confirmation but no interactive prompt is available: {}",
267 redact_secret_text(&crate::session::sanitize_prompt_text(&message))
268 ),
269 };
270 };
271
272 let scope = (session_allow_all.is_some() && !sources.is_empty()).then(|| {
275 crate::session::ConfirmScope {
276 extensions: sources.clone(),
277 tool: tool.clone(),
278 rewritten_by: if modified_input.is_some() {
279 modified_by.clone()
280 } else {
281 Vec::new()
282 },
283 }
284 });
285 let response = prompt
286 .prompt_scoped(
287 crate::session::CONFIRM_PROMPT_TITLE.to_string(),
288 confirm_prompt_body(
289 &message,
290 &sources,
291 tool.as_deref(),
292 modified_input.as_ref().map(|i| (modified_by.as_slice(), i)),
293 ),
294 scope,
295 )
296 .await;
297
298 match response.as_deref().map(str::trim) {
299 Some(answer)
300 if answer.eq_ignore_ascii_case("yes") || answer.eq_ignore_ascii_case("y") =>
301 {
302 approved(modified_input)
303 }
304 Some(answer) if answer.eq_ignore_ascii_case(CONFIRM_ANSWER_ALLOW_ALL) => {
305 match session_allow_all {
306 Some(latch) if latch.allow(&keys) => tracing::warn!(
307 extensions = ?sources,
308 "user chose Allow all: confirms from these extensions are auto-approved for the rest of this conversation"
309 ),
310 Some(_) => tracing::warn!(
311 "user chose Allow all on an unattributed or revoked confirm; approving this call only"
312 ),
313 None => tracing::warn!(
314 "user chose Allow all but this path has no session latch; approving this call only"
315 ),
316 }
317 approved(modified_input)
318 }
319 _ => crate::extensions::hooks::events::HookResult::Block {
320 reason: format!(
321 "Tool call confirmation denied: {}",
322 redact_secret_text(&crate::session::sanitize_prompt_text(&message))
323 ),
324 },
325 }
326 }
327 other => other,
328 }
329}
330
331pub async fn resolve_before_tool_call_decision(
333 original_input: Value,
334 hook_result: crate::extensions::hooks::events::HookResult,
335 secret_prompt: Option<&crate::tools::SecretPromptHandle>,
336 auto_approve_confirms: bool,
337 session_allow_all: Option<&crate::tools::SessionConfirmLatch>,
338) -> BeforeToolCallDecision {
339 match resolve_before_tool_call_result(
340 hook_result,
341 secret_prompt,
342 auto_approve_confirms,
343 session_allow_all,
344 )
345 .await
346 {
347 crate::extensions::hooks::events::HookResult::Block { reason } => {
348 BeforeToolCallDecision::Block { reason }
349 }
350 crate::extensions::hooks::events::HookResult::Modify { input } => {
351 BeforeToolCallDecision::Continue { input }
352 }
353 _ => BeforeToolCallDecision::Continue {
354 input: original_input,
355 },
356 }
357}
358
359const MAX_REPLACE_OUTPUT: usize = 1024 * 1024; pub async fn emit_after_tool_call(
384 hook_bus: &Arc<crate::extensions::hooks::HookBus>,
385 tool_name: &str,
386 runtime_tool_name: Option<&str>,
387 input: Value,
388 output: String,
389 max_tool_output: usize,
390 session_id: Option<&str>,
391) -> String {
392 emit_after_tool_call_outcome(
393 hook_bus,
394 tool_name,
395 runtime_tool_name,
396 input,
397 output,
398 max_tool_output,
399 session_id,
400 false,
401 )
402 .await
403 .output
404}
405
406pub(super) struct AfterToolCallOutcome {
409 pub output: String,
411 pub replaced: bool,
412 pub uncut: Option<String>,
416}
417
418impl AfterToolCallOutcome {
419 pub fn full_text(&self) -> &str {
421 self.uncut.as_deref().unwrap_or(&self.output)
422 }
423}
424
425#[allow(clippy::too_many_arguments)] pub(super) async fn emit_after_tool_call_outcome(
427 hook_bus: &Arc<crate::extensions::hooks::HookBus>,
428 tool_name: &str,
429 runtime_tool_name: Option<&str>,
430 input: Value,
431 output: String,
432 max_tool_output: usize,
433 session_id: Option<&str>,
434 keep_uncut: bool,
435) -> AfterToolCallOutcome {
436 use crate::extensions::hooks::events::HookResult;
437 let original = output.clone();
439 let mut event =
440 crate::extensions::hooks::events::HookEvent::after_tool_call(tool_name, input, output)
441 .with_session(session_id);
442 if let Some(runtime_tool_name) = runtime_tool_name {
443 event.tool_runtime_name = Some(runtime_tool_name.to_string());
444 }
445 let mut replaced = false;
446 let post_hook = match hook_bus.emit(&event).await {
447 HookResult::Replace { mut output } => {
448 replaced = true;
449 if output.len() > MAX_REPLACE_OUTPUT {
450 tracing::warn!(
451 tool = %tool_name,
452 len = output.len(),
453 cap = MAX_REPLACE_OUTPUT,
454 "Extension Replace output exceeds cap — clamping",
455 );
456 let mut boundary = MAX_REPLACE_OUTPUT;
458 while boundary > 0 && !output.is_char_boundary(boundary) {
459 boundary -= 1;
460 }
461 output.truncate(boundary);
462 }
463 output
464 }
465 HookResult::Continue => original,
466 other => {
467 tracing::warn!(
468 tool = %tool_name,
469 ?other,
470 "Unexpected after_tool_call result — using original output",
471 );
472 original
473 }
474 };
475 let output =
479 crate::runtime::helpers::HelperMethods::truncate_tool_result(&post_hook, max_tool_output);
480 let uncut = (keep_uncut && post_hook.len() > max_tool_output).then_some(post_hook);
481 AfterToolCallOutcome {
482 output,
483 replaced,
484 uncut,
485 }
486}
487
488#[derive(Debug, Clone, Copy, PartialEq, Eq)]
491pub struct ReasoningClamp {
492 pub from: agent_core::reasoning::ReasoningLevel,
493 pub to: agent_core::reasoning::ReasoningLevel,
494}
495
496fn validated_single_tool_output(
497 model: &str,
498 output: crate::ToolOutput,
499) -> (String, Option<Vec<Value>>) {
500 let (summary, blocks) = output.into_parts();
501 if let Some(ref blocks) = blocks {
502 if let Err(error) = attachments::validate_tool_blocks(model, blocks) {
503 return (format!("Attachment not sent: {error}"), None);
504 }
505 }
506 (summary, blocks)
507}
508
509fn retain_single_tool_blocks(
512 blocks: Option<Vec<Value>>,
513 original: &str,
514 hooked: &str,
515) -> Option<Vec<Value>> {
516 if original == hooked {
517 blocks
518 } else {
519 None
520 }
521}
522
523fn single_tool_result_content(
524 result: &str,
525 blocks: Option<Vec<Value>>,
526 max_tool_output: usize,
527) -> Value {
528 match blocks {
529 Some(blocks) => Value::Array(blocks),
530 None => Value::String(HelperMethods::truncate_tool_result(result, max_tool_output)),
531 }
532}
533
534#[derive(Clone)]
537struct PromptReloadSource {
538 manifest: PathBuf,
539 context: agent_core::prompt::SelectionContext,
540 user_module: Option<agent_core::prompt::PromptModule>,
541 delegation_policy_digest: Option<String>,
542}
543
544pub struct Runtime {
545 client: Client,
546 anthropic_stream_client: Arc<std::sync::OnceLock<Client>>,
547 auth: Arc<RwLock<AuthState>>,
548 model: String,
549 tools: Arc<RwLock<ToolRegistry>>,
550 system_prompt: Option<String>,
551 effective_prompt: Option<agent_core::prompt::PromptStack>,
553 prompt_generation: u64,
554 prompt_reload_source: Option<PromptReloadSource>,
556 thinking_budget: u32,
557 named_level: Option<agent_core::reasoning::ReasoningLevel>,
561 explicit_reasoning: bool,
566 codex_request_role: crate::runtime::openai::catalog::CodexRequestRole,
569 context_window_override: Option<u64>,
574 pub(crate) continuation: continuation::SharedContinuation,
577 worker_context: worker_context::WorkerContext,
581 compaction_model: Option<String>,
583 compaction_mode: agent_core::compaction::CompactionMode,
585 compaction_exclusions: Vec<agent_core::compaction::ContentClass>,
587 remote_summarization_attempts: Arc<std::sync::atomic::AtomicU64>,
592 subagent_registry: Arc<Mutex<crate::runtime::subagent::SubagentRegistry>>,
594 orchestration: Option<Arc<crate::orchestration::OrchestrationRuntime>>,
596 event_queue: Arc<crate::events::EventQueue>,
598 pub watcher_exit_path: Option<PathBuf>,
600 max_tool_output: usize,
602 tool_spill_policy: tool_spill::ToolSpillPolicy,
608 tool_spill_root: PathBuf,
610 tool_spill_stats: Arc<tool_spill::ToolSpillStats>,
612 read_dedupe: bool,
615 thinking_display: crate::config::ThinkingDisplay,
618 tool_keepwarm: bool,
622 bash_timeout: u64,
623 bash_max_timeout: u64,
624 subagent_timeout: u64,
625 api_retries: u32,
626 refusal_retries: u32,
627 telemetry_level: crate::runtime::telemetry::TelemetryLevel,
629 telemetry_log_generations: u32,
632 telemetry_writer: Option<crate::runtime::telemetry::TelemetryWriter>,
638 trace_ctx: trace::TraceContext,
643 trace_controls: std::sync::Arc<trace::TraceControls>,
647 memory_context_state: std::sync::Arc<std::sync::Mutex<memory_context::SessionMemoryState>>,
659 memory_backend: crate::memory_backend::MemoryBinding,
662 memory_backend_config: Option<crate::config::MemoryBackendConfig>,
664 memory_backend_reconfigure_denied: Arc<std::sync::atomic::AtomicBool>,
666 #[cfg(any(test, feature = "testing"))]
670 capture_provider_for_test: std::sync::Arc<
671 std::sync::Mutex<Option<std::sync::Arc<dyn capture_worker::CaptureProvider>>>,
672 >,
673 pending_history_import_preview:
677 std::sync::Arc<std::sync::Mutex<Option<memory_history::HistoryImportPreview>>>,
678 history_import_plan: std::sync::Arc<std::sync::Mutex<Option<memory_history::ImportPlan>>>,
680 pending_memory_segment:
691 std::sync::Arc<std::sync::Mutex<Option<memory_context::ContextSegment>>>,
692 retained_recall_turn:
699 std::sync::Arc<std::sync::Mutex<Option<memory_context::RetainedRecallTurn>>>,
700 one_shot_trace_writer:
708 std::sync::Arc<std::sync::Mutex<Option<crate::runtime::telemetry::TelemetryWriter>>>,
709 capture_dir: std::path::PathBuf,
715 cache_diagnostics: bool,
717 cache_ttl: crate::core::config::CacheTtlSetting,
721 request_compression: request_encoding::RequestCompression,
725 ttl_downgrade_notified: std::sync::Arc<std::sync::atomic::AtomicBool>,
728 saw_1h_honored: std::sync::Arc<std::sync::atomic::AtomicBool>,
732 cache_anchor: std::sync::Arc<keepwarm::CacheAnchorCell>,
736 last_msg_id: Arc<Mutex<Option<String>>>,
740 cache_diag_rejected: Arc<std::sync::atomic::AtomicBool>,
743 session_manager: std::sync::Arc<crate::tools::shell::SessionManager>,
744 hook_bus: Arc<crate::extensions::hooks::HookBus>,
746 #[allow(dead_code)]
748 reaper_handle: Option<tokio::task::JoinHandle<()>>,
749 #[allow(dead_code)]
750 reaper_cancel: Option<tokio_util::sync::CancellationToken>,
751 credential_source: crate::auth::CredentialSource,
755 token_cache: crate::auth::TokenCache,
758 trusted_worker_models: Vec<agent_core::prompt::QualifiedModelId>,
762 subagent_limits: crate::config::SubagentLimits,
766 progressive_tool_disclosure: bool,
770 deferred_tool_loading: bool,
773 activation_confirm: agent_core::config::ActivationConfirm,
776 delegation_parent: Option<String>,
779 session_allow_all: std::sync::Arc<crate::tools::SessionConfirmLatch>,
784 mcp_runtime: Option<std::sync::Arc<crate::mcp::McpRuntimeManager>>,
788 mcp_session_scope: Option<std::sync::Arc<crate::mcp::McpSessionEndGuard>>,
792 extension_runtime: Option<std::sync::Arc<crate::extensions::lease::ExtensionRuntimeManager>>,
796 extension_session_scope:
800 Option<std::sync::Arc<crate::extensions::lease::ExtensionSessionEndGuard>>,
801 turn_budget: crate::runtime::budget::TurnBudget,
804 owner_steer_budget: Option<Arc<crate::runtime::budget::OwnerSteerBudget>>,
808 message_clock: crate::runtime::message_time::MessageClock,
810 host_tool_session: crate::tools::activation::SessionId,
818 tool_catalog_prompt: Arc<std::sync::Mutex<Option<Option<String>>>>,
823 retained_tool_set: crate::tools::activation::RetainedSessionToolSet,
827 pending_tool_restore: crate::tools::activation::PendingToolRestore,
831 session_id: Option<String>,
835 cwd: Option<PathBuf>,
839 env: Option<crate::session::types::SessionEnv>,
843 env_stripped: Vec<String>,
845 env_warned: std::sync::Arc<std::sync::Mutex<std::collections::HashSet<String>>>,
849}
850
851fn fresh_host_tool_session() -> crate::tools::activation::SessionId {
855 crate::tools::activation::SessionId::parse(&format!(
856 "runtime-{}-{}",
857 std::process::id(),
858 uuid::Uuid::new_v4()
859 ))
860 .expect("generated runtime session id is always valid")
861}
862
863fn fresh_memory_context_state(
870) -> std::sync::Arc<std::sync::Mutex<memory_context::SessionMemoryState>> {
871 let session = memory_context::SessionId::parse(&format!(
872 "memctx-{}-{}",
873 std::process::id(),
874 uuid::Uuid::new_v4()
875 ))
876 .expect("generated memory session id is always valid");
877 std::sync::Arc::new(std::sync::Mutex::new(
878 memory_context::SessionMemoryState::new(session),
879 ))
880}
881
882fn memory_project_scope() -> std::result::Result<agent_core::memory::store::ProjectScope, String> {
886 let cwd = std::env::current_dir()
887 .map_err(|error| format!("cannot resolve current project directory: {error}"))?;
888 agent_core::memory::store::ProjectScope::discover(&cwd)
889 .map_err(|error| format!("cannot resolve project scope: {error}"))
890}
891
892fn memory_project_id() -> memory_context::ProjectId {
893 match memory_project_scope() {
894 Ok(scope) => memory_context::ProjectId::parse(scope.key())
895 .expect("host project scope key is always a valid identifier"),
896 Err(_) => memory_context::ProjectId::parse("project-unresolved")
897 .expect("static fallback project id is valid"),
898 }
899}
900
901fn memory_capture_worker() -> &'static capture_worker::CaptureWorker {
910 static WORKER: std::sync::OnceLock<capture_worker::CaptureWorker> = std::sync::OnceLock::new();
911 WORKER.get_or_init(|| capture_worker::CaptureWorker::new(32))
912}
913
914struct ExtensionCaptureProvider {
915 manager: std::sync::Arc<crate::extensions::lease::ExtensionRuntimeManager>,
916 session: crate::tools::activation::SessionId,
917 plugin: String,
918 digest: crate::tools::catalog::SchemaDigest,
919 handle: tokio::runtime::Handle,
920}
921
922impl capture_worker::CaptureProvider for ExtensionCaptureProvider {
923 fn capture(
924 &self,
925 capture: chat_capture::ChatTurnCapture,
926 ) -> std::result::Result<(), capture_worker::CaptureFailure> {
927 let params = memory_context::capture_request_wire(&capture);
928 self.handle
929 .block_on(
930 crate::extensions::lease::ExtensionLeaseCapability::new(
931 self.session.clone(),
932 self.manager.clone(),
933 )
934 .call_exact(
935 &self.plugin,
936 memory_context::MEMORY_CAPTURE_TOOL_NAME,
937 &self.digest,
938 params,
939 ),
940 )
941 .map(|_| ())
942 .map_err(|_| capture_worker::CaptureFailure {
943 code: "provider_call_failed",
944 })
945 }
946
947 fn contains_capture(
948 &self,
949 capture_id: &[u8; 32],
950 ) -> std::result::Result<capture_worker::CaptureCommitState, capture_worker::CaptureFailure>
951 {
952 let capture_id = chat_capture::CaptureId::from_bytes(*capture_id);
953 let response = self
954 .handle
955 .block_on(
956 crate::extensions::lease::ExtensionLeaseCapability::new(
957 self.session.clone(),
958 self.manager.clone(),
959 )
960 .call_exact(
961 &self.plugin,
962 memory_context::MEMORY_CAPTURE_TOOL_NAME,
963 &self.digest,
964 memory_context::capture_query_wire(&capture_id),
965 ),
966 )
967 .map_err(|_| capture_worker::CaptureFailure {
968 code: "provider_query_failed",
969 })?;
970 Ok(
971 if response
972 .get("committed")
973 .and_then(serde_json::Value::as_bool)
974 == Some(true)
975 {
976 capture_worker::CaptureCommitState::Committed
977 } else {
978 capture_worker::CaptureCommitState::Absent
979 },
980 )
981 }
982
983 fn capture_summary(
984 &self,
985 capture: chat_capture::ConversationSummaryCapture,
986 ) -> std::result::Result<(), capture_worker::CaptureFailure> {
987 let params = memory_context::summary_capture_request_wire(&capture);
988 self.handle
989 .block_on(
990 crate::extensions::lease::ExtensionLeaseCapability::new(
991 self.session.clone(),
992 self.manager.clone(),
993 )
994 .call_exact(
995 &self.plugin,
996 memory_context::MEMORY_CAPTURE_TOOL_NAME,
997 &self.digest,
998 params,
999 ),
1000 )
1001 .map(|_| ())
1002 .map_err(|_| capture_worker::CaptureFailure {
1003 code: "provider_call_failed",
1004 })
1005 }
1006}
1007
1008fn terminal_capture_history(
1009 lease: &memory_context::MemoryContextLease,
1010 messages: &[crate::SharedMessage],
1011 started_at: std::time::SystemTime,
1012) -> chat_capture::TerminalTurnHistory {
1013 use agent_core::core::disclosure::DisclosureClass;
1014 let mut items = Vec::new();
1015 for message in messages {
1016 let Some(role) = message.get("role").and_then(serde_json::Value::as_str) else {
1017 continue;
1018 };
1019 let class = match role {
1020 "user" => chat_capture::CaptureContentClass::UserMessage,
1021 "assistant" => chat_capture::CaptureContentClass::AssistantFinal,
1022 _ => continue,
1023 };
1024 let text = message
1025 .get("content")
1026 .and_then(serde_json::Value::as_str)
1027 .map(str::to_owned)
1028 .unwrap_or_else(|| {
1029 message
1030 .get("content")
1031 .map(ToString::to_string)
1032 .unwrap_or_default()
1033 });
1034 items.push(chat_capture::CanonicalCaptureItem {
1035 project_id: lease.project_id.clone(),
1036 class,
1037 disclosure: DisclosureClass::ModelVisible,
1038 sensitivity: chat_capture::Sensitivity::Normal,
1039 text,
1040 tool_name: None,
1041 });
1042 }
1043 chat_capture::TerminalTurnHistory {
1044 project_id: lease.project_id.clone(),
1045 session_id: lease.session_id.clone(),
1046 turn_id: memory_context::TurnId::parse(&format!("turn-{}", uuid::Uuid::new_v4()))
1047 .expect("generated turn id"),
1048 turn_ordinal: 0,
1049 started_at,
1050 completed_at: std::time::SystemTime::now(),
1051 outcome: agent_core::TurnOutcome::Completed,
1052 items,
1053 compaction: None,
1054 }
1055}
1056
1057fn memory_provider_id() -> memory_context::ContextProviderId {
1058 memory_context::ContextProviderId::parse("axel-memory")
1059 .expect("static provider id is always valid")
1060}
1061
1062fn terminal_capture_start(messages: &[crate::SharedMessage]) -> Option<crate::SharedMessage> {
1063 messages
1064 .iter()
1065 .rfind(|message| {
1066 message["role"] == "user"
1067 && (message["content"].is_string()
1068 || message["content"].as_array().is_some_and(|blocks| {
1069 blocks.iter().any(|block| block["type"] != "tool_result")
1070 }))
1071 })
1072 .cloned()
1073}
1074
1075fn terminal_capture_messages<'a>(
1076 messages: &'a [crate::SharedMessage],
1077 start: &crate::SharedMessage,
1078) -> Option<&'a [crate::SharedMessage]> {
1079 let index = messages
1083 .iter()
1084 .position(|message| Arc::ptr_eq(message, start))?;
1085 Some(&messages[index..])
1086}
1087
1088fn dispatch_completed_terminal_capture(
1091 completed: bool,
1092 final_history: &std::sync::Mutex<Option<Vec<crate::SharedMessage>>>,
1093 state: &std::sync::Mutex<memory_context::SessionMemoryState>,
1094 selected: Option<(
1095 memory_context::MemoryContextLease,
1096 Arc<dyn capture_worker::CaptureProvider>,
1097 )>,
1098 started_at: std::time::SystemTime,
1099 turn_start: Option<&crate::SharedMessage>,
1100) -> bool {
1101 let history = final_history
1102 .lock()
1103 .unwrap_or_else(std::sync::PoisonError::into_inner)
1104 .take();
1105 let (Some(messages), Some((lease, provider))) = (history, selected) else {
1106 return false;
1107 };
1108 if !completed {
1109 return false;
1110 }
1111 let Some(messages) = turn_start.and_then(|start| terminal_capture_messages(&messages, start))
1112 else {
1113 return false;
1114 };
1115 let state = state
1116 .lock()
1117 .unwrap_or_else(std::sync::PoisonError::into_inner);
1118 if state
1119 .capture_lease_at(std::time::SystemTime::now())
1120 .as_ref()
1121 != Some(&lease)
1122 {
1123 return false;
1124 }
1125 memory_capture_worker()
1126 .submit_terminal(
1127 &lease,
1128 terminal_capture_history(&lease, messages, started_at),
1129 memory_context::RetentionClass::Standard,
1130 provider,
1131 )
1132 .unwrap_or(false)
1133}
1134
1135const HTTP_READ_TIMEOUT: Duration = Duration::from_secs(90);
1140
1141fn build_http_client(read_timeout: Duration) -> reqwest::Result<Client> {
1154 Client::builder()
1155 .tls_built_in_webpki_certs(true)
1156 .tls_built_in_native_certs(true)
1157 .connect_timeout(Duration::from_secs(10))
1158 .read_timeout(read_timeout)
1159 .build()
1160}
1161
1162pub fn build_host_http_client() -> Result<Client> {
1166 build_http_client(HTTP_READ_TIMEOUT)
1167 .map_err(|e| RuntimeError::Config(format!("Failed to build HTTP client: {}", e)))
1168}
1169
1170pub(crate) struct RuntimeParts {
1174 pub host: crate::host::HostParts,
1175 pub session_manager: Arc<crate::tools::shell::SessionManager>,
1176 pub reaper: Option<(tokio::task::JoinHandle<()>, CancellationToken)>,
1177}
1178
1179impl RuntimeParts {
1180 pub(crate) fn with_reaper(host: crate::host::HostParts) -> Self {
1182 let session_manager = fresh_session_manager();
1183 let cancel = CancellationToken::new();
1184 let handle =
1185 crate::tools::shell::session::start_reaper(session_manager.clone(), cancel.clone());
1186 Self {
1187 host,
1188 session_manager,
1189 reaper: Some((handle, cancel)),
1190 }
1191 }
1192
1193 #[cfg_attr(not(any(test, feature = "testing")), allow(dead_code))]
1195 pub(crate) fn without_reaper(host: crate::host::HostParts) -> Self {
1196 Self {
1197 host,
1198 session_manager: fresh_session_manager(),
1199 reaper: None,
1200 }
1201 }
1202}
1203
1204fn fresh_session_manager() -> Arc<crate::tools::shell::SessionManager> {
1205 let config = crate::tools::shell::ShellConfig::default();
1206 crate::tools::shell::SessionManager::new(config)
1207}
1208
1209fn fresh_host_parts() -> Result<crate::host::HostParts> {
1213 let client = build_host_http_client()?;
1214 let capture_dir = trace::default_capture_dir();
1219 let _ = trace::sweep_expired_captures(&capture_dir);
1220 let spill_root = sweep_spill_root();
1222 Ok(crate::host::HostParts {
1223 client,
1224 tools: Arc::new(RwLock::new(ToolRegistry::new())),
1225 hook_bus: Arc::new(crate::extensions::hooks::HookBus::new()),
1226 credential_source: crate::auth::CredentialSource::Local,
1227 token_cache: crate::auth::TokenCache::new(),
1228 mcp_runtime: None,
1229 extension_runtime: None,
1230 capture_dir,
1231 spill_root,
1232 progressive_tool_disclosure: false,
1233 })
1234}
1235
1236pub(crate) fn sweep_spill_root() -> PathBuf {
1240 let root = agent_core::core::tool_spill::default_spill_root();
1241 let _ = agent_core::core::tool_spill::sweep_stale_spill(
1242 &root,
1243 agent_core::core::tool_spill::SPILL_MAX_AGE,
1244 std::time::SystemTime::now(),
1245 );
1246 root
1247}
1248
1249fn canonical_trusted_worker_model(model: &str) -> String {
1253 let model = model.trim();
1254 if let Some(id) = model.strip_prefix("claude/") {
1255 format!("anthropic/{id}")
1256 } else if model.starts_with("claude-") {
1257 format!("anthropic/{model}")
1258 } else {
1259 model.to_owned()
1260 }
1261}
1262
1263impl Drop for Runtime {
1267 fn drop(&mut self) {
1268 if let Some(c) = &self.reaper_cancel {
1269 c.cancel();
1270 }
1271 }
1272}
1273
1274impl Runtime {
1275 pub async fn new() -> Result<Self> {
1276 Ok(Self::from_parts(RuntimeParts::with_reaper(
1278 fresh_host_parts()?,
1279 )))
1280 }
1281
1282 pub(crate) fn from_parts(parts: RuntimeParts) -> Self {
1285 let (auth_token, auth_type, refresh_token, token_expires) =
1289 (String::new(), "oauth".to_string(), None, Some(0));
1290 let RuntimeParts {
1291 host,
1292 session_manager,
1293 reaper,
1294 } = parts;
1295 let (reaper_handle, reaper_cancel) = match reaper {
1296 Some((h, c)) => (Some(h), Some(c)),
1297 None => (None, None),
1298 };
1299 let mcp_runtime = host.mcp_runtime;
1300 let extension_runtime = host.extension_runtime;
1301
1302 let mut runtime = Runtime {
1303 client: host.client,
1304 anthropic_stream_client: Default::default(),
1305 auth: Arc::new(RwLock::new(AuthState {
1306 auth_token,
1307 auth_type,
1308 refresh_token,
1309 token_expires,
1310 auth_file_mtime: None,
1311 bound_credential: None,
1312 })),
1313 model: crate::models::default_model().to_string(),
1314 tools: host.tools,
1315 system_prompt: None,
1316 effective_prompt: None,
1317 prompt_generation: 0,
1318 prompt_reload_source: None,
1319 thinking_budget: 4096,
1320 named_level: None,
1321 explicit_reasoning: false,
1322 codex_request_role: crate::runtime::openai::catalog::CodexRequestRole::Foreground,
1323 context_window_override: None,
1324 compaction_model: None,
1325 compaction_mode: agent_core::compaction::CompactionMode::default(),
1326 compaction_exclusions: Vec::new(),
1327 remote_summarization_attempts: Arc::new(std::sync::atomic::AtomicU64::new(0)),
1328 subagent_registry: Arc::new(Mutex::new(
1329 crate::runtime::subagent::SubagentRegistry::new(),
1330 )),
1331 orchestration: None,
1332 event_queue: Arc::new(crate::events::EventQueue::new(1000)),
1333 watcher_exit_path: None,
1334 max_tool_output: 30000,
1335 tool_spill_policy: tool_spill::ToolSpillPolicy::OFF,
1336 tool_spill_root: host.spill_root,
1337 tool_spill_stats: Arc::default(),
1338 read_dedupe: false,
1339 thinking_display: crate::config::ThinkingDisplay::Summarized,
1340 tool_keepwarm: false,
1341 bash_timeout: 30,
1342 bash_max_timeout: 300,
1343 subagent_timeout: 300,
1344 api_retries: 3,
1345 refusal_retries: 2,
1346 telemetry_level: crate::runtime::telemetry::TelemetryLevel::Off,
1347 telemetry_log_generations: crate::runtime::telemetry::DEFAULT_MAX_GENERATIONS,
1348 telemetry_writer: None,
1349 trace_ctx: trace::TraceContext::disabled(),
1350 trace_controls: std::sync::Arc::new(trace::TraceControls::new()),
1351 memory_context_state: fresh_memory_context_state(),
1354 memory_backend: crate::memory_backend::MemoryBinding::legacy_current(),
1356 memory_backend_config: None,
1357 memory_backend_reconfigure_denied: Arc::new(std::sync::atomic::AtomicBool::new(false)),
1358 continuation: Arc::new(Mutex::new(continuation::ContinuationState::default())),
1359 worker_context: Default::default(),
1360 pending_history_import_preview: std::sync::Arc::new(std::sync::Mutex::new(None)),
1361 history_import_plan: std::sync::Arc::new(std::sync::Mutex::new(None)),
1362 #[cfg(any(test, feature = "testing"))]
1363 capture_provider_for_test: std::sync::Arc::new(std::sync::Mutex::new(None)),
1364 pending_memory_segment: std::sync::Arc::new(std::sync::Mutex::new(None)),
1367 retained_recall_turn: std::sync::Arc::new(std::sync::Mutex::new(None)),
1368 one_shot_trace_writer: std::sync::Arc::new(std::sync::Mutex::new(None)),
1369 capture_dir: host.capture_dir,
1370 cache_diagnostics: false,
1371 cache_ttl: crate::core::config::CacheTtlSetting::default(),
1372 request_compression: Default::default(),
1373 ttl_downgrade_notified: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
1374 saw_1h_honored: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
1375 cache_anchor: Default::default(),
1376 last_msg_id: Arc::new(Mutex::new(None)),
1377 cache_diag_rejected: Default::default(),
1378 session_manager,
1379 hook_bus: host.hook_bus,
1380 reaper_handle,
1381 reaper_cancel,
1382 credential_source: host.credential_source,
1383 token_cache: host.token_cache,
1384 trusted_worker_models: Vec::new(),
1385 subagent_limits: crate::config::SubagentLimits::default(),
1386 progressive_tool_disclosure: host.progressive_tool_disclosure,
1387 deferred_tool_loading: true,
1388 activation_confirm: agent_core::config::ActivationConfirm::default(),
1389 delegation_parent: None,
1390 session_allow_all: std::sync::Arc::new(crate::tools::SessionConfirmLatch::new()),
1391 mcp_runtime: None,
1392 mcp_session_scope: None,
1393 extension_runtime: None,
1394 extension_session_scope: None,
1395 turn_budget: crate::runtime::budget::TurnBudget::for_role(
1396 crate::runtime::budget::TurnRole::Foreground,
1397 ),
1398 owner_steer_budget: None,
1399 message_clock: Default::default(),
1400 host_tool_session: fresh_host_tool_session(),
1401 tool_catalog_prompt: Arc::new(std::sync::Mutex::new(None)),
1402 retained_tool_set: Default::default(),
1403 pending_tool_restore: Default::default(),
1404 session_id: None,
1405 cwd: None,
1406 env: None,
1407 env_stripped: Vec::new(),
1408 env_warned: Default::default(),
1409 };
1410 if let Some(m) = mcp_runtime {
1414 runtime.install_mcp_runtime(m);
1415 }
1416 if let Some(m) = extension_runtime {
1417 runtime.install_extension_runtime(m);
1418 }
1419 runtime
1420 }
1421
1422
1423 #[cfg(any(test, feature = "testing"))]
1434 pub fn new_headless() -> Self {
1435 let client = build_http_client(HTTP_READ_TIMEOUT)
1436 .expect("reqwest client construction is infallible with built-in roots");
1437 let host = crate::host::HostParts {
1438 client,
1439 tools: Arc::new(RwLock::new(ToolRegistry::new())),
1440 hook_bus: Arc::new(crate::extensions::hooks::HookBus::new()),
1441 credential_source: crate::auth::CredentialSource::Local,
1442 token_cache: crate::auth::TokenCache::new(),
1443 mcp_runtime: None,
1444 extension_runtime: None,
1445 capture_dir: trace::default_capture_dir(),
1446 spill_root: agent_core::core::tool_spill::default_spill_root(),
1447 progressive_tool_disclosure: false,
1448 };
1449 let mut runtime = Self::from_parts(RuntimeParts::without_reaper(host));
1450 runtime.auth = Arc::new(RwLock::new(AuthState {
1451 auth_token: "test-token".to_string(),
1452 auth_type: "api_key".to_string(),
1453 refresh_token: None,
1454 token_expires: None,
1455 auth_file_mtime: None,
1456 bound_credential: None,
1457 }));
1458 runtime
1459 }
1460
1461 pub fn set_system_prompt(&mut self, prompt: String) {
1462 self.system_prompt = Some(prompt);
1463 }
1464
1465 pub fn system_prompt(&self) -> Option<&str> {
1466 self.system_prompt.as_deref()
1467 }
1468
1469 pub async fn effective_system_prompt(&self) -> Option<String> {
1481 let base = self.orchestrated_system_prompt().await;
1482 if self.effective_prompt.is_some() {
1484 return base;
1485 }
1486 match self.frozen_tool_catalog().await {
1487 Some(catalog) => Some(match base {
1488 Some(base) if !base.is_empty() => format!("{base}\n\n{catalog}"),
1489 _ => catalog,
1490 }),
1491 None => base,
1492 }
1493 }
1494
1495 pub async fn frozen_tool_catalog(&self) -> Option<String> {
1501 if !self.progressive_tool_disclosure
1502 || self.activation_confirm == agent_core::config::ActivationConfirm::Deny
1503 {
1504 return None;
1505 }
1506 if let Some(frozen) = self
1507 .tool_catalog_prompt
1508 .lock()
1509 .unwrap_or_else(std::sync::PoisonError::into_inner)
1510 .clone()
1511 {
1512 return frozen;
1513 }
1514 let rendered = {
1515 let registry = self.tools.read().await;
1516 if registry.get("activate_tools").is_none() {
1517 None
1518 } else {
1519 let core_set = continuation::context_tool_set(
1520 self.host_tool_session.clone(),
1521 registry.catalog(),
1522 true,
1523 self.context_management_enabled(),
1524 );
1525 let core: std::collections::HashSet<String> = core_set
1526 .core_ids()
1527 .map(|id| id.as_str().to_string())
1528 .collect();
1529 crate::tools::prompt_catalog::render(
1530 registry.catalog(),
1531 &core,
1532 crate::tools::activation::is_activatable_record,
1533 )
1534 }
1535 };
1536 let mut slot = self
1537 .tool_catalog_prompt
1538 .lock()
1539 .unwrap_or_else(std::sync::PoisonError::into_inner);
1540 slot.get_or_insert(rendered).clone()
1542 }
1543
1544 async fn orchestrated_system_prompt(&self) -> Option<String> {
1548 if self.effective_prompt.is_some() {
1549 return self.system_prompt.clone();
1550 }
1551 if self.tools.read().await.get("subagent_start").is_none() {
1552 return self.system_prompt.clone();
1553 }
1554 let Ok(model) = crate::orchestration::canonical_foreground_identity(&self.model) else {
1555 return self.system_prompt.clone();
1556 };
1557 let workflow_mode = match self.reasoning_level() {
1558 agent_core::reasoning::ReasoningLevel::UltraCode => {
1559 Some(agent_core::prompt::WorkflowMode::UltraCode)
1560 }
1561 agent_core::reasoning::ReasoningLevel::Max => {
1562 Some(agent_core::prompt::WorkflowMode::Max)
1563 }
1564 agent_core::reasoning::ReasoningLevel::XHigh => {
1565 Some(agent_core::prompt::WorkflowMode::XHigh)
1566 }
1567 _ => None,
1568 };
1569 let Ok(context) = agent_core::prompt::SelectionContext::new(model, None)
1570 .map(|context| context.with_workflow_mode(workflow_mode))
1571 else {
1572 return self.system_prompt.clone();
1573 };
1574 agent_core::prompt::compose_orchestration_prompt(self.system_prompt.as_deref(), &context)
1575 }
1576
1577 pub fn effective_prompt(&self) -> Option<&agent_core::prompt::PromptStack> {
1578 self.effective_prompt.as_ref()
1579 }
1580
1581 pub fn prompt_generation(&self) -> u64 {
1582 self.prompt_generation
1583 }
1584
1585 pub fn retain_prompt_reload_source(
1586 &mut self,
1587 manifest: PathBuf,
1588 context: agent_core::prompt::SelectionContext,
1589 user_module: Option<agent_core::prompt::PromptModule>,
1590 delegation_policy_digest: Option<String>,
1591 ) {
1592 self.prompt_reload_source = Some(PromptReloadSource {
1593 manifest,
1594 context,
1595 user_module,
1596 delegation_policy_digest,
1597 });
1598 }
1599
1600 pub fn reload_prompt(&mut self) -> std::result::Result<u64, agent_core::prompt::PromptError> {
1602 let source = self.prompt_reload_source.clone().ok_or_else(|| {
1603 agent_core::prompt::PromptError::Invalid("no prompt manifest is active".into())
1604 })?;
1605 let raw = std::fs::read_to_string(&source.manifest).map_err(|_| {
1606 agent_core::prompt::PromptError::Invalid("prompt manifest is unavailable".into())
1607 })?;
1608 let manifest = agent_core::prompt::PromptManifest::parse(&raw)?;
1609 let reload_catalog = crate::orchestration::OrchestrationRuntime::trusted_catalog(
1610 source.context.model(),
1611 manifest.delegation_catalog_candidates(),
1612 )
1613 .map_err(|error| agent_core::prompt::PromptError::Invalid(error.into()))?;
1614 let candidate_policy_digest = manifest
1615 .delegation_policy(source.context.model().clone(), &reload_catalog)?
1616 .map(|policy| policy.digest());
1617 if candidate_policy_digest != source.delegation_policy_digest {
1618 return Err(agent_core::prompt::PromptError::Invalid(
1619 "hot reload cannot safely change delegation policy".into(),
1620 ));
1621 }
1622 let registry = manifest.registry(source.manifest.parent())?;
1623 let candidate = agent_core::prompt::compile_prompt_stack(
1624 &manifest,
1625 ®istry,
1626 &source.context,
1627 source.user_module,
1628 )?;
1629 self.apply_prompt_stack(candidate)
1630 }
1631
1632 pub fn apply_prompt_stack(
1634 &mut self,
1635 candidate: agent_core::prompt::PromptStack,
1636 ) -> std::result::Result<u64, agent_core::prompt::PromptError> {
1637 if let Some(current) = &self.effective_prompt {
1638 current.validate_hot_reload(&candidate)?;
1639 }
1640 let composed = candidate.composed().to_owned();
1641 self.effective_prompt = Some(candidate);
1642 self.system_prompt = Some(composed);
1643 self.prompt_generation = self.prompt_generation.saturating_add(1);
1644 Ok(self.prompt_generation)
1645 }
1646
1647 pub fn prompt_inspection_json(&self) -> Option<String> {
1648 let mode = self
1649 .orchestration
1650 .as_ref()
1651 .map(|runtime| runtime.enforcement_mode())
1652 .unwrap_or(agent_core::orchestration::EnforcementMode::Off);
1653 self.effective_prompt.as_ref().and_then(|stack| {
1654 serde_json::to_string(&serde_json::json!({
1655 "generation": self.prompt_generation,
1656 "effective": stack.inspect(mode),
1657 "token_estimate": stack.composed().len().div_ceil(4),
1658 }))
1659 .ok()
1660 })
1661 }
1662
1663 pub fn install_mcp_runtime(&mut self, manager: std::sync::Arc<crate::mcp::McpRuntimeManager>) {
1668 self.mcp_session_scope = Some(std::sync::Arc::new(crate::mcp::McpSessionEndGuard::new(
1669 self.host_tool_session.clone(),
1670 std::sync::Arc::clone(&manager),
1671 )));
1672 self.mcp_runtime = Some(manager);
1673 }
1674
1675 pub fn set_turn_budget(&mut self, budget: crate::runtime::budget::TurnBudget) {
1679 self.turn_budget = budget;
1680 }
1681
1682 pub fn set_owner_steer_budget(
1686 &mut self,
1687 lift: Option<Arc<crate::runtime::budget::OwnerSteerBudget>>,
1688 ) {
1689 self.owner_steer_budget = lift;
1690 }
1691
1692 pub fn message_clock(&self) -> &crate::runtime::message_time::MessageClock {
1694 &self.message_clock
1695 }
1696
1697 pub fn set_message_clock(&mut self, clock: crate::runtime::message_time::MessageClock) {
1698 self.message_clock = clock;
1699 }
1700
1701 pub fn turn_budget(&self) -> &crate::runtime::budget::TurnBudget {
1703 &self.turn_budget
1704 }
1705
1706 pub fn install_extension_runtime(
1711 &mut self,
1712 manager: std::sync::Arc<crate::extensions::lease::ExtensionRuntimeManager>,
1713 ) {
1714 manager.bind_host_scope(self.host_tool_session.clone());
1718 self.extension_session_scope = Some(std::sync::Arc::new(
1719 crate::extensions::lease::ExtensionSessionEndGuard::new(
1720 self.host_tool_session.clone(),
1721 std::sync::Arc::clone(&manager),
1722 ),
1723 ));
1724 self.extension_runtime = Some(manager);
1725 }
1726
1727 pub fn install_orchestration(
1728 &mut self,
1729 runtime: Arc<crate::orchestration::OrchestrationRuntime>,
1730 ) {
1731 for model in &self.trusted_worker_models {
1732 if let Err(error) = runtime.grant_worker_model(model.as_str()) {
1733 tracing::warn!(
1734 model = model.as_str(),
1735 error = %error,
1736 "failed to apply configured worker-model trust"
1737 );
1738 }
1739 }
1740 self.orchestration = Some(runtime);
1741 }
1742
1743 pub fn install_worker_orchestration(
1749 &mut self,
1750 runtime: Arc<crate::orchestration::OrchestrationRuntime>,
1751 ) {
1752 self.orchestration = Some(runtime);
1753 }
1754
1755 pub fn subagent_limits(&self) -> crate::config::SubagentLimits {
1757 self.subagent_limits
1758 }
1759
1760 pub fn set_delegation_parent(&mut self, parent: Option<String>) {
1761 self.delegation_parent = parent;
1762 }
1763
1764 pub fn orchestration(&self) -> Option<&Arc<crate::orchestration::OrchestrationRuntime>> {
1765 self.orchestration.as_ref()
1766 }
1767
1768 pub fn grant_worker_model(&self, model: &str) -> std::result::Result<(), String> {
1773 self.orchestration
1774 .as_ref()
1775 .ok_or_else(|| "delegation policy unavailable".to_string())?
1776 .grant_worker_model(model)
1777 }
1778
1779 pub fn set_model(&mut self, model: String) -> Option<ReasoningClamp> {
1783 self.try_set_model(model).ok().flatten()
1784 }
1785
1786 pub fn try_set_model(
1793 &mut self,
1794 model: String,
1795 ) -> std::result::Result<Option<ReasoningClamp>, String> {
1796 let trimmed = model.trim();
1801 let cleaned = trimmed
1802 .split_once(char::is_whitespace)
1803 .and_then(|(_, rest)| {
1804 let rest = rest.trim_start();
1805 let (latency, candidate) = rest.split_once(char::is_whitespace)?;
1806 let millis = latency.strip_suffix("ms")?;
1807 if !millis.is_empty() && millis.chars().all(|c| c.is_ascii_digit()) {
1808 let candidate = candidate.trim();
1809 if !candidate.is_empty() && !candidate.chars().any(char::is_whitespace) {
1810 return Some(candidate);
1811 }
1812 }
1813 None
1814 })
1815 .unwrap_or(trimmed);
1816 if self.prompt_reload_source.is_none()
1822 && self
1823 .orchestration
1824 .as_ref()
1825 .is_some_and(|current| !current.unreconciled_runtime_handles().is_empty())
1826 {
1827 return Err("model change blocked: workers require collection/reconciliation".into());
1828 }
1829 let orchestration_replacement = if self.prompt_reload_source.is_none()
1830 && self.orchestration.is_some()
1831 {
1832 let foreground = crate::orchestration::canonical_foreground_identity(cleaned)
1833 .map_err(|_| "model change blocked: unresolved foreground model".to_string())?;
1834 let replacement = crate::orchestration::OrchestrationRuntime::from_limits(
1835 foreground,
1836 self.subagent_limits,
1837 )
1838 .map_err(|_| "model change blocked: trusted worker catalog unavailable".to_string())?;
1839 for trusted in &self.trusted_worker_models {
1840 replacement
1841 .grant_worker_model(trusted.as_str())
1842 .map_err(|_| {
1843 "model change blocked: configured worker trust invalid".to_string()
1844 })?;
1845 }
1846 Some(Arc::new(replacement))
1847 } else {
1848 None
1849 };
1850 self.model = cleaned.to_owned();
1851 if let Some(replacement) = orchestration_replacement {
1855 self.orchestration = Some(replacement);
1856 }
1857 if !self.explicit_reasoning {
1863 if let Some(level) =
1864 crate::runtime::openai::catalog::validation::default_level_for_model(cleaned)
1865 {
1866 self.set_reasoning_level(level);
1867 }
1868 }
1869 Ok(self.clamp_reasoning_to_model())
1874 }
1875
1876 fn clamp_reasoning_to_model(&mut self) -> Option<ReasoningClamp> {
1882 use crate::runtime::openai::catalog::validation::{
1883 default_level_for_model, thinking_options_for_model, validate_reasoning_mutation,
1884 };
1885 let model = self.model.as_str();
1886 let from = self.reasoning_level();
1887 if validate_reasoning_mutation(model, from).is_ok() {
1888 return None;
1889 }
1890 let to = default_level_for_model(model)
1891 .filter(|level| validate_reasoning_mutation(model, *level).is_ok())
1892 .or_else(|| {
1893 thinking_options_for_model(model)
1894 .iter()
1895 .filter_map(|opt| agent_core::reasoning::ReasoningLevel::parse(opt))
1896 .find(|level| validate_reasoning_mutation(model, *level).is_ok())
1897 })?;
1898 let explicit = self.explicit_reasoning;
1899 self.set_reasoning_level(to);
1900 self.explicit_reasoning = explicit;
1901 Some(ReasoningClamp { from, to })
1902 }
1903
1904 pub fn set_tools(&mut self, tools: ToolRegistry) {
1905 self.tools = Arc::new(RwLock::new(tools));
1906 }
1907
1908 pub fn subagent_registry(&self) -> &Arc<Mutex<crate::runtime::subagent::SubagentRegistry>> {
1909 &self.subagent_registry
1910 }
1911
1912 pub fn event_queue(&self) -> &Arc<crate::events::EventQueue> {
1913 &self.event_queue
1914 }
1915
1916 pub fn set_event_queue(&mut self, queue: Arc<crate::events::EventQueue>) {
1920 self.event_queue = queue;
1921 }
1922
1923 pub fn hook_bus(&self) -> &Arc<crate::extensions::hooks::HookBus> {
1925 &self.hook_bus
1926 }
1927
1928 pub(crate) fn set_hook_bus(&mut self, bus: Arc<crate::extensions::hooks::HookBus>) {
1933 self.hook_bus = bus;
1934 }
1935
1936 pub fn session_manager(&self) -> &Arc<crate::tools::shell::SessionManager> {
1939 &self.session_manager
1940 }
1941
1942 pub fn host_tool_session_id(&self) -> &crate::tools::activation::SessionId {
1946 &self.host_tool_session
1947 }
1948
1949 pub fn set_session_id(&mut self, id: Option<String>) {
1954 self.session_id = id;
1955 }
1956
1957 pub fn begin_conversation(&mut self, id: Option<String>) {
1963 self.session_allow_all.revoke();
1964 self.session_allow_all = std::sync::Arc::new(crate::tools::SessionConfirmLatch::new());
1965 *self
1968 .last_msg_id
1969 .lock()
1970 .unwrap_or_else(std::sync::PoisonError::into_inner) = None;
1971 self.set_session_id(id);
1972 }
1973
1974 pub fn session_id(&self) -> Option<&str> {
1975 self.session_id.as_deref()
1976 }
1977
1978 pub fn set_cwd(&mut self, cwd: Option<PathBuf>) {
1981 self.cwd = cwd;
1982 }
1983
1984 pub fn cwd(&self) -> Option<&Path> {
1985 self.cwd.as_deref()
1986 }
1987
1988 pub fn set_env(&mut self, env: Option<crate::session::types::SessionEnv>) {
1991 self.env = env;
1992 }
1993
1994 pub fn set_env_stripped(&mut self, stripped: Vec<String>) {
1995 self.env_stripped = stripped;
1996 }
1997
1998 pub fn env(&self) -> Option<&crate::session::types::SessionEnv> {
1999 self.env.as_ref()
2000 }
2001
2002 pub fn env_stripped(&self) -> &[String] {
2003 &self.env_stripped
2004 }
2005
2006 pub fn tools_shared(&self) -> Arc<RwLock<ToolRegistry>> {
2008 Arc::clone(&self.tools)
2009 }
2010
2011 pub fn model(&self) -> &str {
2012 &self.model
2013 }
2014
2015 pub fn http_client(&self) -> &Client {
2016 &self.client
2017 }
2018
2019 pub async fn prepare_model_capabilities(&self, model: &str) {
2027 if model.starts_with("muse/") {
2028 let broker = crate::auth::broker_from_source(
2029 &self.credential_source,
2030 &self.token_cache,
2031 self.client.clone(),
2032 );
2033 crate::runtime::openai::catalog::ensure_muse_catalog_for(model, &broker).await;
2034 }
2035 }
2036
2037 pub fn progressive_tool_disclosure(&self) -> bool {
2039 self.progressive_tool_disclosure
2040 }
2041
2042 pub fn carried_tool_activations(&self) -> Vec<crate::tools::activation::CarriedActivation> {
2048 let retained = self
2049 .retained_tool_set
2050 .lock()
2051 .unwrap_or_else(std::sync::PoisonError::into_inner)
2052 .clone();
2053 if let Some(shared) = retained {
2054 return shared
2055 .read()
2056 .unwrap_or_else(std::sync::PoisonError::into_inner)
2057 .carried_activations();
2058 }
2059 self.pending_tool_restore
2060 .lock()
2061 .unwrap_or_else(std::sync::PoisonError::into_inner)
2062 .clone()
2063 .unwrap_or_default()
2064 }
2065
2066 pub fn restore_tool_activations(
2074 &self,
2075 carried: Vec<crate::tools::activation::CarriedActivation>,
2076 ) {
2077 if carried.is_empty() {
2078 return;
2079 }
2080 let mut pending = self
2081 .pending_tool_restore
2082 .lock()
2083 .unwrap_or_else(std::sync::PoisonError::into_inner);
2084 match pending.as_mut() {
2085 Some(list) => {
2086 for c in carried {
2087 if !list.iter().any(|p| p.tool_id == c.tool_id) {
2088 list.push(c);
2089 }
2090 }
2091 }
2092 None => *pending = Some(carried),
2093 }
2094 }
2095
2096 pub fn activation_confirm(&self) -> agent_core::config::ActivationConfirm {
2098 self.activation_confirm
2099 }
2100
2101 pub fn set_activation_confirm(&mut self, mode: agent_core::config::ActivationConfirm) {
2104 self.activation_confirm = mode;
2105 }
2106
2107 #[cfg(any(test, feature = "testing"))]
2111 pub fn http_client_pool_id(&self) -> usize {
2112 const _: () = assert!(std::mem::size_of::<Client>() == std::mem::size_of::<usize>());
2113 unsafe { std::mem::transmute_copy::<Client, usize>(&self.client) }
2116 }
2117 pub fn set_thinking_budget(&mut self, budget: u32) {
2118 self.thinking_budget = budget;
2119 self.named_level = Some(agent_core::reasoning::ReasoningLevel::from_legacy_budget(
2121 budget,
2122 ));
2123 self.explicit_reasoning = false;
2125 }
2126
2127 pub fn set_reasoning_level(&mut self, level: agent_core::reasoning::ReasoningLevel) {
2133 self.named_level = Some(level);
2134 if let Some(budget) = level.to_legacy_budget() {
2135 self.thinking_budget = budget;
2136 }
2137 self.explicit_reasoning = false;
2139 }
2140
2141 pub fn set_reasoning_level_explicit(&mut self, level: agent_core::reasoning::ReasoningLevel) {
2146 self.set_reasoning_level(level);
2147 self.explicit_reasoning = true;
2148 }
2149
2150 pub fn restore_session_reasoning(&mut self, thinking_level: &str) -> Option<ReasoningClamp> {
2156 if let Some(level) = agent_core::reasoning::ReasoningLevel::parse(thinking_level) {
2157 self.set_reasoning_level_explicit(level);
2158 } else if let Some(budget) = crate::models::budget_for_thinking_level(thinking_level) {
2159 self.set_thinking_budget_explicit(budget);
2160 } else {
2161 return None;
2162 }
2163 self.clamp_reasoning_to_model()
2164 }
2165
2166 pub fn set_thinking_budget_explicit(&mut self, budget: u32) {
2171 self.thinking_budget = budget;
2172 self.named_level = Some(agent_core::reasoning::ReasoningLevel::from_legacy_budget(
2173 budget,
2174 ));
2175 self.explicit_reasoning = true;
2176 }
2177
2178 pub fn is_reasoning_explicit(&self) -> bool {
2180 self.explicit_reasoning
2181 }
2182
2183 pub fn thinking_budget_raw(&self) -> u32 {
2185 self.thinking_budget
2186 }
2187
2188 pub fn set_reasoning_level_checked(
2193 &mut self,
2194 level: agent_core::reasoning::ReasoningLevel,
2195 ) -> std::result::Result<(), String> {
2196 crate::runtime::openai::catalog::validation::validate_reasoning_mutation(
2197 &self.model,
2198 level,
2199 )?;
2200 if self.model.starts_with("anthropic/")
2201 && level == agent_core::reasoning::ReasoningLevel::UltraCode
2202 {
2203 if self.codex_request_role()
2204 != crate::runtime::openai::catalog::ExecutionRole::Foreground
2205 {
2206 return Err("ultracode_requires_foreground".into());
2207 }
2208 if self.effective_prompt.is_some() {
2209 return Err("typed_prompt_manifest_blocks_required_doctrine".into());
2210 }
2211 if self.orchestration.is_none() {
2212 return Err("ultracode_requires_orchestration".into());
2213 }
2214 let tools = self.tools.try_read().map_err(|_| {
2215 "ultracode prerequisite state is busy; refusing mutation".to_string()
2216 })?;
2217 let required = ["subagent_start", "subagent_status", "subagent_collect"];
2218 if !required.iter().all(|name| {
2219 tools
2220 .get(name)
2221 .is_some_and(|tool| tool.extension_id().is_none())
2222 }) {
2223 return Err("ultracode_requires_lifecycle_tools".into());
2224 }
2225 }
2226 self.set_reasoning_level_explicit(level);
2227 Ok(())
2228 }
2229
2230 pub fn set_thinking_budget_checked(&mut self, budget: u32) -> std::result::Result<(), String> {
2237 let level = agent_core::reasoning::ReasoningLevel::from_legacy_budget(budget);
2238 crate::runtime::openai::catalog::validation::validate_reasoning_mutation(
2239 &self.model,
2240 level,
2241 )?;
2242 self.set_thinking_budget_explicit(budget);
2243 Ok(())
2244 }
2245
2246 pub fn reasoning_level(&self) -> agent_core::reasoning::ReasoningLevel {
2247 self.named_level.unwrap_or_else(|| {
2248 agent_core::reasoning::ReasoningLevel::from_legacy_budget(self.thinking_budget)
2249 })
2250 }
2251
2252 pub(crate) fn set_codex_request_role(
2253 &mut self,
2254 role: crate::runtime::openai::catalog::CodexRequestRole,
2255 ) {
2256 self.codex_request_role = role;
2257 }
2258
2259 pub(crate) fn codex_request_role(&self) -> crate::runtime::openai::catalog::CodexRequestRole {
2260 self.codex_request_role
2261 }
2262
2263 pub(crate) fn codex_delegation_plan(
2264 model: &str,
2265 level: agent_core::reasoning::ReasoningLevel,
2266 role: crate::runtime::openai::catalog::CodexRequestRole,
2267 ) -> Option<crate::runtime::openai::catalog::CodexExecutionPlan> {
2268 use crate::runtime::openai::catalog::{plan_codex_execution, CodexRequestRole};
2269 if role != CodexRequestRole::Foreground
2270 || level != agent_core::reasoning::ReasoningLevel::Ultra
2271 || !model.starts_with("openai-codex/")
2272 {
2273 return None;
2274 }
2275 plan_codex_execution(model, level, role, None).ok()
2276 }
2277
2278 async fn authorized_anthropic_plan(
2279 &self,
2280 ) -> Result<Option<crate::runtime::openai::catalog::AnthropicExecutionPlan>> {
2281 if !self.model.starts_with("anthropic/") {
2282 return Ok(None);
2283 }
2284 let tools = self.tools.read().await;
2285 let builtin = |name: &str| {
2286 tools
2287 .get(name)
2288 .is_some_and(|tool| tool.extension_id().is_none())
2289 };
2290 let readiness = self
2291 .orchestration
2292 .as_ref()
2293 .and_then(|runtime| runtime.ultracode_readiness(&self.model).ok());
2294 let prerequisites = crate::runtime::openai::catalog::AnthropicPlanPrerequisites {
2295 orchestration_policy: self.orchestration.is_some(),
2296 foreground_worker_authorized: readiness.is_some(),
2297 concurrent_limit: readiness.map_or(0, |limits| limits.0),
2298 total_limit: readiness.map_or(0, |limits| limits.1),
2299 lifecycle_start: builtin("subagent_start"),
2300 lifecycle_status: builtin("subagent_status"),
2301 lifecycle_steer: builtin("subagent_steer"),
2302 lifecycle_collect: builtin("subagent_collect"),
2303 lifecycle_resume: builtin("subagent_resume"),
2304 };
2305 crate::runtime::openai::catalog::plan_anthropic_execution(
2306 &self.model,
2307 self.reasoning_level(),
2308 self.codex_request_role(),
2309 prerequisites,
2310 crate::runtime::openai::catalog::capability_cache::get(&self.model).and_then(|entry| {
2311 match entry.reasoning {
2312 crate::runtime::openai::catalog::ReasoningSupport::AnthropicAdaptive {
2313 adaptive,
2314 } => Some(adaptive),
2315 crate::runtime::openai::catalog::ReasoningSupport::None => Some(false),
2316 _ => None,
2317 }
2318 }),
2319 )
2320 .map(Some)
2321 .map_err(|error| RuntimeError::Config(error.to_string()))
2322 }
2323
2324 async fn validate_request_preflight(&self) -> Result<()> {
2327 self.validate_request_preflight_for(&self.model, self.codex_request_role)
2328 .await
2329 }
2330
2331 pub(crate) async fn validate_session_driver_preflight(&self) -> Result<()> {
2334 self.validate_request_preflight().await
2335 }
2336
2337 async fn validate_request_preflight_for(
2338 &self,
2339 model: &str,
2340 role: crate::runtime::openai::catalog::CodexRequestRole,
2341 ) -> Result<()> {
2342 if self
2346 .memory_backend_reconfigure_denied
2347 .load(std::sync::atomic::Ordering::SeqCst)
2348 {
2349 return Err(RuntimeError::Config(
2350 "memory backend cannot be reconfigured at runtime (including live legacy extension processes); restart required; request denied".into(),
2351 ));
2352 }
2353 let level = self.reasoning_level();
2354 if model.starts_with("anthropic/")
2355 && level == agent_core::reasoning::ReasoningLevel::UltraCode
2356 && self.effective_prompt.is_some()
2357 {
2358 return Err(RuntimeError::Config(
2359 "Anthropic execution plan rejected: typed_prompt_manifest_blocks_required_doctrine"
2360 .into(),
2361 ));
2362 }
2363 if model.starts_with("anthropic/")
2364 && matches!(
2365 level,
2366 agent_core::reasoning::ReasoningLevel::Max
2367 | agent_core::reasoning::ReasoningLevel::UltraCode
2368 | agent_core::reasoning::ReasoningLevel::Ultra
2369 )
2370 {
2371 let tools = self.tools.read().await;
2372 let builtin = |name: &str| {
2373 tools
2374 .get(name)
2375 .is_some_and(|tool| tool.extension_id().is_none())
2376 };
2377 let readiness = self
2378 .orchestration
2379 .as_ref()
2380 .and_then(|runtime| runtime.ultracode_readiness(model).ok());
2381 let prerequisites = crate::runtime::openai::catalog::AnthropicPlanPrerequisites {
2382 orchestration_policy: self.orchestration.is_some(),
2383 foreground_worker_authorized: readiness.is_some(),
2384 concurrent_limit: readiness.map_or(0, |limits| limits.0),
2385 total_limit: readiness.map_or(0, |limits| limits.1),
2386 lifecycle_start: builtin("subagent_start"),
2387 lifecycle_status: builtin("subagent_status"),
2388 lifecycle_steer: builtin("subagent_steer"),
2389 lifecycle_collect: builtin("subagent_collect"),
2390 lifecycle_resume: builtin("subagent_resume"),
2391 };
2392 drop(tools);
2393 let result = crate::runtime::openai::catalog::plan_anthropic_execution(
2394 model,
2395 level,
2396 role,
2397 prerequisites,
2398 crate::runtime::openai::catalog::capability_cache::get(model).and_then(|entry| {
2399 match entry.reasoning {
2400 crate::runtime::openai::catalog::ReasoningSupport::AnthropicAdaptive {
2401 adaptive,
2402 } => Some(adaptive),
2403 crate::runtime::openai::catalog::ReasoningSupport::None => Some(false),
2404 _ => None,
2405 }
2406 }),
2407 );
2408 match result {
2409 Ok(plan) => {
2410 tracing::debug!(event = "anthropic_mode_plan", qualified_model = %model, requested_level = %level, runtime_role = role.as_str(), execution_mode = ?plan.mode, wire_effort = plan.wire_effort.map_or("omitted", |effort| effort.as_str()), workflow = ?plan.workflow, credentials_attempted = false, network_attempted = false);
2411 return Ok(());
2412 }
2413 Err(error) => {
2414 tracing::debug!(event = "anthropic_mode_plan", qualified_model = %model, requested_level = %level, runtime_role = role.as_str(), decision = "deny", deny_code = error.code().as_str(), credentials_attempted = false, network_attempted = false);
2415 return Err(RuntimeError::Config(error.to_string()));
2416 }
2417 }
2418 }
2419 if model.starts_with("openai-codex/") {
2420 let plan =
2421 crate::runtime::openai::catalog::plan_codex_execution(model, level, role, None)
2422 .map_err(|error| {
2423 tracing::debug!(
2424 event = "codex_mode_plan",
2425 qualified_model = %model,
2426 requested_level = %level,
2427 runtime_role = role.as_str(),
2428 decision = "deny",
2429 deny_code = error.code().as_str(),
2430 network_attempted = false,
2431 "Codex request preflight denied"
2432 );
2433 RuntimeError::Config(error.to_string())
2434 })?;
2435
2436 if plan.automatic_delegation() {
2437 if self.orchestration.is_none() {
2438 tracing::debug!(
2439 event = "codex_mode_plan",
2440 qualified_model = %model,
2441 requested_level = %level,
2442 runtime_role = role.as_str(),
2443 decision = "deny",
2444 deny_code = "ultra_requires_orchestration",
2445 network_attempted = false,
2446 "Codex request preflight denied"
2447 );
2448 return Err(RuntimeError::Config(
2449 "Ultra requires an installed orchestration policy".to_string(),
2450 ));
2451 }
2452
2453 let tools = self.tools.read().await;
2454 let required = ["subagent_start", "subagent_status", "subagent_collect"];
2455 let tools_ready = required.iter().all(|name| {
2456 tools
2457 .get(name)
2458 .is_some_and(|tool| tool.extension_id().is_none())
2459 });
2460 if !tools_ready {
2461 tracing::debug!(
2462 event = "codex_mode_plan",
2463 qualified_model = %model,
2464 requested_level = %level,
2465 runtime_role = role.as_str(),
2466 decision = "deny",
2467 deny_code = "ultra_requires_subagent_tools",
2468 network_attempted = false,
2469 "Codex request preflight denied"
2470 );
2471 return Err(RuntimeError::Config(
2472 "Ultra requires built-in subagent_start, subagent_status, and subagent_collect tools"
2473 .to_string(),
2474 ));
2475 }
2476 }
2477 return Ok(());
2478 }
2479
2480 crate::runtime::openai::catalog::validation::validate_reasoning_mutation(model, level)
2481 .map_err(|error| {
2482 tracing::debug!(
2483 event = "codex_mode_plan",
2484 qualified_model = %model,
2485 requested_level = %level,
2486 runtime_role = role.as_str(),
2487 decision = "deny",
2488 deny_code = "unsupported_reasoning_level",
2489 network_attempted = false,
2490 "request preflight denied"
2491 );
2492 RuntimeError::Config(error)
2493 })
2494 }
2495
2496 pub fn set_compaction_model(&mut self, model: Option<String>) {
2497 self.compaction_model = model;
2498 }
2499
2500 pub fn set_compaction_mode(&mut self, mode: agent_core::compaction::CompactionMode) {
2502 self.compaction_mode = mode;
2503 }
2504
2505 pub fn set_compaction_exclusions(
2507 &mut self,
2508 exclude: Vec<agent_core::compaction::ContentClass>,
2509 ) {
2510 self.compaction_exclusions = exclude;
2511 }
2512
2513 pub fn remote_summarization_attempts(&self) -> u64 {
2516 self.remote_summarization_attempts
2517 .load(std::sync::atomic::Ordering::SeqCst)
2518 }
2519
2520 pub fn compaction_policy(&self) -> crate::runtime::compaction::DisclosurePolicy {
2523 crate::runtime::compaction::DisclosurePolicy {
2524 mode: self.compaction_mode,
2525 exclude: self.compaction_exclusions.clone(),
2526 }
2527 }
2528
2529 pub fn set_context_window(&mut self, window: Option<u64>) {
2530 self.context_window_override = window;
2531 }
2532
2533 pub fn context_window_override(&self) -> Option<u64> {
2535 self.context_window_override
2536 }
2537
2538 pub fn compaction_model(&self) -> &str {
2541 self.compaction_model
2542 .as_deref()
2543 .unwrap_or("claude-sonnet-4-6")
2544 }
2545
2546 pub fn context_window(&self) -> u64 {
2547 self.context_window_override
2548 .unwrap_or_else(|| crate::models::context_window_for_model(&self.model))
2549 }
2550
2551 pub async fn assess_context(
2566 &self,
2567 messages: &[crate::SharedMessage],
2568 ) -> context::ContextAssessment {
2569 let system = self.effective_system_prompt().await;
2570 let schema = self.tools.read().await.tools_schema();
2571 let memory_rendered: Option<String> = self
2576 .pending_memory_segment
2577 .lock()
2578 .unwrap_or_else(std::sync::PoisonError::into_inner)
2579 .as_ref()
2580 .map(|segment| segment.rendered_text().to_string());
2581 let memory_contents: Vec<&str> = memory_rendered.as_deref().into_iter().collect();
2582 let anchor = self
2583 .continuation
2584 .lock()
2585 .unwrap_or_else(std::sync::PoisonError::into_inner)
2586 .usage_anchor_for(&self.model)
2587 .cloned();
2588 context::assess(&context::ContextBudgetInputs {
2589 model: &self.model,
2590 provider_window: self.context_window(),
2591 system_prompt: system.as_deref(),
2592 tools_schema: &schema,
2593 messages,
2594 skill_contents: &[],
2595 memory_contents: &memory_contents,
2596 thinking_budget_tokens: self.thinking_budget as u64,
2597 next_tool_result_bytes: self.max_tool_output as u64,
2598 max_output_tokens: HelperMethods::max_tokens_for_model(&self.model),
2599 })
2600 .anchored(anchor.as_ref(), messages)
2601 }
2602
2603 #[allow(dead_code)] pub(crate) fn hold_memory_contribution(
2621 &self,
2622 contribution: memory_context::MemoryContextContribution,
2623 ) -> std::result::Result<(), memory_context::MemoryContextError> {
2624 let budget = memory_context::memory_budget_tokens(self.context_window())
2625 .ok_or(memory_context::MemoryContextError::RecallBudgetBelowMinimum)?;
2626 memory_context::validate_contribution(
2630 &contribution,
2631 &self.memory_context_project_id(),
2632 budget,
2633 &memory_context::DisclosureGrantSet::model_visible_only(),
2634 )?;
2635 *self
2636 .pending_memory_segment
2637 .lock()
2638 .unwrap_or_else(std::sync::PoisonError::into_inner) =
2639 Some(memory_context::ContextSegment::Memory(contribution));
2640 Ok(())
2641 }
2642
2643 pub(crate) fn clear_memory_contribution(&self) {
2647 *self
2648 .pending_memory_segment
2649 .lock()
2650 .unwrap_or_else(std::sync::PoisonError::into_inner) = None;
2651 }
2652
2653 pub fn memory_recall_why(&self) -> Option<memory_context::RecallTurnMetadata> {
2659 self.retained_recall_turn
2660 .lock()
2661 .unwrap_or_else(std::sync::PoisonError::into_inner)
2662 .as_ref()
2663 .map(|retained| retained.why.clone())
2664 }
2665
2666 async fn apply_turn_memory_recall(&self, messages: &mut Vec<crate::SharedMessage>) {
2684 if (self.memory_backend.exclusive() && !self.memory_backend.is_axel())
2685 || self
2686 .memory_backend_reconfigure_denied
2687 .load(std::sync::atomic::Ordering::SeqCst)
2688 {
2689 self.clear_memory_contribution();
2690 tracing::debug!("memory recall unavailable for the selected host backend");
2691 return;
2692 }
2693 let extension_runtime = self.extension_runtime.clone();
2694 let binding = self.memory_backend.clone();
2695 let session = self.host_tool_session.clone();
2696 let outcome = memory_context::resolve_turn_recall(
2697 &self.memory_context_state,
2698 &self.retained_recall_turn,
2699 &self.memory_context_project_id(),
2700 self.context_window(),
2701 messages,
2702 memory_context::RECALL_HARD_TIMEOUT,
2703 move |lease: memory_context::MemoryContextLease,
2704 request: memory_context::RecallRequest| async move {
2705 if binding.is_axel() {
2706 return axel_context::recall(binding, lease, request).await;
2707 }
2708 let Some(manager) = extension_runtime else {
2709 return Err(memory_context::RecallCallError::ProviderUnavailable);
2710 };
2711 let mut parts = lease.provider_id.as_str().splitn(3, ':');
2712 let (Some("extension"), Some(plugin), Some(local)) =
2713 (parts.next(), parts.next(), parts.next())
2714 else {
2715 return Err(memory_context::RecallCallError::ProviderUnavailable);
2716 };
2717 let Some(digest) =
2718 manager.declared_tool_digest(plugin, memory_context::MEMORY_RECALL_TOOL_NAME)
2719 else {
2720 return Err(memory_context::RecallCallError::ProviderUnavailable);
2721 };
2722 let mut response =
2723 crate::extensions::lease::ExtensionLeaseCapability::new(session, manager)
2724 .call_exact(
2725 plugin,
2726 memory_context::MEMORY_RECALL_TOOL_NAME,
2727 &digest,
2728 memory_context::recall_request_wire(&request),
2729 )
2730 .await
2731 .map_err(|_| memory_context::RecallCallError::CallFailed)?;
2732 if response["provider_id"] != local
2735 && response["provider_id"] != lease.provider_id.as_str()
2736 {
2737 return Err(memory_context::RecallCallError::CallFailed);
2738 }
2739 response["provider_id"] = json!(lease.provider_id.as_str());
2740 Ok(response)
2741 },
2742 )
2743 .await;
2744 match outcome {
2745 memory_context::TurnRecallOutcome::Injected
2746 | memory_context::TurnRecallOutcome::ReusedRetained => {
2747 let contribution = self
2748 .retained_recall_turn
2749 .lock()
2750 .unwrap_or_else(std::sync::PoisonError::into_inner)
2751 .as_ref()
2752 .map(|retained| retained.contribution.clone());
2753 if let Some(contribution) = contribution {
2754 *self
2755 .pending_memory_segment
2756 .lock()
2757 .unwrap_or_else(std::sync::PoisonError::into_inner) =
2758 Some(memory_context::ContextSegment::Memory(contribution));
2759 }
2760 }
2761 memory_context::TurnRecallOutcome::NotEligible => {
2762 self.clear_memory_contribution();
2763 }
2764 memory_context::TurnRecallOutcome::SkippedOpen(reason) => {
2765 self.clear_memory_contribution();
2766 tracing::debug!(
2769 reason = reason.as_str(),
2770 "memory recall skipped; turn proceeds without memory"
2771 );
2772 }
2773 }
2774 }
2775
2776 #[cfg(any(test, feature = "testing"))]
2786 pub async fn apply_turn_memory_recall_for_harness(
2787 &self,
2788 messages: &mut Vec<crate::SharedMessage>,
2789 ) {
2790 self.apply_turn_memory_recall(messages).await;
2791 }
2792
2793 #[cfg(any(test, feature = "testing"))]
2797 pub fn capture_completed_turn_for_harness(
2798 &self,
2799 messages: Vec<crate::SharedMessage>,
2800 turn_ordinal: u64,
2801 ) -> std::result::Result<(), &'static str> {
2802 let started_at = std::time::SystemTime::now();
2803 let lease = self
2804 .memory_context_lock()
2805 .capture_lease_at(started_at)
2806 .ok_or("capture lease unavailable")?;
2807 let provider = self
2808 .extension_capture_provider(&lease)
2809 .ok_or("capture provider unavailable")?;
2810 let mut history = terminal_capture_history(&lease, &messages, started_at);
2811 history.turn_ordinal = turn_ordinal;
2812 memory_capture_worker()
2813 .submit_terminal(
2814 &lease,
2815 history,
2816 memory_context::RetentionClass::Standard,
2817 provider,
2818 )
2819 .map_err(|_| "capture build failed")?;
2820 Ok(())
2821 }
2822
2823 #[cfg(any(test, feature = "testing"))]
2826 pub fn capture_compaction_summary_for_harness(
2827 &self,
2828 first_turn_ordinal: u64,
2829 last_turn_ordinal: u64,
2830 source_digest: &str,
2831 source_message_count: usize,
2832 summary_text: &str,
2833 ) -> std::result::Result<(), &'static str> {
2834 let summarized_at = std::time::SystemTime::now();
2835 let lease = self
2836 .memory_context_lock()
2837 .capture_lease_at(summarized_at)
2838 .ok_or("capture lease unavailable")?;
2839 let source_digest = chat_capture::MessageRangeDigest::from_hex(source_digest)
2840 .ok_or("invalid source digest")?;
2841 let provider = self
2842 .extension_capture_provider(&lease)
2843 .ok_or("capture provider unavailable")?;
2844 let source = chat_capture::CompactionSource {
2845 schema: chat_capture::CompactionSchemaVersion::V1,
2846 project_id: lease.project_id.clone(),
2847 source_session_id: lease.session_id.clone(),
2848 first_turn_ordinal,
2849 last_turn_ordinal,
2850 source_digest,
2851 summary_origin: chat_capture::CompactionSummaryOrigin::LocalOnly,
2852 prompt_stack_digest: chat_capture::PromptStackDigest::from_bytes([7; 32]),
2853 redaction: chat_capture::RedactionPolicy::HostRedacted,
2854 content_classes: vec![agent_core::compaction::ContentClass::UserText],
2855 summarized_at,
2856 };
2857 memory_capture_worker()
2858 .submit_summary(
2859 &lease,
2860 source,
2861 source_message_count,
2862 summary_text,
2863 agent_core::compaction::RedactionPolicy::PolicyExclusions,
2864 memory_context::RetentionClass::Standard,
2865 provider,
2866 )
2867 .map_err(|_| "summary capture build failed")?;
2868 Ok(())
2869 }
2870
2871 #[cfg(any(test, feature = "testing"))]
2874 pub fn cancel_memory_forwarding_for_harness(&self) {
2875 self.clear_memory_contribution();
2876 self.memory_context_disable();
2877 }
2878
2879 #[cfg(any(test, feature = "testing"))]
2881 pub fn extension_lease_count_for_harness(&self) -> usize {
2882 self.extension_runtime
2883 .as_ref()
2884 .map_or(0, |manager| manager.lease_count())
2885 }
2886
2887 pub(crate) fn inherit_memory_backend(&mut self, binding: crate::memory_backend::MemoryBinding) {
2891 self.memory_backend = binding.fork_for_worker();
2892 }
2893
2894 pub fn apply_memory_backend_config(&mut self, config: &crate::config::MemoryBackendConfig) {
2896 if let Some(selected) = &self.memory_backend_config {
2897 if selected != config {
2898 self.memory_backend_reconfigure_denied
2899 .store(true, std::sync::atomic::Ordering::SeqCst);
2900 tracing::warn!("memory backend change denied: live extension processes cannot be reconfigured at runtime; original binding retained; restart required");
2901 self.memory_context_disable();
2902 self.clear_memory_contribution();
2903 }
2904 return;
2905 }
2906 if config.kind != crate::config::MemoryBackendKind::Legacy {
2907 self.memory_context_disable();
2908 }
2909 self.memory_backend = crate::memory_backend::MemoryBinding::from_config_with_cwd(
2913 config,
2914 self.cwd.clone().or_else(|| std::env::current_dir().ok()),
2915 );
2916 self.memory_backend_config = Some(config.clone());
2917 }
2918
2919 #[cfg(test)]
2920 #[allow(dead_code)] pub(crate) fn memory_backend_for_test(&self) -> crate::memory_backend::MemoryBinding {
2922 self.memory_backend.clone()
2923 }
2924
2925 pub fn memory_backend_is_axel(&self) -> bool {
2930 self.memory_backend.is_axel()
2931 }
2932
2933 pub fn memory_backend_exclusive(&self) -> bool {
2934 self.memory_backend.exclusive()
2935 }
2936
2937 fn memory_context_project_id(&self) -> memory_context::ProjectId {
2938 if self.memory_backend.exclusive() {
2939 self.memory_backend
2940 .scope()
2941 .ok()
2942 .and_then(|scope| memory_context::ProjectId::parse(scope.key()).ok())
2943 .unwrap_or_else(|| {
2944 memory_context::ProjectId::parse("project-unresolved").expect("static id")
2945 })
2946 } else {
2947 memory_project_id()
2948 }
2949 }
2950
2951 pub fn memory_tool_capability(&self) -> Option<memory_context::MemoryContextCapability> {
2952 if !self.memory_backend.exclusive() {
2953 return None;
2954 }
2955 let runtime = self.clone();
2956 Some(memory_context::MemoryContextCapability::control_only(
2957 self.memory_context_state.clone(),
2958 self.memory_context_project_id(),
2959 memory_context::ContextProviderId::parse(axel_context::PROVIDER_ID).expect("static id"),
2960 Arc::new(move || runtime.memory_context_disable()),
2961 ))
2962 }
2963
2964 pub fn apply_config(&mut self, config: &crate::config::SynapsConfig) {
2968 self.apply_config_inner(config, true);
2969 }
2970
2971 pub(crate) fn apply_config_keep_tools(&mut self, config: &crate::config::SynapsConfig) {
2978 self.apply_config_inner(config, false);
2979 }
2980
2981 fn apply_config_inner(&mut self, config: &crate::config::SynapsConfig, disable_tools: bool) {
2982 self.apply_memory_backend_config(&config.memory_backend);
2984 if (self.memory_backend.exclusive() && !self.memory_backend.is_axel())
2985 || self
2986 .memory_backend_reconfigure_denied
2987 .load(std::sync::atomic::Ordering::SeqCst)
2988 {
2989 self.memory_context_disable();
2991 self.clear_memory_contribution();
2992 *self
2993 .retained_recall_turn
2994 .lock()
2995 .unwrap_or_else(std::sync::PoisonError::into_inner) = None;
2996 }
2997 if let Some(ref model) = config.model {
2998 self.set_model(model.clone());
2999 }
3000 if let Some(level) = config.thinking_level {
3003 self.set_reasoning_level_explicit(level);
3004 } else if let Some(budget) = config.thinking_budget {
3005 self.set_thinking_budget_explicit(budget);
3006 }
3007 if let Some(clamp) = self.clamp_reasoning_to_model() {
3010 tracing::warn!(
3011 from = clamp.from.as_str(),
3012 to = clamp.to.as_str(),
3013 model = %self.model,
3014 "configured reasoning level not supported by model; clamped"
3015 );
3016 }
3017 self.context_window_override = config.context_window;
3018 self.continuation
3019 .lock()
3020 .unwrap_or_else(std::sync::PoisonError::into_inner)
3021 .config = config.context_management;
3022 self.compaction_model = config.compaction_model.clone();
3023 self.compaction_mode = config.compaction_mode;
3024 self.compaction_exclusions = config.compaction_exclude.clone();
3025 self.max_tool_output = config.max_tool_output;
3026 self.tool_spill_policy = tool_spill::ToolSpillPolicy::from_config(config);
3027 self.read_dedupe = config.read_dedupe;
3028 self.thinking_display = config.thinking_display;
3029 self.tool_keepwarm = config.cache_keepwarm_tools;
3030 self.bash_timeout = config.bash_timeout;
3031 self.bash_max_timeout = config.bash_max_timeout;
3032 self.subagent_timeout = config.subagent_timeout;
3033 self.api_retries = config.api_retries;
3034 self.refusal_retries = config.refusal_retries;
3035 self.telemetry_level =
3036 crate::runtime::telemetry::TelemetryLevel::from_str_key(&config.telemetry);
3037 self.telemetry_log_generations = config.telemetry_log_generations;
3038 self.sync_observability();
3039 self.cache_diagnostics = config.cache_diagnostics;
3040 self.cache_ttl = config.cache_ttl;
3041 self.request_compression = request_encoding::RequestCompression::from_config(
3042 config.request_compression,
3043 config.request_compression_level,
3044 );
3045 self.progressive_tool_disclosure = config.progressive_tool_disclosure;
3046 self.deferred_tool_loading = config.tools_deferred_loading;
3047 self.activation_confirm = config.tools_activation_confirm;
3048 tracing::info!(
3049 mode = config.tools_activation_confirm.as_str(),
3050 auto_approve_confirms = config.server.auto_approve_confirms,
3051 "tools.activation_confirm: model-initiated activate_tools policy"
3052 );
3053 self.trusted_worker_models = config
3054 .favorite_models
3055 .iter()
3056 .filter_map(|model| {
3057 let canonical = canonical_trusted_worker_model(model);
3058 match agent_core::prompt::QualifiedModelId::parse(canonical) {
3059 Ok(model) => Some(model),
3060 Err(_) => {
3061 tracing::warn!(
3062 model = model.as_str(),
3063 "ignoring invalid favorite worker-model identity"
3064 );
3065 None
3066 }
3067 }
3068 })
3069 .collect();
3070 self.trusted_worker_models.sort();
3071 self.trusted_worker_models.dedup();
3072 self.subagent_limits = config.subagent;
3073 if let Some(orchestration) = &self.orchestration {
3074 for model in &self.trusted_worker_models {
3075 if let Err(error) = orchestration.grant_worker_model(model.as_str()) {
3076 tracing::warn!(
3077 model = model.as_str(),
3078 error = %error,
3079 "failed to apply configured worker-model trust"
3080 );
3081 }
3082 }
3083 }
3084 self.apply_auth_config(config);
3085
3086 if disable_tools && !config.disabled_tools.is_empty() {
3090 if let Ok(mut reg) = self.tools.try_write() {
3091 reg.disable(&config.disabled_tools);
3092 }
3093 }
3094 }
3095
3096 pub fn apply_auth_config(&mut self, config: &crate::config::SynapsConfig) {
3104 let new_source = config.auth.credential_source();
3105 if new_source != self.credential_source {
3106 self.token_cache.invalidate("anthropic");
3107 }
3108 self.credential_source = new_source;
3109 if self.credential_source.is_remote() {
3110 if let Ok(mut auth) = self.auth.try_write() {
3111 AuthMethods::scrub_for_remote(&mut auth);
3112 }
3113 }
3114 if let Ok(mut auth) = self.auth.try_write() {
3121 if auth.auth_type == "oauth" && !auth.auth_token.is_empty() {
3122 let selector = config
3123 .auth
3124 .account_selector(crate::auth::OAuthProviderId::Anthropic);
3125 let check = crate::runtime::auth::check_binding(
3126 &self.credential_source,
3127 &selector,
3128 auth.bound_credential.as_deref(),
3129 crate::runtime::auth::RefreshPoint::TurnStart,
3130 );
3131 if check != crate::runtime::auth::BindingCheck::Matches {
3132 AuthMethods::scrub_for_account_switch(&mut auth);
3133 self.token_cache.invalidate("anthropic");
3134 }
3135 }
3136 }
3137 crate::auth::set_global_broker(crate::auth::broker_from_source(
3142 &self.credential_source,
3143 &self.token_cache,
3144 self.client.clone(),
3145 ));
3146 }
3147
3148 pub fn thinking_budget(&self) -> u32 {
3149 self.thinking_budget
3150 }
3151
3152 pub fn max_tool_output(&self) -> usize {
3153 self.max_tool_output
3154 }
3155
3156 pub fn bash_timeout(&self) -> u64 {
3157 self.bash_timeout
3158 }
3159
3160 pub fn bash_max_timeout(&self) -> u64 {
3161 self.bash_max_timeout
3162 }
3163
3164 pub fn subagent_timeout(&self) -> u64 {
3165 self.subagent_timeout
3166 }
3167
3168 pub fn api_retries(&self) -> u32 {
3169 self.api_retries
3170 }
3171
3172 pub fn set_max_tool_output(&mut self, v: usize) {
3173 self.max_tool_output = v;
3174 }
3175
3176 pub fn tool_spill_policy(&self) -> tool_spill::ToolSpillPolicy {
3177 self.tool_spill_policy
3178 }
3179
3180 pub fn set_tool_spill_policy(&mut self, policy: tool_spill::ToolSpillPolicy) {
3181 self.tool_spill_policy = policy;
3182 }
3183
3184 pub fn read_dedupe(&self) -> bool {
3185 self.read_dedupe
3186 }
3187
3188 pub fn set_read_dedupe(&mut self, enabled: bool) {
3190 self.read_dedupe = enabled;
3191 }
3192
3193 pub fn thinking_display(&self) -> crate::config::ThinkingDisplay {
3195 self.thinking_display
3196 }
3197
3198 pub fn set_thinking_display(&mut self, display: crate::config::ThinkingDisplay) {
3199 self.thinking_display = display;
3200 }
3201
3202 pub(crate) fn wire_thinking_display(&self) -> crate::config::ThinkingDisplay {
3209 match self.codex_request_role {
3210 crate::runtime::openai::catalog::CodexRequestRole::Foreground => self.thinking_display,
3211 _ => crate::config::ThinkingDisplay::Summarized,
3212 }
3213 }
3214
3215 pub fn tool_keepwarm(&self) -> bool {
3216 self.tool_keepwarm
3217 }
3218
3219 pub fn set_tool_keepwarm(&mut self, enabled: bool) {
3221 self.tool_keepwarm = enabled;
3222 }
3223
3224 pub fn tool_spill_totals(&self) -> tool_spill::ToolSpillTotals {
3227 self.tool_spill_stats.totals()
3228 }
3229
3230 pub(crate) fn tool_spill(&self) -> tool_spill::ToolSpill {
3234 if !self.tool_spill_policy.is_enabled() {
3235 return tool_spill::ToolSpill::off();
3236 }
3237 let key = self
3238 .session_id
3239 .clone()
3240 .unwrap_or_else(|| self.trace_ctx.session_id().as_str().to_string());
3241 tool_spill::ToolSpill::new(
3242 self.tool_spill_policy,
3243 &self.tool_spill_root,
3244 &key,
3245 Arc::clone(&self.tool_spill_stats),
3246 )
3247 }
3248
3249 pub fn set_bash_timeout(&mut self, v: u64) {
3250 self.bash_timeout = v;
3251 }
3252
3253 pub fn set_bash_max_timeout(&mut self, v: u64) {
3254 self.bash_max_timeout = v;
3255 }
3256
3257 pub fn set_subagent_timeout(&mut self, v: u64) {
3258 self.subagent_timeout = v;
3259 }
3260
3261 pub(crate) fn apply_retry_config(&mut self, config: &crate::SynapsConfig) {
3264 self.api_retries = config.api_retries;
3265 self.refusal_retries = config.refusal_retries;
3266 }
3267
3268 pub fn refusal_retries(&self) -> u32 {
3269 self.refusal_retries
3270 }
3271
3272 pub fn set_api_retries(&mut self, v: u32) {
3273 self.api_retries = v;
3274 }
3275
3276 pub fn telemetry_level(&self) -> crate::runtime::telemetry::TelemetryLevel {
3277 self.telemetry_level
3278 }
3279
3280 pub fn set_telemetry_level(&mut self, level: crate::runtime::telemetry::TelemetryLevel) {
3281 self.telemetry_level = level;
3282 self.sync_observability();
3283 }
3284
3285 pub fn worker_telemetry(&self) -> crate::runtime::telemetry::WorkerTelemetry {
3287 crate::runtime::telemetry::WorkerTelemetry {
3288 level: self.telemetry_level,
3289 writer: self.telemetry_writer.clone(),
3290 parent_session_id: self.session_id.clone(),
3291 parent_runtime_session_id: Some(self.trace_ctx.session_id().as_str().to_string()),
3292 ..Default::default()
3293 }
3294 }
3295
3296 pub fn inherit_telemetry(&mut self, parent: &crate::runtime::telemetry::WorkerTelemetry) {
3307 match (&parent.writer, parent.level.enabled()) {
3308 (Some(writer), true) => {
3309 self.telemetry_level = parent.level;
3310 self.trace_ctx = trace::TraceContext::with_sink(Arc::new(
3311 crate::runtime::telemetry::WriterTraceSink::new(writer.clone()),
3312 ));
3313 self.telemetry_writer = parent.worker_writer();
3316 }
3317 _ => self.set_telemetry_level(crate::runtime::telemetry::TelemetryLevel::Off),
3318 }
3319 }
3320
3321 fn sync_observability(&mut self) {
3328 if self.telemetry_level.enabled() {
3329 if self.telemetry_writer.is_none() {
3330 let writer = crate::runtime::telemetry::TelemetryWriter::new(
3331 crate::runtime::telemetry::WriterOptions {
3332 max_generations: self.telemetry_log_generations,
3333 ..Default::default()
3334 },
3335 );
3336 self.trace_ctx = trace::TraceContext::with_sink(Arc::new(
3337 crate::runtime::telemetry::WriterTraceSink::new(writer.clone()),
3338 ));
3339 self.telemetry_writer = Some(writer);
3340 }
3341 } else {
3342 self.telemetry_writer = None;
3346 self.trace_ctx = trace::TraceContext::disabled();
3347 }
3348 }
3349
3350 pub fn trace_context(&self) -> trace::TraceContext {
3352 self.trace_ctx.clone()
3353 }
3354
3355 pub(crate) fn effective_trace_context(&self) -> trace::TraceContext {
3376 match self.trace_controls.consume() {
3377 None => self.trace_ctx.clone(),
3378 Some(with_content) => {
3379 let base = if self.trace_ctx.enabled() {
3380 self.trace_ctx.clone()
3381 } else {
3382 let writer = crate::runtime::telemetry::TelemetryWriter::new(
3383 crate::runtime::telemetry::WriterOptions::default(),
3384 );
3385 *self
3389 .one_shot_trace_writer
3390 .lock()
3391 .expect("one-shot trace writer lock poisoned") = Some(writer.clone());
3392 self.trace_ctx
3393 .fork_with_sink(Arc::new(crate::runtime::telemetry::WriterTraceSink::new(
3394 writer,
3395 )))
3396 .with_one_shot_request_gate()
3397 };
3398 if with_content {
3399 base.with_content_capture(Arc::new(trace::ContentCapture::new(
3400 self.capture_dir.clone(),
3401 )))
3402 } else {
3403 base
3404 }
3405 }
3406 }
3407 }
3408
3409 pub fn trace_arm_next(&self, with_content: bool) {
3413 self.trace_controls.arm_next(with_content);
3414 }
3415
3416 pub fn trace_status(&self) -> trace::TraceStatusReport {
3421 let _ = trace::sweep_expired_captures(&self.capture_dir);
3422 trace::TraceStatusReport {
3423 persistent_enabled: self.telemetry_writer.is_some(),
3424 arm: self.trace_controls.peek(),
3425 trace_path: crate::runtime::telemetry::default_trace_log_path(),
3426 writer_stats: self.telemetry_writer.as_ref().map(|w| w.stats()),
3427 degraded_records: self.trace_ctx.degraded_records(),
3428 }
3429 }
3430
3431 fn memory_context_lock(&self) -> std::sync::MutexGuard<'_, memory_context::SessionMemoryState> {
3438 self.memory_context_state
3439 .lock()
3440 .unwrap_or_else(std::sync::PoisonError::into_inner)
3441 }
3442
3443 pub fn memory_context_status(&self) -> memory_context::MemoryContextStatus {
3446 self.memory_context_lock().status()
3447 }
3448
3449 pub fn memory_context_disable(&self) -> memory_context::MemoryContextStatus {
3463 let (status, bound, session) = {
3464 let mut state = self.memory_context_lock();
3465 let session = state.session_id().clone();
3466 let bound = state.bound_provider_ids();
3467 let status = match memory_context::apply_memory_context_action(
3468 &mut state,
3469 memory_context::AuthorizedMemoryAction::Disable {
3470 session: session.clone(),
3471 },
3472 ) {
3473 Ok(status) => status,
3474 Err(_) => state.status(),
3478 };
3479 (status, bound, session)
3480 };
3481 memory_context::emit_memory_observability_event(
3484 &memory_context::MemoryObservabilityEvent::context_disabled(
3485 &session,
3486 &self.memory_context_project_id(),
3487 ),
3488 );
3489 self.revoke_memory_provider_leases(&bound);
3493 status
3494 }
3495
3496 fn revoke_memory_provider_leases(&self, bound: &[memory_context::ContextProviderId]) {
3504 let Some(extension_runtime) = &self.extension_runtime else {
3505 return;
3506 };
3507 for provider in bound {
3508 let mut parts = provider.as_str().splitn(3, ':');
3509 if let (Some("extension"), Some(plugin), Some(_local_id)) =
3510 (parts.next(), parts.next(), parts.next())
3511 {
3512 if !plugin.is_empty() {
3513 extension_runtime.revoke_plugin_lease(&self.host_tool_session, plugin);
3514 }
3515 }
3516 }
3517 }
3518
3519 fn resolve_memory_provider(
3539 &self,
3540 requested: Option<&str>,
3541 ) -> std::result::Result<memory_context::ContextProviderId, memory_context::MemoryContextError>
3542 {
3543 use crate::extensions::context_provider as ext_cp;
3544 if memory_context::provider_unavailable_for_test() {
3545 return Err(memory_context::MemoryContextError::ProviderNotRegistered);
3546 }
3547 if (self.memory_backend.exclusive() && !self.memory_backend.is_axel())
3548 || self
3549 .memory_backend_reconfigure_denied
3550 .load(std::sync::atomic::Ordering::SeqCst)
3551 {
3552 return Err(memory_context::MemoryContextError::ProviderNotRegistered);
3554 }
3555 if self.memory_backend.is_axel() {
3556 self.memory_backend
3557 .validate_axel_config()
3558 .map_err(|_| memory_context::MemoryContextError::ProviderNotRegistered)?;
3559 if requested.is_some_and(|id| id != axel_context::PROVIDER_ID) {
3560 return Err(memory_context::MemoryContextError::ProviderNotRegistered);
3561 }
3562 return memory_context::ContextProviderId::parse(axel_context::PROVIDER_ID);
3563 }
3564 let Some(extension_runtime) = &self.extension_runtime else {
3565 return Ok(memory_provider_id());
3566 };
3567 let requested_id = match requested {
3568 None => None,
3569 Some(raw) => Some(
3572 ext_cp::ContextProviderId::parse(raw)
3573 .map_err(|_| memory_context::MemoryContextError::ProviderNotRegistered)?,
3574 ),
3575 };
3576 match extension_runtime.resolve_context_provider(requested_id.as_ref()) {
3577 Ok(descriptor) => {
3578 memory_context::ContextProviderId::parse(&descriptor.runtime_address())
3579 }
3580 Err(ext_cp::ContextProviderLookupError::NotRegistered) => {
3581 Err(memory_context::MemoryContextError::ProviderNotRegistered)
3582 }
3583 Err(ext_cp::ContextProviderLookupError::Ambiguous) => {
3584 Err(memory_context::MemoryContextError::ProviderAmbiguous)
3585 }
3586 }
3587 }
3588
3589 pub(crate) fn memory_context_enable(
3597 &self,
3598 mode: memory_context::MemoryContextMode,
3599 proof: memory_context::UserIntentProof,
3600 ) -> std::result::Result<memory_context::MemoryContextStatus, memory_context::MemoryContextError>
3601 {
3602 self.memory_context_enable_resolved(mode, proof, None)
3603 }
3604
3605 pub(crate) fn memory_context_enable_resolved(
3610 &self,
3611 mode: memory_context::MemoryContextMode,
3612 proof: memory_context::UserIntentProof,
3613 requested_provider: Option<&str>,
3614 ) -> std::result::Result<memory_context::MemoryContextStatus, memory_context::MemoryContextError>
3615 {
3616 let mut state = self.memory_context_lock();
3617 let lease = self.grant_command_memory_lease(&state, mode, proof, requested_provider)?;
3618 let enabled_event = memory_context::MemoryObservabilityEvent::context_enabled(&lease);
3622 let result = memory_context::apply_memory_context_action(
3623 &mut state,
3624 memory_context::AuthorizedMemoryAction::Enable { lease },
3625 );
3626 if result.is_ok() {
3627 memory_context::emit_memory_observability_event(&enabled_event);
3628 }
3629 result
3630 }
3631
3632 pub(crate) fn apply_memory_default_mode(
3644 &self,
3645 config: &agent_core::config::MemoryConfig,
3646 ) -> Option<
3647 std::result::Result<
3648 memory_context::MemoryContextStatus,
3649 memory_context::MemoryContextError,
3650 >,
3651 > {
3652 use memory_context::MemoryContextMode as Mode;
3653 if !config.default_mode_confirmed {
3656 return None;
3657 }
3658 let mode = match config.default_mode.as_str() {
3659 "recall_once" => Mode::RecallOnce,
3660 "recall_each_prompt" => Mode::RecallEachPrompt,
3661 "capture_only" => Mode::CaptureOnly,
3662 "capture_and_recall" => Mode::CaptureAndRecall,
3663 _ => return None,
3664 };
3665 if self.memory_context_status().durable != memory_context::DurableStatus::Off {
3666 return None;
3667 }
3668 let proof = memory_context::UserIntentProof::OperatorConfig;
3669 let result = if mode == Mode::RecallOnce {
3670 self.memory_context_recall_once(proof)
3671 } else {
3672 self.memory_context_enable(mode, proof)
3673 };
3674 match &result {
3675 Ok(_) => tracing::info!(
3676 mode = mode.as_str(),
3677 "memory.default_mode applied to session"
3678 ),
3679 Err(e) => tracing::warn!(
3680 mode = mode.as_str(),
3681 error = %e,
3682 "memory.default_mode not applied; session starts with memory off"
3683 ),
3684 }
3685 Some(result)
3686 }
3687
3688 pub(crate) fn restore_session_memory_mode(&self, mode: crate::session::SessionMemoryMode) {
3695 let Some(lease) = mode.lease_mode() else {
3696 return;
3697 };
3698 if let Err(e) =
3699 self.memory_context_enable(lease, memory_context::mint_explicit_command_proof())
3700 {
3701 tracing::warn!(
3702 mode = lease.as_str(),
3703 error = %e,
3704 "session memory mode not restored; memory stays off"
3705 );
3706 }
3707 }
3708
3709 pub(crate) fn memory_context_recall_once(
3714 &self,
3715 proof: memory_context::UserIntentProof,
3716 ) -> std::result::Result<memory_context::MemoryContextStatus, memory_context::MemoryContextError>
3717 {
3718 let mut state = self.memory_context_lock();
3719 let lease = self.grant_command_memory_lease(
3720 &state,
3721 memory_context::MemoryContextMode::RecallOnce,
3722 proof,
3723 None,
3724 )?;
3725 let enabled_event = memory_context::MemoryObservabilityEvent::context_enabled(&lease);
3728 let result = memory_context::apply_memory_context_action(
3729 &mut state,
3730 memory_context::AuthorizedMemoryAction::RecallOnce { lease },
3731 );
3732 if result.is_ok() {
3733 memory_context::emit_memory_observability_event(&enabled_event);
3734 }
3735 result
3736 }
3737
3738 fn grant_command_memory_lease(
3745 &self,
3746 state: &memory_context::SessionMemoryState,
3747 mode: memory_context::MemoryContextMode,
3748 proof: memory_context::UserIntentProof,
3749 requested_provider: Option<&str>,
3750 ) -> std::result::Result<memory_context::MemoryContextLease, memory_context::MemoryContextError>
3751 {
3752 let provider_id = self.resolve_memory_provider(requested_provider)?;
3753 memory_context::MemoryContextLease::grant(
3754 memory_context::MemoryLeaseId::parse(&format!("memctx-cmd-{}", uuid::Uuid::new_v4()))?,
3755 state.session_id().clone(),
3756 self.memory_context_project_id(),
3757 provider_id,
3758 mode,
3759 memory_context::CapturePolicy::default(),
3760 memory_context::RecallPolicy::default(),
3761 proof,
3762 std::time::SystemTime::now(),
3763 None,
3764 )
3765 }
3766
3767 #[cfg(test)]
3768 pub(crate) fn set_capture_provider_for_test(
3769 &self,
3770 provider: std::sync::Arc<dyn capture_worker::CaptureProvider>,
3771 ) {
3772 *self
3773 .capture_provider_for_test
3774 .lock()
3775 .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(provider);
3776 }
3777
3778 pub(crate) fn compaction_capture_lease_at(
3779 &self,
3780 now: std::time::SystemTime,
3781 ) -> Option<memory_context::MemoryContextLease> {
3782 self.memory_context_lock().capture_lease_at(now)
3783 }
3784
3785 pub(crate) fn submit_compaction_summary_capture(
3786 &self,
3787 capture_lease: Option<memory_context::MemoryContextLease>,
3788 current: &agent_core::session::Session,
3789 api_messages: &[crate::SharedMessage],
3790 outcome: &compaction::CompactionOutcome,
3791 summarized_at: std::time::SystemTime,
3792 ) {
3793 let Some(lease) = capture_lease else {
3794 return;
3795 };
3796 let Ok(source_session_id) = memory_context::SessionId::parse(¤t.id) else {
3797 return;
3798 };
3799 if api_messages.is_empty() {
3804 return;
3805 }
3806 let first_turn_ordinal = 0;
3807 let last_turn_ordinal = (api_messages.len() - 1) as u64;
3808 let Some(source_digest) = chat_capture::MessageRangeDigest::from_hex(
3809 &agent_core::compaction::message_range_digest(api_messages),
3810 ) else {
3811 return;
3812 };
3813 let Some(prompt_stack_digest) =
3814 chat_capture::PromptStackDigest::from_hex(&outcome.prompt_stack_digest)
3815 else {
3816 return;
3817 };
3818 let summary_origin = if outcome.local_only {
3819 chat_capture::CompactionSummaryOrigin::LocalOnly
3820 } else {
3821 chat_capture::CompactionSummaryOrigin::Provider {
3822 provider_id: outcome.summary_provider.clone(),
3823 model_id: outcome.summary_model.clone(),
3824 }
3825 };
3826 let content_classes = outcome.included_classes.clone();
3827 let source = chat_capture::CompactionSource {
3828 schema: chat_capture::CompactionSchemaVersion::V1,
3829 project_id: lease.project_id.clone(),
3830 source_session_id,
3831 first_turn_ordinal,
3832 last_turn_ordinal,
3833 source_digest,
3834 summary_origin,
3835 prompt_stack_digest,
3836 redaction: match outcome.redaction_policy {
3837 agent_core::compaction::RedactionPolicy::TruncationOnly => {
3838 chat_capture::RedactionPolicy::None
3839 }
3840 agent_core::compaction::RedactionPolicy::PolicyExclusions => {
3841 chat_capture::RedactionPolicy::HostRedacted
3842 }
3843 },
3844 content_classes,
3845 summarized_at,
3846 };
3847
3848 #[cfg(any(test, feature = "testing"))]
3849 let test_provider = self
3850 .capture_provider_for_test
3851 .lock()
3852 .unwrap_or_else(std::sync::PoisonError::into_inner)
3853 .clone();
3854 #[cfg(not(any(test, feature = "testing")))]
3855 let test_provider: Option<std::sync::Arc<dyn capture_worker::CaptureProvider>> = None;
3856 let provider = test_provider.or_else(|| self.extension_capture_provider(&lease));
3857 if let Some(provider) = provider {
3858 let _ = memory_capture_worker().submit_summary(
3859 &lease,
3860 source,
3861 api_messages.len(),
3862 &outcome.summary_text,
3863 outcome.redaction_policy,
3864 memory_context::RetentionClass::Standard,
3865 provider,
3866 );
3867 }
3868 }
3869
3870 fn extension_capture_provider(
3871 &self,
3872 lease: &memory_context::MemoryContextLease,
3873 ) -> Option<std::sync::Arc<dyn capture_worker::CaptureProvider>> {
3874 if (self.memory_backend.exclusive() && !self.memory_backend.is_axel())
3875 || self
3876 .memory_backend_reconfigure_denied
3877 .load(std::sync::atomic::Ordering::SeqCst)
3878 {
3879 return None;
3880 }
3881 if self.memory_backend.is_axel() {
3882 if lease.provider_id.as_str() != axel_context::PROVIDER_ID
3883 || lease.project_id != self.memory_context_project_id()
3884 {
3885 return None;
3886 }
3887 return Some(Arc::new(axel_context::AxelCaptureProvider {
3888 binding: self.memory_backend.clone(),
3889 state: self.memory_context_state.clone(),
3890 lease: lease.clone(),
3891 }));
3892 }
3893 let manager = self.extension_runtime.clone()?;
3894 let mut parts = lease.provider_id.as_str().splitn(3, ':');
3895 let (Some("extension"), Some(plugin), Some(_)) = (parts.next(), parts.next(), parts.next())
3896 else {
3897 return None;
3898 };
3899 let digest =
3900 manager.declared_tool_digest(plugin, memory_context::MEMORY_CAPTURE_TOOL_NAME)?;
3901 let handle = tokio::runtime::Handle::try_current().ok()?;
3902 Some(std::sync::Arc::new(ExtensionCaptureProvider {
3903 manager,
3904 session: self.host_tool_session.clone(),
3905 plugin: plugin.to_owned(),
3906 digest,
3907 handle,
3908 }))
3909 }
3910
3911 pub fn memory_history_preview(
3915 &self,
3916 ) -> std::result::Result<memory_history::HistoryImportPreview, memory_history::HistoryImportError>
3917 {
3918 let host = memory_history::HistoryImportHostState::from_current_host()?;
3919 let mut io = memory_history::CanonicalHistoryMetadataIo::new();
3920 let preview = memory_history::preview_history_import(&host, &mut io)?;
3921 *self
3922 .pending_history_import_preview
3923 .lock()
3924 .expect("history preview mutex poisoned") = Some(preview.clone());
3925 Ok(preview)
3926 }
3927
3928 pub(crate) fn memory_history_confirm(
3932 &self,
3933 ) -> std::result::Result<memory_history::HistoryImportReport, memory_history::HistoryImportError>
3934 {
3935 let preview = self
3936 .pending_history_import_preview
3937 .lock()
3938 .expect("history preview mutex poisoned")
3939 .take()
3940 .ok_or(memory_history::HistoryImportError::ConsentRequired)?;
3941 let intent = memory_history::ConfirmedUserIntent::from_explicit_command(
3942 memory_context::mint_explicit_command_id(),
3943 );
3944 let mut io = memory_history::CanonicalHistoryMetadataIo::new();
3945 match memory_history::authorize_history_import(
3946 preview,
3947 memory_history::HistoryImportConsent::Confirmed(intent),
3948 memory_history::RequestAuthority::UserFrontend,
3949 &mut io,
3950 )? {
3951 memory_history::HistoryImportOutcome::Ready(plan) => {
3952 *self
3953 .history_import_plan
3954 .lock()
3955 .expect("history plan mutex poisoned") = Some(plan.clone());
3956 let lease = self
3957 .memory_context_lock()
3958 .capture_lease_at(std::time::SystemTime::now())
3959 .ok_or(memory_history::HistoryImportError::CaptureLeaseUnavailable)?;
3960 let provider = self
3963 .extension_capture_provider(&lease)
3964 .ok_or(memory_history::HistoryImportError::CaptureProviderUnavailable)?;
3965 let sessions_dir = agent_core::config::get_active_config_dir().join("sessions");
3966 memory_history::import_history_from_dir(
3967 &plan,
3968 &lease,
3969 &sessions_dir,
3970 memory_history::IMPORT_BATCH_MAX_RECORDS,
3971 memory_capture_worker(),
3972 provider,
3973 )
3974 }
3975 memory_history::HistoryImportOutcome::Declined => {
3976 Err(memory_history::HistoryImportError::ConsentRequired)
3977 }
3978 }
3979 }
3980
3981 pub(crate) fn memory_history_decline(&self) -> bool {
3983 self.pending_history_import_preview
3984 .lock()
3985 .expect("history preview mutex poisoned")
3986 .take()
3987 .is_some()
3988 }
3989
3990 #[cfg(test)]
3992 pub(crate) fn memory_durable_proof_for_test(&self) -> Option<memory_context::UserIntentProof> {
3993 self.memory_context_lock().durable_proof().cloned()
3994 }
3995
3996 #[cfg(test)]
3998 pub(crate) fn memory_one_shot_proof_for_test(&self) -> Option<memory_context::UserIntentProof> {
3999 self.memory_context_lock().one_shot_pending_proof().cloned()
4000 }
4001
4002 #[cfg(test)]
4005 pub(crate) fn memory_bound_providers_for_test(&self) -> Vec<memory_context::ContextProviderId> {
4006 self.memory_context_lock().bound_provider_ids()
4007 }
4008
4009 pub fn context_report(&self, history: Option<&[crate::SharedMessage]>) -> trace::ContextReport {
4018 use trace::ReportValue;
4019 let (history_messages, history_bytes) = match history {
4020 Some(msgs) => (
4021 ReportValue::Count(msgs.len() as u64),
4022 ReportValue::Count(
4023 msgs.iter()
4024 .map(|m| {
4025 serde_json::to_vec(&**m)
4026 .map(|v| v.len() as u64)
4027 .unwrap_or(0)
4028 })
4029 .sum(),
4030 ),
4031 ),
4032 None => (ReportValue::Unavailable, ReportValue::Unavailable),
4033 };
4034 let tool_count = match self.tools.try_read() {
4037 Ok(tools) => ReportValue::Count(tools.iter_tools_sorted().len() as u64),
4038 Err(_) => ReportValue::Unavailable,
4039 };
4040 trace::ContextReport {
4041 model: self.model.clone(),
4042 system_prompt_bytes: ReportValue::Count(
4043 self.system_prompt().map(|s| s.len() as u64).unwrap_or(0),
4044 ),
4045 tool_count,
4046 history_messages,
4047 history_bytes,
4048 loaded_skills: ReportValue::Unavailable,
4049 loaded_memories: ReportValue::Unavailable,
4050 cache: self.trace_ctx.cache_snapshots().last_activity(),
4051 trace_enabled: self.trace_ctx.enabled(),
4052 writer_stats: self.telemetry_writer.as_ref().map(|w| w.stats()),
4053 degraded_records: self.trace_ctx.degraded_records(),
4054 }
4055 }
4056
4057 pub fn telemetry_writer(&self) -> Option<crate::runtime::telemetry::TelemetryWriter> {
4059 self.telemetry_writer.clone()
4060 }
4061
4062 pub(crate) fn request_telemetry(&self) -> Option<crate::runtime::telemetry::TelemetryWriter> {
4066 self.telemetry_writer
4067 .as_ref()
4068 .map(|w| w.with_session(self.session_id.clone()))
4069 }
4070
4071 pub fn shutdown_observability(
4085 &self,
4086 timeout: std::time::Duration,
4087 ) -> Option<crate::runtime::telemetry::ShutdownOutcome> {
4088 let _ = trace::sweep_expired_captures(&self.capture_dir);
4093 let ephemeral = self
4099 .one_shot_trace_writer
4100 .lock()
4101 .expect("one-shot trace writer lock poisoned")
4102 .take();
4103 let ephemeral_outcome = ephemeral.map(|w| w.shutdown(timeout));
4104 self.telemetry_writer
4105 .as_ref()
4106 .map(|w| w.shutdown(timeout))
4107 .or(ephemeral_outcome)
4108 }
4109
4110 pub async fn shutdown_observability_async(
4123 &self,
4124 timeout: std::time::Duration,
4125 ) -> Option<crate::runtime::telemetry::ShutdownOutcome> {
4126 let capture_dir = self.capture_dir.clone();
4130 let _ =
4131 tokio::task::spawn_blocking(move || trace::sweep_expired_captures(&capture_dir)).await;
4132 let ephemeral = self
4134 .one_shot_trace_writer
4135 .lock()
4136 .expect("one-shot trace writer lock poisoned")
4137 .take();
4138 let ephemeral_outcome = match ephemeral {
4139 Some(writer) => Some(writer.shutdown_async(timeout).await),
4140 None => None,
4141 };
4142 match self.telemetry_writer.clone() {
4143 Some(writer) => Some(writer.shutdown_async(timeout).await),
4144 None => ephemeral_outcome,
4145 }
4146 }
4147
4148 #[cfg(any(test, feature = "testing"))]
4152 pub fn install_observability_for_tests(
4153 &mut self,
4154 writer: crate::runtime::telemetry::TelemetryWriter,
4155 ) {
4156 self.trace_ctx = trace::TraceContext::with_sink(Arc::new(
4157 crate::runtime::telemetry::WriterTraceSink::new(writer.clone()),
4158 ));
4159 self.telemetry_writer = Some(writer);
4160 }
4161
4162 pub fn cache_diagnostics(&self) -> bool {
4163 self.cache_diagnostics
4164 }
4165
4166 pub fn set_cache_diagnostics(&mut self, v: bool) {
4167 self.cache_diagnostics = v;
4168 }
4169
4170 pub fn set_request_compression(
4173 &mut self,
4174 mode: agent_core::config::RequestCompressionMode,
4175 level: u32,
4176 ) {
4177 self.request_compression = request_encoding::RequestCompression::from_config(mode, level);
4178 }
4179
4180 #[cfg(test)]
4182 pub(crate) fn request_compression_for_test(
4183 &self,
4184 ) -> (Option<agent_core::config::RequestCompressionMode>, u32) {
4185 self.request_compression.configured()
4186 }
4187
4188 pub fn cache_ttl(&self) -> crate::core::config::CacheTtlSetting {
4191 self.cache_ttl
4192 }
4193
4194 pub async fn resolved_cache_ttl(&self) -> crate::core::config::CacheTtl {
4201 let auth = self.auth.read().await;
4202 self.cache_ttl.resolve_for_auth_type(&auth.auth_type)
4203 }
4204
4205 pub fn share_ttl_latches(&mut self, parent: &Runtime) {
4209 self.ttl_downgrade_notified = parent.ttl_downgrade_notified.clone();
4210 self.saw_1h_honored = parent.saw_1h_honored.clone();
4211 self.cache_anchor = parent.cache_anchor.clone();
4212 self.last_msg_id = parent.last_msg_id.clone();
4215 self.cache_diag_rejected = parent.cache_diag_rejected.clone();
4216 }
4217
4218 pub(crate) fn cache_diagnosis(&self) -> Option<cache_diagnosis::CacheDiagnosis> {
4221 self.cache_diagnostics.then(|| {
4222 cache_diagnosis::CacheDiagnosis::new(
4223 Arc::clone(&self.last_msg_id),
4224 Arc::clone(&self.cache_diag_rejected),
4225 )
4226 })
4227 }
4228
4229 #[cfg(test)]
4233 pub(crate) async fn set_auth_kind_for_test(&self, kind: &str) {
4234 self.auth.write().await.auth_type = kind.to_string();
4235 }
4236
4237 pub fn set_cache_ttl(&mut self, ttl: impl Into<crate::core::config::CacheTtlSetting>) {
4243 self.cache_ttl = ttl.into();
4244 }
4245
4246 pub fn set_worker_context_settings(
4250 &mut self,
4251 settings: Option<worker_context::WorkerContextSettings>,
4252 ) {
4253 self.worker_context.settings = settings;
4254 }
4255
4256 pub fn attach_worker_transcripts(&mut self, sink: worker_context::WorkerTranscriptSink) {
4259 self.worker_context.sink = Some(sink);
4260 }
4261
4262 pub fn worker_context(&self) -> &worker_context::WorkerContext {
4263 &self.worker_context
4264 }
4265
4266 pub fn thinking_level(&self) -> &str {
4267 match self.named_level {
4270 Some(level) => level.as_str(),
4271 None => crate::core::models::thinking_level_for_budget(self.thinking_budget),
4272 }
4273 }
4274
4275 pub async fn refresh_if_needed(&self) -> Result<()> {
4277 self.refresh_if_needed_for_model(&self.model).await
4278 }
4279
4280 pub async fn reload_auth(&self) -> Result<String> {
4285 AuthMethods::reload_after_rejection(
4286 Arc::clone(&self.auth),
4287 &self.client,
4288 &self.credential_source,
4289 &self.token_cache,
4290 None,
4291 )
4292 .await?;
4293 let expires = self.auth.read().await.token_expires;
4294 let source = if self.credential_source.is_remote() {
4295 "remote broker"
4296 } else {
4297 "auth.json"
4298 };
4299 Ok(match expires {
4300 Some(exp) => {
4301 let mins = exp.saturating_sub(crate::epoch_millis()) / 60_000;
4302 format!("Anthropic credentials reloaded from {source} (valid ~{mins} min)")
4303 }
4304 None => format!("Anthropic credentials reloaded from {source}"),
4305 })
4306 }
4307
4308 pub(crate) async fn refresh_for_keepwarm(&self) -> Result<()> {
4311 if !crate::runtime::auth::model_is_anthropic(&self.model) {
4312 return Ok(());
4313 }
4314 AuthMethods::refresh_if_needed(
4315 Arc::clone(&self.auth),
4316 &self.client,
4317 &self.credential_source,
4318 &self.token_cache,
4319 Some(&self.model),
4320 crate::runtime::auth::RefreshPoint::KeepWarm,
4321 )
4322 .await
4323 }
4324
4325 async fn refresh_if_needed_for_model(&self, model: &str) -> Result<()> {
4326 if !crate::runtime::auth::model_is_anthropic(model) {
4329 return Ok(());
4330 }
4331 AuthMethods::refresh_if_needed(
4332 Arc::clone(&self.auth),
4333 &self.client,
4334 &self.credential_source,
4335 &self.token_cache,
4336 Some(model),
4337 crate::runtime::auth::RefreshPoint::TurnStart,
4338 )
4339 .await
4340 }
4341
4342 pub async fn compact_call(&self, messages: Vec<crate::SharedMessage>) -> Result<String> {
4348 self.remote_summarization_attempts
4351 .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
4352 let model = self.compaction_model();
4353 self.validate_request_preflight_for(
4354 model,
4355 crate::runtime::openai::catalog::CodexRequestRole::Internal,
4356 )
4357 .await?;
4358 self.refresh_if_needed_for_model(model).await?;
4359
4360 use crate::runtime::compaction::COMPACTION_SYSTEM_PROMPT;
4361
4362 ApiMethods::call_api_simple(
4363 &self.auth,
4364 &self.client,
4365 model,
4366 COMPACTION_SYSTEM_PROMPT,
4367 self.thinking_budget,
4368 self.reasoning_level(),
4369 &messages,
4370 self.api_retries,
4371 &api::ApiOptions {
4377 trace: self.trace_ctx.clone(),
4378 telemetry: self.telemetry_writer.clone(),
4379 request_compression: self.request_compression.clone(),
4381 ..api::ApiOptions::default()
4382 },
4383 )
4384 .await
4385 }
4386
4387 pub async fn run_single(&self, prompt: &str) -> Result<String> {
4390 if self
4391 .continuation
4392 .lock()
4393 .unwrap_or_else(std::sync::PoisonError::into_inner)
4394 .durability_blocked
4395 {
4396 return Err(RuntimeError::Session(
4397 "context head save is unresolved; reload the session before further inference"
4398 .into(),
4399 ));
4400 }
4401 self.validate_request_preflight().await?;
4402 let anthropic_execution_plan = self.authorized_anthropic_plan().await?;
4403 self.refresh_if_needed().await?;
4405
4406 let mut messages: Vec<crate::SharedMessage> = vec![std::sync::Arc::new(
4407 json!({"role": "user", "content": prompt}),
4408 )];
4409 let system_prompt = self.effective_system_prompt().await;
4410
4411 loop {
4412 let response = ApiMethods::call_api(
4413 &self.auth,
4414 &self.client,
4415 &self.model,
4416 &*self.tools.read().await,
4417 &system_prompt,
4418 self.thinking_budget,
4419 self.reasoning_level(),
4420 &messages,
4421 self.api_retries,
4422 &api::ApiOptions {
4423 memory_backend: None,
4424 use_1m_context: self.context_window_override == Some(1_000_000),
4425 cache_ttl: self.cache_ttl,
4426 ttl_downgrade_notified: self.ttl_downgrade_notified.clone(),
4427 saw_1h_honored: self.saw_1h_honored.clone(),
4428 credential_source: self.credential_source.clone(),
4429 token_cache: self.token_cache.clone(),
4430 anthropic_base_url: None,
4431 anthropic_stream_client: self.anthropic_stream_client.clone(),
4432 request_compression: self.request_compression.clone(),
4433 anthropic_execution_plan: anthropic_execution_plan.clone(),
4434 codex_request_role: self.codex_request_role(),
4435 thinking_display: self.wire_thinking_display(),
4436 retention_frontier: None,
4439 trace: self.effective_trace_context(),
4444 request_correlation: None,
4445 suppress_stream_deltas: true,
4446 telemetry: self.telemetry_writer.clone(),
4447 tool_session_id: Some(self.host_tool_session.clone()),
4454 session_tool_set: None,
4458 request_tools_schema: None,
4459 anthropic_tool_split: None,
4460 usage_counters: None,
4461 cache_anchor: None,
4462 keepwarm: None,
4463 cache_diagnosis: None,
4465 context_rollover: None,
4466 },
4467 )
4468 .await?;
4469
4470 if let Some(content) = response["content"].as_array() {
4472 let mut response_text = String::new();
4473 let mut tool_uses = Vec::new();
4474
4475 for item in content {
4477 match item["type"].as_str() {
4478 Some("text") => {
4479 if let Some(text) = item["text"].as_str() {
4480 response_text.push_str(text);
4481 }
4482 }
4483 Some("tool_use") => {
4484 tool_uses.push(item.clone());
4485 }
4486 _ => {}
4487 }
4488 }
4489
4490 if tool_uses.is_empty() {
4492 if let Some(orchestration) = &self.orchestration {
4493 if let agent_core::orchestration::CompletionGate::Blocked { workers } =
4494 orchestration.completion_gate()
4495 {
4496 return Err(crate::RuntimeError::Tool(format!(
4497 "completion blocked: {} worker(s) require collection/reconciliation: {} (call subagent_collect with reconciled=true after inspecting each result)",
4498 workers.len(),
4499 workers.join(", ")
4500 )));
4501 }
4502 }
4503 return Ok(response_text);
4504 }
4505
4506 messages.push(std::sync::Arc::new(json!({
4508 "role": "assistant",
4509 "content": content
4510 })));
4511
4512 let mut tool_results = Vec::new();
4514
4515 if tool_uses.len() == 1 {
4516 let tool_use = &tool_uses[0];
4518 if let (Some(tool_name), Some(tool_id)) =
4519 (tool_use["name"].as_str(), tool_use["id"].as_str())
4520 {
4521 let input = &tool_use["input"];
4522 let (result, rich_blocks) = match self.tools.read().await.get(tool_name).cloned() {
4523 Some(tool) => {
4524 let input = self
4525 .tools
4526 .read()
4527 .await
4528 .translate_input_for_api_tool(tool_name, input.clone());
4529 let runtime_name = self
4530 .tools
4531 .read()
4532 .await
4533 .runtime_name_for_api(tool_name)
4534 .to_string();
4535 let ctx = crate::ToolContext {
4536 channels: crate::tools::ToolChannels {
4537 tx_delta: None,
4538 tx_events: None,
4539 },
4540 capabilities: crate::tools::ToolCapabilities {
4541 session_allow_all: Some(self.session_allow_all.clone()),
4542 launch_cancel: None,
4543 memory_backend: Some(self.memory_backend.clone()),
4544 telemetry: Some(self.worker_telemetry()),
4545 watcher_exit_path: self.watcher_exit_path.clone(),
4546 tool_register_tx: None,
4547 session_manager: Some(self.session_manager.clone()),
4548 subagent_registry: Some(self.subagent_registry.clone()),
4549 delegation_parent: None,
4550 codex_parent_plan: Self::codex_delegation_plan(
4551 &self.model,
4552 self.reasoning_level(),
4553 self.codex_request_role(),
4554 ),
4555 event_queue: Some(self.event_queue.clone()),
4556 secret_prompt: None,
4557 orchestration: self.orchestration.clone(),
4558 tool_activation: None,
4559 mcp_leases: None,
4560 extension_leases: None,
4561 memory_context: self.memory_tool_capability(),
4562 cwd: self.cwd.clone(),
4563 env: self.env.clone(),
4564 env_stripped: self.env_stripped.clone(),
4565 env_warned: self.env_warned.clone(),
4566 session_id: self.session_id.clone(),
4567 },
4568 limits: crate::tools::ToolLimits {
4569 max_tool_output: self.max_tool_output,
4570 max_tool_buffer: 256 * 1024,
4571 bash_timeout: self.bash_timeout,
4572 bash_max_timeout: self.bash_max_timeout,
4573 subagent_timeout: self.subagent_timeout,
4574 },
4575 };
4576 let decision = resolve_before_tool_call_decision(
4577 input.clone(),
4578 emit_before_tool_call(
4579 &self.hook_bus,
4580 tool_name,
4581 Some(&runtime_name),
4582 input.clone(),
4583 self.session_id.as_deref(),
4584 )
4585 .await,
4586 None,
4587 false,
4588 Some(&self.session_allow_all),
4589 )
4590 .await;
4591 if let BeforeToolCallDecision::Block { reason } = decision {
4592 (format!("Tool call blocked by extension: {}", reason), None)
4593 } else {
4594 let BeforeToolCallDecision::Continue { input } = decision
4595 else {
4596 unreachable!()
4597 };
4598 let input_for_hook = input.clone();
4599 let (output, rich_blocks) = match tool.execute_rich(input, ctx).await {
4600 Ok(output) => validated_single_tool_output(&self.model, output),
4601 Err(e) => (e.to_string(), None),
4602 };
4603 let hooked_output = emit_after_tool_call(
4604 &self.hook_bus,
4605 tool_name,
4606 Some(&runtime_name),
4607 input_for_hook,
4608 output.clone(),
4609 self.max_tool_output,
4610 self.session_id.as_deref(),
4611 )
4612 .await;
4613 let rich_blocks = retain_single_tool_blocks(rich_blocks, &output, &hooked_output);
4614 (hooked_output, rich_blocks)
4615 }
4616 }
4617 None => (format!("Unknown tool: {}", tool_name), None),
4618 };
4619 tool_results.push(json!({
4620 "type": "tool_result",
4621 "tool_use_id": tool_id,
4622 "content": single_tool_result_content(&result, rich_blocks, self.max_tool_output)
4623 }));
4624 }
4625 } else {
4626 let mut join_set = tokio::task::JoinSet::new();
4628
4629 let cfg_max_tool_output = self.max_tool_output;
4631 let cfg_bash_timeout = self.bash_timeout;
4632 let cfg_bash_max_timeout = self.bash_max_timeout;
4633 let cfg_subagent_timeout = self.subagent_timeout;
4634 let session_mgr = self.session_manager.clone();
4635 let cfg_subagent_registry = self.subagent_registry.clone();
4636 let cfg_event_queue = self.event_queue.clone();
4637 let cfg_hook_bus = self.hook_bus.clone();
4638 let cfg_orchestration = self.orchestration.clone();
4639 let codex_parent_plan = Self::codex_delegation_plan(
4640 &self.model,
4641 self.reasoning_level(),
4642 self.codex_request_role(),
4643 );
4644 let cfg_cwd = self.cwd.clone();
4645 let cfg_env = self.env.clone();
4646 let cfg_session_id = self.session_id.clone();
4647
4648 for tool_use in &tool_uses {
4649 if let (Some(tool_name), Some(tool_id)) = (
4650 tool_use["name"].as_str().map(|s| s.to_string()),
4651 tool_use["id"].as_str().map(|s| s.to_string()),
4652 ) {
4653 let input = tool_use["input"].clone();
4654 let tools_snapshot = self.tools.read().await;
4655 let input =
4656 tools_snapshot.translate_input_for_api_tool(&tool_name, input);
4657 let runtime_name =
4658 tools_snapshot.runtime_name_for_api(&tool_name).to_string();
4659 let tool = tools_snapshot.get(&tool_name).cloned();
4660 drop(tools_snapshot);
4661 let exit_path = self.watcher_exit_path.clone();
4662 let session_mgr_inner = session_mgr.clone();
4663 let registry_inner = cfg_subagent_registry.clone();
4664 let event_queue_inner = cfg_event_queue.clone();
4665 let hook_bus_inner = cfg_hook_bus.clone();
4666 let session_allow_all_inner = self.session_allow_all.clone();
4667 let orchestration_inner = cfg_orchestration.clone();
4668 let memory_backend_inner = self.memory_backend.clone();
4669 let worker_telemetry_inner = self.worker_telemetry();
4670 let memory_context_inner = self.memory_tool_capability();
4671 let codex_parent_plan_inner = codex_parent_plan.clone();
4672 let cwd_inner = cfg_cwd.clone();
4673 let env_inner = cfg_env.clone();
4674 let env_stripped_inner = self.env_stripped.clone();
4675 let env_warned_inner = self.env_warned.clone();
4676 let session_id_inner = cfg_session_id.clone();
4677 let attachment_model = self.model.clone();
4678 let tool_name_for_hook = tool_name.clone();
4679 let runtime_name_for_hook = runtime_name.clone();
4680
4681 join_set.spawn(async move {
4682 let result = match tool {
4683 Some(t) => {
4684 let decision =
4685 crate::runtime::resolve_before_tool_call_decision(
4686 input.clone(),
4687 crate::runtime::emit_before_tool_call(
4688 &hook_bus_inner,
4689 &tool_name_for_hook,
4690 Some(&runtime_name_for_hook),
4691 input.clone(),
4692 session_id_inner.as_deref(),
4693 )
4694 .await,
4695 None,
4696 false,
4697 Some(&session_allow_all_inner),
4698 )
4699 .await;
4700 if let crate::runtime::BeforeToolCallDecision::Block {
4701 reason,
4702 } = decision
4703 {
4704 (format!("Tool call blocked by extension: {}", reason), None)
4705 } else {
4706 let crate::runtime::BeforeToolCallDecision::Continue {
4707 input,
4708 } = decision
4709 else {
4710 unreachable!()
4711 };
4712 let ctx = crate::ToolContext {
4713 channels: crate::tools::ToolChannels {
4714 tx_delta: None,
4715 tx_events: None,
4716 },
4717 capabilities: crate::tools::ToolCapabilities {
4718 session_allow_all: Some(session_allow_all_inner.clone()),
4719 launch_cancel: None,
4720 memory_backend: Some(memory_backend_inner),
4721 telemetry: Some(worker_telemetry_inner),
4722 watcher_exit_path: exit_path,
4723 tool_register_tx: None,
4724 session_manager: Some(session_mgr_inner),
4725 subagent_registry: Some(registry_inner),
4726 delegation_parent: None,
4727 codex_parent_plan: codex_parent_plan_inner,
4728 event_queue: Some(event_queue_inner),
4729 secret_prompt: None,
4730 orchestration: orchestration_inner,
4731 tool_activation: None,
4732 mcp_leases: None,
4733 extension_leases: None,
4734 memory_context: memory_context_inner,
4735 cwd: cwd_inner,
4736 env: env_inner,
4737 env_stripped: env_stripped_inner,
4738 env_warned: env_warned_inner,
4739 session_id: session_id_inner.clone(),
4740 },
4741 limits: crate::tools::ToolLimits {
4742 max_tool_output: cfg_max_tool_output,
4743 max_tool_buffer: 256 * 1024,
4744 bash_timeout: cfg_bash_timeout,
4745 bash_max_timeout: cfg_bash_max_timeout,
4746 subagent_timeout: cfg_subagent_timeout,
4747 },
4748 };
4749 let input_for_hook = input.clone();
4750 let (output, rich_blocks) =
4751 match t.execute_rich(input, ctx).await {
4752 Ok(output) => validated_single_tool_output(
4753 &attachment_model,
4754 output,
4755 ),
4756 Err(e) => (e.to_string(), None),
4757 };
4758 let hooked_output =
4759 crate::runtime::emit_after_tool_call(
4760 &hook_bus_inner,
4761 &tool_name_for_hook,
4762 Some(&runtime_name_for_hook),
4763 input_for_hook,
4764 output.clone(),
4765 cfg_max_tool_output,
4766 session_id_inner.as_deref(),
4767 )
4768 .await;
4769 let rich_blocks = retain_single_tool_blocks(
4770 rich_blocks,
4771 &output,
4772 &hooked_output,
4773 );
4774 (hooked_output, rich_blocks)
4775 }
4776 }
4777 None => (format!("Unknown tool: {}", tool_name), None),
4778 };
4779 (tool_id, result)
4780 });
4781 }
4782 }
4783
4784 let mut results_map = std::collections::HashMap::new();
4786 while let Some(res) = join_set.join_next().await {
4787 match res {
4788 Ok((tool_id, result)) => {
4789 results_map.insert(tool_id, result);
4790 }
4791 Err(e) => {
4792 tracing::error!("Parallel tool task panicked: {}", e);
4794 }
4795 }
4796 }
4797
4798 for tool_use in &tool_uses {
4800 if let Some(tool_id) = tool_use["id"].as_str() {
4801 let (result, rich_blocks) = results_map.remove(tool_id).unwrap_or_else(|| {
4802 ("Tool execution failed: task panicked".to_string(), None)
4803 });
4804 tool_results.push(json!({
4805 "type": "tool_result",
4806 "tool_use_id": tool_id,
4807 "content": single_tool_result_content(&result, rich_blocks, self.max_tool_output)
4808 }));
4809 }
4810 }
4811 }
4812
4813 messages.push(std::sync::Arc::new(json!({
4815 "role": "user",
4816 "content": tool_results
4817 })));
4818
4819 } else {
4821 return Err(RuntimeError::Tool("Invalid response format".to_string()));
4822 }
4823 }
4824 }
4825
4826 pub async fn run_stream(
4829 &self,
4830 prompt: String,
4831 cancel: CancellationToken,
4832 ) -> Pin<Box<dyn Stream<Item = StreamEvent> + Send>> {
4833 self.run_stream_with_messages(
4834 vec![std::sync::Arc::new(
4835 json!({"role": "user", "content": prompt}),
4836 )],
4837 cancel,
4838 None,
4839 None,
4840 false,
4841 )
4842 .await
4843 }
4844
4845 pub fn session_allow_all(&self) -> &std::sync::Arc<crate::tools::SessionConfirmLatch> {
4848 &self.session_allow_all
4849 }
4850
4851 pub(crate) fn inherit_session_allow_all(
4853 &mut self,
4854 parent: Option<&std::sync::Arc<crate::tools::SessionConfirmLatch>>,
4855 ) {
4856 if let Some(parent) = parent {
4857 self.session_allow_all = parent.clone();
4858 }
4859 }
4860
4861 pub async fn run_stream_with_messages(
4865 &self,
4866 messages: Vec<crate::SharedMessage>,
4867 cancel: CancellationToken,
4868 steering_rx: Option<mpsc::UnboundedReceiver<String>>,
4869 secret_prompt: Option<crate::tools::SecretPromptHandle>,
4870 auto_approve_confirms: bool,
4871 ) -> Pin<Box<dyn Stream<Item = StreamEvent> + Send>> {
4872 self.run_stream_tracked(
4873 messages,
4874 cancel,
4875 steering_rx,
4876 secret_prompt,
4877 auto_approve_confirms,
4878 TurnCompletion::new(),
4879 )
4880 .await
4881 }
4882
4883 pub async fn run_stream_tracked(
4887 &self,
4888 messages: Vec<crate::SharedMessage>,
4889 cancel: CancellationToken,
4890 steering_rx: Option<mpsc::UnboundedReceiver<String>>,
4891 secret_prompt: Option<crate::tools::SecretPromptHandle>,
4892 auto_approve_confirms: bool,
4893 completion: TurnCompletion,
4894 ) -> Pin<Box<dyn Stream<Item = StreamEvent> + Send>> {
4895 let (tx, internal_rx) = mpsc::unbounded_channel();
4901 let bounded_rx =
4902 crate::runtime::relay::spawn_bounded_stream_relay(internal_rx, cancel.clone());
4903
4904 let turn_correlation_id = agent_core::next_turn_correlation_id();
4907
4908 if self
4909 .continuation
4910 .lock()
4911 .unwrap_or_else(std::sync::PoisonError::into_inner)
4912 .durability_blocked
4913 {
4914 let error = RuntimeError::Session(
4915 "context head save is unresolved; reload the session before further inference"
4916 .into(),
4917 );
4918 let _ = tx.send(StreamEvent::Session(SessionEvent::Error(
4919 helpers::turn_error_for(&error, &turn_correlation_id),
4920 )));
4921 let _ = tx.send(StreamEvent::Session(SessionEvent::Done));
4922 return Box::pin(tokio_stream::wrappers::ReceiverStream::new(bounded_rx));
4923 }
4924
4925 if let Err(error) = self.validate_request_preflight().await {
4926 let _ = tx.send(StreamEvent::Session(SessionEvent::Error(
4927 helpers::turn_error_for(&error, &turn_correlation_id),
4928 )));
4929 let _ = tx.send(StreamEvent::Session(SessionEvent::Done));
4930 return Box::pin(tokio_stream::wrappers::ReceiverStream::new(bounded_rx));
4931 }
4932
4933 let anthropic_execution_plan = match self.authorized_anthropic_plan().await {
4934 Ok(plan) => plan,
4935 Err(error) => {
4936 let _ = tx.send(StreamEvent::Session(SessionEvent::Error(
4937 helpers::turn_error_for(&error, &turn_correlation_id),
4938 )));
4939 let _ = tx.send(StreamEvent::Session(SessionEvent::Done));
4940 return Box::pin(tokio_stream::wrappers::ReceiverStream::new(bounded_rx));
4941 }
4942 };
4943
4944 if let Err(e) = self.refresh_if_needed().await {
4946 let _ = tx.send(StreamEvent::Session(SessionEvent::Error(
4947 helpers::turn_error_for(&e, &turn_correlation_id),
4948 )));
4949 let _ = tx.send(StreamEvent::Session(SessionEvent::Done));
4950 return Box::pin(tokio_stream::wrappers::ReceiverStream::new(bounded_rx));
4951 }
4952
4953 let capture_started_at = std::time::SystemTime::now();
4963 let capture_turn_start = terminal_capture_start(&messages);
4964 let capture_lease = self
4967 .memory_context_lock()
4968 .capture_lease_at(capture_started_at);
4969 let mut messages = messages;
4970 self.apply_turn_memory_recall(&mut messages).await;
4971 let messages = messages;
4972
4973 let auth = Arc::clone(&self.auth);
4976 let client = self.client.clone();
4977 let credential_source = self.credential_source.clone();
4978 let token_cache = self.token_cache.clone();
4979 let model = self.model.clone();
4980 let tools = self.tools.clone();
4981 let system_prompt = self.effective_system_prompt().await;
4982 let thinking_budget = self.thinking_budget;
4983 let reasoning_level = self.reasoning_level();
4984 let watcher_exit_path = self.watcher_exit_path.clone();
4985 let max_tool_output = self.max_tool_output;
4986 let bash_timeout = self.bash_timeout;
4987 let bash_max_timeout = self.bash_max_timeout;
4988 let subagent_timeout = self.subagent_timeout;
4989 let api_retries = self.api_retries;
4990 let refusal_retries = self.refusal_retries;
4991 let session_manager = self.session_manager.clone();
4992 let subagent_registry = self.subagent_registry.clone();
4996 let reaper_registry = Arc::clone(&subagent_registry);
4999 let reaper_orchestration = self.orchestration.clone();
5000 let capture_provider = capture_lease
5001 .as_ref()
5002 .and_then(|lease| self.extension_capture_provider(lease));
5003 let capture_state = self.memory_context_state.clone();
5004 let event_queue = self.event_queue.clone();
5005 let options = api::ApiOptions {
5006 memory_backend: None,
5007 use_1m_context: self.context_window_override == Some(1_000_000),
5008 cache_ttl: self.cache_ttl,
5009 ttl_downgrade_notified: self.ttl_downgrade_notified.clone(),
5010 saw_1h_honored: self.saw_1h_honored.clone(),
5011 credential_source: self.credential_source.clone(),
5012 token_cache: self.token_cache.clone(),
5013 anthropic_base_url: None,
5014 anthropic_stream_client: self.anthropic_stream_client.clone(),
5015 request_compression: self.request_compression.clone(),
5016 anthropic_execution_plan,
5017 codex_request_role: self.codex_request_role(),
5018 thinking_display: self.wire_thinking_display(),
5019 retention_frontier: None,
5022 trace: self.effective_trace_context(),
5023 request_correlation: None,
5024 suppress_stream_deltas: false,
5025 telemetry: self.request_telemetry(),
5026 tool_session_id: Some(self.host_tool_session.clone()),
5029 session_tool_set: None,
5033 request_tools_schema: None,
5034 anthropic_tool_split: None,
5035 usage_counters: None,
5036 cache_anchor: Some(self.cache_anchor.clone()),
5039 keepwarm: None,
5040 cache_diagnosis: self.cache_diagnosis(),
5043 context_rollover: None,
5044 };
5045
5046 let final_capture_history: Arc<Mutex<Option<Vec<crate::SharedMessage>>>> = Arc::new(Mutex::new(None));
5047 let session = crate::runtime::stream::StreamSession {
5048 memory_backend: self.memory_backend.clone(),
5049 memory_context: self.memory_tool_capability(),
5050 final_capture_history: final_capture_history.clone(),
5051 turn_completion: completion,
5052 context_window: self.context_window(),
5053 continuation: self.continuation.clone(),
5054 worker_context: self.worker_context.clone(),
5055 auth,
5056 client,
5057 credential_source,
5058 token_cache,
5059 options,
5060 api_retries,
5061 refusal_retries,
5062 model,
5063 tools,
5064 system_prompt,
5065 thinking_budget,
5066 reasoning_level,
5067 tx: tx.clone(),
5068 cancel,
5069 steering_rx,
5070 watcher_exit_path,
5071 max_tool_output,
5072 tool_spill: self.tool_spill(),
5073 read_dedupe: self.read_dedupe,
5074 tool_keepwarm: self.tool_keepwarm.then(|| tool_keepwarm::ToolKeepWarm::new(
5075 self.trace_ctx.clone(),
5076 tool_keepwarm::interval_override(),
5077 )),
5078 bash_timeout,
5079 bash_max_timeout,
5080 subagent_timeout,
5081 session_manager,
5082 subagent_registry,
5083 event_queue,
5084 secret_prompt,
5085 hook_bus: self.hook_bus.clone(),
5086 session_id: self.session_id.clone(),
5087 cwd: self.cwd.clone(),
5088 env: self.env.clone(),
5089 env_stripped: self.env_stripped.clone(),
5090 env_warned: self.env_warned.clone(),
5091 auto_approve_confirms,
5092 session_allow_all: self.session_allow_all.clone(),
5093 telemetry_level: self.telemetry_level,
5094 orchestration: self.orchestration.clone(),
5095 delegation_parent: self.delegation_parent.clone(),
5096 turn_correlation_id: turn_correlation_id.clone(),
5097 progressive_tool_disclosure: self.progressive_tool_disclosure,
5098 deferred_tool_loading: self.deferred_tool_loading,
5099 activation_confirm: self.activation_confirm,
5100 tool_session_id: self.host_tool_session.clone(),
5101 retained_tool_set: std::sync::Arc::clone(&self.retained_tool_set),
5102 pending_tool_restore: std::sync::Arc::clone(&self.pending_tool_restore),
5103 mcp_runtime: self.mcp_runtime.clone(),
5104 mcp_session_scope: self.mcp_session_scope.clone(),
5105 extension_runtime: self.extension_runtime.clone(),
5106 extension_session_scope: self.extension_session_scope.clone(),
5107 turn_budget: self.turn_budget.clone(),
5108 owner_steer_budget: self.owner_steer_budget.clone(),
5109 message_clock: self.message_clock.clone(),
5110 };
5111
5112 tokio::spawn(async move {
5113 let completed = match StreamMethods::run_stream_internal(session, messages).await {
5114 Ok(()) => true,
5115 Err(e) => {
5116 let _ = tx.send(StreamEvent::Session(SessionEvent::Error(
5117 helpers::turn_error_for(&e, &turn_correlation_id),
5118 )));
5119 false
5120 }
5121 };
5122 dispatch_completed_terminal_capture(
5125 completed,
5126 &final_capture_history,
5127 &capture_state,
5128 capture_lease.zip(capture_provider),
5129 capture_started_at,
5130 capture_turn_start.as_ref(),
5131 );
5132 crate::runtime::subagent::reap_finished(
5136 &reaper_registry,
5137 reaper_orchestration.as_deref(),
5138 );
5139 agent_core::core::memstat::log_turn_memory();
5140 let _ = tx.send(StreamEvent::Session(SessionEvent::Done));
5141 });
5142
5143 Box::pin(tokio_stream::wrappers::ReceiverStream::new(bounded_rx))
5144 }
5145}
5146
5147impl Clone for Runtime {
5148 fn clone(&self) -> Self {
5149 Self {
5150 client: self.client.clone(),
5151 anthropic_stream_client: self.anthropic_stream_client.clone(),
5152 auth: Arc::clone(&self.auth),
5153 model: self.model.clone(),
5154 tools: self.tools.clone(),
5155 system_prompt: self.system_prompt.clone(),
5156 effective_prompt: self.effective_prompt.clone(),
5157 prompt_generation: self.prompt_generation,
5158 prompt_reload_source: self.prompt_reload_source.clone(),
5159 thinking_budget: self.thinking_budget,
5160 named_level: self.named_level,
5161 explicit_reasoning: self.explicit_reasoning,
5162 codex_request_role: self.codex_request_role,
5163 context_window_override: self.context_window_override,
5164 continuation: Arc::clone(&self.continuation),
5165 worker_context: self.worker_context.clone(),
5166 compaction_model: self.compaction_model.clone(),
5167 compaction_mode: self.compaction_mode,
5168 compaction_exclusions: self.compaction_exclusions.clone(),
5169 remote_summarization_attempts: Arc::clone(&self.remote_summarization_attempts),
5170 subagent_registry: self.subagent_registry.clone(),
5171 orchestration: self.orchestration.clone(),
5172 event_queue: self.event_queue.clone(),
5173 watcher_exit_path: self.watcher_exit_path.clone(),
5174 max_tool_output: self.max_tool_output,
5175 tool_spill_policy: self.tool_spill_policy,
5176 tool_spill_root: self.tool_spill_root.clone(),
5177 tool_spill_stats: Arc::clone(&self.tool_spill_stats),
5178 read_dedupe: self.read_dedupe,
5179 thinking_display: self.thinking_display,
5180 tool_keepwarm: self.tool_keepwarm,
5181 bash_timeout: self.bash_timeout,
5182 bash_max_timeout: self.bash_max_timeout,
5183 subagent_timeout: self.subagent_timeout,
5184 api_retries: self.api_retries,
5185 refusal_retries: self.refusal_retries,
5186 telemetry_level: self.telemetry_level,
5187 telemetry_log_generations: self.telemetry_log_generations,
5188 telemetry_writer: self.telemetry_writer.clone(),
5192 trace_ctx: self.trace_ctx.clone(),
5193 trace_controls: Arc::clone(&self.trace_controls),
5194 memory_context_state: std::sync::Arc::clone(&self.memory_context_state),
5200 memory_backend: self.memory_backend.clone(),
5201 memory_backend_config: self.memory_backend_config.clone(),
5202 memory_backend_reconfigure_denied: Arc::clone(&self.memory_backend_reconfigure_denied),
5203 pending_history_import_preview: std::sync::Arc::clone(
5204 &self.pending_history_import_preview,
5205 ),
5206 history_import_plan: std::sync::Arc::clone(&self.history_import_plan),
5207 #[cfg(any(test, feature = "testing"))]
5208 capture_provider_for_test: std::sync::Arc::clone(&self.capture_provider_for_test),
5209 pending_memory_segment: std::sync::Arc::clone(&self.pending_memory_segment),
5212 retained_recall_turn: std::sync::Arc::clone(&self.retained_recall_turn),
5215 one_shot_trace_writer: Arc::clone(&self.one_shot_trace_writer),
5216 capture_dir: self.capture_dir.clone(),
5217 cache_diagnostics: self.cache_diagnostics,
5218 cache_ttl: self.cache_ttl,
5219 request_compression: self.request_compression.clone(),
5220 ttl_downgrade_notified: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
5224 saw_1h_honored: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
5225 cache_anchor: Default::default(),
5226 last_msg_id: Arc::new(Mutex::new(None)),
5229 cache_diag_rejected: Arc::clone(&self.cache_diag_rejected),
5230 session_manager: self.session_manager.clone(),
5231 hook_bus: self.hook_bus.clone(),
5232 reaper_handle: None, reaper_cancel: None, credential_source: self.credential_source.clone(),
5235 token_cache: self.token_cache.clone(), trusted_worker_models: self.trusted_worker_models.clone(),
5237 subagent_limits: self.subagent_limits,
5238 progressive_tool_disclosure: self.progressive_tool_disclosure,
5239 deferred_tool_loading: self.deferred_tool_loading,
5240 activation_confirm: self.activation_confirm,
5241 delegation_parent: self.delegation_parent.clone(),
5242 session_allow_all: self.session_allow_all.clone(),
5243 mcp_runtime: self.mcp_runtime.clone(),
5244 mcp_session_scope: self.mcp_session_scope.clone(),
5247 extension_runtime: self.extension_runtime.clone(),
5248 extension_session_scope: self.extension_session_scope.clone(),
5250 turn_budget: self.turn_budget.clone(),
5251 owner_steer_budget: self.owner_steer_budget.clone(),
5252 message_clock: self.message_clock.clone(),
5253 host_tool_session: self.host_tool_session.clone(),
5258 tool_catalog_prompt: Arc::clone(&self.tool_catalog_prompt),
5259 retained_tool_set: std::sync::Arc::clone(&self.retained_tool_set),
5260 pending_tool_restore: std::sync::Arc::clone(&self.pending_tool_restore),
5261 session_id: self.session_id.clone(),
5263 cwd: self.cwd.clone(),
5264 env: self.env.clone(),
5265 env_stripped: self.env_stripped.clone(),
5266 env_warned: self.env_warned.clone(),
5267 }
5268 }
5269}
5270
5271#[cfg(test)]
5272mod rich_output_validation_tests {
5273 use super::*;
5274
5275 const MODEL: &str = "anthropic/claude-sonnet-4-6";
5276
5277 fn rich_output(summary: &str) -> crate::ToolOutput {
5278 crate::ToolOutput::Blocks {
5279 summary: summary.to_string(),
5280 blocks: vec![json!({
5282 "type": "image",
5283 "source": {
5284 "type": "base64",
5285 "media_type": "image/png",
5286 "data": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+/l+QAAAAASUVORK5CYII="
5287 }
5288 })],
5289 }
5290 }
5291
5292 #[test]
5293 fn unchanged_summary_retains_rich_array_without_text_truncation() {
5294 let (summary, blocks) = validated_single_tool_output(MODEL, rich_output("image"));
5295 assert_eq!(summary, "image");
5296 let expected = blocks.clone().expect("supported image");
5297 assert_eq!(expected[0], json!({"type": "text", "text": summary}));
5298 assert_eq!(expected[1]["type"], "image");
5299 let retained = retain_single_tool_blocks(blocks, &summary, &summary);
5300 assert_eq!(
5301 single_tool_result_content(&summary, retained, summary.len()),
5302 Value::Array(expected)
5303 );
5304 }
5305
5306 #[test]
5307 fn rewritten_or_truncated_summary_drops_rich_blocks() {
5308 let (summary, blocks) = validated_single_tool_output(MODEL, rich_output("image summary"));
5309 assert!(blocks.is_some());
5310 for hooked in [
5311 "redacted".to_string(),
5312 String::new(),
5313 helpers::HelperMethods::truncate_tool_result(&summary, 5),
5314 ] {
5315 let retained = retain_single_tool_blocks(blocks.clone(), &summary, &hooked);
5316 assert!(retained.is_none());
5317 assert_eq!(
5318 single_tool_result_content(&hooked, retained, 5),
5319 Value::String(helpers::HelperMethods::truncate_tool_result(&hooked, 5))
5320 );
5321 }
5322 }
5323
5324 #[test]
5325 fn unsupported_or_malformed_media_becomes_explicit_text() {
5326 let mut malformed = rich_output("private summary");
5327 if let crate::ToolOutput::Blocks { blocks, .. } = &mut malformed {
5328 blocks[0]["source"]["data"] = json!("private invalid payload");
5329 }
5330 for (model, output) in [
5331 (
5332 "google-gemini/unsupported-rich-test",
5333 rich_output("private summary"),
5334 ),
5335 (MODEL, malformed),
5336 ] {
5337 let (summary, blocks) = validated_single_tool_output(model, output);
5338 assert!(summary.starts_with("Attachment not sent: "), "{summary}");
5339 assert!(!summary.contains("private"));
5340 assert!(blocks.is_none());
5341 assert_eq!(
5342 single_tool_result_content(&summary, blocks, 1024),
5343 Value::String(summary)
5344 );
5345 }
5346 }
5347
5348 #[test]
5349 fn plain_text_and_errors_keep_legacy_truncation() {
5350 for text in ["plain output", "tool error", "Unknown tool: missing"] {
5351 let (summary, blocks) = validated_single_tool_output(
5352 "google-gemini/unsupported-rich-test",
5353 crate::ToolOutput::Text(text.to_string()),
5354 );
5355 assert_eq!(summary, text);
5356 assert!(blocks.is_none());
5357 assert_eq!(
5358 single_tool_result_content(&summary, blocks, 5),
5359 Value::String(helpers::HelperMethods::truncate_tool_result(text, 5))
5360 );
5361 }
5362 }
5363}
5364
5365#[cfg(test)]
5366mod tests {
5367 use super::*;
5368
5369 #[test]
5374 fn host_tool_session_shared_by_clones_fresh_per_runtime() {
5375 let rt = Runtime::new_headless();
5376 let clone = rt.clone();
5377 assert_eq!(
5378 rt.host_tool_session_id(),
5379 clone.host_tool_session_id(),
5380 "clones share the live registry and therefore the host session"
5381 );
5382
5383 let other = Runtime::new_headless();
5384 assert_ne!(
5385 rt.host_tool_session_id(),
5386 other.host_tool_session_id(),
5387 "independently constructed runtimes must never share a session"
5388 );
5389 }
5390
5391 #[test]
5395 fn handed_activations_are_carried_on_before_the_first_turn() {
5396 use crate::tools::activation::{CarriedActivation, CarriedProvenance};
5397 let probe = |name: &str| CarriedActivation {
5398 tool_id: format!("builtin:{name}"),
5399 schema_digest: "ab".repeat(32),
5400 provenance: CarriedProvenance::BuiltinRuntime,
5401 };
5402 let rt = Runtime::new_headless();
5403 assert!(rt.carried_tool_activations().is_empty());
5404 rt.restore_tool_activations(vec![probe("subagent_start")]);
5405 rt.restore_tool_activations(Vec::new());
5406 rt.restore_tool_activations(vec![probe("subagent_start"), probe("subagent_status")]);
5407 assert_eq!(
5408 rt.carried_tool_activations(),
5409 vec![probe("subagent_start"), probe("subagent_status")],
5410 "merged without duplicates; an empty hand-off clears nothing"
5411 );
5412 assert_eq!(rt.clone().carried_tool_activations().len(), 2);
5413 assert!(Runtime::new_headless().carried_tool_activations().is_empty());
5414 }
5415
5416 #[test]
5421 fn telemetry_level_gates_shared_observability_sink() {
5422 let mut rt = Runtime::new_headless();
5423 assert!(!rt.trace_context().enabled(), "off by default");
5424 assert!(rt.telemetry_writer().is_none());
5425
5426 rt.set_telemetry_level(crate::runtime::telemetry::TelemetryLevel::Basic);
5427 assert!(rt.trace_context().enabled(), "basic enables the trace sink");
5428 assert!(rt.telemetry_writer().is_some());
5429
5430 let ctx_before = format!("{:?}", rt.trace_context());
5432 rt.set_telemetry_level(crate::runtime::telemetry::TelemetryLevel::Full);
5433 assert_eq!(format!("{:?}", rt.trace_context()), ctx_before);
5434
5435 let clone = rt.clone();
5437 assert!(clone.trace_context().enabled());
5438 assert_eq!(format!("{:?}", clone.trace_context()), ctx_before);
5439 assert!(clone.telemetry_writer().is_some());
5440
5441 rt.set_telemetry_level(crate::runtime::telemetry::TelemetryLevel::Off);
5442 assert!(!rt.trace_context().enabled(), "off disables both");
5443 assert!(rt.telemetry_writer().is_none());
5444 assert!(rt
5445 .shutdown_observability(std::time::Duration::ZERO)
5446 .is_none());
5447 }
5448
5449 #[tokio::test]
5455 async fn shutdown_observability_async_epilogue_contract() {
5456 let rt = Runtime::new_headless();
5458 let budget = crate::runtime::telemetry::DEFAULT_SHUTDOWN_FLUSH_TIMEOUT;
5459 assert!(rt.shutdown_observability_async(budget).await.is_none());
5460 assert!(rt.shutdown_observability_async(budget).await.is_none());
5461
5462 let tmp = tempfile::tempdir().unwrap();
5464 let telemetry_path = tmp.path().join("synaps/api-log.jsonl");
5465 let mut rt = Runtime::new_headless();
5466 let writer = crate::runtime::telemetry::TelemetryWriter::new(
5467 crate::runtime::telemetry::WriterOptions {
5468 telemetry_path: Some(telemetry_path.clone()),
5469 trace_path: Some(tmp.path().join("synaps/request-trace.jsonl")),
5470 ..Default::default()
5471 },
5472 );
5473 rt.install_observability_for_tests(writer.clone());
5474 writer.enqueue_telemetry(crate::runtime::telemetry::TelemetryRecord {
5475 ts: 7,
5476 model: "claude-sonnet-4-6".to_string(),
5477 ..Default::default()
5478 });
5479 let outcome = rt
5480 .shutdown_observability_async(budget)
5481 .await
5482 .expect("writer installed");
5483 assert!(outcome.is_flushed());
5484 assert_eq!(outcome.stats().written, 1);
5485 assert_eq!(
5486 std::fs::read_to_string(&telemetry_path)
5487 .unwrap()
5488 .lines()
5489 .count(),
5490 1,
5491 "queued final record must be on disk after the epilogue flush"
5492 );
5493 assert!(rt
5495 .shutdown_observability_async(budget)
5496 .await
5497 .expect("writer still installed")
5498 .is_flushed());
5499
5500 let mut rt = Runtime::new_headless();
5502 let slow = crate::runtime::telemetry::TelemetryWriter::new(
5503 crate::runtime::telemetry::WriterOptions {
5504 telemetry_path: Some(tmp.path().join("slow/api-log.jsonl")),
5505 trace_path: Some(tmp.path().join("slow/request-trace.jsonl")),
5506 write_delay: Some(std::time::Duration::from_millis(300)),
5507 ..Default::default()
5508 },
5509 );
5510 rt.install_observability_for_tests(slow.clone());
5511 for _ in 0..10 {
5512 slow.enqueue_telemetry(crate::runtime::telemetry::TelemetryRecord::default());
5513 }
5514 let start = std::time::Instant::now();
5515 let outcome = rt
5516 .shutdown_observability_async(std::time::Duration::from_millis(200))
5517 .await
5518 .expect("writer installed");
5519 assert!(!outcome.is_flushed(), "10×300ms cannot drain in 200ms");
5520 assert!(
5523 start.elapsed() < std::time::Duration::from_millis(1500),
5524 "bounded epilogue must not block exit: {:?}",
5525 start.elapsed()
5526 );
5527 }
5528
5529 #[tokio::test]
5534 async fn one_shot_ephemeral_writer_is_retained_and_drained_at_shutdown() {
5535 let rt = Runtime::new_headless();
5536 assert!(!rt.trace_context().enabled(), "telemetry off");
5537 rt.trace_arm_next(false);
5538 let armed = rt.effective_trace_context();
5539 assert!(armed.enabled(), "armed ephemeral context traces");
5540 let budget = crate::runtime::telemetry::DEFAULT_SHUTDOWN_FLUSH_TIMEOUT;
5543 let outcome = rt
5544 .shutdown_observability_async(budget)
5545 .await
5546 .expect("retained one-shot writer must be drained by the epilogue");
5547 assert!(outcome.is_flushed());
5548 assert!(rt.shutdown_observability_async(budget).await.is_none());
5550 }
5551
5552 #[tokio::test]
5558 #[serial_test::serial(synaps_base_dir)]
5559 async fn startup_and_shutdown_paths_sweep_expired_captures() {
5560 struct BaseDirGuard(Option<String>);
5561 impl Drop for BaseDirGuard {
5562 fn drop(&mut self) {
5563 match self.0.take() {
5564 Some(old) => std::env::set_var("SYNAPS_BASE_DIR", old),
5565 None => std::env::remove_var("SYNAPS_BASE_DIR"),
5566 }
5567 }
5568 }
5569 let tmp = tempfile::tempdir().unwrap();
5570 let guard = BaseDirGuard(std::env::var("SYNAPS_BASE_DIR").ok());
5571 std::env::set_var("SYNAPS_BASE_DIR", tmp.path());
5572
5573 let cap_dir = trace::default_capture_dir();
5574 assert!(
5575 cap_dir.starts_with(tmp.path()),
5576 "capture dir must be private to the test"
5577 );
5578 let stale_id = trace::TraceId::new("req-stale").unwrap();
5579 let plant_stale = || {
5580 agent_core::core::private_fs::ensure_private_dir(&cap_dir).unwrap();
5581 let stale = trace::controls::ContentCaptureBundle {
5582 schema: trace::CONTENT_CAPTURE_SCHEMA.to_string(),
5583 request_id: stale_id.clone(),
5584 created_unix_ms: 1_000,
5585 expires_unix_ms: 2_000, redacted: true,
5587 over_budget: false,
5588 body: Some(serde_json::json!({"old": true})),
5589 };
5590 let path = trace::controls::capture_path(&cap_dir, &stale_id);
5591 std::fs::write(&path, serde_json::to_vec(&stale).unwrap()).unwrap();
5592 path
5593 };
5594
5595 let stale_path = plant_stale();
5597 let rt = Runtime::new().await.expect("runtime constructs");
5598 assert!(
5599 !stale_path.exists(),
5600 "Runtime startup must physically remove expired capture bundles"
5601 );
5602
5603 let stale_path = plant_stale();
5605 let _ = rt.shutdown_observability(std::time::Duration::ZERO);
5606 assert!(
5607 !stale_path.exists(),
5608 "sync shutdown epilogue must remove expired capture bundles"
5609 );
5610
5611 let stale_path = plant_stale();
5613 let _ = rt
5614 .shutdown_observability_async(std::time::Duration::from_millis(100))
5615 .await;
5616 assert!(
5617 !stale_path.exists(),
5618 "async shutdown epilogue must remove expired capture bundles"
5619 );
5620 drop(guard);
5621 }
5622
5623 #[tokio::test]
5628 #[serial_test::serial(synaps_base_dir)]
5629 async fn capture_root_binds_at_construction_and_survives_env_churn() {
5630 let base_a = tempfile::tempdir().unwrap();
5631 let base_b = tempfile::tempdir().unwrap();
5632 let old = std::env::var("SYNAPS_BASE_DIR").ok();
5633
5634 std::env::set_var("SYNAPS_BASE_DIR", base_a.path());
5635 let rt_a = Runtime::new().await.expect("runtime A");
5636 std::env::set_var("SYNAPS_BASE_DIR", base_b.path());
5637 let rt_b = Runtime::new().await.expect("runtime B");
5638
5639 let cap_a = base_a.path().join("trace").join("capture");
5640 let cap_b = base_b.path().join("trace").join("capture");
5641 let plant = |dir: &std::path::Path, name: &str| {
5642 agent_core::core::private_fs::ensure_private_dir(dir).unwrap();
5643 let id = trace::TraceId::new(name).unwrap();
5644 let stale = trace::controls::ContentCaptureBundle {
5645 schema: trace::CONTENT_CAPTURE_SCHEMA.to_string(),
5646 request_id: id.clone(),
5647 created_unix_ms: 1_000,
5648 expires_unix_ms: 2_000,
5649 redacted: true,
5650 over_budget: false,
5651 body: Some(serde_json::json!({"old": true})),
5652 };
5653 let path = trace::controls::capture_path(dir, &id);
5654 std::fs::write(&path, serde_json::to_vec(&stale).unwrap()).unwrap();
5655 path
5656 };
5657
5658 let decoy = tempfile::tempdir().unwrap();
5660 std::env::set_var("SYNAPS_BASE_DIR", decoy.path());
5661
5662 let stale_a = plant(&cap_a, "req-stale-a");
5663 let stale_b = plant(&cap_b, "req-stale-b");
5664 let _ = rt_a.shutdown_observability(std::time::Duration::ZERO);
5665 assert!(
5666 !stale_a.exists(),
5667 "runtime A must sweep the root bound at ITS construction"
5668 );
5669 assert!(
5670 stale_b.exists(),
5671 "runtime A must never sweep another runtime's root"
5672 );
5673 let _ = rt_b
5674 .shutdown_observability_async(std::time::Duration::from_millis(100))
5675 .await;
5676 assert!(
5677 !stale_b.exists(),
5678 "runtime B must sweep the root bound at ITS construction"
5679 );
5680
5681 match old {
5682 Some(v) => std::env::set_var("SYNAPS_BASE_DIR", v),
5683 None => std::env::remove_var("SYNAPS_BASE_DIR"),
5684 }
5685 }
5686
5687 #[test]
5688 fn reload_rejects_policy_change_without_partial_apply() {
5689 let dir = tempfile::tempdir().unwrap();
5690 let path = dir.path().join("prompt.yaml");
5691 let manifest = |total: &str, content: &str| {
5692 format!(
5693 "schema: synaps-prompt/1\nkernel: kernel\nmodules:\n - id: kernel\n version: v1\n source: builtin\n priority: 0\n selectors: {{}}\n mutability: mutable_guidance\n content: {content}\npolicies:\n delegation:\n mode: enforced\n allowed_models: [anthropic/claude-sonnet-4-6]\n max_concurrent_workers: 1\n max_total_workers: {total}\n"
5694 )
5695 };
5696 std::fs::write(&path, manifest("1", "before")).unwrap();
5697 let parsed =
5698 agent_core::prompt::PromptManifest::parse(&std::fs::read_to_string(&path).unwrap())
5699 .unwrap();
5700 let model =
5701 agent_core::prompt::QualifiedModelId::parse("anthropic/claude-sonnet-4-6").unwrap();
5702 let context = agent_core::prompt::SelectionContext::new(model.clone(), None).unwrap();
5703 let stack = agent_core::prompt::compile_prompt_stack(
5704 &parsed,
5705 &parsed.registry(path.parent()).unwrap(),
5706 &context,
5707 None,
5708 )
5709 .unwrap();
5710 let catalog = crate::orchestration::OrchestrationRuntime::trusted_catalog(
5711 &model,
5712 parsed.delegation_catalog_candidates(),
5713 )
5714 .unwrap();
5715 let digest = parsed
5716 .delegation_policy(model, &catalog)
5717 .unwrap()
5718 .map(|p| p.digest());
5719 let mut runtime = Runtime::new_headless();
5720 runtime.apply_prompt_stack(stack).unwrap();
5721 runtime.retain_prompt_reload_source(path.clone(), context, None, digest.clone());
5722 let generation = runtime.prompt_generation();
5723 let composed = runtime.effective_prompt().unwrap().composed().to_owned();
5724
5725 std::fs::write(&path, manifest("2", "after")).unwrap();
5726 let error = runtime.reload_prompt().unwrap_err().to_string();
5727 assert!(error.contains("cannot safely change delegation policy"));
5728 assert_eq!(runtime.prompt_generation(), generation);
5729 assert_eq!(runtime.effective_prompt().unwrap().composed(), composed);
5730 assert_eq!(
5731 runtime
5732 .prompt_reload_source
5733 .as_ref()
5734 .unwrap()
5735 .delegation_policy_digest,
5736 digest
5737 );
5738 }
5739
5740 #[test]
5741 fn worker_install_does_not_replay_process_global_favorite_grants() {
5742 let foreground =
5743 crate::orchestration::canonical_foreground_identity("anthropic/claude-fable-5")
5744 .unwrap();
5745 let session = Arc::new(
5746 crate::orchestration::OrchestrationRuntime::baseline(foreground, 4, 8).unwrap(),
5747 );
5748 let favorite = "openai-codex/gpt-5.6-sol";
5749 assert!(
5750 session.preflight(favorite).is_err(),
5751 "precondition: the parent session did not grant this identity"
5752 );
5753
5754 let mut worker = Runtime::new_headless();
5755 worker.apply_config(&crate::config::SynapsConfig {
5756 favorite_models: vec![favorite.to_owned()],
5757 ..Default::default()
5758 });
5759 worker.install_worker_orchestration(Arc::clone(&session));
5760
5761 assert!(
5762 session.preflight(favorite).is_err(),
5763 "constructing a child must not mint/replay a favorite-model grant"
5764 );
5765 assert!(!session.effective_choices().contains(&favorite.to_owned()));
5766 }
5767
5768 #[test]
5773 #[serial_test::serial(synaps_base_dir)]
5774 fn tool_spill_follows_config_and_is_keyed_by_the_conversation() {
5775 let base = crate::test_env::BaseDirGuard::new();
5776 let mut runtime = Runtime::new_headless();
5777 assert!(!runtime.tool_spill_policy().is_enabled());
5778 assert!(!runtime.tool_spill().is_enabled());
5779 runtime.apply_config(&crate::config::SynapsConfig::default());
5780 assert_eq!(
5781 runtime.tool_spill_policy(),
5782 tool_spill::ToolSpillPolicy {
5783 threshold_bytes: 12 * 1024,
5784 head_bytes: 4 * 1024,
5785 tail_bytes: 3 * 1024,
5786 }
5787 );
5788 runtime.set_session_id(Some("20261029-120000-ab12".into()));
5789 assert_eq!(
5790 runtime.tool_spill().dir(),
5791 Some(base.path().join("spill").join("20261029-120000-ab12").as_path())
5792 );
5793 let clone = runtime.clone();
5795 let text = "line\n".repeat(4_000);
5796 let virtualised = clone
5797 .tool_spill()
5798 .virtualise("toolu_1", &tool_spill::SpillTarget::Spill, &text)
5799 .expect("over the threshold");
5800 assert_eq!(runtime.tool_spill_totals().virtualised, 1);
5801 assert_eq!(
5802 runtime.tool_spill_totals().kept_out_bytes(),
5803 (text.len() - virtualised.preview.len()) as u64
5804 );
5805 runtime.apply_config(&agent_core::config::load_config_from_str(
5806 "tool_output_spill_bytes = 0\n",
5807 ));
5808 assert!(!runtime.tool_spill().is_enabled(), "0 turns it off");
5809 }
5810
5811 #[test]
5812 fn read_dedupe_is_off_until_configured_then_follows_config() {
5813 let mut runtime = Runtime::new_headless();
5814 assert!(!runtime.read_dedupe(), "off until a config is applied");
5815 runtime.apply_config(&agent_core::config::load_config_from_str(""));
5816 assert!(runtime.read_dedupe(), "the config default is on");
5817 assert!(runtime.clone().read_dedupe(), "clones keep it");
5818 runtime.apply_config(&agent_core::config::load_config_from_str("read_dedupe = off\n"));
5819 assert!(!runtime.read_dedupe());
5820 }
5821
5822 #[test]
5826 fn thinking_display_is_foreground_only() {
5827 use crate::config::ThinkingDisplay;
5828 use crate::runtime::openai::catalog::CodexRequestRole;
5829 let mut runtime = Runtime::new_headless();
5830 assert_eq!(runtime.wire_thinking_display(), ThinkingDisplay::Summarized);
5831 runtime.apply_config(&agent_core::config::load_config_from_str("thinking_display = omitted\n"));
5832 assert_eq!(runtime.thinking_display(), ThinkingDisplay::Omitted);
5833 assert_eq!(runtime.wire_thinking_display(), ThinkingDisplay::Omitted);
5834 assert_eq!(runtime.clone().wire_thinking_display(), ThinkingDisplay::Omitted);
5835
5836 let mut worker = runtime.clone();
5838 worker.set_codex_request_role(CodexRequestRole::Worker);
5839 assert_eq!(worker.thinking_display(), ThinkingDisplay::Omitted, "config kept");
5840 assert_eq!(worker.wire_thinking_display(), ThinkingDisplay::Summarized, "wire forced");
5841 let mut internal = runtime.clone();
5842 internal.set_codex_request_role(CodexRequestRole::Internal);
5843 assert_eq!(internal.wire_thinking_display(), ThinkingDisplay::Summarized);
5844
5845 runtime.apply_config(&agent_core::config::load_config_from_str(""));
5846 assert_eq!(runtime.wire_thinking_display(), ThinkingDisplay::Summarized, "default");
5847 }
5848
5849 #[test]
5850 fn tool_keepwarm_is_off_until_configured_then_follows_config() {
5851 let mut runtime = Runtime::new_headless();
5852 assert!(!runtime.tool_keepwarm(), "off until a config is applied");
5853 runtime.apply_config(&agent_core::config::load_config_from_str(""));
5854 assert!(runtime.tool_keepwarm(), "the config default is on");
5855 assert!(runtime.clone().tool_keepwarm(), "clones keep it");
5856 runtime.apply_config(&agent_core::config::load_config_from_str(
5857 "cache_keepwarm_tools = off\n",
5858 ));
5859 assert!(!runtime.tool_keepwarm());
5860 }
5861
5862 #[test]
5863 fn host_install_may_apply_explicit_operator_favorites() {
5864 let mut runtime = Runtime::new_headless();
5865 let config = crate::config::SynapsConfig {
5866 favorite_models: vec![
5867 "openai-codex/gpt-5.6-luna".to_owned(),
5868 "anthropic/claude-opus-4-6".to_owned(),
5869 "claude/claude-fable-5".to_owned(),
5871 "not-qualified".to_owned(),
5873 ],
5874 ..Default::default()
5875 };
5876 runtime.apply_config(&config);
5877 let foreground =
5878 agent_core::prompt::QualifiedModelId::parse("openai-codex/gpt-5.6-sol").unwrap();
5879 runtime.install_orchestration(std::sync::Arc::new(
5880 crate::orchestration::OrchestrationRuntime::baseline(foreground, 3, 8).unwrap(),
5881 ));
5882
5883 let orchestration = runtime.orchestration().unwrap();
5884 for trusted in [
5885 "openai-codex/gpt-5.6-sol",
5886 "openai-codex/gpt-5.6-luna",
5887 "anthropic/claude-opus-4-6",
5888 "anthropic/claude-fable-5",
5889 ] {
5890 assert!(
5891 orchestration
5892 .effective_choices()
5893 .contains(&trusted.to_owned()),
5894 "missing configured worker choice {trusted}: {:?}",
5895 orchestration.effective_choices()
5896 );
5897 }
5898 let authorized = orchestration
5899 .resolve_and_authorize("sa_cross_provider", Some("anthropic/claude-opus-4-6"))
5900 .expect("an explicitly favorited cross-provider model must be authorized");
5901 assert_eq!(authorized.model.as_str(), "anthropic/claude-opus-4-6");
5902 assert!(!orchestration
5903 .effective_choices()
5904 .contains(&"not-qualified".to_owned()));
5905 }
5906
5907 #[test]
5908 fn configured_worker_choices_survive_manifestless_foreground_changes() {
5909 let mut runtime = Runtime::new_headless();
5910 runtime.apply_config(&crate::config::SynapsConfig {
5911 favorite_models: vec!["anthropic/claude-opus-4-6".to_owned()],
5912 ..Default::default()
5913 });
5914 let foreground =
5915 agent_core::prompt::QualifiedModelId::parse("openai-codex/gpt-5.6-sol").unwrap();
5916 runtime.install_orchestration(std::sync::Arc::new(
5917 crate::orchestration::OrchestrationRuntime::baseline(foreground, 3, 8).unwrap(),
5918 ));
5919
5920 runtime
5921 .try_set_model("xai-auth/grok-4.5-latest".to_owned())
5922 .unwrap();
5923
5924 let orchestration = runtime.orchestration().unwrap();
5925 assert_eq!(orchestration.foreground_model(), "xai-auth/grok-4.5-latest");
5926 assert!(orchestration
5927 .effective_choices()
5928 .contains(&"anthropic/claude-opus-4-6".to_owned()));
5929 orchestration
5930 .resolve_and_authorize("sa_after_switch", Some("anthropic/claude-opus-4-6"))
5931 .expect("configured worker trust must survive policy replacement");
5932 }
5933
5934 #[test]
5935 fn configured_subagent_limits_size_the_policy_rebuilt_on_model_change() {
5936 let mut runtime = Runtime::new_headless();
5937 assert_eq!(
5938 runtime.subagent_limits(),
5939 crate::config::SubagentLimits::default(),
5940 "defaults before any config"
5941 );
5942 runtime.apply_config(&crate::config::SynapsConfig {
5943 subagent: crate::config::SubagentLimits {
5944 max_concurrent: 24,
5945 max_total: 96,
5946 max_depth: 2,
5947 },
5948 ..Default::default()
5949 });
5950 let foreground =
5951 agent_core::prompt::QualifiedModelId::parse("openai-codex/gpt-5.6-sol").unwrap();
5952 runtime.install_orchestration(std::sync::Arc::new(
5953 crate::orchestration::OrchestrationRuntime::from_limits(
5954 foreground,
5955 runtime.subagent_limits(),
5956 )
5957 .unwrap(),
5958 ));
5959 runtime
5960 .try_set_model("xai-auth/grok-4.5-latest".to_owned())
5961 .unwrap();
5962
5963 let orchestration = runtime.orchestration().unwrap();
5964 assert_eq!(orchestration.foreground_model(), "xai-auth/grok-4.5-latest");
5965 assert_eq!(
5966 orchestration.ultracode_readiness("xai-auth/grok-4.5-latest"),
5967 Ok((24, 96)),
5968 "the replacement policy uses the configured limits, not a hardcoded 8/64"
5969 );
5970 assert_eq!(orchestration.reserve_delegation("a", None), Ok(1));
5971 assert_eq!(orchestration.reserve_delegation("b", Some("a")), Ok(2));
5972 assert_eq!(
5973 orchestration.reserve_delegation("c", Some("b")),
5974 Err(crate::orchestration::DelegationTreeDenied::DepthLimit { max_depth: 2 })
5975 );
5976 }
5977
5978 #[test]
5979 fn set_model_replaces_manifestless_orchestration_foreground_snapshot() {
5980 let mut runtime = Runtime::new_headless();
5981 let glm = agent_core::prompt::QualifiedModelId::parse("openrouter/z-ai/glm-5.2").unwrap();
5982 runtime.install_orchestration(std::sync::Arc::new(
5983 crate::orchestration::OrchestrationRuntime::baseline(glm, 3, 8).unwrap(),
5984 ));
5985
5986 runtime.set_model("openrouter/deepseek/deepseek-v4-pro".to_owned());
5987
5988 let orchestration = runtime.orchestration().unwrap();
5989 assert_eq!(
5990 orchestration.foreground_model(),
5991 "openrouter/deepseek/deepseek-v4-pro"
5992 );
5993 assert!(orchestration
5994 .effective_choices()
5995 .contains(&"openrouter/moonshotai/kimi-k2.7-code".to_owned()));
5996 assert!(orchestration
5997 .resolve_and_authorize("sa_kimi", Some("openrouter/moonshotai/kimi-k2.7-code"))
5998 .is_ok());
5999 }
6000
6001 #[test]
6006 fn set_model_to_uncached_muse_id_rebuilds_orchestration() {
6007 let _guard = crate::runtime::openai::catalog::muse::test_cache_guard();
6008 crate::runtime::openai::catalog::capability_cache::replace_provider("muse", &[]);
6009 let mut runtime = Runtime::new_headless();
6010 let glm = agent_core::prompt::QualifiedModelId::parse("openrouter/z-ai/glm-5.2").unwrap();
6011 runtime.install_orchestration(std::sync::Arc::new(
6012 crate::orchestration::OrchestrationRuntime::baseline(glm, 3, 8).unwrap(),
6013 ));
6014
6015 for model in ["muse/muse-wonder-1.0", "muse/muse-live-only-zz5"] {
6016 runtime
6017 .try_set_model(model.to_owned())
6018 .unwrap_or_else(|e| panic!("{model}: {e}"));
6019 assert_eq!(runtime.model(), model);
6020 assert_eq!(runtime.orchestration().unwrap().foreground_model(), model);
6021 }
6022 assert!(runtime.try_set_model("muse/Bad Id".to_owned()).is_err());
6024 assert_eq!(runtime.model(), "muse/muse-live-only-zz5");
6025 }
6026
6027 #[test]
6028 fn set_model_refuses_to_orphan_unreconciled_policy_workers() {
6029 let mut runtime = Runtime::new_headless();
6030 let glm = agent_core::prompt::QualifiedModelId::parse("openrouter/z-ai/glm-5.2").unwrap();
6031 let orchestration = std::sync::Arc::new(
6032 crate::orchestration::OrchestrationRuntime::baseline(glm, 3, 8).unwrap(),
6033 );
6034 orchestration
6035 .resolve_and_authorize("sa_active", None)
6036 .unwrap();
6037 runtime.install_orchestration(orchestration.clone());
6038 runtime.model = "openrouter/z-ai/glm-5.2".to_owned();
6039
6040 let error = runtime
6041 .try_set_model("openrouter/deepseek/deepseek-v4-pro".to_owned())
6042 .unwrap_err();
6043
6044 assert!(error.contains("collection/reconciliation"));
6045 assert_eq!(runtime.model(), "openrouter/z-ai/glm-5.2");
6046 assert!(std::sync::Arc::ptr_eq(
6047 runtime.orchestration().unwrap(),
6048 &orchestration
6049 ));
6050 assert!(orchestration.is_unreconciled("sa_active"));
6051 }
6052
6053 #[test]
6054 fn set_model_keeps_active_typed_manifest_orchestration_unchanged() {
6055 let dir = tempfile::tempdir().unwrap();
6056 let manifest_path = dir.path().join("prompt.yaml");
6057 let glm = agent_core::prompt::QualifiedModelId::parse("openrouter/z-ai/glm-5.2").unwrap();
6058 let mut runtime = Runtime::new_headless();
6059 let orchestration = std::sync::Arc::new(
6060 crate::orchestration::OrchestrationRuntime::baseline(glm.clone(), 3, 8).unwrap(),
6061 );
6062 runtime.install_orchestration(orchestration.clone());
6063 runtime.retain_prompt_reload_source(
6064 manifest_path,
6065 agent_core::prompt::SelectionContext::new(glm, None).unwrap(),
6066 None,
6067 None,
6068 );
6069
6070 runtime.set_model("openrouter/deepseek/deepseek-v4-pro".to_owned());
6071
6072 assert!(std::sync::Arc::ptr_eq(
6073 runtime.orchestration().unwrap(),
6074 &orchestration
6075 ));
6076 assert_eq!(
6077 runtime.orchestration().unwrap().foreground_model(),
6078 "openrouter/z-ai/glm-5.2"
6079 );
6080 }
6081
6082 #[test]
6083 fn set_model_preserves_qualified_provider_ids_and_bare_claude() {
6084 let cases = [
6085 (
6086 "github-copilot/claude-opus-4.8",
6087 "github-copilot/claude-opus-4.8",
6088 ),
6089 ("anthropic/claude-sonnet-4-6", "anthropic/claude-sonnet-4-6"),
6090 ("google/gemini-2.5-pro", "google/gemini-2.5-pro"),
6091 ("openai/gpt-5", "openai/gpt-5"),
6092 ("claude-opus-4-8", "claude-opus-4-8"),
6093 ];
6094
6095 for (input, expected) in cases {
6096 let mut runtime = Runtime::new_headless();
6097 runtime.set_model(input.to_owned());
6098 assert_eq!(runtime.model(), expected, "input: {input}");
6099 }
6100 }
6101
6102 #[test]
6103 fn set_model_strips_only_legacy_health_status_decoration() {
6104 let cases = [
6105 ("✅ 339ms groq/llama-3.3-70b", "groq/llama-3.3-70b"),
6106 (
6107 "✅ 339ms github-copilot/claude-opus-4.8",
6108 "github-copilot/claude-opus-4.8",
6109 ),
6110 (
6111 "⚠️ 1200ms anthropic/claude-sonnet-4-6",
6112 "anthropic/claude-sonnet-4-6",
6113 ),
6114 ];
6115
6116 for (input, expected) in cases {
6117 let mut runtime = Runtime::new_headless();
6118 runtime.set_model(input.to_owned());
6119 assert_eq!(runtime.model(), expected, "input: {input}");
6120 }
6121 }
6122
6123 #[tokio::test]
6124 async fn confirm_without_prompt_fails_closed() {
6125 let result = resolve_before_tool_call_result(
6126 crate::extensions::hooks::events::HookResult::confirm("Run deploy?"),
6127 None,
6128 false,
6129 None,
6130 )
6131 .await;
6132
6133 assert!(matches!(
6134 result,
6135 crate::extensions::hooks::events::HookResult::Block { reason }
6136 if reason.contains("requires confirmation") && reason.contains("Run deploy?")
6137 ));
6138 }
6139
6140 #[tokio::test]
6141 async fn modify_result_replaces_tool_input() {
6142 let result = resolve_before_tool_call_decision(
6143 serde_json::json!({"command":"rm -rf /"}),
6144 crate::extensions::hooks::events::HookResult::Modify {
6145 input: serde_json::json!({"command":"echo safe"}),
6146 },
6147 None,
6148 false,
6149 None,
6150 )
6151 .await;
6152
6153 match result {
6154 BeforeToolCallDecision::Continue { input } => {
6155 assert_eq!(input, serde_json::json!({"command":"echo safe"}));
6156 }
6157 BeforeToolCallDecision::Block { reason } => panic!("unexpected block: {reason}"),
6158 }
6159 }
6160
6161 #[tokio::test]
6162 async fn confirm_prompt_yes_continues() {
6163 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
6164 let handle = crate::tools::SecretPromptHandle::new(tx);
6165
6166 let task = tokio::spawn(async move {
6167 let request = rx.recv().await.expect("confirm prompt request");
6168 assert_eq!(request.title, "Confirm tool call");
6169 assert!(request.prompt.contains("Run deploy?"));
6170 let _ = request.response_tx.send(Some("yes".to_string()));
6171 });
6172
6173 let result = resolve_before_tool_call_result(
6174 crate::extensions::hooks::events::HookResult::confirm("Run deploy?"),
6175 Some(&handle),
6176 false,
6177 None,
6178 )
6179 .await;
6180
6181 task.await.unwrap();
6182 assert!(matches!(
6183 result,
6184 crate::extensions::hooks::events::HookResult::Continue
6185 ));
6186 }
6187
6188 #[tokio::test]
6189 async fn confirm_prompt_non_yes_blocks() {
6190 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
6191 let handle = crate::tools::SecretPromptHandle::new(tx);
6192
6193 let task = tokio::spawn(async move {
6194 let request = rx.recv().await.expect("confirm prompt request");
6195 let _ = request.response_tx.send(Some("no".to_string()));
6196 });
6197
6198 let result = resolve_before_tool_call_result(
6199 crate::extensions::hooks::events::HookResult::confirm("Run deploy?"),
6200 Some(&handle),
6201 false,
6202 None,
6203 )
6204 .await;
6205
6206 task.await.unwrap();
6207 assert!(matches!(
6208 result,
6209 crate::extensions::hooks::events::HookResult::Block { reason }
6210 if reason.contains("confirmation denied")
6211 ));
6212 }
6213
6214 fn ext(ids: &[&str]) -> Vec<String> {
6215 ids.iter().map(|s| s.to_string()).collect()
6216 }
6217
6218 fn key(ids: &[&str]) -> Vec<crate::tools::LatchKey> {
6219 ids.iter().map(|s| (s.to_string(), 0)).collect()
6220 }
6221
6222 fn sourced_confirm(
6223 message: &str,
6224 source: &str,
6225 tool: &str,
6226 ) -> crate::extensions::hooks::events::HookResult {
6227 crate::extensions::hooks::events::HookResult::Confirm {
6228 message: message.into(),
6229 sources: ext(&[source]),
6230 tool: Some(tool.into()),
6231 generations: vec![],
6232 modified_input: None,
6233 modified_by: vec![],
6234 }
6235 }
6236
6237 #[test]
6238 fn session_allow_all_clones_share_but_independent_runtimes_do_not() {
6239 use std::sync::Arc;
6240 let parent = Runtime::new_headless();
6241 let cloned = parent.clone();
6242 let independent = Runtime::new_headless();
6243 assert!(Arc::ptr_eq(
6244 parent.session_allow_all(),
6245 cloned.session_allow_all()
6246 ));
6247 assert!(!Arc::ptr_eq(
6248 parent.session_allow_all(),
6249 independent.session_allow_all()
6250 ));
6251 cloned.session_allow_all().allow(&key(&["jev"]));
6252 assert!(parent.session_allow_all().allows_all(&key(&["jev"])));
6253 assert!(!independent.session_allow_all().allows_all(&key(&["jev"])));
6254 }
6255
6256 #[test]
6257 fn session_allow_all_conversation_boundary_isolates_old_clones() {
6258 use std::sync::Arc;
6259 let mut runtime = Runtime::new_headless();
6260 let old_worker = runtime.clone();
6261 old_worker.session_allow_all().allow(&key(&["jev"]));
6262
6263 runtime.begin_conversation(Some("next".into()));
6264 assert_eq!(runtime.session_id(), Some("next"));
6265 assert!(runtime.session_allow_all().is_empty());
6266 assert!(!Arc::ptr_eq(
6267 runtime.session_allow_all(),
6268 old_worker.session_allow_all()
6269 ));
6270 assert!(
6273 old_worker.session_allow_all().is_empty(),
6274 "old latch must be cleared, not just replaced"
6275 );
6276 assert!(!old_worker.session_allow_all().allow(&key(&["jev"])));
6279 assert!(old_worker.session_allow_all().is_empty());
6280 assert!(runtime.session_allow_all().is_empty());
6281
6282 let new_worker = runtime.clone();
6283 new_worker.session_allow_all().allow(&key(&["jev"]));
6284 assert!(runtime.session_allow_all().allows_all(&key(&["jev"])));
6285 runtime.begin_conversation(Some("next".into()));
6287 assert!(runtime.session_allow_all().is_empty());
6288 assert!(!Arc::ptr_eq(
6289 runtime.session_allow_all(),
6290 new_worker.session_allow_all()
6291 ));
6292 }
6293
6294 #[test]
6295 fn conversation_boundary_resets_the_cache_diagnosis_chain() {
6296 let mut runtime = Runtime::new_headless();
6297 runtime.set_cache_diagnostics(true);
6298 let mut driver = runtime.clone();
6299 driver.share_ttl_latches(&runtime);
6300 runtime.cache_diagnosis().unwrap().record("msg_old");
6301 runtime.set_session_id(Some("successor".into()));
6303 assert_eq!(
6304 runtime
6305 .cache_diagnosis()
6306 .unwrap()
6307 .previous_message_id()
6308 .as_deref(),
6309 Some("msg_old")
6310 );
6311 runtime.begin_conversation(Some("next".into()));
6313 assert_eq!(
6314 runtime.cache_diagnosis().unwrap().previous_message_id(),
6315 None
6316 );
6317 assert_eq!(
6318 driver.cache_diagnosis().unwrap().previous_message_id(),
6319 None
6320 );
6321 }
6322
6323 #[test]
6324 fn session_allow_all_identity_change_preserves_compaction_consent() {
6325 use std::sync::Arc;
6326 let mut runtime = Runtime::new_headless();
6327 let worker = runtime.clone();
6328 worker.session_allow_all().allow(&key(&["jev"]));
6329 runtime.set_session_id(Some("linked-successor".into()));
6330 assert!(Arc::ptr_eq(
6331 runtime.session_allow_all(),
6332 worker.session_allow_all()
6333 ));
6334 assert!(runtime.session_allow_all().allows_all(&key(&["jev"])));
6335 }
6336
6337 #[tokio::test]
6341 async fn confirm_prompt_always_latches_only_the_source_extension() {
6342 let latch = crate::tools::SessionConfirmLatch::new();
6343 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
6344 let handle = crate::tools::SecretPromptHandle::new(tx);
6345
6346 let task = tokio::spawn(async move {
6347 let first = rx.recv().await.expect("first confirm prompt");
6348 assert_eq!(
6350 first.confirm_scope,
6351 Some(crate::session::ConfirmScope {
6352 extensions: vec!["jev".into()],
6353 tool: Some("bash".into()),
6354 rewritten_by: vec![],
6355 })
6356 );
6357 assert!(first.prompt.starts_with("Extension: jev\nTool: bash\n"), "{}", first.prompt);
6358 let _ = first
6359 .response_tx
6360 .send(Some(CONFIRM_ANSWER_ALLOW_ALL.to_string()));
6361 let other = rx.recv().await.expect("other extension still prompts");
6364 assert_eq!(
6365 other.confirm_scope.as_ref().map(|s| s.extensions.clone()),
6366 Some(vec!["other".to_string()])
6367 );
6368 let _ = other.response_tx.send(None);
6369 assert!(rx.recv().await.is_none(), "no further prompts");
6370 });
6371
6372 let first = resolve_before_tool_call_result(
6373 sourced_confirm("Run deploy?", "jev", "bash"),
6374 Some(&handle),
6375 false,
6376 Some(&latch),
6377 )
6378 .await;
6379 assert!(matches!(
6380 first,
6381 crate::extensions::hooks::events::HookResult::Continue
6382 ));
6383 assert_eq!(latch.extensions(), ext(&["jev"]), "latch keyed by source");
6384
6385 let second = resolve_before_tool_call_result(
6386 sourced_confirm("Run deploy again?", "jev", "write"),
6387 Some(&handle),
6388 false,
6389 Some(&latch),
6390 )
6391 .await;
6392 assert!(matches!(
6393 second,
6394 crate::extensions::hooks::events::HookResult::Continue
6395 ));
6396
6397 let other = resolve_before_tool_call_result(
6398 sourced_confirm("Run ls?", "other", "bash"),
6399 Some(&handle),
6400 false,
6401 Some(&latch),
6402 )
6403 .await;
6404 assert!(
6405 matches!(other, crate::extensions::hooks::events::HookResult::Block { .. }),
6406 "another extension's confirm is not covered by jev's latch"
6407 );
6408 drop(handle);
6409 task.await.unwrap();
6410 }
6411
6412 #[tokio::test]
6415 async fn latched_plus_unlatched_source_prompts_and_allow_all_latches_both() {
6416 let latch = crate::tools::SessionConfirmLatch::new();
6417 latch.allow(&key(&["a"]));
6418 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
6419 let handle = crate::tools::SecretPromptHandle::new(tx);
6420 let task = tokio::spawn(async move {
6421 let req = rx.recv().await.expect("must prompt: b is not latched");
6422 assert_eq!(
6423 req.confirm_scope.as_ref().map(|s| s.extensions.clone()),
6424 Some(vec!["a".to_string(), "b".to_string()])
6425 );
6426 assert!(req.prompt.starts_with("Extension: a, b\n"), "{}", req.prompt);
6427 let _ = req.response_tx.send(Some(CONFIRM_ANSWER_ALLOW_ALL.to_string()));
6428 });
6429 let merged = || crate::extensions::hooks::events::HookResult::Confirm {
6430 message: "[a] x\n\n[b] y".into(),
6431 sources: ext(&["a", "b"]),
6432 generations: vec![],
6433 tool: Some("bash".into()),
6434 modified_input: None,
6435 modified_by: vec![],
6436 };
6437 let first = resolve_before_tool_call_result(merged(), Some(&handle), false, Some(&latch)).await;
6438 task.await.unwrap();
6439 assert!(matches!(first, crate::extensions::hooks::events::HookResult::Continue));
6440 assert!(latch.allows_all(&key(&["a", "b"])));
6441 drop(handle);
6443 let second = resolve_before_tool_call_result(merged(), None, false, Some(&latch)).await;
6444 assert!(matches!(second, crate::extensions::hooks::events::HookResult::Continue));
6445 }
6446
6447 #[tokio::test]
6450 async fn no_latch_means_no_allow_all_scope() {
6451 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
6452 let handle = crate::tools::SecretPromptHandle::new(tx);
6453 let task = tokio::spawn(async move {
6454 let req = rx.recv().await.unwrap();
6455 assert!(req.confirm_scope.is_none());
6456 let _ = req.response_tx.send(Some("y".into()));
6457 });
6458 let r = resolve_before_tool_call_result(
6459 sourced_confirm("Run?", "jev", "bash"),
6460 Some(&handle),
6461 false,
6462 None,
6463 )
6464 .await;
6465 task.await.unwrap();
6466 assert!(matches!(r, crate::extensions::hooks::events::HookResult::Continue));
6467 }
6468
6469 #[tokio::test]
6474 async fn rewritten_input_is_shown_in_the_confirm_prompt() {
6475 let latch = crate::tools::SessionConfirmLatch::new();
6476 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
6477 let handle = crate::tools::SecretPromptHandle::new(tx);
6478 let task = tokio::spawn(async move {
6479 let req = rx.recv().await.unwrap();
6480 let body = req.prompt.clone();
6481 let scope = req.confirm_scope.clone().unwrap();
6482 let _ = req.response_tx.send(Some("y".into()));
6483 (body, scope)
6484 });
6485 let result = resolve_before_tool_call_result(
6486 crate::extensions::hooks::events::HookResult::Confirm {
6487 message: "Run this bash command?".into(),
6488 sources: ext(&["jev"]),
6489 generations: vec![],
6490 tool: Some("bash".into()),
6491 modified_input: Some(serde_json::json!({"command": "rm -rf build/"})),
6492 modified_by: ext(&["rewriter"]),
6493 },
6494 Some(&handle),
6495 false,
6496 Some(&latch),
6497 )
6498 .await;
6499 let (body, scope) = task.await.unwrap();
6500 let header_end = body.find("Message from jev:\nRun this bash command?").unwrap();
6501 let header = &body[..header_end];
6502 assert!(header.contains("Input rewritten by: rewriter"), "{body}");
6503 assert!(header.contains("Input that will run:\n rm -rf build/"), "{body}");
6504 assert_eq!(scope.rewritten_by, ext(&["rewriter"]));
6505 assert!(matches!(
6506 result,
6507 crate::extensions::hooks::events::HookResult::Modify { input }
6508 if input == serde_json::json!({"command": "rm -rf build/"})
6509 ));
6510 let plain = confirm_prompt_body("m", &ext(&["jev"]), Some("bash"), None);
6512 assert!(!plain.contains("rewritten"), "{plain}");
6513 }
6514
6515 #[test]
6517 fn final_input_preview_redacts_credentials() {
6518 let secrets = [
6519 "SECRET-BEARER-123",
6520 "SECRET-KEY-456",
6521 "SECRET-FIELD-789",
6522 "QmFzaWNTRUNSRVQ=",
6523 "SECRET-TOK-000",
6524 "SECRET-FLAG-111",
6525 "sk-ant-api03-abcdefghijklmnop",
6526 "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.c2lnbmF0dXJl",
6527 "ghp_abcdefghijklmnopqrstuvwxyz0123",
6528 ];
6529 let command = serde_json::json!({
6530 "command": "curl -H 'Authorization: Bearer SECRET-BEARER-123' \
6531 'https://api.example/v1?api_key=SECRET-KEY-456' \
6532 -H \"authorization: Basic QmFzaWNTRUNSRVQ=\" \
6533 && TOKEN=SECRET-TOK-000 deploy --password SECRET-FLAG-111 \
6534 sk-ant-api03-abcdefghijklmnop \
6535 eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.c2lnbmF0dXJl \
6536 ghp_abcdefghijklmnopqrstuvwxyz0123"
6537 });
6538 let json = serde_json::json!({
6539 "url": "https://x/y",
6540 "api_key": "SECRET-FIELD-789",
6541 "headers": {"X-Note": "Bearer SECRET-BEARER-123"},
6542 "args": ["--token=SECRET-TOK-000"],
6543 });
6544 for input in [command, json] {
6545 let p = final_input_preview(&input);
6546 for secret in secrets {
6547 assert!(!p.contains(secret), "{secret} leaked in:\n{p}");
6548 }
6549 assert!(p.contains("[REDACTED]"), "{p}");
6550 }
6551 let plain = final_input_preview(&serde_json::json!({"command": "rm -rf build/ && ls -la"}));
6553 assert!(plain.contains("rm -rf build/ && ls -la"), "{plain}");
6554 }
6555
6556 #[test]
6559 fn redact_secret_text_covers_round4_shapes() {
6560 let rows: &[(&str, &str)] = &[
6561 ("curl -H 'x-api-key: XAPIKEY-SECRET-1' https://a", "XAPIKEY-SECRET-1"),
6562 ("FOO_TOKEN=TOK-SECRET-2 run", "TOK-SECRET-2"),
6563 ("MY_SECRET=SEC-SECRET-3 ./x", "SEC-SECRET-3"),
6564 ("DB_PASSWORD=PW-SECRET-4 ./x", "PW-SECRET-4"),
6565 ("OPENAI_API_KEY=APIKEY-SECRET-5 ./x", "APIKEY-SECRET-5"),
6566 ("curl -H 'Authorization: Basic QkFTSUMtU0VDUkVULTY='", "QkFTSUMtU0VDUkVULTY="),
6567 ("aws_secret_access_key=AWSSEC-SECRET-7", "AWSSEC-SECRET-7"),
6568 ("git clone https://alice:USERINFO-SECRET-8@github.com/o/r", "USERINFO-SECRET-8"),
6569 ("psql postgres://app:PGPW-SECRET-9@db:5432/prod", "PGPW-SECRET-9"),
6570 ("DATABASE_URL=postgres://app:DBURL-SECRET-10@db/prod ./run", "DBURL-SECRET-10"),
6571 ("curl -H 'Cookie: sid=COOKIE-SECRET-11; theme=dark' https://a", "COOKIE-SECRET-11"),
6572 ("printf 'Set-Cookie: sid=SETCOOKIE-SECRET-12'", "SETCOOKIE-SECRET-12"),
6573 ("curl -u alice:CURLU-SECRET-13 https://a", "CURLU-SECRET-13"),
6574 ("mysql -u root -pMYSQL-SECRET-14 prod", "MYSQL-SECRET-14"),
6575 ("sshpass -p SSHPASS-SECRET-15 ssh host", "SSHPASS-SECRET-15"),
6576 ("aws configure set aws_secret_access_key AWSCFG-SECRET-16", "AWSCFG-SECRET-16"),
6577 ("aws x --secret-access-key FLAG-SECRET-17 --region eu", "FLAG-SECRET-17"),
6578 ("aws x --access-key FLAG-SECRET-18", "FLAG-SECRET-18"),
6579 ("STRIPE_KEY=rk_live_STRIPE-SECRET-19 node app", "STRIPE-SECRET-19"),
6580 ("OPENAI_KEY=OAIKEY-SECRET-20 node app", "OAIKEY-SECRET-20"),
6581 ];
6582 let precise = [
6585 "QkFTSUMtU0VDUkVULTY=",
6586 "USERINFO-SECRET-8",
6587 "PGPW-SECRET-9",
6588 "DBURL-SECRET-10",
6589 "COOKIE-SECRET-11",
6590 "SETCOOKIE-SECRET-12",
6591 "CURLU-SECRET-13",
6592 "MYSQL-SECRET-14",
6593 "SSHPASS-SECRET-15",
6594 ];
6595 for (text, secret) in rows {
6596 let out = redact_secret_text(text);
6598 assert!(!out.contains(secret), "{secret} leaked: {out}");
6599 assert!(out.contains("[REDACTED]"), "{out}");
6600 let preview = final_input_preview(&serde_json::json!({ "command": text }));
6601 assert!(!preview.contains(secret), "{secret} leaked in preview: {preview}");
6602 if precise.contains(secret) {
6603 let mut v = serde_json::Value::String(text.to_string());
6604 crate::runtime::trace::export::redact_value(&mut v);
6605 assert!(
6606 !v.as_str().unwrap().contains(secret),
6607 "shared scrub leaked {secret}: {v}"
6608 );
6609 }
6610 }
6611 for ok in [
6613 "mkdir -p build/out",
6614 "ssh -p 2222 host",
6615 "ssh -p2222 host",
6616 "docker run -p 8080:80 nginx",
6617 "mysql -h db -P 3306 -p prod",
6618 "curl -u alice https://example.com/path",
6619 "echo the token is valid and the password field is empty",
6620 "monkey=1 make",
6621 "git log --format=%H -- src/",
6622 "https://example.com/a/b?page=2",
6623 ] {
6624 assert_eq!(redact_secret_text(ok), ok, "over-redacted");
6625 }
6626 }
6627
6628 #[test]
6633 fn shared_scrub_leaves_benign_code_and_prose_alone() {
6634 let benign = [
6635 "max_tokens: 4096",
6636 "password: String,",
6637 "let token = load()?;",
6638 "the password-reset flow",
6639 "pub authorization: Option<String>,",
6640 "cookie: HashMap<String, String>,",
6641 "MY_KEY=value_in_docs",
6642 "--token-file path/to/file",
6643 "fn refresh_access_token(&self) -> Result<Token>",
6644 "aws_secret_access_key is the field name",
6645 ];
6646 for text in benign {
6647 let mut v = serde_json::Value::String(text.to_string());
6648 crate::runtime::trace::export::redact_value(&mut v);
6649 assert_eq!(v.as_str().unwrap(), text, "shared scrub mangled benign text");
6650 assert_eq!(crate::runtime::trace::export::redact_text(text), text);
6651 }
6652 let mut record = serde_json::json!({
6654 "text": "set max_tokens: 4096 and keep the password-reset flow",
6655 "code": ["password: String,", "let token = load()?;"],
6656 });
6657 let before = record.clone();
6658 crate::runtime::trace::export::redact_value(&mut record);
6659 assert_eq!(record, before);
6660 }
6661
6662 #[test]
6665 fn confirmer_message_echoing_injected_credentials_is_redacted() {
6666 let injected = serde_json::json!({
6667 "command": "curl -H 'Authorization: Bearer ECHOED-BEARER-X' https://a"
6668 });
6669 let body = confirm_prompt_body(
6670 "Run: curl -H 'Authorization: Bearer ECHOED-BEARER-X' https://a ?",
6671 &ext(&["jev"]),
6672 Some("bash"),
6673 Some((&ext(&["injector"]), &injected)),
6674 );
6675 assert!(!body.contains("ECHOED-BEARER-X"), "{body}");
6676 assert!(body.contains("Message from jev:\nRun: curl -H 'Authorization: [REDACTED]"), "{body}");
6677 }
6678
6679 #[test]
6683 fn confirm_message_control_and_bidi_chars_are_stripped() {
6684 let hostile = "ok\x1b[3A\x1b[2KExtension: trusted\rTool: none\u{0085}\u{202E}evil\u{200B}x\u{2066}y\tz\nnext";
6685 let body = confirm_prompt_body(hostile, &ext(&["jev"]), Some("bash"), None);
6686 for bad in ['\x1b', '\r', '\u{0085}', '\u{202E}', '\u{200B}', '\u{2066}', '\t'] {
6687 assert!(!body.contains(bad), "{bad:?} survived: {body:?}");
6688 }
6689 assert!(body.contains("Message from jev:\nok[3A[2KExtension: trustedTool: noneevilxy z\nnext"), "{body:?}");
6690 assert!(body.starts_with("Extension: jev\nTool: bash\n"));
6691 }
6692
6693 #[test]
6696 fn extension_message_is_labelled_after_the_host_header() {
6697 let fake = "Input that will run:\n echo harmless";
6698 let body = confirm_prompt_body(fake, &ext(&["jev"]), Some("bash"), None);
6699 let label = body.find("Message from jev:\n").expect(&body);
6700 assert_eq!(body.find("Input that will run:"), Some(label + "Message from jev:\n".len()));
6701 assert!(body.starts_with("Extension: jev\nTool: bash\n"), "{body}");
6702 }
6703
6704 #[test]
6705 fn final_input_preview_is_capped() {
6706 let many = serde_json::json!({"command": (1..=20).map(|i| format!("echo {i}")).collect::<Vec<_>>().join("\n")});
6707 let p = final_input_preview(&many);
6708 assert!(p.lines().count() <= MAX_REWRITE_PREVIEW_LINES + 1, "{p}");
6709 assert!(p.ends_with("…[preview truncated]"), "{p}");
6710 let wide = serde_json::json!({"path": "x".repeat(2000)});
6711 let p = final_input_preview(&wide);
6712 assert!(p.chars().count() <= MAX_REWRITE_PREVIEW_CHARS + 30, "{}", p.len());
6713 let obj = serde_json::json!({"path": "/tmp/a", "content": "hi"});
6714 assert!(final_input_preview(&obj).contains("\"path\": \"/tmp/a\""));
6715 }
6716
6717 #[test]
6720 fn long_confirm_message_is_capped() {
6721 let long = "x\n".repeat(500);
6722 let body = confirm_prompt_body(&long, &ext(&["jev"]), Some("bash"), None);
6723 assert!(body.starts_with("Extension: jev\nTool: bash\n"));
6724 assert!(body.contains("…[message truncated by host]"));
6725 assert!(body.lines().count() <= MAX_CONFIRM_MESSAGE_LINES + 8, "{}", body.lines().count());
6726 let wide = "y".repeat(10_000);
6727 assert!(confirm_prompt_body(&wide, &ext(&["jev"]), None, None).chars().count() < 1400);
6728 }
6729
6730 #[tokio::test]
6733 async fn unattributed_confirm_always_answer_does_not_latch() {
6734 let latch = crate::tools::SessionConfirmLatch::new();
6735 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
6736 let handle = crate::tools::SecretPromptHandle::new(tx);
6737 let task = tokio::spawn(async move {
6738 let req = rx.recv().await.expect("prompt");
6739 assert!(req.confirm_scope.is_none());
6740 let _ = req.response_tx.send(Some(CONFIRM_ANSWER_ALLOW_ALL.to_string()));
6741 });
6742 let result = resolve_before_tool_call_result(
6743 crate::extensions::hooks::events::HookResult::confirm("Run deploy?"),
6744 Some(&handle),
6745 false,
6746 Some(&latch),
6747 )
6748 .await;
6749 task.await.unwrap();
6750 assert!(matches!(
6751 result,
6752 crate::extensions::hooks::events::HookResult::Continue
6753 ));
6754 assert!(latch.is_empty());
6755 }
6756
6757 #[tokio::test]
6761 async fn session_allow_all_does_not_override_block() {
6762 let latch = crate::tools::SessionConfirmLatch::new();
6763 latch.allow(&key(&["jev"]));
6764 let blocked = resolve_before_tool_call_result(
6765 crate::extensions::hooks::events::HookResult::Block {
6766 reason: "nope".into(),
6767 },
6768 None,
6769 false,
6770 Some(&latch),
6771 )
6772 .await;
6773 assert!(matches!(
6774 blocked,
6775 crate::extensions::hooks::events::HookResult::Block { reason } if reason == "nope"
6776 ));
6777 let headless = resolve_before_tool_call_result(
6778 sourced_confirm("Run deploy?", "jev", "bash"),
6779 None,
6780 false,
6781 Some(&latch),
6782 )
6783 .await;
6784 assert!(matches!(
6785 headless,
6786 crate::extensions::hooks::events::HookResult::Continue
6787 ));
6788 let other = resolve_before_tool_call_result(
6789 sourced_confirm("Run deploy?", "other", "bash"),
6790 None,
6791 false,
6792 Some(&latch),
6793 )
6794 .await;
6795 assert!(matches!(
6796 other,
6797 crate::extensions::hooks::events::HookResult::Block { .. }
6798 ));
6799 }
6800
6801 #[test]
6802 fn test_max_tokens_for_model() {
6803 assert_eq!(
6805 HelperMethods::max_tokens_for_model("claude-opus-4-6"),
6806 128000
6807 );
6808 assert_eq!(
6809 HelperMethods::max_tokens_for_model("opus-something"),
6810 128000
6811 );
6812
6813 assert_eq!(
6815 HelperMethods::max_tokens_for_model("claude-sonnet-4-20250514"),
6816 64000
6817 );
6818 assert_eq!(HelperMethods::max_tokens_for_model("haiku"), 64000);
6819 assert_eq!(HelperMethods::max_tokens_for_model("claude-3-haiku"), 64000);
6820 assert_eq!(
6821 HelperMethods::max_tokens_for_model("some-other-model"),
6822 64000
6823 );
6824
6825 assert_eq!(HelperMethods::max_tokens_for_model(""), 64000);
6827 assert_eq!(HelperMethods::max_tokens_for_model("OPUS"), 64000); assert_eq!(
6829 HelperMethods::max_tokens_for_model("model-opus-end"),
6830 128000
6831 ); }
6833
6834 #[test]
6835 fn test_truncate_tool_result() {
6836 let default_max = 30000;
6837
6838 let short = "This is a short string.";
6840 assert_eq!(
6841 HelperMethods::truncate_tool_result(short, default_max),
6842 short
6843 );
6844
6845 let exact = "x".repeat(30000);
6847 assert_eq!(
6848 HelperMethods::truncate_tool_result(&exact, default_max),
6849 exact
6850 );
6851
6852 let too_long = "x".repeat(30001);
6854 let truncated = HelperMethods::truncate_tool_result(&too_long, default_max);
6855
6856 assert!(truncated.starts_with(&"x".repeat(30000)));
6858
6859 assert!(truncated.contains("[truncated — 30001 total bytes, showing first 30000]"));
6861
6862 assert!(truncated.len() > 30000);
6864
6865 let very_long = "a".repeat(50000);
6867 let truncated_very_long = HelperMethods::truncate_tool_result(&very_long, default_max);
6868 assert!(
6869 truncated_very_long.contains("[truncated — 50000 total bytes, showing first 30000]")
6870 );
6871 assert!(truncated_very_long.starts_with(&"a".repeat(30000)));
6872
6873 let custom_truncated = HelperMethods::truncate_tool_result(&very_long, 100);
6875 assert!(custom_truncated.starts_with(&"a".repeat(100)));
6876 assert!(custom_truncated.contains("[truncated — 50000 total bytes, showing first 100]"));
6877 }
6878
6879 #[test]
6883 fn truncate_tool_result_enforces_byte_budget() {
6884 let multibyte = "é".repeat(100); let out = HelperMethods::truncate_tool_result(&multibyte, 50);
6886 let body = out.split("\n\n[truncated").next().unwrap();
6887 assert!(
6888 body.len() <= 50,
6889 "retained body must fit the byte budget (got {} bytes)",
6890 body.len()
6891 );
6892 assert!(std::str::from_utf8(body.as_bytes()).is_ok());
6893 assert!(
6894 out.contains("total bytes"),
6895 "marker must report byte counts, got: {out}"
6896 );
6897 }
6898
6899 #[test]
6900 fn test_thinking_level_ranges() {
6901 use crate::core::models::thinking_level_for_budget;
6902
6903 assert_eq!(thinking_level_for_budget(0), "adaptive");
6905
6906 assert_eq!(thinking_level_for_budget(1), "low");
6908 assert_eq!(thinking_level_for_budget(1024), "low");
6909 assert_eq!(thinking_level_for_budget(2048), "low");
6910
6911 assert_eq!(thinking_level_for_budget(2049), "medium");
6913 assert_eq!(thinking_level_for_budget(3000), "medium");
6914 assert_eq!(thinking_level_for_budget(4096), "medium");
6915
6916 assert_eq!(thinking_level_for_budget(4097), "high");
6918 assert_eq!(thinking_level_for_budget(8192), "high");
6919 assert_eq!(thinking_level_for_budget(16384), "high");
6920
6921 assert_eq!(thinking_level_for_budget(16385), "xhigh");
6923 assert_eq!(thinking_level_for_budget(32768), "xhigh");
6924 assert_eq!(thinking_level_for_budget(100000), "xhigh");
6925 }
6926}
6927
6928#[cfg(test)]
6929mod effective_prompt_tests {
6930 use super::*;
6931
6932 struct LateTool;
6935 #[async_trait::async_trait]
6936 impl crate::Tool for LateTool {
6937 fn name(&self) -> &str {
6938 "late_tool"
6939 }
6940 fn description(&self) -> &str {
6941 "Registered after the catalog froze."
6942 }
6943 fn parameters(&self) -> Value {
6944 json!({"type": "object", "properties": {}})
6945 }
6946 async fn execute(&self, _p: Value, _c: crate::ToolContext) -> crate::Result<String> {
6947 Ok(String::new())
6948 }
6949 }
6950
6951 #[tokio::test]
6952 async fn tool_catalog_lists_activatable_tools_outside_the_core() {
6953 let mut runtime = headless("anthropic/claude-sonnet-5-5", "BASE.");
6954 assert!(
6955 runtime.frozen_tool_catalog().await.is_none(),
6956 "flag off: no catalog"
6957 );
6958 runtime.progressive_tool_disclosure = true;
6959 let catalog = runtime.frozen_tool_catalog().await.expect("catalog");
6960 assert!(catalog.starts_with("## Activatable tools\n"));
6961 assert!(
6962 catalog.contains("not instructions"),
6963 "third-party text is framed as data"
6964 );
6965 assert!(catalog.contains("- builtin:subagent_start"), "{catalog}");
6966 for core in [
6967 "builtin:bash",
6968 "builtin:read",
6969 "builtin:activate_tools",
6970 "builtin:search_tools",
6971 ] {
6972 assert!(
6973 !catalog.contains(&format!("- {core}")),
6974 "core {core} listed"
6975 );
6976 }
6977 let prompt = runtime.effective_system_prompt().await.unwrap();
6978 assert!(prompt.starts_with("BASE."), "base stays first");
6979 assert!(prompt.ends_with(&catalog), "catalog appended last");
6980 }
6981
6982 #[tokio::test]
6983 async fn tool_catalog_is_frozen_per_session() {
6984 let mut runtime = headless("anthropic/claude-sonnet-5-5", "BASE.");
6985 runtime.progressive_tool_disclosure = true;
6986 let first = runtime.effective_system_prompt().await.unwrap();
6987 runtime.tools.write().await.register(Arc::new(LateTool));
6988 let clone = runtime.clone();
6989 assert_eq!(runtime.effective_system_prompt().await.unwrap(), first);
6990 assert_eq!(clone.effective_system_prompt().await.unwrap(), first);
6991 assert!(
6992 !first.contains("late_tool"),
6993 "late registration does not rewrite the prefix"
6994 );
6995 }
6996
6997 #[tokio::test]
6998 async fn tool_catalog_absent_when_activation_is_denied_or_impossible() {
6999 let mut denied = headless("anthropic/claude-sonnet-5-5", "BASE.");
7000 denied.progressive_tool_disclosure = true;
7001 denied.set_activation_confirm(agent_core::config::ActivationConfirm::Deny);
7002 assert!(denied.frozen_tool_catalog().await.is_none());
7003 assert_eq!(
7004 denied.effective_system_prompt().await.as_deref(),
7005 Some("BASE.")
7006 );
7007
7008 let mut worker = headless("anthropic/claude-sonnet-5-5", "BASE.");
7009 worker.progressive_tool_disclosure = true;
7010 worker.set_tools(crate::ToolRegistry::without_subagent());
7011 assert!(
7012 worker.frozen_tool_catalog().await.is_none(),
7013 "no activate_tools, no catalog"
7014 );
7015 }
7016
7017 fn headless(model: &str, base: &str) -> Runtime {
7018 let mut runtime = Runtime::new_headless();
7019 runtime.set_model(model.to_string());
7020 runtime.set_system_prompt(base.to_string());
7021 runtime
7022 }
7023
7024 #[tokio::test]
7025 async fn anthropic_foreground_ultracode_composes_standing_doctrine_once() {
7026 let mut runtime = headless("anthropic/claude-fable-5", "BASE.");
7027 runtime.named_level = Some(agent_core::reasoning::ReasoningLevel::UltraCode);
7028 let prompt = runtime.effective_system_prompt().await.expect("doctrine");
7029 assert_eq!(prompt.matches("<anthropic-ultracode-workflow>").count(), 1);
7030 assert!(prompt.contains("subagent_start"));
7031 assert!(prompt.contains("subagent_resume"));
7032 }
7033
7034 #[tokio::test]
7035 async fn codex_foreground_with_subagent_tools_composes_doctrine() {
7036 let runtime = headless("openai-codex/gpt-5.6-sol", "BASE.");
7037 let prompt = runtime
7038 .effective_system_prompt()
7039 .await
7040 .expect("prompt must compose");
7041 assert!(prompt.starts_with("BASE."), "base must lead");
7042 assert!(prompt.contains("## Subagent supervision"));
7043 assert!(prompt.contains("NEVER end your turn"));
7044 }
7045
7046 #[tokio::test]
7047 async fn worker_runtime_without_subagent_tools_stays_clean() {
7048 let mut runtime = headless("openai-codex/gpt-5.6-sol", "WORKER.");
7049 runtime.set_tools(ToolRegistry::without_subagent());
7050 assert_eq!(
7051 runtime.effective_system_prompt().await.as_deref(),
7052 Some("WORKER."),
7053 "workers have no subagent tools; doctrine would be unactionable noise"
7054 );
7055 }
7056
7057 #[tokio::test]
7058 async fn non_codex_foreground_stays_clean() {
7059 let runtime = headless("xai-auth/grok-4.5-latest", "BASE.");
7060 assert_eq!(
7061 runtime.effective_system_prompt().await.as_deref(),
7062 Some("BASE.")
7063 );
7064 }
7065
7066 #[tokio::test]
7067 async fn manifest_prompt_is_returned_verbatim() {
7068 let mut runtime = headless("openai-codex/gpt-5.6-sol", "IGNORED.");
7069 let module = agent_core::prompt::PromptModule::new(
7070 agent_core::prompt::PromptModuleId::parse("kernel.test").unwrap(),
7071 "1.0.0",
7072 agent_core::prompt::PromptModuleSource::User,
7073 0,
7074 agent_core::prompt::PromptSelectors::default(),
7075 agent_core::prompt::ModuleMutability::MutableGuidance,
7076 "KERNEL.",
7077 )
7078 .unwrap();
7079 let context = agent_core::prompt::SelectionContext::new(
7080 agent_core::prompt::QualifiedModelId::parse("openai-codex/gpt-5.6-sol").unwrap(),
7081 None,
7082 )
7083 .unwrap();
7084 let stack = agent_core::prompt::PromptStack::new(vec![module], context).unwrap();
7085 runtime.apply_prompt_stack(stack).unwrap();
7086 assert_eq!(
7087 runtime.effective_system_prompt().await.as_deref(),
7088 Some("KERNEL."),
7089 "manifest authors own the full stack; no builtin injection on top"
7090 );
7091 }
7092
7093 #[tokio::test]
7094 async fn unresolvable_model_identity_fails_closed_to_base() {
7095 let runtime = headless("definitely-not-a-model", "BASE.");
7096 assert_eq!(
7097 runtime.effective_system_prompt().await.as_deref(),
7098 Some("BASE.")
7099 );
7100 }
7101}
7102
7103#[cfg(test)]
7104mod set_reasoning_level_checked_tests {
7105 use super::*;
7106 use agent_core::reasoning::ReasoningLevel;
7107
7108 fn codex_runtime(model_id: &str) -> Runtime {
7109 let mut rt = Runtime::new_headless();
7110 rt.set_model(format!("openai-codex/{model_id}"));
7111 rt.set_reasoning_level(ReasoningLevel::Low);
7112 rt
7113 }
7114
7115 fn provisioned_fable_runtime() -> Runtime {
7116 let mut rt = Runtime::new_headless();
7117 rt.set_model("anthropic/claude-fable-5".into());
7118 rt.set_reasoning_level(ReasoningLevel::Low);
7119 let foreground =
7120 agent_core::prompt::QualifiedModelId::parse("anthropic/claude-fable-5").unwrap();
7121 let orchestration =
7122 crate::orchestration::OrchestrationRuntime::baseline(foreground, 3, 8).unwrap();
7123 rt.install_orchestration(Arc::new(orchestration));
7124 rt
7125 }
7126
7127 fn install_typed_manifest(rt: &mut Runtime) {
7128 let module = agent_core::prompt::PromptModule::new(
7129 agent_core::prompt::PromptModuleId::parse("kernel.test").unwrap(),
7130 "1.0.0",
7131 agent_core::prompt::PromptModuleSource::User,
7132 0,
7133 agent_core::prompt::PromptSelectors::default(),
7134 agent_core::prompt::ModuleMutability::MutableGuidance,
7135 "KERNEL.",
7136 )
7137 .unwrap();
7138 let context = agent_core::prompt::SelectionContext::new(
7139 agent_core::prompt::QualifiedModelId::parse("anthropic/claude-fable-5").unwrap(),
7140 None,
7141 )
7142 .unwrap();
7143 rt.apply_prompt_stack(agent_core::prompt::PromptStack::new(vec![module], context).unwrap())
7144 .unwrap();
7145 }
7146
7147 #[test]
7148 fn provisioned_fable_accepts_max_and_ultracode() {
7149 for level in [ReasoningLevel::Max, ReasoningLevel::UltraCode] {
7150 let mut rt = provisioned_fable_runtime();
7151 rt.set_reasoning_level_checked(level).unwrap();
7152 assert_eq!(rt.reasoning_level(), level);
7153 }
7154 }
7155
7156 #[test]
7157 fn fable_ultracode_prerequisites_fail_without_mutation() {
7158 let mut missing_orchestration = Runtime::new_headless();
7159 missing_orchestration.set_model("anthropic/claude-fable-5".into());
7160 missing_orchestration.set_reasoning_level(ReasoningLevel::Low);
7161 assert!(missing_orchestration
7162 .set_reasoning_level_checked(ReasoningLevel::UltraCode)
7163 .is_err());
7164 assert_eq!(missing_orchestration.reasoning_level(), ReasoningLevel::Low);
7165
7166 let mut missing_tools = provisioned_fable_runtime();
7167 missing_tools.set_tools(ToolRegistry::without_subagent());
7168 assert!(missing_tools
7169 .set_reasoning_level_checked(ReasoningLevel::UltraCode)
7170 .is_err());
7171 assert_eq!(missing_tools.reasoning_level(), ReasoningLevel::Low);
7172
7173 let mut worker = provisioned_fable_runtime();
7174 worker.set_codex_request_role(crate::runtime::openai::catalog::ExecutionRole::Worker);
7175 assert!(worker
7176 .set_reasoning_level_checked(ReasoningLevel::UltraCode)
7177 .is_err());
7178 assert_eq!(worker.reasoning_level(), ReasoningLevel::Low);
7179
7180 let mut internal = provisioned_fable_runtime();
7181 internal.set_codex_request_role(crate::runtime::openai::catalog::ExecutionRole::Internal);
7182 assert!(internal
7183 .set_reasoning_level_checked(ReasoningLevel::UltraCode)
7184 .is_err());
7185 assert_eq!(internal.reasoning_level(), ReasoningLevel::Low);
7186
7187 let mut manifest = provisioned_fable_runtime();
7188 install_typed_manifest(&mut manifest);
7189 assert!(manifest
7190 .set_reasoning_level_checked(ReasoningLevel::UltraCode)
7191 .is_err());
7192 assert_eq!(manifest.reasoning_level(), ReasoningLevel::Low);
7193 }
7194
7195 #[test]
7196 fn checked_accepts_client_omission_modes_for_codex() {
7197 for level in [ReasoningLevel::Off, ReasoningLevel::Adaptive] {
7198 let mut rt = codex_runtime("gpt-5.6-sol");
7199 rt.set_reasoning_level_checked(level).unwrap();
7200 assert_eq!(rt.reasoning_level(), level);
7201 }
7202 }
7203
7204 #[test]
7206 fn checked_rejects_ultra_for_luna_no_mutation() {
7207 let mut rt = codex_runtime("gpt-5.6-luna");
7208 let before = rt.reasoning_level();
7209 let err = rt
7210 .set_reasoning_level_checked(ReasoningLevel::Ultra)
7211 .unwrap_err();
7212 assert!(
7213 err.contains("ultra"),
7214 "error must name the rejected level; got: {err}"
7215 );
7216 assert_eq!(
7217 rt.reasoning_level(),
7218 before,
7219 "runtime must not be mutated when validation fails"
7220 );
7221 }
7222
7223 #[test]
7225 fn checked_rejects_max_for_gpt55_no_mutation() {
7226 let mut rt = codex_runtime("gpt-5.5");
7227 rt.set_reasoning_level(ReasoningLevel::Medium);
7228 let err = rt
7229 .set_reasoning_level_checked(ReasoningLevel::Max)
7230 .unwrap_err();
7231 assert!(
7232 err.contains("max"),
7233 "error must name the rejected level; got: {err}"
7234 );
7235 assert_eq!(rt.reasoning_level(), ReasoningLevel::Medium);
7236 }
7237
7238 #[test]
7240 fn checked_accepts_ultra_for_sol_and_mutates() {
7241 let mut rt = codex_runtime("gpt-5.6-sol");
7242 rt.set_reasoning_level_checked(ReasoningLevel::Ultra)
7243 .expect("sol must accept ultra");
7244 assert_eq!(rt.reasoning_level(), ReasoningLevel::Ultra);
7245 }
7246
7247 #[test]
7249 fn checked_rejects_extended_levels_without_metadata() {
7250 let mut rt = Runtime::new_headless();
7251 rt.set_model("claude-opus-4-7".to_string());
7252 rt.set_reasoning_level(ReasoningLevel::Low);
7253 for level in [ReasoningLevel::Max, ReasoningLevel::Ultra] {
7254 assert!(rt.set_reasoning_level_checked(level).is_err());
7255 assert_eq!(rt.reasoning_level(), ReasoningLevel::Low);
7256 }
7257 }
7258
7259 #[test]
7262 fn checked_rejects_max_ultra_for_unknown_codex_model() {
7263 let mut rt = codex_runtime("gpt-unknown-future");
7264 rt.set_reasoning_level(ReasoningLevel::Low);
7265 for level in [ReasoningLevel::Max, ReasoningLevel::Ultra] {
7266 let err = rt.set_reasoning_level_checked(level).unwrap_err();
7267 assert!(err.contains("no capability metadata"), "{err}");
7268 assert_eq!(rt.reasoning_level(), ReasoningLevel::Low);
7269 }
7270 }
7271
7272 #[test]
7275 fn checked_enforces_xai_capability_matrix_no_mutation_on_err() {
7276 let mut rt = Runtime::new_headless();
7277 rt.set_model("xai-auth/grok-4.5".to_string());
7278 assert_eq!(rt.reasoning_level(), ReasoningLevel::High);
7280 for level in [
7281 ReasoningLevel::Off,
7282 ReasoningLevel::XHigh,
7283 ReasoningLevel::Ultra,
7284 ] {
7285 let before = rt.reasoning_level();
7286 assert!(rt.set_reasoning_level_checked(level).is_err(), "{level}");
7287 assert_eq!(rt.reasoning_level(), before);
7288 }
7289 rt.set_reasoning_level_checked(ReasoningLevel::Low)
7290 .expect("grok-4.5 supports low");
7291 assert_eq!(rt.reasoning_level(), ReasoningLevel::Low);
7292 }
7293
7294 #[test]
7297 fn checked_rejects_named_on_intrinsic_xai_model_and_defaults_adaptive() {
7298 let mut rt = Runtime::new_headless();
7299 rt.set_model("xai-auth/grok-4.3".to_string());
7300 assert_eq!(rt.reasoning_level(), ReasoningLevel::Adaptive);
7301 assert!(rt
7302 .set_reasoning_level_checked(ReasoningLevel::Medium)
7303 .is_err());
7304 assert_eq!(rt.reasoning_level(), ReasoningLevel::Adaptive);
7305 rt.set_reasoning_level_checked(ReasoningLevel::Adaptive)
7306 .unwrap();
7307 }
7308
7309 #[test]
7311 fn explicit_level_survives_xai_model_switch() {
7312 let mut rt = Runtime::new_headless();
7313 rt.set_model("xai-auth/grok-4.5".to_string());
7314 rt.set_reasoning_level_checked(ReasoningLevel::Low).unwrap();
7315 rt.set_model("xai-auth/grok-4.5-latest".to_string());
7316 assert_eq!(
7317 rt.reasoning_level(),
7318 ReasoningLevel::Low,
7319 "explicit level must not be overwritten by the model default"
7320 );
7321 }
7322
7323 #[test]
7326 fn explicit_unsupported_level_is_clamped_to_model_default_on_switch() {
7327 let mut rt = Runtime::new_headless();
7328 rt.set_reasoning_level_explicit(ReasoningLevel::XHigh);
7329 let clamp = rt
7330 .try_set_model("xai-auth/grok-4.6".to_string())
7331 .expect("model switch succeeds");
7332 assert_eq!(
7333 clamp,
7334 Some(ReasoningClamp {
7335 from: ReasoningLevel::XHigh,
7336 to: ReasoningLevel::High,
7337 })
7338 );
7339 assert_eq!(rt.reasoning_level(), ReasoningLevel::High);
7340 assert!(
7341 rt.is_reasoning_explicit(),
7342 "clamp keeps explicit provenance"
7343 );
7344 assert_eq!(rt.model(), "xai-auth/grok-4.6");
7345 }
7346
7347 #[test]
7349 fn explicit_supported_level_is_not_clamped_on_switch() {
7350 let mut rt = Runtime::new_headless();
7351 rt.set_reasoning_level_explicit(ReasoningLevel::High);
7352 let clamp = rt.try_set_model("xai-auth/grok-4.6".to_string()).unwrap();
7353 assert_eq!(clamp, None);
7354 assert_eq!(rt.reasoning_level(), ReasoningLevel::High);
7355 assert!(rt.is_reasoning_explicit());
7356 }
7357
7358 #[test]
7361 fn non_explicit_default_application_reports_no_clamp() {
7362 let mut rt = Runtime::new_headless();
7363 rt.set_reasoning_level(ReasoningLevel::XHigh);
7364 let clamp = rt.try_set_model("xai-auth/grok-4.6".to_string()).unwrap();
7365 assert_eq!(clamp, None);
7366 assert_eq!(rt.reasoning_level(), ReasoningLevel::High);
7367 assert!(!rt.is_reasoning_explicit());
7368 }
7369
7370 #[test]
7373 fn explicit_max_clamps_to_validating_option_on_multi_agent_model() {
7374 let model = "xai-auth/grok-4.20-multi-agent-0309";
7375 let mut rt = Runtime::new_headless();
7376 rt.set_reasoning_level_explicit(ReasoningLevel::Max);
7377 let clamp = rt
7378 .try_set_model(model.to_string())
7379 .unwrap()
7380 .expect("Max is not supported: must clamp");
7381 assert_eq!(clamp.from, ReasoningLevel::Max);
7382 assert!(
7383 crate::runtime::openai::catalog::validation::validate_reasoning_mutation(
7384 model, clamp.to
7385 )
7386 .is_ok(),
7387 "clamped level {} must validate",
7388 clamp.to
7389 );
7390 assert_eq!(rt.reasoning_level(), clamp.to);
7391 assert!(rt.is_reasoning_explicit());
7392 }
7393
7394 #[test]
7396 fn set_model_returns_clamp() {
7397 let mut rt = Runtime::new_headless();
7398 rt.set_reasoning_level_explicit(ReasoningLevel::XHigh);
7399 let clamp = rt.set_model("xai-auth/grok-4.6".to_string());
7400 assert_eq!(clamp.map(|c| c.to), Some(ReasoningLevel::High));
7401 }
7402
7403 #[test]
7406 fn restore_session_reasoning_clamps_after_reapply() {
7407 let mut rt = Runtime::new_headless();
7408 rt.set_model("xai-auth/grok-4.6".to_string());
7409 let clamp = rt.restore_session_reasoning("xhigh");
7410 assert_eq!(
7411 clamp,
7412 Some(ReasoningClamp {
7413 from: ReasoningLevel::XHigh,
7414 to: ReasoningLevel::High,
7415 })
7416 );
7417 assert_eq!(rt.reasoning_level(), ReasoningLevel::High);
7418 assert!(rt.is_reasoning_explicit());
7419 assert!(rt.set_reasoning_level_checked(rt.reasoning_level()).is_ok());
7420 }
7421
7422 #[test]
7424 fn restore_session_reasoning_keeps_supported_level() {
7425 let mut rt = Runtime::new_headless();
7426 rt.set_model("xai-auth/grok-4.6".to_string());
7427 assert_eq!(rt.restore_session_reasoning("low"), None);
7428 assert_eq!(rt.reasoning_level(), ReasoningLevel::Low);
7429 assert!(rt.is_reasoning_explicit());
7430 }
7431
7432 #[test]
7435 fn apply_config_clamps_unsupported_explicit_level_against_config_model() {
7436 let mut rt = Runtime::new_headless();
7437 rt.apply_config(&crate::config::SynapsConfig {
7438 model: Some("xai-auth/grok-4.6".to_string()),
7439 thinking_level: Some(ReasoningLevel::XHigh),
7440 ..Default::default()
7441 });
7442 assert_eq!(rt.model(), "xai-auth/grok-4.6");
7443 assert_eq!(rt.reasoning_level(), ReasoningLevel::High);
7444 assert!(rt.is_reasoning_explicit());
7445 assert!(rt.set_reasoning_level_checked(rt.reasoning_level()).is_ok());
7446 }
7447}
7448
7449#[cfg(test)]
7450mod codex_execution_preflight_tests {
7451 use super::*;
7452 use crate::runtime::openai::catalog::CodexRequestRole;
7453 use agent_core::prompt::QualifiedModelId;
7454 use agent_core::reasoning::ReasoningLevel;
7455
7456 fn ultra_runtime() -> Runtime {
7457 let mut runtime = Runtime::new_headless();
7458 runtime.set_model("openai-codex/gpt-5.6-sol".to_string());
7459 runtime.set_reasoning_level_explicit(ReasoningLevel::Ultra);
7460 runtime
7461 }
7462
7463 fn install_baseline(runtime: &mut Runtime) {
7464 let foreground =
7465 QualifiedModelId::parse("openai-codex/gpt-5.6-sol").expect("qualified model");
7466 let orchestration = crate::orchestration::OrchestrationRuntime::baseline(foreground, 3, 8)
7467 .expect("baseline orchestration");
7468 runtime.install_orchestration(Arc::new(orchestration));
7469 }
7470
7471 #[tokio::test]
7472 async fn foreground_ultra_preflight_requires_orchestration() {
7473 let runtime = ultra_runtime();
7474 let error = runtime
7475 .validate_request_preflight()
7476 .await
7477 .expect_err("Ultra without orchestration must fail closed");
7478 assert!(error.to_string().contains("orchestration"), "{error}");
7479 }
7480
7481 #[tokio::test]
7482 async fn foreground_ultra_preflight_requires_actionable_subagent_tools() {
7483 let mut runtime = ultra_runtime();
7484 install_baseline(&mut runtime);
7485 runtime.set_tools(ToolRegistry::without_subagent());
7486
7487 let error = runtime
7488 .validate_request_preflight()
7489 .await
7490 .expect_err("Ultra without subagent tools must fail closed");
7491 assert!(error.to_string().contains("subagent"), "{error}");
7492 }
7493
7494 #[tokio::test]
7495 async fn foreground_ultra_preflight_accepts_exact_model_policy_and_tools() {
7496 let mut runtime = ultra_runtime();
7497 install_baseline(&mut runtime);
7498 runtime
7499 .validate_request_preflight()
7500 .await
7501 .expect("fully provisioned Ultra must pass preflight");
7502 }
7503
7504 #[tokio::test]
7505 async fn worker_ultra_preflight_never_requires_or_enables_orchestration() {
7506 let mut runtime = ultra_runtime();
7507 runtime.set_codex_request_role(CodexRequestRole::Worker);
7508 runtime.set_tools(ToolRegistry::without_subagent());
7509 runtime
7510 .validate_request_preflight()
7511 .await
7512 .expect("worker Ultra is underlying max without proactive orchestration");
7513 assert_eq!(runtime.codex_request_role(), CodexRequestRole::Worker);
7514 }
7515
7516 #[tokio::test]
7517 async fn non_codex_ultra_preflight_fails_before_provider_work() {
7518 let mut runtime = Runtime::new_headless();
7519 runtime.set_model("anthropic/claude-opus-4-7".to_string());
7520 runtime.set_reasoning_level_explicit(ReasoningLevel::Ultra);
7521 let error = runtime
7522 .validate_request_preflight()
7523 .await
7524 .expect_err("non-Codex Ultra must fail closed");
7525 assert!(
7526 error.to_string().contains("unsupported_reasoning_level"),
7527 "{error}"
7528 );
7529 }
7530
7531 #[tokio::test]
7532 async fn compaction_preflight_validates_target_before_credentials() {
7533 let mut runtime = ultra_runtime();
7534 runtime.set_compaction_model(Some("claude-sonnet-4-6".to_string()));
7535 {
7536 let mut auth = runtime.auth.write().await;
7537 auth.auth_token.clear();
7538 auth.auth_type = "none".to_string();
7539 auth.refresh_token = None;
7540 auth.token_expires = None;
7541 }
7542
7543 let error = runtime
7544 .compact_call(Vec::new())
7545 .await
7546 .expect_err("Codex Ultra cannot silently cross into Anthropic compaction");
7547 assert!(
7548 error.to_string().contains("authoritative exact-model"),
7549 "preflight must reject the target before auth access: {error}"
7550 );
7551 }
7552}
7553
7554#[cfg(test)]
7555mod http_client_timeout_tests {
7556 use futures::StreamExt;
7570 use std::time::Duration;
7571 use tokio::io::AsyncWriteExt;
7572
7573 const CHUNKED_HEAD: &[u8] =
7574 b"HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ntransfer-encoding: chunked\r\n\r\n";
7575
7576 fn chunk(data: &str) -> Vec<u8> {
7577 format!("{:x}\r\n{}\r\n", data.len(), data).into_bytes()
7578 }
7579
7580 async fn spawn_chunked_server(n_chunks: usize, gap: Duration, terminate: bool) -> String {
7583 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
7584 let addr = listener.local_addr().unwrap();
7585 tokio::spawn(async move {
7586 let (mut sock, _) = listener.accept().await.unwrap();
7587 let mut buf = [0u8; 4096];
7589 use tokio::io::AsyncReadExt;
7590 let _ = sock.read(&mut buf).await;
7591 sock.write_all(CHUNKED_HEAD).await.unwrap();
7592 for i in 0..n_chunks {
7593 sock.write_all(&chunk(&format!("data {i}\n")))
7594 .await
7595 .unwrap();
7596 sock.flush().await.unwrap();
7597 tokio::time::sleep(gap).await;
7598 }
7599 if terminate {
7600 sock.write_all(b"0\r\n\r\n").await.unwrap();
7601 } else {
7602 tokio::time::sleep(Duration::from_secs(60)).await;
7604 }
7605 });
7606 format!("http://{addr}/stream")
7607 }
7608
7609 async fn drain(
7610 client: &reqwest::Client,
7611 url: &str,
7612 ) -> Result<usize, Box<dyn std::error::Error + Send + Sync>> {
7613 let resp = client.get(url).send().await?;
7614 let mut stream = resp.bytes_stream();
7615 let mut total = 0usize;
7616 while let Some(c) = stream.next().await {
7617 total += c?.len();
7618 }
7619 Ok(total)
7620 }
7621
7622 #[tokio::test]
7627 async fn active_stream_outliving_read_timeout_survives() {
7628 let url = spawn_chunked_server(15, Duration::from_millis(100), true).await;
7630 let client = super::build_http_client(Duration::from_millis(400)).expect("client builds");
7631 let total = drain(&client, &url)
7632 .await
7633 .expect("active stream must never be killed by the idle timeout");
7634 assert!(total > 0);
7635 }
7636
7637 #[tokio::test]
7640 async fn stalled_stream_is_killed_by_read_timeout() {
7641 let url = spawn_chunked_server(2, Duration::from_millis(10), false).await;
7643 let client = super::build_http_client(Duration::from_millis(300)).expect("client builds");
7644 let start = std::time::Instant::now();
7645 let err = drain(&client, &url)
7646 .await
7647 .expect_err("stalled stream must be detected");
7648 assert!(
7649 start.elapsed() < Duration::from_secs(5),
7650 "stall must be detected promptly, took {:?}",
7651 start.elapsed()
7652 );
7653 let msg = crate::core::error::error_chain_string(err.as_ref());
7654 assert!(
7655 msg.contains("timed out") || msg.contains("timeout"),
7656 "must be a timeout error: {msg}"
7657 );
7658 }
7659
7660 #[tokio::test]
7664 async fn hung_headers_are_killed_by_read_timeout() {
7665 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
7666 let addr = listener.local_addr().unwrap();
7667 tokio::spawn(async move {
7668 let (_sock, _) = listener.accept().await.unwrap();
7669 tokio::time::sleep(Duration::from_secs(60)).await;
7670 });
7671 let client = super::build_http_client(Duration::from_millis(300)).expect("client builds");
7672 let start = std::time::Instant::now();
7673 let err = client
7674 .get(format!("http://{addr}/codex/responses"))
7675 .send()
7676 .await
7677 .expect_err("hung request must be detected");
7678 assert!(
7679 start.elapsed() < Duration::from_secs(5),
7680 "hung headers must be detected promptly, took {:?}",
7681 start.elapsed()
7682 );
7683 assert!(err.is_timeout(), "must be a timeout: {err:?}");
7684 }
7685}
7686
7687#[cfg(test)]
7689mod memory_context_provider_tests {
7690 use super::{memory_context, Runtime};
7691 use std::sync::Arc;
7692
7693 async fn memory_runtime_with_providers(
7701 plugins: &[(&str, &str)],
7702 ) -> (
7703 Runtime,
7704 std::sync::Arc<crate::extensions::lease::ExtensionRuntimeManager>,
7705 ) {
7706 let mut manager = crate::extensions::manager::ExtensionManager::new(Arc::new(
7707 crate::extensions::hooks::HookBus::new(),
7708 ));
7709 manager.set_progressive_deferral(true);
7710 for (plugin, provider) in plugins {
7711 let manifest: crate::extensions::manifest::ExtensionManifest =
7712 serde_json::from_value(serde_json::json!({
7713 "runtime": "process",
7714 "command": "/bin/false",
7715 "permissions": ["context_providers.register"],
7716 "deferred": {
7717 "context_providers": [{
7718 "id": provider,
7719 "capability": "project-memory",
7720 "description": "test context provider",
7721 "schema_version": 1
7722 }]
7723 }
7724 }))
7725 .expect("manifest parses");
7726 manager
7727 .load(plugin, &manifest)
7728 .await
7729 .expect("deferred context-provider load succeeds without spawning");
7730 }
7731 let extension_runtime = manager.extension_runtime();
7732 let mut runtime = Runtime::new_headless();
7733 runtime.install_extension_runtime(std::sync::Arc::clone(&extension_runtime));
7734 (runtime, extension_runtime)
7735 }
7736
7737 fn assert_memory_off_no_lease(runtime: &Runtime) {
7738 let status = runtime.memory_context_status();
7739 assert_eq!(
7740 status.durable,
7741 memory_context::DurableStatus::Off,
7742 "state must stay Off"
7743 );
7744 assert_eq!(
7745 status.one_shot,
7746 memory_context::OneShotStatus::Idle,
7747 "no one-shot lease may exist"
7748 );
7749 assert!(
7750 runtime.memory_bound_providers_for_test().is_empty(),
7751 "no provider may be bound"
7752 );
7753 }
7754
7755 #[tokio::test]
7758 async fn memory_backend_exclusive_rejects_extension_recall_capture_and_history() {
7759 for selector in ["axel", "invalid-selector"] {
7760 let (mut runtime, _manager) =
7761 memory_runtime_with_providers(&[("memory-test", "notes")]).await;
7762 let provider = runtime.resolve_memory_provider(None).unwrap();
7763 runtime.apply_config(&agent_core::config::load_config_from_str(&format!(
7764 "memory.backend = {selector}\n"
7765 )));
7766 assert!(runtime.memory_backend.exclusive());
7767 assert!(matches!(
7768 runtime.resolve_memory_provider(None),
7769 Err(memory_context::MemoryContextError::ProviderNotRegistered)
7770 ));
7771 assert!(matches!(
7772 runtime.resolve_memory_provider(Some(provider.as_str())),
7773 Err(memory_context::MemoryContextError::ProviderNotRegistered)
7774 ));
7775 assert!(matches!(
7776 runtime.memory_history_confirm(),
7777 Err(
7778 super::memory_history::HistoryImportError::CaptureProviderUnavailable
7779 | super::memory_history::HistoryImportError::ConsentRequired
7780 )
7781 ));
7782 assert_memory_off_no_lease(&runtime);
7783 let mut messages = vec![Arc::new(
7784 serde_json::json!({"role": "user", "content": "test"}),
7785 )];
7786 let original = messages.clone();
7787 runtime.apply_turn_memory_recall(&mut messages).await;
7788 assert_eq!(messages, original);
7789 let clone = runtime.clone();
7790 assert!(clone.memory_backend.exclusive());
7791 assert_eq!(clone.memory_backend.base(), runtime.memory_backend.base());
7792 }
7793 }
7794
7795 #[tokio::test]
7796 async fn memory_backend_first_config_is_immutable_and_changes_require_restart() {
7797 for (first, different) in [("legacy", "axel"), ("axel", "legacy")] {
7798 let mut runtime = Runtime::new_headless();
7799 let config =
7800 agent_core::config::load_config_from_str(&format!("memory.backend = {first}\n"));
7801 runtime.apply_config(&config);
7802 let binding = runtime.memory_backend.clone();
7803 runtime.apply_config(&config);
7804 assert!(!runtime
7805 .memory_backend_reconfigure_denied
7806 .load(std::sync::atomic::Ordering::SeqCst));
7807 if let (Ok(a), Ok(b)) = (binding.scope(), runtime.memory_backend.scope()) {
7808 assert!(std::ptr::eq(a, b));
7809 }
7810 let clone = runtime.clone();
7811 runtime.apply_config(&agent_core::config::load_config_from_str(&format!(
7812 "memory.backend = {different}\n"
7813 )));
7814 assert_eq!(runtime.memory_backend.exclusive(), binding.exclusive());
7815 assert_eq!(
7816 runtime.memory_backend_config.as_ref(),
7817 Some(&config.memory_backend)
7818 );
7819 for denied in [&runtime, &clone] {
7820 let error = denied
7821 .validate_request_preflight()
7822 .await
7823 .unwrap_err()
7824 .to_string();
7825 assert!(error.contains("restart required"), "{error}");
7826 assert!(denied.resolve_memory_provider(None).is_err());
7827 }
7828 runtime.apply_config(&config);
7829 assert!(runtime.validate_request_preflight().await.is_err());
7830 }
7831 }
7832
7833 #[tokio::test]
7834 async fn memory_backend_path_changes_also_require_restart() {
7835 for field in ["executable", "brain"] {
7836 let mut runtime = Runtime::new_headless();
7837 let config = crate::config::MemoryBackendConfig::default();
7838 runtime.apply_memory_backend_config(&config);
7839 let mut different = config.clone();
7840 let path = std::env::temp_dir().join("synaps-memory-config-test");
7841 if field == "executable" {
7842 different.executable = Some(path);
7843 } else {
7844 different.brain = Some(path);
7845 }
7846 runtime.apply_memory_backend_config(&different);
7847 assert_eq!(runtime.memory_backend_config.as_ref(), Some(&config));
7848 assert!(!runtime.memory_backend.exclusive());
7849 let error = runtime
7850 .validate_request_preflight()
7851 .await
7852 .unwrap_err()
7853 .to_string();
7854 assert!(error.contains("restart required"), "{error}");
7855 assert!(runtime.resolve_memory_provider(None).is_err());
7856 }
7857 }
7858
7859 fn default_mode_config(mode: &str, confirmed: bool) -> agent_core::config::MemoryConfig {
7860 agent_core::config::MemoryConfig {
7861 default_mode: mode.to_string(),
7862 default_mode_confirmed: confirmed,
7863 ..Default::default()
7864 }
7865 }
7866
7867 #[test]
7870 fn memory_default_mode_confirmed_grants_session_lease() {
7871 for (name, mode) in [
7872 (
7873 "recall_each_prompt",
7874 memory_context::MemoryContextMode::RecallEachPrompt,
7875 ),
7876 (
7877 "capture_only",
7878 memory_context::MemoryContextMode::CaptureOnly,
7879 ),
7880 (
7881 "capture_and_recall",
7882 memory_context::MemoryContextMode::CaptureAndRecall,
7883 ),
7884 ] {
7885 let runtime = Runtime::new_headless();
7886 let status = runtime
7887 .apply_memory_default_mode(&default_mode_config(name, true))
7888 .expect("a confirmed non-off default applies")
7889 .expect("legacy backend grants");
7890 assert!(
7891 matches!(status.durable, memory_context::DurableStatus::Active { mode: m, .. } if m == mode),
7892 "{name}: got {:?}",
7893 status.durable
7894 );
7895 assert_eq!(
7896 runtime.memory_durable_proof_for_test(),
7897 Some(memory_context::UserIntentProof::OperatorConfig),
7898 "{name}"
7899 );
7900 }
7901
7902 let runtime = Runtime::new_headless();
7903 runtime
7904 .apply_memory_default_mode(&default_mode_config("recall_once", true))
7905 .expect("applies")
7906 .expect("grants");
7907 assert_eq!(
7908 runtime.memory_context_status().durable,
7909 memory_context::DurableStatus::Off
7910 );
7911 assert_eq!(
7912 runtime.memory_one_shot_proof_for_test(),
7913 Some(memory_context::UserIntentProof::OperatorConfig)
7914 );
7915 }
7916
7917 #[test]
7919 fn memory_default_mode_off_unknown_or_unconfirmed_grants_nothing() {
7920 for (mode, confirmed) in [
7921 ("off", true),
7922 ("off", false),
7923 ("capture_and_recall", false),
7924 ("recall_once", false),
7925 ("recall_everything_forever", true),
7926 ] {
7927 let runtime = Runtime::new_headless();
7928 assert!(
7929 runtime
7930 .apply_memory_default_mode(&default_mode_config(mode, confirmed))
7931 .is_none(),
7932 "{mode}/{confirmed} must not apply"
7933 );
7934 assert_memory_off_no_lease(&runtime);
7935 }
7936 }
7937
7938 #[test]
7941 fn memory_default_mode_keeps_existing_state_and_off_revokes() {
7942 let runtime = Runtime::new_headless();
7943 runtime
7944 .memory_context_enable(
7945 memory_context::MemoryContextMode::RecallEachPrompt,
7946 memory_context::mint_explicit_command_proof(),
7947 )
7948 .unwrap();
7949 assert!(runtime
7950 .apply_memory_default_mode(&default_mode_config("capture_and_recall", true))
7951 .is_none());
7952 assert!(matches!(
7953 runtime.memory_context_status().durable,
7954 memory_context::DurableStatus::Active {
7955 mode: memory_context::MemoryContextMode::RecallEachPrompt,
7956 ..
7957 }
7958 ));
7959 assert!(matches!(
7960 runtime.memory_durable_proof_for_test(),
7961 Some(memory_context::UserIntentProof::ExplicitCommand { .. })
7962 ));
7963
7964 let runtime = Runtime::new_headless();
7965 runtime
7966 .apply_memory_default_mode(&default_mode_config("capture_and_recall", true))
7967 .unwrap()
7968 .unwrap();
7969 runtime.memory_context_disable();
7970 assert_memory_off_no_lease(&runtime);
7971 }
7972
7973 #[test]
7976 fn memory_default_mode_unusable_backend_fails_closed() {
7977 let mut runtime = Runtime::new_headless();
7978 runtime.apply_config(&agent_core::config::load_config_from_str(
7979 "memory.backend = invalid-not-axel-not-legacy\n",
7980 ));
7981 assert!(runtime
7982 .apply_memory_default_mode(&default_mode_config("capture_and_recall", true))
7983 .expect("a confirmed default is attempted")
7984 .is_err());
7985 assert_memory_off_no_lease(&runtime);
7986 }
7987
7988 #[test]
7989 fn memory_backend_apply_config_revokes_preexisting_legacy_lease() {
7990 let mut runtime = Runtime::new_headless();
7991 runtime
7992 .memory_context_enable(
7993 memory_context::MemoryContextMode::CaptureAndRecall,
7994 memory_context::UserIntentProof::ExplicitCommand {
7995 command_id: memory_context::mint_explicit_command_id(),
7996 },
7997 )
7998 .unwrap();
7999 let lease = runtime
8000 .memory_context_lock()
8001 .capture_lease_at(std::time::SystemTime::now())
8002 .unwrap();
8003 runtime.apply_config(&agent_core::config::load_config_from_str(
8004 "memory.backend = axel\n",
8005 ));
8006 assert_memory_off_no_lease(&runtime);
8007 assert!(runtime.extension_capture_provider(&lease).is_none());
8008 }
8009
8010 #[test]
8011 fn axel_invalid_config_cannot_grant_consent_or_fallback() {
8012 let mut runtime = Runtime::new_headless();
8013 runtime.apply_config(&agent_core::config::load_config_from_str(
8014 "memory.backend = invalid-not-axel-not-legacy\n",
8015 ));
8016 assert!(runtime.memory_backend.exclusive());
8017 assert!(runtime.resolve_memory_provider(None).is_err());
8018 assert_memory_off_no_lease(&runtime);
8019 }
8020
8021 #[tokio::test]
8027 async fn memory_enable_unregistered_provider_fails_closed_grants_nothing() {
8028 let (runtime, _ext) =
8030 memory_runtime_with_providers(&[("axel-memory-manager", "project-memory")]).await;
8031 for requested in ["no-such-provider", "bad:id"] {
8032 let err = runtime
8033 .memory_context_enable_resolved(
8034 memory_context::MemoryContextMode::CaptureAndRecall,
8035 memory_context::mint_explicit_command_proof(),
8036 Some(requested),
8037 )
8038 .expect_err("unregistered provider must fail closed");
8039 assert_eq!(
8040 err,
8041 memory_context::MemoryContextError::ProviderNotRegistered
8042 );
8043 assert_memory_off_no_lease(&runtime);
8044 }
8045
8046 let (runtime, _ext) = memory_runtime_with_providers(&[]).await;
8049 let err = runtime
8050 .memory_context_enable(
8051 memory_context::MemoryContextMode::CaptureAndRecall,
8052 memory_context::mint_explicit_command_proof(),
8053 )
8054 .expect_err("empty catalog must fail closed");
8055 assert_eq!(
8056 err,
8057 memory_context::MemoryContextError::ProviderNotRegistered
8058 );
8059 assert_memory_off_no_lease(&runtime);
8060 let err = runtime
8062 .memory_context_recall_once(memory_context::mint_explicit_command_proof())
8063 .expect_err("one-shot grant must validate the provider too");
8064 assert_eq!(
8065 err,
8066 memory_context::MemoryContextError::ProviderNotRegistered
8067 );
8068 assert_memory_off_no_lease(&runtime);
8069 }
8070
8071 #[tokio::test]
8075 async fn memory_enable_exactly_one_declared_provider_records_exact_identity() {
8076 const ADDRESS: &str = "extension:axel-memory-manager:project-memory";
8077 let (runtime, _ext) =
8079 memory_runtime_with_providers(&[("axel-memory-manager", "project-memory")]).await;
8080 let status = runtime
8081 .memory_context_enable(
8082 memory_context::MemoryContextMode::CaptureAndRecall,
8083 memory_context::mint_explicit_command_proof(),
8084 )
8085 .expect("unique declared provider must resolve");
8086 assert!(matches!(
8087 status.durable,
8088 memory_context::DurableStatus::Active { .. }
8089 ));
8090 let bound = runtime.memory_bound_providers_for_test();
8091 assert_eq!(bound.len(), 1);
8092 assert_eq!(bound[0].as_str(), ADDRESS, "exact runtime address recorded");
8093
8094 let (runtime, _ext) =
8096 memory_runtime_with_providers(&[("axel-memory-manager", "project-memory")]).await;
8097 runtime
8098 .memory_context_enable_resolved(
8099 memory_context::MemoryContextMode::RecallEachPrompt,
8100 memory_context::mint_explicit_command_proof(),
8101 Some("project-memory"),
8102 )
8103 .expect("explicit exact id must resolve");
8104 assert_eq!(
8105 runtime.memory_bound_providers_for_test()[0].as_str(),
8106 ADDRESS
8107 );
8108 }
8109
8110 #[tokio::test]
8114 async fn memory_enable_ambiguous_overlapping_declarations_fails_closed() {
8115 let (runtime, _ext) = memory_runtime_with_providers(&[
8116 ("mem-a", "project-memory"),
8117 ("mem-b", "project-memory"),
8118 ])
8119 .await;
8120 let err = runtime
8121 .memory_context_enable(
8122 memory_context::MemoryContextMode::CaptureAndRecall,
8123 memory_context::mint_explicit_command_proof(),
8124 )
8125 .expect_err("overlapping declarations must fail closed");
8126 assert_eq!(err, memory_context::MemoryContextError::ProviderAmbiguous);
8127 assert_memory_off_no_lease(&runtime);
8128
8129 let err = runtime
8131 .memory_context_enable_resolved(
8132 memory_context::MemoryContextMode::CaptureOnly,
8133 memory_context::mint_explicit_command_proof(),
8134 Some("project-memory"),
8135 )
8136 .expect_err("same id declared by two plugins is ambiguous");
8137 assert_eq!(err, memory_context::MemoryContextError::ProviderAmbiguous);
8138 assert_memory_off_no_lease(&runtime);
8139 }
8140
8141 #[tokio::test]
8147 async fn memory_disable_revokes_runtime_lease_without_spawn_no_panic() {
8148 let (runtime, ext) =
8149 memory_runtime_with_providers(&[("axel-memory-manager", "project-memory")]).await;
8150 runtime
8151 .memory_context_enable(
8152 memory_context::MemoryContextMode::CaptureAndRecall,
8153 memory_context::mint_explicit_command_proof(),
8154 )
8155 .expect("enable succeeds");
8156 assert_eq!(
8157 ext.lease_count(),
8158 0,
8159 "granting a memory lease spawns nothing"
8160 );
8161
8162 let status = runtime.memory_context_disable();
8163 assert_eq!(status.durable, memory_context::DurableStatus::Off);
8164 assert_memory_off_no_lease(&runtime);
8165 assert_eq!(
8166 ext.lease_count(),
8167 0,
8168 "revocation of a never-spawned lease is a no-op"
8169 );
8170
8171 let status = runtime.memory_context_disable();
8173 assert_eq!(status.durable, memory_context::DurableStatus::Off);
8174 assert_eq!(ext.lease_count(), 0);
8175 }
8176
8177 #[test]
8182 fn memory_context_enable_disable_emit_metadata_only_events() {
8183 memory_context::drain_captured_memory_events_for_test();
8184
8185 let runtime = Runtime::new_headless();
8186 runtime
8188 .memory_context_enable(
8189 memory_context::MemoryContextMode::Off,
8190 memory_context::mint_explicit_command_proof(),
8191 )
8192 .expect_err("Off is not an enable mode");
8193 assert!(
8194 memory_context::drain_captured_memory_events_for_test().is_empty(),
8195 "failed enable must emit no event"
8196 );
8197
8198 runtime
8199 .memory_context_enable(
8200 memory_context::MemoryContextMode::CaptureAndRecall,
8201 memory_context::mint_explicit_command_proof(),
8202 )
8203 .expect("enable succeeds");
8204 runtime.memory_context_disable();
8205
8206 let events = memory_context::drain_captured_memory_events_for_test();
8207 assert_eq!(
8208 events
8209 .iter()
8210 .map(|event| (event.event, event.outcome))
8211 .collect::<Vec<_>>(),
8212 vec![
8213 ("memory_context.enabled", "enabled"),
8214 ("memory_context.disabled", "disabled"),
8215 ],
8216 );
8217 assert_eq!(events[0].mode, Some("capture_and_recall"));
8218 assert_eq!(events[1].mode, Some("off"));
8219
8220 let cwd = std::env::current_dir()
8223 .expect("cwd")
8224 .to_string_lossy()
8225 .into_owned();
8226 for event in &events {
8227 let json = serde_json::to_string(event).expect("event serializes");
8228 assert!(
8229 json.contains("\"project_digest\":\"p") || json.contains("project-cwd-"),
8230 "digest identity expected; got: {json}"
8231 );
8232 assert!(
8233 !json.contains(&cwd),
8234 "raw project path leaked into event JSON: {json}"
8235 );
8236 }
8237 }
8238
8239 #[tokio::test]
8245 async fn memory_session_end_guard_drop_fresh_construction_is_off_no_lease() {
8246 let (runtime, ext) =
8247 memory_runtime_with_providers(&[("axel-memory-manager", "project-memory")]).await;
8248 runtime
8249 .memory_context_enable(
8250 memory_context::MemoryContextMode::CaptureAndRecall,
8251 memory_context::mint_explicit_command_proof(),
8252 )
8253 .expect("enable succeeds");
8254 drop(runtime);
8256 assert_eq!(
8257 ext.lease_count(),
8258 0,
8259 "session end must leave no runtime lease (never-spawned reap is a no-op)"
8260 );
8261
8262 let mut fresh = Runtime::new_headless();
8264 fresh.install_extension_runtime(std::sync::Arc::clone(&ext));
8265 assert_memory_off_no_lease(&fresh);
8266 assert_eq!(ext.lease_count(), 0);
8267 }
8268}