saddle-boundary 0.3.29

//! Borrowed unified outlet: source facts are submitted before conversion.
//! No new logger, task, queue, timer or response ownership.
use saddle_core::{
    BoundedDiagnostic, BoundedDiagnosticCause, CaptureSite, DiagnosticCategory, DiagnosticCode,
    DiagnosticOccurrence, DiagnosticOutcomeAxes, DiagnosticStage, InlineDiagnosticText,
    OperationOutcome, ResponseDelivery,
};
use saddle_observability::{DiagnosticSubmission, EmergencyDiagnosticHandle, EventContext};
use std::io;
use serde::Serialize;

/// Only the parsed input handed to tonic may construct this source target.
#[derive(Debug, Serialize)]
pub(crate) struct ConnectTarget {
    authority: String,
    hostname: String,
    port: u16,
    expanded_zone: Option<String>,
    template_identity: Option<String>,
    deployment_config_version: ProvenanceIdentity,
    deployment_artifact_identity: ProvenanceIdentity,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "snake_case")]
enum ProvenanceIdentity { NotProvided }
impl ConnectTarget {
    pub(crate) fn from_endpoint(
        endpoint: &tonic::transport::Endpoint,
        identity: Option<&crate::ResolvedEndpointIdentity>,
    ) -> Result<Self, crate::BoundaryError> {
        let uri = endpoint.uri();
        if let Some(identity) = identity {
            let expected = crate::ProfuseContractEndpoint::new(
                identity.template.replace("{zone}", &identity.zone),
            )?;
            let expected = tonic::transport::Endpoint::from_shared(expected.0)
                .map_err(|_| crate::BoundaryError::invalid_request("invalid resolved endpoint identity"))?;
            if expected.uri().authority() != uri.authority() {
                return Err(crate::BoundaryError::invalid_request("resolved endpoint identity does not match connection input"));
            }
        }
        Ok(Self {
            authority: uri.authority().expect("validated endpoint authority").as_str().to_owned(),
            hostname: uri.host().expect("validated endpoint host").to_owned(),
            port: uri.port_u16().unwrap_or(80),
            expanded_zone: identity.map(|id| id.zone.clone()),
            template_identity: identity.map(|id| id.template.clone()),
            deployment_config_version: ProvenanceIdentity::NotProvided,
            deployment_artifact_identity: ProvenanceIdentity::NotProvided,
        })
    }
}

/// No task or timer: observes cancellation of the existing caller-owned future.
/// Drop cannot prove whether the caller selected timeout, disconnect, or cancellation.
pub(crate) async fn observe_future<T>(
    source: Option<&dyn SourceCapture>,
    future: impl std::future::Future<Output = T>,
) -> T {
    struct Guard<'a> {
        source: Option<&'a dyn SourceCapture>,
        completed: bool,
    }
    impl Drop for Guard<'_> {
        fn drop(&mut self) {
            if !self.completed {
                if let Some(source) = self.source {
                    let panicked = std::thread::panicking();
                    let receipt = if panicked {
                        source.submit_cause(
                            DiagnosticCategory::UnexpectedError,
                            BoundedDiagnosticCause::new(
                                DiagnosticStage::RequestOutbound,
                                code("transport.future_panicked"),
                            )
                            .with_object(InlineDiagnosticText::metadata("profusecontract")),
                        )
                    } else {
                        source.local_failure(LocalFailure::Cancelled)
                    };
                    let axes = DiagnosticOutcomeAxes {
                        operation: if panicked {
                            OperationOutcome::Panicked
                        } else {
                            OperationOutcome::Cancelled
                        },
                        ..Default::default()
                    };
                    // Degradation is exposed by the shared handle snapshot; no business retry.
                    // Required callers retain the receipt outside this future
                    // and own its single boundary after the borrow is dropped.
                    if !source.required() {
                        let _ = source.boundary(receipt, &axes);
                    }
                }
            }
        }
    }
    let mut guard = Guard {
        source,
        completed: false,
    };
    let result = future.await;
    guard.completed = true;
    result
}

pub struct SourceContext<'a> {
    pub output: &'a EmergencyDiagnosticHandle,
    pub association: Option<(&'a saddle_core::CallContext, &'a EventContext)>,
}

pub(crate) struct WireAssociation<'a> {
    pub(crate) event: &'a EventContext,
    pub(crate) trace_id: &'a str,
    pub(crate) rpc_id: &'a str,
    pub(crate) zone: &'a str,
}

