feat(guest-agent,orchestrator): generalize op_id to all guest agent communications

Every request now carries an op_id (auto-generated UUID4 hex on the Python
side), and every response echoes it back. The Rust guest-agent extracts
op_id from raw JSON before serde deserialization (two-phase parse), then
injects it into all outgoing messages via a new ResponseWriter struct
(using entry().or_insert_with() to preserve existing op_id on file
transfer responses). On the Python side, DualPortChannel._with_op_id()
context manager handles op_id generation, queue registration, command
dispatch, and cleanup — used by both send_request() and stream_messages().

This eliminates the shared default queue that caused stale messages (e.g.
PongMessage from reconnection probes) to pollute subsequent operations.

Key changes:
- Rust: ResponseWriter wraps mpsc::Sender + op_id, replaces direct
  send_response/send_streaming_error calls in all handler paths
- Rust: Remove dead CmdError::Reply.op_id field, with_op_id() method,
  and CmdError::io() second parameter (ResponseWriter handles injection)
- Rust: Remove ActiveWriteHandle.op_id field and unused write_tx param
  from handle_write_file
- Python: GuestAgentRequest base gets optional op_id; GuestAgentResponse
  base class added with optional op_id for all response types
- Python: FileOpDispatcher routes ALL messages by op_id; _default_queue,
  get_default_message(), stream_events() removed as dead code
- Python: Messages without op_id logged and discarded (prevents memory leak)
- Python: ValidationError guard in dispatch loop prevents malformed
  messages from killing all routing
- Python: RuntimeError added to _probe_guest_ready exception tuple
- Python: Remove send_raw_request from both GuestChannel protocol and
  implementations (replaced by op_id-routed send_request)
- Python: OP_QUEUE_DEPTH=64 constant for bounded per-op queues

Author: clemlesne

guest-agent/src/connection.rs

@@ -4,7 +4,7 @@ use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
 use tokio::sync::mpsc;
 
 use crate::constants::*;
-use crate::error::{CmdError, send_response};
+use crate::error::{CmdError, ResponseWriter};
 use crate::types::{FileEntry, GuestResponse};
 
 // ============================================================================
@@ -67,7 +67,6 @@ pub(crate) struct ActiveWriteHandle {
     pub(crate) chunk_tx: mpsc::Sender<Vec<u8>>,
     pub(crate) task: tokio::task::JoinHandle<Result<WriteResult, WriteError>>,
     pub(crate) path_display: String,
-    pub(crate) op_id: String,
 }
 
 pub(crate) struct WriteResult {
@@ -149,105 +148,87 @@ pub(crate) fn validate_file_path(relative_path: &str) -> Result<std::path::PathB
 pub(crate) async fn handle_read_file(
     op_id: &str,
     path: &str,
-    write_tx: &mpsc::Sender<Vec<u8>>,
+    writer: &ResponseWriter,
 ) -> Result<(), CmdError> {
-    let resolved_path = validate_file_path(path).map_err(|e| {
-        CmdError::validation(format!("Invalid path '{path}': {e}")).with_op_id(op_id)
-    })?;
+    let resolved_path = validate_file_path(path)
+        .map_err(|e| CmdError::validation(format!("Invalid path '{path}': {e}")))?;
 
     if resolved_path == std::path::Path::new(SANDBOX_ROOT) {
-        return Err(CmdError::validation("Cannot read a directory").with_op_id(op_id));
+        return Err(CmdError::validation("Cannot read a directory"));
     }
 
-    let canonical_path = tokio::fs::canonicalize(&resolved_path).await.map_err(|e| {
-        CmdError::io(
-            format!("File not found or inaccessible '{path}': {e}"),
-            Some(op_id),
-        )
-    })?;
+    let canonical_path = tokio::fs::canonicalize(&resolved_path)
+        .await
+        .map_err(|e| CmdError::io(format!("File not found or inaccessible '{path}': {e}")))?;
 
     let sandbox_canonical = tokio::fs::canonicalize(SANDBOX_ROOT)
         .await
         .unwrap_or_else(|_| std::path::PathBuf::from(SANDBOX_ROOT));
     if !canonical_path.starts_with(&sandbox_canonical) {
-        return Err(
-            CmdError::validation(format!("Path '{path}' resolves outside sandbox"))
-                .with_op_id(op_id),
-        );
+        return Err(CmdError::validation(format!(
+            "Path '{path}' resolves outside sandbox"
+        )));
     }
 
     let metadata = tokio::fs::metadata(&canonical_path)
         .await
-        .map_err(|e| CmdError::io(format!("Cannot read '{path}': {e}"), Some(op_id)))?;
+        .map_err(|e| CmdError::io(format!("Cannot read '{path}': {e}")))?;
 
     if metadata.is_dir() {
-        return Err(
-            CmdError::validation(format!("'{path}' is a directory, not a file")).with_op_id(op_id),
-        );
+        return Err(CmdError::validation(format!(
+            "'{path}' is a directory, not a file"
+        )));
     }
     if metadata.len() as usize > MAX_FILE_SIZE_BYTES {
         return Err(CmdError::validation(format!(
             "File too large: {} bytes (max {})",
             metadata.len(),
             MAX_FILE_SIZE_BYTES
-        ))
-        .with_op_id(op_id));
+        )));
     }
 
     let file_size = metadata.len();
     let file = std::fs::File::open(&canonical_path)
