| 1 | //! The one pure web adapter bridge and provenance for retry decisions. |
| 2 | //! ToolError remains the execution outcome; this private wrapper has no state |
| 3 | //! or authority and is removed only after the last retry consumer decides. |
| 4 | use crate::tools::spec::{ToolContext, ToolError}; |
| 5 | use serde::de::DeserializeOwned; |
| 6 | use serde_json::Value; |
| 7 | use std::time::Duration; |
| 8 | |
| 9 | // One per-attempt diagnostic bit, never an execution/approval authority. The |
| 10 | // chain owns it while polling a backend so its outer timeout distinguishes a |
| 11 | // pending Host RPC from a Core transport timeout. Cancellation leaves the bit |
| 12 | // set until that caller reads it; there is no global or cached origin. |
| 13 | tokio::task_local! { |
| 14 | pub(super) static HOST_PENDING: std::sync::Arc<std::sync::atomic::AtomicBool>; |
| 15 | } |
| 16 | |
| 17 | #[derive(Debug, Clone, Copy, PartialEq, Eq)] |
| 18 | pub(crate) enum FailureOrigin { |
| 19 | ContentOrProvider, |
| 20 | Host, |
| 21 | CaptureGuard, |
| 22 | } |
| 23 | #[derive(Debug)] |
| 24 | pub(crate) struct AdapterFailure { |
| 25 | pub(crate) origin: FailureOrigin, |
| 26 | pub(crate) error: ToolError, |
| 27 | } |
| 28 | pub(crate) type AdapterResult<T> = Result<T, AdapterFailure>; |
| 29 | impl AdapterFailure { |
| 30 | pub(crate) fn host(error: ToolError) -> Self { |
| 31 | Self { |
| 32 | origin: FailureOrigin::Host, |
| 33 | error, |
| 34 | } |
| 35 | } |
| 36 | pub(crate) fn capture(error: ToolError) -> Self { |
| 37 | Self { |
| 38 | origin: FailureOrigin::CaptureGuard, |
| 39 | error, |
| 40 | } |
| 41 | } |
| 42 | pub(crate) fn content(&self) -> bool { |
| 43 | self.origin == FailureOrigin::ContentOrProvider |
| 44 | } |
| 45 | } |
| 46 | impl From<ToolError> for AdapterFailure { |
| 47 | fn from(error: ToolError) -> Self { |
| 48 | Self { |
| 49 | origin: FailureOrigin::ContentOrProvider, |
| 50 | error, |
| 51 | } |
| 52 | } |
| 53 | } |
| 54 | impl From<AdapterFailure> for ToolError { |
| 55 | fn from(error: AdapterFailure) -> Self { |
| 56 | error.error |
| 57 | } |
| 58 | } |
| 59 | impl std::fmt::Display for AdapterFailure { |
| 60 | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 61 | self.error.fmt(f) |
| 62 | } |
| 63 | } |
| 64 | impl std::error::Error for AdapterFailure {} |
| 65 | |
| 66 | pub(crate) async fn transform<T: DeserializeOwned>( |
| 67 | operation: crate::extension_host::StockOperation, |
| 68 | input: Value, |
| 69 | context: &ToolContext, |
| 70 | budget: Duration, |
| 71 | ) -> AdapterResult<T> { |
| 72 | // Preserve the existing stock cap before broker admission. No partial body |
| 73 | // or automatic backend fallback is a substitute for a full captured value. |
| 74 | let size = serde_json::to_vec(&input) |
| 75 | .map_err(|_| { |
| 76 | AdapterFailure::capture(ToolError::execution_failed( |
| 77 | "Web adapter capture is not JSON", |
| 78 | )) |
| 79 | })? |
| 80 | .len(); |
| 81 | if size > 1024 * 1024 { |
| 82 | return Err(AdapterFailure::capture(ToolError::execution_failed( |
| 83 | "Web adapter capture exceeds the 1 MiB safety limit; no Host fallback was attempted", |
| 84 | ))); |
| 85 | } |
| 86 | let _ = |
| 87 | HOST_PENDING.try_with(|pending| pending.store(true, std::sync::atomic::Ordering::SeqCst)); |
| 88 | let result = crate::extension_host::manager() |
| 89 | .execute_stock(operation, input, context, budget) |
| 90 | .await; |
| 91 | let _ = |
| 92 | HOST_PENDING.try_with(|pending| pending.store(false, std::sync::atomic::Ordering::SeqCst)); |
| 93 | let result = result.map_err(AdapterFailure::host)?; |
| 94 | if !result.success { |
| 95 | return Err(AdapterFailure::host(ToolError::execution_failed( |
| 96 | "Web Host returned an invalid transform result; no fallback was attempted", |
| 97 | ))); |
| 98 | } |
| 99 | serde_json::from_value(result.metadata.ok_or_else(|| { |
| 100 | AdapterFailure::host(ToolError::execution_failed( |
| 101 | "Web Host omitted its transform result", |
| 102 | )) |
| 103 | })?) |
| 104 | .map_err(|_| { |
| 105 | AdapterFailure::host(ToolError::execution_failed( |
| 106 | "Web Host returned a malformed transform proposal; no fallback was attempted", |
| 107 | )) |
| 108 | }) |
| 109 | } |
| 110 | |
| 111 | pub(crate) fn search_selected(context: &ToolContext) -> bool { |
| 112 | context |
| 113 | .features |
| 114 | .enabled(crate::features::Feature::WebSearchHost) |
| 115 | } |
| 116 | |
| 117 | /// Read the complete provider body under the selected Host capture cap. A cap |
| 118 | /// refusal is never an empty result and never earns another backend request. |
| 119 | pub(crate) async fn read_response( |
| 120 | response: reqwest::Response, |
| 121 | context: &ToolContext, |
| 122 | ) -> AdapterResult<String> { |
| 123 | read_response_with_limit(response, context, 1024 * 1024).await |
| 124 | } |
| 125 | |
| 126 | pub(crate) async fn read_response_with_limit( |
| 127 | mut response: reqwest::Response, |
| 128 | context: &ToolContext, |
| 129 | maximum: usize, |
| 130 | ) -> AdapterResult<String> { |
| 131 | if !search_selected(context) { |
| 132 | return response.text().await.map_err(|error| { |
| 133 | ToolError::execution_failed(format!("Failed to read search response: {error}")).into() |
| 134 | }); |
| 135 | } |
| 136 | let content_type = response |
| 137 | .headers() |
| 138 | .get(reqwest::header::CONTENT_TYPE) |
| 139 | .and_then(|value| value.to_str().ok()) |
| 140 | .map(str::to_owned); |
| 141 | let read = async { |
| 142 | let mut bytes = Vec::new(); |
| 143 | while let Some(chunk) = response.chunk().await.map_err(|error| { |
| 144 | ToolError::execution_failed(format!("Failed to read search response: {error}")) |
| 145 | })? { |
| 146 | if chunk.len() > maximum.saturating_sub(bytes.len()) { |
| 147 | return Err(AdapterFailure::capture(ToolError::execution_failed( |
| 148 | format!( |
| 149 | "Web provider response exceeds the {maximum} byte Host capture limit; no fallback was attempted" |
| 150 | ), |
| 151 | ))); |
| 152 | } |
| 153 | bytes.extend_from_slice(&chunk); |
| 154 | } |
| 155 | // Reuse the single bounded charset/NUL decoder. The selected Host path |
| 156 | // refuses binary provider JSON instead of shipping a partial capture. |
| 157 | super::extract::decode_response_body(&bytes, content_type.as_deref(), false) |
| 158 | .map_err(AdapterFailure::capture) |
| 159 | }; |
| 160 | tokio::pin!(read); |
| 161 | let cancel = async { |
| 162 | match context.cancel_token.as_ref() { |
| 163 | Some(token) => token.cancelled().await, |
| 164 | None => std::future::pending().await, |
| 165 | } |
| 166 | }; |
| 167 | let deadline = async { |
| 168 | match context.turn_deadline { |
| 169 | Some(deadline) => tokio::time::sleep_until(deadline).await, |
| 170 | None => std::future::pending().await, |
| 171 | } |
| 172 | }; |
| 173 | tokio::select! { |
| 174 | biased; |
| 175 | () = cancel => Err(ToolError::cancelled("web provider capture").into()), |
| 176 | () = deadline => Err(ToolError::Timeout { seconds: 0 }.into()), |
| 177 | result = &mut read => result, |
| 178 | } |
| 179 | } |
| 180 |