2001. 2001 .iter()
2002. 2002 .find(|h| h["command"] == "echo off")
2003. 2003 .expect("disabled hook must survive the round trip");
2004. 2004 assert_eq!(off["enabled"], false);
2005. 2005 let on = hooks
2006. 2006 .iter()
2007. 2007 .find(|h| h["command"] == "echo on")
2008. 2008 .expect("enabled hook must still be present");
2009. 2009 assert_eq!(on["enabled"], true);
2010. 2010
2011. 2011 // And it is genuinely on disk, not just echoed back from memory.
2012. 2012 let raw = std::fs::read_to_string(cwd.join(".vak/config.toml")).unwrap();
2013. 2013 assert!(
2014. 2014 raw.contains("echo off"),
2015. 2015 "disabled hook missing from config.toml:\n{raw}"
2016. 2016 );
2017. 2017 assert!(raw.contains("enabled = false"), "config.toml:\n{raw}");
2018. 2018
2019. 2019 // A fresh Core loading that same file must not run the disabled hook.
2020. 2020 vak_config::paths::isolate_home_for_tests();
2021. 2021 let restarted = Core::new_with_trust(cwd, true).unwrap();
2022. 2022 let built = vak_core::build_hooks(restarted.config()).unwrap();
2023. 2023 assert_eq!(
2024. 2024 built.len(),
2025. 2025 1,
2026. 2026 "only the enabled hook should become live: {built:?}"
2027. 2027 );
2028. 2028}
2029. 2029
2030. 2030/// One parsed SSE frame from `/stream`.
2031. 2031#[derive(Debug)]
2032. 2032struct Frame {
2033. 2033 event: String,
2034. 2034 id: Option<String>,
2035. 2035 data: serde_json::Value,
2036. 2036}
2037. 2037
2038. 2038/// Read frames from an SSE response until `done` accepts one, or time out.
2039. 2039async fn read_frames(res: reqwest::Response, mut done: impl FnMut(&Frame) -> bool) -> Vec<Frame> {
2040. 2040 use futures::StreamExt;
2041. 2041 let mut body = res.bytes_stream();
2042. 2042 let mut buffer = String::new();
2043. 2043 let mut frames = Vec::new();
2044. 2044 let deadline = tokio::time::Instant::now() + Duration::from_secs(15);
2045. 2045 loop {
2046. 2046 let chunk = tokio::time::timeout_at(deadline, body.next())
2047. 2047 .await
2048. 2048 .expect("stream frames timed out")
2049. 2049 .expect("stream ended")
2050. 2050 .unwrap();
2051. 2051 buffer.push_str(&String::from_utf8_lossy(&chunk));
2052. 2052 while let Some(end) = buffer.find("\n\n") {
2053. 2053 let block: String = buffer.drain(..end + 2).collect();
2054. 2054 let mut frame = Frame {
2055. 2055 event: "message".into(),
2056. 2056 id: None,
2057. 2057 data: serde_json::Value::Null,
2058. 2058 };
2059. 2059 let mut data = String::new();
2060. 2060 for line in block.lines() {
2061. 2061 if let Some(value) = line.strip_prefix("event:") {
2062. 2062 frame.event = value.trim().into();
2063. 2063 } else if let Some(value) = line.strip_prefix("id:") {
2064. 2064 frame.id = Some(value.trim().into());
2065. 2065 } else if let Some(value) = line.strip_prefix("data:") {
2066. 2066 data.push_str(value.trim());
2067. 2067 }
2068. 2068 }
2069. 2069 if data.is_empty() {
2070. 2070 continue; // keep-alive comment
2071. 2071 }
2072. 2072 frame.data = serde_json::from_str(&data).unwrap_or(serde_json::Value::String(data));
2073. 2073 let finished = done(&frame);
2074. 2074 frames.push(frame);
2075. 2075 if finished {
2076. 2076 return frames;
2077. 2077 }
2078. 2078 }
2079. 2079 }
2080. 2080}
2081. 2081
2082. 2082/// One connection carries every subscription: each session's agent and
2083. 2083/// presentation frames, tagged with the session they belong to, plus host
2084. 2084/// changes; the agent frame's id is a cursor vector that resumes each
2085. 2085/// session from its own sequence (docs/design/48-web-client.md §4.7).
2086. 2086#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
2087. 2087async fn one_stream_multiplexes_sessions_and_resumes_each_from_its_cursor() {
2088. 2088 let (base, token, _cwd, _server) = spawn_secured(Arc::new(Scripted {
2089. 2089 responses: Mutex::new(VecDeque::from(vec![text("streamed answer")])),
2090. 2090 }))
2091. 2091 .await;
2092. 2092 let client = client_with(&token);
2093. 2093 let mut ids = Vec::new();
2094. 2094 for _ in 0..2 {
2095. 2095 let id: String = client
2096. 2096 .post(format!("{base}/sessions"))
2097. 2097 .send()
2098. 2098 .await
2099. 2099 .unwrap()
2100. 2100 .json::<serde_json::Value>()
2101. 2101 .await
2102. 2102 .unwrap()["session_id"]
2103. 2103 .as_str()
2104. 2104 .unwrap()
2105. 2105 .to_string();
2106. 2106 ids.push(id);
2107. 2107 }
2108. 2108 let (a, b) = (ids[0].clone(), ids[1].clone());
2109. 2109
2110. 2110 let empty = client.get(format!("{base}/stream")).send().await.unwrap();
2111. 2111 assert_eq!(
2112. 2112 empty.status(),
2113. 2113 400,
2114. 2114 "a stream with no subscription is refused"
2115. 2115 );
2116. 2116
2117. 2117 let url = format!("{base}/stream?session={a}&session={b}&host=1");
2118. 2118 let res = client.get(&url).send().await.unwrap();
2119. 2119 assert_eq!(res.status(), 200);
2120. 2120 let (opened_tx, opened_rx) = tokio::sync::oneshot::channel::<()>();
2121. 2121 let reader = {
2122. 2122 let a = a.clone();
2123. 2123 tokio::spawn(async move {
2124. 2124 let mut opened_tx = Some(opened_tx);
2125. 2125 read_frames(res, move |frame| {
2126. 2126 if frame.event == "agent"
2127. 2127 && frame.data["session"] == a.as_str()
2128. 2128 && frame.data["event"] == "StreamOpened"
2129. 2129 && let Some(tx) = opened_tx.take()
2130. 2130 {
2131. 2131 let _ = tx.send(());
2132. 2132 }
2133. 2133 frame.event == "agent"
2134. 2134 && frame.data["session"] == a.as_str()
2135. 2135 && frame.data["event"].get("RunFinished").is_some()
2136. 2136 })
2137. 2137 .await
2138. 2138 })
2139. 2139 };
2140. 2140 tokio::time::timeout(Duration::from_secs(10), opened_rx)
2141. 2141 .await
2142. 2142 .expect("stream never opened")
2143. 2143 .unwrap();
2144. 2144 let started = client
2145. 2145 .post(format!("{base}/sessions/{a}/run"))
2146. 2146 .json(&serde_json::json!({"prompt": "hello"}))
2147. 2147 .send()
2148. 2148 .await
2149. 2149 .unwrap();
2150. 2150 assert_eq!(started.status(), 202);
2151. 2151 let frames = reader.await.unwrap();
2152. 2152
2153. 2153 assert!(
2154. 2154 frames
2155. 2155 .iter()
2156. 2156 .any(|f| f.event == "host" && f.data["ready"] == true)
2157. 2157 );
2158. 2158 for id in [&a, &b] {
2159. 2159 assert!(
2160. 2160 frames
2161. 2161 .iter()
2162. 2162 .any(|f| f.event == "presentation" && f.data["session"] == id.as_str()),
2163. 2163 "each followed session gets its presentation snapshot"
2164. 2164 );
2165. 2165 }
2166. 2166 let turn = frames
2167. 2167 .iter()
2168. 2168 .filter(|f| f.event == "agent" && f.data["event"] != "StreamOpened")
2169. 2169 .collect::<Vec<_>>();
2170. 2170 assert!(!turn.is_empty());
2171. 2171 assert!(
2172. 2172 turn.iter().all(|f| f.data["session"] == a.as_str()),
2173. 2173 "a run's events are tagged with its own session only"
2174. 2174 );
2175. 2175 let finished = turn.last().unwrap();
2176. 2176 let cursor = finished.id.clone().expect("agent frames carry the cursor");
2177. 2177 let finished_seq: u64 = cursor
2178. 2178 .split(',')
2179. 2179 .find_map(|pair| pair.strip_prefix(&format!("{a}:")))
2180. 2180 .expect("the cursor names the session that moved")
2181. 2181 .parse()
2182. 2182 .unwrap();
2183. 2183
2184. 2184 // A reconnect one event short of the end is replayed exactly that
2185. 2185 // event for `a`, before anything live.
2186. 2186 let resumed = client
2187. 2187 .get(&url)
2188. 2188 .header("Last-Event-ID", format!("{a}:{}", finished_seq - 1))
2189. 2189 .send()
2190. 2190 .await
2191. 2191 .unwrap();
2192. 2192 let a_for_replay = a.clone();
2193. 2193 let replayed = read_frames(resumed, move |frame| {
2194. 2194 frame.event == "agent" && frame.data["session"] == a_for_replay.as_str()
2195. 2195 })
2196. 2196 .await;
2197. 2197 let first = replayed.last().unwrap();
2198. 2198 assert!(first.data["event"].get("RunFinished").is_some());
2199. 2199 assert!(
2200. 2200 first
2201. 2201 .id
2202. 2202 .as_deref()
2203. 2203 .is_some_and(|id| id.contains(&format!("{a}:{finished_seq}")))
2204. 2204 );
2205. 2205}
2206. 2206