assay_core/mcp/
ingest.rs

1//! Resource-bounded ingest for untrusted MCP transcript bytes.
2//!
3//! This layer owns the cost of reaching the existing MCP parser and refuses typed conclusion states
4//! that cannot support a clean reading. A valid JSON-RPC error response is accepted as an observed
5//! protocol outcome, but never treated as a result or as terminality evidence. This layer does not
6//! derive a policy decision or change the public [`McpEvent`] shape. When a caller supplies the
7//! transcript as an optional companion input, a refusal can veto that import but can never author or
8//! alter the producer's evidence.
9
10use super::era::{conclude, conclude_request, RequestAssessment, ResultConclusion};
11use super::parser::{parse_mcp_transcript_detailed_with_depth, EmbeddedJsonDepthExceeded};
12use super::{McpEvent, McpInputFormat};
13use assay_common::limits::{LimitExceeded, LimitKind, LimitReader};
14use serde::de::{DeserializeSeed, Deserializer, IgnoredAny, MapAccess, SeqAccess, Visitor};
15use std::io::Read;
16
17/// Domain-local ceilings for one MCP transcript.
18#[derive(Debug, Clone, Copy)]
19#[non_exhaustive]
20pub struct McpTranscriptLimits {
21    /// Maximum bytes read from the source before UTF-8 materialization.
22    pub max_source_bytes: u64,
23    /// Maximum bytes in one JSON-RPC JSONL line.
24    pub max_line_bytes: u64,
25    /// Maximum messages or transport entries in one transcript.
26    pub max_events: usize,
27    /// Maximum structural JSON nesting depth.
28    pub max_json_depth: usize,
29}
30
31impl Default for McpTranscriptLimits {
32    fn default() -> Self {
33        Self {
34            max_source_bytes: 16 * 1024 * 1024,
35            max_line_bytes: 1024 * 1024,
36            max_events: 100_000,
37            max_json_depth: 64,
38        }
39    }
40}
41
42/// A typed, value-free refusal at the MCP transcript boundary.
43#[derive(Debug, thiserror::Error)]
44#[non_exhaustive]
45pub enum McpTranscriptIngestError {
46    #[error("MCP transcript exceeded source-byte limit of {limit}")]
47    SourceBytes { limit: u64 },
48    #[error("MCP transcript JSONL line exceeded byte limit of {limit}")]
49    LineBytes { limit: u64 },
50    #[error("MCP transcript event count exceeded limit of {limit}")]
51    Events { limit: usize },
52    #[error("MCP transcript JSON nesting exceeded depth limit of {limit}")]
53    JsonDepth { limit: usize },
54    #[error("MCP transcript is not valid UTF-8")]
55    InvalidUtf8,
56    #[error("MCP transcript could not be read")]
57    ReadFailed,
58    #[error("MCP transcript is invalid")]
59    InvalidTranscript,
60    /// At least one request or result needs evidence this transcript does not provide.
61    #[error("MCP transcript conclusion is incomplete")]
62    ConclusionIncomplete,
63    /// At least one request or result carries a protocol-invalid observation.
64    #[error("MCP transcript conclusion is invalid")]
65    ConclusionInvalid,
66}
67
68/// Read and parse one transcript under explicit limits, refusing unusable conclusion states.
69///
70/// `Terminal` and `NonTerminal` results are both accepted. A valid JSON-RPC error response has no
71/// result conclusion and is accepted without being promoted to one. `Incomplete` and `Invalid`
72/// request or result states are mapped to value-free ingest errors; no policy decision or profile
73/// carrier is derived from them.
74pub fn parse_mcp_transcript_bounded<R: Read>(
75    reader: R,
76    format: McpInputFormat,
77    limits: McpTranscriptLimits,
78) -> Result<Vec<McpEvent>, McpTranscriptIngestError> {
79    let mut reader = LimitReader::new(reader, limits.max_source_bytes, LimitKind::SourceBytes);
80    let mut bytes = Vec::new();
81    if let Err(error) = reader.read_to_end(&mut bytes) {
82        if let Some(LimitExceeded { limit, .. }) = LimitExceeded::from_io(&error) {
83            return Err(McpTranscriptIngestError::SourceBytes { limit });
84        }
85        return Err(McpTranscriptIngestError::ReadFailed);
86    }
87    check_json_depth(&bytes, format, limits.max_json_depth)?;
88    if format == McpInputFormat::JsonRpc {
89        check_jsonl_lines(&bytes, limits.max_line_bytes)?;
90    }
91    let text = String::from_utf8(bytes).map_err(|_| McpTranscriptIngestError::InvalidUtf8)?;
92    check_event_count(text.as_bytes(), format, limits.max_events)?;
93    let parsed = parse_mcp_transcript_detailed_with_depth(&text, format, limits.max_json_depth)
94        .map_err(|error| {
95            if error.downcast_ref::<EmbeddedJsonDepthExceeded>().is_some() {
96                McpTranscriptIngestError::JsonDepth {
97                    limit: limits.max_json_depth,
98                }
99            } else {
100                McpTranscriptIngestError::InvalidTranscript
101            }
102        })?;
103    for entry in &parsed {
104        if entry.is_error_response {
105            // The parser has already established a valid JSON-RPC error-response shape. It carries
106            // no MCP result whose terminality can be read. Keep that distinction explicit:
107            // accepting it must not synthesize a result conclusion or a producer decision.
108            continue;
109        }
110        if let Some(metadata) = &entry.context.request_metadata {
111            match conclude_request(
112                &entry.context.era,
113                metadata,
114                entry.context.capability_observation.as_ref(),
115            ) {
116                RequestAssessment::Valid => {}
117                RequestAssessment::Incomplete(_) => {
118                    return Err(McpTranscriptIngestError::ConclusionIncomplete)
119                }
120                RequestAssessment::Invalid(_) => {
121                    return Err(McpTranscriptIngestError::ConclusionInvalid)
122                }
123            }
124        }
125        if let Some(result) = &entry.context.result_observation {
126            match conclude(
127                &entry.context.era,
128                result,
129                entry.context.capability_observation.as_ref(),
130            ) {
131                ResultConclusion::Terminal | ResultConclusion::NonTerminal => {}
132                ResultConclusion::Incomplete(_) => {
133                    return Err(McpTranscriptIngestError::ConclusionIncomplete)
134                }
135                ResultConclusion::Invalid(_) => {
136                    return Err(McpTranscriptIngestError::ConclusionInvalid)
137                }
138            }
139        }
140    }
141    Ok(parsed.into_iter().map(|entry| entry.event).collect())
142}
143
144fn check_jsonl_lines(bytes: &[u8], max_line_bytes: u64) -> Result<(), McpTranscriptIngestError> {
145    for line in bytes.split(|byte| *byte == b'\n') {
146        if line.len() as u64 > max_line_bytes {
147            return Err(McpTranscriptIngestError::LineBytes {
148                limit: max_line_bytes,
149            });
150        }
151    }
152    Ok(())
153}
154
155fn check_json_depth(
156    bytes: &[u8],
157    format: McpInputFormat,
158    max_json_depth: usize,
159) -> Result<(), McpTranscriptIngestError> {
160    if super::json_depth::exceeds_limit(bytes, max_json_depth, format == McpInputFormat::JsonRpc) {
161        return Err(McpTranscriptIngestError::JsonDepth {
162            limit: max_json_depth,
163        });
164    }
165    Ok(())
166}
167
168fn check_event_count(
169    bytes: &[u8],
170    format: McpInputFormat,
171    max_events: usize,
172) -> Result<(), McpTranscriptIngestError> {
173    let count = match format {
174        McpInputFormat::JsonRpc => {
175            let count = bytes
176                .split(|byte| *byte == b'\n')
177                .filter(|line| !line.iter().all(u8::is_ascii_whitespace))
178                .count();
179            if count > max_events {
180                CountOutcome::OverLimit
181            } else {
182                CountOutcome::WithinLimit
183            }
184        }
185        McpInputFormat::Inspector => count_container_array(bytes, "events", true, max_events)?,
186        McpInputFormat::StreamableHttp | McpInputFormat::HttpSse => {
187            count_container_array(bytes, "entries", false, max_events)?
188        }
189    };
190    match count {
191        CountOutcome::WithinLimit => Ok(()),
192        CountOutcome::OverLimit => Err(McpTranscriptIngestError::Events { limit: max_events }),
193    }
194}
195
196#[derive(Debug, Clone, Copy, PartialEq, Eq)]
197enum CountOutcome {
198    WithinLimit,
199    OverLimit,
200}
201
202fn count_container_array(
203    bytes: &[u8],
204    array_key: &'static str,
205    root_array_is_target: bool,
206    max_events: usize,
207) -> Result<CountOutcome, McpTranscriptIngestError> {
208    let mut deserializer = serde_json::Deserializer::from_slice(bytes);
209    deserializer
210        .deserialize_any(ContainerVisitor {
211            array_key,
212            root_array_is_target,
213            max_events,
214        })
215        .map_err(|_| McpTranscriptIngestError::InvalidTranscript)
216}
217
218struct ContainerVisitor {
219    array_key: &'static str,
220    root_array_is_target: bool,
221    max_events: usize,
222}
223
224impl<'de> Visitor<'de> for ContainerVisitor {
225    type Value = CountOutcome;
226
227    fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
228        formatter.write_str("an MCP transcript container")
229    }
230
231    fn visit_seq<A: SeqAccess<'de>>(self, sequence: A) -> Result<Self::Value, A::Error> {
232        if self.root_array_is_target {
233            CountSeed {
234                max_events: self.max_events,
235            }
236            .count(sequence)
237        } else {
238            drain_sequence(sequence)?;
239            Ok(CountOutcome::WithinLimit)
240        }
241    }
242
243    fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<Self::Value, A::Error> {
244        let mut outcome = CountOutcome::WithinLimit;
245        while let Some(key) = map.next_key::<String>()? {
246            if key == self.array_key {
247                let observed = map.next_value_seed(CountSeed {
248                    max_events: self.max_events,
249                })?;
250                if observed == CountOutcome::OverLimit {
251                    return Ok(observed);
252                }
253                outcome = observed;
254            } else {
255                map.next_value::<IgnoredAny>()?;
256            }
257        }
258        Ok(outcome)
259    }
260
261    fn visit_bool<E: serde::de::Error>(self, _value: bool) -> Result<Self::Value, E> {
262        Ok(CountOutcome::WithinLimit)
263    }
264
265    fn visit_i64<E: serde::de::Error>(self, _value: i64) -> Result<Self::Value, E> {
266        Ok(CountOutcome::WithinLimit)
267    }
268
269    fn visit_u64<E: serde::de::Error>(self, _value: u64) -> Result<Self::Value, E> {
270        Ok(CountOutcome::WithinLimit)
271    }
272
273    fn visit_f64<E: serde::de::Error>(self, _value: f64) -> Result<Self::Value, E> {
274        Ok(CountOutcome::WithinLimit)
275    }
276
277    fn visit_str<E: serde::de::Error>(self, _value: &str) -> Result<Self::Value, E> {
278        Ok(CountOutcome::WithinLimit)
279    }
280
281    fn visit_string<E: serde::de::Error>(self, _value: String) -> Result<Self::Value, E> {
282        Ok(CountOutcome::WithinLimit)
283    }
284
285    fn visit_none<E: serde::de::Error>(self) -> Result<Self::Value, E> {
286        Ok(CountOutcome::WithinLimit)
287    }
288
289    fn visit_unit<E: serde::de::Error>(self) -> Result<Self::Value, E> {
290        Ok(CountOutcome::WithinLimit)
291    }
292}
293
294struct CountSeed {
295    max_events: usize,
296}
297
298impl CountSeed {
299    fn count<'de, A: SeqAccess<'de>>(&self, mut sequence: A) -> Result<CountOutcome, A::Error> {
300        let mut count = 0usize;
301        while sequence.next_element::<IgnoredAny>()?.is_some() {
302            count += 1;
303            if count > self.max_events {
304                return Ok(CountOutcome::OverLimit);
305            }
306        }
307        Ok(CountOutcome::WithinLimit)
308    }
309}
310
311impl<'de> DeserializeSeed<'de> for CountSeed {
312    type Value = CountOutcome;
313
314    fn deserialize<D: Deserializer<'de>>(self, deserializer: D) -> Result<Self::Value, D::Error> {
315        deserializer.deserialize_any(CountValueVisitor {
316            max_events: self.max_events,
317        })
318    }
319}
320
321struct CountValueVisitor {
322    max_events: usize,
323}
324
325impl<'de> Visitor<'de> for CountValueVisitor {
326    type Value = CountOutcome;
327
328    fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
329        formatter.write_str("an event array")
330    }
331
332    fn visit_seq<A: SeqAccess<'de>>(self, sequence: A) -> Result<Self::Value, A::Error> {
333        CountSeed {
334            max_events: self.max_events,
335        }
336        .count(sequence)
337    }
338
339    fn visit_map<A: MapAccess<'de>>(self, map: A) -> Result<Self::Value, A::Error> {
340        drain_map(map)?;
341        Ok(CountOutcome::WithinLimit)
342    }
343
344    fn visit_bool<E: serde::de::Error>(self, _value: bool) -> Result<Self::Value, E> {
345        Ok(CountOutcome::WithinLimit)
346    }
347
348    fn visit_i64<E: serde::de::Error>(self, _value: i64) -> Result<Self::Value, E> {
349        Ok(CountOutcome::WithinLimit)
350    }
351
352    fn visit_u64<E: serde::de::Error>(self, _value: u64) -> Result<Self::Value, E> {
353        Ok(CountOutcome::WithinLimit)
354    }
355
356    fn visit_f64<E: serde::de::Error>(self, _value: f64) -> Result<Self::Value, E> {
357        Ok(CountOutcome::WithinLimit)
358    }
359
360    fn visit_str<E: serde::de::Error>(self, _value: &str) -> Result<Self::Value, E> {
361        Ok(CountOutcome::WithinLimit)
362    }
363
364    fn visit_string<E: serde::de::Error>(self, _value: String) -> Result<Self::Value, E> {
365        Ok(CountOutcome::WithinLimit)
366    }
367
368    fn visit_none<E: serde::de::Error>(self) -> Result<Self::Value, E> {
369        Ok(CountOutcome::WithinLimit)
370    }
371
372    fn visit_unit<E: serde::de::Error>(self) -> Result<Self::Value, E> {
373        Ok(CountOutcome::WithinLimit)
374    }
375}
376
377fn drain_sequence<'de, A: SeqAccess<'de>>(mut sequence: A) -> Result<(), A::Error> {
378    while sequence.next_element::<IgnoredAny>()?.is_some() {}
379    Ok(())
380}
381
382fn drain_map<'de, A: MapAccess<'de>>(mut map: A) -> Result<(), A::Error> {
383    while map.next_entry::<IgnoredAny, IgnoredAny>()?.is_some() {}
384    Ok(())
385}
386
387#[cfg(test)]
388mod tests {
389    use super::*;
390    use crate::mcp::parser::parse_mcp_transcript_detailed;
391    use std::io::{Cursor, Read};
392
393    const NOTIFICATION: &str = r#"{"jsonrpc":"2.0","method":"notifications/progress","params":{}}"#;
394
395    fn modern_transport(result: &str) -> String {
396        format!(
397            r#"{{"transport":"streamable-http","transport_context":{{"headers":{{"MCP-Protocol-Version":"2026-07-28"}}}},"entries":[{{"request":{{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{{"name":"example.tool","arguments":{{}},"_meta":{{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{{}}}}}}}}}},{{"response":{{"jsonrpc":"2.0","id":1,"result":{result}}}}}]}}"#
398        )
399    }
400
401    fn modern_transport_with_error(error: &str) -> String {
402        format!(
403            r#"{{"transport":"streamable-http","transport_context":{{"headers":{{"MCP-Protocol-Version":"2026-07-28"}}}},"entries":[{{"request":{{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{{"name":"example.tool","arguments":{{}},"_meta":{{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{{}}}}}}}}}},{{"response":{{"jsonrpc":"2.0","id":1,"error":{error}}}}}]}}"#
404        )
405    }
406
407    fn modern_transport_error() -> String {
408        modern_transport_with_error(r#"{"code":-32603,"message":"ATTACKER_SENTINEL"}"#)
409    }
410
411    fn limits() -> McpTranscriptLimits {
412        McpTranscriptLimits {
413            max_source_bytes: 4096,
414            max_line_bytes: 4096,
415            max_events: 8,
416            max_json_depth: 16,
417        }
418    }
419
420    fn assert_source_limit(error: McpTranscriptIngestError, limit: u64) {
421        assert!(
422            matches!(error, McpTranscriptIngestError::SourceBytes { limit: got } if got == limit),
423            "expected source-byte refusal at {limit}, got {error:?}"
424        );
425    }
426
427    #[test]
428    fn source_limit_is_inclusive_and_one_more_byte_is_refused() {
429        let bytes = NOTIFICATION.as_bytes();
430        let mut exact = limits();
431        exact.max_source_bytes = bytes.len() as u64;
432        assert_eq!(
433            parse_mcp_transcript_bounded(Cursor::new(bytes), McpInputFormat::JsonRpc, exact)
434                .unwrap()
435                .len(),
436            1
437        );
438
439        let mut over = exact;
440        over.max_source_bytes -= 1;
441        let error = parse_mcp_transcript_bounded(Cursor::new(bytes), McpInputFormat::JsonRpc, over)
442            .unwrap_err();
443        assert_source_limit(error, over.max_source_bytes);
444    }
445
446    #[test]
447    fn short_non_seekable_reads_cannot_walk_past_the_source_limit() {
448        struct OneByteAtATime(Cursor<Vec<u8>>);
449        impl Read for OneByteAtATime {
450            fn read(&mut self, output: &mut [u8]) -> std::io::Result<usize> {
451                if output.is_empty() {
452                    return Ok(0);
453                }
454                self.0.read(&mut output[..1])
455            }
456        }
457
458        let bytes = NOTIFICATION.as_bytes().to_vec();
459        let mut exact = limits();
460        exact.max_source_bytes = bytes.len() as u64;
461        assert!(parse_mcp_transcript_bounded(
462            OneByteAtATime(Cursor::new(bytes.clone())),
463            McpInputFormat::JsonRpc,
464            exact,
465        )
466        .is_ok());
467
468        let mut over = exact;
469        over.max_source_bytes -= 1;
470        let error = parse_mcp_transcript_bounded(
471            OneByteAtATime(Cursor::new(bytes)),
472            McpInputFormat::JsonRpc,
473            over,
474        )
475        .unwrap_err();
476        assert_source_limit(error, over.max_source_bytes);
477    }
478
479    #[test]
480    fn source_limit_fires_before_invalid_utf8_is_materialized() {
481        let input = [b'{', b'}', 0xff];
482        let mut bounded = limits();
483        bounded.max_source_bytes = 2;
484        let error =
485            parse_mcp_transcript_bounded(Cursor::new(input), McpInputFormat::Inspector, bounded)
486                .unwrap_err();
487        assert_source_limit(error, 2);
488    }
489
490    #[test]
491    fn jsonl_line_limit_is_inclusive_and_one_more_byte_is_refused() {
492        let mut exact = limits();
493        exact.max_line_bytes = NOTIFICATION.len() as u64;
494        assert!(parse_mcp_transcript_bounded(
495            Cursor::new(NOTIFICATION),
496            McpInputFormat::JsonRpc,
497            exact,
498        )
499        .is_ok());
500
501        let mut over = exact;
502        over.max_line_bytes -= 1;
503        let error =
504            parse_mcp_transcript_bounded(Cursor::new(NOTIFICATION), McpInputFormat::JsonRpc, over)
505                .unwrap_err();
506        assert!(
507            matches!(error, McpTranscriptIngestError::LineBytes { limit } if limit == over.max_line_bytes)
508        );
509    }
510
511    #[test]
512    fn jsonl_line_limit_fires_before_invalid_json_is_parsed() {
513        let mut bounded = limits();
514        bounded.max_line_bytes = 3;
515        let error =
516            parse_mcp_transcript_bounded(Cursor::new("not-json"), McpInputFormat::JsonRpc, bounded)
517                .unwrap_err();
518        assert!(matches!(
519            error,
520            McpTranscriptIngestError::LineBytes { limit: 3 }
521        ));
522    }
523
524    #[test]
525    fn jsonrpc_event_limit_is_applied_before_the_parser_builds_the_event_vector() {
526        let transcript = format!("{NOTIFICATION}\n{NOTIFICATION}");
527        let mut exact = limits();
528        exact.max_events = 2;
529        assert_eq!(
530            parse_mcp_transcript_bounded(Cursor::new(&transcript), McpInputFormat::JsonRpc, exact,)
531                .unwrap()
532                .len(),
533            2
534        );
535
536        let mut over = exact;
537        over.max_events = 1;
538        let error =
539            parse_mcp_transcript_bounded(Cursor::new(&transcript), McpInputFormat::JsonRpc, over)
540                .unwrap_err();
541        assert!(matches!(
542            error,
543            McpTranscriptIngestError::Events { limit: 1 }
544        ));
545
546        let malformed_second = format!("{NOTIFICATION}\nnot-json");
547        let error = parse_mcp_transcript_bounded(
548            Cursor::new(malformed_second),
549            McpInputFormat::JsonRpc,
550            over,
551        )
552        .unwrap_err();
553        assert!(
554            matches!(error, McpTranscriptIngestError::Events { limit: 1 }),
555            "event ceiling must decide before a later row reaches the semantic parser"
556        );
557    }
558
559    #[test]
560    fn inspector_event_limit_is_applied_to_the_events_array() {
561        let transcript = format!(r#"{{"events":[{NOTIFICATION},{NOTIFICATION}]}}"#);
562        let mut bounded = limits();
563        bounded.max_events = 1;
564        let error = parse_mcp_transcript_bounded(
565            Cursor::new(transcript),
566            McpInputFormat::Inspector,
567            bounded,
568        )
569        .unwrap_err();
570        assert!(matches!(
571            error,
572            McpTranscriptIngestError::Events { limit: 1 }
573        ));
574    }
575
576    #[test]
577    fn transport_event_limit_is_applied_to_the_entries_array() {
578        let entry = format!(r#"{{"request":{NOTIFICATION}}}"#);
579        let transcript =
580            format!(r#"{{"transport":"streamable-http","entries":[{entry},{entry}]}}"#);
581        let mut bounded = limits();
582        bounded.max_events = 1;
583        let error = parse_mcp_transcript_bounded(
584            Cursor::new(transcript),
585            McpInputFormat::StreamableHttp,
586            bounded,
587        )
588        .unwrap_err();
589        assert!(matches!(
590            error,
591            McpTranscriptIngestError::Events { limit: 1 }
592        ));
593    }
594
595    #[test]
596    fn json_depth_limit_is_inclusive_and_one_more_level_is_refused() {
597        let transcript =
598            r#"{"jsonrpc":"2.0","method":"notifications/progress","params":{"outer":{"leaf":1}}}"#;
599        let mut exact = limits();
600        exact.max_json_depth = 3;
601        assert!(parse_mcp_transcript_bounded(
602            Cursor::new(transcript),
603            McpInputFormat::JsonRpc,
604            exact,
605        )
606        .is_ok());
607
608        let mut over = exact;
609        over.max_json_depth = 2;
610        let error =
611            parse_mcp_transcript_bounded(Cursor::new(transcript), McpInputFormat::JsonRpc, over)
612                .unwrap_err();
613        assert!(matches!(
614            error,
615            McpTranscriptIngestError::JsonDepth { limit: 2 }
616        ));
617    }
618
619    #[test]
620    fn sse_string_payload_obeys_the_same_json_depth_limit() {
621        let request = serde_json::json!({
622            "jsonrpc": "2.0",
623            "id": 1,
624            "method": "tools/call",
625            "params": {
626                "name": "example.tool",
627                "arguments": {"one": {"two": {"three": true}}},
628                "_meta": {
629                    "io.modelcontextprotocol/protocolVersion": "2026-07-28",
630                    "io.modelcontextprotocol/clientCapabilities": {}
631                }
632            }
633        });
634        let response = serde_json::json!({
635            "jsonrpc": "2.0",
636            "id": 1,
637            "result": {"content": [], "resultType": "complete"}
638        });
639        let transcript = serde_json::json!({
640            "transport": "http-sse",
641            "transport_context": {
642                "headers": {"MCP-Protocol-Version": "2026-07-28"}
643            },
644            "entries": [
645                {"sse": {"event": "message", "data": request.to_string()}},
646                {"sse": {"event": "message", "data": response.to_string()}}
647            ]
648        })
649        .to_string();
650        let mut exact = limits();
651        exact.max_json_depth = 5;
652        assert!(parse_mcp_transcript_bounded(
653            Cursor::new(transcript.as_bytes()),
654            McpInputFormat::HttpSse,
655            exact,
656        )
657        .is_ok());
658
659        let mut over = exact;
660        over.max_json_depth = 4;
661
662        let error =
663            parse_mcp_transcript_bounded(Cursor::new(transcript), McpInputFormat::HttpSse, over)
664                .unwrap_err();
665        assert!(matches!(
666            error,
667            McpTranscriptIngestError::JsonDepth { limit: 4 }
668        ));
669    }
670
671    #[test]
672    fn framed_json_depth_accumulates_across_newlines() {
673        let transcript = r#"{
674  "transport": "streamable-http",
675  "transport_context": {
676    "headers": {
677      "MCP-Protocol-Version": "2026-07-28"
678    }
679  },
680  "entries": [
681    {
682      "request": {
683        "jsonrpc": "2.0",
684        "id": 1,
685        "method": "tools/call",
686        "params": {
687          "name": "example.tool",
688          "arguments": {},
689          "_meta": {
690            "io.modelcontextprotocol/protocolVersion": "2026-07-28",
691            "io.modelcontextprotocol/clientCapabilities": {}
692          }
693        }
694      }
695    }
696  ]
697}"#;
698        let mut bounded = limits();
699        bounded.max_json_depth = 1;
700
701        let error = parse_mcp_transcript_bounded(
702            Cursor::new(transcript),
703            McpInputFormat::StreamableHttp,
704            bounded,
705        )
706        .unwrap_err();
707        assert!(matches!(
708            error,
709            McpTranscriptIngestError::JsonDepth { limit: 1 }
710        ));
711    }
712
713    #[test]
714    fn every_jsonrpc_message_shape_requires_the_2_0_marker() {
715        let request = serde_json::json!({
716            "jsonrpc": "2.0",
717            "id": 1,
718            "method": "tools/call",
719            "params": {
720                "name": "example.tool",
721                "arguments": {},
722                "_meta": {
723                    "io.modelcontextprotocol/protocolVersion": "2026-07-28",
724                    "io.modelcontextprotocol/clientCapabilities": {}
725                }
726            }
727        });
728        let shapes = [
729            ("request", request.clone(), false),
730            (
731                "notification",
732                serde_json::json!({
733                    "jsonrpc": "2.0",
734                    "method": "notifications/progress",
735                    "params": {}
736                }),
737                false,
738            ),
739            (
740                "success response",
741                serde_json::json!({
742                    "jsonrpc": "2.0",
743                    "id": 1,
744                    "result": {"content": [], "resultType": "complete"}
745                }),
746                true,
747            ),
748            (
749                "error response",
750                serde_json::json!({
751                    "jsonrpc": "2.0",
752                    "id": 1,
753                    "error": {"code": -32603, "message": "refused"}
754                }),
755                true,
756            ),
757        ];
758        let markers = [
759            ("missing", None),
760            ("non-string", Some(serde_json::json!(7))),
761            ("wrong", Some(serde_json::json!("1.0"))),
762        ];
763
764        for (shape_name, shape, needs_request) in shapes {
765            for (marker_name, marker) in &markers {
766                let mut malformed = shape.clone();
767                let object = malformed.as_object_mut().unwrap();
768                match marker {
769                    Some(value) => {
770                        object.insert("jsonrpc".to_string(), value.clone());
771                    }
772                    None => {
773                        object.remove("jsonrpc");
774                    }
775                }
776                let transcript = if needs_request {
777                    format!("{request}\n{malformed}\n")
778                } else {
779                    format!("{malformed}\n")
780                };
781                let error = parse_mcp_transcript_bounded(
782                    Cursor::new(transcript),
783                    McpInputFormat::JsonRpc,
784                    limits(),
785                )
786                .unwrap_err();
787                assert!(
788                    matches!(error, McpTranscriptIngestError::InvalidTranscript),
789                    "{shape_name} with {marker_name} marker returned {error:?}"
790                );
791            }
792        }
793    }
794
795    #[test]
796    fn braces_inside_strings_do_not_consume_json_depth_budget() {
797        let transcript =
798            r#"{"jsonrpc":"2.0","method":"notifications/progress","params":{"text":"{{[[}}]]"}}"#;
799        let mut bounded = limits();
800        bounded.max_json_depth = 2;
801        assert!(parse_mcp_transcript_bounded(
802            Cursor::new(transcript),
803            McpInputFormat::JsonRpc,
804            bounded,
805        )
806        .is_ok());
807    }
808
809    #[test]
810    fn unterminated_jsonl_string_cannot_mask_depth_on_the_next_line() {
811        let transcript = "{\"text\":\"unterminated\n[[[]]]";
812        let mut bounded = limits();
813        bounded.max_json_depth = 2;
814        let error =
815            parse_mcp_transcript_bounded(Cursor::new(transcript), McpInputFormat::JsonRpc, bounded)
816                .unwrap_err();
817        assert!(matches!(
818            error,
819            McpTranscriptIngestError::JsonDepth { limit: 2 }
820        ));
821    }
822
823    #[test]
824    fn unbalanced_jsonl_depth_does_not_carry_into_the_next_line() {
825        let transcript = format!("{{\"open\":{{\n{NOTIFICATION}");
826        let mut bounded = limits();
827        bounded.max_json_depth = 2;
828        let error =
829            parse_mcp_transcript_bounded(Cursor::new(transcript), McpInputFormat::JsonRpc, bounded)
830                .unwrap_err();
831        assert!(matches!(error, McpTranscriptIngestError::InvalidTranscript));
832    }
833
834    #[test]
835    fn diagnostics_do_not_echo_attacker_controlled_input() {
836        let input = br#"{"ATTACKER_SENTINEL":"#;
837        let error =
838            parse_mcp_transcript_bounded(Cursor::new(input), McpInputFormat::Inspector, limits())
839                .unwrap_err();
840        assert!(matches!(error, McpTranscriptIngestError::InvalidTranscript));
841        assert!(!error.to_string().contains("ATTACKER_SENTINEL"));
842        assert!(!format!("{error:?}").contains("ATTACKER_SENTINEL"));
843    }
844
845    #[test]
846    fn unresolved_request_era_is_refused_after_bounded_parsing() {
847        let request = r#"{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"example.tool","arguments":{}}}"#;
848        let error =
849            parse_mcp_transcript_bounded(Cursor::new(request), McpInputFormat::JsonRpc, limits())
850                .unwrap_err();
851        assert_eq!(error.to_string(), "MCP transcript conclusion is incomplete");
852    }
853
854    #[test]
855    fn modern_result_without_result_type_is_refused_after_bounded_parsing() {
856        let transcript = modern_transport(r#"{"content":[]}"#);
857        let error = parse_mcp_transcript_bounded(
858            Cursor::new(transcript),
859            McpInputFormat::StreamableHttp,
860            limits(),
861        )
862        .unwrap_err();
863        assert_eq!(error.to_string(), "MCP transcript conclusion is invalid");
864    }
865
866    #[test]
867    fn unrecognized_modern_result_is_not_promoted_to_a_clean_reading() {
868        let transcript = modern_transport(r#"{"content":[],"resultType":"future-state"}"#);
869        let error = parse_mcp_transcript_bounded(
870            Cursor::new(transcript),
871            McpInputFormat::StreamableHttp,
872            limits(),
873        )
874        .unwrap_err();
875        assert_eq!(error.to_string(), "MCP transcript conclusion is incomplete");
876        assert!(!format!("{error:?}").contains("future-state"));
877    }
878
879    #[test]
880    fn modern_error_response_is_observed_without_becoming_a_result_conclusion() {
881        let transcript = modern_transport_error();
882        let parsed =
883            parse_mcp_transcript_detailed(&transcript, McpInputFormat::StreamableHttp).unwrap();
884        let error_entry = parsed
885            .iter()
886            .find(|entry| entry.is_error_response)
887            .expect("the parser must observe the JSON-RPC error response");
888        assert!(
889            error_entry.context.result_observation.is_none(),
890            "an error response must not acquire an MCP result conclusion"
891        );
892        assert!(parse_mcp_transcript_bounded(
893            Cursor::new(transcript),
894            McpInputFormat::StreamableHttp,
895            limits(),
896        )
897        .is_ok());
898    }
899
900    #[test]
901    fn malformed_error_members_are_refused_before_error_acceptance() {
902        for malformed in [
903            "null",
904            "7",
905            r#""denied""#,
906            r#"{"message":"missing code"}"#,
907            r#"{"code":-32603}"#,
908            r#"{"code":1.0,"message":"decimal code"}"#,
909            r#"{"code":1.5,"message":"float code"}"#,
910            r#"{"code":-32603,"message":7}"#,
911        ] {
912            let error = parse_mcp_transcript_bounded(
913                Cursor::new(modern_transport_with_error(malformed)),
914                McpInputFormat::StreamableHttp,
915                limits(),
916            )
917            .unwrap_err();
918            assert!(
919                matches!(error, McpTranscriptIngestError::InvalidTranscript),
920                "malformed error member must be a transcript refusal, got {error:?}"
921            );
922        }
923    }
924
925    #[test]
926    fn valid_modern_input_required_with_continuation_is_accepted() {
927        let transcript =
928            modern_transport(r#"{"resultType":"input_required","requestState":"opaque"}"#);
929        assert!(parse_mcp_transcript_bounded(
930            Cursor::new(transcript),
931            McpInputFormat::StreamableHttp,
932            limits(),
933        )
934        .is_ok());
935    }
936}
assay_core/mcp/ingest.rs

assay_core/mcp/
ingest.rs
