use super::{
    History, MAX_BYTES, MAX_CYCLES, MAX_GROUPS, Message, QUEUE, Recorder, add, apply, read, save,
};
use serde_json::json;
use std::sync::{
    Arc, Mutex,
    atomic::{AtomicU64, Ordering},
    mpsc,
};
use tracing_subscriber::Layer;
use tracing_subscriber::layer::SubscriberExt;

#[test]
fn history_roundtrip_groups_observations_and_bounds_restarts() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("support.json");
    let mut h = History {
        schema_version: 1,
        ..History::default()
    };
    for n in 0..(MAX_CYCLES + 4) {
        apply(
            &mut h,
            Message::Begin(json!({"threads": 2, "username": "private-username"})),
        );
        for _ in 0..3 {
            add(&mut h, "exact_lookup_rejection_v1", json!({"rejected_requests": n, "expected_owner": "absent", "stage": "master_reference_scope", "reason": "owner_unqualified"}).as_object().unwrap().clone());
        }
        apply(
            &mut h,
            Message::Complete(
                json!({"bytes_downloaded": 604, "disk_bytes_written": 17}),
                "partial_failure",
            ),
        );
    }
    save(&path, &mut h).unwrap();
    let (reopened, status) = read(&path);
    assert_eq!(status, "available");
    assert_eq!(reopened.cycles.len(), MAX_CYCLES);
    assert_eq!(reopened.cycles_evicted, 4);
    assert_eq!(
        reopened.cycles.last().unwrap().diagnostics[0].observations,
        3
    );
    assert_eq!(
        reopened.cycles.last().unwrap().stats["disk_bytes_written"],
        17
    );
    assert!(std::fs::metadata(&path).unwrap().len() <= MAX_BYTES);
    assert!(
        !serde_json::to_string(&reopened)
            .unwrap()
            .contains("private-username")
    );
}

#[test]
fn interrupted_staging_keeps_previous_snapshot_and_is_bounded_on_restart() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("support.json");
    let mut h = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut h, Message::Begin(json!({"threads": 2})));
    save(&path, &mut h).unwrap();
    std::fs::write(path.with_extension("support-tmp"), br#"{"half-written":1"#).unwrap();
    let (mut reopened, status) = read(&path);
    assert_eq!(status, "available");
    assert!(reopened.cycles[0].completed_at.is_none());
    apply(&mut reopened, Message::Begin(json!({"threads": 4})));
    save(&path, &mut reopened).unwrap();
    assert_eq!(read(&path).0.cycles.len(), 2);
    assert!(!path.with_extension("support-tmp").exists());
}

#[test]
fn groups_and_queue_have_explicit_omission_counts() {
    let mut h = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut h, Message::Begin(json!({})));
    for n in 0..(MAX_GROUPS + 5) {
        add(
            &mut h,
            "metadata_capture_ambiguity_counts_v1",
            json!({"matching_children": n}).as_object().unwrap().clone(),
        );
    }
    assert_eq!(h.cycles[0].diagnostics.len(), MAX_GROUPS);
    assert_eq!(h.groups_omitted, 5);
    let (sender, _receiver) = mpsc::sync_channel(QUEUE);
    let recorder = Recorder {
        sender,
        generation: Arc::new(AtomicU64::new(0)),
        dropped: Arc::new(AtomicU64::new(0)),
        aliases: Arc::new(Mutex::new(std::collections::HashMap::new())),
        alias_namespace: uuid::Uuid::new_v4(),
    };
    for _ in 0..(QUEUE + 5) {
        recorder.send(Message::Observe(
            "sparse_identity_state_failed",
            Default::default(),
        ));
    }
    assert_eq!(recorder.dropped.load(Ordering::Relaxed), 5);
}