-        .map_err(|e| CmdError::io(format!("Failed to read '{path}': {e}"), Some(op_id)))?;
+        .map_err(|e| CmdError::io(format!("Failed to read '{path}': {e}")))?;
 
-    let mut encoder =
-        zstd::stream::read::Encoder::new(file, FILE_TRANSFER_ZSTD_LEVEL).map_err(|e| {
-            CmdError::io(
-                format!("Compression init failed for '{path}': {e}"),
-                Some(op_id),
-            )
-        })?;
+    let mut encoder = zstd::stream::read::Encoder::new(file, FILE_TRANSFER_ZSTD_LEVEL)
+        .map_err(|e| CmdError::io(format!("Compression init failed for '{path}': {e}")))?;
 
     let mut buf = vec![0u8; FILE_TRANSFER_CHUNK_SIZE];
     loop {
         let n = match std::io::Read::read(&mut encoder, &mut buf) {
             Ok(0) => break,
             Ok(n) => n,
             Err(e) => {
-                return Err(CmdError::io(format!("Compression error: {e}"), Some(op_id)));
+                return Err(CmdError::io(format!("Compression error: {e}")));
             }
         };
         let chunk_b64 = BASE64.encode(&buf[..n]);
-        send_response(
-            write_tx,
-            &GuestResponse::FileChunk {
+        writer
+            .send(&GuestResponse::FileChunk {
                 op_id: op_id.to_string(),
                 data: chunk_b64,
-            },
-        )
-        .await?;
+            })
+            .await?;
     }
 
-    send_response(
-        write_tx,
-        &GuestResponse::FileReadComplete {
+    writer
+        .send(&GuestResponse::FileReadComplete {
             op_id: op_id.to_string(),
             path: path.to_string(),
             size: file_size,
-        },
-    )
-    .await
+        })
+        .await
 }
 
 /// Handle list_files command: read directory entries.
