alktty 0.5.0

Terminal session protocol: wire format, TtyBackend trait, TtyAdapter, and typed consumer client. Producer/consumer protocol crate on top of alkcall channels.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233

#![cfg(feature = "local")]

//! End-to-end integration tests for pipe mode (`terminal: None`) —
//! `LocalTtyBackend` + `TtyAdapter::drive_session` over a
//! `tokio::io::duplex` transport stand-in, running real commands.
//!
//! Covers scenarios 9–13 of `tasks/tty/integration-test.md`:
//!   9.  Happy path (echo): stdout "hello", stderr empty, exit 0
//!   10. Separate stderr: stdout "out", stderr "err", exit 0
//!   11. Signal (SIGTERM, Unix): exit signal-terminated
//!   12. Cancel cleanup: drop duplex → child killed
//!   13. Resize no-op: control chunk accepted, no error

mod common;

use std::sync::Arc;
use std::time::Duration;

use alktty::local::LocalTtyBackend;
use alktty::wire::STREAM_STDOUT;
use common::{nanos_seed, negotiate_pipe_json, spawn_session};

/// 9. Happy path (echo): negotiate `{backend:"local", tty:null,
/// cmd:["echo","hello"]}`, read stdout chunks, assert "hello", exit 0.
/// Assert stderr is empty.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn pipe_happy_path_echo() {
    let backend = Arc::new(LocalTtyBackend::new());
    let (mut client, server) = spawn_session("local", backend);
    client
        .write_negotiation(negotiate_pipe_json("local", &["echo", "hello"]).as_str())
        .await;

    let (stdout, stderr, code) = client
        .read_until_exit()
        .await
        .expect("expected exit chunk before stream close");
    let out = String::from_utf8_lossy(&stdout);
    assert!(
        out.contains("hello"),
        "stdout should contain 'hello'; got: {out:?}"
    );
    assert!(stderr.is_empty(), "stderr should be empty; got: {stderr:?}");
    assert_eq!(code, 0, "echo should exit 0");
    client.assert_no_more_chunks().await;
    let _ = server.await;
}

/// 10. Separate stderr: negotiate `cmd:["sh","-c","echo out; echo err >&2"]`,
/// assert stdout stream receives "out", stderr stream receives "err"
/// (as stderr chunks, stream_type 2), exit 0.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn pipe_separate_stderr() {
    let backend = Arc::new(LocalTtyBackend::new());
    let (mut client, server) = spawn_session("local", backend);
    client
        .write_negotiation(
            negotiate_pipe_json("local", &["sh", "-c", "echo out; echo err >&2"]).as_str(),
        )
        .await;

    let (stdout, stderr, code) = client
        .read_until_exit()
        .await
        .expect("expected exit chunk before stream close");
    let out = String::from_utf8_lossy(&stdout);
    let err = String::from_utf8_lossy(&stderr);
    assert!(
        out.contains("out"),
        "stdout should contain 'out'; got: {out:?}"
    );
    assert!(
        err.contains("err"),
        "stderr should contain 'err'; got: {err:?}"
    );
    assert_eq!(code, 0, "sh should exit 0");
    let _ = server.await;
}

/// 11. Signal (SIGTERM, Unix): negotiate `sh -c 'echo ready > <marker>;
/// exec sleep 60'`, wait for the marker (the child has exec'd), send
/// `signal:"TERM"`, await exit, assert signal-terminated.
#[cfg(unix)]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn pipe_signal_sigterm_kills_child() {
    let marker = std::env::temp_dir().join(format!(
        "alktty_pipe_sigterm_ready_{}_{}.txt",
        std::process::id(),
        nanos_seed()
    ));
    let cmd = format!("echo ready > '{}'; exec sleep 60", marker.display());
    let backend = Arc::new(LocalTtyBackend::new());
    let (mut client, server) = spawn_session("local", backend);
    client
        .write_negotiation(negotiate_pipe_json("local", &["sh", "-c", cmd.as_str()]).as_str())
        .await;

    // Readiness signal (N4): the child exec'd `sleep` once the marker
    // file exists — no fixed sleep.
    assert!(
        common::wait_for_file(&marker, Duration::from_secs(5)).await,
        "child never became ready"
    );
    let _ = std::fs::remove_file(&marker);

    client
        .write_control(br#"{"type":"signal","name":"TERM"}"#)
        .await;

    let (_out, _err, code) = client
        .read_until_exit_timeout(Duration::from_secs(5))
        .await
        .expect("expected exit chunk after SIGTERM");
    assert_ne!(
        code, 0,
        "child killed by SIGTERM should report non-zero exit; got {code}"
    );
    let _ = server.await;
}