#[test]
fn diagnostic_layer_runs_below_log_filter_and_ignores_private_fields() {
    let (sender, receiver) = mpsc::sync_channel(QUEUE);
    let recorder = Recorder {
        sender,
        generation: Arc::new(AtomicU64::new(0)),
        dropped: Arc::new(AtomicU64::new(0)),
        aliases: Arc::new(Mutex::new(std::collections::HashMap::new())),
        alias_namespace: uuid::Uuid::new_v4(),
    };
    let subscriber = tracing_subscriber::registry().with(recorder).with(
        tracing_subscriber::fmt::layer()
            .with_writer(std::io::sink)
            .with_filter(tracing_subscriber::filter::LevelFilter::ERROR),
    );
    tracing::subscriber::with_default(subscriber, || {
        tracing::debug!(
            diagnostic = "exact_lookup_rejection_v1",
            target = "paired",
            stage = "master_reference_scope",
            reason = "owner_unqualified",
            expected_owner = "absent",
            lookup_zone = "primary",
            rejected_requests = 604u64,
            asset_id = "private-id",
            path = "/private/path",
            error = "private-password",
            token = "private-token",
            "private free text"
        );
        tracing::error!(
            diagnostic = "private-new-diagnostic",
            error = "private-body",
            "private message"
        );
    });
    let Message::Tagged(_, message) = receiver.try_recv().unwrap() else {
        panic!("expected attribution envelope")
    };
    let Message::Event(kind, fields) = *message else {
        panic!("expected diagnostic")
    };
    assert_eq!(kind, "exact_lookup_rejection_v1");
    assert_eq!(fields["rejected_requests"], 604);
    assert_eq!(fields.len(), 6);
    assert!(!serde_json::to_string(&fields).unwrap().contains("private"));
    assert!(receiver.try_recv().is_err());
}

#[test]
fn untrusted_saved_strings_and_unknown_schema_never_escape() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("support.json");
    let mut h = History {
        schema_version: 1,
        ..History::default()
    };
    apply(
        &mut h,
        Message::Begin(json!({"username": "private-id", "resolution": "https://private-url"})),
    );
    let cycle = h.cycles.last_mut().unwrap();
    cycle.build_version = "0.private-token".into();
    cycle.outcome = "private-error".into();
    cycle.stats = json!({"inventory_drop_library": "private-album", "bytes_downloaded": 1});
    add(&mut h, "exact_lookup_rejection_v1", json!({"reason": "private-error", "target": "private-id", "rejected_requests": 4, "path": "/private/path"}).as_object().unwrap().clone());
    std::fs::write(&path, serde_json::to_vec(&h).unwrap()).unwrap();
    let (clean, status) = read(&path);
    assert_eq!(status, "available");
    assert!(!serde_json::to_string(&clean).unwrap().contains("private"));
    std::fs::write(&path, b"{\"schema_version\":999,\"cycles\":[],\"cycles_evicted\":0,\"queue_dropped\":0,\"groups_omitted\":0,\"previous_history_unavailable\":false}").unwrap();
    assert_eq!(read(&path).1, "unsupported_schema");
    std::fs::write(&path, b"{incomplete").unwrap();
    assert_eq!(read(&path).1, "invalid_or_interrupted");
    std::fs::write(&path, vec![b'x'; usize::try_from(MAX_BYTES + 1).unwrap()]).unwrap();
    assert_eq!(read(&path).1, "truncated_or_unreadable");
}

#[tokio::test]
async fn normal_worker_flushes_restart_history() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("support.json");
    let (recorder, guard) = super::start(path.clone(), json!({"threads": 2})).unwrap();
    recorder
        .sender
        .send(Message::Event(
            "sparse_deletion_validation_failed".into(),
            Default::default(),
        ))
        .unwrap();
    guard.finish().await;
    let (recorder, guard) = super::start(path.clone(), json!({"threads": 4})).unwrap();
    recorder
        .sender
        .send(Message::Complete(json!({"failed": 1}), "partial_failure"))
        .unwrap();
    guard.finish().await;
    let (history, status) = read(&path);
    assert_eq!(status, "available");
    assert_eq!(history.cycles.len(), 2);
    assert_eq!(history.cycles[1].stats["failed"], 1);
    assert_eq!(
        history.cycles[0].diagnostics[0].kind,
        "sparse_deletion_validation_failed"
    );
}

#[test]
fn serialized_byte_limit_rotates_complete_records() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("support.json");
    let mut history = History {
        schema_version: 1,
        ..History::default()
    };
    for _ in 0..MAX_CYCLES {
        apply(&mut history, Message::Begin(json!({})));
        // Fill fixed scalar fields with long numeric encodings, not private text.
        for n in 0..MAX_GROUPS {
            let fields = json!({"pages":n,"records":u64::MAX,"transferred_bytes":u64::MAX,"retained_bytes":u64::MAX,
                "page_budget":u64::MAX,"record_budget":u64::MAX,"page_byte_budget":u64::MAX,"retained_byte_budget":u64::MAX,
                "elapsed_secs":1.7976931348623157e308,"reason":"invalid_inventory_evidence","phase":"family_scan","subreason":"child_reference_scope_mismatch",
                "eof_observed":true,"family_context":"unrelated","scope_mismatch_component":"owner","child_soft_deleted":false,
                "expected_owner":"private_default","expected_owner_type":"string","supplied_owner":"other_string","supplied_owner_type":"string"});
            add(
                &mut history,
                "legacy_inventory_failure_v2",
                fields.as_object().unwrap().clone(),
            );
        }
    }
    assert!(serde_json::to_vec(&history).unwrap().len() as u64 > MAX_BYTES);
    save(&path, &mut history).unwrap();
    assert!(std::fs::metadata(&path).unwrap().len() <= MAX_BYTES);
    assert!(history.cycles_evicted > 0);
    let (reopened, status) = read(&path);
    assert_eq!(status, "available");
    assert_eq!(reopened.cycles_evicted, history.cycles_evicted);
}

