fno_agents/
claude_attach.rs

1//! Speak Claude's daemon `control.sock`: the `op:'attach'` handshake + the
2//! newline-delimited JSON transport.
3//!
4//! G1 substrate (epic x-07c1, node x-26df). The Phase-0 spike retired the
5//! held-attach *keepalive* premise (an idle `claude --bg` session and an un-held
6//! control both survived 65min idle, so the attach was not what kept it live).
7//! So footnote does NOT hold a session for liveness. `op:'attach'` survives for a
8//! different reason: G2/grid attaches to pull the PTY frame STREAM for rendering a
9//! tile. The drive primitive (`op:'reply'` + transcript confirm) lives in
10//! [`crate::claude_drive`]; the roster read is [`crate::claude_roster`].
11//!
12//! Wire contracts pinned to claude-code **2.1.195** (readiness brief
13//! `2026-06-27-phase0-held-attach-readiness.md`):
14//!   - framing: newline-delimited JSON over `control.sock` (candidate A). `[corroborated]`
15//!   - attach request zod: `{proto:1, op:'attach', short:/^[a-f0-9]{8}$/,
16//!     auth?:string, cols:int, rows:int, caps:{terminal,mux,ssh,...}}`. `[corroborated]`
17//!   - attach-OK reply: `{ok:true, op:'attach', decModes, via, tempo, state}`. `[corroborated]`
18//!   - auth = the daemon `control.key` (32-hex), NOT the per-worker `ptyAuth`.
19//!
20//! The live socket sits behind [`ControlTransport`] so the handshake is
21//! unit-tested with a fake; the live `UnixStream` path is one thin type.
22
23use std::io::{self, BufRead, BufReader, Write};
24use std::os::unix::net::UnixStream;
25use std::path::Path;
26use std::time::Duration;
27
28use serde::Deserialize;
29
30/// `proto` field value the 2.1.195 daemon expects (`gp=1`); a mismatch yields an
31/// `EPROTO` "restart claude". `[corroborated]`
32pub const ATTACH_PROTO: u32 = 1;
33
34/// An `op:'attach'` request. `auth` is optional: a same-uid socket attach may
35/// omit it ("legacy client, allowed via peerUid"); when the daemon `control.key`
36/// is readable we present it.
37#[derive(Debug, Clone, PartialEq)]
38pub struct AttachRequest {
39    /// 8-hex worker short id (the roster map key) -- a wire value, used only here
40    /// at the `control.sock` boundary, never as a footnote-side identity.
41    pub short: String,
42    /// Daemon control key, or `None` for the same-uid no-auth path.
43    pub auth: Option<String>,
44    pub cols: u32,
45    pub rows: u32,
46}
47
48impl AttachRequest {
49    /// Construct an attach with an explicit window size.
50    pub fn new(short: impl Into<String>, auth: Option<String>, cols: u32, rows: u32) -> Self {
51        AttachRequest {
52            short: short.into(),
53            auth,
54            cols,
55            rows,
56        }
57    }
58
59    /// Attach to pull the PTY frame stream (G2 tile render). A modest default
60    /// window; the daemon accepts a non-TTY attacher and streams frames.
61    pub fn for_frame_stream(short: impl Into<String>, auth: Option<String>) -> Self {
62        Self::new(short, auth, 80, 24)
63    }
64
65    /// Serialize to the newline-terminated JSON line the daemon reads. `caps` is a
66    /// REQUIRED object; we send the minimal valid shape (`terminal`/`mux` are
67    /// nullable-required, `ssh` required) -- the finding's `colorLevel`/`browser`
68    /// keys are NOT in the schema and are omitted. `[corroborated]`
69    pub fn to_json_line(&self) -> String {
70        let mut obj = serde_json::Map::new();
71        obj.insert("proto".into(), ATTACH_PROTO.into());
72        obj.insert("op".into(), "attach".into());
73        obj.insert("short".into(), self.short.clone().into());
74        if let Some(a) = &self.auth {
75            obj.insert("auth".into(), a.clone().into());
76        }
77        obj.insert("cols".into(), self.cols.into());
78        obj.insert("rows".into(), self.rows.into());
79        obj.insert(
80            "caps".into(),
81            serde_json::json!({"terminal": null, "mux": null, "ssh": false}),
82        );
83        let mut line = serde_json::Value::Object(obj).to_string();
84        line.push('\n');
85        line
86    }
87}
88
89/// The daemon's attach-OK reply (`ok:true`). Response-only fields per the lane-c
90/// correction: `decModes`/`tempo`/`state`/`via` are NOT request fields.
91#[derive(Debug, Clone, PartialEq, Deserialize)]
92pub struct AttachOk {
93    #[serde(default)]
94    pub dec_modes: Vec<String>,
95    #[serde(default)]
96    pub via: Option<String>,
97    /// `"active" | "blocked"` -- the session's input tempo at attach time.
98    #[serde(default)]
99    pub tempo: Option<String>,
100    /// `"running"` for a live session.
101    #[serde(default)]
102    pub state: Option<String>,
103}
104
105/// Why an attach did not complete. `Refused` carries the daemon's own code where
106/// it sent one (`EPROTO`, `ERESPAWNING`, ...); `Malformed` is a reply we could not
107/// parse (or an I/O error around the handshake).
108#[derive(Debug, Clone, PartialEq)]
109pub enum AttachError {
110    Refused {
111        code: Option<String>,
112        detail: String,
113    },
114    Malformed(String),
115}
116
117impl std::fmt::Display for AttachError {
118    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
119        match self {
120            AttachError::Refused { code, detail } => match code {
121                Some(c) => write!(f, "attach refused ({c}): {detail}"),
122                None => write!(f, "attach refused: {detail}"),
123            },
124            AttachError::Malformed(m) => write!(f, "malformed attach reply: {m}"),
125        }
126    }
127}
128
129impl std::error::Error for AttachError {}
130
131/// Parse one reply line into an `AttachOk` or an `AttachError`. `ok:true` ->
132/// `AttachOk`; `ok:false` (or absent) -> `Refused`, mining `code`/`error`/`reason`
133/// for the daemon's reason; non-JSON -> `Malformed`.
134pub fn parse_attach_reply(line: &str) -> Result<AttachOk, AttachError> {
135    let v: serde_json::Value = match serde_json::from_str(line.trim()) {
136        Ok(v) => v,
137        Err(e) => return Err(AttachError::Malformed(format!("{e}: {line:?}"))),
138    };
139    let ok = v
140        .get("ok")
141        .and_then(serde_json::Value::as_bool)
142        .unwrap_or(false);
143    if ok {
144        // The reply is camelCase; pull decModes -> dec_modes explicitly so AttachOk
145        // stays snake-cased without a rename attr on a hand-parsed value.
146        let dec_modes = v
147            .get("decModes")
148            .and_then(serde_json::Value::as_array)
149            .map(|a| {
150                a.iter()
151                    .filter_map(|x| x.as_str().map(str::to_string))
152                    .collect()
153            })
154            .unwrap_or_default();
155        let str_field = |k: &str| {
156            v.get(k)
157                .and_then(serde_json::Value::as_str)
158                .map(str::to_string)
159        };
160        Ok(AttachOk {
161            dec_modes,
162            via: str_field("via"),
163            tempo: str_field("tempo"),
164            state: str_field("state"),
165        })
166    } else {
167        let code = v
168            .get("code")
169            .and_then(serde_json::Value::as_str)
170            .map(str::to_string);
171        let detail = v
172            .get("error")
173            .or_else(|| v.get("reason"))
174            .or_else(|| v.get("message"))
175            .and_then(serde_json::Value::as_str)
176            .unwrap_or("attach not accepted")
177            .to_string();
178        Err(AttachError::Refused { code, detail })
179    }
180}
181
182/// The newline-delimited JSON transport over a daemon socket. Behind a trait so
183/// the handshake + the drive inject ([`crate::claude_drive`]) are exercised with a
184/// fake in unit tests and the real `UnixStream` path is the only thing that needs
185/// a live daemon.
186pub trait ControlTransport {
187    /// Write one already-newline-terminated line.
188    fn send_line(&mut self, line: &str) -> io::Result<()>;
189    /// Read the next line (without the trailing newline). `Ok(None)` == EOF.
190    fn recv_line(&mut self) -> io::Result<Option<String>>;
191}
192
193/// Live `control.sock` transport: a `UnixStream` with a buffered line reader.
194pub struct UnixControlTransport {
195    write: UnixStream,
196    read: BufReader<UnixStream>,
197}
198
199impl UnixControlTransport {
200    /// Connect to the daemon `control.sock` at `path`. A non-draining reader gets
201    /// `c.destroy()`'d (lane-a backpressure), so a frame-stream consumer must keep
202    /// calling `recv_line`.
203    pub fn connect(path: &Path) -> io::Result<Self> {
204        let stream = UnixStream::connect(path)?;
205        stream.set_read_timeout(Some(Duration::from_secs(30)))?;
206        let read = BufReader::new(stream.try_clone()?);
207        Ok(UnixControlTransport {
208            write: stream,
209            read,
210        })
211    }
212}
213
214impl ControlTransport for UnixControlTransport {
215    fn send_line(&mut self, line: &str) -> io::Result<()> {
216        self.write.write_all(line.as_bytes())?;
217        self.write.flush()
218    }
219
220    fn recv_line(&mut self) -> io::Result<Option<String>> {
221        let mut buf = String::new();
222        let n = self.read.read_line(&mut buf)?;
223        if n == 0 {
224            return Ok(None); // EOF
225        }
226        // Trim trailing newline in place rather than allocating a fresh String.
227        let len = buf.trim_end_matches(['\n', '\r']).len();
228        buf.truncate(len);
229        Ok(Some(buf))
230    }
231}
232
233/// Perform the attach handshake over `t`: send the request, read + parse the
234/// first reply. The precursor a frame-stream consumer (G2) or the drive primitive
235/// runs before it streams frames / injects a turn.
236pub fn perform_attach<T: ControlTransport>(
237    t: &mut T,
238    req: &AttachRequest,
239) -> Result<AttachOk, AttachError> {
240    let line = req.to_json_line();
241    t.send_line(&line)
242        .map_err(|e| AttachError::Malformed(format!("send: {e}")))?;
243    match t.recv_line() {
244        Ok(Some(reply)) => parse_attach_reply(&reply),
245        Ok(None) => Err(AttachError::Refused {
246            code: Some("EOF".into()),
247            detail: "daemon closed before attach reply".into(),
248        }),
249        Err(e) => Err(AttachError::Malformed(format!("recv: {e}"))),
250    }
251}
252
253/// A raw PTY byte stream from a `control.sock` `op:'attach'`. After the single
254/// newline-terminated attach-OK JSON line, the daemon streams the terminal's raw
255/// VT/ANSI bytes -- pinned live on 2.1.195: NOT JSON-wrapped, NOT length-prefixed
256/// (the daemon unwraps the internal ptySock framing before streaming to
257/// attachers). So G2 renders a tile by feeding these bytes straight into its
258/// terminal-emulator pane. `\r`/`\n` are ordinary terminal content here, so the
259/// stream is read as raw bytes, never line-split.
260#[derive(Debug)]
261pub struct FrameStream<R: io::Read> {
262    reader: BufReader<R>,
263}
264
265impl<R: io::Read> io::Read for FrameStream<R> {
266    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
267        self.reader.read(buf)
268    }
269}
270
271/// Attach for the frame stream: send the `op:'attach'` request on `writer`, read
272/// + parse the one handshake line from `reader`, then hand back a [`FrameStream`]
273/// positioned at the first raw PTY byte. Bytes the handshake read buffered past
274/// the reply's `\n` are preserved -- they are the first frame bytes, so they must
275/// not be dropped. Reader and writer are split so a live caller passes a
276/// `try_clone`'d `UnixStream` writer plus the stream as the reader, while tests
277/// pass a `Vec`/`Cursor`.
278pub fn attach_for_frames<R: io::Read, W: Write>(
279    mut writer: W,
280    reader: R,
281    req: &AttachRequest,
282) -> Result<(AttachOk, FrameStream<R>), AttachError> {
283    writer
284        .write_all(req.to_json_line().as_bytes())
285        .and_then(|()| writer.flush())
286        .map_err(|e| AttachError::Malformed(format!("send: {e}")))?;
287    let mut reader = BufReader::new(reader);
288    let mut line = String::new();
289    match reader.read_line(&mut line) {
290        Ok(0) => Err(AttachError::Refused {
291            code: Some("EOF".into()),
292            detail: "daemon closed before attach reply".into(),
293        }),
294        // A reply without the protocol newline is a truncated handshake (the
295        // daemon closed mid-write): `read_line` returns `Ok(n>0)` and the JSON
296        // may even parse, but the stream that follows is empty. Treat it as a
297        // refusal rather than handing back a FrameStream that instantly EOFs.
298        Ok(_) if !line.ends_with('\n') => Err(AttachError::Refused {
299            code: Some("EOF".into()),
300            detail: "daemon closed before complete attach reply".into(),
301        }),
302        Ok(_) => {
303            let ok = parse_attach_reply(&line)?;
304            Ok((ok, FrameStream { reader }))
305        }
306        Err(e) => Err(AttachError::Malformed(format!("recv: {e}"))),
307    }
308}
309
310#[cfg(test)]
311mod tests {
312    use super::*;
313    use std::collections::VecDeque;
314    use std::io::{Cursor, Read};
315
316    /// A scripted in-memory transport: replays queued reply lines, records sends.
317    struct FakeTransport {
318        replies: VecDeque<Option<String>>,
319        sent: Vec<String>,
320        recv_err: bool,
321    }
322    impl FakeTransport {
323        fn new(replies: Vec<Option<&str>>) -> Self {
324            FakeTransport {
325                replies: replies.into_iter().map(|r| r.map(str::to_string)).collect(),
326                sent: Vec::new(),
327                recv_err: false,
328            }
329        }
330    }
331    impl ControlTransport for FakeTransport {
332        fn send_line(&mut self, line: &str) -> io::Result<()> {
333            self.sent.push(line.to_string());
334            Ok(())
335        }
336        fn recv_line(&mut self) -> io::Result<Option<String>> {
337            if self.recv_err {
338                return Err(io::Error::other("boom"));
339            }
340            Ok(self.replies.pop_front().flatten())
341        }
342    }
343
344    #[test]
345    fn attach_request_serializes_to_pinned_schema() {
346        let req = AttachRequest::for_frame_stream("a1b2c3d4", Some("deadbeef".into()));
347        let line = req.to_json_line();
348        assert!(line.ends_with('\n'));
349        let v: serde_json::Value = serde_json::from_str(line.trim()).unwrap();
350        assert_eq!(v["proto"], 1);
351        assert_eq!(v["op"], "attach");
352        assert_eq!(v["short"], "a1b2c3d4");
353        assert_eq!(v["auth"], "deadbeef");
354        assert_eq!(v["cols"], 80);
355        assert_eq!(v["rows"], 24);
356        assert!(v["caps"]["terminal"].is_null());
357        assert!(v["caps"]["mux"].is_null());
358        assert_eq!(v["caps"]["ssh"], false);
359        assert!(v["caps"].get("colorLevel").is_none());
360    }
361
362    #[test]
363    fn attach_request_omits_auth_for_same_uid_path() {
364        let req = AttachRequest::for_frame_stream("a1b2c3d4", None);
365        let v: serde_json::Value = serde_json::from_str(req.to_json_line().trim()).unwrap();
366        assert!(v.get("auth").is_none(), "no-auth path must omit the key");
367    }
368
369    #[test]
370    fn parse_ok_reply() {
371        let ok = parse_attach_reply(
372            r#"{"ok":true,"op":"attach","decModes":["1049","2004"],"via":"spare","tempo":"active","state":"running"}"#,
373        )
374        .unwrap();
375        assert_eq!(ok.dec_modes, vec!["1049", "2004"]);
376        assert_eq!(ok.via.as_deref(), Some("spare"));
377        assert_eq!(ok.tempo.as_deref(), Some("active"));
378        assert_eq!(ok.state.as_deref(), Some("running"));
379    }
380
381    #[test]
382    fn parse_refused_reply_mines_code_and_reason() {
383        let err = parse_attach_reply(r#"{"ok":false,"code":"EPROTO","error":"restart claude"}"#)
384            .unwrap_err();
385        assert_eq!(
386            err,
387            AttachError::Refused {
388                code: Some("EPROTO".into()),
389                detail: "restart claude".into()
390            }
391        );
392    }
393
394    #[test]
395    fn parse_non_json_is_malformed() {
396        assert!(matches!(
397            parse_attach_reply("not a frame"),
398            Err(AttachError::Malformed(_))
399        ));
400    }
401
402    #[test]
403    fn perform_attach_happy_path() {
404        let mut t =
405            FakeTransport::new(vec![Some(r#"{"ok":true,"op":"attach","state":"running"}"#)]);
406        let req = AttachRequest::for_frame_stream("a1b2c3d4", None);
407        let ok = perform_attach(&mut t, &req).unwrap();
408        assert_eq!(ok.state.as_deref(), Some("running"));
409        assert_eq!(t.sent.len(), 1);
410        assert!(t.sent[0].contains("\"op\":\"attach\""));
411    }
412
413    #[test]
414    fn perform_attach_eof_is_refused() {
415        let mut t = FakeTransport::new(vec![None]);
416        let req = AttachRequest::for_frame_stream("a1b2c3d4", None);
417        let err = perform_attach(&mut t, &req).unwrap_err();
418        assert!(matches!(err, AttachError::Refused { .. }));
419    }
420
421    #[test]
422    fn perform_attach_recv_error_is_malformed() {
423        let mut t = FakeTransport::new(vec![]);
424        t.recv_err = true;
425        let req = AttachRequest::for_frame_stream("a1b2c3d4", None);
426        let err = perform_attach(&mut t, &req).unwrap_err();
427        assert!(matches!(err, AttachError::Malformed(_)));
428    }
429
430    // Build the server-side bytes a daemon would write back: one attach-OK line
431    // then the raw PTY tail.
432    fn server_bytes(reply: &str, raw_tail: &[u8]) -> Cursor<Vec<u8>> {
433        let mut v = format!("{reply}\n").into_bytes();
434        v.extend_from_slice(raw_tail);
435        Cursor::new(v)
436    }
437
438    #[test]
439    fn attach_for_frames_returns_ok_then_raw_tail() {
440        let raw = b"\x1b[2J\x1b[Hhello world";
441        let reader = server_bytes(r#"{"ok":true,"op":"attach","state":"running"}"#, raw);
442        let mut writer: Vec<u8> = Vec::new();
443        let req = AttachRequest::for_frame_stream("a1b2c3d4", Some("k".into()));
444        let (ok, mut stream) = attach_for_frames(&mut writer, reader, &req).unwrap();
445        assert_eq!(ok.state.as_deref(), Some("running"));
446        // The attach request went out on the writer.
447        assert!(String::from_utf8_lossy(&writer).contains("\"op\":\"attach\""));
448        // Everything after the handshake line is the raw frame stream, intact.
449        let mut got = Vec::new();
450        stream.read_to_end(&mut got).unwrap();
451        assert_eq!(got, raw);
452    }
453
454    #[test]
455    fn attach_for_frames_keeps_tail_buffered_with_handshake() {
456        // The raw tail arrives in the SAME recv as the handshake line; it must not
457        // be lost when read_line stops at the newline.
458        let raw = b"first-frame-bytes";
459        let reader = server_bytes(r#"{"ok":true,"op":"attach"}"#, raw);
460        let req = AttachRequest::for_frame_stream("a1b2c3d4", None);
461        let (_ok, mut stream) = attach_for_frames(Vec::new(), reader, &req).unwrap();
462        let mut got = Vec::new();
463        stream.read_to_end(&mut got).unwrap();
464        assert_eq!(got, raw);
465    }
466
467    #[test]
468    fn attach_for_frames_tail_with_embedded_newlines_is_not_split() {
469        // PTY content contains \r and \n as ordinary bytes; the stream must hand
470        // them back verbatim, never line-framed.
471        let raw = b"line1\r\nline2\nline3";
472        let reader = server_bytes(r#"{"ok":true,"op":"attach"}"#, raw);
473        let req = AttachRequest::for_frame_stream("a1b2c3d4", None);
474        let (_ok, mut stream) = attach_for_frames(Vec::new(), reader, &req).unwrap();
475        let mut got = Vec::new();
476        stream.read_to_end(&mut got).unwrap();
477        assert_eq!(got, raw);
478    }
479
480    #[test]
481    fn attach_for_frames_refused_propagates() {
482        let reader = server_bytes(
483            r#"{"ok":false,"code":"EPROTO","error":"restart claude"}"#,
484            b"",
485        );
486        let req = AttachRequest::for_frame_stream("a1b2c3d4", None);
487        let err = attach_for_frames(Vec::new(), reader, &req).unwrap_err();
488        assert_eq!(
489            err,
490            AttachError::Refused {
491                code: Some("EPROTO".into()),
492                detail: "restart claude".into()
493            }
494        );
495    }
496
497    #[test]
498    fn attach_for_frames_eof_before_reply_is_refused() {
499        let reader = Cursor::new(Vec::new());
500        let req = AttachRequest::for_frame_stream("a1b2c3d4", None);
501        let err = attach_for_frames(Vec::new(), reader, &req).unwrap_err();
502        assert!(matches!(err, AttachError::Refused { .. }));
503    }
504
505    #[test]
506    fn attach_for_frames_truncated_reply_without_newline_is_refused() {
507        // Valid JSON but the daemon closed before the protocol newline: the
508        // handshake is incomplete and the frame stream would be empty, so this
509        // must refuse rather than report a successful attach.
510        let reader = Cursor::new(br#"{"ok":true,"op":"attach","state":"running"}"#.to_vec());
511        let req = AttachRequest::for_frame_stream("a1b2c3d4", None);
512        let err = attach_for_frames(Vec::new(), reader, &req).unwrap_err();
513        assert!(matches!(err, AttachError::Refused { .. }));
514    }
515}
fno_agents/claude_attach.rs

fno_agents/
claude_attach.rs
