frame-conv 0.2.0

//! Raw-SDK characterization harness: the F-0c spike's participant shapes
//! (record -> send -> receive with interleaved-delivery collection,
//! content-verified enrollment) re-hosted at the workspace pins — obtained
//! from `examples/spike-liminal-conversation/src/lib.rs`, not re-derived
//! (the spike keeps its own =0.3.0 pins). Shared by the F-3b
//! characterization probes (`characterize_leave.rs`,
//! `characterize_admission_recovery.rs`).

use std::error::Error;
use std::fs;
use std::io::Write as _;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use liminal_protocol::wire::{
    ClientRequest, EnrollmentRequest, EnrollmentToken, Generation, ParticipantDelivery, ServerPush,
    ServerValue,
};
use liminal_sdk::remote::{
    ParticipantResumeStore, RemoteConfig, RemoteOperationRecordOutcome, RemoteParticipantError,
    RemoteParticipantHandle, RemoteParticipantInbound, RemoteParticipantSendOutcome,
};
use liminal_sdk::{ConnectionPoolConfig, SdkError};

/// Time allowed for a participant's initial TCP connect + handshake
/// (the F-0c spike's `CONNECT_TIMEOUT`, re-sized for load). The spike's
/// 5s budget was measured on a quiet box; under a full cold battery the
/// dev server is CPU-starved and a handshake read can return EAGAIN for
/// longer than that (seen: census worker at the 2026-07-24 final
/// battery). Sized to `ANSWER_BUDGET` for the same reason that one is
/// 30s — a longer deadline only delays failure, never masks one, because
/// the loop returns on first success and reports the last error at
/// exhaustion.
pub const CONNECT_TIMEOUT: Duration = ANSWER_BUDGET;

/// Total silence budget when waiting for one correlated answer; elapsed IO
/// quanta inside it are benign re-arms (`super::QUANTUM` is the
/// substrate's hardcoded 5s receive quantum — upstream ASK-2).
pub const ANSWER_BUDGET: Duration = Duration::from_secs(30);

/// File-backed LPCR store on the SDK's own trait. `super::FileStore`
/// serves frame-conv's `ResumeStore`; the raw probes persist the SDK's
/// canonical bytes directly so pre-crash state can be snapshotted and
/// restored.
pub struct RawStore {
    path: PathBuf,
}

impl ParticipantResumeStore for RawStore {
    fn persist(&mut self, canonical_lpcr: &[u8]) -> Result<(), SdkError> {
        let mut file = fs::File::create(&self.path).map_err(|error| SdkError::Store {
            description: format!("create {}: {error}", self.path.display()),
        })?;
        file.write_all(canonical_lpcr)
            .and_then(|()| file.sync_all())
            .map_err(|error| SdkError::Store {
                description: format!("write {}: {error}", self.path.display()),
            })
    }
}

fn remote_config(endpoint: &str) -> Result<RemoteConfig, Box<dyn Error>> {
    // The channel-era config wart (upstream ASK-3): channel and conversation
    // names the participant path never uses.
    Ok(RemoteConfig::new(
        endpoint.to_owned(),
        "events",
        "unused-conversation-name",
        ConnectionPoolConfig::new(1, 10, 16),
    )?)
}

fn connect_with(
    endpoint: &str,
    build: impl Fn(&RemoteConfig) -> Result<RemoteParticipantHandle<RawStore>, RemoteParticipantError>,
) -> Result<RemoteParticipantHandle<RawStore>, Box<dyn Error>> {
    let deadline = Instant::now() + CONNECT_TIMEOUT;
    let mut last_error: Option<Box<dyn Error>> = None;
    while Instant::now() < deadline {
        let config = remote_config(endpoint)?;
        // A handshake failure inside `build` (e.g. an EAGAIN frame read
        // on a starved box) is the same transient class as a refused
        // connect — both stay inside the deadline loop.
        let attempt = config
            .connect_tcp()
            .map_err(Box::<dyn Error>::from)
            .and_then(|connected| build(&connected).map_err(Box::<dyn Error>::from));
        match attempt {
            Ok(handle) => return Ok(handle),
            Err(error) => {
                last_error = Some(error);
                std::thread::sleep(Duration::from_millis(20));
            }
        }
    }
    Err(last_error.map_or_else(
        || "participant never connected within timeout".into(),
        |error| format!("participant never connected within timeout: {error}").into(),
    ))
}

