frame-conv 0.2.0

//! The handle's receive engine: one classification path for every inbound
//! frame (nothing observed on the wire is dropped), and the bounded submit
//! loop for correlated operations.

use std::time::{Duration, Instant};

use liminal_protocol::wire::{
    ClientRequest, ParticipantDelivery, ParticipantRecord, RecordAdmission,
    RecordAdmissionAttemptToken, ServerPush, ServerValue,
};
use liminal_sdk::remote::{
    RemoteOperationRecordOutcome, RemoteParticipantInbound, RemoteParticipantSendOutcome,
};

use crate::anomaly::Anomaly;
use crate::envelope::Envelope;
use crate::error::PublishError;
use crate::id::{ConversationSeq, MessageKind, ParticipantRef, PatternId};
use crate::outcome::PublishReceipt;
use crate::seam::{
    ReceiveFate, classify_receive_error, classify_refusal, depart_from_detached, failure_from_died,
    mint_token16,
};
use crate::store::ResumeStore;
use crate::track::ReplyClass;

use super::state::{ConversationHandle, DeathNote, HeldReply, HeldRequest, QueuedItem};

/// One step of the receive engine.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum PumpStep {
    /// A frame was received and classified.
    Classified,
    /// The IO quantum elapsed with nothing to read — benign re-arm.
    Quiet,
}

/// A submit loop failure, in seam vocabulary; callers map it onto their
/// typed error family without losing detail.
#[derive(Debug)]
pub(crate) enum SubmitFailure {
    /// The client crate refused to record the operation (the G2 failed-path
    /// client-side guard: no wire traffic, slot never consumed). The typed
    /// reason rides along so surfaces with a typed answer for a specific
    /// refusal can match it (the leave surface's `AlreadyDead` — probe
    /// finding 2) instead of re-deriving it from the detail string.
    RecordRefused {
        reason: liminal_protocol::client::ClientOperationRecordRefusalReason,
        detail: String,
    },
    /// The transport was lost; the substrate recorded operation and
    /// reconnect fates (finding G2 — the slot is driven to recovery through
    /// the transport-loss surface, never spun on).
    TransportLost { detail: String },
    /// The declared answer window elapsed without the correlated answer.
    AnswerTimeout { waited: Duration },
    /// The client crate refused the correlated answer (conservative
    /// ambiguity) — surfaced, never swallowed.
    InboundRefused { detail: String },
    /// A state, storage, or protocol failure at the seam.
    State { detail: String },
    /// The connection died while waiting for the answer.
    ConnectionFate { detail: String },
}

impl<S: ResumeStore> ConversationHandle<S> {
    /// Receives at most one frame: classifies it, or reports benign quiet.
    ///
    /// Every non-elapse receive error is a connection fate: the handle
    /// latches detached and the error surfaces typed to the caller.
    pub(crate) fn pump_step(&mut self) -> Result<PumpStep, String> {
        match self.sdk.receive() {
            Ok(RemoteParticipantInbound::Push {
                value: ServerPush::ParticipantDelivery(delivery),
                ..
            }) => {
                self.ingest_delivery(delivery);
                Ok(PumpStep::Classified)
            }
            Ok(RemoteParticipantInbound::Push { value, .. }) => {
                self.note_anomaly(Anomaly::UnexpectedFrame {
                    detail: format!("non-delivery push outside an operation: {value:?}"),
                });
                Ok(PumpStep::Classified)
            }
            Ok(RemoteParticipantInbound::Applied { value, .. }) => {
                self.note_anomaly(Anomaly::UnexpectedFrame {
                    detail: format!("operation answer with no operation outstanding: {value:?}"),
                });
                Ok(PumpStep::Classified)
            }
            Ok(RemoteParticipantInbound::Refused { value, reason, .. }) => {
                self.note_anomaly(Anomaly::UnexpectedFrame {
                    detail: format!("inbound refused by client crate: {value:?} ({reason:?})"),
                });
                Ok(PumpStep::Classified)
            }
            Err(error) => match classify_receive_error(&error) {
                ReceiveFate::QuantumElapsed => Ok(PumpStep::Quiet),
                ReceiveFate::ConnectionFate => {
                    self.attached = false;
                    Err(format!("receive failed: {error}"))
                }
            },
        }
    }

