polyc-query 2026.9.0

//! Durable per-read audit record.
//!
//! Every read the query layer serves leaves a durable audit record, whatever
//! its mechanism. [`ReadAuditRecord`] is that record's Rust shape, aligned
//! with the wire message it encodes to, `polychrome.events.v1.ReadAuditEvent`
//! (`crates/proto/proto/events.proto`, kind `read_audit`) —
//! [`ReadAuditRecord::into_event_payload`] is the encode step, and
//! [`ReadAuditRecord::kind`] names the eventlog `kind` string an
//! [`polyc_eventlog::Event`] carrying that payload uses. This module owns the
//! record shape only; appending it to the dedicated `query-audit` partition is
//! a control-plane concern — no persistence lives here.
//!
//! # Two records per read, not one
//!
//! A single post-hoc audit append (execute, THEN append, log-and-continue on
//! failure) leaves a gap: a crash between returning a result and appending its
//! audit record — or an append that simply fails — lets a read return
//! sensitive data with no durable trace at all. [`ReadAuditRecord::intent`]
//! and [`ReadAuditRecord::completion`] build the two correlated halves of the
//! fix instead: an INTENT record, minted and appended BEFORE the read runs
//! (the control plane's centralized lifecycle helper fails closed — refuses to
//! run the read at all — when this append fails), and a COMPLETION record,
//! appended after the read returns, correlated back to its intent by
//! [`ReadAuditRecord::read_id`]. An intent with no matching completion is
//! itself the durable evidence of an indeterminate read.
//!
//! # Why the detail is a union
//!
//! This audit surface began SQL-shaped: `sql`, `row_count` and
//! `skipped_partitions` sat directly in the envelope, so a read that is not a
//! statement had nowhere to describe itself. [`ReadDetail`] is the
//! extensibility point that was missing. Everything common to every read stays
//! in the envelope; everything mechanism-specific lives in the variant, and
//! the wire `operation` field is derived from which variant is present rather
//! than stored beside it — so the two cannot disagree.

use std::time::SystemTime;

use buffa::Message as _;
use polyc_proto::proto::polychrome::events::v1::{
    ConversationSearchDetail, QuerySurface, ReadAuditEvent, ReadAuditOutcome, ReadAuditPhase,
    ReadOperationKind, SearchHitFetchDetail, SqlReadDetail, read_audit_event,
};

/// Who issued a read and on whose behalf — the envelope fields that mean the
/// same thing for every read, whatever its mechanism.
///
/// Grouped into one value rather than passed as five loose arguments because
/// they travel together unchanged from a read's intent to its completion:
/// building both halves from the same [`ReadAuditContext`] is what makes them
/// agree by construction.
#[derive(Debug, Clone)]
pub struct ReadAuditContext {
    /// The caller's identity, where one exists. `None` when the surface has
    /// none.
    pub caller_identity: Option<String>,
    /// The calling conversation, set on an agent-tool read. `None` on a fleet
    /// read.
    pub conversation_id: Option<String>,
    /// The calling turn, alongside `conversation_id`. `None` on a fleet read,
    /// and on an explorer-page read — those carry `web_session_id` instead.
    pub turn_id: Option<String>,
    /// The calling web session, alongside `conversation_id`, set only for an
    /// explorer-page read. Never set together with `turn_id`: a scoped session
    /// is built for exactly one subject, so a durable audit record never
    /// fabricates a turn id for a session that has none.
    pub web_session_id: Option<String>,
    /// Which registration issued the read.
    pub surface: QuerySurface,
}

/// The SQL half of [`ReadDetail`].
#[derive(Debug, Clone, Default)]
pub struct SqlDetail {
    /// The SQL text as submitted, verbatim.
    pub sql: String,
    /// Mirrors `crate::output::QueryResultJson::truncated`. Meaningful only on
    /// a successful completion.
    pub truncated: bool,
    /// Mirrors `crate::output::QueryResultJson::skipped_partitions`.
    /// Meaningful only on a successful completion.
    pub skipped_partitions: u32,
    /// The result's row count. Meaningful only on a successful completion.
    pub row_count: u64,
}