#[cfg(unix)]
#[test]
fn staging_symlink_cannot_modify_another_file() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("support.json");
    let target = dir.path().join("media");
    std::fs::write(&target, b"private media").unwrap();
    std::os::unix::fs::symlink(&target, path.with_extension("support-tmp")).unwrap();
    let mut history = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut history, Message::Begin(json!({})));
    assert!(save(&path, &mut history).is_err());
    assert_eq!(std::fs::read(target).unwrap(), b"private media");
}

#[tokio::test]
async fn existing_writer_lock_preserves_prior_history() {
    use fs4::fs_std::FileExt;
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("support.json");
    let mut history = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut history, Message::Begin(json!({})));
    save(&path, &mut history).unwrap();
    let before = std::fs::read(&path).unwrap();
    let lock = super::private_options()
        .create(true)
        .truncate(false)
        .open(path.with_extension("support-lock"))
        .unwrap();
    lock.lock_exclusive().unwrap();
    let (_, guard) = super::start(path.clone(), json!({"threads":42})).unwrap();
    guard.finish().await;
    assert_eq!(std::fs::read(path).unwrap(), before);
}

#[test]
fn restarting_does_not_complete_an_interrupted_prior_startup() {
    let mut history = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut history, Message::Start(json!({})));
    add(
        &mut history,
        "support_startup_v1",
        json!({"phase":"starting"}).as_object().unwrap().clone(),
    );
    apply(&mut history, Message::Start(json!({})));
    assert!(history.cycles[0].completed_at.is_none());
    assert_eq!(history.cycles[0].outcome, "running");
    add(
        &mut history,
        "support_startup_v1",
        json!({"phase":"starting"}).as_object().unwrap().clone(),
    );
    apply(&mut history, Message::Begin(json!({})));
    assert!(history.cycles[0].completed_at.is_none());
    assert_eq!(history.cycles[1].outcome, "success");
    assert!(history.cycles[1].completed_at.is_some());
}

#[test]
fn dropped_cycle_handoff_cannot_overwrite_prior_completed_evidence() {
    let (sender, receiver) = mpsc::sync_channel(1);
    let recorder = Recorder {
        sender,
        generation: Arc::new(AtomicU64::new(0)),
        dropped: Arc::new(AtomicU64::new(0)),
        aliases: Arc::new(Mutex::new(Default::default())),
        alias_namespace: uuid::Uuid::new_v4(),
    };
    let mut history = History {
        schema_version: 1,
        ..History::default()
    };
    recorder.send(Message::Begin(json!({"threads":2})));
    apply(&mut history, receiver.recv().unwrap());
    recorder.send(Message::Complete(json!({"downloaded":17}), "success"));
    apply(&mut history, receiver.recv().unwrap());
    let prior = serde_json::to_value(&history.cycles[0]).unwrap();
    // A stalled writer has a full observation queue at the next cycle boundary.
    recorder.send(Message::Observe(
        "sparse_identity_state_failed",
        Default::default(),
    ));
    recorder.send(Message::Begin(json!({"threads":4})));
    assert_eq!(recorder.dropped.load(Ordering::Relaxed), 1);
    apply(&mut history, receiver.recv().unwrap());
    recorder.send(Message::Complete(
        json!({"downloaded":604}),
        "partial_failure",
    ));
    apply(&mut history, receiver.recv().unwrap());
    assert_eq!(
        serde_json::to_value(&history.cycles[0]).unwrap()["stats"],
        prior["stats"]
    );
    assert_eq!(
        history.cycles[0].completed_at,
        prior["completed_at"].as_str().map(str::to_owned)
    );
    assert_eq!(history.cycles[0].outcome, "success");
    assert_eq!(history.cycles.len(), 1);
    assert_eq!(history.queue_dropped, 1);
    // A later accepted handoff resumes accurate collection with a separate record.
    recorder.send(Message::Begin(json!({"threads":8})));
    apply(&mut history, receiver.recv().unwrap());
    recorder.send(Message::Complete(json!({"downloaded":2}), "success"));
    apply(&mut history, receiver.recv().unwrap());
    assert_eq!(history.cycles.len(), 2);
    assert_eq!(history.cycles[1].stats["downloaded"], 2);
}

