hatchery_node/server/
node_wake.rs

1//! Letter -> node -> g4a wake -> session turn.
2//!
3//! m4a only hands a letter over (its `M4A_LETTER_HOOK`); the node knows which of its sessions
4//! owns the addressed nick (it assigned that account at spawn) and asks g4a
5//! (`gate4agent_wake::hosted`) what to do. A live session gets the framed prompt as its next
6//! turn; the node's policy ([`crate::wake_pipeline`]) gates everything else: typing into a raw
7//! terminal needs an explicit terminal grant, a nick bound to a routine gets one routine POST via
8//! g4a, and anything else is refused (m4a keeps the letter in its durable inbox).
9//! Callers present the node's wake token (a 0600 file next to the accounts root), so another
10//! local user cannot type into a session through the door.
11
12use std::collections::BTreeMap;
13use std::sync::Arc;
14
15use gate4agent_wake::hosted::{plan_hosted_wake, HostedSessions, HostedTarget, HostedWake};
16use gate4agent_wake::{ProviderKind, WakeLetter};
17
18use crate::wake_pipeline::{deliver_routine, launch_spawn_profile, route, Route, RoutineLetter, SessionShape};
19
20use super::*;
21
22/// One letter for a session of this node.
23pub struct NodeLetter {
24    pub nick: String,
25    pub from: String,
26    pub event_id: String,
27    pub body: String,
28    /// The node spawned this session itself: a raw terminal gets the short wake line typed by
29    /// the node (no terminal grant needed; the line carries no letter text).
30    pub own_spawn: bool,
31}
32
33#[derive(Clone)]
34pub struct NodeWake {
35    shared: Arc<NodeShared>,
36    token: Arc<str>,
37    policy: Arc<crate::wake_pipeline::WakePolicy>,
38}
39
40struct Snapshot(BTreeMap<String, HostedTarget>);
41
42impl HostedSessions for Snapshot {
43    fn lookup(&self, nick: &str) -> Option<HostedTarget> {
44        self.0.get(nick).cloned()
45    }
46}
47
48fn kind_of(agent: &AgentId) -> Option<ProviderKind> {
49    ProviderKind::parse(crate::m4a_accounts::m4a_provider(agent.as_str())?)
50}
51
52impl NodeWake {
53    pub(super) fn new(shared: Arc<NodeShared>, token: String, policy: Arc<crate::wake_pipeline::WakePolicy>) -> Self {
54        Self { shared, token: token.into(), policy }
55    }
56
57    /// Constant-time check of the caller's token.
58    pub fn authorized(&self, token: &str) -> bool {
59        let (a, b) = (self.token.as_bytes(), token.as_bytes());
60        a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
61    }
62
63    fn snapshot(&self) -> (Snapshot, BTreeMap<String, SessionAddress>) {
64        let live = self.shared.m4a_live.lock().unwrap_or_else(|p| p.into_inner()).clone();
65        let bindings = self.shared.session_bindings.lock().unwrap_or_else(|p| p.into_inner());
66        let mut targets = BTreeMap::new();
67        let mut addresses = BTreeMap::new();
68        for (nick, (instance_id, agent)) in live {
69            let Some(provider) = kind_of(&agent) else { continue };
70            let binding = bindings.get(&instance_id);
71            if let Some(b) = binding {
72                addresses.insert(
73                    nick.clone(),
74                    SessionAddress { workspace_id: b.workspace_id.clone(), session: SessionKey { instance_id, generation: b.generation } },
75                );
76            }
77            let spawn_allowed = self.policy.spawn_profile(&nick).is_some();
78            targets.insert(nick.clone(), HostedTarget { nick, provider, live: binding.is_some(), spawn_allowed });
79        }
80        (Snapshot(targets), addresses)
81    }
82
83    /// Wake signal for a session this node spawned (from its mailbox listener): no letter text.
84    pub async fn wake_spawned(&self, nick: String, from: String, event_id: String) -> Result<&'static str, String> {
85        self.wake(NodeLetter { nick, from, event_id, body: String::new(), own_spawn: true }).await
86    }
87
88    /// Wakes the session that owns `letter.nick`. `Ok` names what happened; `Err` is safe to
89    /// log (it never carries the letter text).
90    pub async fn wake(&self, letter: NodeLetter) -> Result<&'static str, String> {
91        let (sessions, addresses) = self.snapshot();
92        let wake = WakeLetter { body: &letter.body, from_nick: &letter.from, event_id: &letter.event_id, room: None, reaction: None };
93        match plan_hosted_wake(&sessions, &letter.nick, &wake) {
94            HostedWake::Submit { nick, prompt } => {
95                let prompt = if self.policy.deliver_body { prompt } else { crate::wake_pipeline::wake_line(&letter.from, &letter.event_id) };
96                let address = addresses.get(&nick).cloned().ok_or("session vanished")?;
97                let agent = self.shared.validate_address(&address).map_err(|_| "session is not live".to_owned())?;
98                let refused = |e: NodeFailure| {
99                    tracing::warn!(nick = %nick, code = ?e.code, "letter wake input refused");
100                    "session did not take the turn".to_owned()
101                };
102                let shape = if self.shared.require_session_runtime_policy(&address, ProviderRuntimeRequirement::SemanticPrompt).is_ok() {
103                    SessionShape::Semantic
104                } else {
105                    SessionShape::RawTerminal
106                };
107                let routed = match route(&self.policy, &nick, shape) {
108                    Route::Refuse(_) if letter.own_spawn && shape == SessionShape::RawTerminal => Route::TerminalTyping,
109                    other => other,
110                };
111                let prompt = if letter.own_spawn { crate::wake_pipeline::wake_line(&letter.from, &letter.event_id) } else { prompt };
112                match routed {
113                    Route::Deliver => {
114                        let action = InputAction::SubmitPrompt(PromptPayload { text: prompt, framing: prompt_framing(&agent) });
115                        self.shared.dispatch_input_bounded(&address, action).await.map_err(refused)?;
116                    }
117                    Route::TerminalTyping => {
118                        // Terminal capability, granted explicitly for this nick: type the prompt
119                        // as one line, then Enter.
120                        let text = prompt.replace(['\n', '\r'], " ");
121                        self.shared.dispatch_input_bounded(&address, InputAction::TerminalText(TerminalText { text })).await.map_err(refused)?;
122                        self.shared.dispatch_input_bounded(&address, InputAction::TerminalControl(TerminalControl::ControlM)).await.map_err(refused)?;
123                    }
124                    Route::Refuse(why) => {
125                        tracing::info!(nick = %nick, "letter not typed: no terminal grant");
126                        return Err(why);
127                    }
128                    Route::Routine(_) => return Err("unexpected routine route for a live session".into()),
129                    Route::Spawn(_) => return Err("unexpected spawn route for a live session".into()),
130                }
131                tracing::info!(nick = %nick, "letter woke a live session");
132                Ok("submitted")
133            }
134            HostedWake::Spawn { nick, .. } => match self.policy.spawn_profile(&nick).cloned() {
135                Some(profile) => {
136                    tokio::task::spawn_blocking(move || launch_spawn_profile(&profile))
137                        .await
138                        .map_err(|_| "spawn profile task failed".to_owned())?
139                        .map_err(|e| e)?;
140                    tracing::info!(nick = %nick, "letter woke an offline session via spawn profile");
141                    Ok("spawned")
142                }
143                None => Err("spawn on letter is not enabled on this node".into()),
144            },
145            HostedWake::Refuse(why) => match route(&self.policy, &letter.nick, SessionShape::None) {
146                Route::Spawn(profile) => {
147                    let nick = letter.nick.clone();
148                    tokio::task::spawn_blocking(move || launch_spawn_profile(&profile))
149                        .await
150                        .map_err(|_| "spawn profile task failed".to_owned())?
151                        .map_err(|e| e)?;
152                    tracing::info!(nick = %nick, "letter woke an offline session via spawn profile");
153                    Ok("spawned")
154                }
155                Route::Routine(target) => {
156                    let routine_letter = RoutineLetter {
157                        to: letter.nick.clone(),
158                        from: letter.from.clone(),
159                        event_id: letter.event_id.clone(),
160                        subject: None,
161                        room: None,
162                        body: letter.body.clone(),
163                        reply: None,
164                        note: None,
165                    };
166                    let nick = letter.nick.clone();
167                    match tokio::task::spawn_blocking(move || deliver_routine(&target, &routine_letter)).await {
168                        Ok(Ok(_)) => {
169                            tracing::info!(nick = %nick, "letter delivered to a routine");
170                            Ok("routine")
171                        }
172                        Ok(Err(e)) => Err(e),
173                        Err(_) => Err("routine delivery task failed".into()),
174                    }
175                }
176                _ => Err(why),
177            },
178        }
179    }
180}
hatchery_node/server/node_wake.rs

hatchery_node/server/
node_wake.rs
