rightkit_hooks/
receipt.rs

1//! Delivery receipts at two levels.
2//!
3//! - `emitted`: the hook printed schema-valid output. Pulse's chat `delivered` is this level: the
4//!   claim happens before the print, so a rejected or truncated output still counts.
5//! - `observed`: the context's marker line was found in the host's own transcript (Claude JSONL
6//!   transcript, Codex rollout JSONL), read back after the turn. Only [`ReceiptStore::observe`]
7//!   sets it.
8//!
9//! The marker is the last line of the emitted context (`[<prefix>:<id>]`, Membrane's
10//! `receipt_marker` with a caller-chosen prefix). A receipt is content-free: it keeps the
11//! length of the context, never the context. [`ObserveModule`] is the transcript-scan hook.
12//!
13//! A module that timed out in the dispatcher keeps running on its detached thread and can still
14//! write a receipt after the hook has replied; such a receipt is real but late.
15
16use crate::dispatch::{HookModule, ModuleContext, ModuleOutput};
17use crate::event::{EventKind, HookEvent, Wire};
18use serde::{Deserialize, Serialize};
19use serde_json::json;
20use std::io::{Read, Seek, SeekFrom};
21use std::path::{Path, PathBuf};
22use std::sync::atomic::{AtomicU64, Ordering};
23use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
24
25/// Most of a transcript's tail one scan reads (Membrane's `CONFIRM_TAIL_BYTES`).
26pub const SCAN_TAIL_BYTES: u64 = 4 * 1024 * 1024;
27/// Wall budget of one scan (Membrane's `CONFIRM_BUDGET`).
28pub const SCAN_BUDGET: Duration = Duration::from_secs(2);
29const RECEIPTS_DIR: &str = "receipts";
30
31#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
32#[serde(rename_all = "snake_case")]
33pub enum DeliveryLevel {
34    Emitted,
35    Observed,
36}
37
38/// Content-free record of one emitted context.
39#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
40#[serde(rename_all = "camelCase")]
41pub struct Receipt {
42    pub id: String,
43    pub session: String,
44    pub wire: String,
45    pub event: String,
46    pub level: DeliveryLevel,
47    /// Bytes of context emitted, marker included.
48    pub chars: usize,
49    pub ts: u64,
50}
51
52fn unix_seconds() -> u64 {
53    SystemTime::now()
54        .duration_since(UNIX_EPOCH)
55        .map_or(0, |d| d.as_secs())
56}
57
58/// A fresh alphanumeric receipt id: nanosecond clock, process id and a per-process counter,
59/// concatenated (not mixed), so ids from different processes or calls cannot collide unless the
60/// clock, pid and counter all agree.
61pub fn new_receipt_id() -> String {
62    static COUNTER: AtomicU64 = AtomicU64::new(0);
63    let nanos = SystemTime::now()
64        .duration_since(UNIX_EPOCH)
65        .map_or(0, |d| d.as_nanos() as u64);
66    let count = COUNTER.fetch_add(1, Ordering::Relaxed) & 0xffff;
67    format!("{nanos:016x}{:08x}{count:04x}", std::process::id())
68}
69
70/// The marker line for `id`: `[prefix:id]`.
71pub fn marker(prefix: &str, id: &str) -> String {
72    format!("[{prefix}:{id}]")
73}
74
75/// Append the marker as the final line of `context`, cutting the body (never the marker) so the
76/// whole stays within `cap_bytes`. When the cap is smaller than the marker line itself, the marker
77/// is skipped and the context is cut to the cap: the result never exceeds `cap_bytes`.
78pub fn tag_context(context: &str, prefix: &str, id: &str, cap_bytes: usize) -> String {
79    let tail = format!("\n{}", marker(prefix, id));
80    let (room, tail) = match cap_bytes.checked_sub(tail.len()) {
81        Some(room) => (room, tail),
82        None => (cap_bytes, String::new()),
83    };
84    let mut cut = context.len().min(room);
85    while !context.is_char_boundary(cut) {
86        cut -= 1;
87    }
88    format!("{}{tail}", &context[..cut])
89}
90
91/// Pure scan: which of `markers` appear in the last [`SCAN_TAIL_BYTES`] of the file, within
92/// `budget`. A missing or unreadable file finds nothing.
93pub fn scan_transcript_tail(path: &Path, markers: &[String], budget: Duration) -> Vec<String> {
94    let started = Instant::now();
95    let Ok(mut file) = std::fs::File::open(path) else {
96        return Vec::new();
97    };
98    let Ok(length) = file.metadata().map(|m| m.len()) else {
99        return Vec::new();
100    };
101    if file
102        .seek(SeekFrom::Start(length.saturating_sub(SCAN_TAIL_BYTES)))
103        .is_err()
104    {
105        return Vec::new();
106    }
107    let mut tail = Vec::new();
108    if file.take(SCAN_TAIL_BYTES).read_to_end(&mut tail).is_err() {
109        return Vec::new();
110    }
111    let tail = String::from_utf8_lossy(&tail);
112    markers
113        .iter()
114        .take_while(|_| started.elapsed() < budget)
115        .filter(|marker| tail.contains(marker.as_str()))
116        .cloned()
117        .collect()
118}
119
120/// A host session id as a file name: characters outside `[A-Za-z0-9._-]` become `_`, cut at 128;
121/// empty or all-dot ids have no name.
122fn session_file_name(session: &str) -> Option<String> {
123    let name: String = session
124        .chars()
125        .map(|c| {
126            if c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-') {
127                c
128            } else {
129                '_'
130            }
131        })
132        .take(128)
133        .collect();
134    (!name.chars().all(|c| c == '.')).then_some(name)
135}
136
137/// Receipts on disk: `<root>/receipts/<session>/<id>.json`.
138#[derive(Clone, Debug)]
139pub struct ReceiptStore {
140    root: PathBuf,
141    prefix: String,
142}
143
144impl ReceiptStore {
145    /// `prefix` is the marker prefix this producer uses (for example `pulse`).
146    pub fn new(root: impl Into<PathBuf>, prefix: impl Into<String>) -> Self {
147        Self {
148            root: root.into(),
149            prefix: prefix.into(),
150        }
151    }
152
153    pub fn prefix(&self) -> &str {
154        &self.prefix
155    }
156
157    fn path(&self, session: &str, id: &str) -> Option<PathBuf> {
158        if id.is_empty() || !id.chars().all(|c| c.is_ascii_alphanumeric()) {
159            return None;
160        }
161        Some(
162            self.root
163                .join(RECEIPTS_DIR)
164                .join(session_file_name(session)?)
165                .join(format!("{id}.json")),
166        )
167    }
168
169    fn write(&self, receipt: &Receipt) -> std::io::Result<()> {
170        let Some(path) = self.path(&receipt.session, &receipt.id) else {
171            return Err(std::io::Error::other(
172                "receipt id or session is not a file name",
173            ));
174        };
175        let parent = path
176            .parent()
177            .ok_or_else(|| std::io::Error::other("no parent"))?;
178        std::fs::create_dir_all(parent)?;
179        let tmp = parent.join(format!(".{}.tmp", receipt.id));
180        std::fs::write(
181            &tmp,
182            serde_json::to_vec(receipt).map_err(std::io::Error::other)?,
183        )?;
184        std::fs::rename(&tmp, &path)
185    }
186
187    /// Record context just printed: level `emitted`. `context` is the text including its marker.
188    pub fn record_emitted(
189        &self,
190        event: &HookEvent,
191        id: &str,
192        context: &str,
193    ) -> std::io::Result<Receipt> {
194        let session = event
195            .session_id
196            .as_deref()
197            .ok_or_else(|| std::io::Error::other("event has no session id"))?;
198        let receipt = Receipt {
199            id: id.to_owned(),
200            session: session.to_owned(),
201            wire: event.wire.as_str().to_owned(),
202            event: event.kind.to_string(),
203            level: DeliveryLevel::Emitted,
204            chars: context.len(),
205            ts: unix_seconds(),
206        };
207        self.write(&receipt)?;
208        Ok(receipt)
209    }
210
211    /// Tag `context` with a fresh marker, record it `emitted`, and return the tagged text to print.
212    pub fn tag_and_record(
213        &self,
214        event: &HookEvent,
215        context: &str,
216        cap_bytes: usize,
217    ) -> std::io::Result<String> {
218        let id = new_receipt_id();
219        let tagged = tag_context(context, &self.prefix, &id, cap_bytes);
220        self.record_emitted(event, &id, &tagged)?;
221        Ok(tagged)
222    }
223
224    /// Every receipt of `session`, oldest first.
225    pub fn list(&self, session: &str) -> Vec<Receipt> {
226        let Some(name) = session_file_name(session) else {
227            return Vec::new();
228        };
229        let mut receipts: Vec<Receipt> = std::fs::read_dir(self.root.join(RECEIPTS_DIR).join(name))
230            .into_iter()
231            .flatten()
232            .flatten()
233            .filter(|e| e.path().extension().is_some_and(|x| x == "json"))
234            .filter_map(|e| std::fs::read(e.path()).ok())
235            .filter_map(|bytes| serde_json::from_slice(&bytes).ok())
236            .collect();
237        receipts.sort_by(|a: &Receipt, b| (a.ts, &a.id).cmp(&(b.ts, &b.id)));
238        receipts
239    }
240
241    /// Flip to `observed` every `emitted` receipt of `session` and `wire` whose marker is in the
242    /// transcript tail. A miss leaves it `emitted`. Returns how many flipped.
243    pub fn observe(&self, session: &str, wire: Wire, transcript: &Path) -> usize {
244        let pending: Vec<Receipt> = self
245            .list(session)
246            .into_iter()
247            .filter(|r| r.level == DeliveryLevel::Emitted && r.wire == wire.as_str())
248            .collect();
249        if pending.is_empty() {
250            return 0;
251        }
252        let markers: Vec<String> = pending
253            .iter()
254            .map(|r| marker(&self.prefix, &r.id))
255            .collect();
256        let found = scan_transcript_tail(transcript, &markers, SCAN_BUDGET);
257        pending
258            .into_iter()
259            .zip(markers)
260            .filter(|(_, marker)| found.contains(marker))
261            .filter(|(receipt, _)| {
262                let mut receipt = receipt.clone();
263                receipt.level = DeliveryLevel::Observed;
264                self.write(&receipt).is_ok()
265            })
266            .count()
267    }
268}
269
270/// The transcript-scan hook: on the chat's own `Stop`, mark emitted receipts `observed`.
271pub struct ObserveModule {
272    store: ReceiptStore,
273}
274
275impl ObserveModule {
276    pub fn new(store: ReceiptStore) -> Self {
277        Self { store }
278    }
279}
280
281impl HookModule for ObserveModule {
282    fn id(&self) -> &str {
283        "rightkit.receipts.observe"
284    }
285
286    fn handle(&self, event: &HookEvent, _ctx: &ModuleContext) -> Result<ModuleOutput, String> {
287        let (EventKind::Stop, true) = (&event.kind, event.is_own_turn()) else {
288            return Ok(ModuleOutput::empty());
289        };
290        let (Some(session), Some(path)) = (&event.session_id, &event.transcript_path) else {
291            return Ok(ModuleOutput::empty());
292        };
293        let observed = self.store.observe(session, event.wire, Path::new(path));
294        Ok(ModuleOutput {
295            detail: json!({"observed": observed}),
296            ..ModuleOutput::default()
297        })
298    }
299}
rightkit_hooks/receipt.rs

rightkit_hooks/
receipt.rs
