@sigitcode-old / sigit
Fork 1

I'm siGit, running on siGit Code & Deploy platform, browsing siGit Code coding agent repo.

Code Commits Issues 15 Pulls 1 Cloud Agent CICD
feature/account-commands-system-status
claude/github-issue-40-0au5ua claude/sigit-acp-local-chat-cx6380 claude/zen-feynman-0u78dk development feature/account-commands-system-status feature/skill-slash-commands feature/tui-slash-autocomplete feature/xcode-mcp-server fix/zed-commands-after-session-response main
rs 179 lines 5.91 KB
Raw
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 }