/// Connects one FRESH SDK participant persisting its LPCR at `lpcr_path`.
pub fn connect_fresh(
    endpoint: &str,
    lpcr_path: &Path,
) -> Result<RemoteParticipantHandle<RawStore>, Box<dyn Error>> {
    connect_with(endpoint, |config| {
        RemoteParticipantHandle::new(
            config,
            RawStore {
                path: lpcr_path.to_path_buf(),
            },
        )
    })
}

/// Reconnects an SDK participant by RESTORING the aggregate from
/// previously persisted canonical LPCR bytes.
pub fn restore_from(
    endpoint: &str,
    lpcr_path: &Path,
    canonical_lpcr: &[u8],
) -> Result<RemoteParticipantHandle<RawStore>, Box<dyn Error>> {
    connect_with(endpoint, |config| {
        RemoteParticipantHandle::restore(
            config,
            RawStore {
                path: lpcr_path.to_path_buf(),
            },
            canonical_lpcr,
        )
    })
}

/// Whether a receive error is the SDK's ELAPSED IO quantum rather than a
/// dead connection — string classification, exactly the ASK-2 wart the
/// F-0c harness records.
pub fn is_quantum_elapse(error: &RemoteParticipantError) -> bool {
    match error {
        RemoteParticipantError::Transport(sdk_error) => {
            let text = sdk_error.to_string();
            text.contains("os error 35")
                || text.contains("Resource temporarily unavailable")
                || text.contains("timed out")
        }
        _ => false,
    }
}

/// Everything one probe submission can observe — every branch is a
/// recorded finding, none is an error of the probe itself. Unmatched
/// variants' fields exist to be carried into the transcript/failure Debug
/// output (dead-code analysis ignores derived Debug — the same
/// structural-dead-code situation `support/mod.rs` declares).
#[derive(Debug)]
#[allow(dead_code)]
pub enum ProbeAnswer {
    /// The client crate refused to RECORD the operation locally; the wire
    /// never saw it.
    CrateRefusedRecord { detail: String },
    /// Transport died at send or receive.
    TransportLost { detail: String },
    /// No correlated answer arrived within [`ANSWER_BUDGET`].
    NoAnswer {
        deliveries: Vec<ParticipantDelivery>,
    },
    /// A correlated answer arrived (`Applied` or `Refused` at the crate).
    Answer {
        response: Box<RemoteParticipantInbound>,
        deliveries: Vec<ParticipantDelivery>,
    },
}

/// Runs one operation through the SDK's mandatory shape
/// (`record -> send -> receive`), collecting interleaved deliveries until
/// the correlated answer arrives (the F-0c 0.3.0 interleaving finding).
pub fn probe_submit(
    handle: &RemoteParticipantHandle<RawStore>,
    request: ClientRequest,
) -> Result<ProbeAnswer, Box<dyn Error>> {
    let operation = match handle.record_operation(request)? {
        RemoteOperationRecordOutcome::Recorded(operation)
        | RemoteOperationRecordOutcome::Continuous(operation) => operation,
        RemoteOperationRecordOutcome::Refused { request, reason } => {
            return Ok(ProbeAnswer::CrateRefusedRecord {
                detail: format!("request {request:?} refused: {reason:?}"),
            });
        }
    };
    match handle.send_operation(operation)? {
        RemoteParticipantSendOutcome::Sent { .. } => {}
        RemoteParticipantSendOutcome::TransportLost {
            error,
            operation_fate,
            reconnect,
        } => {
            return Ok(ProbeAnswer::TransportLost {
                detail: format!(
                    "send lost transport: {error}; operation fate {operation_fate:?}; \
                     reconnect {reconnect:?}"
                ),
            });
        }
    }
    let start = Instant::now();
    let mut deliveries = Vec::new();
    while start.elapsed() < ANSWER_BUDGET {
        match handle.receive() {
            Ok(RemoteParticipantInbound::Push {
                value: ServerPush::ParticipantDelivery(delivery),
                ..
            }) => deliveries.push(delivery),
            Ok(RemoteParticipantInbound::Push { value, .. }) => {
                return Err(format!("unexpected non-delivery push: {value:?}").into());
            }
            Ok(response) => {
                return Ok(ProbeAnswer::Answer {
                    response: Box::new(response),
                    deliveries,
                });
            }
            Err(error) if is_quantum_elapse(&error) => {}
            Err(error) => {
                return Ok(ProbeAnswer::TransportLost {
                    detail: format!("receive lost transport: {error}"),
                });
            }
        }
    }
    Ok(ProbeAnswer::NoAnswer { deliveries })
}

