polyc_eventlog_model/

event.rs

//! The [`Event`] item stored in a journal and its [`commonware_codec`] wiring.
//!
//! An [`Event`] is the unit the journal persists: a `kind` discriminator plus
//! an opaque `payload`. This crate is deliberately **payload-agnostic** — the
//! payload is a `Vec<u8>` of buffa-encoded bytes whose concrete schema lives in
//! `polyc-proto`'s `events.proto`. The event log neither encodes nor
//! interprets it; it round-trips the bytes verbatim.
//!
//! The journal primitive (`commonware_storage::journal::contiguous::variable`)
//! requires its item type to implement [`commonware_codec`]'s [`Write`],
//! [`EncodeSize`], and [`Read`] traits. [`Event`] implements them by length-
//! prefixing the `kind` UTF-8 bytes and the `payload` bytes, exactly the
//! variable-size pattern documented in the `commonware-codec` crate.

use bytes::{Buf, BufMut};
use commonware_codec::{
    EncodeSize, Error as CodecError, Read, ReadExt as _, ReadRangeExt as _, Write,
};

use crate::taint::TrustTag;

/// A single conversation event: a `kind` tag, a [`TrustTag`] provenance
/// capability, and an opaque encoded payload.
///
/// `kind` is a short string discriminator (e.g. `"user_msg"`,
/// `"planner_decision"`, `"tool_call"`). `payload` is opaque to this crate —
/// it is the buffa-encoded body of whichever `events.proto` message the `kind`
/// names, stored and replayed byte-for-byte. `trust` is the CaMeL-style trust
/// tag assigned at ingress (see [`TrustTag`]); it travels with the event
/// through the durable log so a data-flow policy can reason over provenance
/// without re-decoding payloads.
///
/// The position (turn/seq ordering) is **not** carried in the event itself: the
/// journal assigns each appended event a monotonically increasing position, and
/// the physical journal adapter yields events in that append order. See the
/// adapter's documentation for the ordering contract.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Event {
    /// Discriminator naming the payload's schema (e.g. `"tool_call"`).
    pub kind: String,
    /// Provenance/trust capability assigned at ingress.
    pub trust: TrustTag,
    /// Opaque, buffa-encoded payload bytes. Round-tripped verbatim.
    pub payload: Vec<u8>,
}

impl Event {
    /// Construct an unclassified ([`TrustTag::Unspecified`]) event from a
    /// `kind` and an owned payload — for control/marker events that carry no
    /// external content provenance.
    #[must_use]
    pub fn new(kind: impl Into<String>, payload: Vec<u8>) -> Self {
        Self::with_trust(kind, payload, TrustTag::Unspecified)
    }

    /// Construct a [`TrustTag::TrustedUser`] event — an authenticated
    /// principal's own message.
    #[must_use]
    pub fn trusted(kind: impl Into<String>, payload: Vec<u8>) -> Self {
        Self::with_trust(kind, payload, TrustTag::TrustedUser)
    }

    /// Construct a [`TrustTag::QuarantinedContent`] event — tool output,
    /// fetched content, or otherwise untrusted inbound content.
    #[must_use]
    pub fn quarantined(kind: impl Into<String>, payload: Vec<u8>) -> Self {
        Self::with_trust(kind, payload, TrustTag::QuarantinedContent)
    }

    /// Construct an event with an explicit [`TrustTag`].
    #[must_use]
    pub fn with_trust(kind: impl Into<String>, payload: Vec<u8>, trust: TrustTag) -> Self {
        Self {
            kind: kind.into(),
            trust,
            payload,
        }
    }
}

/// Decode-time bounds for an [`Event`], supplied as the journal's
/// `codec_config`.
///
/// `commonware-codec` requires an explicit maximum length when reading any
/// variable-length field, so that a corrupt or hostile length prefix cannot
/// trigger an unbounded allocation. These caps are applied when decoding the
/// `kind` and `payload` of each event during a physical journal replay.
///
/// [`EventCfg::DEFAULT`] provides generous defaults suitable for conversation
/// events; tighten them per deployment if desired.
#[derive(Debug, Clone, Copy)]
pub struct EventCfg {
    /// Maximum byte length permitted for a decoded `kind` string.
    pub max_kind_len: usize,
    /// Maximum byte length permitted for a decoded `payload`.
    pub max_payload_len: usize,
}

impl EventCfg {
    /// Default decode bounds: 256-byte `kind`, 16 MiB `payload`.
    pub const DEFAULT: Self = Self {
        max_kind_len: 256,
        max_payload_len: 16 * 1024 * 1024,
    };
}

impl Default for EventCfg {
    fn default() -> Self {
        Self::DEFAULT
    }
}

impl Write for Event {
    fn write(&self, buf: &mut impl BufMut) {
        // Forward-compatible layout: the original fields (`kind` UTF-8 bytes as a
        // length-prefixed `Vec<u8>`, then the `payload`) are written FIRST, in
        // their original order and encoding, and the trust discriminant byte is
        // APPENDED last. Keeping the trust byte at the tail — rather than
        // prepending it — means a record written before the trust field existed
        // (kind + payload, no trailing byte) is still a valid prefix of this
        // layout: the journal frames each item, so on read "no bytes left after
        // payload" is unambiguously the absence of a trust byte (→ Unspecified).
        // Deploying the trust substrate therefore does not invalidate a single
        // pre-existing event-log record.
        self.kind.as_bytes().to_vec().write(buf);
        self.payload.write(buf);
        self.trust.as_u8().write(buf);
    }
}

