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/// Marker prefix of the store Membrane reads and writes (`[membrane:<id>]`).
31pub const MEMBRANE_MARKER_PREFIX: &str = "membrane";
32
33#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
34#[serde(rename_all = "snake_case")]
35pub enum DeliveryLevel {
36    Emitted,
37    Observed,
38}
39
40/// Content-free record of one emitted context.
41#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
42#[serde(rename_all = "camelCase")]
43pub struct Receipt {
44    pub id: String,
45    pub session: String,
46    pub wire: String,
47    pub event: String,
48    pub level: DeliveryLevel,
49    /// Bytes of context emitted, marker included.
50    pub chars: usize,
51    pub ts: u64,
52}
53
54/// The on-disk form of a [`Receipt`], shared with Membrane's `ContextReceipt`.
55///
56/// Membrane names the fields `harness` and `state` (`emitted` or `delivered`); this crate names
57/// them `wire` and `level` (`emitted` or `observed`) and keeps `chars`. Writes carry both
58/// spellings, so Membrane reads every file this crate writes and this crate reads Membrane's.
59/// Reads accept either spelling; a file with only `state` gets its `level` from `state`.
60#[derive(Serialize, Deserialize)]
61#[serde(rename_all = "camelCase")]
62struct DiskReceipt {
63    id: String,
64    session: String,
65    #[serde(default, skip_serializing_if = "Option::is_none")]
66    wire: Option<String>,
67    #[serde(default, skip_serializing_if = "Option::is_none")]
68    harness: Option<String>,
69    event: String,
70    #[serde(default, skip_serializing_if = "Option::is_none")]
71    level: Option<String>,
72    #[serde(default, skip_serializing_if = "Option::is_none")]
73    state: Option<String>,
74    #[serde(default)]
75    chars: usize,
76    ts: u64,
77}
78
79impl DiskReceipt {
80    fn from_receipt(receipt: &Receipt) -> Self {
81        let (level, state) = match receipt.level {
82            DeliveryLevel::Emitted => ("emitted", "emitted"),
83            DeliveryLevel::Observed => ("observed", "delivered"),
84        };
85        Self {
86            id: receipt.id.clone(),
87            session: receipt.session.clone(),
88            wire: Some(receipt.wire.clone()),
89            harness: Some(receipt.wire.clone()),
90            event: receipt.event.clone(),
91            level: Some(level.to_owned()),
92            state: Some(state.to_owned()),
93            chars: receipt.chars,
94            ts: receipt.ts,
95        }
96    }
97
98    fn into_receipt(self) -> Option<Receipt> {
99        let wire = self.wire.or(self.harness)?;
100        // Membrane's confirm flips only `state`, so a read record whose `level` still says
101        // `emitted` is observed when `state` says `delivered`: any observed spelling wins.
102        let spellings = [self.level.as_deref(), self.state.as_deref()];
103        let level = if spellings
104            .iter()
105            .flatten()
106            .any(|s| matches!(*s, "observed" | "delivered"))
107        {
108            DeliveryLevel::Observed
109        } else if spellings.iter().flatten().any(|s| *s == "emitted") {
110            DeliveryLevel::Emitted
111        } else {
112            return None;
113        };
114        Some(Receipt {
115            id: self.id,
116            session: self.session,
117            wire,
118            event: self.event,
119            level,
120            chars: self.chars,
121            ts: self.ts,
122        })
123    }
124}
125
126fn unix_seconds() -> u64 {
127    SystemTime::now()
128        .duration_since(UNIX_EPOCH)
129        .map_or(0, |d| d.as_secs())
130}
131
132/// A fresh alphanumeric receipt id: nanosecond clock, process id and a per-process counter,
133/// concatenated (not mixed), so ids from different processes or calls cannot collide unless the
134/// clock, pid and counter all agree.
135pub fn new_receipt_id() -> String {
136    static COUNTER: AtomicU64 = AtomicU64::new(0);
137    let nanos = SystemTime::now()
138        .duration_since(UNIX_EPOCH)
139        .map_or(0, |d| d.as_nanos() as u64);
140    let count = COUNTER.fetch_add(1, Ordering::Relaxed) & 0xffff;
141    format!("{nanos:016x}{:08x}{count:04x}", std::process::id())
142}
143
144/// The marker line for `id`: `[prefix:id]`.
145pub fn marker(prefix: &str, id: &str) -> String {
146    format!("[{prefix}:{id}]")
147}
148
149/// Append the marker as the final line of `context`, cutting the body (never the marker) so the
150/// whole stays within `cap_bytes`. When the cap is smaller than the marker line itself, the marker
151/// is skipped and the context is cut to the cap: the result never exceeds `cap_bytes`.
152pub fn tag_context(context: &str, prefix: &str, id: &str, cap_bytes: usize) -> String {
153    let tail = format!("\n{}", marker(prefix, id));
154    let (room, tail) = match cap_bytes.checked_sub(tail.len()) {
155        Some(room) => (room, tail),
156        None => (cap_bytes, String::new()),
157    };
158    let mut cut = context.len().min(room);
159    while !context.is_char_boundary(cut) {
160        cut -= 1;
161    }
162    format!("{}{tail}", &context[..cut])
163}
164
165/// Pure scan: which of `markers` appear in the last [`SCAN_TAIL_BYTES`] of the file, within
166/// `budget`. A missing or unreadable file finds nothing.
167pub fn scan_transcript_tail(path: &Path, markers: &[String], budget: Duration) -> Vec<String> {
168    let started = Instant::now();
169    let Ok(mut file) = std::fs::File::open(path) else {
170        return Vec::new();
171    };
172    let Ok(length) = file.metadata().map(|m| m.len()) else {
173        return Vec::new();
174    };
175    if file
176        .seek(SeekFrom::Start(length.saturating_sub(SCAN_TAIL_BYTES)))
177        .is_err()
178    {
179        return Vec::new();
180    }
181    let mut tail = Vec::new();
182    if file.take(SCAN_TAIL_BYTES).read_to_end(&mut tail).is_err() {
183        return Vec::new();
184    }
185    let tail = String::from_utf8_lossy(&tail);
186    markers
187        .iter()
188        .take_while(|_| started.elapsed() < budget)
189        .filter(|marker| tail.contains(marker.as_str()))
190        .cloned()
191        .collect()
192}
193
194/// A host session id as a file name: characters outside `[A-Za-z0-9._-]` become `_`, cut at 128;
195/// empty or all-dot ids have no name.
196fn session_file_name(session: &str) -> Option<String> {
197    let name: String = session
198        .chars()
199        .map(|c| {
200            if c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-') {
201                c
202            } else {
203                '_'
204            }
205        })
206        .take(128)
207        .collect();
208    (!name.chars().all(|c| c == '.')).then_some(name)
209}
210
211/// Receipts on disk: `<root>/receipts/<session>/<id>.json`.
212#[derive(Clone, Debug)]
213pub struct ReceiptStore {
214    root: PathBuf,
215    prefix: String,
216}
217
218impl ReceiptStore {
219    /// `prefix` is the marker prefix this producer uses (for example `pulse`).
220    pub fn new(root: impl Into<PathBuf>, prefix: impl Into<String>) -> Self {
221        Self {
222            root: root.into(),
223            prefix: prefix.into(),
224        }
225    }
226
227    /// The store Membrane reads and writes: marker prefix `membrane`, the same
228    /// `receipts/<session>/<id>.json` layout and the shared field spellings of [`DiskReceipt`].
229    pub fn membrane(root: impl Into<PathBuf>) -> Self {
230        Self::new(root, MEMBRANE_MARKER_PREFIX)
231    }
232
233    pub fn prefix(&self) -> &str {
234        &self.prefix
235    }
236
237    fn path(&self, session: &str, id: &str) -> Option<PathBuf> {
238        if id.is_empty() || !id.chars().all(|c| c.is_ascii_alphanumeric()) {
239            return None;
240        }
241        Some(
242            self.root
243                .join(RECEIPTS_DIR)
244                .join(session_file_name(session)?)
245                .join(format!("{id}.json")),
246        )
247    }
248
249    fn write(&self, receipt: &Receipt) -> std::io::Result<()> {
250        let Some(path) = self.path(&receipt.session, &receipt.id) else {
251            return Err(std::io::Error::other(
252                "receipt id or session is not a file name",
253            ));
254        };
255        let parent = path
256            .parent()
257            .ok_or_else(|| std::io::Error::other("no parent"))?;
258        std::fs::create_dir_all(parent)?;
259        let tmp = parent.join(format!(".{}.tmp", receipt.id));
260        let bytes = serde_json::to_vec(&DiskReceipt::from_receipt(receipt))
261            .map_err(std::io::Error::other)?;
262        std::fs::write(&tmp, bytes)?;
263        std::fs::rename(&tmp, &path)
264    }
265
266    /// Record context just printed: level `emitted`. `context` is the text including its marker.
267    pub fn record_emitted(
268        &self,
269        event: &HookEvent,
270        id: &str,
271        context: &str,
272    ) -> std::io::Result<Receipt> {
273        let session = event
274            .session_id
275            .as_deref()
276            .ok_or_else(|| std::io::Error::other("event has no session id"))?;
277        let receipt = Receipt {
278            id: id.to_owned(),
279            session: session.to_owned(),
280            wire: event.wire.as_str().to_owned(),
281            event: event.kind.to_string(),
282            level: DeliveryLevel::Emitted,
283            chars: context.len(),
284            ts: unix_seconds(),
285        };
286        self.write(&receipt)?;
287        Ok(receipt)
288    }
289
290    /// Tag `context` with a fresh marker, record it `emitted`, and return the tagged text to print.
291    pub fn tag_and_record(
292        &self,
293        event: &HookEvent,
294        context: &str,
295        cap_bytes: usize,
296    ) -> std::io::Result<String> {
297        let id = new_receipt_id();
298        let tagged = tag_context(context, &self.prefix, &id, cap_bytes);
299        self.record_emitted(event, &id, &tagged)?;
300        Ok(tagged)
301    }
302
303    /// Every receipt of `session`, oldest first.
304    pub fn list(&self, session: &str) -> Vec<Receipt> {
305        let Some(name) = session_file_name(session) else {
306            return Vec::new();
307        };
308        let mut receipts: Vec<Receipt> = std::fs::read_dir(self.root.join(RECEIPTS_DIR).join(name))
309            .into_iter()
310            .flatten()
311            .flatten()
312            .filter(|e| e.path().extension().is_some_and(|x| x == "json"))
313            .filter_map(|e| std::fs::read(e.path()).ok())
314            .filter_map(|bytes| serde_json::from_slice::<DiskReceipt>(&bytes).ok())
315            .filter_map(DiskReceipt::into_receipt)
316            .collect();
317        receipts.sort_by(|a: &Receipt, b| (a.ts, &a.id).cmp(&(b.ts, &b.id)));
318        receipts
319    }
320
321    /// Flip to `observed` every `emitted` receipt of `session` and `wire` whose marker is in the
322    /// transcript tail. A miss leaves it `emitted`. Returns how many flipped.
323    pub fn observe(&self, session: &str, wire: Wire, transcript: &Path) -> usize {
324        let pending: Vec<Receipt> = self
325            .list(session)
326            .into_iter()
327            .filter(|r| r.level == DeliveryLevel::Emitted && r.wire == wire.as_str())
328            .collect();
329        if pending.is_empty() {
330            return 0;
331        }
332        let markers: Vec<String> = pending
333            .iter()
334            .map(|r| marker(&self.prefix, &r.id))
335            .collect();
336        let found = scan_transcript_tail(transcript, &markers, SCAN_BUDGET);
337        pending
338            .into_iter()
339            .zip(markers)
340            .filter(|(_, marker)| found.contains(marker))
341            .filter(|(receipt, _)| {
342                let mut receipt = receipt.clone();
343                receipt.level = DeliveryLevel::Observed;
344                self.write(&receipt).is_ok()
345            })
346            .count()
347    }
348}
349
350/// The transcript-scan hook: on the chat's own `Stop`, mark emitted receipts `observed`.
351pub struct ObserveModule {
352    store: ReceiptStore,
353}
354
355impl ObserveModule {
356    pub fn new(store: ReceiptStore) -> Self {
357        Self { store }
358    }
359}
360
361impl HookModule for ObserveModule {
362    fn id(&self) -> &str {
363        "rightkit.receipts.observe"
364    }
365
366    fn handle(&self, event: &HookEvent, _ctx: &ModuleContext) -> Result<ModuleOutput, String> {
367        let (EventKind::Stop, true) = (&event.kind, event.is_own_turn()) else {
368            return Ok(ModuleOutput::empty());
369        };
370        let (Some(session), Some(path)) = (&event.session_id, &event.transcript_path) else {
371            return Ok(ModuleOutput::empty());
372        };
373        let observed = self.store.observe(session, event.wire, Path::new(path));
374        Ok(ModuleOutput {
375            detail: json!({"observed": observed}),
376            ..ModuleOutput::default()
377        })
378    }
379}
rightkit_hooks/receipt.rs

rightkit_hooks/
receipt.rs
