fno_agents/
worker.rs

1//! Per-agent worker shim (Wave 3, task 3.3 — Outcome B).
2//!
3//! Wave 0 proved that a child on a PTY whose master the *daemon* owns is
4//! SIGHUP'd and dies the instant the daemon closes the master. The locked fix
5//! (Outcome B) is this: a per-agent worker process owns the PTY master and
6//! **outlives the daemon**. The daemon spawns one worker per PTY-managed agent,
7//! puts it in its own process group (so a kill of the daemon's group does not
8//! reach it), and talks to it over `<short_id>/worker.sock`. On daemon restart,
9//! the recovery sweep rediscovers live workers by scanning for their sockets
10//! and reattaches.
11//!
12//! The worker is intentionally tiny and single-client: only the daemon connects
13//! to it, so requests are handled serially on a current-thread runtime. There
14//! is no `Send` requirement (the [`PtySession`] is never moved across tasks),
15//! which sidesteps the `MasterPty: !Sync` constraint cleanly.
16
17use crate::events::EventEmitter;
18use crate::paths::{self, AgentsHome};
19use crate::protocol::{read_request, write_response, ErrorCode, ProtocolError, Request, Response};
20use crate::pty::{PtySession, DEFAULT_OUTPUT_RING_BYTES};
21use crate::state::{self, AgentState};
22use crate::AgentStatus;
23use base64::Engine as _;
24use portable_pty::CommandBuilder;
25use serde_json::json;
26use std::path::PathBuf;
27use std::time::Duration;
28use tokio::net::{UnixListener, UnixStream};
29
30/// How the daemon launches a worker: provider argv + where + terminal size.
31#[derive(Debug, Clone)]
32pub struct WorkerConfig {
33    pub short_id: String,
34    pub home: PathBuf,
35    pub cwd: PathBuf,
36    /// Provider command line: `argv[0]` is the program, the rest its args.
37    pub argv: Vec<String>,
38    pub rows: u16,
39    pub cols: u16,
40    pub ring_bytes: usize,
41}
42
43impl WorkerConfig {
44    /// Build from the worker binary's parsed args, defaulting the ring size.
45    pub fn new(
46        short_id: impl Into<String>,
47        home: impl Into<PathBuf>,
48        cwd: impl Into<PathBuf>,
49        argv: Vec<String>,
50    ) -> Self {
51        WorkerConfig {
52            short_id: short_id.into(),
53            home: home.into(),
54            cwd: cwd.into(),
55            argv,
56            rows: 24,
57            cols: 80,
58            ring_bytes: DEFAULT_OUTPUT_RING_BYTES,
59        }
60    }
61}
62
63#[derive(Debug, thiserror::Error)]
64pub enum WorkerError {
65    #[error("worker config: no provider argv given")]
66    NoArgv,
67    #[error("pty: {0}")]
68    Pty(#[from] crate::pty::PtyError),
69    #[error("io: {0}")]
70    Io(#[from] std::io::Error),
71    #[error("state: {0}")]
72    State(#[from] state::StateError),
73}
74
75/// Run the worker: spawn the PTY child, publish `state.json`, bind
76/// `worker.sock`, and serve daemon RPCs until the child exits or a
77/// `worker.shutdown` arrives. Returns when the worker should exit.
78pub async fn run(cfg: WorkerConfig) -> Result<(), WorkerError> {
79    if cfg.argv.is_empty() {
80        return Err(WorkerError::NoArgv);
81    }
82    let home = AgentsHome::at(&cfg.home);
83    let sock_path = home.worker_sock(&cfg.short_id);
84    let state_path = home.state_json(&cfg.short_id);
85
86    // Spawn the PTY child the worker will own for its whole lifetime.
87    let cmd = build_child_command(&cfg);
88    let pty = PtySession::spawn(cmd, cfg.rows, cfg.cols, cfg.ring_bytes)?;
89
90    // Publish live state (status=live, pty.active=true) so the daemon and the
91    // daemon-down read path both see a coherent picture.
92    let mut st = AgentState::new_pty(&cfg.short_id);
93    st.status = AgentStatus::Live;
94    st.ready = true;
95    if let Some(p) = st.pty.as_mut() {
96        p.active = true;
97    }
98    state::write_state_atomic(&state_path, &st)?;
99
100    // Bind the worker socket (replace any stale socket from a prior incarnation)
101    // and lock it to mode 0600.
102    let _ = std::fs::remove_file(&sock_path);
103    if let Some(parent) = sock_path.parent() {
104        std::fs::create_dir_all(parent)?;
105    }
106    let listener = UnixListener::bind(&sock_path)?;
107    let _ = paths::set_file_mode_0600(&sock_path);
108
109    let mut liveness = tokio::time::interval(Duration::from_millis(250));
110    liveness.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
111
112    loop {
113        tokio::select! {
114            accepted = listener.accept() => {
115                match accepted {
116                    Ok((stream, _addr)) => {
117                        if serve_connection(&pty, stream).await {
118                            // shutdown requested
119                            break;
120                        }
121                    }
122                    Err(_) => continue,
123                }
124            }
125            _ = liveness.tick() => {
126                if !pty.is_child_alive() {
127                    break;
128                }
129            }
130        }
131    }
132
133    // Child exited or shutdown requested. Emit an operator-visible exit event
134    // (so events.jsonl is not silent on steady-state agent death), flip the
135    // registry row to Exited (so `agent.list` does not keep reporting Live
136    // until the next daemon restart/reconcile), mark state.json, drop the
137    // socket, and exit. (silent-failure #1 + #2.)
138    let child_alive_at_exit = pty.is_child_alive();
139    let emitter = EventEmitter::new(home.events_jsonl(), format!("worker:{}", cfg.short_id));
140    let _ = emitter.emit(
141        "agent_exited",
142        &serde_json::json!({
143            "short_id": cfg.short_id,
144            "reason": if child_alive_at_exit { "shutdown" } else { "child_exited" },
145        }),
146    );
147    if let Err(e) = state::update_registry(&home.registry_json(), |r| {
148        if let Some(entry) = r.entries.iter_mut().find(|e| e.short_id == cfg.short_id) {
149            entry.status = AgentStatus::Exited;
150        }
151    }) {
152        eprintln!(
153            "fno-agents-worker: registry exit-update failed for {}: {e}",
154            cfg.short_id
155        );
156    }
157    if let Err(e) = finalize(&pty, &state_path, &cfg.short_id) {
158        eprintln!(
159            "fno-agents-worker: state.json exit-write failed for {}: {e}",
160            cfg.short_id
161        );
162    }
163    let _ = std::fs::remove_file(&sock_path);
164    Ok(())
165}
166
167/// Serve requests on one daemon connection until EOF or `worker.shutdown`.
168/// Returns `true` if shutdown was requested (the run loop then exits).
169async fn serve_connection(pty: &PtySession, mut stream: UnixStream) -> bool {
170    loop {
171        let req = match read_request(&mut stream).await {
172            Ok(r) => r,
173            // Clean hangup or any read fault: end this connection, keep serving.
174            Err(ProtocolError::UnexpectedEof) | Err(_) => return false,
175        };
176        let (resp, shutdown) = handle(pty, &req);
177        if write_response(&mut stream, &resp).await.is_err() {
178            return shutdown;
179        }
180        if shutdown {
181            return true;
182        }
183    }
184}
185
186/// Handle one worker RPC. Returns the response and whether shutdown was asked.
187fn handle(pty: &PtySession, req: &Request) -> (Response, bool) {
188    match req.method.as_str() {
189        "worker.ping" => (Response::ok(req.id, json!({"pong": true})), false),
190        "worker.write" => {
191            // Two input shapes: `data` (a UTF-8 string, the ask path) or
192            // `bytes_b64` (base64 of raw keystroke bytes, the drive path, which
193            // must carry control chars and arbitrary non-UTF-8 bytes a JSON
194            // string cannot). Exactly one is expected; `bytes_b64` wins if both
195            // are present.
196            let bytes: Option<Vec<u8>> = match req.params.get("bytes_b64").and_then(|v| v.as_str())
197            {
198                Some(b64) => match base64::engine::general_purpose::STANDARD.decode(b64) {
199                    Ok(raw) => Some(raw),
200                    Err(e) => {
201                        return (
202                            Response::err(
203                                req.id,
204                                ErrorCode::InvalidParams,
205                                format!("invalid base64 in `bytes_b64`: {e}"),
206                            ),
207                            false,
208                        )
209                    }
210                },
211                None => req
212                    .params
213                    .get("data")
214                    .and_then(|v| v.as_str())
215                    .map(|s| s.as_bytes().to_vec()),
216            };
217            match bytes {
218                Some(raw) => match pty.write_input(&raw) {
219                    Ok(()) => (Response::ok(req.id, json!({"written": raw.len()})), false),
220                    Err(e) => (
221                        Response::err(req.id, ErrorCode::Internal, format!("write failed: {e}")),
222                        false,
223                    ),
224                },
225                None => (
226                    Response::err(
227                        req.id,
228                        ErrorCode::InvalidParams,
229                        "missing `data` (string) or `bytes_b64` (base64)",
230                    ),
231                    false,
232                ),
233            }
234        }
235        "worker.read_since" => {
236            // Incremental PTY-output read for drive streaming. `cursor` is the
237            // absolute byte offset returned by the prior call (0 for a fresh
238            // reader); the response carries the new bytes (base64), the next
239            // cursor, whether a gap (dropped bytes) preceded this read, and
240            // child liveness so the drive pump can detect exit.
241            let cursor = req
242                .params
243                .get("cursor")
244                .and_then(|v| v.as_u64())
245                .unwrap_or(0);
246            let rs = pty.read_since(cursor);
247            let b64 = base64::engine::general_purpose::STANDARD.encode(&rs.bytes);
248            (
249                Response::ok(
250                    req.id,
251                    json!({
252                        "bytes_b64": b64,
253                        "next_offset": rs.next,
254                        "gap": rs.gap,
255                        "child_alive": pty.is_child_alive(),
256                    }),
257                ),
258                false,
259            )
260        }
261        "worker.snapshot" => {
262            let snap = pty.snapshot();
263            let text = String::from_utf8_lossy(&snap).into_owned();
264            (
265                Response::ok(
266                    req.id,
267                    json!({
268                        "text": text,
269                        "dropped_bytes": pty.dropped_bytes(),
270                        "child_alive": pty.is_child_alive(),
271                    }),
272                ),
273                false,
274            )
275        }
276        "worker.status" => (
277            Response::ok(
278                req.id,
279                json!({
280                    "child_pid": pty.child_pid(),
281                    "child_alive": pty.is_child_alive(),
282                    "drain_outcome": format!("{:?}", pty.drain_outcome()),
283                }),
284            ),
285            false,
286        ),
287        "worker.resize" => {
288            let rows = req.params.get("rows").and_then(|v| v.as_u64());
289            let cols = req.params.get("cols").and_then(|v| v.as_u64());
290            match (rows, cols) {
291                (Some(r), Some(c)) => match pty.resize(r as u16, c as u16) {
292                    Ok(()) => (Response::ok(req.id, json!({"resized": true})), false),
293                    Err(e) => (
294                        Response::err(req.id, ErrorCode::Internal, format!("resize: {e}")),
295                        false,
296                    ),
297                },
298                _ => (
299                    Response::err(req.id, ErrorCode::InvalidParams, "need rows and cols"),
300                    false,
301                ),
302            }
303        }
304        "worker.shutdown" => {
305            let _ = pty.kill();
306            (Response::ok(req.id, json!({"shutdown": true})), true)
307        }
308        other => (
309            Response::err(
310                req.id,
311                ErrorCode::UnknownMethod,
312                format!("unknown worker method: {other}"),
313            ),
314            false,
315        ),
316    }
317}
318
319/// Best-effort terminal state write on exit. A failure here is logged via the
320/// return path's caller, never fatal — the worker is exiting regardless.
321/// Terminal state write on exit. Returns the write result so `run()` can log a
322/// failure rather than discarding it (the prior version's doc claimed
323/// caller-side logging that did not exist — now it does).
324fn finalize(
325    pty: &PtySession,
326    state_path: &std::path::Path,
327    short_id: &str,
328) -> Result<(), state::StateError> {
329    let mut st = state::load_state(state_path)
330        .ok()
331        .flatten()
332        .unwrap_or_else(|| AgentState::new_pty(short_id));
333    st.status = AgentStatus::Exited;
334    st.ready = false;
335    if let Some(p) = st.pty.as_mut() {
336        p.active = false;
337    }
338    let _ = pty.kill();
339    state::write_state_atomic(state_path, &st)
340}
341
342/// Build the PTY child command from a [`WorkerConfig`], stamping the
343/// drive-authority identity variables into its environment.
344///
345/// Extracted into a dedicated function so the stamp contract is unit-testable
346/// without spawning a PTY (ab-1e86b88e: locks the drive-authority LD3 identity
347/// stamp read by scripts/lib/drive-authority.sh against silent removal).
348fn build_child_command(cfg: &WorkerConfig) -> CommandBuilder {
349    let mut cmd = CommandBuilder::new(&cfg.argv[0]);
350    for a in &cfg.argv[1..] {
351        cmd.arg(a);
352    }
353    cmd.cwd(&cfg.cwd);
354    // Stamp this agent's identity into the PTY child's environment (cv-140f09c3).
355    // The child (claude/codex) and any Stop / graph-write-protect hook it spawns
356    // inherit FNO_AGENTS_SELF_SHORT_ID, so the drive-authority guard can scope
357    // itself to THIS agent: it fires only when an open operator drive window
358    // targets this short_id, never on a window driving some unrelated agent.
359    cmd.env("FNO_AGENTS_SELF_SHORT_ID", &cfg.short_id);
360    cmd.env("FNO_AGENTS_HOME", cfg.home.as_os_str());
361    cmd
362}
363
364#[cfg(test)]
365mod tests {
366    use super::*;
367    use crate::protocol::{read_response, write_request};
368    use std::ffi::OsStr;
369    use std::time::Instant;
370
371    // --- ab-1e86b88e: drive-authority LD3 identity stamp regression tests ----
372    //
373    // These tests lock the contract that build_child_command stamps BOTH
374    // FNO_AGENTS_SELF_SHORT_ID (the per-agent identity read by
375    // scripts/lib/drive-authority.sh) and FNO_AGENTS_HOME into every PTY child's
376    // environment.  The stamp must never be silently dropped: a child process
377    // (claude/codex) and every Stop / graph-write-protect hook it spawns inherit
378    // the variable so the drive-authority guard can scope itself to THIS agent.
379    // End-to-end propagation (worker -> PTY child -> bash Stop hook) was verified
380    // live on 2026-05-31 against Claude Code 2.1.156; see
381    // tests/hooks/verify-self-short-id-propagation.sh for the manual verifier.
382
383    #[test]
384    fn build_child_command_stamps_self_short_id() {
385        // ab-1e86b88e: locks FNO_AGENTS_SELF_SHORT_ID stamp in build_child_command.
386        // If this test starts failing it means the drive-authority LD3 identity
387        // guard (drive-authority.sh) will no longer be able to scope itself to
388        // the correct agent - do NOT remove this assertion.
389        let cfg = WorkerConfig::new(
390            "wk-1a2b3c",
391            PathBuf::from("/tmp/abi-test-home"),
392            PathBuf::from("/tmp"),
393            vec!["claude".to_string()],
394        );
395        let cmd = build_child_command(&cfg);
396        assert_eq!(
397            cmd.get_env("FNO_AGENTS_SELF_SHORT_ID"),
398            Some(OsStr::new("wk-1a2b3c")),
399            "FNO_AGENTS_SELF_SHORT_ID must be stamped with the worker's short_id \
400             (read by scripts/lib/drive-authority.sh for LD3 scope guard)"
401        );
402    }
403
404    #[test]
405    fn build_child_command_stamps_fno_agents_home() {
406        // ab-1e86b88e: companion to the short_id test - home must also be stamped
407        // so the child and its hooks can locate the shared agents store.
408        let home = PathBuf::from("/tmp/abi-test-home-2");
409        let cfg = WorkerConfig::new(
410            "wk-deadbeef",
411            home.clone(),
412            PathBuf::from("/tmp"),
413            vec!["codex".to_string()],
414        );
415        let cmd = build_child_command(&cfg);
416        assert_eq!(
417            cmd.get_env("FNO_AGENTS_HOME"),
418            Some(home.as_os_str()),
419            "FNO_AGENTS_HOME must be stamped with the worker's home path"
420        );
421    }
422
423    // Unix-socket paths are bounded by `sun_path` (~104 bytes on macOS), so a
424    // socket-binding test must root under a SHORT path. `/tmp/<short>` keeps
425    // `<root>/<short_id>/worker.sock` comfortably under the limit; the verbose
426    // `std::env::temp_dir()` path on macOS (`/var/folders/...`) does not. The
427    // pid + atomic counter make the path collision-proof across parallel tests
428    // (a plain timestamp can collide when tests start in the same millisecond).
429    fn tmp_home(tag: &str) -> PathBuf {
430        use std::sync::atomic::{AtomicU32, Ordering};
431        static COUNTER: AtomicU32 = AtomicU32::new(0);
432        let n = COUNTER.fetch_add(1, Ordering::Relaxed);
433        PathBuf::from(format!("/tmp/abiw{tag}{}_{}", std::process::id(), n))
434    }
435
436    /// Spawn the worker on a background current-thread runtime and return its
437    /// home + short_id once the socket is up. Drives a real `cat` child (stays
438    /// alive on its stdin).
439    async fn start_worker(home: &PathBuf, short_id: &str) {
440        let cfg = WorkerConfig::new(
441            short_id,
442            home.clone(),
443            std::env::temp_dir(),
444            vec!["cat".to_string()],
445        );
446        std::thread::spawn(move || {
447            let rt = tokio::runtime::Builder::new_current_thread()
448                .enable_all()
449                .build()
450                .unwrap();
451            rt.block_on(async {
452                if let Err(e) = run(cfg).await {
453                    eprintln!("WORKER RUN ERROR: {e}");
454                }
455            });
456        });
457        // Wait for the socket to appear.
458        let sock = AgentsHome::at(home).worker_sock(short_id);
459        let start = Instant::now();
460        while !sock.exists() && start.elapsed() < Duration::from_secs(5) {
461            tokio::time::sleep(Duration::from_millis(20)).await;
462        }
463        assert!(sock.exists(), "worker socket never appeared");
464    }
465
466    /// Connect to a just-bound socket, retrying briefly. A freshly-bound
467    /// listener under heavy parallel test load can momentarily refuse before its
468    /// accept loop is scheduled; the production client retries the same way
469    /// (`client::ensure_daemon`).
470    async fn connect_retry(sock: &std::path::Path) -> UnixStream {
471        let start = Instant::now();
472        loop {
473            match UnixStream::connect(sock).await {
474                Ok(c) => return c,
475                Err(_) if start.elapsed() < Duration::from_secs(3) => {
476                    tokio::time::sleep(Duration::from_millis(20)).await;
477                }
478                Err(e) => panic!("connect to {} failed: {e}", sock.display()),
479            }
480        }
481    }
482
483    #[tokio::test(flavor = "current_thread")]
484    async fn worker_serves_ping_write_snapshot() {
485        let home = tmp_home("rpc");
486        start_worker(&home, "wkA").await;
487        let sock = AgentsHome::at(&home).worker_sock("wkA");
488
489        let mut conn = connect_retry(&sock).await;
490        write_request(&mut conn, &Request::new(1, "worker.ping", json!({})))
491            .await
492            .unwrap();
493        let resp = read_response(&mut conn).await.unwrap();
494        assert!(!resp.is_err());
495        assert_eq!(resp.result().unwrap()["pong"], true);
496
497        // Write to cat's stdin; it echoes back onto the PTY.
498        write_request(
499            &mut conn,
500            &Request::new(2, "worker.write", json!({"data": "hello-pty\n"})),
501        )
502        .await
503        .unwrap();
504        let _ = read_response(&mut conn).await.unwrap();
505
506        // Snapshot should eventually contain the echoed text.
507        let mut seen = false;
508        for i in 0..50 {
509            write_request(
510                &mut conn,
511                &Request::new(100 + i, "worker.snapshot", json!({})),
512            )
513            .await
514            .unwrap();
515            let r = read_response(&mut conn).await.unwrap();
516            if r.result().unwrap()["text"]
517                .as_str()
518                .unwrap()
519                .contains("hello-pty")
520            {
521                seen = true;
522                break;
523            }
524            tokio::time::sleep(Duration::from_millis(20)).await;
525        }
526        assert!(seen, "echoed PTY output never appeared in snapshot");
527
528        // Status reports a live child.
529        write_request(&mut conn, &Request::new(3, "worker.status", json!({})))
530            .await
531            .unwrap();
532        let r = read_response(&mut conn).await.unwrap();
533        assert_eq!(r.result().unwrap()["child_alive"], true);
534
535        // Shutdown ends the worker.
536        write_request(&mut conn, &Request::new(4, "worker.shutdown", json!({})))
537            .await
538            .unwrap();
539        let r = read_response(&mut conn).await.unwrap();
540        assert_eq!(r.result().unwrap()["shutdown"], true);
541
542        std::fs::remove_dir_all(&home).ok();
543    }
544
545    #[tokio::test(flavor = "current_thread")]
546    async fn worker_publishes_live_state() {
547        let home = tmp_home("state");
548        start_worker(&home, "wkB").await;
549        let state_path = AgentsHome::at(&home).state_json("wkB");
550        let st = state::load_state(&state_path).unwrap().unwrap();
551        assert_eq!(st.status, AgentStatus::Live);
552        assert!(st.pty.unwrap().active);
553
554        // shut it down
555        let sock = AgentsHome::at(&home).worker_sock("wkB");
556        let mut conn = connect_retry(&sock).await;
557        write_request(&mut conn, &Request::new(1, "worker.shutdown", json!({})))
558            .await
559            .unwrap();
560        let _ = read_response(&mut conn).await;
561        std::fs::remove_dir_all(&home).ok();
562    }
563
564    #[tokio::test(flavor = "current_thread")]
565    async fn worker_read_since_streams_incrementally_and_binary_write_roundtrips() {
566        let home = tmp_home("drive");
567        start_worker(&home, "wkD").await;
568        let sock = AgentsHome::at(&home).worker_sock("wkD");
569        let mut conn = connect_retry(&sock).await;
570
571        // Drive-path input: raw bytes via base64 (here a control byte + text).
572        let raw = b"\x01drive-bytes\n";
573        let b64 = base64::engine::general_purpose::STANDARD.encode(raw);
574        write_request(
575            &mut conn,
576            &Request::new(1, "worker.write", json!({ "bytes_b64": b64 })),
577        )
578        .await
579        .unwrap();
580        let r = read_response(&mut conn).await.unwrap();
581        assert!(!r.is_err());
582        assert_eq!(r.result().unwrap()["written"], raw.len());
583
584        // Incremental read: poll read_since from cursor 0 until cat echoes the
585        // text back, advancing the cursor each poll. The accumulated decoded
586        // output must contain the echoed payload.
587        let mut cursor = 0u64;
588        let mut acc: Vec<u8> = Vec::new();
589        let mut seen = false;
590        for i in 0..50 {
591            write_request(
592                &mut conn,
593                &Request::new(100 + i, "worker.read_since", json!({ "cursor": cursor })),
594            )
595            .await
596            .unwrap();
597            let resp = read_response(&mut conn).await.unwrap();
598            let res = resp.result().unwrap();
599            cursor = res["next_offset"].as_u64().unwrap();
600            let chunk = base64::engine::general_purpose::STANDARD
601                .decode(res["bytes_b64"].as_str().unwrap())
602                .unwrap();
603            acc.extend_from_slice(&chunk);
604            if acc
605                .windows(b"drive-bytes".len())
606                .any(|w| w == b"drive-bytes")
607            {
608                seen = true;
609                break;
610            }
611            tokio::time::sleep(Duration::from_millis(20)).await;
612        }
613        assert!(seen, "echoed drive bytes never streamed via read_since");
614
615        // Drain any trailing echo bytes before asserting an empty tail. The PTY
616        // child (`cat`) can emit trailing bytes (e.g. a `\r\n` line ending) AFTER
617        // the loop above breaks on seeing the payload but BEFORE the tail re-read
618        // below, so a single re-read races the child's output (the intermittent
619        // "left 2 right 0" CI failure). Poll read_since until a read returns zero
620        // new bytes, advancing the cursor past the trailing echo, so the tail
621        // assertion is deterministic. (Fixes flaky cv-ea4e1f0c.)
622        for _ in 0..50 {
623            write_request(
624                &mut conn,
625                &Request::new(150, "worker.read_since", json!({ "cursor": cursor })),
626            )
627            .await
628            .unwrap();
629            let resp = read_response(&mut conn).await.unwrap();
630            let res = resp.result().unwrap();
631            let chunk_len = base64::engine::general_purpose::STANDARD
632                .decode(res["bytes_b64"].as_str().unwrap())
633                .unwrap()
634                .len();
635            cursor = res["next_offset"].as_u64().unwrap();
636            if chunk_len == 0 {
637                break;
638            }
639            tokio::time::sleep(Duration::from_millis(20)).await;
640        }
641
642        // Cursor is now at the tail: a re-read returns no new bytes.
643        write_request(
644            &mut conn,
645            &Request::new(200, "worker.read_since", json!({ "cursor": cursor })),
646        )
647        .await
648        .unwrap();
649        let resp = read_response(&mut conn).await.unwrap();
650        let res = resp.result().unwrap();
651        assert_eq!(
652            base64::engine::general_purpose::STANDARD
653                .decode(res["bytes_b64"].as_str().unwrap())
654                .unwrap()
655                .len(),
656            0,
657            "re-read at the tail must be empty"
658        );
659        assert_eq!(res["child_alive"], true);
660
661        write_request(&mut conn, &Request::new(3, "worker.shutdown", json!({})))
662            .await
663            .unwrap();
664        let _ = read_response(&mut conn).await;
665        std::fs::remove_dir_all(&home).ok();
666    }
667
668    #[tokio::test(flavor = "current_thread")]
669    async fn unknown_method_is_structured_error() {
670        let home = tmp_home("unknown");
671        start_worker(&home, "wkC").await;
672        let sock = AgentsHome::at(&home).worker_sock("wkC");
673        let mut conn = connect_retry(&sock).await;
674        write_request(&mut conn, &Request::new(1, "worker.bogus", json!({})))
675            .await
676            .unwrap();
677        let r = read_response(&mut conn).await.unwrap();
678        assert!(r.is_err());
679        assert_eq!(r.error().unwrap().code, ErrorCode::UnknownMethod);
680
681        write_request(&mut conn, &Request::new(2, "worker.shutdown", json!({})))
682            .await
683            .unwrap();
684        let _ = read_response(&mut conn).await;
685        std::fs::remove_dir_all(&home).ok();
686    }
687}
fno_agents/worker.rs

fno_agents/
worker.rs
