yeti_types/backend/

log.rs

//! Transaction log primitives — the unified durable record of every
//! committed write in a yeti deployment.
//!
//! ## Why one log
//!
//! Before this surface, audit and replication would have lived as
//! separate sinks: audit writing to its own tables, replication
//! writing to a separate per-CF append-only log. The Harper precedent
//! ([`harperfast/harper`]'s `RocksTransactionLogStore`) demonstrates
//! the unified-store pattern works in production: one log records
//! every commit, and both audit (range-by-time) and replication
//! (tail-by-HLC) read from it. Yeti adopts the same shape because:
//!
//! - One write per commit (vs. two) — half the I/O on the write path.
//! - Audit and replication can never disagree about what happened.
//! - Single retention policy fits both (a node retains the log only
//!   as long as its peers might need it for catch-up; audit needs
//!   the same window).
//! - New observers (custom audit consumers, real-time dashboards)
//!   plug into the existing log instead of bolting on parallel
//!   sinks.
//!
//! ## Position in the stack
//!
//! - **Trait + entry shape**: this module, `yeti-types::backend::log`.
//!   Sits alongside the sibling `observer` module
//!   ([`TransactionObserver`](crate::backend::TransactionObserver) /
//!   [`TransactionEvent`](crate::backend::TransactionEvent)) — the
//!   trio that defines the post-commit observation surface.
//! - **Default implementation**: `yeti-store::transaction_log::KvTransactionLog`.
//!   RocksDB-backed (or any `KvBackend`). Opens in its own
//!   per-deployment directory so the log's atomicity is local
//!   (data committed but log missing is a single-machine failure
//!   mode, not a cross-system one).
//! - **Commit-path integration**: future slice. The log is built
//!   here as a standalone primitive so its contract can be reviewed
//!   and tested before deciding how to intercept writes
//!   (`LoggingBackend` wrapper vs. explicit hook surface on
//!   [`BackendManager`](crate::backend::BackendManager)).
//!
//! ## Wire stability
//!
//! [`LogEntry`] is also the wire frame: the replication transport
//! streams `LogEntry`s between peers. The shape is therefore stable
//! across the cluster — adding a field requires a coordinated
//! deployment. Use `Option<...>` for new optional fields so that
//! older readers continue to parse newer entries.
//!
//! [`harperfast/harper`]: https://github.com/harperdb

use std::sync::Arc;

use async_trait::async_trait;
use serde::{Deserialize, Serialize};

use crate::error::Result;
use crate::hlc::HlcTimestamp;
use crate::types::TableId;

use super::WriteOp;

// ============================================================================
// Originator
// ============================================================================

/// Identity of the party that initiated a write.
///
/// Captured on every [`LogEntry`] so audit queries can answer
/// "who/what wrote this record" without a sidecar table, and
/// replication can distinguish between locally-originated writes
/// (which must be streamed to peers) and remotely-applied writes
/// (which were received over the wire and must NOT be re-streamed).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum Originator {
    /// Locally-initiated write through the request pipeline. `user`
    /// is the authenticated principal when available; `None` for
    /// anonymous writes (e.g., public tables with
    /// `@export(public: [create])`).
    Local {
        /// Authenticated username, if the write went through the auth
        /// pipeline.
        user: Option<String>,
    },
    /// Write applied locally in response to a peer's replication
    /// stream. Carries the source node id so replication's outgoing
    /// log writer can filter out these entries — we don't re-stream
    /// what we received.
    Replicated {
        /// Node id of the source peer that originated the write.
        from_node: String,
    },
    /// Internal write performed by the platform itself — startup
    /// bootstrap, system-table maintenance, scheduled compaction.
    /// `component` is a short label for diagnostics
    /// (`"auth-bootstrap"`, `"ttl-sweep"`, etc.).
    Internal {
        /// Name of the platform component that issued the write.
        component: String,
    },
}

impl Originator {
    /// Convenience constructor for anonymous local writes.
    #[must_use]
    pub const fn anonymous() -> Self {
        Self::Local { user: None }
    }

    /// Convenience constructor for authenticated local writes.
    #[must_use]
    pub fn local(user: impl Into<String>) -> Self {
        Self::Local {
            user: Some(user.into()),
        }
    }

    /// True if this write was received over the replication wire
    /// (i.e., should not be re-streamed). Used by the outgoing-log
    /// writer to filter out apply-from-peer writes.
    #[must_use]
    pub const fn is_replicated(&self) -> bool {
        matches!(self, Self::Replicated { .. })
    }
}