pub(crate) struct RoutedSourceContext<'a> {
    pub(crate) base: SourceContext<'a>,
    pub(crate) wire: Option<WireAssociation<'a>>,
}

/// Enqueued is not written; retain this status independently of business Result.
#[derive(Clone, Copy)]
pub struct SourceReceipt {
    pub(crate) diagnostic_id: u64,
    pub(crate) submission: DiagnosticSubmission,
    pub(crate) occurrence: DiagnosticOccurrence,
}

impl SourceReceipt {
    pub fn diagnostic_id(&self) -> u64 {
        self.diagnostic_id
    }
    pub fn submission(&self) -> DiagnosticSubmission {
        self.submission
    }
    pub fn occurrence(&self) -> DiagnosticOccurrence {
        self.occurrence
    }
}

impl std::fmt::Debug for SourceReceipt {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SourceReceipt")
            .field("diagnostic_id", &self.diagnostic_id)
            .field("submission", &self.submission)
            .finish_non_exhaustive()
    }
}

#[derive(Clone, Copy)]
pub enum IoOperation {
    Accept,
    ReadHead,
    ReadBody,
    WriteResponse,
    ShutdownResponse,
    DeadlineWrite,
}

#[derive(Clone, Copy)]
pub enum LocalFailure {
    InvalidRequest,
    InvalidAuthority,
    Deadline,
    Cancelled,
}

pub(crate) fn code(value: &'static str) -> DiagnosticCode {
    DiagnosticCode::new(value).expect("closed transport code")
}

pub(crate) fn io_kind(error: &io::Error) -> DiagnosticCode {
    code(match error.kind() {
        io::ErrorKind::ConnectionRefused => "connection_refused",
        io::ErrorKind::ConnectionReset => "connection_reset",
        io::ErrorKind::ConnectionAborted => "connection_aborted",
        io::ErrorKind::BrokenPipe => "broken_pipe",
        io::ErrorKind::UnexpectedEof => "unexpected_eof",
        io::ErrorKind::WriteZero => "write_zero",
        io::ErrorKind::TimedOut => "timed_out",
        io::ErrorKind::Interrupted => "interrupted",
        io::ErrorKind::WouldBlock => "would_block",
        io::ErrorKind::InvalidData => "invalid_data",
        io::ErrorKind::InvalidInput => "invalid_input",
        io::ErrorKind::PermissionDenied => "permission_denied",
        io::ErrorKind::NotConnected => "not_connected",
        io::ErrorKind::AddrInUse => "addr_in_use",
        io::ErrorKind::AddrNotAvailable => "addr_not_available",
        _ => "unknown",
    })
}

impl SourceContext<'_> {
    #[cfg(test)]
    #[track_caller]
    pub(crate) fn rpc_failure(&self, status: &tonic::Status) -> SourceReceipt {
        (self as &dyn SourceCapture)
            .rpc_failure(status, None)
            .expect("legacy receipt")
    }
    #[track_caller]
    pub fn local_failure(&self, failure: LocalFailure) -> SourceReceipt {
        SourceCapture::local_failure(self, failure).expect("legacy receipt")
    }
    #[track_caller]
    pub fn io_failure(&self, operation: IoOperation, error: &io::Error) -> SourceReceipt {
        (self as &dyn SourceCapture)
            .io_failure(operation, error)
            .expect("legacy receipt")
    }
    pub fn boundary(
        &self,
        source: Option<SourceReceipt>,
        axes: &DiagnosticOutcomeAxes,
    ) -> DiagnosticSubmission {
        SourceCapture::boundary(self, source, axes)
    }
}

