1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
//! Notify receipts (fork #50, confirmation depth 3): the sender's
//! `notify_id` becomes a checkable contract instead of a fire-and-forget
//! hope. The session message queue is per-target, so when the tool-loop
//! drain point fires for a target, every notify queued for that target is
//! consumed by definition — no id field on `QueuedUserMessage` is needed,
//! the drain stamps by target and the sender polls by id.
use std::collections::HashMap;
use std::sync::{Mutex, OnceLock};
use uuid::Uuid;
/// Lifecycle of a routed notify as observed by the receiving machinery.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ReceiptState {
/// Routed into the queue or the wake path; not yet observed at a drain
/// point.
Queued,
/// Observed at the tool-loop drain point — injected into the model
/// context between tool iterations.
Injected,
}
impl ReceiptState {
pub(crate) fn as_str(self) -> &'static str {
match self {
ReceiptState::Queued => "queued",
ReceiptState::Injected => "injected",
}
}
}
#[derive(Debug, Clone)]
pub(crate) struct NotifyReceipt {
pub target: Uuid,
pub state: ReceiptState,
pub queued_at: chrono::DateTime<chrono::Utc>,
pub injected_at: Option<chrono::DateTime<chrono::Utc>>,
}
fn receipts() -> &'static Mutex<HashMap<Uuid, NotifyReceipt>> {
static RECEIPTS: OnceLock<Mutex<HashMap<Uuid, NotifyReceipt>>> = OnceLock::new();
RECEIPTS.get_or_init(|| Mutex::new(HashMap::new()))
}
/// Record that `id` was routed to `target` and awaits consumption.
pub(crate) fn record_queued(id: Uuid, target: Uuid) {
let receipt = NotifyReceipt {
target,
state: ReceiptState::Queued,
queued_at: chrono::Utc::now(),
injected_at: None,
};
match receipts().lock() {
Ok(mut map) => {
map.insert(id, receipt);
}
Err(e) => {
tracing::error!(
target: "quiet_delivery",
"notify receipts poisoned, receipt {id} not recorded: {e}"
);
}
}
}
/// Stamp every queued receipt for `target` as injected. Called at the
/// tool-loop drain point, which consumes the session's whole queue — a
/// queued receipt that is still `Queued` after a drain for its target
/// would be a lie to the sender. Returns the number stamped.
pub(crate) fn mark_injected_for_target(target: Uuid) -> usize {
let now = chrono::Utc::now();
match receipts().lock() {
Ok(mut map) => {
let mut stamped = 0;
for receipt in map.values_mut() {
if receipt.target == target && receipt.state == ReceiptState::Queued {
receipt.state = ReceiptState::Injected;
receipt.injected_at = Some(now);
stamped += 1;
}
}
stamped
}
Err(e) => {
tracing::error!(
target: "quiet_delivery",
"notify receipts poisoned, injection stamp lost for {target}: {e}"
);
0
}
}
}
/// Sender-facing status check by id.
pub(crate) fn status(id: Uuid) -> Option<NotifyReceipt> {
receipts().lock().ok()?.get(&id).cloned()
}