tooltest_core/
lints.rs

1use std::collections::{HashMap, HashSet};
2
3use chrono::NaiveDate;
4use serde_json::json;
5
6use crate::coverage_filter::is_coverage_tool_eligible;
7use crate::output_schema::compile_output_schema;
8pub use crate::schema_dialect::DEFAULT_JSON_SCHEMA_DIALECT;
9use crate::schema_dialect::{
10    normalize_schema_id, DRAFT4_HTTP, DRAFT4_HTTPS, DRAFT6_HTTP, DRAFT6_HTTPS, DRAFT7_HTTP,
11    DRAFT7_HTTPS,
12};
13use crate::{
14    CoverageRule, LintDefinition, LintFinding, LintLevel, LintPhase, LintRule, ListLintContext,
15    ResponseLintContext, RunLintContext,
16};
17
18fn schema_id_from_object(schema: &crate::JsonObject) -> Option<&str> {
19    schema.get("$schema").and_then(|value| value.as_str())
20}
21
22/// Lint: checks the raw tools/list count against a configured maximum.
23#[derive(Clone, Debug)]
24pub struct MaxToolsLint {
25    definition: LintDefinition,
26    max_tools: usize,
27}
28
29impl MaxToolsLint {
30    pub fn new(definition: LintDefinition, max_tools: usize) -> Self {
31        Self {
32            definition,
33            max_tools,
34        }
35    }
36}
37
38impl LintRule for MaxToolsLint {
39    fn definition(&self) -> &LintDefinition {
40        &self.definition
41    }
42
43    fn check_list(&self, context: &ListLintContext<'_>) -> Vec<LintFinding> {
44        if context.raw_tool_count <= self.max_tools {
45            return Vec::new();
46        }
47        let message = format!(
48            "tools/list returned {} tools (max {})",
49            context.raw_tool_count, self.max_tools
50        );
51        vec![LintFinding::new(message).with_details(json!({
52            "count": context.raw_tool_count,
53            "max": self.max_tools,
54        }))]
55    }
56}
57
58/// Lint: enforces a minimum MCP protocol version based on initialize response.
59#[derive(Clone, Debug)]
60pub struct McpSchemaMinVersionLint {
61    definition: LintDefinition,
62    min_version: NaiveDate,
63    min_version_raw: String,
64}
65
66impl McpSchemaMinVersionLint {
67    pub fn new(definition: LintDefinition, min_version: impl Into<String>) -> Result<Self, String> {
68        let min_version_raw = min_version.into();
69        let min_version =
70            NaiveDate::parse_from_str(&min_version_raw, "%Y-%m-%d").map_err(|_| {
71                format!("invalid minimum protocol version '{min_version_raw}'; expected YYYY-MM-DD")
72            })?;
73        Ok(Self {
74            definition,
75            min_version,
76            min_version_raw,
77        })
78    }
79}
80
81impl LintRule for McpSchemaMinVersionLint {
82    fn definition(&self) -> &LintDefinition {
83        &self.definition
84    }
85
86    fn check_list(&self, context: &ListLintContext<'_>) -> Vec<LintFinding> {
87        let Some(protocol_version) = context.protocol_version else {
88            return vec![LintFinding::new("server did not report protocolVersion")];
89        };
90        let trimmed = protocol_version.trim();
91        if trimmed.is_empty() {
92            return vec![LintFinding::new("server reported an empty protocolVersion")];
93        }
94        let parsed = match NaiveDate::parse_from_str(trimmed, "%Y-%m-%d") {
95            Ok(parsed) => parsed,
96            Err(_) => {
97                return vec![LintFinding::new(format!(
98                    "server protocolVersion '{trimmed}' is not YYYY-MM-DD"
99                ))
100                .with_details(json!({
101                    "reported": trimmed,
102                    "expected_format": "YYYY-MM-DD",
103                }))];
104            }
105        };
106        if parsed < self.min_version {
107            return vec![LintFinding::new(format!(
108                "server protocolVersion '{trimmed}' is below minimum {}",
109                self.min_version_raw
110            ))
111            .with_details(json!({
112                "reported": trimmed,
113                "minimum": self.min_version_raw,
114            }))];
115        }
116        Vec::new()
117    }
118}
119
120/// Lint: validates tool schema dialects against an allowlist.
121#[derive(Clone, Debug)]
122pub struct JsonSchemaDialectCompatLint {
123    definition: LintDefinition,
124    allowlist: HashSet<String>,
125}
126
127impl JsonSchemaDialectCompatLint {
128    pub fn new(definition: LintDefinition, allowlist: impl IntoIterator<Item = String>) -> Self {
129        let allowlist = allowlist
130            .into_iter()
131            .map(|entry| normalize_schema_id(&entry).to_string())
132            .collect();
133        Self {
134            definition,
135            allowlist,
136        }
137    }
138
139    fn check_schema(
140        &self,
141        tool_name: &str,
142        schema: &crate::JsonObject,
143        label: &str,
144    ) -> Option<LintFinding> {
145        let declared = schema_id_from_object(schema)
146            .map(normalize_schema_id)
147            .unwrap_or_else(|| normalize_schema_id(DEFAULT_JSON_SCHEMA_DIALECT));
148        if self.allowlist.contains(declared) {
149            return None;
150        }
151        Some(
152            LintFinding::new(format!(
153                "tool '{}' {label} schema declares unsupported dialect '{declared}'",
154                tool_name
155            ))
156            .with_details(json!({
157                "tool": tool_name,
158                "schema": declared,
159                "schema_label": label,
160            })),
161        )
162    }
163}
164
165impl LintRule for JsonSchemaDialectCompatLint {
166    fn definition(&self) -> &LintDefinition {
167        &self.definition
168    }
169
170    fn check_list(&self, context: &ListLintContext<'_>) -> Vec<LintFinding> {
171        let mut findings = Vec::new();
172        for tool in context.tools {
173            if let Some(finding) =
174                self.check_schema(tool.name.as_ref(), tool.input_schema.as_ref(), "input")
175            {
176                findings.push(finding);
177            }
178            if let Some(schema) = tool.output_schema.as_ref() {
179                if let Some(finding) =
180                    self.check_schema(tool.name.as_ref(), schema.as_ref(), "output")
181                {
182                    findings.push(finding);
183                }
184            }
185        }
186        findings
187    }
188}
189
190/// Lint: reports `$defs` usage with legacy JSON Schema drafts.
191#[derive(Clone, Debug)]
192pub struct JsonSchemaKeywordCompatLint {
193    definition: LintDefinition,
194}
195
196impl JsonSchemaKeywordCompatLint {
197    pub fn new(definition: LintDefinition) -> Self {
198        Self { definition }
199    }
200
201    fn is_legacy_schema_id(schema_id: &str) -> bool {
202        matches!(
203            schema_id,
204            DRAFT7_HTTP | DRAFT7_HTTPS | DRAFT6_HTTP | DRAFT6_HTTPS | DRAFT4_HTTP | DRAFT4_HTTPS
205        )
206    }
207
208    fn check_schema(
209        &self,
210        tool_name: &str,
211        schema: &crate::JsonObject,
212        label: &str,
213    ) -> Option<LintFinding> {
214        if !schema.contains_key("$defs") {
215            return None;
216        }
217        let declared = schema_id_from_object(schema)
218            .map(normalize_schema_id)
219            .unwrap_or(DEFAULT_JSON_SCHEMA_DIALECT);
220        if !Self::is_legacy_schema_id(declared) {
221            return None;
222        }
223        Some(
224            LintFinding::new(format!(
225                "tool '{}' {label} schema declares {declared} but uses '$defs'; draft-07 and earlier use 'definitions'",
226                tool_name
227            ))
228            .with_details(json!({
229                "tool": tool_name,
230                "schema": declared,
231                "schema_label": label,
232                "keyword": "$defs",
233            })),
234        )
235    }
236}
237
238impl LintRule for JsonSchemaKeywordCompatLint {
239    fn definition(&self) -> &LintDefinition {
240        &self.definition
241    }
242
243    fn check_list(&self, context: &ListLintContext<'_>) -> Vec<LintFinding> {
244        let mut findings = Vec::new();
245        for tool in context.tools {
246            if let Some(finding) =
247                self.check_schema(tool.name.as_ref(), tool.input_schema.as_ref(), "input")
248            {
249                findings.push(finding);
250            }
251            if let Some(schema) = tool.output_schema.as_ref() {
252                if let Some(finding) =
253                    self.check_schema(tool.name.as_ref(), schema.as_ref(), "output")
254                {
255                    findings.push(finding);
256                }
257            }
258        }
259        findings
260    }
261}
262
263/// Lint: reports output schemas that fail to compile.
264#[derive(Clone, Debug)]
265pub struct OutputSchemaCompileLint {
266    definition: LintDefinition,
267}
268
269impl OutputSchemaCompileLint {
270    pub fn new(definition: LintDefinition) -> Self {
271        Self { definition }
272    }
273}
274
275impl LintRule for OutputSchemaCompileLint {
276    fn definition(&self) -> &LintDefinition {
277        &self.definition
278    }
279
280    fn check_list(&self, context: &ListLintContext<'_>) -> Vec<LintFinding> {
281        let mut findings = Vec::new();
282        for tool in context.tools {
283            let Some(schema) = tool.output_schema.as_ref() else {
284                continue;
285            };
286            if let Err(error) = compile_output_schema(schema.as_ref()) {
287                findings.push(
288                    LintFinding::new(format!(
289                        "tool '{}' output schema failed to compile",
290                        tool.name.as_ref()
291                    ))
292                    .with_details(json!({
293                        "tool": tool.name.as_ref(),
294                        "error": error,
295                    })),
296                );
297            }
298        }
299        findings
300    }
301}
302
303/// Lint: enforces a maximum structuredContent byte size per response.
304#[derive(Clone, Debug)]
305pub struct MaxStructuredContentBytesLint {
306    definition: LintDefinition,
307    max_bytes: usize,
308    serialize: fn(&serde_json::Value) -> Result<Vec<u8>, serde_json::Error>,
309}
310
311impl MaxStructuredContentBytesLint {
312    pub fn new(definition: LintDefinition, max_bytes: usize) -> Self {
313        Self {
314            definition,
315            max_bytes,
316            serialize: serde_json::to_vec
317                as fn(&serde_json::Value) -> Result<Vec<u8>, serde_json::Error>,
318        }
319    }
320
321    #[cfg(test)]
322    pub(crate) fn new_with_serializer(
323        definition: LintDefinition,
324        max_bytes: usize,
325        serialize: fn(&serde_json::Value) -> Result<Vec<u8>, serde_json::Error>,
326    ) -> Self {
327        Self {
328            definition,
329            max_bytes,
330            serialize,
331        }
332    }
333}
334
335impl LintRule for MaxStructuredContentBytesLint {
336    fn definition(&self) -> &LintDefinition {
337        &self.definition
338    }
339
340    fn check_response(&self, context: &ResponseLintContext<'_>) -> Vec<LintFinding> {
341        let Some(value) = context.response.structured_content.as_ref() else {
342            return Vec::new();
343        };
344        let size = match (self.serialize)(value) {
345            Ok(encoded) => encoded.len(),
346            Err(error) => {
347                return vec![LintFinding::new(format!(
348                    "tool '{}' structuredContent failed to serialize",
349                    context.tool.name.as_ref()
350                ))
351                .with_details(json!({
352                    "tool": context.tool.name.as_ref(),
353                    "error": error.to_string(),
354                }))];
355            }
356        };
357        if size <= self.max_bytes {
358            return Vec::new();
359        }
360        vec![LintFinding::new(format!(
361            "tool '{}' structuredContent is {size} bytes (max {})",
362            context.tool.name.as_ref(),
363            self.max_bytes
364        ))
365        .with_details(json!({
366            "tool": context.tool.name.as_ref(),
367            "size": size,
368            "max": self.max_bytes,
369        }))]
370    }
371}
372
373/// Lint: reports missing structuredContent when an output schema exists.
374#[derive(Clone, Debug)]
375pub struct MissingStructuredContentLint {
376    definition: LintDefinition,
377}
378
379impl MissingStructuredContentLint {
380    pub fn new(definition: LintDefinition) -> Self {
381        Self { definition }
382    }
383}
384
385impl LintRule for MissingStructuredContentLint {
386    fn definition(&self) -> &LintDefinition {
387        &self.definition
388    }
389
390    fn check_response(&self, context: &ResponseLintContext<'_>) -> Vec<LintFinding> {
391        if context.tool.output_schema.is_some() && context.response.structured_content.is_none() {
392            return vec![LintFinding::new(format!(
393                "tool '{}' returned no structuredContent for output schema",
394                context.tool.name.as_ref()
395            ))
396            .with_details(json!({
397                "tool": context.tool.name.as_ref(),
398            }))];
399        }
400        Vec::new()
401    }
402}
403
404/// Lint: enforces coverage validation rules at run completion.
405#[derive(Clone, Debug)]
406pub struct CoverageLint {
407    definition: LintDefinition,
408    rules: Vec<CoverageRule>,
409}
410
411impl CoverageLint {
412    pub fn new(definition: LintDefinition, rules: Vec<CoverageRule>) -> Result<Self, String> {
413        if definition.phase != LintPhase::Run {
414            return Err("coverage lint must be configured for run phase".to_string());
415        }
416        for rule in &rules {
417            if let CoverageRule::PercentCalled { min_percent } = rule {
418                if !min_percent.is_finite() || *min_percent < 0.0 || *min_percent > 100.0 {
419                    return Err(format!(
420                        "coverage lint min_percent out of range: {min_percent}"
421                    ));
422                }
423            }
424        }
425        Ok(Self { definition, rules })
426    }
427
428    pub fn rules(&self) -> &[CoverageRule] {
429        &self.rules
430    }
431
432    fn effective_rules(&self) -> Vec<CoverageRule> {
433        if self.rules.is_empty() {
434            vec![CoverageRule::PercentCalled { min_percent: 100.0 }]
435        } else {
436            self.rules.clone()
437        }
438    }
439}
440
441impl LintRule for CoverageLint {
442    fn definition(&self) -> &LintDefinition {
443        &self.definition
444    }
445
446    fn check_run(&self, context: &RunLintContext<'_>) -> Vec<LintFinding> {
447        let Some(coverage) = context.coverage else {
448            return Vec::new();
449        };
450        let eligible: Vec<String> = coverage
451            .counts
452            .keys()
453            .filter(|name| {
454                is_coverage_tool_eligible(
455                    name.as_str(),
456                    context.coverage_allowlist,
457                    context.coverage_blocklist,
458                )
459            })
460            .cloned()
461            .collect();
462
463        let uncallable: HashSet<&str> = coverage
464            .warnings
465            .iter()
466            .map(|warning| warning.tool.as_str())
467            .collect();
468        let callable: Vec<String> = eligible
469            .into_iter()
470            .filter(|name| !uncallable.contains(name.as_str()))
471            .collect();
472
473        let counts: HashMap<&str, u64> = coverage
474            .counts
475            .iter()
476            .map(|(name, count)| (name.as_str(), *count))
477            .collect();
478
479        let mut findings = Vec::new();
480        for rule in self.effective_rules() {
481            match rule {
482                CoverageRule::MinCallsPerTool { min } => {
483                    let mut violations = Vec::new();
484                    for tool in &callable {
485                        let count = *counts.get(tool.as_str()).unwrap_or(&0);
486                        if count < min {
487                            violations.push(json!({ "tool": tool, "count": count }));
488                        }
489                    }
490                    if !violations.is_empty() {
491                        findings.push(
492                            LintFinding::new("coverage rule min_calls_per_tool failed")
493                                .with_code("coverage_validation_failed")
494                                .with_details(json!({
495                                    "rule": "min_calls_per_tool",
496                                    "min": min,
497                                    "violations": violations,
498                                })),
499                        );
500                    }
501                }
502                CoverageRule::NoUncalledTools => {
503                    let uncalled: Vec<String> = callable
504                        .iter()
505                        .filter(|tool| *counts.get(tool.as_str()).unwrap_or(&0) == 0)
506                        .cloned()
507                        .collect();
508                    if !uncalled.is_empty() {
509                        findings.push(
510                            LintFinding::new("coverage rule no_uncalled_tools failed")
511                                .with_code("coverage_validation_failed")
512                                .with_details(json!({
513                                    "rule": "no_uncalled_tools",
514                                    "uncalled": uncalled,
515                                })),
516                        );
517                    }
518                }
519                CoverageRule::PercentCalled { min_percent } => {
520                    let denom = callable.len() as f64;
521                    if denom == 0.0 {
522                        continue;
523                    }
524                    let called = callable
525                        .iter()
526                        .filter(|tool| *counts.get(tool.as_str()).unwrap_or(&0) > 0)
527                        .count() as f64;
528                    let percent = (called / denom) * 100.0;
529                    if percent < min_percent {
530                        findings.push(
531                            LintFinding::new("coverage rule percent_called failed")
532                                .with_code("coverage_validation_failed")
533                                .with_details(json!({
534                                    "rule": "percent_called",
535                                    "min_percent": min_percent,
536                                    "percent": percent,
537                                    "called": called,
538                                    "eligible": denom,
539                                })),
540                        );
541                    }
542                }
543            }
544        }
545
546        findings
547    }
548}
549
550/// Lint: fails the run when any failure occurred.
551#[derive(Clone, Debug)]
552pub struct NoCrashLint {
553    definition: LintDefinition,
554}
555
556impl NoCrashLint {
557    pub fn new(definition: LintDefinition) -> Result<Self, String> {
558        if definition.phase != LintPhase::Run {
559            return Err("no_crash lint must be configured for run phase".to_string());
560        }
561        if definition.level != LintLevel::Error {
562            return Err("no_crash lint must be configured at error level".to_string());
563        }
564        Ok(Self { definition })
565    }
566}
567
568impl LintRule for NoCrashLint {
569    fn definition(&self) -> &LintDefinition {
570        &self.definition
571    }
572
573    fn check_run(&self, context: &RunLintContext<'_>) -> Vec<LintFinding> {
574        if matches!(context.outcome, crate::RunOutcome::Failure(_)) {
575            return vec![LintFinding::new("run failed")];
576        }
577        Vec::new()
578    }
579}
tooltest_core/lints.rs

tooltest_core/
lints.rs