    /// Classifies one delivery into the handle's typed surfaces. Nothing is
    /// dropped: every record either lands in an inbox, fulfils an exchange,
    /// or is recorded on the anomaly surface.
    pub(crate) fn ingest_delivery(&mut self, delivery: ParticipantDelivery) {
        if delivery.conversation_id != self.conversation.to_wire() {
            self.note_anomaly(Anomaly::UnexpectedFrame {
                detail: format!(
                    "delivery for foreign conversation {}: {:?}",
                    delivery.conversation_id, delivery.record
                ),
            });
            return;
        }
        let seq = ConversationSeq::new(delivery.delivery_seq);
        self.observe_seq(seq);
        match delivery.record {
            ParticipantRecord::OrdinaryRecord {
                sender_participant_id,
                payload,
            } => {
                let sender = ParticipantRef::from_wire(sender_participant_id);
                self.ingest_ordinary(sender, seq, &payload);
            }
            ParticipantRecord::Attached {
                affected_participant_id,
                ..
            } => {
                self.enqueue_shared(QueuedItem::Joined {
                    peer: ParticipantRef::from_wire(affected_participant_id),
                    seq,
                });
            }
            ParticipantRecord::Detached {
                affected_participant_id,
                cause,
                ..
            } => {
                self.enqueue_shared(QueuedItem::Departed {
                    peer: ParticipantRef::from_wire(affected_participant_id),
                    seq,
                    reason: depart_from_detached(cause),
                });
            }
            ParticipantRecord::Died {
                affected_participant_id,
                cause,
                ..
            } => {
                let peer = ParticipantRef::from_wire(affected_participant_id);
                let failure = failure_from_died(cause);
                self.deaths.push(DeathNote {
                    peer,
                    failure: failure.clone(),
                    seq,
                });
                self.enqueue_shared(QueuedItem::Failed { peer, seq, failure });
            }
            ParticipantRecord::Left {
                affected_participant_id,
                ..
            } => {
                self.enqueue_shared(QueuedItem::Departed {
                    peer: ParticipantRef::from_wire(affected_participant_id),
                    seq,
                    reason: crate::outcome::DepartReason::Left,
                });
            }
            ParticipantRecord::HistoryCompacted { .. } => {
                self.enqueue_shared(QueuedItem::Compacted { seq });
            }
        }
    }

    /// Enqueues a conversation-level item (lifecycle, compaction, gap) into
    /// BOTH pattern inboxes: each consumption surface presents a complete
    /// stream. Pattern-owned records (subscription events, publications)
    /// route to exactly one inbox instead.
    fn enqueue_shared(&mut self, item: QueuedItem) {
        self.publications_inbox.push_back(item.clone());
        self.events_inbox.push_back(item);
    }

    /// Routes one ordinary record through the envelope codec and the
    /// pattern classifier.
    fn ingest_ordinary(&mut self, sender: ParticipantRef, seq: ConversationSeq, payload: &[u8]) {
        let envelope = match Envelope::decode(payload) {
            Ok(envelope) => envelope,
            Err(refusal) => {
                self.counters.malformed_inbound += 1;
                self.anomalies.push(Anomaly::MalformedInbound {
                    seq,
                    slug: refusal.slug,
                    detail: refusal.detail,
                });
                return;
            }
        };
        match envelope.kind {
            MessageKind::Reply => {
                let correlation = envelope.correlation;
                match self.exchanges.classify_reply(correlation) {
                    ReplyClass::First => {
                        self.replies.insert(
                            correlation,
                            HeldReply {
                                envelope,
                                responder: sender,
                                seq,
                            },
                        );
                    }
                    ReplyClass::Duplicate => {
                        self.counters.duplicate_replies += 1;
                        self.anomalies
                            .push(Anomaly::DuplicateReply { correlation, seq });
                    }
                    ReplyClass::Late => {
                        self.counters.late_replies += 1;
                        self.anomalies.push(Anomaly::LateReply { correlation, seq });
                    }
                    ReplyClass::Foreign => {
                        self.counters.foreign_replies += 1;
                        self.anomalies
                            .push(Anomaly::ForeignReply { correlation, seq });
                    }
                }
            }
            MessageKind::Request => {
                self.requests_inbox.push_back(HeldRequest {
                    envelope,
                    requester: sender,
                    seq,
                });
            }
            MessageKind::Event => {
                let inbox = if envelope.pattern == PatternId::PubSub {
                    &mut self.publications_inbox
                } else {
                    &mut self.events_inbox
                };
                inbox.push_back(QueuedItem::Event {
                    publisher: sender,
                    seq,
                    envelope,
                });
            }
        }
    }

