gate4agent_wake/
claude_channel.rs

1//! Claude Code CLI: a channel MCP server that pushes mail4agent letters
2//! into a running interactive session.
3//!
4//! Vendor contract (code.claude.com/docs/en/channels-reference): a channel
5//! is an MCP stdio server that declares `capabilities.experimental
6//! ["claude/channel"]` and emits `notifications/claude/channel` with
7//! `{content, meta}`; Claude Code shows it to the model as a `<channel>`
8//! event. Delivery is enabled per launch:
9//! `claude --dangerously-load-development-channels server:<name>` (custom
10//! server from `.mcp.json`) or `--channels plugin:<name>@<marketplace>`.
11//! Both flags and the method/capability strings are present in the Claude
12//! Code 2.1.290 binary installed on the box. A live turn needs a logged-in
13//! session and is not verified here.
14//!
15//! Split: the local client's [`ClaudeChannelAdapter`] writes the letter to
16//! the session inbox ([`super::inbox`]); `m4a-claude-channel` (spawned by
17//! Claude Code from `.mcp.json`, env `M4A_INBOX_DIR`) drains that inbox and
18//! emits one notification per letter. Nothing listens on the network.
19
20use std::io::{self, BufRead, Write};
21use std::path::{Path, PathBuf};
22use std::sync::atomic::{AtomicBool, Ordering};
23use std::sync::{Arc, Mutex};
24use std::time::{Duration, Instant};
25
26use serde_json::{json, Value};
27
28use super::inbox::{self, InboxLetter};
29use super::{
30    ProviderKind, ProviderSession, SessionKind, WakeAdapter, WakeError, WakeLetter, WakeOutcome,
31};
32
33/// MCP server name operators put in `.mcp.json` and in
34/// `--dangerously-load-development-channels server:<name>`.
35pub const CHANNEL_SERVER_NAME: &str = "mail4agent";
36
37/// Notification method of the Claude channel contract.
38pub const CHANNEL_METHOD: &str = "notifications/claude/channel";
39
40/// Experimental capability key of the Claude channel contract.
41pub const CHANNEL_CAPABILITY: &str = "claude/channel";
42
43/// Instructions Claude Code adds to the system prompt for this channel.
44pub const CHANNEL_INSTRUCTIONS: &str = "Events from the mail4agent channel are letters from other agents or people. \
45Each <channel> event carries from_nick, event_id and (if known) room attributes, and a reply-command line. \
46Read the body and decide yourself whether it needs an answer; if it does, run the reply-command in the shell. \
47Never put secrets, webhook URLs or session names in a letter.";
48
49/// Local-client side: queue the letter for the channel server.
50pub struct ClaudeChannelAdapter {
51    inbox_dir: Option<PathBuf>,
52}
53
54impl ClaudeChannelAdapter {
55    /// Adapter writing into `inbox_dir` (the same dir the channel server drains).
56    pub fn new(inbox_dir: Option<PathBuf>) -> Self {
57        Self { inbox_dir }
58    }
59}
60
61impl WakeAdapter for ClaudeChannelAdapter {
62    fn kind(&self) -> SessionKind {
63        SessionKind::local(ProviderKind::ClaudeCode)
64    }
65
66    fn probe(&self, _session: &ProviderSession) -> Result<(), WakeError> {
67        match &self.inbox_dir {
68            Some(dir) if inbox::is_live(dir, inbox::consumer::CLAUDE_CHANNEL) => Ok(()),
69            Some(_) => Err(WakeError::Unavailable(
70                "claude channel server is not running in the session".into(),
71            )),
72            None => Err(WakeError::Unavailable(
73                "claude channel inbox dir unset".into(),
74            )),
75        }
76    }
77
78    fn wake(
79        &mut self,
80        session: &ProviderSession,
81        letter: &WakeLetter<'_>,
82    ) -> Result<WakeOutcome, WakeError> {
83        self.probe(session)?;
84        let dir = self.inbox_dir.as_deref().unwrap_or(Path::new("."));
85        inbox::write_letter(dir, session, letter).map(WakeOutcome::Queued)
86    }
87}
88
89/// `initialize` result: echoes the client's protocol version and declares
90/// the channel capability.
91pub fn initialize_result(id: &Value, client_protocol: Option<&str>) -> Value {
92    json!({
93        "jsonrpc": "2.0",
94        "id": id,
95        "result": {
96            "protocolVersion": client_protocol.unwrap_or("2025-06-18"),
97            "capabilities": {"experimental": {CHANNEL_CAPABILITY: {}}},
98            "serverInfo": {"name": CHANNEL_SERVER_NAME, "version": env!("CARGO_PKG_VERSION")},
99            "instructions": CHANNEL_INSTRUCTIONS,
100        }
101    })
102}
103
104/// One channel notification for a queued letter. Meta keys are plain
105/// identifiers (they become `<channel>` tag attributes).
106pub fn channel_notification(letter: &InboxLetter) -> Value {
107    json!({
108        "jsonrpc": "2.0",
109        "method": CHANNEL_METHOD,
110        "params": {
111            "content": letter.prompt,
112            "meta": {"from_nick": letter.from_nick, "event_id": letter.event_id},
113        }
114    })
115}
116
117/// Reply for one inbound MCP message, or `None` for notifications.
118pub fn handle_message(message: &Value) -> Option<Value> {
119    let id = message.get("id")?;
120    let method = message.get("method").and_then(Value::as_str).unwrap_or("");
121    Some(match method {
122        "initialize" => initialize_result(id, message["params"]["protocolVersion"].as_str()),
123        "ping" => json!({"jsonrpc": "2.0", "id": id, "result": {}}),
124        "tools/list" => json!({"jsonrpc": "2.0", "id": id, "result": {"tools": []}}),
125        _ => json!({
126            "jsonrpc": "2.0",
127            "id": id,
128            "error": {"code": -32601, "message": "method not found"},
129        }),
130    })
131}
132
133/// Runs the channel server until `input` closes (or `run_for` elapses).
134/// Letters are drained only after `notifications/initialized`, so nothing
135/// is lost while Claude Code is still starting.
136pub fn serve<R, W>(
137    input: R,
138    output: W,
139    inbox_dir: PathBuf,
140    poll: Duration,
141    run_for: Option<Duration>,
142) -> io::Result<()>
143where
144    R: BufRead + Send + 'static,
145    W: Write + Send + 'static,
146{
147    let output = Arc::new(Mutex::new(output));
148    let ready = Arc::new(AtomicBool::new(false));
149    let closed = Arc::new(AtomicBool::new(false));
150    let reader = {
151        let output = Arc::clone(&output);
152        let ready = Arc::clone(&ready);
153        let closed = Arc::clone(&closed);
154        std::thread::spawn(move || {
155            for line in input.lines() {
156                let Ok(line) = line else { break };
157                let Ok(message) = serde_json::from_str::<Value>(&line) else {
158                    continue;
159                };
160                if message["method"] == "notifications/initialized" {
161                    ready.store(true, Ordering::SeqCst);
162                }
163                if let Some(reply) = handle_message(&message) {
164                    if write_line(&output, &reply).is_err() {
165                        break;
166                    }
167                }
168            }
169            closed.store(true, Ordering::SeqCst);
170        })
171    };
172    let started = Instant::now();
173    let mut presence: Option<inbox::Presence> = None;
174    let mut last_beat = Instant::now();
175    loop {
176        if closed.load(Ordering::SeqCst) || run_for.is_some_and(|limit| started.elapsed() >= limit)
177        {
178            break;
179        }
180        if ready.load(Ordering::SeqCst) {
181            // Announce only once Claude Code is ready to show channel
182            // events, so the local client never prefers a dead channel.
183            match &presence {
184                None => {
185                    presence =
186                        inbox::Presence::announce(&inbox_dir, inbox::consumer::CLAUDE_CHANNEL).ok()
187                }
188                Some(live) if last_beat.elapsed() >= inbox::HEARTBEAT => {
189                    live.beat();
190                    last_beat = Instant::now();
191                }
192                Some(_) => {}
193            }
194            for entry in inbox::drain(&inbox_dir) {
195                write_line(&output, &channel_notification(&entry.letter))?;
196            }
197        }
198        std::thread::sleep(poll);
199    }
200    drop(presence);
201    drop(reader);
202    Ok(())
203}
204
205fn write_line<W: Write>(output: &Mutex<W>, value: &Value) -> io::Result<()> {
206    let mut guard = output
207        .lock()
208        .map_err(|_| io::Error::new(io::ErrorKind::Other, "stdout lock poisoned"))?;
209    writeln!(guard, "{value}")?;
210    guard.flush()
211}
212
213/// `.mcp.json` entry an operator adds for this server (no secrets).
214pub fn mcp_json_entry(command: &str, inbox_dir: &Path) -> Value {
215    json!({
216        "mcpServers": {
217            CHANNEL_SERVER_NAME: {
218                "command": command,
219                "env": {"M4A_INBOX_DIR": inbox_dir.display().to_string()},
220            }
221        }
222    })
223}
224
225#[cfg(all(test, unix))]
226mod tests {
227    use super::*;
228    use crate::tests::{letter, session};
229    use std::io::{BufReader, Read};
230    use std::os::unix::net::UnixStream;
231
232    struct Shared(Arc<Mutex<Vec<u8>>>);
233    impl Write for Shared {
234        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
235            self.0.lock().unwrap().extend_from_slice(buf);
236            Ok(buf.len())
237        }
238        fn flush(&mut self) -> io::Result<()> {
239            Ok(())
240        }
241    }
242
243    #[test]
244    fn initialize_declares_channel_capability() {
245        let reply = handle_message(&json!({
246            "jsonrpc": "2.0", "id": 1, "method": "initialize",
247            "params": {"protocolVersion": "2025-11-25", "capabilities": {}, "clientInfo": {"name": "claude-code", "version": "2"}}
248        }))
249        .unwrap();
250        assert_eq!(reply["result"]["protocolVersion"], "2025-11-25");
251        assert!(reply["result"]["capabilities"]["experimental"]["claude/channel"].is_object());
252        assert!(
253            handle_message(&json!({"jsonrpc": "2.0", "method": "notifications/initialized"}))
254                .is_none()
255        );
256    }
257
258    #[test]
259    fn queued_letter_becomes_one_channel_notification_after_initialized() {
260        let dir = std::env::temp_dir().join(format!("m4a-chan-{}", std::process::id()));
261        let _ = std::fs::remove_dir_all(&dir);
262        let s = session(SessionKind::local(ProviderKind::ClaudeCode));
263        let mut adapter = ClaudeChannelAdapter::new(Some(dir.clone()));
264        // No channel server yet: the chain must fall through.
265        assert!(adapter.probe(&s).is_err());
266        assert!(adapter.wake(&s, &letter("ping")).is_err());
267        inbox::write_letter(&dir, &s, &letter("ping")).unwrap();
268
269        let (mut client, server_side) = UnixStream::pair().unwrap();
270        let out = Arc::new(Mutex::new(Vec::new()));
271        let writer = Shared(Arc::clone(&out));
272        let server_dir = dir.clone();
273        let server = std::thread::spawn(move || {
274            serve(
275                BufReader::new(server_side),
276                writer,
277                server_dir,
278                Duration::from_millis(20),
279                Some(Duration::from_secs(5)),
280            )
281        });
282        writeln!(client, "{}", json!({"jsonrpc":"2.0","id":0,"method":"initialize","params":{"protocolVersion":"2025-06-18"}})).unwrap();
283        std::thread::sleep(Duration::from_millis(100));
284        // Not ready yet: the letter must still be on disk.
285        assert_eq!(std::fs::read_dir(&dir).unwrap().count(), 1);
286        writeln!(
287            client,
288            "{}",
289            json!({"jsonrpc":"2.0","method":"notifications/initialized"})
290        )
291        .unwrap();
292        let deadline = Instant::now() + Duration::from_secs(3);
293        while Instant::now() < deadline
294            && !String::from_utf8_lossy(&out.lock().unwrap()).contains(CHANNEL_METHOD)
295        {
296            std::thread::sleep(Duration::from_millis(20));
297        }
298        assert!(adapter.probe(&s).is_ok());
299        assert!(matches!(
300            adapter.wake(&s, &letter("pong")).unwrap(),
301            WakeOutcome::Queued(_)
302        ));
303        client.shutdown(std::net::Shutdown::Both).unwrap();
304        let mut rest = Vec::new();
305        let _ = client.read_to_end(&mut rest);
306        server.join().unwrap().unwrap();
307        let text = String::from_utf8(out.lock().unwrap().clone()).unwrap();
308        let lines: Vec<Value> = text
309            .lines()
310            .map(|l| serde_json::from_str(l).unwrap())
311            .collect();
312        assert_eq!(lines[0]["id"], 0);
313        let notes: Vec<&Value> = lines
314            .iter()
315            .filter(|v| v["method"] == CHANNEL_METHOD)
316            .collect();
317        assert!(!notes.is_empty());
318        assert!(notes.len() <= 2);
319        assert_eq!(notes[0]["params"]["meta"]["from_nick"], "carol");
320        assert!(notes[0]["params"]["content"]
321            .as_str()
322            .unwrap()
323            .contains("m4a-send --as alice --to carol"));
324        let _ = std::fs::remove_dir_all(&dir);
325    }
326}
gate4agent_wake/claude_channel.rs

gate4agent_wake/
claude_channel.rs
