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}
29
30#[derive(Clone)]
31pub struct NodeWake {
32    shared: Arc<NodeShared>,
33    token: Arc<str>,
34    policy: Arc<crate::wake_pipeline::WakePolicy>,
35}
36
37struct Snapshot(BTreeMap<String, HostedTarget>);
38
39impl HostedSessions for Snapshot {
40    fn lookup(&self, nick: &str) -> Option<HostedTarget> {
41        self.0.get(nick).cloned()
42    }
43}
44
45fn kind_of(agent: &AgentId) -> Option<ProviderKind> {
46    ProviderKind::parse(crate::m4a_accounts::m4a_provider(agent.as_str())?)
47}
48
49impl NodeWake {
50    pub(super) fn new(shared: Arc<NodeShared>, token: String, policy: Arc<crate::wake_pipeline::WakePolicy>) -> Self {
51        Self { shared, token: token.into(), policy }
52    }
53
54    /// Constant-time check of the caller's token.
55    pub fn authorized(&self, token: &str) -> bool {
56        let (a, b) = (self.token.as_bytes(), token.as_bytes());
57        a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
58    }
59
60    fn snapshot(&self) -> (Snapshot, BTreeMap<String, SessionAddress>) {
61        let live = self.shared.m4a_live.lock().unwrap_or_else(|p| p.into_inner()).clone();
62        let bindings = self.shared.session_bindings.lock().unwrap_or_else(|p| p.into_inner());
63        let mut targets = BTreeMap::new();
64        let mut addresses = BTreeMap::new();
65        for (nick, (instance_id, agent)) in live {
66            let Some(provider) = kind_of(&agent) else { continue };
67            let binding = bindings.get(&instance_id);
68            if let Some(b) = binding {
69                addresses.insert(
70                    nick.clone(),
71                    SessionAddress { workspace_id: b.workspace_id.clone(), session: SessionKey { instance_id, generation: b.generation } },
72                );
73            }
74            let spawn_allowed = self.policy.spawn_profile(&nick).is_some();
75            targets.insert(nick.clone(), HostedTarget { nick, provider, live: binding.is_some(), spawn_allowed });
76        }
77        (Snapshot(targets), addresses)
78    }
79
80    /// Wakes the session that owns `letter.nick`. `Ok` names what happened; `Err` is safe to
81    /// log (it never carries the letter text).
82    pub async fn wake(&self, letter: NodeLetter) -> Result<&'static str, String> {
83        let (sessions, addresses) = self.snapshot();
84        let wake = WakeLetter { body: &letter.body, from_nick: &letter.from, event_id: &letter.event_id, room: None, reaction: None };
85        match plan_hosted_wake(&sessions, &letter.nick, &wake) {
86            HostedWake::Submit { nick, prompt } => {
87                let prompt = if self.policy.deliver_body { prompt } else { crate::wake_pipeline::wake_line(&letter.from, &letter.event_id) };
88                let address = addresses.get(&nick).cloned().ok_or("session vanished")?;
89                let agent = self.shared.validate_address(&address).map_err(|_| "session is not live".to_owned())?;
90                let refused = |e: NodeFailure| {
91                    tracing::warn!(nick = %nick, code = ?e.code, "letter wake input refused");
92                    "session did not take the turn".to_owned()
93                };
94                let shape = if self.shared.require_session_runtime_policy(&address, ProviderRuntimeRequirement::SemanticPrompt).is_ok() {
95                    SessionShape::Semantic
96                } else {
97                    SessionShape::RawTerminal
98                };
99                match route(&self.policy, &nick, shape) {
100                    Route::Deliver => {
101                        let action = InputAction::SubmitPrompt(PromptPayload { text: prompt, framing: prompt_framing(&agent) });
102                        self.shared.dispatch_input_bounded(&address, action).await.map_err(refused)?;
103                    }
104                    Route::TerminalTyping => {
105                        // Terminal capability, granted explicitly for this nick: type the prompt
106                        // as one line, then Enter.
107                        let text = prompt.replace(['\n', '\r'], " ");
108                        self.shared.dispatch_input_bounded(&address, InputAction::TerminalText(TerminalText { text })).await.map_err(refused)?;
109                        self.shared.dispatch_input_bounded(&address, InputAction::TerminalControl(TerminalControl::ControlM)).await.map_err(refused)?;
110                    }
111                    Route::Refuse(why) => {
112                        tracing::info!(nick = %nick, "letter not typed: no terminal grant");
113                        return Err(why);
114                    }
115                    Route::Routine(_) => return Err("unexpected routine route for a live session".into()),
116                    Route::Spawn(_) => return Err("unexpected spawn route for a live session".into()),
117                }
118                tracing::info!(nick = %nick, "letter woke a live session");
119                Ok("submitted")
120            }
121            HostedWake::Spawn { nick, .. } => match self.policy.spawn_profile(&nick).cloned() {
122                Some(profile) => {
123                    tokio::task::spawn_blocking(move || launch_spawn_profile(&profile))
124                        .await
125                        .map_err(|_| "spawn profile task failed".to_owned())?
126                        .map_err(|e| e)?;
127                    tracing::info!(nick = %nick, "letter woke an offline session via spawn profile");
128                    Ok("spawned")
129                }
130                None => Err("spawn on letter is not enabled on this node".into()),
131            },
132            HostedWake::Refuse(why) => match route(&self.policy, &letter.nick, SessionShape::None) {
133                Route::Spawn(profile) => {
134                    let nick = letter.nick.clone();
135                    tokio::task::spawn_blocking(move || launch_spawn_profile(&profile))
136                        .await
137                        .map_err(|_| "spawn profile task failed".to_owned())?
138                        .map_err(|e| e)?;
139                    tracing::info!(nick = %nick, "letter woke an offline session via spawn profile");
140                    Ok("spawned")
141                }
142                Route::Routine(target) => {
143                    let routine_letter = RoutineLetter {
144                        to: letter.nick.clone(),
145                        from: letter.from.clone(),
146                        event_id: letter.event_id.clone(),
147                        subject: None,
148                        room: None,
149                        body: letter.body.clone(),
150                        reply: None,
151                        note: None,
152                    };
153                    let nick = letter.nick.clone();
154                    match tokio::task::spawn_blocking(move || deliver_routine(&target, &routine_letter)).await {
155                        Ok(Ok(_)) => {
156                            tracing::info!(nick = %nick, "letter delivered to a routine");
157                            Ok("routine")
158                        }
159                        Ok(Err(e)) => Err(e),
160                        Err(_) => Err("routine delivery task failed".into()),
161                    }
162                }
163                _ => Err(why),
164            },
165        }
166    }
167}
hatchery_node/server/node_wake.rs

hatchery_node/server/
node_wake.rs