/// Internal common source path: legacy callers and required callers use the
/// same classifiers, before conversion. No network implementation is copied.
pub(crate) trait SourceCapture: Sync {
    fn guarded(&self) -> bool {
        false
    }
    fn required(&self) -> bool {
        false
    }
    #[track_caller]
    fn submit_cause(
        &self,
        category: DiagnosticCategory,
        cause: BoundedDiagnosticCause,
    ) -> Option<SourceReceipt>;
    /// Original is borrowed before lossy classification. Guarded consumers retain
    /// their mandatory receipt outside the future, not in the legacy error.
    #[track_caller]
    fn submit_error(
        &self,
        error: &(dyn std::error::Error + 'static),
        category: DiagnosticCategory,
        cause: BoundedDiagnosticCause,
    ) -> Option<SourceReceipt> {
        let _ = error;
        self.submit_cause(category, cause)
    }
    #[track_caller]
    fn submit_connect_error(
        &self,
        error: &(dyn std::error::Error + 'static),
        target: &ConnectTarget,
        category: DiagnosticCategory,
        cause: BoundedDiagnosticCause,
    ) -> Option<SourceReceipt>;
    #[track_caller]
    fn submit_remote_error(
        &self,
        error: &(dyn std::error::Error + 'static),
        category: DiagnosticCategory,
        cause: BoundedDiagnosticCause,
        business_unit: &str,
        function: &str,
    ) -> Option<SourceReceipt> {
        let _ = (business_unit, function);
        self.submit_error(error, category, cause)
    }
    /// The peer response has protocol fields, but no Error interface or message.
    /// Keep the borrowed, validated target and actual enum values until the
    /// original description and Debug record have been written.
    #[track_caller]
    fn remote_failure(
        &self,
        failure: &crate::TechnicalFailure,
        business_unit: &str,
        function: &str,
        cause: BoundedDiagnosticCause,
    ) -> Option<SourceReceipt> {
        let _ = (failure, business_unit, function);
        self.submit_cause(DiagnosticCategory::ExpectedRejection, cause)
    }
    fn bind_child(
        &self,
        _: &saddle_core::CallContext,
        _: &EventContext,
        _: &str,
        _: &str,
    ) -> Result<(), crate::BoundaryError> {
        Ok(())
    }
    fn boundary(
        &self,
        source: Option<SourceReceipt>,
        axes: &DiagnosticOutcomeAxes,
    ) -> DiagnosticSubmission;
    #[track_caller]
    fn local_failure(&self, failure: LocalFailure) -> Option<SourceReceipt> {
        let reason = match failure {
            LocalFailure::InvalidRequest => "transport.request_invalid",
            LocalFailure::InvalidAuthority => "transport.authority_invalid",
            LocalFailure::Deadline => "transport.deadline_elapsed",
            LocalFailure::Cancelled => "transport.cancelled",
        };
        self.submit_cause(
            DiagnosticCategory::ExpectedRejection,
            BoundedDiagnosticCause::new(DiagnosticStage::RequestOutbound, code(reason))
                .with_object(InlineDiagnosticText::metadata("profusecontract")),
        )
    }
    #[track_caller]
    fn local_error(
        &self,
        error: &crate::BoundaryError,
        failure: LocalFailure,
    ) -> Option<SourceReceipt> {
        let _ = error;
        self.local_failure(failure)
    }
    #[track_caller]
    fn protocol_failure(&self, _description: &'static str) -> Option<SourceReceipt> {
        self.submit_cause(
            DiagnosticCategory::UnexpectedError,
            BoundedDiagnosticCause::new(
                DiagnosticStage::RequestOutbound,
                code("transport.response_invalid"),
            )
            .with_object(InlineDiagnosticText::metadata("profusecontract")),
        )
    }
}

impl SourceCapture for SourceContext<'_> {
    fn submit_connect_error(
        &self,
        error: &(dyn std::error::Error + 'static),
        target: &ConnectTarget,
        category: DiagnosticCategory,
        cause: BoundedDiagnosticCause,
    ) -> Option<SourceReceipt> {
        self.capture_original(error, target, category, cause)
    }
    fn boundary(
        &self,
        source: Option<SourceReceipt>,
        axes: &DiagnosticOutcomeAxes,
    ) -> DiagnosticSubmission {
        self.output
            .submit_boundary(source.map(|value| value.occurrence), axes, self.association)
    }

    fn submit_cause(
        &self,
        category: DiagnosticCategory,
        cause: BoundedDiagnosticCause,
    ) -> Option<SourceReceipt> {
        let diagnostic = BoundedDiagnostic::capture(category, CaptureSite::FirstObserved, cause);
        let submission = self.output.submit_bounded(
            Some(&diagnostic),
            &DiagnosticOutcomeAxes::default(),
            self.association,
        );
        Some(SourceReceipt {
            diagnostic_id: diagnostic.id(),
            occurrence: diagnostic.occurrence(),
            submission,
        })
    }
}

impl SourceContext<'_> {
    fn capture_original<T: Serialize>(
        &self, error: &(dyn std::error::Error + 'static), target: &T,
        category: DiagnosticCategory, cause: BoundedDiagnosticCause,
    ) -> Option<SourceReceipt> {
        let scope = if let Some((call, event)) = self.association {
            saddle_observability::RequestDiagnosticScope::established(self.output, call, event)
        } else {
            saddle_observability::RequestDiagnosticScope::unassociated_transport(self.output)
        };
        let receipt = scope.source_error_with_target(error, target, category, cause);
        Some(SourceReceipt {
            diagnostic_id: receipt.source_diagnostic().id(),
            submission: receipt.submission(),
            occurrence: receipt.source_diagnostic().occurrence(),
        })
    }
}

impl SourceCapture for RoutedSourceContext<'_> {
    fn submit_cause(&self, category: DiagnosticCategory, cause: BoundedDiagnosticCause) -> Option<SourceReceipt> {
        self.base.submit_cause(category, cause)
    }
    fn submit_error(&self, error: &(dyn std::error::Error + 'static), category: DiagnosticCategory, cause: BoundedDiagnosticCause) -> Option<SourceReceipt> {
        self.base.submit_error(error, category, cause)
    }
    fn submit_connect_error(
        &self, error: &(dyn std::error::Error + 'static), target: &ConnectTarget,
        category: DiagnosticCategory, cause: BoundedDiagnosticCause,
    ) -> Option<SourceReceipt> {
        if self.base.association.is_some() {
            return self.base.submit_connect_error(error, target, category, cause);
        }
        let Some(wire) = self.wire.as_ref() else {
            return self.base.submit_connect_error(error, target, category, cause);
        };
        let Some(scope) = saddle_observability::RequestDiagnosticScope::outbound_wire(
            self.base.output, wire.event, wire.trace_id, wire.rpc_id, wire.zone,
        ) else {
            return self.base.submit_connect_error(error, target, category, cause);
        };
        let receipt = scope.source_error_with_target(error, target, category, cause);
        Some(SourceReceipt {
            diagnostic_id: receipt.source_diagnostic().id(),
            submission: receipt.submission(),
            occurrence: receipt.source_diagnostic().occurrence(),
        })
    }
    fn boundary(&self, source: Option<SourceReceipt>, axes: &DiagnosticOutcomeAxes) -> DiagnosticSubmission {
        self.base.boundary(source, axes)
    }
}

impl dyn SourceCapture + '_ {
    /// Classification is bounded; original capture is delegated before mapping.
    #[track_caller]
    pub(crate) fn connect_failure(
        &self,
        error: &(dyn std::error::Error + 'static),
        target: &ConnectTarget,
    ) -> Option<SourceReceipt> {
        let cause = BoundedDiagnosticCause::new(
            DiagnosticStage::RequestOutbound,
            code("transport.connect_failed"),
        )
        .with_object(InlineDiagnosticText::metadata("profusecontract"));
        let (cause, expected) = error_cause(cause, error);
        self.submit_connect_error(
            error,
            target,
            if expected {
                DiagnosticCategory::ExpectedRejection
            } else {
                DiagnosticCategory::UnexpectedError
            },
            cause,
        )
    }

    #[track_caller]
    pub(crate) fn rpc_failure(
        &self, status: &tonic::Status, remote_target: Option<(&str, &str)>,
    ) -> Option<SourceReceipt> {
        use tonic::Code;
        let reason = match status.code() {
            Code::Ok => "transport.rpc.ok",
            Code::Cancelled => "transport.rpc.cancelled",
            Code::Unknown => "transport.rpc.unknown",
            Code::InvalidArgument => "transport.rpc.invalid_argument",
            Code::DeadlineExceeded => "transport.rpc.deadline_exceeded",
            Code::NotFound => "transport.rpc.not_found",
            Code::AlreadyExists => "transport.rpc.already_exists",
            Code::PermissionDenied => "transport.rpc.permission_denied",
            Code::ResourceExhausted => "transport.rpc.resource_exhausted",
            Code::FailedPrecondition => "transport.rpc.failed_precondition",
            Code::Aborted => "transport.rpc.aborted",
            Code::OutOfRange => "transport.rpc.out_of_range",
            Code::Unimplemented => "transport.rpc.unimplemented",
            Code::Internal => "transport.rpc.internal",
            Code::Unavailable => "transport.rpc.unavailable",
            Code::DataLoss => "transport.rpc.data_loss",
            Code::Unauthenticated => "transport.rpc.unauthenticated",
        };
        let cause = BoundedDiagnosticCause::new(DiagnosticStage::RequestOutbound, code(reason))
            .with_object(InlineDiagnosticText::metadata("profusecontract"));
        let (cause, _) = error_cause(cause, status);
        let expected = matches!(
            status.code(),
            Code::Cancelled
                | Code::DeadlineExceeded
                | Code::ResourceExhausted
                | Code::InvalidArgument
                | Code::Unauthenticated
                | Code::PermissionDenied
        );
        let category = if expected {
                DiagnosticCategory::ExpectedRejection
            } else {
                DiagnosticCategory::UnexpectedError
            };
        if let Some((business_unit, function)) = remote_target {
            self.submit_remote_error(status, category, cause, business_unit, function)
        } else {
            self.submit_error(status, category, cause)
        }
    }
    #[track_caller]
    pub fn io_failure(&self, operation: IoOperation, error: &io::Error) -> Option<SourceReceipt> {
        let (stage, reason) = match operation {
            IoOperation::Accept => (DiagnosticStage::StartupListener, "transport.accept_io"),
            IoOperation::ReadHead => (DiagnosticStage::RequestDecode, "transport.read_head_io"),
            IoOperation::ReadBody => (DiagnosticStage::RequestDecode, "transport.read_body_io"),
            IoOperation::WriteResponse => (DiagnosticStage::RequestResponse, "transport.write_io"),
            IoOperation::ShutdownResponse => {
                (DiagnosticStage::RequestResponse, "transport.shutdown_io")
            }
            IoOperation::DeadlineWrite => (
                DiagnosticStage::RequestResponse,
                "transport.deadline_write_io",
            ),
        };
        let expected = !matches!(operation, IoOperation::Accept)
            && matches!(
                error.kind(),
                io::ErrorKind::ConnectionReset
                    | io::ErrorKind::ConnectionAborted
                    | io::ErrorKind::BrokenPipe
                    | io::ErrorKind::UnexpectedEof
                    | io::ErrorKind::TimedOut
                    | io::ErrorKind::Interrupted
                    | io::ErrorKind::WouldBlock
                    | io::ErrorKind::InvalidData
                    | io::ErrorKind::InvalidInput
            );
        self.submit_error(
            error,
            if expected {
                DiagnosticCategory::ExpectedRejection
            } else {
                DiagnosticCategory::UnexpectedError
            },
            BoundedDiagnosticCause::new(stage, code(reason))
                .with_system(io_kind(error), error.raw_os_error())
                .with_object(InlineDiagnosticText::metadata("profusegw")),
        )
    }
}

fn error_cause(
    mut cause: BoundedDiagnosticCause,
    error: &(dyn std::error::Error + 'static),
) -> (BoundedDiagnosticCause, bool) {
    let mut current = Some(error);
    for _ in 0..16 {
        let Some(error) = current else { break };
        if let Some(io) = error.downcast_ref::<io::Error>() {
            cause = cause.with_system(io_kind(io), io.raw_os_error());
            return (
                cause,
                matches!(
                    io.kind(),
                    io::ErrorKind::TimedOut | io::ErrorKind::Interrupted
                ),
            );
        }
        current = error.source();
    }
    (cause.with_system(code("unknown"), None), false)
}

/// Kept outside the write future, so cancelling that future cannot erase progress.
/// Observer only: caller retains the existing deadline and response ownership.
pub struct DeliveryProgress {
    written: u64,
    state: ResponseDelivery,
    operation: OperationOutcome,
    first_failure: Option<SourceReceipt>,
    terminal: bool,
}
impl Default for DeliveryProgress {
    fn default() -> Self {
        Self {
            written: 0,
            state: ResponseDelivery::NotStarted,
            operation: OperationOutcome::Unknown,
            first_failure: None,
            terminal: false,
        }
    }
}
impl DeliveryProgress {
    /// Call only for actual positive write results, never planned body length.
    pub fn wrote(&mut self, count: usize) {
        if self.terminal {
            return;
        }
        self.written = self.written.saturating_add(count as u64);
        if count > 0 {
            self.state = ResponseDelivery::Partial;
        }
    }
    pub fn local_write_complete(&mut self) {
        if self.terminal {
            return;
        }
        self.terminal = true;
        self.state = ResponseDelivery::LocalWriteComplete;
        self.operation = OperationOutcome::Succeeded;
    }
    pub fn failed(&mut self, receipt: SourceReceipt) {
        if self.terminal {
            return;
        }
        self.terminal = true;
        if self.first_failure.is_none() {
            self.first_failure = Some(receipt);
        }
        self.state = ResponseDelivery::Failed;
        self.operation = OperationOutcome::Failed;
    }
    pub fn timed_out(&mut self) {
        if self.terminal {
            return;
        }
        self.terminal = true;
        self.state = ResponseDelivery::TimedOut;
        self.operation = OperationOutcome::TimedOut;
    }
    pub fn cancelled(&mut self) {
        if self.terminal {
            return;
        }
        self.terminal = true;
        self.state = ResponseDelivery::Cancelled;
        self.operation = OperationOutcome::Cancelled;
    }
    pub fn first_failure(&self) -> Option<SourceReceipt> {
        self.first_failure
    }
    pub fn axes(&self) -> DiagnosticOutcomeAxes {
        DiagnosticOutcomeAxes {
            operation: self.operation,
            delivery: self.state,
            bytes_written: Some(self.written),
            ..DiagnosticOutcomeAxes::default()
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn real_tonic_dns_chain_matches_borrowed_error_and_target() {
        use saddle_observability::{EmergencyDiagnostics, EventContext, FileLoggingConfig, RequestIdentity, Rotation, RouteIdentity};
        struct Tap<'a> {
            source: SourceContext<'a>,
            expected: std::sync::Mutex<Option<(serde_json::Value, Vec<(String, String)>)>>,
        }
        impl SourceCapture for Tap<'_> {
            fn submit_cause(&self, category: DiagnosticCategory, cause: BoundedDiagnosticCause) -> Option<SourceReceipt> {
                self.source.submit_cause(category, cause)
            }
            fn boundary(&self, receipt: Option<SourceReceipt>, axes: &DiagnosticOutcomeAxes) -> DiagnosticSubmission {
                self.source.boundary(receipt, axes)
            }
            fn submit_connect_error(
                &self, error: &(dyn std::error::Error + 'static), target: &ConnectTarget,
                category: DiagnosticCategory, cause: BoundedDiagnosticCause,
            ) -> Option<SourceReceipt> {
                let mut chain = Vec::new();
                let mut cursor = Some(error);
                while let Some(current) = cursor {
                    chain.push((current.to_string(), format!("{current:?}")));
                    cursor = current.source();
                    assert!(chain.len() < 32, "unexpected cause cycle");
                }
                *self.expected.lock().unwrap() = Some((serde_json::to_value(target).unwrap(), chain));
                self.source.submit_connect_error(error, target, category, cause)
            }
        }
        let _ = saddle_observability::init(Default::default());
        let directory = std::env::temp_dir().join(format!(
            "tonic-dns-original-{}-{}", std::process::id(),
            std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos(),
        ));
        std::fs::create_dir_all(&directory).unwrap();
        let mut writer = EmergencyDiagnostics::start(&FileLoggingConfig::new(&directory, Rotation::Daily)).unwrap();
        let handle = writer.handle();
        let observer = saddle_observability::global().unwrap();
        let (call, _) = observer.start_external_call_with_rpc(
            "fixture", "dns", "profusecontract", "connect", Some("dns-trace"),
            saddle_core::RpcCorrelationId::new("0.7").unwrap(),
        ).unwrap();
        let event = EventContext::new(
            RequestIdentity::new("dns-chain").unwrap(), RouteIdentity::new("connect").unwrap(), 1,
        ).unwrap();
        let tap = Tap {
            source: SourceContext { output: &handle, association: Some((call.context(), &event)) },
            expected: std::sync::Mutex::new(None),
        };
        let host = "saddle-chain-no-record.invalid";
        let template = crate::ProfuseContractAuthorityTemplate::new("http://{zone}:45052").unwrap();
        let endpoint = template.resolve(host).unwrap();
        let deadline = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() as i64 + 15_000;
        let error = tokio::time::timeout(std::time::Duration::from_secs(16),
            crate::TonicBoundary::connect_until(endpoint, deadline, Some(&tap)),
        ).await.unwrap().err().expect("real resolver must fail");
        let receipt = error.source_receipt().expect("source receipt");
        assert_eq!(receipt.submission, DiagnosticSubmission::Written);
        let (expected_target, expected_chain) = tap.expected.lock().unwrap().take().unwrap();
        let raw = std::fs::read_to_string(writer.target()).unwrap();
        let rows: Vec<serde_json::Value> = raw.lines().map(|line| serde_json::from_str(line).unwrap()).collect();
        let original: Vec<_> = rows.iter().filter(|row| row["event"] == "request_error_original").collect();
        assert_eq!(original.last().unwrap()["state"], "exposed_chain_complete");
        let header: serde_json::Value = serde_json::from_str(original[0]["payload"].as_str().unwrap()).unwrap();
        assert_eq!(header["target"], expected_target);
        assert_eq!(header["target"]["hostname"], host);
        assert_eq!(header["context"]["trace_id"]["value"], "dns-trace");
        assert_eq!(header["context"]["rpc_id"]["value"], "0.7");
        assert!(original.iter().all(|row| row["occurrence"] == serde_json::to_value(receipt.occurrence).unwrap()));
        assert!(expected_chain.len() >= 2, "expected tonic and resolver causes");
        for (depth, (description, debug)) in expected_chain.iter().enumerate() {
            for (channel, expected) in [("description", description), ("debug", debug)] {
                let actual: String = original.iter()
                    .filter(|row| row["cause_depth"] == depth && row["channel"] == channel)
                    .map(|row| row["payload"].as_str().unwrap()).collect();
                assert_eq!(&actual, expected, "{channel} at depth {depth}");
            }
        }
        call.succeed();
        while matches!(writer.shutdown(), saddle_observability::DiagnosticShutdown::Pending) {
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        }
        std::fs::remove_dir_all(directory).unwrap();
    }

    #[test]
    fn resolved_identity_must_match_actual_connect_input() {
        let endpoint = tonic::transport::Endpoint::from_shared("http://actual.invalid:45051".to_owned()).unwrap();
        let wrong = crate::ResolvedEndpointIdentity {
            zone: "other.invalid".into(), template: "http://{zone}:45051".into(),
        };
        assert!(ConnectTarget::from_endpoint(&endpoint, Some(&wrong)).is_err());
        let correct = crate::ResolvedEndpointIdentity {
            zone: "actual.invalid".into(), template: "http://{zone}:45051".into(),
        };
        let target = ConnectTarget::from_endpoint(&endpoint, Some(&correct)).unwrap();
        let value = serde_json::to_value(target).unwrap();
        assert_eq!(value["authority"], "actual.invalid:45051");
        assert_eq!(value["expanded_zone"], "actual.invalid");
    }

    #[tokio::test]
    async fn normal_connection_has_no_failure_receipt() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let accepted = tokio::spawn(async move { listener.accept().await.unwrap().0 });
        let endpoint = crate::ProfuseContractEndpoint::new(format!("http://{address}")).unwrap();
        let deadline = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() as i64 + 3000;
        let boundary = crate::TonicBoundary::connect_until(endpoint, deadline, None).await.unwrap();
        drop(boundary);
        let _ = accepted.await.unwrap();
    }

    #[tokio::test]
    async fn unavailable_output_keeps_tonic_original_in_safe_carrier() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let address = listener.local_addr().unwrap();
        drop(listener);
        let endpoint = crate::ProfuseContractEndpoint::new(format!("http://{address}")).unwrap();
        let deadline = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() as i64 + 3000;
        let failure = crate::TonicBoundary::connect_until(endpoint, deadline, None)
            .await.err().expect("refused connection");
        let original = failure.unconfirmed_connect_error().expect("original retained without output");
        assert!(format!("{original:?}").contains("ConnectionRefused"));
        let classified = crate::request_diagnostics::BoundaryFailure::from_error(&failure);
        assert!(classified.unconfirmed_connect_error().is_some());
        assert!(!format!("{failure:?}").contains(&address.to_string()));
        assert!(!format!("{classified:?}").contains(&address.to_string()));
    }

    #[tokio::test]
    async fn source_immediate_submission_and_reference_are_secret_free() {
        use saddle_observability::{EmergencyDiagnostics, FileLoggingConfig, Rotation};
        let directory = std::env::temp_dir().join(format!(
            "transport-source-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir(&directory).unwrap();
        let mut writer =
            EmergencyDiagnostics::start(&FileLoggingConfig::new(&directory, Rotation::Daily))
                .unwrap();
        let handle = writer.handle();
        let source = SourceContext {
            output: &handle,
            association: None,
        };
        let receipt = source.rpc_failure(&tonic::Status::internal(
            "Bearer SECRET_TOKEN https://secret.invalid/payload",
        ));
        assert_eq!(receipt.submission, DiagnosticSubmission::Enqueued);
        assert_eq!(
            source.boundary(Some(receipt), &DiagnosticOutcomeAxes::default()),
            DiagnosticSubmission::Enqueued
        );
        // Actual connection error is captured before conversion, without formatting URL.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let address = listener.local_addr().unwrap();
        drop(listener);
        let endpoint = crate::ProfuseContractEndpoint::new(format!("http://{address}")).unwrap();
        let deadline = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as i64
            + 1000;
        let error =
            match crate::TonicBoundary::connect_until(endpoint, deadline, Some(&source)).await {
                Ok(_) => panic!("unexpected connect"),
                Err(error) => error,
            };
        assert!(error.diagnostic().is_none());
        assert_eq!(
            error.source_receipt().unwrap().submission,
            DiagnosticSubmission::Written
        );
        use std::io::Write;
        let (mut socket, peer) = std::os::unix::net::UnixStream::pair().unwrap();
        let mut delivery = DeliveryProgress::default();
        delivery.wrote(socket.write(b"part").unwrap());
        drop(peer);
        let io = socket.write(b"remaining").unwrap_err();
        delivery.failed(source.io_failure(IoOperation::WriteResponse, &io));
        let axes = serde_json::to_value(delivery.axes()).unwrap();
        assert_eq!(axes["bytes_written"], 4);
        assert_eq!(axes["delivery"], "failed");
        assert_eq!(axes["business"], "unknown");
        // Actual cancellation of a polled future emits source then ID-only result.
        let mut future = Box::pin(observe_future(Some(&source), std::future::pending::<()>()));
        use std::future::Future;
        assert!(
            future
                .as_mut()
                .poll(&mut std::task::Context::from_waker(std::task::Waker::noop()))
                .is_pending()
        );
        drop(future);
        let stop = std::time::Instant::now() + std::time::Duration::from_secs(2);
        while matches!(
            writer.shutdown(),
            saddle_observability::DiagnosticShutdown::Pending
        ) {
            assert!(
                std::time::Instant::now() < stop,
                "test writer did not drain"
            );
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        }
        let raw = std::fs::read_to_string(writer.target()).unwrap();
        for secret in ["SECRET_TOKEN", "secret.invalid"] {
            assert!(!raw.contains(secret));
        }
        let rows: Vec<serde_json::Value> = raw
            .lines()
            .map(|line| serde_json::from_str(line).unwrap())
            .collect();
        assert!(rows.len() > 6);
        let originals: Vec<_> = rows.iter().filter(|row| row["event"] == "request_error_original").collect();
        assert_eq!(originals.last().unwrap()["state"], "exposed_chain_complete");
        let target: serde_json::Value = serde_json::from_str(originals[0]["payload"].as_str().unwrap()).unwrap();
        assert_eq!(target["target"]["authority"], address.to_string());
        assert!(rows[0]["diagnostic"].is_object());
        assert!(rows[1]["diagnostic"].is_null());
        assert_eq!(
            rows[0]["diagnostic_reference"],
            rows[1]["diagnostic_reference"]
        );
        assert!(raw.contains("connection_refused"));
        assert!(raw.contains("broken_pipe"));
        assert!(raw.contains("transport.cancelled"));
        assert_eq!(rows.last().unwrap()["axes"]["operation"], "cancelled");
        assert!(rows.last().unwrap()["diagnostic"].is_null());
        assert_eq!(writer.snapshot().written, rows.len() as u64);
        assert_eq!(
            source
                .rpc_failure(&tonic::Status::cancelled("SECRET_TOKEN"))
                .submission,
            DiagnosticSubmission::Closed
        );
        std::fs::remove_dir_all(&directory).unwrap();
    }

    #[test]
    fn delivery_progress_survives_cancel_without_business_claim() {
        let mut progress = DeliveryProgress::default();
        progress.wrote(7);
        progress.timed_out();
        progress.wrote(99);
        progress.local_write_complete();
        let axes = serde_json::to_value(progress.axes()).unwrap();
        assert_eq!(axes["bytes_written"], 7);
        assert_eq!(axes["delivery"], "timed_out");
        assert_eq!(axes["business"], "unknown");
        assert_eq!(axes["physical"], "unknown");
        assert_eq!(axes["cleanup"], "unknown");
        let mut complete = DeliveryProgress::default();
        complete.wrote(9);
        complete.local_write_complete();
        // A later shutdown must not erase successful local write facts.
        complete.cancelled();
        let axes = serde_json::to_value(complete.axes()).unwrap();
        assert_eq!(axes["delivery"], "local_write_complete");
        assert_eq!(axes["business"], "unknown");
    }
}