// ============================================================================
// LogEntry
// ============================================================================

/// A single durable record of a committed write.
///
/// Persisted to the per-deployment [`TransactionLog`] at commit time.
/// Both audit and replication observers consume from this log:
///
/// - Replication tail-reads by HLC: streams new entries to peers,
///   filtering out [`Originator::Replicated`] writes so applied-
///   from-peer entries don't echo back.
/// - Audit range-reads by HLC + filters by table / originator / user
///   for query responses.
///
/// Wire-stable (serde-derived) because [`LogEntry`] frames also flow
/// across the replication network protocol between nodes.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LogEntry {
    /// HLC timestamp assigned at commit time. Strictly greater than
    /// any previously-appended entry's HLC on the same node.
    pub hlc: HlcTimestamp,

    /// Node id of the node that PERSISTED this entry (the node that
    /// observed the commit). Both locally-originated and replication-
    /// applied writes log here, so `node_id` is always the local
    /// node. Distinguished from [`Originator::Replicated::from_node`]
    /// which records the conceptual writer.
    pub node_id: String,

    /// In-deployment table identity.
    pub table: TableId,

    /// The write operation that was committed.
    pub op: WriteOp,

    /// HLC of the existing record at this key prior to the write, if
    /// any. `None` for inserts. Used by anti-entropy to
    /// record causal predecessor relationships for the receive-side
    /// HLC-LWW comparison.
    pub prev_hlc: Option<HlcTimestamp>,

    /// Raw bytes of the existing record at this key prior to the
    /// write, if the table opted into `@audit(capture_state: true)`
    /// and the `LoggingBackend` was configured with
    /// `capture_before: true`. `None` when capture was disabled OR
    /// the key didn't exist (insert path). Decoded as JSON by audit
    /// readers to surface as `AuditEntry.before`.
    ///
    /// Storage cost is significant — each captured write doubles
    /// the log entry size plus pays one extra `get` at write time
    /// — so this is opt-in per table via the `@audit` directive.
    ///
    /// Note: encoded inline (not `skip_serializing_if`) because the
    /// wire format is rmp-serde tuple-positional; skipping fields
    /// breaks the decoder's positional expectations.
    pub prev_value: Option<Vec<u8>>,

    /// Who / what initiated the write.
    pub originator: Originator,

    /// Optional correlation id for traceability — typically the
    /// request id from the HTTP / MQTT / gRPC layer.
    pub request_id: Option<String>,

    /// Which protocol / interface the write came in through — `"rest"`,
    /// `"graphql"`, `"mqtt"`, `"mcp"`, or `"internal"` for platform-
    /// originated writes. `None` when the originator couldn't be
    /// determined (background tasks, startup bootstrap that didn't
    /// set the task-local). Audit queries surface this for filtering.
    pub interface: Option<String>,
}

impl LogEntry {
    /// Serialize to `MessagePack` bytes for storage / wire.
    ///
    /// # Errors
    ///
    /// Returns [`crate::error::YetiError::Internal`] if encoding fails
    /// (in practice, impossible for fully-owned `LogEntry` values; the
    /// `Result` is preserved for forward-compatibility with future
    /// fields that may have encoding constraints).
    pub fn encode(&self) -> Result<Vec<u8>> {
        rmp_serde::to_vec(self)
            .map_err(|e| crate::error::YetiError::Internal(format!("LogEntry encode failed: {e}")))
    }

    /// Deserialize from `MessagePack` bytes.
    ///
    /// # Errors
    ///
    /// Returns [`crate::error::YetiError::Internal`] if the input is
    /// not a valid `LogEntry` encoding (truncation, version skew, or
    /// corruption on disk).
    pub fn decode(bytes: &[u8]) -> Result<Self> {
        rmp_serde::from_slice(bytes)
            .map_err(|e| crate::error::YetiError::Internal(format!("LogEntry decode failed: {e}")))
    }
}

// ============================================================================
// RetentionPolicy
// ============================================================================