#[cfg(unix)]
#[test]
fn failed_history_rename_never_uses_an_unowned_copy_fallback() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("support.json");
    std::fs::create_dir(&path).unwrap();
    let media = dir.path().join("media.jpg");
    std::fs::write(&media, b"private media").unwrap();
    let sibling = path.with_extension(format!("json.kei-xdev-tmp-{}", std::process::id()));
    std::os::unix::fs::symlink(&media, &sibling).unwrap();
    let mut history = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut history, Message::Start(json!({})));
    assert!(save(&path, &mut history).is_err());
    assert_eq!(std::fs::read(media).unwrap(), b"private media");
    assert!(
        std::fs::symlink_metadata(sibling)
            .unwrap()
            .file_type()
            .is_symlink()
    );
    assert!(path.is_dir());
}

#[test]
fn late_failures_and_summary_survive_sampled_details_and_report_counts() {
    let mut h = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut h, Message::Begin(json!({})));
    for n in 0..200 {
        let fields = json!({"phase":"scan","reason":"candidate","scan_index":n});
        for _ in 0..2 {
            add(
                &mut h,
                "support_import_v1",
                fields.as_object().unwrap().clone(),
            );
        }
    }
    add(
        &mut h,
        "support_import_v1",
        json!({"phase":"adoption","reason":"strict_refusal"})
            .as_object()
            .unwrap()
            .clone(),
    );
    add(
        &mut h,
        "support_import_v1",
        json!({"phase":"complete","total":200,"strict_refused":1})
            .as_object()
            .unwrap()
            .clone(),
    );
    let c = &h.cycles[0];
    assert_eq!(c.diagnostics.len(), super::MAX_DETAIL_GROUPS + 2);
    assert_eq!(c.observations_total, Some(402));
    assert_eq!(c.observations_shown, Some(130));
    assert_eq!(c.observations_omitted, Some(272));
    assert!(
        c.diagnostics
            .iter()
            .any(|d| d.fields.get("strict_refused") == Some(&json!(1)))
    );
    let tmp = tempfile::tempdir().unwrap();
    let path = tmp.path().join("support.json");
    save(&path, &mut h).unwrap();
    let (clean, status) = read(&path);
    assert_eq!(status, "available");
    assert_eq!(clean.cycles[0].observations_omitted, Some(272));
    // Exhaust failure groups too: late summary displaces an earlier failure.
    for n in 0..MAX_GROUPS {
        add(
            &mut h,
            "exact_lookup_rejection_v1",
            json!({"rejected_requests":n}).as_object().unwrap().clone(),
        );
    }
    add(
        &mut h,
        "support_import_v1",
        json!({"phase":"complete","total":201,"strict_refused":2})
            .as_object()
            .unwrap()
            .clone(),
    );
    let c = &h.cycles[0];
    assert_eq!(c.diagnostics.len(), MAX_GROUPS);
    assert_eq!(
        c.observations_total,
        c.observations_shown
            .zip(c.observations_omitted)
            .map(|(a, b)| a + b)
    );
    assert!(
        c.diagnostics
            .iter()
            .any(|d| d.fields.get("total") == Some(&json!(201)))
    );
}

#[test]
fn older_history_observation_accounting_is_explicitly_unavailable() {
    let mut h = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut h, Message::Begin(json!({})));
    let mut saved = serde_json::to_value(&h).unwrap();
    let c = saved["cycles"][0].as_object_mut().unwrap();
    for field in [
        "observations_total",
        "observations_shown",
        "observations_omitted",
    ] {
        c.remove(field);
    }
    let tmp = tempfile::tempdir().unwrap();
    let path = tmp.path().join("support.json");
    std::fs::write(&path, serde_json::to_vec(&saved).unwrap()).unwrap();
    let (read, status) = read(&path);
    assert_eq!(status, "available");
    assert!(read.cycles[0].observations_total.is_none());
    assert!(read.cycles[0].observations_shown.is_none());
    assert!(read.cycles[0].observations_omitted.is_none());
}