/// Records and sends one operation WITHOUT waiting for its answer — the
/// crash window's first half (the caller then drops the handle to die
/// before the receipt).
pub fn send_without_answer(
    handle: &RemoteParticipantHandle<RawStore>,
    request: ClientRequest,
) -> Result<(), Box<dyn Error>> {
    let operation = match handle.record_operation(request)? {
        RemoteOperationRecordOutcome::Recorded(operation)
        | RemoteOperationRecordOutcome::Continuous(operation) => operation,
        RemoteOperationRecordOutcome::Refused { request, reason } => {
            return Err(format!("crate refused to record {request:?}: {reason:?}").into());
        }
    };
    match handle.send_operation(operation)? {
        RemoteParticipantSendOutcome::Sent { .. } => Ok(()),
        RemoteParticipantSendOutcome::TransportLost { error, .. } => {
            Err(format!("send lost transport: {error}").into())
        }
    }
}

/// A bound identity, content-verified from its `EnrollBound` receipt.
pub struct Enrolled {
    pub participant_id: u64,
    pub attach_secret: [u8; 32],
    pub generation: Generation,
}

/// Enrolls a participant, proving the binding by decoded receipt content.
pub fn enroll(
    handle: &RemoteParticipantHandle<RawStore>,
    conversation_id: u64,
    token: [u8; 16],
) -> Result<Enrolled, Box<dyn Error>> {
    let answer = probe_submit(
        handle,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new(token),
        }),
    )?;
    let ProbeAnswer::Answer { response, .. } = answer else {
        return Err(format!("enrollment got no answer: {answer:?}").into());
    };
    let RemoteParticipantInbound::Applied { value, .. } = *response else {
        return Err(format!("enrollment was not applied: {response:?}").into());
    };
    let ServerValue::EnrollBound(receipt) = value else {
        return Err(format!("enrollment did not bind: {value:?}").into());
    };
    Ok(Enrolled {
        participant_id: receipt.participant_id(),
        attach_secret: receipt.attach_secret().into_bytes(),
        generation: receipt.origin_binding_epoch().capability_generation,
    })
}

/// Outcome of watching a connection for deliveries across a silence
/// budget: everything observed, in arrival order, plus any terminal
/// non-elapse fate (returned, never swallowed).
#[derive(Debug)]
#[allow(dead_code)]
pub struct DeliveryWatch {
    pub deliveries: Vec<ParticipantDelivery>,
    pub terminal: Option<String>,
}

/// Receives until the silence `budget` elapses or a non-elapse transport
/// fate surfaces; elapsed IO quanta are benign re-arms.
pub fn watch_deliveries(
    handle: &RemoteParticipantHandle<RawStore>,
    budget: Duration,
) -> Result<DeliveryWatch, Box<dyn Error>> {
    let start = Instant::now();
    let mut deliveries = Vec::new();
    let mut terminal = None;
    while start.elapsed() < budget {
        match handle.receive() {
            Ok(RemoteParticipantInbound::Push {
                value: ServerPush::ParticipantDelivery(delivery),
                ..
            }) => deliveries.push(delivery),
            Ok(other) => {
                return Err(format!("unexpected non-delivery inbound: {other:?}").into());
            }
            Err(error) if is_quantum_elapse(&error) => {}
            Err(error) => {
                terminal = Some(format!("{error}"));
                break;
            }
        }
    }
    Ok(DeliveryWatch {
        deliveries,
        terminal,
    })
}