/// Retention policy for the transaction log.
///
/// At least one bound must be set for [`TransactionLog::sweep`] to do
/// useful work; with both bounds set, sweep removes entries that
/// exceed *either* bound (logical OR — whichever is tighter wins).
///
/// Time-based sweep is the cheaper path: a single HLC threshold,
/// `delete_range`-friendly. Size-based sweep is more expensive because
/// it requires totaling on-disk bytes; expect implementations to
/// approximate (e.g., count entries and assume average size) rather
/// than exact byte accounting.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct RetentionPolicy {
    /// Remove entries with physical-time component older than
    /// `now - max_age_secs`. `None` disables age-based sweep.
    pub max_age_secs: Option<u64>,

    /// Trim the log to at most this many on-disk bytes by removing
    /// the oldest entries. `None` disables size-based sweep. Best-
    /// effort: implementations may overshoot by one entry rather
    /// than partial-truncate.
    pub max_size_bytes: Option<u64>,
}

impl RetentionPolicy {
    /// Time-only retention policy.
    #[must_use]
    pub const fn time_only(max_age_secs: u64) -> Self {
        Self {
            max_age_secs: Some(max_age_secs),
            max_size_bytes: None,
        }
    }

    /// Size-only retention policy.
    #[must_use]
    pub const fn size_only(max_size_bytes: u64) -> Self {
        Self {
            max_age_secs: None,
            max_size_bytes: Some(max_size_bytes),
        }
    }

    /// Combined retention — sweep removes entries that exceed
    /// **either** bound.
    #[must_use]
    pub const fn dual(max_age_secs: u64, max_size_bytes: u64) -> Self {
        Self {
            max_age_secs: Some(max_age_secs),
            max_size_bytes: Some(max_size_bytes),
        }
    }

    /// True if no bound is set; [`TransactionLog::sweep`] is a no-op
    /// in this case.
    #[must_use]
    pub const fn is_disabled(&self) -> bool {
        self.max_age_secs.is_none() && self.max_size_bytes.is_none()
    }
}

// ============================================================================
// TransactionLog trait
// ============================================================================

/// Per-deployment append-only log of every committed write.
///
/// One instance per deployment process; lives for the deployment's
/// lifetime. Entries are HLC-ordered and survive process restart
/// (durability is the foundational contract — replication's catch-
/// up window relies on it).
///
/// ## Contract
///
/// - [`append`](Self::append) is the durability sink. Returns only
///   after the entry has been fsynced (or the implementation's
///   chosen durability point); subsequent [`read_after`](Self::read_after)
///   calls on the same instance must see the new entry.
/// - [`read_after`](Self::read_after) returns entries strictly
///   greater than the given HLC, in ascending HLC order, up to
///   `limit`. Used by replication for streaming tail-reads and
///   catch-up on peer reconnection.
/// - [`sweep`](Self::sweep) removes entries that exceed the
///   retention bounds. Returns the count of removed entries. Sweep
///   is the only operation that DELETES from the log — append is
///   the only operation that ADDS.
///
/// ## Implementations
///
/// The canonical implementation is
/// `yeti_store::transaction_log::KvTransactionLog`, backed by a
/// dedicated [`KvBackend`](crate::backend::KvBackend) opened at
/// `{root_directory}/logs/transactions/{deployment_hash}/`. Tests use
/// the in-memory [`MemoryBackend`](crate::backend::MemoryBackend)
/// under the same wrapper.
#[async_trait]
pub trait TransactionLog: Send + Sync + std::fmt::Debug {
    /// Append a new entry. The entry's `hlc` is assigned by the
    /// caller (typically from a [`HybridLogicalClock::now`](crate::hlc::HybridLogicalClock::now)
    /// call at commit time). The log persists the entry verbatim.
    ///
    /// # Errors
    ///
    /// - [`crate::error::YetiError::Internal`] if encoding fails.
    /// - Whatever the underlying storage backend returns from a write.
    async fn append(&self, entry: LogEntry) -> Result<()>;

    /// Append many entries in one call. Implementations should fuse
    /// the appends into a single batched backend write where possible
    /// — the default implementation falls back to a per-entry loop and
    /// is correct but pays one awaited backend call per entry.
    ///
    /// Used by [`LoggingBackend::write_batch`](../../yeti_store/struct.LoggingBackend.html)
    /// so a 100-record table batch produces one storage write per
    /// table + one batched write on the log, instead of 100 serial
    /// awaited log puts.
    ///
    /// # Errors
    ///
    /// - [`crate::error::YetiError::Internal`] if encoding any entry fails.
    /// - Whatever the underlying storage backend returns from a write.
    async fn append_batch(&self, entries: Vec<LogEntry>) -> Result<()> {
        for entry in entries {
            self.append(entry).await?;
        }
        Ok(())
    }