#[test]
fn issue956_pair_reference_reasons_survive_history_reopen_without_private_fields() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("history.json");
    let mut history = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut history, Message::Begin(json!({})));
    let reasons = [
        "master_reference_missing",
        "master_reference_malformed",
        "child_master_mismatch",
        "record_provider_error",
    ];
    for reason in reasons {
        add(
            &mut history,
            "exact_lookup_rejection_v1",
            json!({
                "reason":reason,"target":"paired","stage":"pairing","rejected_requests":1,
                "recordName":"PRIVATE_RECORD_CANARY"
            })
            .as_object()
            .unwrap()
            .clone(),
        );
    }
    save(&path, &mut history).unwrap();
    let (reopened, status) = read(&path);
    assert_eq!(status, "available");
    let saved: Vec<_> = reopened.cycles[0]
        .diagnostics
        .iter()
        .map(|observation| observation.fields["reason"].as_str().unwrap())
        .collect();
    assert_eq!(saved, reasons);
    assert!(
        !serde_json::to_string(&reopened)
            .unwrap()
            .contains("PRIVATE_RECORD_CANARY")
    );
}

#[test]
fn missing_reference_history_survives_restart_and_revalidates_export() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("support.json");
    let mut history = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut history, Message::Begin(json!({})));
    let fields = json!({"record_history":"replacement", "prior_reference":"unavailable",
        "sparse_marker":"not_recorded", "linkage":"not_recorded", "lookup_outcome":"scope_unavailable", "lookup_family":"not_established", "lookup_scope":"child_unavailable", "lookup_shape":"provider_error"});
    add(
        &mut history,
        "support_missing_reference_v1",
        fields.as_object().unwrap().clone(),
    );
    apply(
        &mut history,
        Message::Complete(json!({}), "partial_failure"),
    );
    save(&path, &mut history).unwrap();
    let (reopened, status) = read(&path);
    assert_eq!(status, "available");
    let diagnostic = &reopened.cycles[0].diagnostics[0];
    assert_eq!(diagnostic.kind, "support_missing_reference_v1");
    assert_eq!(serde_json::Value::Object(diagnostic.fields.clone()), fields);
    let mut poisoned: serde_json::Value =
        serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
    for key in ["lookup_family", "lookup_scope", "lookup_shape"] {
        poisoned["cycles"][0]["diagnostics"][0]["fields"][key] = json!("private-provider-id");
    }
    std::fs::write(&path, serde_json::to_vec(&poisoned).unwrap()).unwrap();
    assert!(
        !serde_json::to_string(&read(&path).0)
            .unwrap()
            .contains("private-provider-id")
    );
}

#[test]
fn legacy_receipt_prerequisites_roundtrip_only_fixed_labels() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("support.json");
    let mut history = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut history, Message::Begin(json!({})));
    let labels = [
        "child_not_ready",
        "rendition_missing",
        "path_receipt_missing",
        "receipt_binding_mismatch",
        "source_receipt_missing",
        "source_receipt_invalid",
        "receipt_state_changed",
        "diagnostic_unavailable",
    ];
    for label in labels {
        add(&mut history, "legacy_receipt_prerequisite_v1", json!({"prerequisite":label,"id":"private-id","path":"/private/path","checksum":"private-checksum","error":"private-error"}).as_object().unwrap().clone());
    }
    add(
        &mut history,
        "legacy_receipt_prerequisite_v1",
        json!({"prerequisite":"private-new-label"})
            .as_object()
            .unwrap()
            .clone(),
    );
    save(&path, &mut history).unwrap();
    let (reopened, status) = read(&path);
    assert_eq!(status, "available");
    let observed: Vec<_> = reopened.cycles[0]
        .diagnostics
        .iter()
        .filter_map(|d| {
            d.fields
                .get("prerequisite")
                .and_then(serde_json::Value::as_str)
        })
        .collect();
    assert_eq!(observed, labels);
    assert!(
        !serde_json::to_string(&reopened)
            .unwrap()
            .contains("private")
    );
}

