mcp_langbase_reasoning/self_improvement/
pipes.rs

1//! Langbase pipe integration for the self-improvement system.
2//!
3//! This module provides a specialized interface to Langbase pipes for the
4//! self-improvement loop. It uses existing pipes with tailored prompts:
5//!
6//! - **reflection-v1**: Diagnosis generation and learning synthesis
7//! - **decision-framework-v1**: Action selection with multi-criteria analysis
8//! - **detection-v1**: Decision validation (bias/fallacy detection)
9//!
10//! # Strategy: Existing Pipes First
11//!
12//! Rather than creating new specialized pipes, we use existing pipes with
13//! carefully crafted prompts. This approach:
14//! - Avoids pipe proliferation
15//! - Reuses proven reasoning patterns
16//! - Allows gradual specialization based on metrics
17
18use std::sync::Arc;
19use std::time::{Duration, Instant};
20
21use serde::{Deserialize, Serialize};
22use tracing::{debug, error, info, warn};
23
24use crate::error::LangbaseError;
25use crate::langbase::{LangbaseClient, Message, PipeRequest, PipeResponse};
26
27use super::allowlist::ActionAllowlist;
28use super::config::SelfImprovementPipeConfig;
29use super::types::{HealthReport, MetricsSnapshot, NormalizedReward, SelfDiagnosis, SuggestedAction};
30
31// ============================================================================
32// Error Types
33// ============================================================================
34
35/// Errors that can occur during pipe operations.
36#[derive(Debug, thiserror::Error)]
37pub enum PipeError {
38    /// Pipe call failed due to network or API error
39    #[error("Pipe '{pipe}' unavailable: {message}")]
40    Unavailable {
41        /// Name of the pipe
42        pipe: String,
43        /// Error message
44        message: String,
45        /// Whether a fallback was used
46        fallback_used: bool,
47    },
48
49    /// Pipe call timed out
50    #[error("Pipe '{pipe}' timed out after {timeout_ms}ms")]
51    Timeout {
52        /// Name of the pipe
53        pipe: String,
54        /// Timeout in milliseconds
55        timeout_ms: u64,
56    },
57
58    /// Failed to parse pipe response
59    #[error("Failed to parse response from '{pipe}': {error}")]
60    ParseFailed {
61        /// Name of the pipe
62        pipe: String,
63        /// Parse error details
64        error: String,
65    },
66
67    /// Langbase client error
68    #[error("Langbase error: {0}")]
69    Langbase(#[from] LangbaseError),
70}
71
72impl PipeError {
73    /// Check if this error indicates the pipe is unavailable.
74    pub fn is_unavailable(&self) -> bool {
75        matches!(self, PipeError::Unavailable { .. } | PipeError::Timeout { .. })
76    }
77}
78
79// ============================================================================
80// Response Types
81// ============================================================================
82
83/// Response from the diagnosis pipe.
84#[derive(Debug, Clone, Serialize, Deserialize)]
85pub struct DiagnosisResponse {
86    /// Root cause analysis
87    pub suspected_cause: String,
88    /// Severity assessment: info, warning, high, critical
89    pub severity: String,
90    /// Confidence in the diagnosis (0.0-1.0)
91    pub confidence: f64,
92    /// Supporting evidence
93    pub evidence: Vec<String>,
94    /// Recommended action type
95    pub recommended_action_type: String,
96    /// Target for the action (parameter name, feature name, etc.)
97    pub action_target: Option<String>,
98    /// Rationale for the recommendation
99    pub rationale: String,
100}
101
102impl Default for DiagnosisResponse {
103    fn default() -> Self {
104        Self {
105            suspected_cause: "Unable to determine cause".to_string(),
106            severity: "info".to_string(),
107            confidence: 0.0,
108            evidence: vec![],
109            recommended_action_type: "no_op".to_string(),
110            action_target: None,
111            rationale: "Diagnosis unavailable".to_string(),
112        }
113    }
114}
115
116/// Action scores from the decision framework.
117#[derive(Debug, Clone, Serialize, Deserialize)]
118pub struct ActionScores {
119    /// Effectiveness score
120    pub effectiveness: f64,
121    /// Risk score (lower = safer)
122    pub risk: f64,
123    /// Reversibility score
124    pub reversibility: f64,
125    /// Historical success rate
126    pub historical_success: f64,
127}
128
129/// Response from the action selection pipe.
130#[derive(Debug, Clone, Serialize, Deserialize)]
131pub struct ActionSelectionResponse {
132    /// Selected action option
133    pub selected_option: String,
134    /// Scores for the selected action
135    pub scores: ActionScores,
136    /// Total composite score
137    pub total_score: f64,
138    /// Rationale for selection
139    pub rationale: String,
140    /// Other options that were considered
141    pub alternatives_considered: Vec<String>,
142}
143
144impl Default for ActionSelectionResponse {
145    fn default() -> Self {
146        Self {
147            selected_option: "no_op".to_string(),
148            scores: ActionScores {
149                effectiveness: 0.0,
150                risk: 1.0,
151                reversibility: 0.0,
152                historical_success: 0.0,
153            },
154            total_score: 0.0,
155            rationale: "Action selection unavailable".to_string(),
156            alternatives_considered: vec![],
157        }
158    }
159}
160
161/// Bias/fallacy detection from validation pipe.
162#[derive(Debug, Clone, Serialize, Deserialize)]
163pub struct BiasDetection {
164    /// Type of bias detected
165    pub bias_type: String,
166    /// Severity (1-5)
167    pub severity: i32,
168    /// Explanation
169    pub explanation: String,
170}
171
172/// Fallacy detection from validation pipe.
173#[derive(Debug, Clone, Serialize, Deserialize)]
174pub struct FallacyDetection {
175    /// Type of fallacy detected
176    pub fallacy_type: String,
177    /// Severity (1-5)
178    pub severity: i32,
179    /// Explanation
180    pub explanation: String,
181}
182
183/// Response from the validation pipe.
184#[derive(Debug, Clone, Serialize, Deserialize)]
185pub struct ValidationResponse {
186    /// Detected biases
187    pub biases_detected: Vec<BiasDetection>,
188    /// Detected fallacies
189    pub fallacies_detected: Vec<FallacyDetection>,
190    /// Overall quality score (0.0-1.0)
191    pub overall_quality: f64,
192    /// Whether to proceed with the action
193    pub should_proceed: bool,
194    /// Warnings to log
195    pub warnings: Vec<String>,
196}
197
198impl Default for ValidationResponse {
199    fn default() -> Self {
200        Self {
201            biases_detected: vec![],
202            fallacies_detected: vec![],
203            overall_quality: 0.5,
204            should_proceed: false, // Default to safe behavior
205            warnings: vec!["Validation unavailable - defaulting to safe behavior".to_string()],
206        }
207    }
208}
209
210/// Recommendations from the learning synthesis.
211#[derive(Debug, Clone, Serialize, Deserialize)]
212pub struct LearningRecommendations {
213    /// Whether to adjust the allowlist
214    pub adjust_allowlist: bool,
215    /// Parameters to adjust
216    pub param_adjustments: Vec<ParamAdjustment>,
217    /// Whether to adjust cooldown
218    pub adjust_cooldown: bool,
219    /// New cooldown value if adjusting
220    pub new_cooldown_secs: Option<u64>,
221}
222
223/// Suggested parameter adjustment from learning.
224#[derive(Debug, Clone, Serialize, Deserialize)]
225pub struct ParamAdjustment {
226    /// Parameter key
227    pub key: String,
228    /// Direction: "increase" or "decrease"
229    pub direction: String,
230    /// Reason for the adjustment
231    pub reason: String,
232}
233
234/// Response from the learning synthesis pipe.
235#[derive(Debug, Clone, Serialize, Deserialize)]
236pub struct LearningResponse {
237    /// Assessment of the outcome
238    pub outcome_assessment: String,
239    /// Accuracy of the root cause analysis (0.0-1.0)
240    pub root_cause_accuracy: f64,
241    /// Effectiveness of the action (0.0-1.0)
242    pub action_effectiveness: f64,
243    /// Key lessons learned
244    pub lessons: Vec<String>,
245    /// Recommendations for future actions
246    pub recommendations: LearningRecommendations,
247    /// Confidence in the analysis (0.0-1.0)
248    pub confidence: f64,
249}
250
251impl Default for LearningResponse {
252    fn default() -> Self {
253        Self {
254            outcome_assessment: "Learning synthesis unavailable".to_string(),
255            root_cause_accuracy: 0.0,
256            action_effectiveness: 0.0,
257            lessons: vec![],
258            recommendations: LearningRecommendations {
259                adjust_allowlist: false,
260                param_adjustments: vec![],
261                adjust_cooldown: false,
262                new_cooldown_secs: None,
263            },
264            confidence: 0.0,
265        }
266    }
267}
268
269// ============================================================================
270// Pipe Metrics
271// ============================================================================
272
273/// Metrics for a single pipe call.
274#[derive(Debug, Clone)]
275pub struct PipeCallMetrics {
276    /// Which pipe was called
277    pub pipe_name: String,
278    /// Latency in milliseconds
279    pub latency_ms: u64,
280    /// Whether parsing succeeded
281    pub parse_success: bool,
282    /// Whether the call succeeded overall
283    pub call_success: bool,
284}
285
286// ============================================================================
287// Self-Improvement Pipes
288// ============================================================================
289
290/// Self-improvement pipe operations using existing Langbase pipes.
291///
292/// This struct provides specialized methods for each phase of the
293/// self-improvement loop, using existing pipes with tailored prompts.
294#[derive(Clone)]
295pub struct SelfImprovementPipes {
296    langbase: Arc<LangbaseClient>,
297    config: SelfImprovementPipeConfig,
298}
299
300impl SelfImprovementPipes {
301    /// Create a new SelfImprovementPipes instance.
302    pub fn new(langbase: Arc<LangbaseClient>, config: SelfImprovementPipeConfig) -> Self {
303        Self { langbase, config }
304    }
305
306    /// Generate a diagnosis using reflection-v1 pipe.
307    ///
308    /// This analyzes the health report and trigger to determine root cause
309    /// and recommend an action.
310    pub async fn generate_diagnosis(
311        &self,
312        health_report: &HealthReport,
313        trigger: &super::types::TriggerMetric,
314    ) -> Result<(DiagnosisResponse, PipeCallMetrics), PipeError> {
315        let prompt = self.build_diagnosis_prompt(health_report, trigger);
316        let start = Instant::now();
317
318        let response = self
319            .call_pipe_with_timeout(&self.config.diagnosis_pipe, prompt)
320            .await?;
321
322        let latency_ms = start.elapsed().as_millis() as u64;
323
324        let diagnosis = self.parse_diagnosis_response(&response)?;
325
326        let metrics = PipeCallMetrics {
327            pipe_name: self.config.diagnosis_pipe.clone(),
328            latency_ms,
329            parse_success: true,
330            call_success: true,
331        };
332
333        Ok((diagnosis, metrics))
334    }
335
336    /// Select an action using decision-framework-v1 pipe.
337    ///
338    /// This evaluates the diagnosis against the allowlist and historical
339    /// effectiveness to select the best action.
340    pub async fn select_action(
341        &self,
342        diagnosis: &SelfDiagnosis,
343        allowlist: &ActionAllowlist,
344        history: &[ActionEffectiveness],
345    ) -> Result<(ActionSelectionResponse, PipeCallMetrics), PipeError> {
346        let prompt = self.build_action_selection_prompt(diagnosis, allowlist, history);
347        let start = Instant::now();
348
349        let response = self
350            .call_pipe_with_timeout(&self.config.decision_pipe, prompt)
351            .await?;
352
353        let latency_ms = start.elapsed().as_millis() as u64;
354
355        let selection = self.parse_action_selection_response(&response)?;
356
357        let metrics = PipeCallMetrics {
358            pipe_name: self.config.decision_pipe.clone(),
359            latency_ms,
360            parse_success: true,
361            call_success: true,
362        };
363
364        Ok((selection, metrics))
365    }
366
367    /// Validate a decision using detection-v1 pipe.
368    ///
369    /// This checks for cognitive biases and logical fallacies in the
370    /// diagnosis and action selection process.
371    pub async fn validate_decision(
372        &self,
373        diagnosis: &SelfDiagnosis,
374        action: &SuggestedAction,
375    ) -> Result<(ValidationResponse, PipeCallMetrics), PipeError> {
376        if !self.config.enable_validation {
377            // Validation disabled - return default approval
378            return Ok((
379                ValidationResponse {
380                    biases_detected: vec![],
381                    fallacies_detected: vec![],
382                    overall_quality: 1.0,
383                    should_proceed: true,
384                    warnings: vec![],
385                },
386                PipeCallMetrics {
387                    pipe_name: self.config.detection_pipe.clone(),
388                    latency_ms: 0,
389                    parse_success: true,
390                    call_success: true,
391                },
392            ));
393        }
394
395        let prompt = self.build_validation_prompt(diagnosis, action);
396        let start = Instant::now();
397
398        let response = self
399            .call_pipe_with_timeout(&self.config.detection_pipe, prompt)
400            .await?;
401
402        let latency_ms = start.elapsed().as_millis() as u64;
403
404        let validation = self.parse_validation_response(&response)?;
405
406        let metrics = PipeCallMetrics {
407            pipe_name: self.config.detection_pipe.clone(),
408            latency_ms,
409            parse_success: true,
410            call_success: true,
411        };
412
413        Ok((validation, metrics))
414    }
415
416    /// Synthesize learning from an executed action using reflection-v1 pipe.
417    ///
418    /// This analyzes the before/after metrics to extract lessons and
419    /// recommendations for future improvements.
420    pub async fn synthesize_learning(
421        &self,
422        action: &SuggestedAction,
423        diagnosis: &SelfDiagnosis,
424        pre_metrics: &MetricsSnapshot,
425        post_metrics: &MetricsSnapshot,
426        reward: &NormalizedReward,
427    ) -> Result<(LearningResponse, PipeCallMetrics), PipeError> {
428        let prompt =
429            self.build_learning_prompt(action, diagnosis, pre_metrics, post_metrics, reward);
430        let start = Instant::now();
431
432        let response = self
433            .call_pipe_with_timeout(&self.config.learning_pipe, prompt)
434            .await?;
435
436        let latency_ms = start.elapsed().as_millis() as u64;
437
438        let learning = self.parse_learning_response(&response)?;
439
440        let metrics = PipeCallMetrics {
441            pipe_name: self.config.learning_pipe.clone(),
442            latency_ms,
443            parse_success: true,
444            call_success: true,
445        };
446
447        Ok((learning, metrics))
448    }
449
450    // ========================================================================
451    // Internal: Pipe Calls
452    // ========================================================================
453
454    async fn call_pipe_with_timeout(
455        &self,
456        pipe_name: &str,
457        prompt: String,
458    ) -> Result<PipeResponse, PipeError> {
459        let timeout = Duration::from_millis(self.config.pipe_timeout_ms);
460
461        let messages = vec![Message::user(prompt)];
462        let request = PipeRequest::new(pipe_name, messages);
463
464        debug!(pipe = %pipe_name, "Calling self-improvement pipe");
465
466        match tokio::time::timeout(timeout, self.langbase.call_pipe(request)).await {
467            Ok(Ok(response)) => {
468                info!(pipe = %pipe_name, "Self-improvement pipe call succeeded");
469                Ok(response)
470            }
471            Ok(Err(e)) => {
472                error!(pipe = %pipe_name, error = %e, "Self-improvement pipe call failed");
473                Err(PipeError::Unavailable {
474                    pipe: pipe_name.to_string(),
475                    message: e.to_string(),
476                    fallback_used: false,
477                })
478            }
479            Err(_) => {
480                warn!(pipe = %pipe_name, timeout_ms = self.config.pipe_timeout_ms, "Self-improvement pipe call timed out");
481                Err(PipeError::Timeout {
482                    pipe: pipe_name.to_string(),
483                    timeout_ms: self.config.pipe_timeout_ms,
484                })
485            }
486        }
487    }
488
489    // ========================================================================
490    // Internal: Prompt Building
491    // ========================================================================
492
493    fn build_diagnosis_prompt(
494        &self,
495        health_report: &HealthReport,
496        trigger: &super::types::TriggerMetric,
497    ) -> String {
498        let trigger_json =
499            serde_json::to_string_pretty(trigger).unwrap_or_else(|_| format!("{:?}", trigger));
500        let baselines_json = serde_json::to_string_pretty(&health_report.baselines)
501            .unwrap_or_else(|_| "{}".to_string());
502        let metrics_json = serde_json::to_string_pretty(&health_report.current_metrics)
503            .unwrap_or_else(|_| "{}".to_string());
504
505        format!(
506            r#"## Self-Improvement System Diagnosis Request
507
508### Context
509You are analyzing system health data to diagnose issues and recommend actions.
510This is an autonomous self-improvement system for an MCP reasoning server.
511
512### Trigger Event
513```json
514{trigger_json}
515```
516
517### Current Metrics
518```json
519{metrics_json}
520```
521
522### Baseline Values
523```json
524{baselines_json}
525```
526
527### Task
528Analyze this data and provide a diagnosis. Respond with a JSON object:
529
530```json
531{{
532  "suspected_cause": "Root cause analysis - what is likely causing this trigger",
533  "severity": "info|warning|high|critical",
534  "confidence": 0.0-1.0,
535  "evidence": ["evidence point 1", "evidence point 2"],
536  "recommended_action_type": "adjust_param|toggle_feature|scale_resource|restart_service|clear_cache|no_op",
537  "action_target": "parameter or feature name if applicable",
538  "rationale": "Why this action would help"
539}}
540```
541
542Focus on:
5431. Identifying the most likely root cause
5442. Recommending safe, reversible actions
5453. Providing clear rationale"#
546        )
547    }
548
549    fn build_action_selection_prompt(
550        &self,
551        diagnosis: &SelfDiagnosis,
552        allowlist: &ActionAllowlist,
553        history: &[ActionEffectiveness],
554    ) -> String {
555        let diagnosis_json =
556            serde_json::to_string_pretty(diagnosis).unwrap_or_else(|_| "{}".to_string());
557        let allowlist_summary = format!("{}", allowlist.summary());
558        let history_json =
559            serde_json::to_string_pretty(history).unwrap_or_else(|_| "[]".to_string());
560
561        format!(
562            r#"## Self-Improvement Action Selection
563
564### Diagnosis
565```json
566{diagnosis_json}
567```
568
569### Available Actions (Allowlist)
570{allowlist_summary}
571
572### Historical Effectiveness
573```json
574{history_json}
575```
576
577### Task
578Select the best action from the allowlist. Respond with a JSON object:
579
580```json
581{{
582  "selected_option": "action type and target",
583  "scores": {{
584    "effectiveness": 0.0-1.0,
585    "risk": 0.0-1.0,
586    "reversibility": 0.0-1.0,
587    "historical_success": 0.0-1.0
588  }},
589  "total_score": 0.0-1.0,
590  "rationale": "Why this action is the best choice",
591  "alternatives_considered": ["other options that were evaluated"]
592}}
593```
594
595Important:
5961. Only select actions within the allowlist bounds
5972. Prefer reversible actions
5983. Consider historical success rates
5994. Balance effectiveness against risk"#
600        )
601    }
602
603    fn build_validation_prompt(
604        &self,
605        diagnosis: &SelfDiagnosis,
606        action: &SuggestedAction,
607    ) -> String {
608        let diagnosis_json =
609            serde_json::to_string_pretty(diagnosis).unwrap_or_else(|_| "{}".to_string());
610        let action_json =
611            serde_json::to_string_pretty(action).unwrap_or_else(|_| "{}".to_string());
612
613        format!(
614            r#"## Self-Improvement Decision Validation
615
616### Diagnosis
617```json
618{diagnosis_json}
619```
620
621### Proposed Action
622```json
623{action_json}
624```
625
626### Task
627Validate this diagnosis and action for cognitive biases and logical fallacies.
628Respond with a JSON object:
629
630```json
631{{
632  "biases_detected": [
633    {{"bias_type": "name", "severity": 1-5, "explanation": "why this is a concern"}}
634  ],
635  "fallacies_detected": [
636    {{"fallacy_type": "name", "severity": 1-5, "explanation": "why this is a concern"}}
637  ],
638  "overall_quality": 0.0-1.0,
639  "should_proceed": true/false,
640  "warnings": ["any important caveats"]
641}}
642```
643
644Check for:
6451. Confirmation bias (only seeing supporting evidence)
6462. Anchoring bias (over-relying on first data point)
6473. Hasty generalization (insufficient samples)
6484. False cause fallacy (correlation != causation)
6495. Bandwagon fallacy (because it worked before)"#
650        )
651    }
652
653    fn build_learning_prompt(
654        &self,
655        action: &SuggestedAction,
656        diagnosis: &SelfDiagnosis,
657        pre_metrics: &MetricsSnapshot,
658        post_metrics: &MetricsSnapshot,
659        reward: &NormalizedReward,
660    ) -> String {
661        let action_json =
662            serde_json::to_string_pretty(action).unwrap_or_else(|_| "{}".to_string());
663        let diagnosis_json =
664            serde_json::to_string_pretty(diagnosis).unwrap_or_else(|_| "{}".to_string());
665        let pre_json =
666            serde_json::to_string_pretty(pre_metrics).unwrap_or_else(|_| "{}".to_string());
667        let post_json =
668            serde_json::to_string_pretty(post_metrics).unwrap_or_else(|_| "{}".to_string());
669        let reward_json =
670            serde_json::to_string_pretty(reward).unwrap_or_else(|_| "{}".to_string());
671
672        format!(
673            r#"## Self-Improvement Learning Synthesis
674
675### Original Diagnosis
676```json
677{diagnosis_json}
678```
679
680### Executed Action
681```json
682{action_json}
683```
684
685### Metrics Before
686```json
687{pre_json}
688```
689
690### Metrics After
691```json
692{post_json}
693```
694
695### Calculated Reward
696```json
697{reward_json}
698```
699
700### Task
701Synthesize learning from this action execution. Respond with a JSON object:
702
703```json
704{{
705  "outcome_assessment": "summary of what happened",
706  "root_cause_accuracy": 0.0-1.0,
707  "action_effectiveness": 0.0-1.0,
708  "lessons": ["lesson 1", "lesson 2"],
709  "recommendations": {{
710    "adjust_allowlist": true/false,
711    "param_adjustments": [
712      {{"key": "param name", "direction": "increase|decrease", "reason": "why"}}
713    ],
714    "adjust_cooldown": true/false,
715    "new_cooldown_secs": null or number
716  }},
717  "confidence": 0.0-1.0
718}}
719```
720
721Focus on:
7221. Was the root cause diagnosis accurate?
7232. Did the action have the intended effect?
7243. What can we learn for future actions?
7254. Should we adjust any parameters or thresholds?"#
726        )
727    }
728
729    // ========================================================================
730    // Internal: Response Parsing
731    // ========================================================================
732
733    fn parse_diagnosis_response(
734        &self,
735        response: &PipeResponse,
736    ) -> Result<DiagnosisResponse, PipeError> {
737        let json_str = extract_json(&response.completion);
738
739        serde_json::from_str::<DiagnosisResponse>(&json_str).map_err(|e| {
740            warn!(
741                error = %e,
742                completion_preview = %response.completion.chars().take(200).collect::<String>(),
743                "Failed to parse diagnosis response"
744            );
745            PipeError::ParseFailed {
746                pipe: self.config.diagnosis_pipe.clone(),
747                error: e.to_string(),
748            }
749        })
750    }
751
752    fn parse_action_selection_response(
753        &self,
754        response: &PipeResponse,
755    ) -> Result<ActionSelectionResponse, PipeError> {
756        let json_str = extract_json(&response.completion);
757
758        serde_json::from_str::<ActionSelectionResponse>(&json_str).map_err(|e| {
759            warn!(
760                error = %e,
761                completion_preview = %response.completion.chars().take(200).collect::<String>(),
762                "Failed to parse action selection response"
763            );
764            PipeError::ParseFailed {
765                pipe: self.config.decision_pipe.clone(),
766                error: e.to_string(),
767            }
768        })
769    }
770
771    fn parse_validation_response(
772        &self,
773        response: &PipeResponse,
774    ) -> Result<ValidationResponse, PipeError> {
775        let json_str = extract_json(&response.completion);
776
777        serde_json::from_str::<ValidationResponse>(&json_str).map_err(|e| {
778            warn!(
779                error = %e,
780                completion_preview = %response.completion.chars().take(200).collect::<String>(),
781                "Failed to parse validation response"
782            );
783            PipeError::ParseFailed {
784                pipe: self.config.detection_pipe.clone(),
785                error: e.to_string(),
786            }
787        })
788    }
789
790    fn parse_learning_response(
791        &self,
792        response: &PipeResponse,
793    ) -> Result<LearningResponse, PipeError> {
794        let json_str = extract_json(&response.completion);
795
796        serde_json::from_str::<LearningResponse>(&json_str).map_err(|e| {
797            warn!(
798                error = %e,
799                completion_preview = %response.completion.chars().take(200).collect::<String>(),
800                "Failed to parse learning response"
801            );
802            PipeError::ParseFailed {
803                pipe: self.config.learning_pipe.clone(),
804                error: e.to_string(),
805            }
806        })
807    }
808
809    /// Get the configuration.
810    pub fn config(&self) -> &SelfImprovementPipeConfig {
811        &self.config
812    }
813}
814
815// ============================================================================
816// Helper Types
817// ============================================================================
818
819/// Historical effectiveness data for an action type.
820#[derive(Debug, Clone, Serialize, Deserialize)]
821pub struct ActionEffectiveness {
822    /// Action type
823    pub action_type: String,
824    /// Action signature (for grouping similar actions)
825    pub action_signature: String,
826    /// Total attempts
827    pub total_attempts: u32,
828    /// Successful attempts
829    pub successful_attempts: u32,
830    /// Average reward
831    pub avg_reward: f64,
832    /// Effectiveness score
833    pub effectiveness_score: f64,
834}
835
836// ============================================================================
837// Utility Functions
838// ============================================================================
839
840/// Extract JSON from a completion that may contain markdown code blocks.
841fn extract_json(completion: &str) -> String {
842    // Try to find JSON in markdown code block
843    if let Some(start) = completion.find("```json") {
844        if let Some(end) = completion[start + 7..].find("```") {
845            return completion[start + 7..start + 7 + end].trim().to_string();
846        }
847    }
848
849    // Try to find JSON in generic code block
850    if let Some(start) = completion.find("```") {
851        let after_start = &completion[start + 3..];
852        // Skip language identifier if present
853        let json_start = after_start.find('\n').map(|n| n + 1).unwrap_or(0);
854        if let Some(end) = after_start[json_start..].find("```") {
855            return after_start[json_start..json_start + end].trim().to_string();
856        }
857    }
858
859    // Try to find JSON object directly
860    if let Some(start) = completion.find('{') {
861        if let Some(end) = completion.rfind('}') {
862            if end > start {
863                return completion[start..=end].to_string();
864            }
865        }
866    }
867
868    // Return original if no JSON found
869    completion.to_string()
870}
871
872#[cfg(test)]
873mod tests {
874    use super::*;
875
876    #[test]
877    fn test_extract_json_from_markdown() {
878        let completion = r#"Here is my analysis:
879
880```json
881{
882  "suspected_cause": "High error rate",
883  "severity": "warning"
884}
885```
886
887That's my diagnosis."#;
888
889        let json = extract_json(completion);
890        assert!(json.contains("suspected_cause"));
891        assert!(json.contains("High error rate"));
892    }
893
894    #[test]
895    fn test_extract_json_from_code_block() {
896        let completion = r#"```
897{
898  "key": "value"
899}
900```"#;
901
902        let json = extract_json(completion);
903        assert!(json.contains("key"));
904    }
905
906    #[test]
907    fn test_extract_json_direct() {
908        let completion = r#"{"key": "value"}"#;
909
910        let json = extract_json(completion);
911        assert_eq!(json, r#"{"key": "value"}"#);
912    }
913
914    #[test]
915    fn test_extract_json_with_text() {
916        let completion = r#"Here is some text before {"key": "value"} and after"#;
917
918        let json = extract_json(completion);
919        assert_eq!(json, r#"{"key": "value"}"#);
920    }
921
922    #[test]
923    fn test_diagnosis_response_default() {
924        let response = DiagnosisResponse::default();
925        assert_eq!(response.recommended_action_type, "no_op");
926        assert_eq!(response.confidence, 0.0);
927    }
928
929    #[test]
930    fn test_validation_response_default() {
931        let response = ValidationResponse::default();
932        assert!(!response.should_proceed);
933        assert_eq!(response.warnings.len(), 1);
934    }
935
936    #[test]
937    fn test_learning_response_default() {
938        let response = LearningResponse::default();
939        assert!(!response.recommendations.adjust_allowlist);
940        assert_eq!(response.confidence, 0.0);
941    }
942
943    #[test]
944    fn test_pipe_error_is_unavailable() {
945        let unavailable = PipeError::Unavailable {
946            pipe: "test".to_string(),
947            message: "failed".to_string(),
948            fallback_used: false,
949        };
950        assert!(unavailable.is_unavailable());
951
952        let timeout = PipeError::Timeout {
953            pipe: "test".to_string(),
954            timeout_ms: 30000,
955        };
956        assert!(timeout.is_unavailable());
957
958        let parse_failed = PipeError::ParseFailed {
959            pipe: "test".to_string(),
960            error: "invalid json".to_string(),
961        };
962        assert!(!parse_failed.is_unavailable());
963    }
964
965    #[test]
966    fn test_diagnosis_response_serialization() {
967        let response = DiagnosisResponse {
968            suspected_cause: "High latency".to_string(),
969            severity: "warning".to_string(),
970            confidence: 0.85,
971            evidence: vec!["P95 increased by 50%".to_string()],
972            recommended_action_type: "adjust_param".to_string(),
973            action_target: Some("REQUEST_TIMEOUT_MS".to_string()),
974            rationale: "Increase timeout to handle slow responses".to_string(),
975        };
976
977        let json = serde_json::to_string(&response).unwrap();
978        let parsed: DiagnosisResponse = serde_json::from_str(&json).unwrap();
979
980        assert_eq!(parsed.suspected_cause, response.suspected_cause);
981        assert_eq!(parsed.confidence, response.confidence);
982    }
983
984    #[test]
985    fn test_action_effectiveness_serialization() {
986        let effectiveness = ActionEffectiveness {
987            action_type: "adjust_param".to_string(),
988            action_signature: "REQUEST_TIMEOUT_MS:increase".to_string(),
989            total_attempts: 5,
990            successful_attempts: 4,
991            avg_reward: 0.3,
992            effectiveness_score: 0.8,
993        };
994
995        let json = serde_json::to_string(&effectiveness).unwrap();
996        let parsed: ActionEffectiveness = serde_json::from_str(&json).unwrap();
997
998        assert_eq!(parsed.action_type, effectiveness.action_type);
999        assert_eq!(parsed.effectiveness_score, effectiveness.effectiveness_score);
1000    }
1001}
mcp_langbase_reasoning/self_improvement/pipes.rs

mcp_langbase_reasoning/self_improvement/
pipes.rs