/// The conversation-search half of [`ReadDetail`].
///
/// Carries the canonical query text deliberately rather than a hash: for a
/// tool whose purpose is moving content between conversations, "what was
/// searched for" is the single most valuable thing an auditor can ask. Never
/// carries snippets or returned message text.
#[derive(Debug, Clone, Default)]
pub struct SearchDetail {
    /// The query text as the trusted side canonicalized it, verbatim.
    pub canonical_query: String,
    /// BLAKE3 over the canonical authorized conversation set, lower-hex — the
    /// same digest bound into the approved call, so an auditor can tell
    /// whether a search ran against the scope its approver saw.
    pub scope_hash: String,
    /// How many conversations were in that authorized set.
    pub scope_count: u32,
    /// False when any in-scope partition was not covered through its last
    /// committed turn boundary.
    pub coverage_complete: bool,
    /// How many hits were returned. Zero on any non-success outcome.
    pub hit_count: u32,
    /// How many partitions were actually read after the membership prefilter.
    pub partitions_read: u32,
}

/// The search-hit-fetch half of [`ReadDetail`].
///
/// Never carries snippets or returned message text.
#[derive(Debug, Clone, Default)]
pub struct FetchDetail {
    /// [`ReadAuditRecord::read_id`] of the search this hit came from, so a
    /// fetch is always traceable to the approved search that produced its
    /// handle.
    pub origin_read_id: String,
    /// BLAKE3 over the hit handle, lower-hex — a stable reference to which hit
    /// was read, without reproducing the handle's own claims in the audit row.
    pub handle_hash: String,
    /// True when the returned text was cut off at the byte cap.
    pub truncated: bool,
    /// How many bytes of message text were returned.
    pub byte_count: u64,
}

/// The mechanism-specific half of a [`ReadAuditRecord`].
#[derive(Debug, Clone)]
pub enum ReadDetail {
    /// A SQL statement through the `DataFusion` engine.
    Sql(SqlDetail),
    /// A participation-scoped search across the conversations the initiating
    /// persona took part in.
    ConversationSearch(SearchDetail),
    /// A targeted read of one hit returned by an earlier search.
    SearchHitFetch(FetchDetail),
}

impl ReadDetail {
    /// Which operation this variant describes.
    ///
    /// Derived rather than stored, so the wire message's `operation` field can
    /// never name a different operation than the populated `detail` variant.
    #[must_use]
    pub const fn operation(&self) -> ReadOperationKind {
        match *self {
            Self::Sql(_) => ReadOperationKind::Sql,
            Self::ConversationSearch(_) => ReadOperationKind::ConversationSearch,
            Self::SearchHitFetch(_) => ReadOperationKind::SearchHitFetch,
        }
    }

    /// This detail with every field a read cannot know until it finishes
    /// zeroed, leaving only what is already true when the intent is minted.
    ///
    /// [`ReadAuditRecord::intent`] applies this itself rather than trusting
    /// the caller to pass a half-filled detail: an intent record carrying a
    /// row count or a hit count would be claiming an outcome it cannot have
    /// observed, and the two-phase lifecycle exists precisely so that an
    /// unmatched intent means "outcome unknown".
    #[must_use]
    fn intent_only(self) -> Self {
        match self {
            Self::Sql(detail) => Self::Sql(SqlDetail {
                sql: detail.sql,
                truncated: false,
                skipped_partitions: 0,
                row_count: 0,
            }),
            Self::ConversationSearch(detail) => Self::ConversationSearch(SearchDetail {
                canonical_query: detail.canonical_query,
                scope_hash: detail.scope_hash,
                scope_count: detail.scope_count,
                coverage_complete: false,
                hit_count: 0,
                partitions_read: 0,
            }),
            Self::SearchHitFetch(detail) => Self::SearchHitFetch(FetchDetail {
                origin_read_id: detail.origin_read_id,
                handle_hash: detail.handle_hash,
                truncated: false,
                byte_count: 0,
            }),
        }
    }
}

