From 07c6c6d8735cc704cff5bb703f62e128212ab303 Mon Sep 17 00:00:00 2001 From: Clifton King Date: Tue, 4 Aug 2026 13:08:29 -0500 Subject: [PATCH] Add sanitized fixture record and replay --- Cargo.toml | 4 + README.md | 43 +- docs/COOKBOOK.md | 32 + examples/fixture_record_replay.rs | 54 + src/backend/fixture.rs | 1312 +++++++++++++++++ src/backend/mod.rs | 8 +- src/backend/usage.rs | 48 +- src/diagnostics.rs | 6 +- src/error/mod.rs | 5 +- src/lib.rs | 7 +- tests/documentation_gallery_tests.rs | 5 + tests/fixture_replay_tests.rs | 59 + .../record_replay/apple_2023_10k.fixture.json | 85 ++ 13 files changed, 1653 insertions(+), 15 deletions(-) create mode 100644 examples/fixture_record_replay.rs create mode 100644 src/backend/fixture.rs create mode 100644 tests/fixture_replay_tests.rs create mode 100644 tests/fixtures/record_replay/apple_2023_10k.fixture.json diff --git a/Cargo.toml b/Cargo.toml index a09cb24..1f16a66 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -108,6 +108,10 @@ required-features = ["derive", "openai", "tools"] name = "mock_testing_example" required-features = ["derive", "mock"] +[[example]] +name = "fixture_record_replay" +required-features = ["derive", "mock"] + [[example]] name = "retry_attempt_ledger" required-features = ["derive", "openai"] diff --git a/README.md b/README.md index 6f5c1fb..64a53d6 100644 --- a/README.md +++ b/README.md @@ -203,6 +203,7 @@ complete copy-paste recipes. | Read a chart with Kimi K3 | [`kimi_k3_multimodal_example.rs`](examples/kimi_k3_multimodal_example.rs) | Downloads a labeled revenue chart, sends it through Moonshot's OpenAI-compatible endpoint, and returns typed values plus calculated insights. | | Put extraction behind an axum handler | [`axum_handler_example.rs`](examples/axum_handler_example.rs) | Injects any `LLMClient` into typed JSON request and response handling, tested in-process. | | Test without network | [`mock_testing_example.rs`](examples/mock_testing_example.rs) | Scripts realistic responses through the real deserialization, validation, and re-ask path. | +| Record and replay a sanitized fixture | [`fixture_record_replay.rs`](examples/fixture_record_replay.rs) | Persists a versioned interaction with usage and attempts, then strictly replays it offline. | | Use a local model with Ollama | [`ollama_local_example.rs`](examples/ollama_local_example.rs) | Connects to the keyless local endpoint through the same structured-output API. | | Choose a provider at runtime | [`runtime_provider_example.rs`](examples/runtime_provider_example.rs) | Parses `provider/model` into one `AnyClient`, including aggregator model IDs with slashes. | | Reuse an existing schemars model | [`schemars_bridge_example.rs`](examples/schemars_bridge_example.rs) | Materializes `JsonSchema + Serde` types through the transparent `Schemars` adapter. | @@ -958,6 +959,46 @@ feature pulls in only the lightweight path-aware decoder and works without the H client; streaming and tool-loop mocking light up when the `streaming` / `tools` features are also enabled. See `examples/mock_testing_example.rs`. +### Record, sanitize, and replay fixtures + +`FixtureRecorder` is an `LLMClient` wrapper for turning representative +non-streaming calls into versioned JSON fixtures. Sanitization is mandatory and +runs before a request or response is retained. Credential-shaped JSON fields are +redacted structurally, inline media bytes are never stored, and the callback lets +you remove domain-specific identifiers: + +```rust +use rstructor::{ + Fixture, FixtureRecorder, FixtureSanitizer, Instructor, LLMClient, OpenAIClient, +}; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, PartialEq, Instructor, Serialize, Deserialize)] +struct Position { account_id: String, symbol: String, quantity: i64 } + +fn sanitizer() -> FixtureSanitizer { + FixtureSanitizer::new(|text| text.replace("HF-ALPHA-001", "[ACCOUNT]")) +} + +let recorder = FixtureRecorder::new(OpenAIClient::from_env()?, sanitizer()); +let position: Position = recorder.extract("HF-ALPHA-001 owns 1,000 AAPL shares").await?; +recorder.save("tests/fixtures/position.fixture.json")?; + +// CI: no API key or network. +let fixture = Fixture::load("tests/fixtures/position.fixture.json")?; +let replay = fixture.replay_with_sanitizer(sanitizer()); +let replayed: Position = replay.extract("HF-ALPHA-001 owns 1,000 AAPL shares").await?; +assert_eq!(replayed, position); +replay.assert_finished()?; +``` + +Replay is ordered and strict. Operation, sanitized prompt, target schema, and +media metadata must match before an interaction is consumed, so prompt or type +drift fails locally instead of silently returning a stale fixture. The fixture +also preserves token usage, attempt reports, typed provider errors, HTTP status, +and request IDs. See `examples/fixture_record_replay.rs` for a runnable, +key-free round trip based on a real 10-K metric. + ## Feature Flags ```toml @@ -970,7 +1011,7 @@ rstructor = { version = "0.5.1", features = ["openai", "anthropic", "grok", "gem - `logging` — Tracing integration - `streaming` — Streaming via `generate_stream` / `materialize_iter` / `materialize_stream` (opt-in) - `tools` — Tool/function calling via `Toolbox` + `client.with_tools(..).run(..)` (opt-in) -- `mock` — `MockClient` for offline unit testing (opt-in; see [Testing](#testing-offline)) +- `mock` — `MockClient` plus record/sanitize/replay fixtures for offline testing (opt-in; see [Testing](#testing-offline)) All features are on by default. For a **schema-only build** — generate JSON Schema from your types with no networking, `tokio`, or `reqwest` — disable the providers: diff --git a/docs/COOKBOOK.md b/docs/COOKBOOK.md index d916b50..c6a767c 100644 --- a/docs/COOKBOOK.md +++ b/docs/COOKBOOK.md @@ -309,6 +309,38 @@ async fn main() -> Result<(), Box> { Scripted payloads still pass through the production Serde and `Instructor` validation path; only the provider transport is replaced. +## Record and replay a sanitized fixture + +The [fixture example](../examples/fixture_record_replay.rs) records a complete +non-streaming interaction and replays it without a key or network. The required +sanitizer runs before values enter the in-memory fixture; inline media bytes are +never persisted. + +```rust +use rstructor::{Fixture, FixtureRecorder, FixtureSanitizer, LLMClient, OpenAIClient}; + +fn sanitizer() -> FixtureSanitizer { + FixtureSanitizer::new(|text| text.replace("PRIVATE-ACCOUNT", "[ACCOUNT]")) +} + +# async fn example() -> Result<(), Box> { +let recorder = FixtureRecorder::new(OpenAIClient::from_env()?, sanitizer()); +let fill: Fill = recorder.extract("PRIVATE-ACCOUNT bought 5,000 AAPL at 238").await?; +recorder.save("tests/fixtures/fill.fixture.json")?; + +let fixture = Fixture::load("tests/fixtures/fill.fixture.json")?; +let replay = fixture.replay_with_sanitizer(sanitizer()); +let replayed: Fill = replay.extract("PRIVATE-ACCOUNT bought 5,000 AAPL at 238").await?; +assert_eq!(replayed, fill); +replay.assert_finished()?; +# Ok(()) +# } +``` + +Replay matches the sanitized operation, prompt, schema, and media metadata in +order. A mismatch leaves the interaction unconsumed and reports only the field +that differed, so assertion failures do not echo fixture contents. + ## Use a local model through Ollama The [Ollama example](../examples/ollama_local_example.rs) is safe to run in CI: diff --git a/examples/fixture_record_replay.rs b/examples/fixture_record_replay.rs new file mode 100644 index 0000000..1acbb12 --- /dev/null +++ b/examples/fixture_record_replay.rs @@ -0,0 +1,54 @@ +//! Record, sanitize, save, and strictly replay an offline fixture. +//! +//! Run with: +//! `cargo run --example fixture_record_replay --features mock` + +use rstructor::{ + Fixture, FixtureRecorder, FixtureSanitizer, Instructor, LLMClient, MockClient, TokenUsage, +}; +use serde::{Deserialize, Serialize}; + +const APPLE_2023_PROMPT: &str = "From Apple Inc.'s 2023 Form 10-K: Total net sales were $383,285 million for the fiscal year ended September 30, 2023. Extract the issuer, fiscal year, and net sales in USD millions."; + +#[derive(Debug, PartialEq, Instructor, Serialize, Deserialize)] +struct FilingMetric { + issuer: String, + fiscal_year: u16, + net_sales_usd_millions: u64, +} + +fn fixture_sanitizer() -> FixtureSanitizer { + // Replace account IDs, customer names, or other private strings here. + // Credential-shaped JSON fields and inline media bytes are redacted separately. + FixtureSanitizer::new(str::to_owned) +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + // A live recording run can wrap OpenAIClient, AnthropicClient, GeminiClient, + // or GrokClient in exactly the same way. MockClient keeps this example key-free. + let source = MockClient::new().with_response_and_usage( + r#"{"issuer":"Apple Inc.","fiscal_year":2023,"net_sales_usd_millions":383285}"#, + TokenUsage::new("recorded-provider-model", 67, 24), + ); + let recorder = FixtureRecorder::new(source, fixture_sanitizer()); + let recorded = recorder + .extract_with_report::(APPLE_2023_PROMPT) + .await?; + assert_eq!(recorded.data.net_sales_usd_millions, 383_285); + + std::fs::create_dir_all("tmp")?; + recorder.save("tmp/apple-2023-10k.fixture.json")?; + + let fixture = Fixture::load("tmp/apple-2023-10k.fixture.json")?; + let replay = fixture.replay_with_sanitizer(fixture_sanitizer()); + let replayed = replay + .extract_with_report::(APPLE_2023_PROMPT) + .await?; + assert_eq!(replayed.data, recorded.data); + assert_eq!(replayed.report, recorded.report); + replay.assert_finished()?; + + println!("recorded and replayed {} interaction", fixture.len()); + Ok(()) +} diff --git a/src/backend/fixture.rs b/src/backend/fixture.rs new file mode 100644 index 0000000..da4d73d --- /dev/null +++ b/src/backend/fixture.rs @@ -0,0 +1,1312 @@ +//! Versioned, privacy-aware fixtures for recording and replaying LLM calls. +//! +//! Wrap a live or mock client in [`FixtureRecorder`], make normal non-streaming +//! calls through the wrapper, and persist the resulting [`Fixture`]. Tests can +//! load that file and replay it through [`ReplayClient`] without a network or +//! API key. Replay is ordered and strict: the operation, prompt, schema, and +//! media metadata must match before a recorded response is consumed. + +use std::collections::{BTreeSet, VecDeque}; +use std::fmt; +use std::path::Path; +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use thiserror::Error; + +use crate::ResponseMetadata; +use crate::backend::ModelInfo; +use crate::backend::client::{LLMClient, MediaFile}; +use crate::backend::usage::{ + AttemptOutcome, ExtractionReport, GenerateResult, MaterializeFailure, MaterializeReport, + MaterializeResult, RunUsage, TokenUsage, +}; +use crate::error::{ApiErrorKind, RStructorError, Result, StreamErrorKind}; +use crate::model::Instructor; +use crate::schema::SchemaType; + +/// Current on-disk fixture schema version. +pub const FIXTURE_SCHEMA_VERSION: u32 = 1; + +type TextSanitizer = dyn Fn(&str) -> String + Send + Sync + 'static; + +/// Mandatory privacy boundary used before requests and responses enter a fixture. +/// +/// The callback runs on every retained string. Common credential-shaped JSON +/// fields are also replaced structurally, and inline media bytes are never +/// stored. Build the same sanitizer in replay tests when it changes request +/// text that must be matched. +#[derive(Clone)] +pub struct FixtureSanitizer { + sanitize_text: Arc, + redacted_json_keys: BTreeSet, +} + +impl FixtureSanitizer { + /// Create a sanitizer with a required string callback and safe JSON-key defaults. + pub fn new(sanitize_text: F) -> Self + where + F: Fn(&str) -> String + Send + Sync + 'static, + { + let redacted_json_keys = [ + "api_key", + "authorization", + "cookie", + "password", + "secret", + "set_cookie", + "token", + ] + .into_iter() + .map(str::to_string) + .collect(); + + Self { + sanitize_text: Arc::new(sanitize_text), + redacted_json_keys, + } + } + + /// Add case-insensitive JSON keys whose values must be replaced entirely. + #[must_use] + pub fn redact_json_keys(mut self, keys: I) -> Self + where + I: IntoIterator, + S: AsRef, + { + self.redacted_json_keys + .extend(keys.into_iter().map(|key| normalize_json_key(key.as_ref()))); + self + } + + fn text(&self, text: &str) -> String { + (self.sanitize_text)(text) + } + + fn json(&self, value: &Value) -> Value { + match value { + Value::Object(object) => Value::Object( + object + .iter() + .map(|(key, value)| { + let value = if self.redacted_json_keys.contains(&normalize_json_key(key)) { + Value::String("[REDACTED]".to_string()) + } else { + self.json(value) + }; + (key.clone(), value) + }) + .collect(), + ), + Value::Array(items) => Value::Array(items.iter().map(|item| self.json(item)).collect()), + Value::String(text) => Value::String(self.text(text)), + primitive => primitive.clone(), + } + } + + fn request(&self, mut request: StoredRequest) -> StoredRequest { + request.prompt = self.text(&request.prompt); + request.schema_name = request.schema_name.map(|name| self.text(&name)); + request.schema = request.schema.map(|schema| self.json(&schema)); + for media in &mut request.media { + media.uri = self.text(&media.uri); + media.mime_type = self.text(&media.mime_type); + } + request + } + + fn report(&self, mut report: ExtractionReport) -> ExtractionReport { + sanitize_usage(self, report.final_usage.as_mut()); + if let Some(cumulative) = report.cumulative_usage.as_mut() { + sanitize_run_usage(self, cumulative); + } + for attempt in &mut report.attempts { + sanitize_usage(self, attempt.usage.as_mut()); + if let AttemptOutcome::Failed { message, .. } = &mut attempt.outcome { + *message = self.text(message); + } + if let Some(response) = attempt.response.as_mut() { + sanitize_response_metadata(self, response); + } + } + report + } +} + +impl fmt::Debug for FixtureSanitizer { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("FixtureSanitizer") + .field("sanitize_text", &"") + .field("redacted_json_keys", &self.redacted_json_keys) + .finish() + } +} + +fn normalize_json_key(key: &str) -> String { + key.trim().to_ascii_lowercase().replace(['-', ' '], "_") +} + +fn sanitize_usage(sanitizer: &FixtureSanitizer, usage: Option<&mut TokenUsage>) { + if let Some(usage) = usage { + usage.model = sanitizer.text(&usage.model); + } +} + +fn sanitize_run_usage(sanitizer: &FixtureSanitizer, usage: &mut RunUsage) { + let old = std::mem::take(&mut usage.by_model); + usage.by_model = old + .into_iter() + .map(|(model, mut token_usage)| { + token_usage.model = sanitizer.text(&token_usage.model); + (sanitizer.text(&model), token_usage) + }) + .collect(); +} + +fn sanitize_response_metadata(sanitizer: &FixtureSanitizer, response: &mut ResponseMetadata) { + for request_id in response.request_ids.values_mut() { + *request_id = sanitizer.text(request_id); + } + if let Some(body) = response.sanitized_body.as_mut() { + body.text = sanitize_body(sanitizer, &body.text); + } +} + +fn sanitize_body(sanitizer: &FixtureSanitizer, body: &str) -> String { + serde_json::from_str(body).map_or_else( + |_| sanitizer.text(body), + |value| sanitizer.json(&value).to_string(), + ) +} + +/// File-format and replay-completion errors for fixtures. +#[derive(Debug, Error)] +#[non_exhaustive] +pub enum FixtureError { + /// The fixture could not be read or written. + #[error("fixture I/O error: {0}")] + Io(#[from] std::io::Error), + /// The fixture JSON was invalid. + #[error("invalid fixture JSON: {0}")] + Json(#[from] serde_json::Error), + /// The fixture was written with an unsupported schema version. + #[error("unsupported fixture schema version {found}; this build supports {supported}")] + UnsupportedVersion { + /// Version found in the fixture. + found: u32, + /// Version supported by this library build. + supported: u32, + }, + /// A replay finished while recorded interactions remained unused. + #[error("fixture replay left {remaining} interaction(s) unused")] + ReplayIncomplete { + /// Number of unused interactions. + remaining: usize, + }, +} + +/// A versioned collection of sanitized, ordered request/response interactions. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[non_exhaustive] +pub struct Fixture { + schema_version: u32, + interactions: Vec, +} + +impl Default for Fixture { + fn default() -> Self { + Self::new() + } +} + +impl Fixture { + /// Create an empty fixture using the current schema version. + #[must_use] + pub fn new() -> Self { + Self { + schema_version: FIXTURE_SCHEMA_VERSION, + interactions: Vec::new(), + } + } + + /// Return the on-disk schema version. + #[must_use] + pub fn schema_version(&self) -> u32 { + self.schema_version + } + + /// Number of recorded interactions. + #[must_use] + pub fn len(&self) -> usize { + self.interactions.len() + } + + /// Whether no interactions have been recorded. + #[must_use] + pub fn is_empty(&self) -> bool { + self.interactions.is_empty() + } + + /// Parse and version-check a fixture from JSON. + pub fn from_json(json: &str) -> std::result::Result { + let fixture: Self = serde_json::from_str(json)?; + fixture.validate_version()?; + Ok(fixture) + } + + /// Serialize the fixture as stable, pretty JSON with a trailing newline. + pub fn to_json(&self) -> std::result::Result { + self.validate_version()?; + let mut json = serde_json::to_string_pretty(self)?; + json.push('\n'); + Ok(json) + } + + /// Load and version-check a fixture file. + pub fn load(path: impl AsRef) -> std::result::Result { + Self::from_json(&std::fs::read_to_string(path)?) + } + + /// Write a fixture as stable, pretty JSON. + pub fn save(&self, path: impl AsRef) -> std::result::Result<(), FixtureError> { + std::fs::write(path, self.to_json()?)?; + Ok(()) + } + + /// Create a strict offline replay client using identity string sanitization. + #[must_use] + pub fn replay(&self) -> ReplayClient { + self.replay_with_sanitizer(FixtureSanitizer::new(str::to_owned)) + } + + /// Create a replay client that sanitizes incoming requests before matching. + /// + /// Use the same sanitizer construction as the recording run when request + /// prompts or schema string values were transformed. + #[must_use] + pub fn replay_with_sanitizer(&self, sanitizer: FixtureSanitizer) -> ReplayClient { + ReplayClient::new(self.clone(), sanitizer) + } + + fn validate_version(&self) -> std::result::Result<(), FixtureError> { + if self.schema_version == FIXTURE_SCHEMA_VERSION { + Ok(()) + } else { + Err(FixtureError::UnsupportedVersion { + found: self.schema_version, + supported: FIXTURE_SCHEMA_VERSION, + }) + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +enum StoredOperation { + Materialize, + MaterializeWithMedia, + MaterializeWithMetadata, + MaterializeWithAttempts, + MaterializeWithMediaAndAttempts, + Generate, + GenerateWithMedia, + GenerateWithMetadata, + ListModels, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +struct StoredRequest { + operation: StoredOperation, + prompt: String, + #[serde(skip_serializing_if = "Option::is_none")] + schema_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + schema: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + media: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +struct StoredMedia { + uri: String, + mime_type: String, + inline_data: bool, +} + +impl StoredMedia { + fn from_media(media: &MediaFile) -> Self { + Self { + uri: media.uri.clone(), + mime_type: media.mime_type.clone(), + inline_data: media.data.is_some(), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct StoredInteraction { + request: StoredRequest, + response: StoredResponse, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum StoredResponse { + Success { + body: String, + #[serde(skip_serializing_if = "Option::is_none")] + usage: Option, + #[serde(skip_serializing_if = "Option::is_none")] + report: Option, + }, + Error { + error: StoredError, + #[serde(skip_serializing_if = "Option::is_none")] + report: Option, + }, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +enum StoredError { + Api { + provider: String, + api_kind: ApiErrorKind, + response: Option>, + }, + Validation { + message: String, + }, + Schema { + message: String, + }, + SchemaCompatibility { + provider: String, + context: String, + path: String, + message: String, + }, + Serialization { + message: String, + }, + OutputDecode { + path: String, + message: String, + }, + ToolArgumentDecode { + path: String, + message: String, + }, + Streaming { + stream_kind: StreamErrorKind, + message: String, + }, + Timeout, + Unsupported { + message: String, + }, +} + +impl StoredError { + fn from_error(error: &RStructorError, sanitizer: &FixtureSanitizer) -> Self { + match error { + RStructorError::ApiError { + provider, + kind, + response, + } => { + let mut kind = kind.clone(); + sanitize_api_error_kind(sanitizer, &mut kind); + let mut response = response.clone(); + if let Some(response) = response.as_mut() { + sanitize_response_metadata(sanitizer, response); + } + Self::Api { + provider: sanitizer.text(provider), + api_kind: kind, + response, + } + } + RStructorError::ValidationError(message) => Self::Validation { + message: sanitizer.text(message), + }, + RStructorError::SchemaError(message) => Self::Schema { + message: sanitizer.text(message), + }, + RStructorError::SchemaCompatibilityError { + provider, + context, + path, + message, + } => Self::SchemaCompatibility { + provider: sanitizer.text(provider), + context: sanitizer.text(context), + path: sanitizer.text(path), + message: sanitizer.text(message), + }, + RStructorError::SerializationError(message) => Self::Serialization { + message: sanitizer.text(message), + }, + RStructorError::OutputDecodeError { path, message } => Self::OutputDecode { + path: sanitizer.text(path), + message: sanitizer.text(message), + }, + RStructorError::ToolArgumentDecodeError { path, message } => Self::ToolArgumentDecode { + path: sanitizer.text(path), + message: sanitizer.text(message), + }, + RStructorError::StreamingError { kind, message } => Self::Streaming { + stream_kind: *kind, + message: sanitizer.text(message), + }, + RStructorError::Timeout => Self::Timeout, + RStructorError::Unsupported(message) => Self::Unsupported { + message: sanitizer.text(message), + }, + #[cfg(feature = "_client")] + RStructorError::HttpError(error) => Self::Unsupported { + message: sanitizer.text(&format!("recorded HTTP transport error: {error}")), + }, + RStructorError::JsonError(error) => Self::Serialization { + message: sanitizer.text(&format!("recorded JSON error: {error}")), + }, + } + } + + fn into_error(self) -> RStructorError { + match self { + Self::Api { + provider, + api_kind, + response, + } => RStructorError::ApiError { + provider, + kind: api_kind, + response, + }, + Self::Validation { message } => RStructorError::ValidationError(message), + Self::Schema { message } => RStructorError::SchemaError(message), + Self::SchemaCompatibility { + provider, + context, + path, + message, + } => RStructorError::SchemaCompatibilityError { + provider: provider.into(), + context: context.into(), + path: path.into(), + message: message.into(), + }, + Self::Serialization { message } => RStructorError::SerializationError(message), + Self::OutputDecode { path, message } => { + RStructorError::OutputDecodeError { path, message } + } + Self::ToolArgumentDecode { path, message } => { + RStructorError::ToolArgumentDecodeError { path, message } + } + Self::Streaming { + stream_kind, + message, + } => RStructorError::StreamingError { + kind: stream_kind, + message: message.into(), + }, + Self::Timeout => RStructorError::Timeout, + Self::Unsupported { message } => RStructorError::Unsupported(message), + } + } +} + +fn sanitize_api_error_kind(sanitizer: &FixtureSanitizer, kind: &mut ApiErrorKind) { + match kind { + ApiErrorKind::InvalidModel { model, suggestion } => { + *model = sanitizer.text(model); + *suggestion = suggestion.take().map(|value| sanitizer.text(&value)); + } + ApiErrorKind::BadRequest { details } | ApiErrorKind::UnexpectedResponse { details } => { + *details = sanitizer.text(details); + } + ApiErrorKind::Other { message, .. } => { + *message = sanitizer.text(message); + } + _ => {} + } +} + +fn structured_request( + operation: StoredOperation, + prompt: &str, + media: &[MediaFile], +) -> Result +where + T: Instructor, +{ + Ok(StoredRequest { + operation, + prompt: prompt.to_string(), + schema_name: ::schema_name(), + schema: Some(::try_schema()?.to_json()), + media: media.iter().map(StoredMedia::from_media).collect(), + }) +} + +fn text_request(operation: StoredOperation, prompt: &str, media: &[MediaFile]) -> StoredRequest { + StoredRequest { + operation, + prompt: prompt.to_string(), + schema_name: None, + schema: None, + media: media.iter().map(StoredMedia::from_media).collect(), + } +} + +fn sanitized_structured_body(value: &T, sanitizer: &FixtureSanitizer) -> Result +where + T: Serialize, +{ + let value = serde_json::to_value(value) + .map_err(|error| RStructorError::SerializationError(error.to_string()))?; + serde_json::to_string(&sanitizer.json(&value)) + .map_err(|error| RStructorError::SerializationError(error.to_string())) +} + +fn success_report(report: &MaterializeReport) -> ExtractionReport { + ExtractionReport { + final_usage: report.final_usage.clone(), + cumulative_usage: report.cumulative_usage.clone(), + attempts: report.attempts.clone(), + attempts_complete: report.attempts_complete, + } +} + +fn failure_report(failure: &MaterializeFailure) -> ExtractionReport { + ExtractionReport { + final_usage: failure + .attempts + .last() + .and_then(|attempt| attempt.usage.clone()), + cumulative_usage: failure.cumulative_usage.clone(), + attempts: failure.attempts.clone(), + attempts_complete: failure.attempts_complete, + } +} + +/// An [`LLMClient`] wrapper that records sanitized non-streaming interactions. +/// +/// Recording is in memory until [`fixture`](Self::fixture) or [`save`](Self::save) +/// is called. The sanitizer is applied synchronously before an interaction is +/// retained; the recorder never keeps the original prompt or output. +pub struct FixtureRecorder { + inner: C, + sanitizer: FixtureSanitizer, + fixture: Arc>, +} + +impl FixtureRecorder { + /// Wrap a client with a mandatory fixture sanitizer. + #[must_use] + pub fn new(inner: C, sanitizer: FixtureSanitizer) -> Self { + Self { + inner, + sanitizer, + fixture: Arc::new(Mutex::new(Fixture::new())), + } + } + + /// Borrow the wrapped client. + #[must_use] + pub fn inner(&self) -> &C { + &self.inner + } + + /// Snapshot the interactions recorded so far. + #[must_use] + pub fn fixture(&self) -> Fixture { + self.fixture.lock().unwrap().clone() + } + + /// Number of interactions recorded so far. + #[must_use] + pub fn len(&self) -> usize { + self.fixture.lock().unwrap().len() + } + + /// Whether no interactions have been recorded. + #[must_use] + pub fn is_empty(&self) -> bool { + self.fixture.lock().unwrap().is_empty() + } + + /// Persist a snapshot of the interactions recorded so far. + pub fn save(&self, path: impl AsRef) -> std::result::Result<(), FixtureError> { + self.fixture().save(path) + } + + fn record(&self, request: StoredRequest, response: StoredResponse) { + self.fixture + .lock() + .unwrap() + .interactions + .push(StoredInteraction { + request: self.sanitizer.request(request), + response, + }); + } + + fn success( + &self, + request: StoredRequest, + body: String, + mut usage: Option, + report: Option, + ) { + sanitize_usage(&self.sanitizer, usage.as_mut()); + let report = report.map(|report| self.sanitizer.report(report)); + self.record( + request, + StoredResponse::Success { + body, + usage, + report, + }, + ); + } + + fn error( + &self, + request: StoredRequest, + error: &RStructorError, + report: Option, + ) { + let report = report.map(|report| self.sanitizer.report(report)); + self.record( + request, + StoredResponse::Error { + error: StoredError::from_error(error, &self.sanitizer), + report, + }, + ); + } +} + +impl fmt::Debug for FixtureRecorder { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("FixtureRecorder") + .field("inner", &std::any::type_name::()) + .field("recorded_interactions", &self.len()) + .field("sanitizer", &self.sanitizer) + .finish() + } +} + +#[async_trait] +impl LLMClient for FixtureRecorder +where + C: LLMClient + Sync, +{ + async fn materialize(&self, prompt: &str) -> Result + where + T: Instructor + serde::de::DeserializeOwned + Send + 'static, + { + let request = structured_request::(StoredOperation::Materialize, prompt, &[])?; + match self.inner.materialize(prompt).await { + Ok(data) => { + let body = sanitized_structured_body(&data, &self.sanitizer)?; + self.success(request, body, None, None); + Ok(data) + } + Err(error) => { + self.error(request, &error, None); + Err(error) + } + } + } + + async fn materialize_with_media(&self, prompt: &str, media: &[MediaFile]) -> Result + where + T: Instructor + serde::de::DeserializeOwned + Send + 'static, + { + let request = + structured_request::(StoredOperation::MaterializeWithMedia, prompt, media)?; + match self.inner.materialize_with_media(prompt, media).await { + Ok(data) => { + let body = sanitized_structured_body(&data, &self.sanitizer)?; + self.success(request, body, None, None); + Ok(data) + } + Err(error) => { + self.error(request, &error, None); + Err(error) + } + } + } + + async fn materialize_with_metadata(&self, prompt: &str) -> Result> + where + T: Instructor + serde::de::DeserializeOwned + Send + 'static, + { + let request = + structured_request::(StoredOperation::MaterializeWithMetadata, prompt, &[])?; + match self.inner.materialize_with_metadata(prompt).await { + Ok(result) => { + let body = sanitized_structured_body(&result.data, &self.sanitizer)?; + self.success(request, body, result.usage.clone(), None); + Ok(result) + } + Err(error) => { + self.error(request, &error, None); + Err(error) + } + } + } + + async fn materialize_with_attempts( + &self, + prompt: &str, + ) -> std::result::Result, MaterializeFailure> + where + T: Instructor + serde::de::DeserializeOwned + Send + 'static, + { + let request = + structured_request::(StoredOperation::MaterializeWithAttempts, prompt, &[]) + .map_err(MaterializeFailure::from_error)?; + match self.inner.materialize_with_attempts(prompt).await { + Ok(report) => { + let body = sanitized_structured_body(&report.data, &self.sanitizer) + .map_err(MaterializeFailure::from_error)?; + self.success( + request, + body, + report.final_usage.clone(), + Some(success_report(&report)), + ); + Ok(report) + } + Err(failure) => { + self.error(request, failure.error(), Some(failure_report(&failure))); + Err(failure) + } + } + } + + async fn materialize_with_media_and_attempts( + &self, + prompt: &str, + media: &[MediaFile], + ) -> std::result::Result, MaterializeFailure> + where + T: Instructor + serde::de::DeserializeOwned + Send + 'static, + { + let request = structured_request::( + StoredOperation::MaterializeWithMediaAndAttempts, + prompt, + media, + ) + .map_err(MaterializeFailure::from_error)?; + match self + .inner + .materialize_with_media_and_attempts(prompt, media) + .await + { + Ok(report) => { + let body = sanitized_structured_body(&report.data, &self.sanitizer) + .map_err(MaterializeFailure::from_error)?; + self.success( + request, + body, + report.final_usage.clone(), + Some(success_report(&report)), + ); + Ok(report) + } + Err(failure) => { + self.error(request, failure.error(), Some(failure_report(&failure))); + Err(failure) + } + } + } + + async fn generate(&self, prompt: &str) -> Result { + let request = text_request(StoredOperation::Generate, prompt, &[]); + match self.inner.generate(prompt).await { + Ok(text) => { + self.success(request, self.sanitizer.text(&text), None, None); + Ok(text) + } + Err(error) => { + self.error(request, &error, None); + Err(error) + } + } + } + + async fn generate_with_media(&self, prompt: &str, media: &[MediaFile]) -> Result { + let request = text_request(StoredOperation::GenerateWithMedia, prompt, media); + match self.inner.generate_with_media(prompt, media).await { + Ok(text) => { + self.success(request, self.sanitizer.text(&text), None, None); + Ok(text) + } + Err(error) => { + self.error(request, &error, None); + Err(error) + } + } + } + + async fn generate_with_metadata(&self, prompt: &str) -> Result { + let request = text_request(StoredOperation::GenerateWithMetadata, prompt, &[]); + match self.inner.generate_with_metadata(prompt).await { + Ok(result) => { + self.success( + request, + self.sanitizer.text(&result.text), + result.usage.clone(), + None, + ); + Ok(result) + } + Err(error) => { + self.error(request, &error, None); + Err(error) + } + } + } + + fn from_env() -> Result + where + Self: Sized, + { + Err(RStructorError::Unsupported( + "FixtureRecorder requires an explicit inner client and sanitizer".to_string(), + )) + } + + async fn list_models(&self) -> Result> { + let request = text_request(StoredOperation::ListModels, "", &[]); + match self.inner.list_models().await { + Ok(models) => { + let body = sanitized_structured_body(&models, &self.sanitizer)?; + self.success(request, body, None, None); + Ok(models) + } + Err(error) => { + self.error(request, &error, None); + Err(error) + } + } + } +} + +struct ReplayState { + remaining: Mutex>, + total: usize, +} + +/// Strict, ordered, offline replay of a [`Fixture`]. +/// +/// Clones share replay position. Call [`assert_finished`](Self::assert_finished) +/// at the end of a test to catch interactions that were silently skipped. +#[derive(Clone)] +pub struct ReplayClient { + state: Arc, + sanitizer: FixtureSanitizer, +} + +impl ReplayClient { + fn new(fixture: Fixture, sanitizer: FixtureSanitizer) -> Self { + let total = fixture.interactions.len(); + Self { + state: Arc::new(ReplayState { + remaining: Mutex::new(fixture.interactions.into()), + total, + }), + sanitizer, + } + } + + /// Number of recorded interactions not yet consumed. + #[must_use] + pub fn remaining(&self) -> usize { + self.state.remaining.lock().unwrap().len() + } + + /// Fail if replay did not consume every recorded interaction. + pub fn assert_finished(&self) -> std::result::Result<(), FixtureError> { + let remaining = self.remaining(); + if remaining == 0 { + Ok(()) + } else { + Err(FixtureError::ReplayIncomplete { remaining }) + } + } + + fn take(&self, request: StoredRequest) -> Result { + let request = self.sanitizer.request(request); + let mut remaining = self.state.remaining.lock().unwrap(); + let interaction_number = self.state.total - remaining.len() + 1; + let expected = remaining.front().ok_or_else(|| { + RStructorError::Unsupported(format!( + "fixture replay received unexpected interaction {interaction_number}: fixture exhausted" + )) + })?; + if let Some(field) = request_mismatch(&expected.request, &request) { + return Err(RStructorError::Unsupported(format!( + "fixture request mismatch at interaction {interaction_number}: {field} differs" + ))); + } + Ok(remaining.pop_front().unwrap().response) + } + + fn structured(&self, request: StoredRequest) -> Result + where + T: Instructor + serde::de::DeserializeOwned, + { + match self.take(request)? { + StoredResponse::Success { body, .. } => parse_and_validate(&body), + StoredResponse::Error { error, .. } => Err(error.into_error()), + } + } + + fn structured_with_attempts( + &self, + request: StoredRequest, + ) -> std::result::Result, MaterializeFailure> + where + T: Instructor + serde::de::DeserializeOwned, + { + let response = self.take(request).map_err(MaterializeFailure::from_error)?; + match response { + StoredResponse::Success { + body, + usage, + report, + } => { + let data = parse_and_validate(&body).map_err(MaterializeFailure::from_error)?; + Ok(match report { + Some(report) => MaterializeReport::from_fixture_parts( + data, + report.final_usage, + report.cumulative_usage, + report.attempts, + report.attempts_complete, + ), + None => MaterializeReport::from_result(MaterializeResult::new(data, usage)), + }) + } + StoredResponse::Error { error, report } => { + let error = error.into_error(); + Err(match report { + Some(report) => MaterializeFailure::from_fixture_parts( + error, + report.cumulative_usage, + report.attempts, + report.attempts_complete, + ), + None => MaterializeFailure::from_error(error), + }) + } + } + } +} + +impl fmt::Debug for ReplayClient { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("ReplayClient") + .field("remaining", &self.remaining()) + .field("total", &self.state.total) + .finish() + } +} + +fn request_mismatch(expected: &StoredRequest, actual: &StoredRequest) -> Option<&'static str> { + if expected.operation != actual.operation { + Some("operation") + } else if expected.prompt != actual.prompt { + Some("prompt") + } else if expected.schema_name != actual.schema_name { + Some("schema name") + } else if expected.schema != actual.schema { + Some("schema") + } else if expected.media != actual.media { + Some("media metadata") + } else { + None + } +} + +fn parse_and_validate(body: &str) -> Result +where + T: Instructor + serde::de::DeserializeOwned, +{ + let value: T = crate::decode::output_from_str(body)?; + value.validate()?; + Ok(value) +} + +#[async_trait] +impl LLMClient for ReplayClient { + async fn materialize(&self, prompt: &str) -> Result + where + T: Instructor + serde::de::DeserializeOwned + Send + 'static, + { + self.structured(structured_request::( + StoredOperation::Materialize, + prompt, + &[], + )?) + } + + async fn materialize_with_media(&self, prompt: &str, media: &[MediaFile]) -> Result + where + T: Instructor + serde::de::DeserializeOwned + Send + 'static, + { + self.structured(structured_request::( + StoredOperation::MaterializeWithMedia, + prompt, + media, + )?) + } + + async fn materialize_with_metadata(&self, prompt: &str) -> Result> + where + T: Instructor + serde::de::DeserializeOwned + Send + 'static, + { + let request = + structured_request::(StoredOperation::MaterializeWithMetadata, prompt, &[])?; + match self.take(request)? { + StoredResponse::Success { body, usage, .. } => { + Ok(MaterializeResult::new(parse_and_validate(&body)?, usage)) + } + StoredResponse::Error { error, .. } => Err(error.into_error()), + } + } + + async fn materialize_with_attempts( + &self, + prompt: &str, + ) -> std::result::Result, MaterializeFailure> + where + T: Instructor + serde::de::DeserializeOwned + Send + 'static, + { + let request = + structured_request::(StoredOperation::MaterializeWithAttempts, prompt, &[]) + .map_err(MaterializeFailure::from_error)?; + self.structured_with_attempts(request) + } + + async fn materialize_with_media_and_attempts( + &self, + prompt: &str, + media: &[MediaFile], + ) -> std::result::Result, MaterializeFailure> + where + T: Instructor + serde::de::DeserializeOwned + Send + 'static, + { + let request = structured_request::( + StoredOperation::MaterializeWithMediaAndAttempts, + prompt, + media, + ) + .map_err(MaterializeFailure::from_error)?; + self.structured_with_attempts(request) + } + + async fn generate(&self, prompt: &str) -> Result { + match self.take(text_request(StoredOperation::Generate, prompt, &[]))? { + StoredResponse::Success { body, .. } => Ok(body), + StoredResponse::Error { error, .. } => Err(error.into_error()), + } + } + + async fn generate_with_media(&self, prompt: &str, media: &[MediaFile]) -> Result { + match self.take(text_request( + StoredOperation::GenerateWithMedia, + prompt, + media, + ))? { + StoredResponse::Success { body, .. } => Ok(body), + StoredResponse::Error { error, .. } => Err(error.into_error()), + } + } + + async fn generate_with_metadata(&self, prompt: &str) -> Result { + match self.take(text_request( + StoredOperation::GenerateWithMetadata, + prompt, + &[], + ))? { + StoredResponse::Success { body, usage, .. } => Ok(GenerateResult::new(body, usage)), + StoredResponse::Error { error, .. } => Err(error.into_error()), + } + } + + fn from_env() -> Result + where + Self: Sized, + { + Err(RStructorError::Unsupported( + "ReplayClient must be created from a Fixture".to_string(), + )) + } + + async fn list_models(&self) -> Result> { + match self.take(text_request(StoredOperation::ListModels, "", &[]))? { + StoredResponse::Success { body, .. } => serde_json::from_str(&body) + .map_err(|error| RStructorError::SerializationError(error.to_string())), + StoredResponse::Error { error, .. } => Err(error.into_error()), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{Instructor, LLMClient, MockClient}; + use serde::{Deserialize, Serialize}; + + #[derive(Debug, Clone, PartialEq, Instructor, Serialize, Deserialize)] + struct Position { + account_id: String, + symbol: String, + quantity: i64, + } + + fn sanitizer() -> FixtureSanitizer { + FixtureSanitizer::new(|text| text.replace("HF-ALPHA-001", "[ACCOUNT]")) + } + + #[tokio::test] + async fn records_sanitized_report_and_replays_strictly() { + let usage = TokenUsage::new("gpt-5.6-2026-07-15", 82, 19); + let inner = MockClient::new() + .with_response_and_usage( + r#"{"account_id":"HF-ALPHA-001","symbol":"AAPL","quantity":125000}"#, + usage.clone(), + ) + .with_retries(0); + let recorder = FixtureRecorder::new(inner, sanitizer()); + + let extraction = recorder + .extract_with_report::("HF-ALPHA-001 owns 125,000 AAPL shares") + .await + .unwrap(); + assert_eq!(extraction.report.final_usage, Some(usage)); + + let fixture = recorder.fixture(); + let json = fixture.to_json().unwrap(); + assert!(!json.contains("HF-ALPHA-001")); + assert!(json.contains("[ACCOUNT]")); + + let replay = fixture.replay_with_sanitizer(sanitizer()); + let replayed = replay + .extract_with_report::("HF-ALPHA-001 owns 125,000 AAPL shares") + .await + .unwrap(); + assert_eq!(replayed.data.account_id, "[ACCOUNT]"); + assert_eq!(replayed.report, extraction.report); + replay.assert_finished().unwrap(); + } + + #[tokio::test] + async fn request_mismatch_does_not_consume_or_echo_prompts() { + let inner = MockClient::new().with_response("approved"); + let recorder = FixtureRecorder::new(inner, sanitizer()); + recorder.generate("approve HF-ALPHA-001").await.unwrap(); + let replay = recorder.fixture().replay_with_sanitizer(sanitizer()); + + let error = replay.generate("reject HF-ALPHA-001").await.unwrap_err(); + let message = error.to_string(); + assert!(message.contains("prompt differs")); + assert!(!message.contains("approve")); + assert!(!message.contains("reject")); + assert_eq!(replay.remaining(), 1); + } + + #[test] + fn common_sensitive_json_keys_and_inline_media_are_not_retained() { + let sanitizer = FixtureSanitizer::new(str::to_owned); + let value = serde_json::json!({ + "authorization": "Bearer live-secret", + "nested": {"api-key": "live-key", "safe": "kept"} + }); + let sanitized = sanitizer.json(&value); + assert_eq!(sanitized["authorization"], "[REDACTED]"); + assert_eq!(sanitized["nested"]["api-key"], "[REDACTED]"); + assert_eq!(sanitized["nested"]["safe"], "kept"); + + let media = MediaFile { + uri: String::new(), + mime_type: "image/png".to_string(), + data: Some("base64-secret".to_string()), + }; + let stored = StoredMedia::from_media(&media); + let json = serde_json::to_string(&stored).unwrap(); + assert!(stored.inline_data); + assert!(!json.contains("base64-secret")); + } + + #[test] + fn rejects_unknown_schema_version_and_malformed_json() { + let unsupported = r#"{"schema_version":99,"interactions":[]}"#; + assert!(matches!( + Fixture::from_json(unsupported), + Err(FixtureError::UnsupportedVersion { + found: 99, + supported: FIXTURE_SCHEMA_VERSION + }) + )); + assert!(matches!( + Fixture::from_json("{"), + Err(FixtureError::Json(_)) + )); + } + + #[tokio::test] + async fn typed_api_error_round_trips_with_sanitized_diagnostics() { + let mut metadata = ResponseMetadata::new(429); + metadata.request_ids.insert( + "x-request-id".to_string(), + "HF-ALPHA-001-request".to_string(), + ); + let error = RStructorError::api_error_with_response( + "OpenAI", + ApiErrorKind::RateLimited { retry_after: None }, + metadata, + ); + let inner = MockClient::new().with_error(error); + let recorder = FixtureRecorder::new(inner, sanitizer()); + let original = recorder.generate("HF-ALPHA-001").await.unwrap_err(); + assert_eq!(original.status_code(), Some(429)); + + let replay = recorder.fixture().replay_with_sanitizer(sanitizer()); + let replayed = replay.generate("HF-ALPHA-001").await.unwrap_err(); + assert_eq!(replayed.status_code(), Some(429)); + assert_eq!(replayed.request_id(), Some("[ACCOUNT]-request")); + replay.assert_finished().unwrap(); + } + + #[test] + fn incomplete_replay_reports_remaining_interactions() { + let fixture = Fixture { + schema_version: FIXTURE_SCHEMA_VERSION, + interactions: vec![StoredInteraction { + request: text_request(StoredOperation::Generate, "hello", &[]), + response: StoredResponse::Success { + body: "world".to_string(), + usage: None, + report: None, + }, + }], + }; + assert!(matches!( + fixture.replay().assert_finished(), + Err(FixtureError::ReplayIncomplete { remaining: 1 }) + )); + } +} diff --git a/src/backend/mod.rs b/src/backend/mod.rs index 63f1435..8c23a40 100644 --- a/src/backend/mod.rs +++ b/src/backend/mod.rs @@ -1,6 +1,8 @@ #[cfg(feature = "_client")] mod any_client; pub mod client; +#[cfg(feature = "mock")] +mod fixture; #[cfg(feature = "_client")] mod media; mod messages; @@ -36,6 +38,10 @@ pub mod openai; #[cfg(feature = "_client")] pub use any_client::{AnyClient, Provider}; pub use client::{LLMClient, MediaFile}; +#[cfg(feature = "mock")] +pub use fixture::{ + FIXTURE_SCHEMA_VERSION, Fixture, FixtureError, FixtureRecorder, FixtureSanitizer, ReplayClient, +}; pub use messages::{ChatMessage, ChatRole}; #[cfg(feature = "_client")] pub(crate) use messages::{MaterializeAttemptError, request_messages}; @@ -76,7 +82,7 @@ pub use usage::{ /// # Ok(()) /// # } /// ``` -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct ModelInfo { /// The model identifier used in API requests pub id: String, diff --git a/src/backend/usage.rs b/src/backend/usage.rs index fe61d6b..7faca59 100644 --- a/src/backend/usage.rs +++ b/src/backend/usage.rs @@ -1,6 +1,8 @@ use std::collections::BTreeMap; use std::fmt; +use serde::{Deserialize, Serialize}; + use crate::error::RStructorError; /// Token usage information from an LLM API call. @@ -26,7 +28,7 @@ use crate::error::RStructorError; /// # Ok(()) /// # } /// ``` -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct TokenUsage { /// The model used for this request pub model: String, @@ -81,7 +83,7 @@ impl TokenUsage { /// accounting if a provider reports different concrete model versions across /// retries. Keys use the response's model identifier when present and the /// configured model as a fallback. -#[derive(Debug, Clone, PartialEq, Eq, Default)] +#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] #[non_exhaustive] pub struct RunUsage { /// Number of attempts whose provider response included token usage. @@ -206,7 +208,7 @@ fn saturating_add(overflowed: &mut bool, left: u64, right: u64) -> u64 { /// Whether an attempt reached structured-output validation. #[non_exhaustive] -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum AttemptKind { /// A structured response reached decoding and custom validation. Semantic, @@ -216,7 +218,7 @@ pub enum AttemptKind { /// Why execution did or did not continue after a failed attempt. #[non_exhaustive] -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum RetryDisposition { /// Another provider attempt was made. Retried, @@ -228,7 +230,7 @@ pub enum RetryDisposition { /// Outcome of one materialization attempt. #[non_exhaustive] -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub enum AttemptOutcome { /// Structured output decoded and validated successfully. Succeeded, @@ -242,7 +244,7 @@ pub enum AttemptOutcome { } /// Immutable record of one materialization attempt. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[non_exhaustive] pub struct AttemptRecord { /// One-indexed attempt number. @@ -316,7 +318,7 @@ impl AttemptRecord { /// The same report shape is available on [`Extraction`] and /// [`ExtractionError`], so callers can inspect attempts and usage without /// maintaining separate success and failure accounting code. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[non_exhaustive] pub struct ExtractionReport { /// Usage from the final provider response, when it was reported. @@ -493,6 +495,23 @@ impl MaterializeReport { } } + #[cfg(feature = "mock")] + pub(crate) fn from_fixture_parts( + data: T, + final_usage: Option, + cumulative_usage: Option, + attempts: Vec, + attempts_complete: bool, + ) -> Self { + Self { + data, + final_usage, + cumulative_usage, + attempts, + attempts_complete, + } + } + /// Build a report from final-only metadata with unavailable attempt history. /// /// This is used by the default [`LLMClient`](crate::LLMClient) @@ -557,6 +576,21 @@ impl MaterializeFailure { } } + #[cfg(feature = "mock")] + pub(crate) fn from_fixture_parts( + error: RStructorError, + cumulative_usage: Option, + attempts: Vec, + attempts_complete: bool, + ) -> Self { + Self { + error: Box::new(error), + cumulative_usage, + attempts, + attempts_complete, + } + } + /// Create an empty-ledger failure when a client cannot expose attempt metadata. #[must_use] pub fn from_error(error: RStructorError) -> Self { diff --git a/src/diagnostics.rs b/src/diagnostics.rs index f0a1c80..1437a72 100644 --- a/src/diagnostics.rs +++ b/src/diagnostics.rs @@ -4,12 +4,14 @@ use std::collections::BTreeMap; #[cfg(feature = "_client")] use std::{fmt, sync::Arc}; +use serde::{Deserialize, Serialize}; + /// A response body that has passed through a caller-provided sanitizer. /// /// rstructor never stores provider response bodies by default. When capture is /// explicitly enabled, the sanitizer runs before the value is retained and the /// sanitized value is bounded by the configured byte limit. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[non_exhaustive] pub struct SanitizedResponseBody { /// Sanitized response text retained for diagnostics. @@ -23,7 +25,7 @@ pub struct SanitizedResponseBody { /// Status and recognized request-ID headers are captured for both successful /// and failed structured-output attempts. Response bodies remain absent unless /// the client was configured with opt-in response body capture. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[non_exhaustive] pub struct ResponseMetadata { /// Exact HTTP status code returned by the provider. diff --git a/src/error/mod.rs b/src/error/mod.rs index 2f83998..817bb61 100644 --- a/src/error/mod.rs +++ b/src/error/mod.rs @@ -1,3 +1,4 @@ +use serde::{Deserialize, Serialize}; use std::{fmt, time::Duration}; use thiserror::Error; @@ -38,7 +39,7 @@ use crate::ResponseMetadata; /// } /// } /// ``` -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub enum ApiErrorKind { /// Rate limit exceeded (HTTP 429) /// @@ -126,7 +127,7 @@ pub enum ApiErrorKind { /// provider's remaining response is malformed or truncated. Callers should stop /// consuming the stream when a [`RStructorError::StreamingError`] is yielded and /// decide whether to retain or roll back any earlier items. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[non_exhaustive] pub enum StreamErrorKind { /// An SSE event contained bytes that were not valid UTF-8. diff --git a/src/lib.rs b/src/lib.rs index 0fcb3cb..a585e24 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -100,7 +100,10 @@ pub use backend::{ pub use backend::{DEFAULT_CONNECT_TIMEOUT, DEFAULT_REQUEST_TIMEOUT}; #[cfg(feature = "tools")] pub use backend::{DynTool, FnTool, Tool, ToolRunner, Toolbox}; +#[cfg(feature = "mock")] +pub use backend::{ + FIXTURE_SCHEMA_VERSION, Fixture, FixtureError, FixtureRecorder, FixtureSanitizer, MockClient, + MockRequestView, MockResponse, RecordedRequest, ReplayClient, RequestKind, +}; #[cfg(feature = "streaming")] pub use backend::{ItemStream, ObjectStream, StreamedObject, TextStream}; -#[cfg(feature = "mock")] -pub use backend::{MockClient, MockRequestView, MockResponse, RecordedRequest, RequestKind}; diff --git a/tests/documentation_gallery_tests.rs b/tests/documentation_gallery_tests.rs index ab4f61a..0f17438 100644 --- a/tests/documentation_gallery_tests.rs +++ b/tests/documentation_gallery_tests.rs @@ -12,6 +12,7 @@ const GALLERY_EXAMPLES: &[(&str, &[&str])] = &[ ("kimi_k3_multimodal_example", &["derive", "openai"]), ("axum_handler_example", &["derive", "mock"]), ("mock_testing_example", &["derive", "mock"]), + ("fixture_record_replay", &["derive", "mock"]), ("ollama_local_example", &["derive", "openai"]), ("runtime_provider_example", &["derive", "openai"]), ("schemars_bridge_example", &["mock", "schemars"]), @@ -52,6 +53,10 @@ const COOKBOOK_RECIPES: &[(&str, &str)] = &[ "## Test extraction offline with `MockClient`", "mock_testing_example", ), + ( + "## Record and replay a sanitized fixture", + "fixture_record_replay", + ), ( "## Use a local model through Ollama", "ollama_local_example", diff --git a/tests/fixture_replay_tests.rs b/tests/fixture_replay_tests.rs new file mode 100644 index 0000000..7299ffd --- /dev/null +++ b/tests/fixture_replay_tests.rs @@ -0,0 +1,59 @@ +#![cfg(all(feature = "derive", feature = "mock"))] + +use rstructor::{Fixture, Instructor, LLMClient}; +use serde::{Deserialize, Serialize}; + +const APPLE_2023_PROMPT: &str = "From Apple Inc.'s 2023 Form 10-K: Total net sales were $383,285 million for the fiscal year ended September 30, 2023. Extract the issuer, fiscal year, and net sales in USD millions."; +const APPLE_2023_FIXTURE: &str = include_str!("fixtures/record_replay/apple_2023_10k.fixture.json"); + +#[derive(Debug, PartialEq, Instructor, Serialize, Deserialize)] +struct FilingMetric { + issuer: String, + fiscal_year: u16, + net_sales_usd_millions: u64, +} + +#[tokio::test] +async fn replays_real_world_filing_metric_with_usage_and_attempts() { + let fixture = Fixture::from_json(APPLE_2023_FIXTURE).unwrap(); + assert_eq!(fixture.to_json().unwrap(), APPLE_2023_FIXTURE); + + let replay = fixture.replay(); + let extraction = replay + .extract_with_report::(APPLE_2023_PROMPT) + .await + .unwrap(); + + assert_eq!( + extraction.data, + FilingMetric { + issuer: "Apple Inc.".to_string(), + fiscal_year: 2023, + net_sales_usd_millions: 383_285, + } + ); + assert_eq!(extraction.report.final_usage.unwrap().total_tokens(), 91); + assert_eq!(extraction.report.attempts.len(), 1); + assert!(extraction.report.attempts_complete); + replay.assert_finished().unwrap(); +} + +#[tokio::test] +async fn fixture_replay_rejects_schema_drift_without_consuming_the_exchange() { + #[derive(Debug, Instructor, Serialize, Deserialize)] + struct DriftedMetric { + issuer: String, + fiscal_year: u16, + net_sales_usd_millions: u64, + operating_income_usd_millions: u64, + } + + let replay = Fixture::from_json(APPLE_2023_FIXTURE).unwrap().replay(); + let error = replay + .extract_with_report::(APPLE_2023_PROMPT) + .await + .unwrap_err(); + + assert!(error.to_string().contains("schema name differs")); + assert_eq!(replay.remaining(), 1); +} diff --git a/tests/fixtures/record_replay/apple_2023_10k.fixture.json b/tests/fixtures/record_replay/apple_2023_10k.fixture.json new file mode 100644 index 0000000..03962f7 --- /dev/null +++ b/tests/fixtures/record_replay/apple_2023_10k.fixture.json @@ -0,0 +1,85 @@ +{ + "schema_version": 1, + "interactions": [ + { + "request": { + "operation": "materialize_with_attempts", + "prompt": "From Apple Inc.'s 2023 Form 10-K: Total net sales were $383,285 million for the fiscal year ended September 30, 2023. Extract the issuer, fiscal year, and net sales in USD millions.", + "schema_name": "FilingMetric", + "schema": { + "properties": { + "fiscal_year": { + "type": "integer" + }, + "issuer": { + "type": "string" + }, + "net_sales_usd_millions": { + "type": "integer" + } + }, + "required": [ + "issuer", + "fiscal_year", + "net_sales_usd_millions" + ], + "title": "FilingMetric", + "type": "object" + } + }, + "response": { + "type": "success", + "body": "{\"fiscal_year\":2023,\"issuer\":\"Apple Inc.\",\"net_sales_usd_millions\":383285}", + "usage": { + "model": "recorded-provider-model", + "input_tokens": 67, + "cached_input_tokens": 0, + "cache_write_input_tokens": 0, + "output_tokens": 24 + }, + "report": { + "final_usage": { + "model": "recorded-provider-model", + "input_tokens": 67, + "cached_input_tokens": 0, + "cache_write_input_tokens": 0, + "output_tokens": 24 + }, + "cumulative_usage": { + "reported_attempts": 1, + "input_tokens": 67, + "cached_input_tokens": 0, + "cache_write_input_tokens": 0, + "output_tokens": 24, + "by_model": { + "recorded-provider-model": { + "model": "recorded-provider-model", + "input_tokens": 67, + "cached_input_tokens": 0, + "cache_write_input_tokens": 0, + "output_tokens": 24 + } + }, + "overflowed": false + }, + "attempts": [ + { + "number": 1, + "kind": "Semantic", + "outcome": "Succeeded", + "usage": { + "model": "recorded-provider-model", + "input_tokens": 67, + "cached_input_tokens": 0, + "cache_write_input_tokens": 0, + "output_tokens": 24 + }, + "response": null + } + ], + "attempts_complete": true + } + } + } + ] +}