#[test]
fn issue956_lookup_record_statuses_survive_reopen_and_reject_untrusted_values() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("history.json");
    let mut history = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut history, Message::Begin(json!({})));
    for (master, child) in [("not_found", "returned"), ("returned", "access_denied")] {
        add(
            &mut history,
            "exact_lookup_rejection_v1",
            json!({
                "target":"paired","stage":"record_error","reason":"record_provider_error",
                "master_status":master,"child_status":child,"rejected_requests":3,
                "serverErrorCode":"PRIVATE_ERROR_CANARY","recordName":"PRIVATE_RECORD_CANARY"
            })
            .as_object()
            .unwrap()
            .clone(),
        );
    }
    save(&path, &mut history).unwrap();
    let (reopened, status) = read(&path);
    assert_eq!(status, "available");
    let events = &reopened.cycles[0].diagnostics;
    assert_eq!(events.len(), 2);
    assert_eq!(events[0].fields["master_status"], "not_found");
    assert_eq!(events[0].fields["child_status"], "returned");
    assert_eq!(events[1].fields["master_status"], "returned");
    assert_eq!(events[1].fields["child_status"], "access_denied");
    assert!(
        !serde_json::to_string(&reopened)
            .unwrap()
            .contains("PRIVATE_")
    );

    let mut untrusted = serde_json::to_value(&reopened).unwrap();
    untrusted["cycles"][0]["diagnostics"][0]["fields"]["master_status"] =
        json!("PRIVATE_STATUS_CANARY");
    untrusted["cycles"][0]["diagnostics"][0]["fields"]["child_status"] = json!(401);
    std::fs::write(&path, serde_json::to_vec(&untrusted).unwrap()).unwrap();
    let (clean, status) = read(&path);
    assert_eq!(status, "available");
    assert!(
        !clean.cycles[0].diagnostics[0]
            .fields
            .contains_key("master_status")
    );
    assert!(
        !clean.cycles[0].diagnostics[0]
            .fields
            .contains_key("child_status")
    );
    assert_eq!(
        clean.cycles[0].diagnostics[1].fields["child_status"],
        "access_denied"
    );
    assert!(!serde_json::to_string(&clean).unwrap().contains("PRIVATE_"));
}

#[test]
fn issue956_lookup_statuses_cross_the_production_tracing_recorder_and_reopen() {
    let statuses = [
        "not_requested",
        "omitted",
        "returned",
        "tombstone",
        "not_found",
        "access_denied",
        "authentication_failed",
        "zone_not_found",
        "retry_later",
        "throttled",
        "server_error",
        "other_error",
        "malformed_error",
    ];
    let _parallel_dispatch = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
    let (recorder, drain) = super::test_layer();
    let subscriber = tracing_subscriber::registry().with(recorder);
    tracing::subscriber::with_default(subscriber, || {
        for status in statuses {
            tracing::info!(
                diagnostic = "exact_lookup_rejection_v1",
                target = "paired",
                master_status = status,
                child_status = status,
                rejected_requests = 1u64,
                record_name = "PRIVATE_RECORD_CANARY",
                error = "PRIVATE_ERROR_CANARY",
                "PRIVATE_MESSAGE_CANARY"
            );
        }
        tracing::info!(
            diagnostic = "exact_lookup_rejection_v1",
            master_status = "PRIVATE_STATUS_CANARY",
            child_status = 401u64,
            rejected_requests = 2u64,
            "untrusted categories"
        );
    });
    let events = drain();
    assert_eq!(events.len(), statuses.len() + 1);
    for (event, status) in events.iter().zip(statuses) {
        assert_eq!(event["fields"]["master_status"], status);
        assert_eq!(event["fields"]["child_status"], status);
    }
    let last = events.last().unwrap()["fields"].as_object().unwrap();
    assert_eq!(last.len(), 1);
    assert_eq!(last["rejected_requests"], 2);
    assert!(!serde_json::to_string(&events).unwrap().contains("PRIVATE_"));

    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("history.json");
    let mut history = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut history, Message::Begin(json!({})));
    for event in events {
        add(
            &mut history,
            event["kind"].as_str().unwrap(),
            event["fields"].as_object().unwrap().clone(),
        );
    }
    save(&path, &mut history).unwrap();
    let (reopened, status) = read(&path);
    assert_eq!(status, "available");
    for (event, status) in reopened.cycles[0].diagnostics.iter().zip(statuses) {
        assert_eq!(event.fields["master_status"], status);
        assert_eq!(event.fields["child_status"], status);
    }
}

