netget/client/http_proxy/mod.rs
1//! HTTP proxy client implementation
2pub mod actions;
3
4pub use actions::HttpProxyClientProtocol;
5
6use anyhow::{Context, Result};
7use std::net::SocketAddr;
8use std::sync::Arc;
9use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
10use tokio::net::TcpStream;
11use tokio::sync::{mpsc, Mutex};
12use tracing::{debug, error, info, trace};
13
14use crate::client::http_proxy::actions::{
15 HTTP_PROXY_CLIENT_CONNECTED_EVENT, HTTP_PROXY_RESPONSE_RECEIVED_EVENT,
16 HTTP_PROXY_TUNNEL_ESTABLISHED_EVENT,
17};
18use crate::client::llm_budget::call_llm_for_client;
19use crate::llm::actions::client_trait::{Client, ClientActionResult};
20use crate::llm::ollama_client::OllamaClient;
21use crate::llm::ClientLlmResult;
22use crate::protocol::Event;
23use crate::state::app_state::AppState;
24use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
25use crate::state::{AccessLogOwner, ClientId, ClientStatus};
26
27/// Connection state for LLM processing
28#[derive(Debug, Clone, PartialEq)]
29enum ConnectionState {
30 Idle,
31 Processing,
32 Accumulating,
33}
34
35/// Per-client data for LLM handling
36struct ClientData {
37 state: ConnectionState,
38 queued_data: Vec<u8>,
39 memory: String,
40 tunnel_established: bool,
41}
42
43/// HTTP proxy client that connects to a proxy server
44pub struct HttpProxyClient;
45
46impl HttpProxyClient {
47 /// Connect to an HTTP proxy server with integrated LLM actions
48 pub async fn connect_with_llm_actions(
49 remote_addr: String,
50 llm_client: OllamaClient,
51 app_state: Arc<AppState>,
52 status_tx: mpsc::UnboundedSender<String>,
53 client_id: ClientId,
54 startup_params: Option<crate::protocol::StartupParams>,
55 ) -> Result<SocketAddr> {
56 // `proxy_auth` and `default_target` were both declared and neither was read: a
57 // proxy that requires authentication answered 407 to every CONNECT with nothing in
58 // the parameter list to explain it, and `default_target` named a tunnel nothing
59 // ever opened.
60 //
61 // `proxy_auth` is `username:password`, so it is resolved once here into the
62 // `Proxy-Authorization: Basic ...` value every CONNECT now carries.
63 let (proxy_auth, default_target) = match &startup_params {
64 Some(params) => (
65 params.get_optional_string("proxy_auth")?,
66 params.get_optional_string("default_target")?,
67 ),
68 None => (None, None),
69 };
70 let proxy_authorization = proxy_auth.map(|credentials| {
71 use base64::Engine;
72 let encoded = base64::engine::general_purpose::STANDARD.encode(credentials.as_bytes());
73 format!("Basic {encoded}")
74 });
75
76 // Connect to the proxy server
77 let stream = TcpStream::connect(&remote_addr).await.context(format!(
78 "Failed to connect to HTTP proxy at {}",
79 remote_addr
80 ))?;
81
82 let local_addr = stream.local_addr()?;
83 let remote_sock_addr = stream.peer_addr()?;
84
85 info!(
86 "HTTP proxy client {} connected to proxy {} (local: {})",
87 client_id, remote_sock_addr, local_addr
88 );
89
90 // Stored so both the tunnel task and an injected `establish_tunnel` read the same
91 // configuration rather than each carrying its own copy.
92 app_state
93 .with_client_mut(client_id, |client| {
94 if let Some(value) = &proxy_authorization {
95 client.set_protocol_field(
96 "proxy_authorization".to_string(),
97 serde_json::json!(value),
98 );
99 }
100 if let Some(target) = &default_target {
101 client.set_protocol_field(
102 "default_target".to_string(),
103 serde_json::json!(target),
104 );
105 }
106 })
107 .await;
108
109 // Update client state
110 app_state
111 .update_client_status(client_id, ClientStatus::Connected)
112 .await;
113 let _ = status_tx.send(format!(
114 "[CLIENT] HTTP proxy client {} connected to {}",
115 client_id, remote_sock_addr
116 ));
117 let _ = status_tx.send("__UPDATE_UI__".to_string());
118
119 // Split stream. Done before the connected-event LLM call so the write half can be
120 // shared with the command task below while that call is (possibly) parked.
121 let (read_half, write_half) = tokio::io::split(stream);
122 let write_half_arc = Arc::new(Mutex::new(write_half));
123
124 // Command channel for the dashboard's [ send_http_request ] / [ send_data ] /
125 // [ establish_tunnel ] / [ disconnect ] rows. Registered BEFORE the connected call: a
126 // manual handler can park that call for minutes, and an injected send must work for
127 // the whole park. The read loop mixes `read_line` (CONNECT headers, not
128 // cancellation-safe) and `read` (tunnel bytes), so the channel is drained by its own
129 // task sharing the write half rather than by a `select!` arm.
130 let mut command_rx =
131 crate::client::command_support::register_command_channel(&app_state, client_id).await;
132 let cmd_state = app_state.clone();
133 let cmd_tx = status_tx.clone();
134 let cmd_write = write_half_arc.clone();
135 let cmd_task = tokio::spawn(async move {
136 while let Some(cmd) = command_rx.recv().await {
137 let disconnect =
138 handle_injected_command(&cmd_write, cmd, client_id, &cmd_state, &cmd_tx).await;
139 if disconnect {
140 // Half-close: the proxy reads EOF and closes; the read loop then sees 0
141 // and runs its normal disconnect path (which drops the handle).
142 let _ = cmd_write.lock().await.shutdown().await;
143 break;
144 }
145 }
146 });
147 app_state.register_client_task(client_id, cmd_task).await;
148
149 // Call LLM with connected event
150 if let Some(instruction) = app_state.get_instruction_for_client(client_id).await {
151 let protocol =
152 Arc::new(crate::client::http_proxy::actions::HttpProxyClientProtocol::new());
153 let event = Event::new(
154 &HTTP_PROXY_CLIENT_CONNECTED_EVENT,
155 serde_json::json!({
156 "proxy_addr": remote_sock_addr.to_string(),
157 }),
158 );
159
160 match call_llm_for_client(
161 &llm_client,
162 &app_state,
163 client_id.to_string(),
164 &instruction,
165 "",
166 Some(&event),
167 protocol.as_ref(),
168 &status_tx,
169 )
170 .await
171 {
172 Ok(ClientLlmResult {
173 actions,
174 memory_updates,
175 }) => {
176 // Store memory
177 if let Some(mem) = memory_updates {
178 app_state.set_memory_for_client(client_id, mem).await;
179 }
180
181 // Execute initial actions
182 for action in actions {
183 if let Ok(result) = protocol.as_ref().execute_action(action) {
184 match result {
185 ClientActionResult::Custom { name, data }
186 if name == "establish_tunnel" =>
187 {
188 // Handle tunnel establishment. A target the model did
189 // not fully specify falls back to `default_target`,
190 // which is what that parameter is for.
191 let target = match (
192 data.get("target_host").and_then(|v| v.as_str()),
193 data.get("target_port").and_then(|v| v.as_u64()),
194 ) {
195 (Some(host), Some(port)) => Some(format!("{host}:{port}")),
196 _ => default_target.clone(),
197 };
198 if let Some(target) = target {
199 info!(
200 "HTTP proxy client {} establishing tunnel to {}",
201 client_id, target
202 );
203 // We'll establish the tunnel in the spawn task below
204 app_state
205 .with_client_mut(client_id, |client| {
206 client.set_protocol_field(
207 "tunnel_target".to_string(),
208 serde_json::json!(target),
209 );
210 })
211 .await;
212 }
213 }
214 _ => {}
215 }
216 }
217 }
218 }
219 Err(e) => {
220 error!("LLM error for HTTP proxy client {}: {}", client_id, e);
221 }
222 }
223 }
224
225 // `default_target` means "open this tunnel", not "open it only if the model also
226 // asks". If nothing selected a target above, it is the target.
227 if let Some(target) = &default_target {
228 app_state
229 .with_client_mut(client_id, |client| {
230 if client.get_protocol_field("tunnel_target").is_none() {
231 client.set_protocol_field(
232 "tunnel_target".to_string(),
233 serde_json::json!(target),
234 );
235 }
236 })
237 .await;
238 }
239
240 // Initialize client data
241 let client_data = Arc::new(Mutex::new(ClientData {
242 state: ConnectionState::Idle,
243 queued_data: Vec::new(),
244 memory: String::new(),
245 tunnel_established: false,
246 }));
247
248 // Clone for spawn
249 let app_state_clone = app_state.clone();
250 let write_half_clone = write_half_arc.clone();
251 let client_data_clone = client_data.clone();
252
253 // Spawn task to handle tunnel establishment if needed
254 // Registered with AppState so stop_client can abort this task —
255 // dropping a JoinHandle only detaches it in Tokio.
256 let task_registrar = app_state.clone();
257 let task_handle = tokio::spawn(async move {
258 // Check if we have a tunnel target to establish
259 if let Some(tunnel_target) = app_state_clone
260 .with_client_mut(client_id, |client| {
261 client
262 .get_protocol_field("tunnel_target")
263 .and_then(|v| v.as_str())
264 .map(|s| s.to_string())
265 })
266 .await
267 .flatten()
268 {
269 let parts: Vec<&str> = tunnel_target.split(':').collect();
270 if parts.len() == 2 {
271 let target_host = parts[0];
272 let target_port = parts[1];
273
274 // Send CONNECT request
275 let connect_request =
276 connect_request(target_host, target_port, proxy_authorization.as_deref());
277
278 debug!(
279 "HTTP proxy client {} sending CONNECT request: {}",
280 client_id,
281 connect_request.trim()
282 );
283
284 if let Err(e) = write_half_clone
285 .lock()
286 .await
287 .write_all(connect_request.as_bytes())
288 .await
289 {
290 error!(
291 "HTTP proxy client {} failed to send CONNECT: {}",
292 client_id, e
293 );
294 }
295 }
296 }
297 });
298 task_registrar
299 .register_client_task(client_id, task_handle)
300 .await;
301
302 // Spawn read loop
303 let app_state_clone = app_state.clone();
304 let status_tx_clone = status_tx.clone();
305 let write_half_clone = write_half_arc.clone();
306
307 // Registered with AppState so stop_client can abort this task —
308 // dropping a JoinHandle only detaches it in Tokio.
309 let task_registrar = app_state.clone();
310 let task_handle = tokio::spawn(async move {
311 let handle_state = app_state_clone.clone();
312 // The loop body is an inner block so its `return`s all land on the handle removal
313 // below: once the socket is dead the dashboard must stop offering [ send ].
314 async move {
315 let mut reader = BufReader::new(crate::client::response_reader::FrameReader::new(read_half));
316
317 // First, check if we need to read CONNECT response
318 if let Some(target) = app_state_clone
319 .with_client_mut(client_id, |client| {
320 client
321 .get_protocol_field("tunnel_target")
322 .and_then(|v| v.as_str())
323 .map(|s| s.to_string())
324 })
325 .await
326 .flatten()
327 {
328 reader.get_mut().start_frame(crate::client::response_reader::RESPONSE_DEADLINE);
329 // Read CONNECT response
330 let mut status_line = String::new();
331 match crate::client::response_reader::read_response_line(&mut reader, &mut status_line).await {
332 Ok(0) => {
333 error!(
334 "HTTP proxy client {} disconnected during CONNECT",
335 client_id
336 );
337 app_state_clone
338 .update_client_status(
339 client_id,
340 ClientStatus::Error("Proxy disconnected".to_string()),
341 )
342 .await;
343 return;
344 }
345 Ok(_) => {
346 debug!(
347 "HTTP proxy client {} received status: {}",
348 client_id,
349 status_line.trim()
350 );
351
352 // Parse status code
353 let parts: Vec<&str> = status_line.split_whitespace().collect();
354 let status_code = if parts.len() >= 2 {
355 parts[1].parse::<u16>().unwrap_or(0)
356 } else {
357 0
358 };
359
360 // Bound the aggregate head too: many short headers are otherwise
361 // just as unbounded as one line with no terminator.
362 let mut header_bytes = status_line.len();
363 loop {
364 let mut header_line = String::new();
365 match crate::client::response_reader::read_response_line(&mut reader, &mut header_line).await {
366 Ok(0) => {
367 app_state_clone.update_client_status(client_id, ClientStatus::Error("Proxy closed before CONNECT headers completed".to_string())).await;
368 return;
369 },
370 Ok(_) => {
371 header_bytes += header_line.len();
372 if header_bytes > crate::client::response_reader::MAX_RESPONSE_LINE_BYTES {
373 app_state_clone.update_client_status(client_id, ClientStatus::Error("Proxy CONNECT response headers exceed the byte cap".to_string())).await;
374 return;
375 }
376 if header_line.trim().is_empty() {
377 break;
378 }
379 }
380 Err(e) => {
381 error!(
382 "HTTP proxy client {} error reading headers: {}",
383 client_id, e
384 );
385 app_state_clone.update_client_status(client_id, ClientStatus::Error(e.to_string())).await;
386 return;
387 }
388 }
389 }
390
391 if status_code == 200 {
392 info!(
393 "HTTP proxy client {} tunnel established successfully",
394 client_id
395 );
396 client_data_clone.lock().await.tunnel_established = true;
397
398 // Call LLM with tunnel established event
399 if let Some(instruction) =
400 app_state_clone.get_instruction_for_client(client_id).await
401 {
402 let protocol = Arc::new(crate::client::http_proxy::actions::HttpProxyClientProtocol::new());
403
404 let parts: Vec<&str> = target.split(':').collect();
405 let (target_host, target_port) = if parts.len() == 2 {
406 (parts[0].to_string(), parts[1].parse::<u16>().unwrap_or(0))
407 } else {
408 (target.clone(), 0)
409 };
410
411 let event = Event::new(
412 &HTTP_PROXY_TUNNEL_ESTABLISHED_EVENT,
413 serde_json::json!({
414 "target_host": target_host,
415 "target_port": target_port,
416 "status_code": status_code,
417 }),
418 );
419
420 // Copy the memory out under its own guard first. Passing
421 // `&client_data_clone.lock().await.memory` into the call keeps
422 // that guard alive for the whole `match` -- a temporary in a
423 // match scrutinee lives until the match ends -- and the `Ok`
424 // arm locks the same mutex again to store the model's memory
425 // update. `tokio::sync::Mutex` is not reentrant, so the tunnel
426 // task deadlocks the moment the model returns memory.
427 let memory =
428 { client_data_clone.lock().await.memory.clone() };
429 match call_llm_for_client(
430 &llm_client,
431 &app_state_clone,
432 client_id.to_string(),
433 &instruction,
434 &memory,
435 Some(&event),
436 protocol.as_ref(),
437 &status_tx_clone,
438 )
439 .await
440 {
441 Ok(ClientLlmResult {
442 actions,
443 memory_updates,
444 }) => {
445 // Update memory
446 if let Some(mem) = memory_updates {
447 client_data_clone.lock().await.memory = mem;
448 }
449
450 // Execute actions
451 for action in actions {
452 if let Ok(result) =
453 protocol.as_ref().execute_action(action)
454 {
455 match result {
456 ClientActionResult::SendData(bytes) => {
457 if write_half_clone
458 .lock()
459 .await
460 .write_all(&bytes)
461 .await
462 .is_ok()
463 {
464 trace!("HTTP proxy client {} sent {} bytes via tunnel", client_id, bytes.len());
465 }
466 }
467 ClientActionResult::Disconnect => {
468 info!(
469 "HTTP proxy client {} disconnecting",
470 client_id
471 );
472 return;
473 }
474 _ => {}
475 }
476 }
477 }
478 }
479 Err(e) => {
480 error!(
481 "LLM error for HTTP proxy client {}: {}",
482 client_id, e
483 );
484 }
485 }
486 }
487 } else {
488 error!(
489 "HTTP proxy client {} tunnel failed with status {}",
490 client_id, status_code
491 );
492 app_state_clone
493 .update_client_status(
494 client_id,
495 ClientStatus::Error(format!("Tunnel failed: {}", status_code)),
496 )
497 .await;
498 return;
499 }
500 }
501 Err(e) => {
502 error!(
503 "HTTP proxy client {} error reading CONNECT response: {}",
504 client_id, e
505 );
506 app_state_clone
507 .update_client_status(client_id, ClientStatus::Error(e.to_string()))
508 .await;
509 return;
510 }
511 }
512 }
513
514 // Now read data through the tunnel
515 let mut buffer = vec![0u8; 8192];
516
517 reader.get_mut().end_frame();
518 'tunnel: loop {
519 match reader.read(&mut buffer).await {
520 Ok(0) => {
521 info!("HTTP proxy client {} disconnected", client_id);
522 app_state_clone
523 .update_client_status(client_id, ClientStatus::Disconnected)
524 .await;
525 let _ = status_tx_clone.send(format!(
526 "[CLIENT] HTTP proxy client {} disconnected",
527 client_id
528 ));
529 let _ = status_tx_clone.send("__UPDATE_UI__".to_string());
530 break;
531 }
532 Ok(n) => {
533 let data = buffer[..n].to_vec();
534 trace!(
535 "HTTP proxy client {} received {} bytes via tunnel",
536 client_id,
537 n
538 );
539
540 // Handle data with LLM
541 let mut client_data_lock = client_data_clone.lock().await;
542
543 match client_data_lock.state {
544 ConnectionState::Idle => {
545 // Process immediately
546 client_data_lock.state = ConnectionState::Processing;
547 drop(client_data_lock);
548
549 // Call LLM
550 if let Some(instruction) =
551 app_state_clone.get_instruction_for_client(client_id).await
552 {
553 let protocol = Arc::new(crate::client::http_proxy::actions::HttpProxyClientProtocol::new());
554 let event = Event::new(
555 &HTTP_PROXY_RESPONSE_RECEIVED_EVENT,
556 serde_json::json!({
557 "data_hex": hex::encode(&data),
558 "data_length": data.len(),
559 }),
560 );
561
562 // Same reasoning as the tunnel-established call above:
563 // the guard must not survive into the match arms.
564 let memory =
565 { client_data_clone.lock().await.memory.clone() };
566 match call_llm_for_client(
567 &llm_client,
568 &app_state_clone,
569 client_id.to_string(),
570 &instruction,
571 &memory,
572 Some(&event),
573 protocol.as_ref(),
574 &status_tx_clone,
575 )
576 .await
577 {
578 Ok(ClientLlmResult {
579 actions,
580 memory_updates,
581 }) => {
582 // Update memory
583 if let Some(mem) = memory_updates {
584 client_data_clone.lock().await.memory = mem;
585 }
586
587 // Execute actions
588 for action in actions {
589 if let Ok(result) =
590 protocol.as_ref().execute_action(action)
591 {
592 match result {
593 ClientActionResult::SendData(bytes) => {
594 if write_half_clone
595 .lock()
596 .await
597 .write_all(&bytes)
598 .await
599 .is_ok()
600 {
601 trace!("HTTP proxy client {} sent {} bytes", client_id, bytes.len());
602 }
603 }
604 ClientActionResult::Disconnect => {
605 // Labeled: a bare `break` here only
606 // exits the actions for-loop and the
607 // tunnel keeps running.
608 info!("HTTP proxy client {} disconnecting", client_id);
609 app_state_clone
610 .update_client_status(client_id, ClientStatus::Disconnected)
611 .await;
612 let _ = status_tx_clone.send(format!(
613 "[CLIENT] HTTP proxy client {} disconnected",
614 client_id
615 ));
616 let _ = status_tx_clone.send("__UPDATE_UI__".to_string());
617 break 'tunnel;
618 }
619 _ => {}
620 }
621 }
622 }
623 }
624 Err(e) => {
625 error!(
626 "LLM error for HTTP proxy client {}: {}",
627 client_id, e
628 );
629 }
630 }
631 }
632
633 // Process queued data if any
634 let mut client_data_lock = client_data_clone.lock().await;
635 if !client_data_lock.queued_data.is_empty() {
636 client_data_lock.queued_data.clear();
637 }
638 client_data_lock.state = ConnectionState::Idle;
639 }
640 ConnectionState::Processing => {
641 // Queue data
642 client_data_lock.queued_data.extend_from_slice(&data);
643 client_data_lock.state = ConnectionState::Accumulating;
644 }
645 ConnectionState::Accumulating => {
646 // Continue queuing
647 client_data_lock.queued_data.extend_from_slice(&data);
648 }
649 }
650 }
651 Err(e) => {
652 error!("HTTP proxy client {} read error: {}", client_id, e);
653 app_state_clone
654 .update_client_status(client_id, ClientStatus::Error(e.to_string()))
655 .await;
656 let _ = status_tx_clone.send("__UPDATE_UI__".to_string());
657 break;
658 }
659 }
660 }
661 }
662 .await;
663 handle_state.remove_client_handle(client_id).await;
664 });
665 task_registrar
666 .register_client_task(client_id, task_handle)
667 .await;
668
669 Ok(local_addr)
670 }
671}
672
673/// The CONNECT request the client puts on the wire — the one encoder for both the LLM's
674/// `establish_tunnel` (via the tunnel task) and a dashboard-injected one.
675///
676/// `proxy_authorization` is the already-encoded credential from the `proxy_auth` startup
677/// parameter (`Basic <base64(user:pass)>`). RFC 7235 §4.4 puts it on the CONNECT itself,
678/// which is the only request the proxy ever sees — everything after it is tunnel bytes.
679fn connect_request(
680 target_host: &str,
681 target_port: &str,
682 proxy_authorization: Option<&str>,
683) -> String {
684 let auth = match proxy_authorization {
685 Some(value) => format!("Proxy-Authorization: {value}\r\n"),
686 None => String::new(),
687 };
688 format!(
689 "CONNECT {}:{} HTTP/1.1\r\nHost: {}:{}\r\n{}\r\n",
690 target_host, target_port, target_host, target_port, auth
691 )
692}
693
694/// Command-task arm. `send_http_request` / `send_data` / `disconnect` / `wait_for_more` are
695/// `SendData`-shaped and go through the generic helper. `establish_tunnel` returns
696/// `ClientActionResult::Custom`, which the generic arm cannot execute, so it is mapped here
697/// through the same CONNECT encoder the tunnel task uses; the proxy's reply then reaches the
698/// read loop as ordinary tunnel bytes (an `http_proxy_response_received` event).
699async fn handle_injected_command<W>(
700 write_half: &Arc<Mutex<W>>,
701 command: ClientCommand,
702 client_id: ClientId,
703 state: &AppState,
704 status_tx: &mpsc::UnboundedSender<String>,
705) -> bool
706where
707 W: tokio::io::AsyncWrite + Unpin,
708{
709 let protocol = HttpProxyClientProtocol::new();
710 if command.action.get("type").and_then(|v| v.as_str()) != Some("establish_tunnel") {
711 return crate::client::command_support::handle_stream_client_command(
712 &protocol, write_half, command, client_id, state, status_tx,
713 )
714 .await;
715 }
716
717 let action = command.action.clone();
718 let outcome: anyhow::Result<ClientSendOutcome> = match protocol.execute_action(action.clone()) {
719 Ok(ClientActionResult::Custom { data, .. }) => {
720 // The same two startup parameters the LLM path honours: an injected CONNECT
721 // that names no target uses `default_target`, and every injected CONNECT
722 // carries the `proxy_auth` credential.
723 let (default_target, proxy_authorization) = state
724 .with_client_mut(client_id, |client| {
725 (
726 client
727 .get_protocol_field("default_target")
728 .and_then(|v| v.as_str())
729 .map(|s| s.to_string()),
730 client
731 .get_protocol_field("proxy_authorization")
732 .and_then(|v| v.as_str())
733 .map(|s| s.to_string()),
734 )
735 })
736 .await
737 .unwrap_or((None, None));
738
739 let (host, port) = match (
740 data.get("target_host").and_then(|v| v.as_str()),
741 data.get("target_port").and_then(|v| v.as_u64()),
742 ) {
743 (Some(host), Some(port)) => (host.to_string(), port.to_string()),
744 _ => match default_target.as_deref().and_then(|t| t.rsplit_once(':')) {
745 Some((host, port)) => (host.to_string(), port.to_string()),
746 None => (String::new(), "0".to_string()),
747 },
748 };
749 let bytes = connect_request(&host, &port, proxy_authorization.as_deref()).into_bytes();
750 let mut guard = write_half.lock().await;
751 match guard.write_all(&bytes).await {
752 Ok(()) => match guard.flush().await {
753 Ok(()) => Ok(ClientSendOutcome::Sent {
754 bytes_sent: bytes.len(),
755 }),
756 Err(e) => Err(anyhow::anyhow!("flush failed: {e}")),
757 },
758 Err(e) => Err(anyhow::anyhow!("write failed: {e}")),
759 }
760 }
761 Ok(_) => Ok(ClientSendOutcome::Executed {
762 detail: "executed".to_string(),
763 }),
764 Err(e) => Ok(ClientSendOutcome::Rejected {
765 error: e.to_string(),
766 }),
767 };
768
769 // Same access-log shape as the generic arm, so the injection shows in the request pane.
770 let outcome_json = match &outcome {
771 Ok(outcome) => serde_json::to_value(outcome).unwrap_or(serde_json::Value::Null),
772 Err(e) => serde_json::json!({"error": e.to_string()}),
773 };
774 state
775 .record_access_log(
776 AccessLogOwner::Client(client_id.as_u32()),
777 crate::llm::actions::protocol_trait::Protocol::protocol_name(&protocol),
778 None,
779 "injected_action",
780 action,
781 vec![outcome_json],
782 )
783 .await;
784 match &outcome {
785 Ok(ClientSendOutcome::Sent { bytes_sent }) => info!(
786 "HTTP proxy client {} sent injected CONNECT ({} bytes)",
787 client_id, bytes_sent
788 ),
789 Ok(_) => {}
790 Err(e) => {
791 error!(
792 "HTTP proxy client {} injected establish_tunnel failed: {}",
793 client_id, e
794 );
795 let _ = status_tx.send(format!(
796 "[WARN] Client {} injected action failed: {}",
797 client_id, e
798 ));
799 }
800 }
801 let _ = status_tx.send("__UPDATE_UI__".to_string());
802 crate::client::command_support::reply(command, outcome);
803 false
804}