perspt_agent/
transport.rs

1//! The transport adapter (PSP-9 system 4).
2//!
3//! This is the only place where the SDK's provider-neutral contract and
4//! `perspt-core`'s `genai` driver are both in scope. `GenAiTransport` wraps a
5//! [`perspt_core::ModelPortfolio`] and implements
6//! [`perspt_sdk::ModelTransport`], translating the SDK types to and from the
7//! core mirrors. Everything above the transport — the turn loop, routing,
8//! adjudication, calibration — sees only SDK types (Gate S).
9
10use std::sync::Arc;
11use std::time::Duration;
12
13use perspt_core::{CoreMessage, CoreToolCall, CoreToolChoice, CoreToolSpec, CoreTurnOutput};
14use perspt_sdk::{
15    Conversation, Message, ModelFamily, ModelId, ModelTransport, ProviderCapabilities,
16    ProviderToolCall, SdkError, ToolChoicePolicy, ToolSpec, TransportFuture, TurnOutput,
17};
18
19/// Retries per turn for transient provider failures. A 429 or 5xx is a
20/// throughput delay, not a route failure (Gate P: rate-limit waits stay
21/// inside the transport); only after these bounded retries does an error
22/// reach the recovery cascade, where it may consume a sticky failover.
23const TRANSIENT_RETRIES: u32 = 3;
24const BACKOFF_BASE: Duration = Duration::from_secs(2);
25
26/// The backoff before retry `attempt` (0-based), or `None` when the cause
27/// is not transient or the retry budget is exhausted. Backoff doubles per
28/// attempt: 2s, 4s, 8s.
29fn transient_retry_delay(cause: &str, attempt: u32) -> Option<Duration> {
30    if attempt >= TRANSIENT_RETRIES {
31        return None;
32    }
33    let lowered = cause.to_ascii_lowercase();
34    const TRANSIENT_MARKERS: &[&str] = &[
35        "429",
36        "too many requests",
37        "rate limit",
38        "resource exhausted",
39        "resource_exhausted",
40        "status code '500",
41        "internal server error",
42        "502",
43        "bad gateway",
44        "503",
45        "service unavailable",
46        "504",
47        "gateway timeout",
48        "timed out",
49        "timeout",
50        "connection reset",
51        "connection closed before",
52        "overloaded",
53        "temporarily unavailable",
54    ];
55    if !TRANSIENT_MARKERS
56        .iter()
57        .any(|marker| lowered.contains(marker))
58    {
59        return None;
60    }
61    Some(BACKOFF_BASE * 2u32.pow(attempt))
62}
63
64/// `perspt_sdk::ModelTransport` implemented over the core portfolio.
65pub struct GenAiTransport {
66    portfolio: Arc<perspt_core::ModelPortfolio>,
67}
68
69impl GenAiTransport {
70    /// Wrap a live portfolio.
71    pub fn new(portfolio: Arc<perspt_core::ModelPortfolio>) -> Self {
72        Self { portfolio }
73    }
74
75    /// The wrapped portfolio, for surfaces such as `perspt providers`.
76    pub fn portfolio(&self) -> &perspt_core::ModelPortfolio {
77        &self.portfolio
78    }
79}
80
81/// Render the SDK conversation into core mirror messages.
82fn render_conversation(conversation: &Conversation) -> Vec<CoreMessage> {
83    conversation
84        .messages()
85        .iter()
86        .map(|message| match message {
87            Message::System { content } => CoreMessage::System {
88                content: content.clone(),
89            },
90            Message::User { content } => CoreMessage::User {
91                content: content.clone(),
92            },
93            Message::Assistant { content } => CoreMessage::Assistant {
94                content: content.clone(),
95            },
96            Message::AssistantToolCalls { calls } => CoreMessage::AssistantToolCalls {
97                calls: calls.iter().map(render_call).collect(),
98            },
99            Message::ToolResponse { call_id, content } => CoreMessage::ToolResponse {
100                call_id: call_id.clone(),
101                content: content.clone(),
102            },
103        })
104        .collect()
105}
106
107fn render_call(call: &ProviderToolCall) -> CoreToolCall {
108    CoreToolCall {
109        call_id: call.call_id.clone(),
110        name: call.name.clone(),
111        arguments: call.arguments.clone(),
112    }
113}
114
115fn render_specs(tools: &[ToolSpec]) -> Vec<CoreToolSpec> {
116    tools
117        .iter()
118        .map(|spec| CoreToolSpec {
119            name: spec.name.clone(),
120            description: spec.description.clone(),
121            schema: spec.schema.clone(),
122            strict: spec.strict,
123        })
124        .collect()
125}
126
127fn render_choice(choice: ToolChoicePolicy) -> CoreToolChoice {
128    match choice {
129        ToolChoicePolicy::Auto => CoreToolChoice::Auto,
130        ToolChoicePolicy::None => CoreToolChoice::None,
131        ToolChoicePolicy::Required => CoreToolChoice::Required,
132        ToolChoicePolicy::Specific(name) => CoreToolChoice::Specific(name),
133    }
134}
135
136fn lift_output(output: CoreTurnOutput) -> TurnOutput {
137    match output {
138        CoreTurnOutput::Text(text) => TurnOutput::Text(text),
139        CoreTurnOutput::ToolCalls(calls) => TurnOutput::ToolCalls(
140            calls
141                .into_iter()
142                .map(|call| ProviderToolCall {
143                    call_id: call.call_id,
144                    name: call.name,
145                    arguments: call.arguments,
146                })
147                .collect(),
148        ),
149    }
150}
151
152/// Lift the core capability mirror into the SDK record.
153fn lift_caps(caps: &perspt_core::ProviderCaps) -> ProviderCapabilities {
154    ProviderCapabilities {
155        tool_calling: caps.tool_calling,
156        strict_schema: caps.strict_schema,
157        parallel_tool_calls: caps.parallel_tool_calls,
158        streaming_tool_calls: caps.streaming_tool_calls,
159        prompt_caching: caps.prompt_caching,
160        structured_output: caps.structured_output,
161        max_context_tokens: caps.max_context_tokens,
162    }
163}
164
165impl ModelTransport for GenAiTransport {
166    fn chat_turn<'a>(
167        &'a self,
168        model: &'a ModelId,
169        conversation: &'a Conversation,
170        tools: &'a [ToolSpec],
171        choice: ToolChoicePolicy,
172    ) -> TransportFuture<'a, TurnOutput> {
173        self.chat_turn_with_options(
174            model,
175            conversation,
176            tools,
177            choice,
178            perspt_sdk::GenerationOptions::default(),
179        )
180    }
181
182    fn chat_turn_with_options<'a>(
183        &'a self,
184        model: &'a ModelId,
185        conversation: &'a Conversation,
186        tools: &'a [ToolSpec],
187        choice: ToolChoicePolicy,
188        options: perspt_sdk::GenerationOptions,
189    ) -> TransportFuture<'a, TurnOutput> {
190        let core_choice = render_choice(choice);
191        Box::pin(async move {
192            let handle = self
193                .portfolio
194                .resolve(&model.provider)
195                .map_err(|e| SdkError::Domain(format!("route resolution failed: {e}")))?;
196            let messages = render_conversation(conversation);
197            let specs = render_specs(tools);
198            let mut attempt = 0u32;
199            loop {
200                match handle
201                    .provider
202                    .chat_turn_configured(
203                        &model.model,
204                        &messages,
205                        &specs,
206                        core_choice.clone(),
207                        perspt_core::tools_driver::CoreGenerationConfig {
208                            max_tokens: options.max_tokens,
209                            temperature: options.temperature,
210                            stop_sequences: options.stop_sequences.clone(),
211                        },
212                    )
213                    .await
214                {
215                    Ok(output) => return Ok(lift_output(output)),
216                    Err(error) => {
217                        let cause = format!("{error:#}");
218                        let Some(delay) = transient_retry_delay(&cause, attempt) else {
219                            return Err(SdkError::Domain(format!("chat turn failed: {cause}")));
220                        };
221                        log::warn!(
222                            "transient provider error from {model} (retry {} of {TRANSIENT_RETRIES} \
223                             in {delay:?}): {cause}",
224                            attempt + 1
225                        );
226                        tokio::time::sleep(delay).await;
227                        attempt += 1;
228                    }
229                }
230            }
231        })
232    }
233
234    fn capabilities(&self, model: &ModelId) -> ProviderCapabilities {
235        match self.portfolio.resolve(&model.provider) {
236            Ok(handle) => lift_caps(&handle.caps),
237            // An unknown route declares nothing; the caller's route
238            // resolution surfaces the real error.
239            Err(_) => ProviderCapabilities::text_only(0),
240        }
241    }
242
243    fn family_of(&self, model: &ModelId) -> ModelFamily {
244        ModelFamily::from_model_name(&model.model)
245    }
246
247    /// The genai driver's adapter identity (PSP-10 system 24). One identity
248    /// for the whole driver: endpoint/provider ids stay credentials, and a
249    /// gateway-served model keeps its own family via `family_of`.
250    fn adapter_kind(&self) -> &'static str {
251        "genai"
252    }
253}
254
255#[cfg(test)]
256mod tests {
257    use super::*;
258
259    #[test]
260    fn transient_causes_get_bounded_doubling_backoff() {
261        let cause = "Request failed with status code '429 Too Many Requests'";
262        assert_eq!(
263            transient_retry_delay(cause, 0),
264            Some(Duration::from_secs(2))
265        );
266        assert_eq!(
267            transient_retry_delay(cause, 1),
268            Some(Duration::from_secs(4))
269        );
270        assert_eq!(
271            transient_retry_delay(cause, 2),
272            Some(Duration::from_secs(8))
273        );
274        assert_eq!(transient_retry_delay(cause, 3), None, "retries are bounded");
275        assert!(transient_retry_delay("504 Gateway Timeout", 0).is_some());
276        assert!(transient_retry_delay("connection reset by peer", 0).is_some());
277        assert!(transient_retry_delay("RESOURCE_EXHAUSTED", 0).is_some());
278    }
279
280    #[test]
281    fn non_transient_causes_surface_immediately() {
282        assert_eq!(transient_retry_delay("401 Unauthorized", 0), None);
283        assert_eq!(transient_retry_delay("invalid request schema", 0), None);
284        assert_eq!(transient_retry_delay("model not found", 0), None);
285    }
286
287    #[test]
288    fn conversation_renders_role_for_role() {
289        let mut conversation = Conversation::with_system("harness");
290        conversation.push_user("go");
291        conversation.push_tool_calls(vec![ProviderToolCall {
292            call_id: "c1".into(),
293            name: "read_file".into(),
294            arguments: serde_json::json!({"path": "a"}),
295        }]);
296        conversation.push_tool_response("c1", "contents");
297        let rendered = render_conversation(&conversation);
298        assert_eq!(rendered.len(), 4);
299        assert!(
300            matches!(&rendered[2], CoreMessage::AssistantToolCalls { calls } if calls.len() == 1)
301        );
302        assert!(
303            matches!(&rendered[3], CoreMessage::ToolResponse { call_id, .. } if call_id == "c1")
304        );
305    }
306
307    #[test]
308    fn tool_call_output_lifts_back_to_sdk_types() {
309        let output = lift_output(CoreTurnOutput::ToolCalls(vec![CoreToolCall {
310            call_id: "c9".into(),
311            name: "grep".into(),
312            arguments: serde_json::json!({"query": "fn main"}),
313        }]));
314        assert_eq!(output.tool_calls().len(), 1);
315        assert_eq!(output.tool_calls()[0].name, "grep");
316    }
317
318    #[test]
319    fn family_is_derived_from_the_model_name_not_the_provider_key() {
320        let config =
321            perspt_core::Config::from_toml_str("[providers.local]\nadapter = \"ollama\"\n")
322                .unwrap();
323        let portfolio = Arc::new(perspt_core::ModelPortfolio::from_config(&config).unwrap());
324        let transport = GenAiTransport::new(portfolio);
325        let model = ModelId::new("local", "llama3.3:70b");
326        assert_eq!(transport.family_of(&model), ModelFamily::Llama);
327    }
328}
perspt_agent/transport.rs

perspt_agent/
transport.rs
