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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
//! Connection-keyed notifier hook for worker registration lifecycle.
//!
//! This is the application seam for self-describing worker registration. When a
//! worker sends a [`Frame::WorkerRegister`](liminal::protocol::Frame) over its
//! established connection, the server associates the registration with the
//! connection's beamr process id and invokes the configured
//! [`ConnectionNotifier`]; on connection close it invokes the matching
//! deregistration. The notifier is connection-keyed (by pid), which is distinct
//! from the subject-keyed responder registry in [`super::services`].
//!
//! Keeping the hook a `liminal-server` trait — rather than a liminal-core
//! concern — preserves liminal's generality: liminal still runs standalone with
//! no notifier configured, and the application (aion, in Stage 2) plugs its
//! registry in without liminal depending on it.
//!
//! The same trait carries the presence seam for pass-stamped connections: a
//! connection whose `Connect` verified a registry pass (a browser or agent
//! seat, which sends no `WorkerRegister`) announces itself once on attach and
//! once on detach through [`ConnectionNotifier::on_pass_attached`] and
//! [`ConnectionNotifier::on_pass_detached`], so an application derives
//! presence from those two events and never polls for it.
use liminal::protocol::WorkerRegistration;
use crate::ServerError;
use crate::auth_pass::{PassPrincipal, VerifiedHolder};
/// Application hook invoked when a worker registers or unregisters on a
/// connection.
///
/// Implementations associate the connection's beamr process id (`pid`) with the
/// worker's declared [`WorkerRegistration`] so the application can route work to
/// it, and release that association on disconnect. The hook is synchronous: a
/// registration is acknowledged to the worker only after
/// [`on_worker_registered`](Self::on_worker_registered) returns, so a rejecting
/// application surfaces a `Rejected` ack instead of leaving the worker silently
/// connected but never dispatched-to.
pub trait ConnectionNotifier: std::fmt::Debug + Send + Sync {
/// Called when a worker registers on the connection identified by `pid`.
///
/// Returning `Ok(())` accepts the registration (the worker receives an
/// `Accepted` ack). Returning [`ServerError`] rejects it (the worker receives
/// a `Rejected` ack carrying the error text), so a failed association never
/// leaves the worker believing it is registered.
///
/// # Errors
/// Returns [`ServerError`] when the application declines the registration.
fn on_worker_registered(
&self,
pid: u64,
registration: &WorkerRegistration,
) -> Result<(), ServerError>;
/// Called when a worker registers on the connection identified by `pid`,
/// with the holder that connection's `Connect` verified, or `None` for a
/// connection that proved no one (a bearer, or an open server). The
/// server calls this hook, never [`on_worker_registered`] directly
/// (ACCESS-004 R4).
///
/// When it is called, the registration's name is the holder's: a
/// mismatch was refused before any hook ran. The default forwards to
/// [`on_worker_registered`], so an existing implementor compiles and
/// behaves unchanged; one that decides rights by holder overrides this.
///
/// [`on_worker_registered`]: Self::on_worker_registered
///
/// # Errors
/// Returns [`ServerError`] when the application declines the registration.
fn on_worker_registered_by(
&self,
pid: u64,
registration: &WorkerRegistration,
_holder: Option<&VerifiedHolder>,
) -> Result<(), ServerError> {
self.on_worker_registered(pid, registration)
}
/// Called when the connection identified by `pid` — which had a stored
/// registration — closes, so the application can release the association.
///
/// Deregistration is best-effort and infallible from the connection's
/// perspective: it runs on the close path where there is no peer to report an
/// error to.
fn on_worker_unregistered(&self, pid: u64);
/// Called when the connection identified by `pid` publishes to `channel`,
/// carrying the opaque envelope `payload`, BEFORE the normal channel fan-out.
///
/// Returns `true` when the application CONSUMED the publish out-of-band (an
/// observability-drain tap): the connection process then does NOT route it to the
/// channel-fan-out cluster and answers with no wire response, so a tapped channel
/// need not be a declared fan-out channel. Returns `false` (the default) to let
/// the publish flow through the normal channel machinery unchanged.
///
/// This is the observability-drain hook: a worker publishing an agent transcript
/// event to the reserved observability channel is consumed here — the hosting
/// application (aion) persists and live-fans-out the event without a second
/// connection. It is fire-and-forget: a publish is a one-way notification, so
/// there is no reply and a failed persist is the application's concern to log.
///
/// The default returns `false`, so liminal still runs standalone: with no
/// notifier, or a notifier that does not recognise the channel, every publish
/// routes to the normal fan-out exactly as before.
fn on_channel_publish(&self, pid: u64, channel: &str, payload: &[u8]) -> bool {
let _ = (pid, channel, payload);
false
}
/// Called exactly once when a `Connect` carrying a registry pass has
/// SUCCEEDED on the connection identified by `pid` — after the pass
/// verified, version negotiation passed, and the handshake was admitted —
/// carrying the principal the pass stamped: participant, public key, live
/// prefix, conversation scope, `may_enroll`. `pid` is the connection's own
/// id, the same one every other hook on this trait is keyed by, so the
/// matching [`on_pass_detached`](Self::on_pass_detached) can be paired.
///
/// A `Connect` that presents a pass and is refused (or a bearer connection,
/// which stamps no principal) never reaches this hook. The default does
/// nothing, so existing implementors compile unchanged and liminal still
/// runs standalone.
fn on_pass_attached(&self, pid: u64, principal: &PassPrincipal) {
let _ = (pid, principal);
}
/// Called exactly once when a connection that announced
/// [`on_pass_attached`](Self::on_pass_attached) closes — for any reason: a
/// clean `Disconnect`, the peer ending the transport, a transport error, a
/// process crash, or supervisor shutdown — carrying the same `pid` and the
/// principal the attach carried. Best-effort and infallible like
/// [`on_worker_unregistered`](Self::on_worker_unregistered): it runs on the
/// close path where there is no peer to report to. The default does
/// nothing.
fn on_pass_detached(&self, pid: u64, principal: &PassPrincipal) {
let _ = (pid, principal);
}
}