1
+//! The command list must follow the `session/new` and `session/fork` responses.
2
+//!
3
+//! Zed creates a thread only once it has the response, and drops
4
+//! `available_commands_update` for a session id it does not know yet. Sent
5
+//! before the response, the list never arrives and Zed rejects `/whoami`
6
+//! unless the user types a leading space. This reads the raw wire in order.
7
+
8
+use std::io::{BufRead, BufReader, Write};
9
+use std::process::{Child, ChildStdin, Command, Stdio};
10
+use std::sync::mpsc::{Receiver, channel};
11
+use std::time::{Duration, Instant};
12
+
13
+use serde_json::{Value, json};
14
+
15
+const TIMEOUT: Duration = Duration::from_secs(60);
16
+
17
+struct Wire {
18
+ child: Child,
19
+ stdin: ChildStdin,
20
+ incoming: Receiver<Value>,
21
+ next_id: u64,
22
+}
23
+
24
+fn spawn(config_dir: &std::path::Path) -> Wire {
25
+ let mut child = Command::new(env!("CARGO_BIN_EXE_sigit"))
26
+ .env("OPENAI_BASE_URL", "http://127.0.0.1:9")
27
+ .env("OPENAI_API_KEY", "test-key")
28
+ .env("SIGIT_MODEL", "scripted-model")
29
+ .env("SIGIT_CONFIG_DIR", config_dir)
30
+ .env("SIGIT_MCP", "off")
31
+ .env_remove("SIGIT_LOCAL_INFERENCE")
32
+ .stdin(Stdio::piped())
33
+ .stdout(Stdio::piped())
34
+ .stderr(Stdio::null())
35
+ .spawn()
36
+ .expect("spawn sigit in ACP mode");
37
+ let stdout = child.stdout.take().unwrap();
38
+ let (tx, incoming) = channel();
39
+ std::thread::spawn(move || {
40
+ for line in BufReader::new(stdout).lines() {
41
+ let Ok(line) = line else { break };
42
+ if let Ok(message) = serde_json::from_str::<Value>(&line)
43
+ && tx.send(message).is_err()
44
+ {
45
+ break;
46
+ }
47
+ }
48
+ });
49
+ let stdin = child.stdin.take().unwrap();
50
+ Wire {
51
+ child,
52
+ stdin,
53
+ incoming,
54
+ next_id: 0,
55
+ }
56
+}
57
+
58
+impl Wire {
59
+ fn request(&mut self, method: &str, params: Value) -> u64 {
60
+ self.next_id += 1;
61
+ let id = self.next_id;
62
+ let mut line =
63
+ json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}).to_string();
64
+ line.push('\n');
65
+ self.stdin.write_all(line.as_bytes()).unwrap();
66
+ self.stdin.flush().unwrap();
67
+ id
68
+ }
69
+
70
+ fn next(&mut self, what: &str) -> Value {
71
+ self.incoming
72
+ .recv_timeout(TIMEOUT)
73
+ .unwrap_or_else(|_| panic!("timed out waiting for {what}"))
74
+ }
75
+
76
+ /// Everything on the wire up to and including the response to `id`.
77
+ fn until_response(&mut self, id: u64) -> (Value, Vec<Value>) {
78
+ let mut before = Vec::new();
79
+ let deadline = Instant::now() + TIMEOUT;
80
+ loop {
81
+ assert!(Instant::now() < deadline, "no response to request {id}");
82
+ let message = self.next(&format!("the response to request {id}"));
83
+ if message["id"] == id && message.get("method").is_none() {
84
+ assert!(
85
+ message.get("error").is_none(),
86
+ "request {id} failed: {message}"
87
+ );
88
+ return (message, before);
89
+ }
90
+ before.push(message);
91
+ }
92
+ }
93
+
94
+ /// The next `available_commands_update` for `session_id`.
95
+ fn commands_update_for(&mut self, session_id: &str) -> Value {
96
+ loop {
97
+ let message = self.next("available_commands_update");
98
+ if message["method"] == "session/update"
99
+ && message["params"]["sessionId"] == session_id
100
+ && message["params"]["update"]["sessionUpdate"] == "available_commands_update"
101
+ {
102
+ return message["params"]["update"].clone();
103
+ }
104
+ }
105
+ }
106
+}
107
+
108
+impl Drop for Wire {
109
+ fn drop(&mut self) {
110
+ let _ = self.child.kill();
111
+ let _ = self.child.wait();
112
+ }
113
+}
114
+
115
+fn is_commands_update_for(message: &Value, session_id: &str) -> bool {
116
+ message["method"] == "session/update"
117
+ && message["params"]["sessionId"] == session_id
118
+ && message["params"]["update"]["sessionUpdate"] == "available_commands_update"
119
+}
120
+
121
+fn command_names(update: &Value) -> Vec<&str> {
122
+ update["availableCommands"]
123
+ .as_array()
124
+ .expect("availableCommands")
125
+ .iter()
126
+ .filter_map(|command| command["name"].as_str())
127
+ .collect()
128
+}
129
+
130
+#[test]
131
+fn commands_follow_the_new_and_fork_responses() {
132
+ let scratch = std::env::temp_dir().join(format!("sigit_acp_cmds_{}", std::process::id()));
133
+ let config_dir = scratch.join("config");
134
+ let project = scratch.join("project");
135
+ std::fs::create_dir_all(&config_dir).unwrap();
136
+ std::fs::create_dir_all(&project).unwrap();
137
+
138
+ let mut wire = spawn(&config_dir);
139
+ let id = wire.request(
140
+ "initialize",
141
+ json!({"protocolVersion": 1, "clientCapabilities": {}}),
142
+ );
143
+ wire.until_response(id);
144
+
145
+ let id = wire.request("session/new", json!({"cwd": project, "mcpServers": []}));
146
+ let (response, before) = wire.until_response(id);
147
+ let session_id = response["result"]["sessionId"]
148
+ .as_str()
149
+ .unwrap()
150
+ .to_string();
151
+ assert!(
152
+ !before
153
+ .iter()
154
+ .any(|m| is_commands_update_for(m, &session_id)),
155
+ "the command list must not precede the session/new response: {before:#?}"
156
+ );
157
+ let update = wire.commands_update_for(&session_id);
158
+ assert!(command_names(&update).contains(&"whoami"), "{update}");
159
+
160
+ let id = wire.request(
161
+ "session/fork",
162
+ json!({"sessionId": session_id, "cwd": project, "mcpServers": []}),
163
+ );
164
+ let (response, before) = wire.until_response(id);
165
+ let forked = response["result"]["sessionId"]
166
+ .as_str()
167
+ .unwrap()
168
+ .to_string();
169
+ assert_ne!(forked, session_id);
170
+ assert!(
171
+ !before.iter().any(|m| is_commands_update_for(m, &forked)),
172
+ "the command list must not precede the session/fork response: {before:#?}"
173
+ );
174
+ let update = wire.commands_update_for(&forked);
175
+ assert!(command_names(&update).contains(&"whoami"), "{update}");
176
+
177
+ drop(wire);
178
+ let _ = std::fs::remove_dir_all(&scratch);
179
+}