impl EncodeSize for Event {
    fn encode_size(&self) -> usize {
        self.trust.as_u8().encode_size()
            + self.kind.as_bytes().to_vec().encode_size()
            + self.payload.encode_size()
    }
}

impl Read for Event {
    type Cfg = EventCfg;

    fn read_cfg(buf: &mut impl Buf, cfg: &EventCfg) -> Result<Self, CodecError> {
        let kind_bytes = <Vec<u8>>::read_range(buf, 0..=cfg.max_kind_len)?;
        let kind = String::from_utf8(kind_bytes)
            .map_err(|_| CodecError::Invalid("Event", "kind is not valid UTF-8"))?;
        let payload = <Vec<u8>>::read_range(buf, 0..=cfg.max_payload_len)?;
        // Forward-compatible trust tag: the journal frames each item, so any
        // bytes remaining after the payload are the trailing trust discriminant
        // written by the current layout. A record with none left predates the
        // trust field and is read as `Unspecified` — NOT a decode failure — so
        // existing logs survive the upgrade. A present-but-unknown byte is still
        // corruption (`from_u8` rejects it); `Decode` enforces that exactly one
        // trailing byte was consumed (any extra is `ExtraData`).
        let trust = if buf.has_remaining() {
            TrustTag::from_u8(u8::read(buf)?)
                .ok_or(CodecError::Invalid("Event", "unknown trust tag"))?
        } else {
            TrustTag::Unspecified
        };
        Ok(Self {
            kind,
            trust,
            payload,
        })
    }
}

#[cfg(test)]
mod tests {
    use super::{Event, EventCfg};
    use crate::taint::TrustTag;
    use bytes::BytesMut;
    use commonware_codec::{Decode as _, Encode as _, Write as _};

    // Deploy-safety: a record written in the OLD on-disk layout — `kind` then
    // `payload`, with NO trailing trust byte (every record that predates the
    // trust field) — must still decode, classified as `TrustTag::Unspecified`.
    // The journal frames each item, so the decoder sees "no bytes remain after
    // payload" and reads the absence as Unspecified rather than failing. This is
    // what stops deploying the trust substrate from wiping existing event logs.
    #[test]
    fn old_layout_without_trust_byte_decodes_as_unspecified() {
        // Reproduce the pre-trust-field encoding exactly: the `kind` UTF-8 bytes
        // as a length-prefixed `Vec<u8>`, then the `payload` — and nothing else.
        let mut buf = BytesMut::new();
        b"user_msg".to_vec().write(&mut buf);
        b"summarize my inbox".to_vec().write(&mut buf);
        let decoded =
            Event::decode_cfg(buf.freeze(), &EventCfg::DEFAULT).expect("old record must decode");
        assert_eq!(decoded.kind, "user_msg");
        assert_eq!(decoded.payload, b"summarize my inbox".to_vec());
        assert_eq!(
            decoded.trust,
            TrustTag::Unspecified,
            "a record with no trust byte must read as Unspecified"
        );
    }

    #[test]
    fn round_trips_through_codec() {
        let event = Event::new("tool_call", vec![1, 2, 3, 0xff, 0]);
        let bytes = event.encode();
        let decoded = Event::decode_cfg(bytes, &EventCfg::DEFAULT).expect("decode");
        assert_eq!(event, decoded);
    }

    #[test]
    fn trust_tag_survives_codec_round_trip() {
        for event in [
            Event::trusted("user_msg", b"hi".to_vec()),
            Event::quarantined("output_msg", b"<tool result>".to_vec()),
            Event::new("turn_start", Vec::new()),
        ] {
            let decoded = Event::decode_cfg(event.encode(), &EventCfg::DEFAULT).expect("decode");
            assert_eq!(decoded, event);
            assert_eq!(decoded.trust, event.trust);
        }
    }

    #[test]
    fn rejects_unknown_trust_tag_byte() {
        // The trust discriminant is the LAST byte of the encoding (trailing, for
        // backward-readability); a present value naming no known tag is
        // corruption and must fail to decode.
        let mut bytes = Event::trusted("user_msg", b"x".to_vec()).encode().to_vec();
        let last = bytes.len() - 1;
        bytes[last] = 0xff;
        assert!(Event::decode_cfg(&bytes[..], &EventCfg::DEFAULT).is_err());
    }

    #[test]
    fn round_trips_empty_payload() {
        let event = Event::new("user_msg", Vec::new());
        let bytes = event.encode();
        let decoded = Event::decode_cfg(bytes, &EventCfg::DEFAULT).expect("decode");
        assert_eq!(event, decoded);
        assert!(decoded.payload.is_empty());
    }

    #[test]
    fn rejects_payload_over_cap() {
        let event = Event::new("k", vec![0u8; 64]);
        let bytes = event.encode();
        let tight = EventCfg {
            max_kind_len: 256,
            max_payload_len: 8,
        };
        assert!(Event::decode_cfg(bytes, &tight).is_err());
    }
}