    /// Read up to `limit` entries with `hlc > after`, in ascending
    /// HLC order. `after = HlcTimestamp::ZERO` reads from the
    /// beginning of the log.
    ///
    /// # Errors
    ///
    /// - [`crate::error::YetiError::Internal`] if decoding any entry
    ///   fails (indicates on-disk corruption or version skew).
    /// - Whatever the underlying storage backend returns from a scan.
    async fn read_after(&self, after: HlcTimestamp, limit: usize) -> Result<Vec<LogEntry>>;

    /// Remove entries that exceed the retention policy. Returns the
    /// number of entries removed.
    ///
    /// # Errors
    ///
    /// Whatever the underlying storage backend returns from scan +
    /// batch-delete operations.
    async fn sweep(&self, policy: RetentionPolicy) -> Result<usize>;
}

/// Shareable handle for a [`TransactionLog`] implementation.
pub type TransactionLogHandle = Arc<dyn TransactionLog>;

// ============================================================================
// WriteContext — task-local plumbing for originator / interface / request id
// ============================================================================

/// Per-request write context — set at the request entry point
/// (HTTP/MQTT/gRPC dispatcher) and read by the backend logging layer
/// to populate the corresponding fields on the [`LogEntry`].
///
/// Threaded as a tokio task-local so that intermediate layers don't
/// have to pass `Originator` / interface / request id through every
/// function signature. The HTTP dispatcher wraps each request future
/// in `WRITE_CTX::scope`; any backend write that happens inside the
/// scoped future reads the context and stamps the resulting log entry
/// accordingly.
///
/// When no scope is active (background tasks, startup bootstrap,
/// internal sweeps), [`current_write_context`] returns
/// [`WriteContext::internal`] with `component = "yeti-host"` so the
/// log still records a meaningful originator.
#[derive(Debug, Clone)]
pub struct WriteContext {
    /// Who initiated the write.
    pub originator: Originator,
    /// Which protocol the write came in through — `"rest"`, `"graphql"`,
    /// `"mqtt"`, `"mcp"`, `"internal"`. `None` when not derivable.
    pub interface: Option<String>,
    /// Optional request correlation id (e.g., the HTTP `x-request-id`
    /// header value or a generated UUID).
    pub request_id: Option<String>,
}

impl WriteContext {
    /// The default fallback used when no scope is active. Marks the
    /// write as originating from the yeti-host platform itself.
    #[must_use]
    pub fn internal() -> Self {
        Self {
            originator: Originator::Internal {
                component: "yeti-host".to_owned(),
            },
            interface: Some("internal".to_owned()),
            request_id: None,
        }
    }

    /// Construct a [`WriteContext`] for a request that came in through
    /// the named interface, identifying the authenticated user when
    /// known.
    #[must_use]
    pub fn for_request(
        interface: impl Into<String>,
        user: Option<String>,
        request_id: Option<String>,
    ) -> Self {
        let originator = user.map_or_else(Originator::anonymous, |u| Originator::Local {
            user: Some(u),
        });
        Self {
            originator,
            interface: Some(interface.into()),
            request_id,
        }
    }

    /// Construct a [`WriteContext`] for a write applied locally in
    /// response to a peer's replication stream.
    #[must_use]
    pub fn for_replication(from_node: impl Into<String>) -> Self {
        Self {
            originator: Originator::Replicated {
                from_node: from_node.into(),
            },
            interface: Some("replication".to_owned()),
            request_id: None,
        }
    }
}

tokio::task_local! {
    /// Per-task [`WriteContext`]. Set via [`WRITE_CTX::scope`] at the
    /// request entry point.
    pub static WRITE_CTX: WriteContext;
}

/// Read the active [`WriteContext`], or return [`WriteContext::internal`]
/// if no scope is currently active.
///
/// Backend logging layers call this to populate the originator /
/// interface / `request_id` fields on each [`LogEntry`]; it never
/// fails — a missing scope means "platform-originated" rather than
/// an error.
#[must_use]
pub fn current_write_context() -> WriteContext {
    WRITE_CTX
        .try_with(Clone::clone)
        .unwrap_or_else(|_| WriteContext::internal())
}

