返回 CodeWhale
bridge.rs
根目录 / crates / tui / src / rlm / bridge.rs
1 //! RPC bridge that services `llm_query` / `rlm_query` calls coming back
2 //! from the long-lived Python REPL during an RLM turn.
3 //!
4 //! This is the spiritual successor to the HTTP sidecar from earlier
5 //! versions — except instead of binding a localhost port and routing
6 //! through `urllib`, requests come in through stdin/stdout and we just
7 //! call the LLM client directly here in Rust.
8 //!
9 //! The bridge tracks cumulative token usage and the recursion budget. For
10 //! `Rlm` / `RlmBatch` requests it recursively calls `run_rlm_turn_inner`
11 //! at depth-1; the future-type cycle (bridge → run_rlm_turn_inner →
12 //! bridge) is broken by `run_rlm_turn_inner` returning a boxed dyn future.
13
14 use std::sync::Arc;
15 use std::time::Duration;
16 use std::{future::Future, pin::Pin};
17
18 use anyhow::Result;
19 use futures_util::future::join_all;
20 use tokio::sync::Mutex;
21
22 use crate::llm_client::LlmClient;
23 use crate::models::{ContentBlock, Message, MessageRequest, MessageResponse, SystemPrompt, Usage};
24 use crate::repl::runtime::{BatchResp, RpcDispatcher, RpcRequest, RpcResponse, SingleResp};
25 use crate::utils::spawn_supervised;
26
27 /// Object-safe runtime-model adapter for a working kernel.
28 ///
29 /// The normal turn loop owns a `SharedModelClient`, while the original RLM
30 /// bridge predates that boundary and accepts the concrete [`LlmClient`] trait.
31 /// Keeping this small adapter here means a persistent kernel follows exactly
32 /// the selected model route (including custom providers) without teaching the
33 /// kernel about provider transports or falling back to a side channel.
34 pub(crate) struct ModelClientRlmAdapter {
35 client: crate::core::model_client::SharedModelClient,
36 }
37
38 impl ModelClientRlmAdapter {
39 pub(crate) fn new(client: crate::core::model_client::SharedModelClient) -> Self {
40 Self { client }
41 }
42 }
43
44 /// Per-child completion timeout — same as the previous sidecar default.
45 const CHILD_TIMEOUT_SECS: u64 = 120;
46 /// Default `max_tokens` for one-shot child completions.
47 const DEFAULT_CHILD_MAX_TOKENS: u32 = 4096;
48 /// Hard cap on prompts per batch RPC.
49 pub const MAX_BATCH: usize = 16;
50
51 /// Object-safe slice of the LLM client interface that the RLM bridge needs.
52 ///
53 /// `LlmClient` itself uses native async trait methods, which are not dyn-safe.
54 /// The bridge only needs non-streaming completions, so this boxed-future shim
55 /// gives tests a clean mock seam without changing the wider provider trait.
56 pub(crate) trait RlmLlmClient: Send + Sync {
57 fn create_message_boxed(
58 &self,
59 request: MessageRequest,
60 ) -> Pin<Box<dyn Future<Output = Result<MessageResponse>> + Send + '_>>;
61 }
62
63 impl RlmLlmClient for ModelClientRlmAdapter {
64 fn create_message_boxed(
65 &self,
66 request: MessageRequest,
67 ) -> Pin<Box<dyn Future<Output = Result<MessageResponse>> + Send + '_>> {
68 let client = Arc::clone(&self.client);
69 Box::pin(async move { client.create_message(request).await })
70 }
71 }
72
73 impl<T> RlmLlmClient for T
74 where
75 T: LlmClient + Send + Sync,
76 {
77 fn create_message_boxed(
78 &self,
79 request: MessageRequest,
80 ) -> Pin<Box<dyn Future<Output = Result<MessageResponse>> + Send + '_>> {
81 Box::pin(self.create_message(request))
82 }
83 }
84
85 /// State shared with the bridge across all RPC calls in one turn.
86 pub struct RlmBridge {
87 client: Arc<dyn RlmLlmClient>,
88 child_model: String,
89 /// Recursion budget remaining for `Rlm` / `RlmBatch` requests. When
90 /// zero, those requests fall back to plain `Llm` completions.
91 depth_remaining: u32,
92 usage: Arc<Mutex<Usage>>,
93 }
94
95 impl RlmBridge {
96 pub(crate) fn new(
97 client: Arc<dyn RlmLlmClient>,
98 child_model: String,
99 depth_remaining: u32,
100 ) -> Self {
101 Self {
102 client,
103 child_model,
104 depth_remaining,
105 usage: Arc::new(Mutex::new(Usage::default())),
106 }
107 }
108
109 pub fn usage_handle(&self) -> Arc<Mutex<Usage>> {
110 Arc::clone(&self.usage)
111 }
112
113 async fn dispatch_llm(
114 &self,
115 prompt: String,
116 _model: Option<String>,
117 max_tokens: Option<u32>,
118 system: Option<String>,
119 ) -> SingleResp {
120 let request = MessageRequest {
121 // The Python helper accepts `model=` for older snippets, but it is
122 // intentionally not authoritative. RLM child calls are pinned to
123 // the tool's configured child model so model-generated Python
124 // cannot silently upgrade cheap fanout work to an expensive model.
125 model: self.child_model.clone(),
126 messages: vec![Message {
127 role: "user".to_string(),
128 content: vec![ContentBlock::Text {
129 text: prompt,
130 cache_control: None,
131 }],
132 }],
133 max_tokens: max_tokens.unwrap_or(DEFAULT_CHILD_MAX_TOKENS),
134 system: system.map(SystemPrompt::Text),
135 tools: None,
136 tool_choice: None,
137 metadata: None,
138 thinking: None,
139 reasoning_effort: None,
140 stream: Some(false),
141 temperature: Some(0.4_f32),
142 top_p: Some(0.9_f32),
143 };
144
145 let fut = self.client.create_message_boxed(request);
146 let response =
147 match tokio::time::timeout(Duration::from_secs(CHILD_TIMEOUT_SECS), fut).await {
148 Ok(Ok(r)) => r,
149 Ok(Err(e)) => {
150 return SingleResp {
151 text: String::new(),
152 error: Some(format!("llm_query failed: {e}")),
153 };
154 }
155 Err(_) => {
156 return SingleResp {
157 text: String::new(),
158 error: Some(format!("llm_query timed out after {CHILD_TIMEOUT_SECS}s")),
159 };
160 }
161 };
162
163 let text = response
164 .content
165 .iter()
166 .filter_map(|b| match b {
167 ContentBlock::Text { text, .. } => Some(text.as_str()),
168 _ => None,
169 })
170 .collect::<Vec<_>>()
171 .join("\n");
172
173 {
174 let mut u = self.usage.lock().await;
175 super::add_usage_with_prompt_cache(&mut u, &response.usage);
176 }
177
178 SingleResp { text, error: None }
179 }
180
181 async fn dispatch_llm_batch(
182 &self,
183 prompts: Vec<String>,
184 _model: Option<String>,
185 dependency_mode: Option<String>,
186 ) -> BatchResp {
187 if let Some(resp) = batch_guard(prompts.len(), dependency_mode.as_deref()) {
188 return resp;
189 }
190
191 let model = Arc::new(self.child_model.clone());
192
193 let futures = prompts.into_iter().map(|prompt| {
194 let model = Arc::clone(&model);
195 async move {
196 self.dispatch_llm((*prompt).to_string(), Some((*model).clone()), None, None)
197 .await
198 }
199 });
200
201 BatchResp {
202 results: join_all(futures).await,
203 }
204 }
205
206 async fn dispatch_rlm(&self, prompt: String, _model: Option<String>) -> SingleResp {
207 if self.depth_remaining == 0 {
208 // Budget exhausted — fall back to a one-shot child completion
209 // rather than returning an error. Matches the paper's behaviour
210 // ("sub_RLM gracefully degrades to llm_query at depth=0").
211 return self.dispatch_llm(prompt, None, None, None).await;
212 }
213
214 // Build a drain channel to absorb status events from the nested
215 // turn (we don't surface them; this dispatch is invisible to the
216 // outer agent stream).
217 let (tx, mut rx) = tokio::sync::mpsc::channel(64);
218 let drain = spawn_supervised(
219 "rlm-bridge-drain",
220 std::panic::Location::caller(),
221 async move { while rx.recv().await.is_some() {} },
222 );
223
224 let child_model = self.child_model.clone();
225
226 // Recursive call. The dyn-erasure on `run_rlm_turn_inner` breaks
227 // the `bridge → turn → bridge` opaque-future cycle.
228 let result = super::turn::run_rlm_turn_inner(
229 Arc::clone(&self.client),
230 child_model.clone(),
231 prompt,
232 None,
233 child_model,
234 tx,
235 self.depth_remaining.saturating_sub(1),
236 )
237 .await;
238
239 drain.abort();
240
241 {
242 let mut u = self.usage.lock().await;
243 super::add_usage_with_prompt_cache(&mut u, &result.usage);
244 }
245
246 SingleResp {
247 text: result.answer,
248 error: result.error,
249 }
250 }
251
252 async fn dispatch_rlm_batch(
253 &self,
254 prompts: Vec<String>,
255 _model: Option<String>,
256 dependency_mode: Option<String>,
257 ) -> BatchResp {
258 if let Some(resp) = batch_guard(prompts.len(), dependency_mode.as_deref()) {
259 return resp;
260 }
261
262 let futures = prompts
263 .into_iter()
264 .map(|p| async move { self.dispatch_rlm(p, None).await });
265 BatchResp {
266 results: join_all(futures).await,
267 }
268 }
269 }
270
271 fn batch_guard(prompt_count: usize, dependency_mode: Option<&str>) -> Option<BatchResp> {
272 if prompt_count == 0 {
273 return Some(BatchResp { results: vec![] });
274 }
275 if prompt_count > MAX_BATCH {
276 return Some(BatchResp {
277 results: (0..prompt_count)
278 .map(|_| SingleResp {
279 text: String::new(),
280 error: Some(format!("batch too large: {prompt_count} > {MAX_BATCH}")),
281 })
282 .collect(),
283 });
284 }
285 let mode = dependency_mode
286 .unwrap_or_default()
287 .trim()
288 .to_ascii_lowercase()
289 .replace(['-', ' '], "_");
290 if !matches!(
291 mode.as_str(),
292 "independent" | "parallel_safe" | "map_reduce"
293 ) {
294 return Some(BatchResp {
295 results: (0..prompt_count)
296 .map(|_| SingleResp {
297 text: String::new(),
298 error: Some(
299 "batch requires dependency_mode='independent'; use sub_query_sequence or sequential sub_query calls for dependent work"
300 .to_string(),
301 ),
302 })
303 .collect(),
304 });
305 }
306 None
307 }
308
309 impl RpcDispatcher for RlmBridge {
310 fn dispatch<'a>(
311 &'a self,
312 req: RpcRequest,
313 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = RpcResponse> + Send + 'a>> {
314 Box::pin(async move {
315 match req {
316 RpcRequest::Llm {
317 prompt,
318 model,
319 max_tokens,
320 system,
321 } => {
322 RpcResponse::Single(self.dispatch_llm(prompt, model, max_tokens, system).await)
323 }
324 RpcRequest::LlmBatch {
325 prompts,
326 model,
327 dependency_mode,
328 safety_note: _,
329 } => RpcResponse::Batch(
330 self.dispatch_llm_batch(prompts, model, dependency_mode)
331 .await,
332 ),
333 RpcRequest::Rlm { prompt, model } => {
334 RpcResponse::Single(self.dispatch_rlm(prompt, model).await)
335 }
336 RpcRequest::RlmBatch {
337 prompts,
338 model,
339 dependency_mode,
340 safety_note: _,
341 } => RpcResponse::Batch(
342 self.dispatch_rlm_batch(prompts, model, dependency_mode)
343 .await,
344 ),
345 }
346 })
347 }
348 }
349
350 #[cfg(test)]
351 mod tests {
352 use super::*;
353 use crate::llm_client::mock::MockLlmClient;
354
355 fn mock_response_with_usage(text: &str, usage: Usage) -> MessageResponse {
356 MessageResponse {
357 id: "mock_msg".to_string(),
358 r#type: "message".to_string(),
359 role: "assistant".to_string(),
360 content: vec![ContentBlock::Text {
361 text: text.to_string(),
362 cache_control: None,
363 }],
364 model: "mock-model".to_string(),
365 stop_reason: Some("end_turn".to_string()),
366 stop_sequence: None,
367 container: None,
368 usage,
369 }
370 }
371
372 fn mock_response(text: &str, input_tokens: u32, output_tokens: u32) -> MessageResponse {
373 mock_response_with_usage(
374 text,
375 Usage {
376 input_tokens,
377 output_tokens,
378 ..Usage::default()
379 },
380 )
381 }
382
383 fn bridge_for(mock: Arc<MockLlmClient>, depth_remaining: u32) -> RlmBridge {
384 let client: Arc<dyn RlmLlmClient> = mock;
385 RlmBridge::new(client, "child-model".to_string(), depth_remaining)
386 }
387
388 #[test]
389 fn batch_guard_allows_non_empty_batches_at_the_cap() {
390 assert!(batch_guard(MAX_BATCH, Some("independent")).is_none());
391 }
392
393 #[test]
394 fn batch_guard_returns_empty_response_for_empty_batches() {
395 let response = batch_guard(0, None).expect("empty batch should be handled");
396 assert!(response.results.is_empty());
397 }
398
399 #[test]
400 fn batch_guard_returns_one_error_per_oversized_prompt() {
401 let response = batch_guard(MAX_BATCH + 2, Some("independent"))
402 .expect("oversized batch should be handled");
403 assert_eq!(response.results.len(), MAX_BATCH + 2);
404 assert!(response.results.iter().all(|result| {
405 result.text.is_empty()
406 && result
407 .error
408 .as_deref()
409 .is_some_and(|err| err.contains("batch too large"))
410 }));
411 }
412
413 #[test]
414 fn batch_guard_requires_explicit_independence_for_parallel_work() {
415 let response = batch_guard(2, None).expect("missing dependency mode should be handled");
416 assert_eq!(response.results.len(), 2);
417 assert!(response.results.iter().all(|result| {
418 result.text.is_empty()
419 && result
420 .error
421 .as_deref()
422 .is_some_and(|err| err.contains("dependency_mode='independent'"))
423 }));
424
425 let response = batch_guard(2, Some("sequential"))
426 .expect("dependent dependency mode should be handled");
427 assert!(response.results.iter().all(|result| {
428 result
429 .error
430 .as_deref()
431 .is_some_and(|err| err.contains("sub_query_sequence"))
432 }));
433 }
434
435 #[tokio::test]
436 async fn llm_dispatch_pins_configured_child_model() {
437 let mock = Arc::new(MockLlmClient::new(Vec::new()));
438 mock.push_message_response(mock_response("child answer", 7, 11));
439 let bridge = bridge_for(Arc::clone(&mock), 1);
440
441 let response = bridge
442 .dispatch(RpcRequest::Llm {
443 prompt: "child prompt".to_string(),
444 model: Some("override-model".to_string()),
445 max_tokens: Some(123),
446 system: Some("child system".to_string()),
447 })
448 .await;
449
450 match response {
451 RpcResponse::Single(single) => {
452 assert_eq!(single.text, "child answer");
453 assert!(single.error.is_none());
454 }
455 other => panic!("expected single response, got {other:?}"),
456 }
457
458 let captured = mock.captured_requests();
459 assert_eq!(captured.len(), 1);
460 assert_eq!(captured[0].model, "child-model");
461 assert_eq!(captured[0].max_tokens, 123);
462 assert_eq!(
463 captured[0].system,
464 Some(SystemPrompt::Text("child system".to_string()))
465 );
466
467 let usage = bridge.usage.lock().await;
468 assert_eq!(usage.input_tokens, 7);
469 assert_eq!(usage.output_tokens, 11);
470 }
471
472 #[tokio::test]
473 async fn llm_dispatch_preserves_prompt_cache_usage() {
474 let mock = Arc::new(MockLlmClient::new(Vec::new()));
475 mock.push_message_response(mock_response_with_usage(
476 "cached child answer",
477 Usage {
478 input_tokens: 1000,
479 output_tokens: 100,
480 prompt_cache_hit_tokens: Some(800),
481 prompt_cache_miss_tokens: Some(200),
482 ..Usage::default()
483 },
484 ));
485 let bridge = bridge_for(Arc::clone(&mock), 1);
486
487 let response = bridge
488 .dispatch(RpcRequest::Llm {
489 prompt: "child prompt".to_string(),
490 model: None,
491 max_tokens: None,
492 system: None,
493 })
494 .await;
495
496 match response {
497 RpcResponse::Single(single) => {
498 assert_eq!(single.text, "cached child answer");
499 assert!(single.error.is_none());
500 }
501 other => panic!("expected single response, got {other:?}"),
502 }
503
504 let usage = bridge.usage.lock().await;
505 assert_eq!(usage.input_tokens, 1000);
506 assert_eq!(usage.output_tokens, 100);
507 assert_eq!(usage.prompt_cache_hit_tokens, Some(800));
508 assert_eq!(usage.prompt_cache_miss_tokens, Some(200));
509 }
510
511 #[tokio::test]
512 async fn llm_batch_dispatch_pins_configured_child_model() {
513 let mock = Arc::new(MockLlmClient::new(Vec::new()));
514 mock.push_message_response(mock_response("one", 1, 2));
515 mock.push_message_response(mock_response("two", 3, 4));
516 mock.push_message_response(mock_response("three", 5, 6));
517 let bridge = bridge_for(Arc::clone(&mock), 1);
518
519 let response = bridge
520 .dispatch(RpcRequest::LlmBatch {
521 prompts: vec!["a".to_string(), "b".to_string(), "c".to_string()],
522 model: Some("batch-model".to_string()),
523 dependency_mode: Some("independent".to_string()),
524 safety_note: Some("test prompts are independent".to_string()),
525 })
526 .await;
527
528 match response {
529 RpcResponse::Batch(batch) => {
530 let texts: Vec<_> = batch
531 .results
532 .iter()
533 .map(|result| result.text.as_str())
534 .collect();
535 assert_eq!(texts, ["one", "two", "three"]);
536 assert!(batch.results.iter().all(|result| result.error.is_none()));
537 }
538 other => panic!("expected batch response, got {other:?}"),
539 }
540
541 let captured = mock.captured_requests();
542 assert_eq!(captured.len(), 3);
543 assert!(
544 captured
545 .iter()
546 .all(|request| request.model == "child-model")
547 );
548
549 let usage = bridge.usage.lock().await;
550 assert_eq!(usage.input_tokens, 9);
551 assert_eq!(usage.output_tokens, 12);
552 }
553
554 #[tokio::test]
555 async fn rlm_dispatch_at_depth_zero_pins_configured_child_model() {
556 let mock = Arc::new(MockLlmClient::new(Vec::new()));
557 mock.push_message_response(mock_response("fallback answer", 3, 5));
558 let bridge = bridge_for(Arc::clone(&mock), 0);
559
560 let response = bridge
561 .dispatch(RpcRequest::Rlm {
562 prompt: "nested prompt".to_string(),
563 model: Some("override-model".to_string()),
564 })
565 .await;
566
567 match response {
568 RpcResponse::Single(single) => {
569 assert_eq!(single.text, "fallback answer");
570 assert!(single.error.is_none());
571 }
572 other => panic!("expected single response, got {other:?}"),
573 }
574
575 let usage = bridge.usage.lock().await;
576 assert_eq!(usage.input_tokens, 3);
577 assert_eq!(usage.output_tokens, 5);
578
579 let captured = mock.captured_requests();
580 assert_eq!(captured.len(), 1);
581 assert_eq!(captured[0].model, "child-model");
582 }
583 }
584
584 lines RUST