#[test]
fn issue956_child_reference_status_crosses_tracing_and_reopen_without_private_values() {
    use tracing_subscriber::layer::SubscriberExt;
    let labels = [
        "not_requested",
        "omitted",
        "unavailable",
        "invalid_child",
        "missing",
        "malformed",
        "sparse",
        "matches_requested",
        "different",
    ];
    let _parallel_dispatch = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
    let (recorder, drain) = super::test_layer();
    tracing::subscriber::with_default(tracing_subscriber::registry().with(recorder), || {
        for label in labels {
            tracing::info!(
                diagnostic = "exact_lookup_rejection_v1",
                child_reference_status = label,
                record_name = "PRIVATE_RECORD_CANARY",
                "PRIVATE_MESSAGE_CANARY"
            );
        }
    });
    let events = drain();
    assert_eq!(events.len(), labels.len());
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("history.json");
    let mut history = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut history, Message::Begin(json!({})));
    for (event, label) in events.iter().zip(labels) {
        assert_eq!(event["fields"]["child_reference_status"], label);
        add(
            &mut history,
            "exact_lookup_rejection_v1",
            event["fields"].as_object().unwrap().clone(),
        );
    }
    save(&path, &mut history).unwrap();
    let (reopened, status) = read(&path);
    assert_eq!(status, "available");
    assert_eq!(reopened.cycles[0].diagnostics.len(), labels.len());
    for (event, label) in reopened.cycles[0].diagnostics.iter().zip(labels) {
        assert_eq!(event.fields["child_reference_status"], label);
    }
    assert!(
        !serde_json::to_string(&reopened)
            .unwrap()
            .contains("PRIVATE_")
    );
    let mut untrusted = serde_json::to_value(&reopened).unwrap();
    for (index, invalid) in [json!("PRIVATE_REFERENCE_CANARY"), json!(42)]
        .into_iter()
        .enumerate()
    {
        untrusted["cycles"][0]["diagnostics"][index]["fields"]["child_reference_status"] = invalid;
    }
    std::fs::write(&path, serde_json::to_vec(&untrusted).unwrap()).unwrap();
    let (clean, status) = read(&path);
    assert_eq!(status, "available");
    assert!(
        clean.cycles[0].diagnostics[..2]
            .iter()
            .all(|event| !event.fields.contains_key("child_reference_status"))
    );
    assert_eq!(
        clean.cycles[0].diagnostics[2].fields["child_reference_status"],
        labels[2]
    );
    assert!(!serde_json::to_string(&clean).unwrap().contains("PRIVATE_"));
}

#[test]
fn storage_reuse_counters_saturate_and_preserve_last_stop() {
    let mut reuse = super::StorageReuse::default();
    reuse.inbox.new = u64::MAX;
    reuse.inbox.exact = u64::MAX;
    reuse.inbox.equivalent = u64::MAX;
    reuse.inbox.capacity_rejected = u64::MAX;
    reuse.inbox.single_item_limit = u64::MAX;
    reuse.record(
        super::StorageComponent::Inbox,
        super::StorageOutcome::New,
        Some((7, 11)),
    );
    reuse.record(
        super::StorageComponent::Inbox,
        super::StorageOutcome::Exact,
        None,
    );
    reuse.record(
        super::StorageComponent::Inbox,
        super::StorageOutcome::Equivalent,
        None,
    );
    reuse.record(
        super::StorageComponent::Inbox,
        super::StorageOutcome::CapacityRejected,
        Some((8, 11)),
    );
    assert!(reuse.inbox.observed);
    assert!(!reuse.catalogue.observed);
    assert_eq!(reuse.inbox.new, u64::MAX);
    assert_eq!(reuse.inbox.exact, u64::MAX);
    assert_eq!(reuse.inbox.equivalent, u64::MAX);
    assert_eq!(reuse.inbox.capacity_rejected, u64::MAX);
    assert_eq!(reuse.inbox.last_reason, super::ReuseReason::Capacity);
    assert_eq!(reuse.inbox.last_observed_used, Some(8));
    assert_eq!(reuse.inbox.last_observed_limit, Some(11));
    reuse.record(
        super::StorageComponent::Selection,
        super::StorageOutcome::SingleItemLimit,
        None,
    );
    assert_eq!(reuse.selection.single_item_limit, 1);
    assert_eq!(
        reuse.selection.last_reason,
        super::ReuseReason::SingleItemLimit
    );
    assert_eq!(reuse.selection.last_observed_used, None);
    reuse.record(
        super::StorageComponent::Selection,
        super::StorageOutcome::Exact,
        Some((3, 9)),
    );
    assert_eq!(
        reuse.selection.last_reason,
        super::ReuseReason::SingleItemLimit
    );
    assert_eq!(reuse.selection.last_observed_used, Some(3));
    assert_eq!(reuse.selection.last_observed_limit, Some(9));
}