-pub(crate) async fn handle_list_files(
-    path: &str,
-    write_tx: &mpsc::Sender<Vec<u8>>,
-) -> Result<(), CmdError> {
+pub(crate) async fn handle_list_files(path: &str, writer: &ResponseWriter) -> Result<(), CmdError> {
     let resolved_path = validate_file_path(path)
         .map_err(|e| CmdError::validation(format!("Invalid path '{path}': {e}")))?;
 
     let mut read_dir = tokio::fs::read_dir(&resolved_path)
         .await
-        .map_err(|e| CmdError::io(format!("Cannot list '{path}': {e}"), None))?;
+        .map_err(|e| CmdError::io(format!("Cannot list '{path}': {e}")))?;
 
     let mut entries = Vec::new();
     while let Ok(Some(entry)) = read_dir.next_entry().await {
@@ -264,14 +245,12 @@ pub(crate) async fn handle_list_files(
 
     entries.sort_by(|a, b| a.name.cmp(&b.name));
 
-    send_response(
-        write_tx,
-        &GuestResponse::FileList {
+    writer
+        .send(&GuestResponse::FileList {
             path: path.to_string(),
             entries,
-        },
-    )
-    .await
+        })
+        .await
 }
 
 /// Set up a pipelined file write operation.
@@ -281,25 +260,20 @@ pub(crate) async fn handle_write_file(
     op_id: &str,
     path: &str,
     make_executable: bool,
-    write_tx: &mpsc::Sender<Vec<u8>>,
 ) -> Result<ActiveWriteHandle, CmdError> {
-    let resolved_path = validate_file_path(path).map_err(|e| {
-        CmdError::validation(format!("Invalid path '{path}': {e}")).with_op_id(op_id)
-    })?;
+    let resolved_path = validate_file_path(path)
+        .map_err(|e| CmdError::validation(format!("Invalid path '{path}': {e}")))?;
 
     if resolved_path == std::path::Path::new(SANDBOX_ROOT) {
-        return Err(
-            CmdError::validation("Cannot write to sandbox root directory").with_op_id(op_id),
-        );
+        return Err(CmdError::validation(
+            "Cannot write to sandbox root directory",
+        ));
     }
 
     if let Some(parent) = resolved_path.parent()
         && let Err(e) = std::fs::create_dir_all(parent)
     {
-        return Err(CmdError::io(
-            format!("Failed to create directories: {e}"),
-            Some(op_id),
-        ));
+        return Err(CmdError::io(format!("Failed to create directories: {e}")));
     }
 
     let tmp_path = resolved_path
@@ -308,7 +282,7 @@ pub(crate) async fn handle_write_file(
         .join(format!(".wr.{op_id}.tmp"));
 
     let file = std::fs::File::create(&tmp_path)
-        .map_err(|e| CmdError::io(format!("Failed to create temp file: {e}"), Some(op_id)))?;
+        .map_err(|e| CmdError::io(format!("Failed to create temp file: {e}")))?;
 
     let counting_writer = CountingWriter {
         inner: file,
@@ -319,10 +293,7 @@ pub(crate) async fn handle_write_file(
         Ok(d) => d,
         Err(e) => {
             let _ = std::fs::remove_file(&tmp_path);
-            return Err(CmdError::io(
-                format!("Decompression init failed: {e}"),
-                Some(op_id),
-            ));
+            return Err(CmdError::io(format!("Decompression init failed: {e}")));
         }
     };
 
@@ -391,17 +362,10 @@ pub(crate) async fn handle_write_file(
         })
     });
 
-    // Note: handle_write_file previously returned Ok(None) on validation errors.
-    // Now those are Err(CmdError::Reply), and success always returns the handle.
-    // The unused write_tx parameter is kept for API consistency; validation errors
-    // now propagate via CmdError to the dispatcher.
-    let _ = write_tx;
-
     Ok(ActiveWriteHandle {
         chunk_tx,
         task,
         path_display: path.to_string(),
-        op_id: op_id.to_string(),
     })
 }
 

guest-agent/src/error.rs

@@ -1,7 +1,6 @@
 //! Package installation (pip, bun) with validation.
 
 use tokio::process::Command;
-use tokio::sync::mpsc;
 
 use crate::constants::*;
 use crate::error::*;
@@ -115,7 +114,7 @@ pub(crate) fn validate_package_name(pkg: &str) -> Result<(), String> {
 pub(crate) async fn install_packages(
     language: Language,
     packages: &[String],
-    write_tx: &mpsc::Sender<Vec<u8>>,
+    writer: &ResponseWriter,
 ) -> Result<(), CmdError> {
     use std::time::Instant;
     use tokio::time::Duration;
@@ -227,35 +226,29 @@ pub(crate) async fn install_packages(
     }
 
     if !stdout_lines.is_empty() {
-        send_response(
-            write_tx,
-            &GuestResponse::Stdout {
+        writer
+            .send(&GuestResponse::Stdout {
                 chunk: stdout_lines.join("\n") + "\n",
-            },
-        )
-        .await?;
+            })
+            .await?;
     }
 
     if !stderr_lines.is_empty() {
-        send_response(
-            write_tx,
-            &GuestResponse::Stderr {
+        writer
+            .send(&GuestResponse::Stderr {
                 chunk: stderr_lines.join("\n") + "\n",
-            },
-        )
-        .await?;
+            })
+            .await?;
     }
 
-    send_response(
-        write_tx,
-        &GuestResponse::Complete {
+    writer
+        .send(&GuestResponse::Complete {
             exit_code,
             execution_time_ms: duration_ms,
             spawn_ms: None,
             process_ms: None,
-        },
-    )
-    .await?;
+        })
+        .await?;
 
     Ok(())
 }