    /// Feeds one DELIVERED position to the contiguity tracker; a
    /// detected skip of a non-own position surfaces as a typed gap item
    /// and a named counter (constraint 5's gap surface).
    pub(crate) fn observe_seq(&mut self, seq: ConversationSeq) {
        if let Some(gap) = self.tracker.observe(seq) {
            self.counters.gaps += 1;
            self.enqueue_shared(QueuedItem::Gap {
                expected: gap.expected,
                observed: gap.observed,
            });
        }
    }

    /// Feeds one of this participant's OWN committed positions (its
    /// receipt is the witness — own records are never delivered) via the
    /// tracker's receipt seam: never a frontier advance past unobserved
    /// foreign positions, so a receipt racing a HELD fan-out delivery
    /// cannot mint a false gap (tear-F1). The only gap this path can
    /// surface is the own-ahead ledger's overflow — a delivery stall
    /// indistinguishable from loss, surfaced loudly.
    pub(crate) fn observe_own_seq(&mut self, seq: ConversationSeq) {
        if let Some(gap) = self.tracker.observe_own(seq) {
            self.counters.gaps += 1;
            self.enqueue_shared(QueuedItem::Gap {
                expected: gap.expected,
                observed: gap.observed,
            });
        }
    }

    /// Records an anomaly on both halves of the anomaly home.
    pub(crate) fn note_anomaly(&mut self, anomaly: Anomaly) {
        if matches!(anomaly, Anomaly::UnexpectedFrame { .. }) {
            self.counters.unexpected_frames += 1;
        }
        self.anomalies.push(anomaly);
    }

    /// Runs one operation through the substrate's mandatory shape
    /// (`record → send → receive-until-correlated-answer`), classifying
    /// every interleaved delivery on the way (the 0.3.0 interleaving
    /// re-derivation — F-0c-FINDINGS §Full characterization). The wait for
    /// the correlated answer is bounded by the declared answer window;
    /// elapsed IO quanta within the window are benign re-arms.
    pub(crate) fn submit(&mut self, request: ClientRequest) -> Result<ServerValue, SubmitFailure> {
        let operation = match self.sdk.record_operation(request) {
            Ok(
                RemoteOperationRecordOutcome::Recorded(operation)
                | RemoteOperationRecordOutcome::Continuous(operation),
            ) => operation,
            Ok(RemoteOperationRecordOutcome::Refused { request, reason }) => {
                return Err(SubmitFailure::RecordRefused {
                    reason,
                    detail: format!("client crate refused {request:?}: {reason:?}"),
                });
            }
            Err(error) => {
                return Err(SubmitFailure::State {
                    detail: format!("record failed: {error}"),
                });
            }
        };
        match self.sdk.send_operation(operation) {
            Ok(RemoteParticipantSendOutcome::Sent { .. }) => {
                // `Sent` is transport-write testimony only (finding G1);
                // the operation's evidence is the correlated answer below.
            }
            Ok(RemoteParticipantSendOutcome::TransportLost {
                error,
                operation_fate,
                reconnect,
            }) => {
                self.attached = false;
                return Err(SubmitFailure::TransportLost {
                    detail: format!(
                        "send lost transport: {error}; operation fate: {operation_fate:?}; \
                         reconnect: {reconnect:?}"
                    ),
                });
            }
            Err(error) => {
                return Err(SubmitFailure::State {
                    detail: format!("send failed: {error}"),
                });
            }
        }
        let started = Instant::now();
        loop {
            match self.sdk.receive() {
                Ok(RemoteParticipantInbound::Push {
                    value: ServerPush::ParticipantDelivery(delivery),
                    ..
                }) => self.ingest_delivery(delivery),
                Ok(RemoteParticipantInbound::Push { value, .. }) => {
                    self.note_anomaly(Anomaly::UnexpectedFrame {
                        detail: format!("non-delivery push during an operation: {value:?}"),
                    });
                }
                Ok(RemoteParticipantInbound::Applied { value, .. }) => return Ok(value),
                Ok(RemoteParticipantInbound::Refused { value, reason, .. }) => {
                    return Err(SubmitFailure::InboundRefused {
                        detail: format!("answer refused by client crate: {value:?} ({reason:?})"),
                    });
                }
                Err(error) => match classify_receive_error(&error) {
                    ReceiveFate::QuantumElapsed => {
                        let waited = started.elapsed();
                        if waited >= self.answer_window {
                            return Err(SubmitFailure::AnswerTimeout { waited });
                        }
                    }
                    ReceiveFate::ConnectionFate => {
                        self.attached = false;
                        return Err(SubmitFailure::ConnectionFate {
                            detail: format!("connection lost awaiting answer: {error}"),
                        });
                    }
                },
            }
        }
    }