#[test]
fn migration_evidence_replaces_counts_and_budgets_with_the_last_result() {
    let mut history = History {
        schema_version: 1,
        ..History::default()
    };
    apply(&mut history, Message::Begin(json!({})));
    super::apply_migration(
        &mut history,
        super::MigrationEvent::Committed(super::CompactionSummary {
            examined_raw_pages: 10,
            logical_charge_before: Some(100),
            logical_charge_after: Some(40),
            ..Default::default()
        }),
    );
    super::apply_migration(
        &mut history,
        super::MigrationEvent::Committed(super::CompactionSummary {
            examined_raw_pages: 2,
            logical_charge_before: Some(9),
            logical_charge_after: Some(4),
            ..Default::default()
        }),
    );
    let evidence = &history.cycles[0].transport_compaction;
    assert_eq!(evidence.committed, 2);
    assert_eq!(evidence.examined_raw_pages, 2);
    assert_eq!(evidence.logical_charge_before, Some(9));
    assert_eq!(evidence.logical_charge_after, Some(4));
    assert_eq!(evidence.committed_reclaim, Some(5));
    super::apply_migration(
        &mut history,
        super::MigrationEvent::Failed(super::MigrationFailure::MigrationFailed),
    );
    let evidence = &history.cycles[0].transport_compaction;
    assert_eq!(evidence.failed, 1);
    assert_eq!(evidence.reason, super::CompactionReason::MigrationFailed);
    assert_eq!(evidence.examined_raw_pages, 0);
    assert_eq!(evidence.logical_charge_before, None);
    assert_eq!(evidence.logical_charge_after, None);
    assert_eq!(evidence.committed_reclaim, Some(0));
    assert_eq!(evidence.committed, 2);
}

#[test]
fn older_storage_and_migration_fields_default_without_rejecting_history() {
    let value = json!({
        "schema_version": 1,
        "cycles_evicted": 0,
        "queue_dropped": 0,
        "groups_omitted": 0,
        "previous_history_unavailable": false,
        "cycles": [{
            "id": uuid::Uuid::new_v4().to_string(),
            "started_at": "2026-10-22T00:00:00Z",
            "completed_at": null,
            "outcome": "running",
            "build_version": "0.25.1",
            "configuration": {},
            "stats": {},
            "diagnostics": [],
            "storage_reuse": {
                "inbox": {"new":1,"exact":0,"equivalent":0,"capacity_rejected":0},
                "selection": {"new":0,"exact":0,"equivalent":0,"capacity_rejected":0},
                "work": {"new":0,"exact":0,"equivalent":0,"capacity_rejected":0}
            },
            "transport_compaction": {
                "committed": 1,
                "failed": 0,
                "reason": "committed",
                "examined_raw_pages": 2
            }
        }]
    });
    let history: History = serde_json::from_value(value).unwrap();
    assert_eq!(history.cycles[0].storage_reuse.inbox.new, 1);
    assert_eq!(history.cycles[0].storage_reuse.catalogue.new, 0);
    assert_eq!(history.cycles[0].storage_reuse.inbox.single_item_limit, 0);
    assert_eq!(
        history.cycles[0].transport_compaction.logical_charge_before,
        None
    );
}

#[test]
fn unknown_fixed_enum_labels_do_not_escape_from_stored_history() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("support.json");
    let cycle = json!({
        "id": uuid::Uuid::new_v4().to_string(),
        "started_at": "2026-10-22T00:00:00Z",
        "completed_at": null,
        "outcome": "running",
        "build_version": "0.25.1",
        "configuration": {},
        "stats": {},
        "diagnostics": []
    });
    let mut value = json!({
        "schema_version": 1,
        "cycles_evicted": 0,
        "queue_dropped": 0,
        "groups_omitted": 0,
        "previous_history_unavailable": false,
        "cycles": [cycle]
    });
    value["cycles"][0]["storage_reuse"] = json!({"inbox":{"last_reason":"PRIVATE_ENUM_CANARY"}});
    std::fs::write(&path, serde_json::to_vec(&value).unwrap()).unwrap();
    let (clean, status) = read(&path);
    assert_eq!(status, "invalid_or_interrupted");
    assert!(
        !serde_json::to_string(&clean)
            .unwrap()
            .contains("PRIVATE_ENUM_CANARY")
    );
    value["cycles"][0]["storage_reuse"] = json!({});
    value["cycles"][0]["transport_compaction"] = json!({"reason":"PRIVATE_ENUM_CANARY"});
    std::fs::write(&path, serde_json::to_vec(&value).unwrap()).unwrap();
    let (clean, status) = read(&path);
    assert_eq!(status, "invalid_or_interrupted");
    assert!(
        !serde_json::to_string(&clean)
            .unwrap()
            .contains("PRIVATE_ENUM_CANARY")
    );
}