Vakyartha documentation

1//! Discord/Slack bridge end-to-end against faked remote APIs
2//! (docs/design/34-channel-onboarding.md Phase 3), following
3//! `telegram_bridge.rs`: an axum router stands in for the remote surface,
4//! a scripted provider stands in for the model, and the real gateway
5//! router sits in between — so the assertion is on the whole contract,
6//! not on a mocked seam.
7#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
8
9use std::collections::{HashMap, VecDeque};
10use std::sync::{Arc, Mutex};
11
12use tokio_util::sync::CancellationToken;
13
14use vak_core::Core;
15use vak_llm::stream;
16use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, Usage};
17use vak_llm::{EventStream, LlmError, Provider};
18use vak_server::surfaces::discord::DiscordBridge;
19use vak_server::surfaces::slack::SlackBridge;
20
21struct Scripted {
22 responses: Mutex<VecDeque<AssistantMessage>>,
23}
24
25#[async_trait::async_trait]
26impl Provider for Scripted {
27 fn name(&self) -> &str {
28 "scripted"
29 }
30
31 async fn stream(
32 &self,
33 _request: ChatRequest,
34 _cancel: CancellationToken,
35 ) -> Result<EventStream, LlmError> {
36 let next = self.responses.lock().unwrap().pop_front();
37 let (mut sink, rx) = stream::channel(64);
38 match next {
39 Some(m) => {
40 sink.push(stream::StreamEvent::Start { partial: m.clone() });
41 sink.close_message(m).await;
42 }
43 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
44 }
45 Ok(rx)
46 }
47}
48
49fn text(t: &str) -> AssistantMessage {
50 AssistantMessage {
51 content: vec![ContentBlock::text(t)],
52 stop_reason: vak_llm::types::StopReason::EndTurn,
53 usage: Usage {
54 input_tokens: 7,
55 output_tokens: 3,
56 ..Default::default()
57 },
58 model: "test-model".into(),
59 response_id: None,
60 }
61}
62
63/// A gateway with `key` pre-allowlisted and one scripted reply queued.
64async fn spawn_gateway(allow_key: &str, reply: &str) -> String {
65 let dir = tempfile::tempdir().unwrap();
66 let cwd = dir.path().to_path_buf();
67 // Hermetic against the developer's own global config.
68 let _ = std::fs::create_dir_all(cwd.join(".vak"));
69 let _ = std::fs::write(
70 cwd.join(".vak/config.toml"),
71 format!("[memory]\nreflection = false\n\n[gateway]\nchat_allowlist = [\"{allow_key}\"]\n"),
72 );
73 vak_config::paths::isolate_home_for_tests();
74 let core = Core::new_with_trust(cwd, true).unwrap();
75 core.set_sessions_home(dir.path().join("home"));
76 core.set_provider_instance(Arc::new(Scripted {
77 responses: Mutex::new(VecDeque::from(vec![text(reply)])),
78 }));
79 std::mem::forget(dir);
80 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
81 let addr = listener.local_addr().unwrap();
82 tokio::spawn(async move {
83 axum::serve(listener, vak_server::gateway_router(core))
84 .await
85 .unwrap();
86 });
87 format!("http://{addr}")
88}
89
90type Sent = Arc<Mutex<Vec<serde_json::Value>>>;
91
92/// Minimal Discord REST double: `GET /channels/{id}/messages` serves one
93/// scripted message the first time and nothing after, `POST` records the
94/// replies the bridge sends.
95async fn spawn_mock_discord() -> (String, Sent) {
96 let served = Arc::new(std::sync::atomic::AtomicUsize::new(0));
97 let sent: Sent = Arc::new(Mutex::new(Vec::new()));
98 let s = served.clone();
99 let recorder = sent.clone();
100 let app = axum::Router::new().route(
101 "/channels/{id}/messages",
102 axum::routing::get(move || {
103 let s = s.clone();
104 async move {
105 if s.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
106 return axum::Json(serde_json::json!([{
107 "id": "1001",
108 "content": "ping",
109 "author": { "id": "u-7" },
110 }]));
111 }
112 axum::Json(serde_json::json!([]))
113 }
114 })
115 .post(move |axum::Json(body): axum::Json<serde_json::Value>| {
116 let recorder = recorder.clone();
117 async move {
118 recorder.lock().unwrap().push(body);
119 axum::Json(serde_json::json!({ "id": "2001" }))
120 }
121 }),
122 );
123 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
124 let addr = listener.local_addr().unwrap();
125 tokio::spawn(async move {
126 axum::serve(listener, app).await.unwrap();
127 });
128 (format!("http://{addr}"), sent)
129}
130
131/// Minimal Slack API double: `conversations.history` + `chat.postMessage`.
132async fn spawn_mock_slack() -> (String, Sent) {
133 let served = Arc::new(std::sync::atomic::AtomicUsize::new(0));
134 let sent: Sent = Arc::new(Mutex::new(Vec::new()));
135 let s = served.clone();
136 let recorder = sent.clone();
137 let app = axum::Router::new()
138 .route(
139 "/conversations.history",
140 axum::routing::get(move || {
141 let s = s.clone();
142 async move {
143 if s.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
144 return axum::Json(serde_json::json!({
145 "ok": true,
146 "messages": [{ "ts": "1700.0001", "text": "ping", "user": "U7" }],
147 }));
148 }
149 axum::Json(serde_json::json!({ "ok": true, "messages": [] }))
150 }
151 }),
152 )
153 .route(
154 "/chat.postMessage",
155 axum::routing::post(move |axum::Json(body): axum::Json<serde_json::Value>| {
156 let recorder = recorder.clone();
157 async move {
158 recorder.lock().unwrap().push(body);
159 axum::Json(serde_json::json!({ "ok": true }))
160 }
161 }),
162 );
163 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
164 let addr = listener.local_addr().unwrap();
165 tokio::spawn(async move {
166 axum::serve(listener, app).await.unwrap();
167 });
168 (format!("http://{addr}"), sent)
169}
170
171#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
172async fn discord_bridge_routes_message_and_delivers_reply() {
173 let (api_base, sent) = spawn_mock_discord().await;
174 let gateway_url = spawn_gateway("discord:555", "pong from agent").await;
175 let bridge = DiscordBridge {
176 api_base,
177 bot_token: "bottok".into(),
178 channel_ids: vec!["555".into()],
179 gateway_url,
180 gateway_token: "vk_test".into(),
181 poll_secs: 0,
182 bot_id: None,
183 };
184
185 // First pass seeds the cursor from the channel's latest message
186 // without replaying a backlog into the agent.
187 let mut cursors = HashMap::new();
188 bridge.tick(&mut cursors).await.unwrap();
189 assert_eq!(cursors.get("555").map(String::as_str), Some("1001"));
190 assert!(sent.lock().unwrap().is_empty(), "cold start must not reply");
191
192 // The mock now returns nothing new, so a second pass is quiet — the
193 // cursor is what makes that true, not luck.
194 bridge.tick(&mut cursors).await.unwrap();
195 assert!(sent.lock().unwrap().is_empty());
196}
197
198#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
199async fn discord_bridge_replies_to_a_message_after_the_cursor() {
200 let (api_base, sent) = spawn_mock_discord().await;
201 let gateway_url = spawn_gateway("discord:555", "pong from agent").await;
202 let bridge = DiscordBridge {
203 api_base,
204 bot_token: "bottok".into(),
205 channel_ids: vec!["555".into()],
206 gateway_url,
207 gateway_token: "vk_test".into(),
208 poll_secs: 0,
209 bot_id: None,
210 };
211 // Pre-seed the cursor so the scripted message counts as new.
212 let mut cursors = HashMap::from([("555".to_string(), "1".to_string())]);
213 bridge.tick(&mut cursors).await.unwrap();
214
215 let delivered = sent.lock().unwrap();
216 assert_eq!(delivered.len(), 1);
217 assert_eq!(delivered[0]["content"], "pong from agent");
218 assert_eq!(cursors.get("555").map(String::as_str), Some("1001"));
219}
220
221#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
222async fn slack_bridge_replies_to_a_message_after_the_cursor() {
223 let (api_base, sent) = spawn_mock_slack().await;
224 let gateway_url = spawn_gateway("slack:C1", "pong from agent").await;
225 let bridge = SlackBridge {
226 api_base,
227 bot_token: "bottok".into(),
228 channel_ids: vec!["C1".into()],
229 gateway_url,
230 gateway_token: "vk_test".into(),
231 poll_secs: 0,
232 bot_id: None,
233 };
234 let mut cursors = HashMap::from([("C1".to_string(), "1.0".to_string())]);
235 bridge.tick(&mut cursors).await.unwrap();
236
237 let delivered = sent.lock().unwrap();
238 assert_eq!(delivered.len(), 1);
239 assert_eq!(delivered[0]["channel"], "C1");
240 assert_eq!(delivered[0]["text"], "pong from agent");
241 assert_eq!(cursors.get("C1").map(String::as_str), Some("1700.0001"));
242}
243
244#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
245async fn a_chat_not_on_the_allowlist_gets_a_rejection_not_an_agent_turn() {
246 // Phase 1's lifecycle applies uniformly to any surface's key: an
247 // unknown Discord channel becomes pending and is answered with the
248 // gateway's rejection, never with a model reply.
249 let (api_base, sent) = spawn_mock_discord().await;
250 let gateway_url = spawn_gateway("discord:other", "should not be reached").await;
251 let bridge = DiscordBridge {
252 api_base,
253 bot_token: "bottok".into(),
254 channel_ids: vec!["555".into()],
255 gateway_url,
256 gateway_token: "vk_test".into(),
257 poll_secs: 0,
258 bot_id: None,
259 };
260 let mut cursors = HashMap::from([("555".to_string(), "1".to_string())]);
261 bridge.tick(&mut cursors).await.unwrap();
262
263 let delivered = sent.lock().unwrap();
264 assert_eq!(delivered.len(), 1);
265 let content = delivered[0]["content"].as_str().unwrap();
266 assert!(
267 content.contains("gateway error"),
268 "unallowed chat must surface the rejection, got: {content}"
269 );
270}
271

