feat: add SGLang conformance recordings and CI replay

.github/workflows/rust.yml

@@ -129,9 +129,13 @@ jobs:
       - name: Verify PostgreSQL migrations and restart persistence
         run: cargo test -p agentic-server-core --test postgres_storage_integration -- --ignored --test-threads=1
 
-  dynamo-upstream:
+  provider-upstream:
     runs-on: ubuntu-latest
     timeout-minutes: 20
+    strategy:
+      fail-fast: false
+      matrix:
+        provider: [dynamo, sglang]
     env:
       CARGO_INCREMENTAL: "0"
       CARGO_PROFILE_TEST_DEBUG: "0"
@@ -154,5 +158,5 @@ jobs:
           python3 -m pip install 'PyYAML==6.0.3'
           python3 scripts/validate-cassettes.py
 
-      - name: Replay Dynamo upstream cassettes
-        run: cargo test -p agentic-server-core --test dynamo_cassette_test
+      - name: Replay provider upstream cassettes
+        run: cargo test -p agentic-server-core --test ${{ matrix.provider }}_cassette_test

README.md

@@ -176,6 +176,9 @@ Serving through [NVIDIA Dynamo](https://github.com/ai-dynamo/dynamo) instead of
 gateway at the Dynamo frontend the same way; see
 [Running Agentic API in front of NVIDIA Dynamo](docs/guides/dynamo-upstream.md).
 
+For SGLang, see [Running Agentic API in front of SGLang](docs/guides/sglang-upstream.md), including the pinned
+launch configuration, cassette recorder, and shared provider replay tests.
+
 **2. Start Agentic API**, pointing it at the vLLM server (set the `YOU_*` variables to enable built-in web search):
 
 ```bash

crates/agentic-server-core/src/events/types.rs

@@ -243,14 +243,27 @@ impl TryFrom<SSEEventType> for &'static str {
 pub struct WireEvent {
     #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
     pub event_type: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(
+        default,
+        deserialize_with = "deserialize_sequence_number",
+        skip_serializing_if = "Option::is_none"
+    )]
     pub sequence_number: Option<u64>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub output_index: Option<u64>,
     #[serde(flatten)]
     pub rest: Map<String, Value>,
 }
 
+// Some upstreams use negative integer sentinels for an unspecified sequence.
+// Preserve the full u64 range; client numbering belongs to the stream relay.
+fn deserialize_sequence_number<'de, D>(deserializer: D) -> Result<Option<u64>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    Option::<i128>::deserialize(deserializer).map(|value| value.and_then(|number| u64::try_from(number).ok()))
+}
+
 impl WireEvent {
     #[must_use]
     pub fn new(event_type: impl Into<String>) -> Self {

crates/agentic-server-core/tests/cassettes/README.md

@@ -211,6 +211,7 @@ turns:
 | `record_mcp_cassettes.sh` | Native MCP counter tool discovery and calls (streaming + non-streaming) | gateway and OpenAI reference |
 | `record_web_search_cassettes.sh` | Matching web-search calls (streaming + non-streaming) | gateway and OpenAI reference |
 | `record_dynamo_cassettes.sh` | Stateful two-turn and client-executed function tool call cassettes (streaming + non-streaming) | NVIDIA Dynamo frontend |
+| `record_sglang_cassettes.sh` | Same shared executor scenarios as Dynamo, staged validation and sanitized provenance | SGLang |
 | `record_tool_search_cassettes.sh` | Four-turn mixed function/namespace client tool-search characterization; gateway blocking, HTTP/SSE, and WebSocket acceptance | gateway and OpenAI reference |
 
 ### Text-only (OpenAI)
@@ -279,6 +280,11 @@ hydrated item history the gateway sends upstream), records it, and merges both i
 DYNAMO_URL=http://127.0.0.1:8000 MODEL=openai/gpt-oss-20b bash tests/cassettes/record_dynamo_cassettes.sh
 ```
 
+### SGLang
+
+For SGLang launch, recording, sanitation, and replay instructions, see
+[the SGLang upstream guide](../../../../docs/guides/sglang-upstream.md).
+
 ### Client tool search (OpenAI reference and gateway)
 
 The recorder captures four turns: a search call, a linked search output followed by one loaded ordinary function

crates/agentic-server-core/tests/cassettes/prepare_sglang_recordings.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""Attach provenance and sanitize recorder-produced SGLang cassettes in staging."""
+
+import json
+import sys
+from pathlib import Path
+from urllib.request import urlopen
+
+import yaml
+
+
+def prepare(root: Path, version: str, model: str, upstream: str) -> None:
+    with urlopen(upstream.rstrip("/") + "/server_info", timeout=15) as response:
+        info = json.load(response)
+    if info["version"] != version or info["model_path"] != model:
+        raise ValueError("Running SGLang version/model does not match the requested recording profile")
+    launch = {
+        key: info[key]
+        for key in (
+            "revision", "reasoning_parser", "tool_call_parser", "context_length",
+            "mem_fraction_static", "dtype", "quantization", "tp_size", "random_seed",
+        )
+    }
+    for path in sorted(root.glob("*.yaml")):
+        cassette = yaml.safe_load(path.read_text())
+        identifiers = {}
+
+        def sanitize(value, key=""):
+            if isinstance(value, dict):
+                return {name: sanitize(item, name) for name, item in value.items()}
+            if isinstance(value, list):
+                return [sanitize(item, key) for item in value]
+            if key in {"created_at", "completed_at"} and isinstance(value, (float, int)):
+                # Keep provider wire number types while removing wall-clock time.
+                return 0.0 if isinstance(value, float) else 0
+            if key in {"id", "item_id", "call_id", "previous_response_id"} and isinstance(value, str):
+                if value not in identifiers:
+                    identifiers[value] = f"recorded_{len(identifiers) + 1}"
+                return identifiers[value]
+            return value
+
+        for turn in cassette["turns"]:
+            turn["request"]["headers"] = {"content-type": "application/json"}
+            turn["request"]["query_params"] = {}
+            turn["request"]["body"] = sanitize(turn["request"]["body"])
+            response = turn["response"]
+            if response.get("body") is not None:
+                response["body"] = sanitize(response["body"])
+            else:
+                sanitized = []
+                for raw in response["sse"]:
+                    lines = []
+                    for line in raw.splitlines(keepends=True):
+                        if line.startswith("data:") and line[5:].strip() != "[DONE]":
+                            ending = "\n" if line.endswith("\n") else ""
+                            line = "data: " + json.dumps(sanitize(json.loads(line[5:])), separators=(",", ":")) + ending
+                        lines.append(line)
+                    sanitized.append("".join(lines))
+                response["sse"] = sanitized
+        cassette["provider"] = {
+            "name": "sglang",
+            "version": version,
+            "model": model,
+            "transport": "http-sse" if cassette["turns"][0]["request"]["body"]["stream"] else "http-json",
+            "launch": launch,
+            "capabilities_exercised": (
+                ["text", "gateway_stateful_continuation"]
+                if "stateful" in path.name else ["client_function_call"]
+            ),
+            "unverified": ["parallel_function_calls", "structured_text", "upstream_websocket", "reasoning_summary"],
+        }
+        path.write_text(yaml.safe_dump(cassette, sort_keys=False, allow_unicode=True, width=10**9))
+
+
+if __name__ == "__main__":
+    prepare(Path(sys.argv[1]), sys.argv[2], sys.argv[3], sys.argv[4])

crates/agentic-server-core/tests/cassettes/record_sglang_cassettes.sh

@@ -0,0 +1,131 @@
+#!/usr/bin/env bash
+# Record real SGLang traffic using the shared recorder and Dynamo scenario definitions.
+# See docs/guides/sglang-upstream.md for the pinned launch configuration.
+
+set -euo pipefail
+
+SCRIPTS_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+DEST_DIR="$SCRIPTS_DIR/sglang"
+STAGING_ROOT="$(mktemp -d -t agentic-sglang-recording.XXXXXXXX)"
+BASE_DIR="$STAGING_ROOT/sglang"
+TOOLS_FILE="$SCRIPTS_DIR/tool_calls/tools.json"
+SGLANG_URL="${SGLANG_URL:-http://127.0.0.1:30000}"
+MODEL="${MODEL:-Qwen/Qwen3-8B}"
+MODEL_SLUG="$(echo "$MODEL" | tr '/: ' '---')"
+PYTHON="${PYTHON:-python}"
+SGLANG_VERSION="${SGLANG_VERSION:-0.5.18}"
+
+green() { printf '\033[32m%s\033[0m\n' "$*"; }
+bold()  { printf '\033[1m%s\033[0m\n'  "$*"; }
+
+mkdir -p "$BASE_DIR"
+
+bold "SGLang URL: $SGLANG_URL"
+bold "Model:      $MODEL"
+echo
+
+# record NAME STREAM_FLAG PROMPT [extra recorder args...]
+record() {
+    local name="$1" stream_flag="$2" prompt="$3"; shift 3
+    local suffix; [[ -n "$stream_flag" ]] && suffix=nonstreaming || suffix=streaming
+    bold "── $name ($suffix) ──"
+    record_into "$BASE_DIR/${name}-${MODEL_SLUG}-${suffix}.yaml" "$stream_flag" "$prompt" "$@"
+    green "✓ $name ($suffix) done."
+}
+
+# append_turn NAME STREAM_FLAG PROMPT [extra recorder args...]
+# The recorder truncates its output file, so extra turns are recorded separately and merged.
+append_turn() {
+    local name="$1" stream_flag="$2" prompt="$3"; shift 3
+    local suffix; [[ -n "$stream_flag" ]] && suffix=nonstreaming || suffix=streaming
+    local target="$BASE_DIR/${name}-${MODEL_SLUG}-${suffix}.yaml"
+    local extra="${target%.yaml}.next-turn.yaml"
+    bold "── $name ($suffix), next turn ──"
+    record_into "$extra" "$stream_flag" "$prompt" "$@"
+    $PYTHON - "$target" "$extra" <<'PY'
+import sys, yaml
+target, extra = sys.argv[1], sys.argv[2]
+merged = yaml.safe_load(open(target))
+for turn in yaml.safe_load(open(extra))["turns"]:
+    turn["filename"] = f"t{len(merged['turns']) + 1}"
+    merged["turns"].append(turn)
+yaml.safe_dump(merged, open(target, "w"), sort_keys=False, allow_unicode=True, width=10**9)
+PY
+    rm -f "$extra"
+    green "✓ $name ($suffix) turn appended."
+}
+
+record_into() {
+    local output="$1" stream_flag="$2" prompt="$3"; shift 3
+    # shellcheck disable=SC2086
+    printf '%s\n' "$prompt" | $PYTHON "$SCRIPTS_DIR/record_cassette.py" \
+        --mode responses \
+        --turns 1 \
+        --model "$MODEL" \
+        --vllm "$SGLANG_URL" \
+        --max-output-tokens 2048 \
+        $stream_flag \
+        "$@" \
+        --output "$output"
+}
+
+# hydrated_turn2_input CASSETTE OUT_JSON
+# Writes the item history the gateway sends for turn 2: the user prompt, the assistant message exactly as
+# recorded in turn 1 (same id and text), and the follow-up user prompt.
+hydrated_turn2_input() {
+    $PYTHON - "$1" "$2" "$TURN1_PROMPT" "$TURN2_PROMPT" <<'PY'
+import json, sys, yaml
+cassette, out, turn1, turn2 = sys.argv[1:5]
+response = yaml.safe_load(open(cassette))["turns"][0]["response"]
+if response.get("body"):
+    completed = response["body"]
+else:
+    completed = next(
+        json.loads(line[len("data: "):])["response"]
+        for raw in response["sse"]
+        for line in raw.splitlines()
+        if line.startswith("data: ") and json.loads(line[len("data: "):]).get("type") == "response.completed"
+    )
+history = [{"type": "message", "role": "user", "content": turn1}]
+for item in completed["output"]:
+    if item["type"] == "reasoning":
+        history.append(item)
+    elif item["type"] == "message":
+        history.append({
+        "type": "message",
+        "id": item["id"],
+        "role": "assistant",
+        "status": "completed",
+        "content": [{"type": "output_text", "text": part["text"]} for part in item["content"]],
+        })
+    else:
+        raise ValueError(f"Unexpected output item in text scenario: {item['type']}")
+history.append({"type": "message", "role": "user", "content": turn2})
+json.dump(history, open(out, "w"), indent=2)
+PY
+}
+
+TURN1_PROMPT="Remember the word APPLE. Just say: OK"
+TURN2_PROMPT="What word did I ask you to remember? Reply with just the word."
+
+for stream_flag in --no-stream ""; do
+    [[ -n "$stream_flag" ]] && suffix=nonstreaming || suffix=streaming
+    record sglang-stateful "$stream_flag" "$TURN1_PROMPT"
+    turn2_input="$(mktemp --suffix=.json)"
+    hydrated_turn2_input "$BASE_DIR/sglang-stateful-${MODEL_SLUG}-${suffix}.yaml" "$turn2_input"
+    append_turn sglang-stateful "$stream_flag" "" --input-file "$turn2_input"
+    rm -f "$turn2_input"
+    record sglang-tool-call-auto "$stream_flag" "What is the current NVIDIA stock price? Use the tool." \
+        --tools "$TOOLS_FILE" --tool-choice auto
+done
+
+echo
+"$PYTHON" "$SCRIPTS_DIR/prepare_sglang_recordings.py" "$BASE_DIR" "$SGLANG_VERSION" "$MODEL" "$SGLANG_URL"
+"$PYTHON" "$SCRIPTS_DIR/../../../../scripts/validate-cassettes.py" "$BASE_DIR"
+(
+    cd "$SCRIPTS_DIR/../../../.."
+    PROVIDER_CASSETTE_ROOT="$STAGING_ROOT" cargo test -p agentic-server-core --test sglang_cassette_test
+)
+mkdir -p "$DEST_DIR"
+cp "$BASE_DIR"/*.yaml "$DEST_DIR/"
+green "All SGLang cassettes validated and published -> $DEST_DIR (staging: $STAGING_ROOT)"