/// 12. Cancel cleanup (ADR-056): negotiate `sleep 60`, drop the duplex
/// mid-session, assert the child is killed (no orphan). The child
/// writes its pid to a temp file so we can probe it after the drop.
#[cfg(unix)]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn pipe_cancel_cleanup_kills_child_no_orphan() {
    let pid_file = std::env::temp_dir().join(format!(
        "alktty_pipe_cancel_pid_{}_{}.txt",
        std::process::id(),
        nanos_seed()
    ));
    let cmd = format!("echo $$ > '{}'; exec sleep 60", pid_file.display());
    let backend = Arc::new(LocalTtyBackend::new());
    let (mut client, server) = spawn_session("local", backend);
    client
        .write_negotiation(negotiate_pipe_json("local", &["sh", "-c", cmd.as_str()]).as_str())
        .await;

    for _ in 0..200 {
        if pid_file.exists() {
            break;
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    let pid_str = std::fs::read_to_string(&pid_file).expect("pid file written");
    let pid: i32 = pid_str.trim().parse().expect("pid parses");
    let _ = std::fs::remove_file(&pid_file);

    drop(client);
    server.abort();
    let _ = server.await;

    // Poll for the child's death (the pid file is written before exec,
    // so the kill lands on the exec'd process; bounded, no fixed sleep).
    let mut alive = true;
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    while alive {
        let r = unsafe { libc::kill(pid, 0) };
        if r != 0 && std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) {
            alive = false;
            break;
        }
        if tokio::time::Instant::now() >= deadline {
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert!(!alive, "child (pid={pid}) should be killed after cancel");
}

/// 13. Resize no-op: negotiate `cmd:["cat"]`, send a `resize` control
/// chunk, assert no error (PipeControl::resize is a no-op). Close the
/// write half to signal end-of-input (the adapter's input pump drops
/// the `ChildStdin` on `ConnectionClosed`, which closes the pipe —
/// tokio's `ChildStdin::poll_shutdown` is a no-op on Unix). Await exit.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn pipe_resize_noop() {
    let backend = Arc::new(LocalTtyBackend::new());
    let (mut client, server) = spawn_session("local", backend);
    client
        .write_negotiation(negotiate_pipe_json("local", &["cat"]).as_str())
        .await;

    // No readiness sleep needed (N4): the adapter's input pump
    // processes chunks in order, and PipeControl::resize is a no-op.
    client
        .write_control(br#"{"type":"resize","cols":120,"rows":40}"#)
        .await;

    client.write_control(br#"{"type":"eof"}"#).await;
    client.close_write_half().await;

    let (_out, _err, code) = client
        .read_until_exit_timeout(Duration::from_secs(5))
        .await
        .expect("expected exit chunk after resize + eof");
    assert_eq!(
        code, 0,
        "cat should exit 0 after no-op resize + write-half close"
    );

    let _ = server.await;
}

/// Sanity: an `echo` in pipe mode should produce at least one
/// stdout chunk with non-empty bytes (the adapter emits a
/// zero-length stdout sentinel after the backend stream ends).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn pipe_echo_emits_stdout_chunk_then_sentinel() {
    let backend = Arc::new(LocalTtyBackend::new());
    let (mut client, server) = spawn_session("local", backend);
    client
        .write_negotiation(negotiate_pipe_json("local", &["echo", "hi"]).as_str())
        .await;

    let mut saw_nonempty_stdout = false;
    while let Some((st, bytes)) = client.read_chunk_timeout(Duration::from_secs(5)).await {
        if st == STREAM_STDOUT && !bytes.is_empty() {
            saw_nonempty_stdout = true;
        }
        if st == alktty::wire::STREAM_CTRL_OUT {
            let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
            if v["type"] == "exit" {
                break;
            }
        }
    }
    assert!(
        saw_nonempty_stdout,
        "expected at least one non-empty stdout chunk"
    );
    let _ = server.await;
}