impl From<&SqlDetail> for SqlReadDetail {
    fn from(detail: &SqlDetail) -> Self {
        Self {
            sql: detail.sql.clone(),
            truncated: detail.truncated,
            skipped_partitions: detail.skipped_partitions,
            row_count: detail.row_count,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl From<&SearchDetail> for ConversationSearchDetail {
    fn from(detail: &SearchDetail) -> Self {
        Self {
            canonical_query: detail.canonical_query.clone(),
            scope_hash: detail.scope_hash.clone(),
            scope_count: detail.scope_count,
            coverage_complete: detail.coverage_complete,
            hit_count: detail.hit_count,
            partitions_read: detail.partitions_read,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl From<&FetchDetail> for SearchHitFetchDetail {
    fn from(detail: &FetchDetail) -> Self {
        Self {
            origin_read_id: detail.origin_read_id.clone(),
            handle_hash: detail.handle_hash.clone(),
            truncated: detail.truncated,
            byte_count: detail.byte_count,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl From<&ReadDetail> for read_audit_event::Detail {
    fn from(detail: &ReadDetail) -> Self {
        match *detail {
            ReadDetail::Sql(ref inner) => Self::Sql(Box::new(inner.into())),
            ReadDetail::ConversationSearch(ref inner) => {
                Self::ConversationSearch(Box::new(inner.into()))
            }
            ReadDetail::SearchHitFetch(ref inner) => Self::SearchHitFetch(Box::new(inner.into())),
        }
    }
}

/// One durable audit record for a single read.
///
/// Either the INTENT half or the COMPLETION half of the two-phase lifecycle
/// (see the module doc). [`ReadAuditRecord::intent`] and
/// [`ReadAuditRecord::completion`] are the two constructors, and the fields
/// are private, so a caller cannot build a record without going through one of
/// them and picking a phase.
#[derive(Debug, Clone)]
pub struct ReadAuditRecord {
    read_id: String,
    context: ReadAuditContext,
    detail: ReadDetail,
    timestamp: SystemTime,
    phase: ReadAuditPhase,
    outcome: ReadAuditOutcome,
    duration_ms: u64,
}

impl ReadAuditRecord {
    /// Build the INTENT half of a read's audit record — appended BEFORE the
    /// read runs.
    ///
    /// Carries no outcome, and no completeness field inside `detail`:
    /// `intent_only` zeroes those here rather than trusting the caller to have
    /// left them alone.
    #[must_use]
    pub fn intent(
        read_id: String,
        context: ReadAuditContext,
        detail: ReadDetail,
        timestamp: SystemTime,
    ) -> Self {
        Self {
            read_id,
            context,
            detail: detail.intent_only(),
            timestamp,
            phase: ReadAuditPhase::Intent,
            outcome: ReadAuditOutcome::Unspecified,
            duration_ms: 0,
        }
    }

    /// Build the COMPLETION half of a read's audit record — appended after the
    /// read returns, correlated to its intent by `read_id` (the SAME id
    /// [`ReadAuditRecord::intent`] minted for this read).
    #[must_use]
    pub const fn completion(
        read_id: String,
        context: ReadAuditContext,
        detail: ReadDetail,
        timestamp: SystemTime,
        outcome: ReadAuditOutcome,
        duration_ms: u64,
    ) -> Self {
        Self {
            read_id,
            context,
            detail,
            timestamp,
            phase: ReadAuditPhase::Completion,
            outcome,
            duration_ms,
        }
    }

    /// The id correlating this record with the other half of its read.
    #[must_use]
    pub fn read_id(&self) -> &str {
        &self.read_id
    }

    /// Which half of the two-phase lifecycle this record is.
    #[must_use]
    pub const fn phase(&self) -> ReadAuditPhase {
        self.phase
    }

    /// How the read resolved. Always [`ReadAuditOutcome::Unspecified`] on an
    /// intent record.
    #[must_use]
    pub const fn outcome(&self) -> ReadAuditOutcome {
        self.outcome
    }

    /// The mechanism-specific half of this record.
    #[must_use]
    pub const fn detail(&self) -> &ReadDetail {
        &self.detail
    }

    /// Who issued this read and on whose behalf.
    #[must_use]
    pub const fn context(&self) -> &ReadAuditContext {
        &self.context
    }

    /// The eventlog `kind` string every [`ReadAuditRecord`] is appended under
    /// — `polyc_proto::kinds::READ_AUDIT`, bare (no `:{turn_uuid}` suffix):
    /// the proto message already carries `conversation_id`/`turn_id` as
    /// fields, so unlike per-turn kinds (`usage`, `user_msg`, ...) this kind
    /// needs no tag to correlate a record back to its turn. Both halves of the
    /// lifecycle share this kind and partition — [`ReadAuditRecord::phase`] is
    /// what a reader distinguishes them by, not the kind string.
    #[must_use]
    pub const fn kind() -> &'static str {
        polyc_proto::kinds::READ_AUDIT
    }

    /// Encode this record to `polychrome.events.v1.ReadAuditEvent` wire bytes
    /// — the payload an [`polyc_eventlog::Event`] of kind
    /// [`ReadAuditRecord::kind`] carries.
    ///
    /// The optional context fields map `None` to proto3's empty-string
    /// default, matching the wire message's own "empty until it applies"
    /// contract for each. `timestamp` converts to Unix milliseconds,
    /// saturating at `u64::MAX` in the — practically unreachable — case of a
    /// timestamp before the Unix epoch or past the year 584 million, rather
    /// than panicking on a clock anomaly.
    ///
    /// Named `into_*` despite taking `&self`, not `self`, because encoding a
    /// durable audit record to bytes is naturally a read of it, not a
    /// consuming conversion — the record is typically still needed afterward
    /// (e.g. for logging) alongside its encoded form.
    #[must_use]
    #[allow(
        clippy::wrong_self_convention,
        reason = "encode is non-consuming by design; see the doc comment above"
    )]
    pub fn into_event_payload(&self) -> Vec<u8> {
        ReadAuditEvent {
            read_id: self.read_id.clone(),
            phase: self.phase.into(),
            operation: self.detail.operation().into(),
            caller_identity: self.context.caller_identity.clone().unwrap_or_default(),
            conversation_id: self.context.conversation_id.clone().unwrap_or_default(),
            turn_id: self.context.turn_id.clone().unwrap_or_default(),
            web_session_id: self.context.web_session_id.clone().unwrap_or_default(),
            surface: self.context.surface.into(),
            recorded_at_ms: epoch_millis(self.timestamp),
            outcome: self.outcome.into(),
            duration_ms: self.duration_ms,
            detail: Some((&self.detail).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
        .encode_to_vec()
    }
}

/// `timestamp` as Unix milliseconds, saturating rather than panicking on the
/// clock anomalies `SystemTime::duration_since` and a lossy `u128 -> u64`
/// narrowing could otherwise surface: a `timestamp` before the Unix epoch
/// saturates to `0`; one whose millisecond count overflows `u64` (year 584
/// million or later) saturates to `u64::MAX`.
fn epoch_millis(timestamp: SystemTime) -> u64 {
    timestamp
        .duration_since(SystemTime::UNIX_EPOCH)
        .map_or(0, |elapsed| {
            u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX)
        })
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use polyc_proto::events_decode::try_decode_event_payload;

    use super::*;

    const FIXED: Duration = Duration::from_mins(29_200_000);

    fn fleet_context() -> ReadAuditContext {
        ReadAuditContext {
            caller_identity: None,
            conversation_id: None,
            turn_id: None,
            web_session_id: None,
            surface: QuerySurface::Http,
        }
    }

    fn sql_detail() -> ReadDetail {
        ReadDetail::Sql(SqlDetail {
            sql: "SELECT 1".to_string(),
            truncated: false,
            skipped_partitions: 0,
            row_count: 0,
        })
    }

    fn sample() -> ReadAuditRecord {
        ReadAuditRecord::intent(
            "rid-1".to_string(),
            fleet_context(),
            sql_detail(),
            SystemTime::UNIX_EPOCH + FIXED,
        )
    }

    fn decode(record: &ReadAuditRecord) -> ReadAuditEvent {
        try_decode_event_payload::<ReadAuditEvent>(&record.into_event_payload())
            .expect("a freshly encoded ReadAuditEvent must decode")
    }

    #[test]
    fn kind_is_the_registered_read_audit_constant() {
        assert_eq!(ReadAuditRecord::kind(), polyc_proto::kinds::READ_AUDIT);
    }

    #[test]
    fn into_event_payload_round_trips_through_the_real_proto_decoder() {
        let decoded = decode(&sample());

        assert_eq!(decoded.read_id, "rid-1");
        assert_eq!(decoded.caller_identity, "");
        assert_eq!(decoded.conversation_id, "");
        assert_eq!(decoded.turn_id, "");
        assert_eq!(decoded.recorded_at_ms, 1_752_000_000_000);
        assert_eq!(decoded.surface, buffa::EnumValue::from(QuerySurface::Http));
        assert_eq!(
            decoded.phase,
            buffa::EnumValue::from(ReadAuditPhase::Intent)
        );
        assert_eq!(
            decoded.outcome,
            buffa::EnumValue::from(ReadAuditOutcome::Unspecified)
        );
        assert_eq!(
            decoded.operation,
            buffa::EnumValue::from(ReadOperationKind::Sql)
        );
        assert!(matches!(
            decoded.detail,
            Some(read_audit_event::Detail::Sql(ref sql)) if sql.sql == "SELECT 1"
        ));
    }

    #[test]
    fn none_identity_fields_encode_to_proto3_empty_string_not_absence() {
        let record = ReadAuditRecord::intent(
            "rid-2".to_string(),
            ReadAuditContext {
                caller_identity: Some("persona-42".to_string()),
                conversation_id: Some("conv-9".to_string()),
                turn_id: Some("turn-3".to_string()),
                web_session_id: Some("web-7".to_string()),
                surface: QuerySurface::Http,
            },
            sql_detail(),
            SystemTime::UNIX_EPOCH + FIXED,
        );
        let decoded = decode(&record);

        assert_eq!(decoded.caller_identity, "persona-42");
        assert_eq!(decoded.conversation_id, "conv-9");
        assert_eq!(decoded.turn_id, "turn-3");
        assert_eq!(decoded.web_session_id, "web-7");
    }

    #[test]
    fn timestamp_before_epoch_saturates_to_zero_instead_of_panicking() {
        let record = ReadAuditRecord::intent(
            "rid-3".to_string(),
            fleet_context(),
            sql_detail(),
            SystemTime::UNIX_EPOCH - Duration::from_secs(1),
        );

        assert_eq!(decode(&record).recorded_at_ms, 0);
    }

    /// A completion carries the same `read_id` as its matching intent, plus
    /// the outcome and completeness fields an intent never sets.
    #[test]
    fn completion_carries_its_outcome_and_completeness_fields() {
        let record = ReadAuditRecord::completion(
            "rid-4".to_string(),
            fleet_context(),
            ReadDetail::Sql(SqlDetail {
                sql: "SELECT 1".to_string(),
                truncated: true,
                skipped_partitions: 2,
                row_count: 9,
            }),
            SystemTime::UNIX_EPOCH + FIXED,
            ReadAuditOutcome::Success,
            123,
        );
        let decoded = decode(&record);

        assert_eq!(decoded.read_id, "rid-4");
        assert_eq!(
            decoded.phase,
            buffa::EnumValue::from(ReadAuditPhase::Completion)
        );
        assert_eq!(
            decoded.outcome,
            buffa::EnumValue::from(ReadAuditOutcome::Success)
        );
        assert_eq!(decoded.duration_ms, 123);
        let Some(read_audit_event::Detail::Sql(sql)) = decoded.detail else {
            panic!("a SQL read must encode a SQL detail");
        };
        assert!(sql.truncated);
        assert_eq!(sql.skipped_partitions, 2);
        assert_eq!(sql.row_count, 9);
    }

    /// Both halves share the SAME `kind()` and partition — a reader
    /// distinguishes them by `phase`, never by a different eventlog `kind`.
    #[test]
    fn intent_and_completion_share_the_same_kind() {
        let intent = sample();
        let completion = ReadAuditRecord::completion(
            "rid-1".to_string(),
            fleet_context(),
            sql_detail(),
            SystemTime::UNIX_EPOCH + FIXED,
            ReadAuditOutcome::Success,
            0,
        );

        assert_eq!(ReadAuditRecord::kind(), polyc_proto::kinds::READ_AUDIT);
        assert_eq!(intent.read_id(), completion.read_id());
        assert_eq!(intent.phase(), ReadAuditPhase::Intent);
        assert_eq!(completion.phase(), ReadAuditPhase::Completion);
    }

    /// The wire `operation` is derived from the populated `detail` variant, so
    /// a record can never claim one operation while carrying another's detail.
    #[test]
    fn operation_is_derived_from_the_detail_variant() {
        for (detail, expected) in [
            (sql_detail(), ReadOperationKind::Sql),
            (
                ReadDetail::ConversationSearch(SearchDetail::default()),
                ReadOperationKind::ConversationSearch,
            ),
            (
                ReadDetail::SearchHitFetch(FetchDetail::default()),
                ReadOperationKind::SearchHitFetch,
            ),
        ] {
            let record = ReadAuditRecord::intent(
                "rid-op".to_string(),
                fleet_context(),
                detail,
                SystemTime::UNIX_EPOCH + FIXED,
            );

            assert_eq!(
                decode(&record).operation,
                buffa::EnumValue::from(expected),
                "the wire operation must name the populated detail variant"
            );
        }
    }

    /// An intent can never claim an outcome it has not observed: the
    /// constructor zeroes every completion-only field, whatever the caller
    /// passed.
    #[test]
    fn intent_zeroes_completion_only_fields_the_caller_left_set() {
        let record = ReadAuditRecord::intent(
            "rid-5".to_string(),
            fleet_context(),
            ReadDetail::ConversationSearch(SearchDetail {
                canonical_query: "where did we decide the timeout".to_string(),
                scope_hash: "ab".repeat(32),
                scope_count: 12,
                coverage_complete: true,
                hit_count: 7,
                partitions_read: 3,
            }),
            SystemTime::UNIX_EPOCH + FIXED,
        );

        let Some(read_audit_event::Detail::ConversationSearch(search)) = decode(&record).detail
        else {
            panic!("a conversation search must encode a search detail");
        };
        assert_eq!(search.canonical_query, "where did we decide the timeout");
        assert_eq!(
            search.scope_count, 12,
            "scope is known before the read runs"
        );
        assert_eq!(search.hit_count, 0, "an intent has observed no hits");
        assert_eq!(search.partitions_read, 0, "an intent has read nothing");
        assert!(
            !search.coverage_complete,
            "an intent has established no coverage"
        );
    }

    /// A fetch detail names the search that produced its handle, and never the
    /// text that came back.
    #[test]
    fn fetch_detail_records_its_origin_search_and_byte_count() {
        let record = ReadAuditRecord::completion(
            "rid-6".to_string(),
            fleet_context(),
            ReadDetail::SearchHitFetch(FetchDetail {
                origin_read_id: "rid-4".to_string(),
                handle_hash: "cd".repeat(32),
                truncated: true,
                byte_count: 8_000,
            }),
            SystemTime::UNIX_EPOCH + FIXED,
            ReadAuditOutcome::Success,
            5,
        );

        let Some(read_audit_event::Detail::SearchHitFetch(fetch)) = decode(&record).detail else {
            panic!("a hit fetch must encode a fetch detail");
        };
        assert_eq!(fetch.origin_read_id, "rid-4");
        assert_eq!(fetch.handle_hash, "cd".repeat(32));
        assert!(fetch.truncated);
        assert_eq!(fetch.byte_count, 8_000);
    }
}