    /// Admits one encoded envelope as an ordinary record and waits for the
    /// correlated terminal answer. The answer must echo the exact
    /// client-selected attempt identity (the 0.3.0 committed-admission
    /// contract — F-0c-FINDINGS §tripwire succession); a mis-correlated
    /// commit is a protocol violation, never silently accepted.
    pub(crate) fn admit(&mut self, payload: Vec<u8>) -> Result<PublishReceipt, PublishError> {
        self.admit_with_token(payload, RecordAdmissionAttemptToken::new(mint_token16()))
    }

    /// The same admission shape with a CALLER-DETERMINED attempt token: the
    /// pub/sub path, where the `PublicationId`'s bytes ARE the token
    /// (amendment A4 — the substrate's admission is the reuse-refusal
    /// home).
    pub(crate) fn admit_with_token(
        &mut self,
        payload: Vec<u8>,
        token: RecordAdmissionAttemptToken,
    ) -> Result<PublishReceipt, PublishError> {
        let answer = self
            .submit(ClientRequest::RecordAdmission(RecordAdmission {
                conversation_id: self.conversation.to_wire(),
                participant_id: self.me.to_wire(),
                capability_generation: self.generation,
                record_admission_attempt_token: token,
                payload,
            }))
            .map_err(submit_to_publish)?;
        match answer {
            ServerValue::RecordCommitted(committed) => {
                if committed.request().record_admission_attempt_token != token {
                    return Err(PublishError::Protocol {
                        detail: format!(
                            "commit answer echoed a foreign attempt token: {committed:?}"
                        ),
                    });
                }
                let seq = ConversationSeq::new(committed.delivery_seq());
                // The sender is excluded from its own record's delivery
                // (F-0c assertion 5); the committed position feeds the
                // tracker through the RECEIPT seam — commit answers and
                // fan-out deliveries carry no cross-stream order, so a
                // receipt must never advance the delivery frontier past
                // a held foreign position (tear-F1).
                self.observe_own_seq(seq);
                Ok(PublishReceipt { seq })
            }
            other => {
                let (class, detail) = classify_refusal(&other);
                Err(PublishError::Refused { class, detail })
            }
        }
    }
}

/// Maps a submit failure onto the publish error family without losing the
/// exact detail.
pub(crate) fn submit_to_publish(failure: SubmitFailure) -> PublishError {
    match failure {
        SubmitFailure::RecordRefused { detail, .. }
        | SubmitFailure::State { detail }
        | SubmitFailure::InboundRefused { detail } => PublishError::Protocol { detail },
        SubmitFailure::TransportLost { detail } | SubmitFailure::ConnectionFate { detail } => {
            PublishError::ConnectionLost { detail }
        }
        SubmitFailure::AnswerTimeout { waited } => PublishError::AnswerTimeout { waited },
    }
}