netget/state/
intercepts.rs

1//! Events parked for a **human** to answer.
2//!
3//! A `manual` event handler (`EventHandlerType::Manual`) is the fourth way to answer a
4//! request, next to static / script / LLM: the event becomes a pending question at the
5//! dashboard, the connection waits exactly as it would for a slow model, and whatever
6//! actions the operator composes are executed as the answer.
7//!
8//! Mechanically this is a rendezvous: the dispatcher (inside the connection's event task)
9//! calls [`AppState::park_intercept`], gets a `oneshot::Receiver`, and awaits it under the
10//! handler's timeout. The dashboard lists pending intercepts from its snapshot and calls
11//! [`AppState::resolve_intercept`] with the composed actions — or
12//! [`AppState::dismiss_intercept`] to refuse, which the dispatcher sees as "no answer" and
13//! turns into the fail-closed path. Nothing here touches a socket; the waiting dispatcher
14//! owns the wire.
15//!
16//! The types live here; the `impl AppState` accessors live in `app_state.rs` (same
17//! split as `client_handles.rs`), because `AppState::inner` is module-private.
18
19use serde_json::Value;
20use tokio::sync::oneshot;
21
22use crate::state::{ClientId, ServerId};
23
24/// Which instance the parked event belongs to.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum InterceptOwner {
27    Server(ServerId),
28    Client(ClientId),
29}
30
31/// One parked event, held in `AppState` until answered, dismissed, or timed out.
32pub struct PendingIntercept {
33    pub id: u64,
34    pub owner: InterceptOwner,
35    pub connection_id: Option<u32>,
36    pub event_type: String,
37    pub description: String,
38    pub event_data: Option<Value>,
39    pub created_unix_ms: u64,
40    /// How long the waiting dispatcher gives the operator before failing closed.
41    pub timeout_secs: u64,
42    /// Taken by `resolve_intercept`; dropped by `dismiss_intercept`. When the
43    /// *receiver* side is gone (the dispatcher timed out or its connection task
44    /// was cancelled), the entry is dead and is pruned lazily.
45    pub(crate) reply_tx: Option<oneshot::Sender<Vec<Value>>>,
46}
47
48/// A cloneable view of a pending intercept, for the dashboard's snapshot.
49#[derive(Debug, Clone)]
50pub struct InterceptView {
51    pub id: u64,
52    pub owner: InterceptOwner,
53    pub connection_id: Option<u32>,
54    pub event_type: String,
55    pub description: String,
56    pub event_data: Option<Value>,
57    pub created_unix_ms: u64,
58    /// The handler's `timeout_secs`: the event fails closed this long after
59    /// `created_unix_ms` if nobody answers.
60    pub timeout_secs: u64,
61}
62
63impl InterceptView {
64    /// Whole seconds until the waiting dispatcher gives up and fails closed,
65    /// measured against `now_unix_ms` (0 once the deadline has passed).
66    pub fn seconds_until_fail_closed(&self, now_unix_ms: u64) -> u64 {
67        let deadline = self
68            .created_unix_ms
69            .saturating_add(self.timeout_secs.saturating_mul(1000));
70        deadline.saturating_sub(now_unix_ms) / 1000
71    }
72}
73
74impl PendingIntercept {
75    pub(crate) fn view(&self) -> InterceptView {
76        InterceptView {
77            id: self.id,
78            owner: self.owner,
79            connection_id: self.connection_id,
80            event_type: self.event_type.clone(),
81            description: self.description.clone(),
82            event_data: self.event_data.clone(),
83            created_unix_ms: self.created_unix_ms,
84            timeout_secs: self.timeout_secs,
85        }
86    }
87
88    /// Dead entries: answered/dismissed (`reply_tx` taken) or abandoned (the
89    /// dispatcher's receiver dropped on timeout or task cancellation).
90    pub(crate) fn is_dead(&self) -> bool {
91        match &self.reply_tx {
92            None => true,
93            Some(tx) => tx.is_closed(),
94        }
95    }
96}
97
98pub fn now_unix_ms() -> u64 {
99    crate::utils::clock::SystemTime::now()
100        .duration_since(crate::utils::clock::UNIX_EPOCH)
101        .map(|d| d.as_millis() as u64)
102        .unwrap_or(0)
103}
netget/state/intercepts.rs

netget/state/
intercepts.rs