1//! Discord/Slack bridge end-to-end against faked remote APIs
2//! (docs/design/34-channel-onboarding.md Phase 3), following
3//! `telegram_bridge.rs`: an axum router stands in for the remote surface,
4//! a scripted provider stands in for the model, and the real gateway
5//! router sits in between — so the assertion is on the whole contract,
6//! not on a mocked seam.
7#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
8
9use std::collections::{HashMap, VecDeque};
10use std::sync::{Arc, Mutex};
11
12use tokio_util::sync::CancellationToken;
13
14use vak_core::Core;
15use vak_llm::stream;
16use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, Usage};
17use vak_llm::{EventStream, LlmError, Provider};
18use vak_server::surfaces::discord::DiscordBridge;
19use vak_server::surfaces::slack::SlackBridge;
20
21struct Scripted {
22 responses: Mutex<VecDeque<AssistantMessage>>,
23}
24
25#[async_trait::async_trait]
26impl Provider for Scripted {
27 fn name(&self) -> &str {
28 "scripted"
29 }
30
31 async fn stream(
32 &self,
33 _request: ChatRequest,
34 _cancel: CancellationToken,
35 ) -> Result<EventStream, LlmError> {
36 let next = self.responses.lock().unwrap().pop_front();
37 let (mut sink, rx) = stream::channel(64);
38 match next {
39 Some(m) => {
40 sink.push(stream::StreamEvent::Start { partial: m.clone() });
41 sink.close_message(m).await;
42 }
43 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
44 }
45 Ok(rx)
46 }
47}
48
49fn text(t: &str) -> AssistantMessage {
50 AssistantMessage {
51 content: vec![ContentBlock::text(t)],
52 stop_reason: vak_llm::types::StopReason::EndTurn,
53 usage: Usage {
54 input_tokens: 7,
55 output_tokens: 3,
56 ..Default::default()
57 },
58 model: "test-model".into(),
59 response_id: None,
60 }
61}
62
63/// A gateway with `key` pre-allowlisted and one scripted reply queued.
64async fn spawn_gateway(allow_key: &str, reply: &str) -> String {
65 let dir = tempfile::tempdir().unwrap();
66 let cwd = dir.path().to_path_buf();
67 // Hermetic against the developer's own global config.
68 let _ = std::fs::create_dir_all(cwd.join(".vak"));
69 let _ = std::fs::write(
70 cwd.join(".vak/config.toml"),
71 format!("[memory]\nreflection = false\n\n[gateway]\nchat_allowlist = [\"{allow_key}\"]\n"),
72 );
73 vak_config::paths::isolate_home_for_tests();
74 let core = Core::new_with_trust(cwd, true).unwrap();
75 core.set_sessions_home(dir.path().join("home"));
76 core.set_provider_instance(Arc::new(Scripted {
77 responses: Mutex::new(VecDeque::from(vec![text(reply)])),
78 }));
79 std::mem::forget(dir);
80 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
81 let addr = listener.local_addr().unwrap();
82 tokio::spawn(async move {
83 axum::serve(listener, vak_server::gateway_router(core))
84 .await
85 .unwrap();
86 });
87 format!("http://{addr}")
88}
89
90type Sent = Arc<Mutex<Vec<serde_json::Value>>>;
91
92/// Minimal Discord REST double: `GET /channels/{id}/messages` serves one
93/// scripted message the first time and nothing after, `POST` records the
94/// replies the bridge sends.
95async fn spawn_mock_discord() -> (String, Sent) {
96 let served = Arc::new(std::sync::atomic::AtomicUsize::new(0));
97 let sent: Sent = Arc::new(Mutex::new(Vec::new()));
98 let s = served.clone();
99 let recorder = sent.clone();
100 let app = axum::Router::new().route(
101 "/channels/{id}/messages",
102 axum::routing::get(move || {
103 let s = s.clone();
104 async move {
105 if s.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
106 return axum::Json(serde_json::json!([{
107 "id": "1001",
108 "content": "ping",
109 "author": { "id": "u-7" },
110 }]));
111 }
112 axum::Json(serde_json::json!([]))
113 }
114 })
115 .post(move |axum::Json(body): axum::Json<serde_json::Value>| {
116 let recorder = recorder.clone();
117 async move {
118 recorder.lock().unwrap().push(body);
119 axum::Json(serde_json::json!({ "id": "2001" }))
120 }
121 }),
122 );
123 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
124 let addr = listener.local_addr().unwrap();
125 tokio::spawn(async move {
126 axum::serve(listener, app).await.unwrap();
127 });
128 (format!("http://{addr}"), sent)
129}
130
131/// Minimal Slack API double: `conversations.history` + `chat.postMessage`.
132async fn spawn_mock_slack() -> (String, Sent) {
133 let served = Arc::new(std::sync::atomic::AtomicUsize::new(0));
134 let sent: Sent = Arc::new(Mutex::new(Vec::new()));
135 let s = served.clone();
136 let recorder = sent.clone();
137 let app = axum::Router::new()
138 .route(
139 "/conversations.history",
140 axum::routing::get(move || {
141 let s = s.clone();
142 async move {
143 if s.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
144 return axum::Json(serde_json::json!({
145 "ok": true,
146 "messages": [{ "ts": "1700.0001", "text": "ping", "user": "U7" }],
147 }));
148 }
149 axum::Json(serde_json::json!({ "ok": true, "messages": [] }))
150 }
151 }),
152 )
153 .route(
154 "/chat.postMessage",
155 axum::routing::post(move |axum::Json(body): axum::Json<serde_json::Value>| {
156 let recorder = recorder.clone();
157 async move {
158 recorder.lock().unwrap().push(body);
159 axum::Json(serde_json::json!({ "ok": true }))
160 }
161 }),
162 );
163 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
164 let addr = listener.local_addr().unwrap();
165 tokio::spawn(async move {
166 axum::serve(listener, app).await.unwrap();
167 });
168 (format!("http://{addr}"), sent)
169}
170
171#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
172async fn discord_bridge_routes_message_and_delivers_reply() {
173 let (api_base, sent) = spawn_mock_discord().await;
174 let gateway_url = spawn_gateway("discord:555", "pong from agent").await;
175 let bridge = DiscordBridge {
176 api_base,
177 bot_token: "bottok".into(),
178 channel_ids: vec!["555".into()],
179 gateway_url,
180 gateway_token: "vk_test".into(),
181 poll_secs: 0,
182 bot_id: None,
183 };
184
185 // First pass seeds the cursor from the channel's latest message
186 // without replaying a backlog into the agent.
187 let mut cursors = HashMap::new();
188 bridge.tick(&mut cursors).await.unwrap();
189 assert_eq!(cursors.get("555").map(String::as_str), Some("1001"));
190 assert!(sent.lock().unwrap().is_empty(), "cold start must not reply");
191
192 // The mock now returns nothing new, so a second pass is quiet — the
193 // cursor is what makes that true, not luck.
194 bridge.tick(&mut cursors).await.unwrap();
195 assert!(sent.lock().unwrap().is_empty());
196}
197
198#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
199async fn discord_bridge_replies_to_a_message_after_the_cursor() {
200 let (api_base, sent) = spawn_mock_discord().await;
201 let gateway_url = spawn_gateway("discord:555", "pong from agent").await;
202 let bridge = DiscordBridge {
203 api_base,
204 bot_token: "bottok".into(),
205 channel_ids: vec!["555".into()],
206 gateway_url,
207 gateway_token: "vk_test".into(),
208 poll_secs: 0,
209 bot_id: None,
210 };
211 // Pre-seed the cursor so the scripted message counts as new.
212 let mut cursors = HashMap::from([("555".to_string(), "1".to_string())]);
213 bridge.tick(&mut cursors).await.unwrap();
214
215 let delivered = sent.lock().unwrap();
216 assert_eq!(delivered.len(), 1);
217 assert_eq!(delivered[0]["content"], "pong from agent");
218 assert_eq!(cursors.get("555").map(String::as_str), Some("1001"));
219}
220
221#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
222async fn slack_bridge_replies_to_a_message_after_the_cursor() {
223 let (api_base, sent) = spawn_mock_slack().await;
224 let gateway_url = spawn_gateway("slack:C1", "pong from agent").await;
225 let bridge = SlackBridge {
226 api_base,
227 bot_token: "bottok".into(),
228 channel_ids: vec!["C1".into()],
229 gateway_url,
230 gateway_token: "vk_test".into(),
231 poll_secs: 0,
232 bot_id: None,
233 };
234 let mut cursors = HashMap::from([("C1".to_string(), "1.0".to_string())]);
235 bridge.tick(&mut cursors).await.unwrap();
236
237 let delivered = sent.lock().unwrap();
238 assert_eq!(delivered.len(), 1);
239 assert_eq!(delivered[0]["channel"], "C1");
240 assert_eq!(delivered[0]["text"], "pong from agent");
241 assert_eq!(cursors.get("C1").map(String::as_str), Some("1700.0001"));
242}
243
244#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
245async fn a_chat_not_on_the_allowlist_gets_a_rejection_not_an_agent_turn() {
246 // Phase 1's lifecycle applies uniformly to any surface's key: an
247 // unknown Discord channel becomes pending and is answered with the
248 // gateway's rejection, never with a model reply.
249 let (api_base, sent) = spawn_mock_discord().await;
250 let gateway_url = spawn_gateway("discord:other", "should not be reached").await;
251 let bridge = DiscordBridge {
252 api_base,
253 bot_token: "bottok".into(),
254 channel_ids: vec!["555".into()],
255 gateway_url,
256 gateway_token: "vk_test".into(),
257 poll_secs: 0,
258 bot_id: None,
259 };
260 let mut cursors = HashMap::from([("555".to_string(), "1".to_string())]);
261 bridge.tick(&mut cursors).await.unwrap();
262
263 let delivered = sent.lock().unwrap();
264 assert_eq!(delivered.len(), 1);
265 let content = delivered[0]["content"].as_str().unwrap();
266 assert!(
267 content.contains("gateway error"),
268 "unallowed chat must surface the rejection, got: {content}"
269 );
270}
271

Chat Bridges

Indexing the workspace…

Chat Bridges