agent_core/core/
context_head.rs

1//! Frontend acknowledgement for a context-head persistence barrier.
2//!
3//! Cloneable because stream events are cloneable, but completion is single-use.
4//! Dropping every copy without acknowledgement closes the receiver: consumers
5//! without durable session persistence fail closed rather than permit inference.
6use std::sync::{Arc, Mutex};
7use tokio::sync::oneshot;
8
9type HeadResult = Result<(), String>;
10type HeadSender = Arc<Mutex<Option<oneshot::Sender<HeadResult>>>>;
11
12#[derive(Clone, Debug)]
13pub struct ContextHeadReceipt {
14    sender: HeadSender,
15}
16
17impl ContextHeadReceipt {
18    pub fn channel() -> (Self, oneshot::Receiver<Result<(), String>>) {
19        let (sender, receiver) = oneshot::channel();
20        (
21            Self {
22                sender: Arc::new(Mutex::new(Some(sender))),
23            },
24            receiver,
25        )
26    }
27
28    /// Acknowledge only after the candidate session head is durably saved.
29    /// An I/O error may be post-rename: it stops inference, not promises rollback.
30    pub fn complete(&self, result: std::io::Result<()>) {
31        if let Some(sender) = self
32            .sender
33            .lock()
34            .unwrap_or_else(std::sync::PoisonError::into_inner)
35            .take()
36        {
37            let _ = sender.send(result.map_err(|error| error.to_string()));
38        }
39    }
40}
41
42#[cfg(test)]
43mod tests {
44    use super::*;
45
46    #[tokio::test]
47    async fn receipt_is_single_use_across_clones() {
48        let (receipt, result) = ContextHeadReceipt::channel();
49        receipt.clone().complete(Ok(()));
50        receipt.complete(Err(std::io::Error::other("late error")));
51        assert_eq!(result.await.unwrap(), Ok(()));
52    }
53
54    #[tokio::test]
55    async fn unhandled_receipt_fails_closed() {
56        let (receipt, result) = ContextHeadReceipt::channel();
57        drop(receipt);
58        assert!(result.await.is_err());
59    }
60}
agent_core/core/context_head.rs

agent_core/core/
context_head.rs