/// Build a [`WriteContext`] that overlays the authenticated user
/// onto whatever interface / `request_id` an outer scope (typically
/// the HTTP/MQTT/etc. dispatcher) has already set.
///
/// If no outer scope is active, falls back to
/// [`WriteContext::internal`] for interface / `request_id` so the
/// returned context is still well-formed — but in production every
/// real write path should have an outer dispatcher scope. The
/// fallback exists for tests and background tasks that call into the
/// yeti-table handlers without going through HTTP.
#[must_use]
pub fn current_write_context_with_user(user: Option<String>) -> WriteContext {
    let outer = current_write_context();
    let originator = user.map_or_else(Originator::anonymous, |u| Originator::Local {
        user: Some(u),
    });
    WriteContext {
        originator,
        interface: outer.interface,
        request_id: outer.request_id,
    }
}

// ============================================================================
// Tests
// ============================================================================

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;

    fn sample_entry() -> LogEntry {
        LogEntry {
            hlc: HlcTimestamp::new(1_700_000_000_000, 5),
            node_id: "node-a".to_owned(),
            table: TableId::new("demo-app", "data", "users"),
            op: WriteOp::Put {
                key: b"alice".to_vec(),
                value: b"{\"role\":\"admin\"}".to_vec(),
            },
            prev_hlc: Some(HlcTimestamp::new(1_699_000_000_000, 0)),
            prev_value: None,
            originator: Originator::local("admin"),
            request_id: Some("req-7f3".to_owned()),
            interface: Some("rest".to_owned()),
        }
    }

    #[test]
    fn originator_serde_round_trip() {
        let cases = [
            Originator::anonymous(),
            Originator::local("alice"),
            Originator::Replicated {
                from_node: "node-b".to_owned(),
            },
            Originator::Internal {
                component: "auth-bootstrap".to_owned(),
            },
        ];
        for o in cases {
            let bytes = rmp_serde::to_vec(&o).unwrap();
            let back: Originator = rmp_serde::from_slice(&bytes).unwrap();
            assert_eq!(o, back);
        }
    }

    #[test]
    fn originator_is_replicated_only_for_replicated_variant() {
        assert!(!Originator::anonymous().is_replicated());
        assert!(!Originator::local("alice").is_replicated());
        assert!(
            Originator::Replicated {
                from_node: "node-b".to_owned()
            }
            .is_replicated()
        );
        assert!(
            !Originator::Internal {
                component: "x".to_owned()
            }
            .is_replicated()
        );
    }

    #[test]
    fn log_entry_encode_decode_round_trip() {
        let entry = sample_entry();
        let bytes = entry.encode().unwrap();
        let back = LogEntry::decode(&bytes).unwrap();
        assert_eq!(entry, back);
    }

    #[test]
    fn log_entry_decode_rejects_garbage() {
        let result = LogEntry::decode(b"not-a-real-msgpack-payload");
        assert!(result.is_err());
        let msg = result.unwrap_err().to_string();
        assert!(msg.contains("LogEntry decode failed"), "got: {msg}");
    }

    #[test]
    fn log_entry_for_delete_op() {
        let entry = LogEntry {
            op: WriteOp::Delete {
                key: b"alice".to_vec(),
            },
            ..sample_entry()
        };
        let bytes = entry.encode().unwrap();
        let back = LogEntry::decode(&bytes).unwrap();
        assert_eq!(entry, back);
        assert!(back.op.is_delete());
    }

    #[test]
    fn retention_policy_constructors() {
        let t = RetentionPolicy::time_only(3600);
        assert_eq!(t.max_age_secs, Some(3600));
        assert_eq!(t.max_size_bytes, None);

        let s = RetentionPolicy::size_only(1_000_000);
        assert_eq!(s.max_age_secs, None);
        assert_eq!(s.max_size_bytes, Some(1_000_000));

        let d = RetentionPolicy::dual(3600, 1_000_000);
        assert_eq!(d.max_age_secs, Some(3600));
        assert_eq!(d.max_size_bytes, Some(1_000_000));
    }

    #[test]
    fn retention_policy_default_is_disabled() {
        assert!(RetentionPolicy::default().is_disabled());
        assert!(!RetentionPolicy::time_only(60).is_disabled());
        assert!(!RetentionPolicy::size_only(1).is_disabled());
    }

    #[test]
    fn write_op_key_accessor_works_for_both_variants() {
        let put = WriteOp::Put {
            key: b"k".to_vec(),
            value: b"v".to_vec(),
        };
        let del = WriteOp::Delete { key: b"k".to_vec() };
        assert_eq!(put.key(), b"k");
        assert_eq!(del.key(), b"k");
        assert!(!put.is_delete());
        assert!(del.is_delete());
    }

    #[tokio::test]
    async fn current_write_context_returns_internal_when_no_scope() {
        let ctx = current_write_context();
        match ctx.originator {
            Originator::Internal { component } => assert_eq!(component, "yeti-host"),
            _ => panic!("expected Internal originator outside scope"),
        }
        assert_eq!(ctx.interface.as_deref(), Some("internal"));
    }

    #[tokio::test]
    async fn current_write_context_reads_active_scope() {
        let scoped =
            WriteContext::for_request("rest", Some("alice".to_owned()), Some("req-abc".to_owned()));
        let observed = WRITE_CTX
            .scope(scoped.clone(), async { current_write_context() })
            .await;
        assert_eq!(observed.interface.as_deref(), Some("rest"));
        assert_eq!(observed.request_id.as_deref(), Some("req-abc"));
        match observed.originator {
            Originator::Local { user } => assert_eq!(user.as_deref(), Some("alice")),
            _ => panic!("expected Local originator"),
        }
    }

    #[tokio::test]
    async fn write_context_for_request_anonymous_when_user_is_none() {
        let ctx = WriteContext::for_request("graphql", None, None);
        assert!(matches!(ctx.originator, Originator::Local { user: None }));
        assert_eq!(ctx.interface.as_deref(), Some("graphql"));
    }

    #[tokio::test]
    async fn write_context_for_replication_marks_originator() {
        let ctx = WriteContext::for_replication("node-east");
        assert!(ctx.originator.is_replicated());
        assert_eq!(ctx.interface.as_deref(), Some("replication"));
    }

    #[tokio::test]
    async fn current_write_context_with_user_inherits_outer_interface_and_request_id() {
        let outer = WriteContext {
            originator: Originator::anonymous(),
            interface: Some("rest".to_owned()),
            request_id: Some("req-42".to_owned()),
        };
        WRITE_CTX
            .scope(outer, async {
                let inner = current_write_context_with_user(Some("alice".to_owned()));
                assert_eq!(inner.interface.as_deref(), Some("rest"));
                assert_eq!(inner.request_id.as_deref(), Some("req-42"));
                assert!(matches!(
                    inner.originator,
                    Originator::Local { ref user } if user.as_deref() == Some("alice")
                ));
            })
            .await;
    }

    #[tokio::test]
    async fn current_write_context_with_user_falls_back_to_internal_outside_scope() {
        let ctx = current_write_context_with_user(Some("alice".to_owned()));
        // Outer scope absent, so interface inherits the internal default.
        assert_eq!(ctx.interface.as_deref(), Some("internal"));
        assert!(matches!(
            ctx.originator,
            Originator::Local { ref user } if user.as_deref() == Some("alice")
        ));
    }

    #[tokio::test]
    async fn nested_write_ctx_scope_inner_overrides_outer() {
        let outer = WriteContext {
            originator: Originator::anonymous(),
            interface: Some("rest".to_owned()),
            request_id: Some("req-1".to_owned()),
        };
        WRITE_CTX
            .scope(outer, async {
                let inner = current_write_context_with_user(Some("bob".to_owned()));
                WRITE_CTX
                    .scope(inner, async {
                        let observed = current_write_context();
                        // Inner originator wins.
                        assert!(matches!(
                            observed.originator,
                            Originator::Local { ref user } if user.as_deref() == Some("bob")
                        ));
                        // Outer interface + request_id still visible.
                        assert_eq!(observed.interface.as_deref(), Some("rest"));
                        assert_eq!(observed.request_id.as_deref(), Some("req-1"));
                    })
                    .await;
            })
            .await;
    }

    #[tokio::test]
    async fn write_ctx_scope_does_not_leak_across_tasks() {
        let outer = WriteContext::for_request("rest", Some("alice".to_owned()), None);
        WRITE_CTX
            .scope(outer, async {
                // Inside the scope: see alice.
                let inner_ctx = current_write_context();
                assert!(matches!(
                    inner_ctx.originator,
                    Originator::Local { user: Some(ref u) } if u == "alice"
                ));

                // A spawned task does NOT inherit the task-local; it
                // should fall back to Internal.
                let handle = tokio::spawn(async { current_write_context() });
                let spawned_ctx = handle.await.unwrap();
                assert!(matches!(
                    spawned_ctx.originator,
                    Originator::Internal { .. }
                ));
            })
            .await;
    }
}