ferrum_cli/commands/
replay_bundle.rs

1//! Offline request replay bundle validation.
2
3use crate::config::CliConfig;
4use clap::Args;
5use ferrum_types::{FerrumError, ProfileEntrypoint, Result, OBSERVABILITY_PROFILE_SCHEMA_VERSION};
6use serde_json::{json, Value};
7use std::fs;
8use std::path::{Path, PathBuf};
9
10const REQUIRED_BUNDLE_FILES: &[&str] = &[
11    "request.json",
12    "prompt_token_ids.json",
13    "sampling_params.json",
14    "runtime_effective_config.json",
15    "backend_selection.json",
16    "output_token_ids.json",
17    "output_text.txt",
18    "bad_output_scan.json",
19    "replay.command.json",
20];
21
22#[derive(Args, Debug)]
23pub struct ReplayBundleCommand {
24    /// Request replay bundle directory.
25    pub bundle_dir: PathBuf,
26
27    /// Write an offline synthetic/no-weight replay artifact to this directory.
28    #[arg(long, value_name = "DIR")]
29    pub out: Option<PathBuf>,
30
31    /// Print a JSON summary instead of a PASS line.
32    #[arg(long)]
33    pub json: bool,
34}
35
36pub async fn execute(cmd: ReplayBundleCommand, _config: CliConfig) -> Result<()> {
37    let summary = replay_bundle(&cmd.bundle_dir, cmd.out.as_deref())?;
38    if cmd.json {
39        println!(
40            "{}",
41            serde_json::to_string_pretty(&summary)
42                .map_err(|err| FerrumError::serialization(err.to_string()))?
43        );
44    } else {
45        println!(
46            "FERRUM REPLAY BUNDLE PASS: {}",
47            cmd.bundle_dir.to_string_lossy()
48        );
49    }
50    Ok(())
51}
52
53fn replay_bundle(bundle_dir: &Path, out: Option<&Path>) -> Result<Value> {
54    let validated = validate_bundle(bundle_dir)?;
55    let generated = if let Some(out) = out {
56        Some(write_offline_replay_artifact(out, validated.entrypoint)?)
57    } else {
58        None
59    };
60    Ok(json!({
61        "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
62        "status": "pass",
63        "bundle_dir": bundle_dir.to_string_lossy(),
64        "request_id": validated.request_id,
65        "entrypoint": validated.entrypoint_label,
66        "model": validated.model,
67        "output_token_count": validated.output_token_count,
68        "bad_output": validated.bad_output,
69        "offline_replay_artifact": generated,
70        "pass_line": format!("FERRUM REPLAY BUNDLE PASS: {}", bundle_dir.to_string_lossy())
71    }))
72}
73
74#[derive(Debug)]
75struct ValidatedBundle {
76    request_id: String,
77    entrypoint: ProfileEntrypoint,
78    entrypoint_label: String,
79    model: String,
80    output_token_count: usize,
81    bad_output: bool,
82}
83
84fn validate_bundle(bundle_dir: &Path) -> Result<ValidatedBundle> {
85    if !bundle_dir.is_dir() {
86        return Err(FerrumError::invalid_parameter(format!(
87            "bundle dir does not exist or is not a directory: {}",
88            bundle_dir.display()
89        )));
90    }
91    for file in REQUIRED_BUNDLE_FILES {
92        let path = bundle_dir.join(file);
93        if !path.is_file() {
94            return Err(FerrumError::invalid_parameter(format!(
95                "missing replay bundle file: {}",
96                path.display()
97            )));
98        }
99    }
100
101    let request = read_json(&bundle_dir.join("request.json"))?;
102    let request_id = required_string(&request, "request_id", "request.json")?;
103    if request.get("schema_version").and_then(Value::as_u64)
104        != Some(OBSERVABILITY_PROFILE_SCHEMA_VERSION as u64)
105    {
106        return Err(FerrumError::invalid_parameter(
107            "request.json schema_version mismatch",
108        ));
109    }
110    if request.get("sanitized").and_then(Value::as_bool) != Some(true) {
111        return Err(FerrumError::invalid_parameter(
112            "request.json sanitized must be true",
113        ));
114    }
115    let entrypoint_label = required_string(&request, "entrypoint", "request.json")?;
116    let entrypoint = match entrypoint_label.as_str() {
117        "run" => ProfileEntrypoint::Run,
118        "serve" => ProfileEntrypoint::Serve,
119        other => {
120            return Err(FerrumError::invalid_parameter(format!(
121                "unsupported replay entrypoint: {other}"
122            )));
123        }
124    };
125    let model = request
126        .get("model")
127        .and_then(Value::as_str)
128        .unwrap_or("unknown")
129        .to_string();
130
131    for file in [
132        "prompt_token_ids.json",
133        "sampling_params.json",
134        "runtime_effective_config.json",
135        "backend_selection.json",
136        "output_token_ids.json",
137        "bad_output_scan.json",
138        "replay.command.json",
139    ] {
140        let value = read_json(&bundle_dir.join(file))?;
141        let other_id = required_string(&value, "request_id", file)?;
142        if other_id != request_id {
143            return Err(FerrumError::invalid_parameter(format!(
144                "{file} request_id mismatch: {other_id} != {request_id}"
145            )));
146        }
147    }
148
149    let output_tokens = read_json(&bundle_dir.join("output_token_ids.json"))?;
150    let output_token_count = validate_token_ids(&output_tokens, "output_token_ids.json")?;
151    let bad_scan = read_json(&bundle_dir.join("bad_output_scan.json"))?;
152    let bad_output = bad_scan
153        .get("bad_output")
154        .and_then(Value::as_bool)
155        .ok_or_else(|| FerrumError::invalid_parameter("bad_output_scan.bad_output missing"))?;
156    let replay = read_json(&bundle_dir.join("replay.command.json"))?;
157    let argv = replay
158        .get("argv")
159        .and_then(Value::as_array)
160        .ok_or_else(|| FerrumError::invalid_parameter("replay.command argv missing"))?;
161    if argv.is_empty() || !argv.iter().all(Value::is_string) {
162        return Err(FerrumError::invalid_parameter(
163            "replay.command argv must be a non-empty string array",
164        ));
165    }
166
167    Ok(ValidatedBundle {
168        request_id,
169        entrypoint,
170        entrypoint_label,
171        model,
172        output_token_count,
173        bad_output,
174    })
175}
176
177fn write_offline_replay_artifact(out: &Path, entrypoint: ProfileEntrypoint) -> Result<Value> {
178    let profile = out.join("profile.jsonl");
179    let memory = out.join("memory_profile.jsonl");
180    let scheduler = out.join("scheduler_trace.jsonl");
181    let request_dump = out.join("request_dump");
182    let config = crate::observability_product::ProductObservabilityConfig::new(
183        entrypoint,
184        "synthetic/no-weight",
185        Some(&profile),
186        crate::observability_product::ProfileDetailArg::Basic,
187        Some(&memory),
188        Some(&scheduler),
189        Some(&request_dump),
190        1.0,
191    );
192    let written = crate::observability_product::write_synthetic_product_observability(&config)?;
193    let summary = json!({
194        "out": out.to_string_lossy(),
195        "entrypoint": entrypoint.as_str(),
196        "artifact_count": written.len(),
197        "profile_jsonl": profile.to_string_lossy(),
198        "request_dump_dir": request_dump.to_string_lossy()
199    });
200    fs::create_dir_all(out).map_err(|err| FerrumError::io(err.to_string()))?;
201    fs::write(
202        out.join("replay_bundle_summary.json"),
203        serde_json::to_vec_pretty(&summary)
204            .map_err(|err| FerrumError::serialization(err.to_string()))?,
205    )
206    .map_err(|err| FerrumError::io(err.to_string()))?;
207    Ok(summary)
208}
209
210fn validate_token_ids(value: &Value, label: &str) -> Result<usize> {
211    let token_ids = value
212        .get("token_ids")
213        .and_then(Value::as_array)
214        .ok_or_else(|| FerrumError::invalid_parameter(format!("{label}.token_ids missing")))?;
215    if !token_ids
216        .iter()
217        .all(|item| item.as_u64().is_some_and(|token| token <= u32::MAX as u64))
218    {
219        return Err(FerrumError::invalid_parameter(format!(
220            "{label}.token_ids must be non-negative u32 values"
221        )));
222    }
223    let token_count = value
224        .get("token_count")
225        .and_then(Value::as_u64)
226        .ok_or_else(|| FerrumError::invalid_parameter(format!("{label}.token_count missing")))?
227        as usize;
228    if token_count != token_ids.len() {
229        return Err(FerrumError::invalid_parameter(format!(
230            "{label}.token_count must match token_ids length"
231        )));
232    }
233    Ok(token_count)
234}
235
236fn read_json(path: &Path) -> Result<Value> {
237    let body = fs::read_to_string(path)
238        .map_err(|err| FerrumError::io(format!("failed to read {}: {err}", path.display())))?;
239    serde_json::from_str(&body).map_err(|err| {
240        FerrumError::serialization(format!("failed to parse {}: {err}", path.display()))
241    })
242}
243
244fn required_string(value: &Value, key: &str, label: &str) -> Result<String> {
245    value
246        .get(key)
247        .and_then(Value::as_str)
248        .filter(|text| !text.trim().is_empty())
249        .map(ToString::to_string)
250        .ok_or_else(|| FerrumError::invalid_parameter(format!("{label}.{key} must be a string")))
251}
252
253#[cfg(test)]
254mod tests {
255    use super::*;
256    use uuid::Uuid;
257
258    #[test]
259    fn replay_bundle_validates_and_writes_offline_artifact() {
260        let root = std::env::temp_dir().join(format!("ferrum-replay-bundle-{}", Uuid::new_v4()));
261        let bundle = root.join("req-test");
262        write_test_bundle(&bundle, "run");
263        let out = root.join("offline");
264
265        let summary = replay_bundle(&bundle, Some(&out)).unwrap();
266
267        assert_eq!(summary["status"], "pass");
268        assert_eq!(summary["entrypoint"], "run");
269        assert!(out.join("profile.jsonl").is_file());
270        assert!(out.join("request_dump").is_dir());
271        fs::remove_dir_all(root).ok();
272    }
273
274    #[test]
275    fn replay_bundle_rejects_missing_required_file() {
276        let root = std::env::temp_dir().join(format!("ferrum-replay-bundle-{}", Uuid::new_v4()));
277        let bundle = root.join("req-test");
278        write_test_bundle(&bundle, "serve");
279        fs::remove_file(bundle.join("output_token_ids.json")).unwrap();
280
281        let err = validate_bundle(&bundle).expect_err("missing output tokens should fail");
282
283        assert!(err.to_string().contains("missing replay bundle file"));
284        fs::remove_dir_all(root).ok();
285    }
286
287    fn write_test_bundle(bundle: &Path, entrypoint: &str) {
288        fs::create_dir_all(bundle).unwrap();
289        let request_id = "req-test";
290        let common = json!({
291            "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
292            "request_id": request_id,
293            "sanitized": true
294        });
295        write_json(
296            &bundle.join("request.json"),
297            json!({
298                "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
299                "request_id": request_id,
300                "entrypoint": entrypoint,
301                "model": "synthetic/no-weight",
302                "backend": "synthetic",
303                "sanitized": true
304            }),
305        );
306        write_json(
307            &bundle.join("prompt_token_ids.json"),
308            json!({"schema_version": 1, "request_id": request_id, "token_ids": [1], "token_count": 1, "sanitized": true}),
309        );
310        write_json(
311            &bundle.join("sampling_params.json"),
312            json!({"schema_version": 1, "request_id": request_id, "sampling_params": {"max_tokens": 1}}),
313        );
314        write_json(
315            &bundle.join("runtime_effective_config.json"),
316            json!({"schema_version": 1, "request_id": request_id, "entrypoint": entrypoint, "sanitized": true}),
317        );
318        write_json(
319            &bundle.join("backend_selection.json"),
320            json!({"schema_version": 1, "request_id": request_id, "backend": "synthetic"}),
321        );
322        write_json(
323            &bundle.join("output_token_ids.json"),
324            json!({"schema_version": 1, "request_id": request_id, "token_ids": [2], "token_count": 1, "finish_reason": "stop"}),
325        );
326        write_json(
327            &bundle.join("bad_output_scan.json"),
328            json!({
329                "schema_version": 1,
330                "request_id": request_id,
331                "bad_output": false,
332                "reasons": [],
333                "classified_output_sha256": "2689367b205c16ce32ed4200942b8b8b1e262dfc70d9bc9fbc77c49699a4f1df",
334                "output_sha256": "dc51b8c96c2d745df3bd5590d990230a482fd247123599548e0632fdbf97fc22"
335            }),
336        );
337        write_json(
338            &bundle.join("replay.command.json"),
339            json!({"schema_version": 1, "request_id": request_id, "entrypoint": entrypoint, "argv": ["ferrum", "replay-bundle", bundle], "command": "ferrum replay-bundle"}),
340        );
341        fs::write(bundle.join("output_text.txt"), "ok\n").unwrap();
342        let _ = common;
343    }
344
345    fn write_json(path: &Path, value: Value) {
346        fs::write(path, serde_json::to_vec_pretty(&value).unwrap()).unwrap();
347    }
348}
ferrum_cli/commands/replay_bundle.rs

ferrum_cli/commands/
replay_bundle.rs
