Stage 3

codetracer-python-recorder/Cargo.lock: 
codetracer-python-recorder/Cargo.toml: 
codetracer-python-recorder/src/ffi.rs: 
codetracer-python-recorder/src/policy.rs: 
codetracer-python-recorder/src/runtime/io_capture/events.rs: 
codetracer-python-recorder/src/runtime/io_capture/install.rs: 
codetracer-python-recorder/src/runtime/io_capture/mod.rs: 
codetracer-python-recorder/src/runtime/io_capture/mute.rs: 
codetracer-python-recorder/src/runtime/io_capture/proxies.rs: 
codetracer-python-recorder/src/runtime/io_capture/sink.rs: 
codetracer-python-recorder/src/runtime/logging.rs: 
codetracer-python-recorder/src/runtime/mod.rs: 
codetracer-python-recorder/src/session.rs: 
design-docs/io-capture-line-proxy-implementation-plan.status.md: 

Signed-off-by: Tzanko Matev <[email protected]>

Author: tzanko-matev

codetracer-python-recorder/Cargo.lock

@@ -31,6 +31,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 uuid = { version = "1.10", features = ["v4"] }
 recorder-errors = { version = "0.1.0", path = "crates/recorder-errors" }
+base64 = "0.22"
 
 [dev-dependencies]
 pyo3 = { version = "0.25.1", features = ["auto-initialize"] }

codetracer-python-recorder/Cargo.toml

@@ -145,12 +145,9 @@ mod tests {
     #[test]
     fn map_recorder_error_sets_python_attributes() {
         Python::with_gil(|py| {
-            let err = usage!(
-                ErrorCode::UnsupportedFormat,
-                "invalid trace format"
-            )
-            .with_context("format", "yaml")
-            .with_source(std::io::Error::new(std::io::ErrorKind::Other, "boom"));
+            let err = usage!(ErrorCode::UnsupportedFormat, "invalid trace format")
+                .with_context("format", "yaml")
+                .with_source(std::io::Error::new(std::io::ErrorKind::Other, "boom"));
             let pyerr = map_recorder_error(err);
             let ty = pyerr.get_type(py);
             assert!(ty.is(py.get_type::<PyUsageError>()));
@@ -191,9 +188,8 @@ mod tests {
     #[test]
     fn dispatch_converts_recorder_error_to_pyerr() {
         Python::with_gil(|py| {
-            let result: PyResult<()> = dispatch("dispatch_env", || {
-                Err(enverr!(ErrorCode::Io, "disk full"))
-            });
+            let result: PyResult<()> =
+                dispatch("dispatch_env", || Err(enverr!(ErrorCode::Io, "disk full")));
             let err = result.expect_err("expected PyErr");
             let ty = err.get_type(py);
             assert!(ty.is(py.get_type::<PyEnvironmentError>()));

codetracer-python-recorder/src/ffi.rs

@@ -20,6 +20,8 @@ pub const ENV_LOG_LEVEL: &str = "CODETRACER_LOG_LEVEL";
 pub const ENV_LOG_FILE: &str = "CODETRACER_LOG_FILE";
 /// Environment variable enabling JSON error trailers on stderr.
 pub const ENV_JSON_ERRORS: &str = "CODETRACER_JSON_ERRORS";
+/// Environment variable toggling IO capture strategies.
+pub const ENV_CAPTURE_IO: &str = "CODETRACER_CAPTURE_IO";
 
 static POLICY: OnceCell<RwLock<RecorderPolicy>> = OnceCell::new();
 
@@ -61,6 +63,21 @@ impl FromStr for OnRecorderError {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct IoCapturePolicy {
+    pub line_proxies: bool,
+    pub fd_fallback: bool,
+}
+
+impl Default for IoCapturePolicy {
+    fn default() -> Self {
+        Self {
+            line_proxies: true,
+            fd_fallback: false,
+        }
+    }
+}
+
 /// Recorder-wide runtime configuration.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct RecorderPolicy {
@@ -70,6 +87,7 @@ pub struct RecorderPolicy {
     pub log_level: Option<String>,
     pub log_file: Option<PathBuf>,
     pub json_errors: bool,
+    pub io_capture: IoCapturePolicy,
 }
 
 impl Default for RecorderPolicy {
@@ -81,6 +99,7 @@ impl Default for RecorderPolicy {
             log_level: None,
             log_file: None,
             json_errors: false,
+            io_capture: IoCapturePolicy::default(),
         }
     }
 }
@@ -111,6 +130,16 @@ impl RecorderPolicy {
         if let Some(json_errors) = update.json_errors {
             self.json_errors = json_errors;
         }
+        if let Some(line_proxies) = update.io_capture_line_proxies {
+            self.io_capture.line_proxies = line_proxies;
+            if !self.io_capture.line_proxies {
+                self.io_capture.fd_fallback = false;
+            }
+        }
+        if let Some(fd_fallback) = update.io_capture_fd_fallback {
+            // fd fallback requires proxies to be on.
+            self.io_capture.fd_fallback = fd_fallback && self.io_capture.line_proxies;
+        }
     }
 }
 
@@ -130,6 +159,8 @@ struct PolicyUpdate {
     log_level: Option<String>,
     log_file: Option<PolicyPath>,
     json_errors: Option<bool>,
+    io_capture_line_proxies: Option<bool>,
+    io_capture_fd_fallback: Option<bool>,
 }
 
 /// Snapshot the current policy.
@@ -178,6 +209,12 @@ pub fn configure_policy_from_env() -> RecorderResult<()> {
         update.json_errors = Some(parse_bool(&value)?);
     }
 
+    if let Ok(value) = env::var(ENV_CAPTURE_IO) {
+        let (line_proxies, fd_fallback) = parse_capture_io(&value)?;
+        update.io_capture_line_proxies = Some(line_proxies);
+        update.io_capture_fd_fallback = Some(fd_fallback);
+    }
+
     apply_policy_update(update);
     Ok(())
 }
@@ -194,6 +231,54 @@ fn parse_bool(value: &str) -> RecorderResult<bool> {
     }
 }
 
+fn parse_capture_io(value: &str) -> RecorderResult<(bool, bool)> {
+    let trimmed = value.trim();
+    if trimmed.is_empty() {
+        let default = IoCapturePolicy::default();
+        return Ok((default.line_proxies, default.fd_fallback));
+    }
+
+    let lower = trimmed.to_ascii_lowercase();
+    if matches!(
+        lower.as_str(),
+        "0" | "off" | "false" | "disable" | "disabled" | "none"
+    ) {
+        return Ok((false, false));
+    }
+    if matches!(lower.as_str(), "1" | "on" | "true" | "enable" | "enabled") {
+        return Ok((true, false));
+    }
+
+    let mut line_proxies = false;
+    let mut fd_fallback = false;
+    for token in lower.split(',') {
+        match token.trim() {
+            "" => {}
+            "proxies" | "proxy" => line_proxies = true,
+            "fd" | "mirror" | "fallback" => {
+                line_proxies = true;
+                fd_fallback = true;
+            }
+            other => {
+                return Err(usage!(
+                    ErrorCode::InvalidPolicyValue,
+                    "invalid CODETRACER_CAPTURE_IO value '{}'",
+                    other
+                ));
+            }
+        }
+    }
+
+    if !line_proxies && !fd_fallback {
+        return Err(usage!(
+            ErrorCode::InvalidPolicyValue,
+            "CODETRACER_CAPTURE_IO must enable at least 'proxies' or 'fd'"
+        ));
+    }
+
+    Ok((line_proxies, fd_fallback))
+}
+
 // === PyO3 helpers ===
 
 use pyo3::prelude::*;
@@ -202,14 +287,16 @@ use pyo3::types::PyDict;
 use crate::ffi;
 
 #[pyfunction(name = "configure_policy")]
-#[pyo3(signature = (on_recorder_error=None, require_trace=None, keep_partial_trace=None, log_level=None, log_file=None, json_errors=None))]
+#[pyo3(signature = (on_recorder_error=None, require_trace=None, keep_partial_trace=None, log_level=None, log_file=None, json_errors=None, io_capture_line_proxies=None, io_capture_fd_fallback=None))]
 pub fn configure_policy_py(
     on_recorder_error: Option<&str>,
     require_trace: Option<bool>,
     keep_partial_trace: Option<bool>,
     log_level: Option<&str>,
     log_file: Option<&str>,
     json_errors: Option<bool>,
+    io_capture_line_proxies: Option<bool>,
+    io_capture_fd_fallback: Option<bool>,
 ) -> PyResult<()> {
     let mut update = PolicyUpdate::default();
 
@@ -245,6 +332,14 @@ pub fn configure_policy_py(
         update.json_errors = Some(value);
     }
 
+    if let Some(value) = io_capture_line_proxies {
+        update.io_capture_line_proxies = Some(value);
+    }
+
+    if let Some(value) = io_capture_fd_fallback {
+        update.io_capture_fd_fallback = Some(value);
+    }
+
     apply_policy_update(update);
     Ok(())
 }
@@ -278,6 +373,11 @@ pub fn py_policy_snapshot(py: Python<'_>) -> PyResult<PyObject> {
         dict.set_item("log_file", py.None())?;
     }
     dict.set_item("json_errors", snapshot.json_errors)?;
+
+    let io_dict = PyDict::new(py);
+    io_dict.set_item("line_proxies", snapshot.io_capture.line_proxies)?;
+    io_dict.set_item("fd_fallback", snapshot.io_capture.fd_fallback)?;
+    dict.set_item("io_capture", io_dict)?;
     Ok(dict.into())
 }
 
@@ -301,6 +401,8 @@ mod tests {
         assert!(!snap.json_errors);
         assert!(snap.log_level.is_none());
         assert!(snap.log_file.is_none());
+        assert!(snap.io_capture.line_proxies);
+        assert!(!snap.io_capture.fd_fallback);
     }
 
     #[test]
@@ -313,6 +415,8 @@ mod tests {
         update.log_level = Some("debug".to_string());
         update.log_file = Some(PolicyPath::Value(PathBuf::from("/tmp/log.txt")));
         update.json_errors = Some(true);
+        update.io_capture_line_proxies = Some(true);
+        update.io_capture_fd_fallback = Some(true);
 
         apply_policy_update(update);
 
@@ -323,6 +427,8 @@ mod tests {
         assert_eq!(snap.log_level.as_deref(), Some("debug"));
         assert_eq!(snap.log_file.as_deref(), Some(Path::new("/tmp/log.txt")));
         assert!(snap.json_errors);
+        assert!(snap.io_capture.line_proxies);
+        assert!(snap.io_capture.fd_fallback);
         reset_policy();
     }
 
@@ -336,6 +442,7 @@ mod tests {
         env::set_var(ENV_LOG_LEVEL, "info");
         env::set_var(ENV_LOG_FILE, "/tmp/out.log");
         env::set_var(ENV_JSON_ERRORS, "yes");
+        env::set_var(ENV_CAPTURE_IO, "proxies,fd");
 
         configure_policy_from_env().expect("configure from env");
 
@@ -348,6 +455,8 @@ mod tests {
         assert_eq!(snap.log_level.as_deref(), Some("info"));
         assert_eq!(snap.log_file.as_deref(), Some(Path::new("/tmp/out.log")));
         assert!(snap.json_errors);
+        assert!(snap.io_capture.line_proxies);
+        assert!(snap.io_capture.fd_fallback);
         reset_policy();
     }
 
@@ -364,6 +473,19 @@ mod tests {
         reset_policy();
     }
 
+    #[test]
+    fn configure_policy_from_env_rejects_invalid_capture_io() {
+        reset_policy();
+        let env_guard = env_lock();
+        env::set_var(ENV_CAPTURE_IO, "invalid-token");
+
+        let err = configure_policy_from_env().expect_err("invalid capture io should error");
+        assert_eq!(err.code, ErrorCode::InvalidPolicyValue);
+
+        drop(env_guard);
+        reset_policy();
+    }
+
     fn env_lock() -> EnvGuard {
         EnvGuard
     }
@@ -379,6 +501,7 @@ mod tests {
                 ENV_LOG_LEVEL,
                 ENV_LOG_FILE,
                 ENV_JSON_ERRORS,
+                ENV_CAPTURE_IO,
             ] {
                 env::remove_var(key);
             }

codetracer-python-recorder/src/policy.rs

@@ -1,5 +1,6 @@
 use crate::runtime::line_snapshots::FrameId;
 use pyo3::Python;
+use runtime_tracing::{Line, PathId};
 use std::fmt;
 use std::thread::ThreadId;
 use std::time::Instant;
@@ -43,6 +44,9 @@ pub struct ProxyEvent {
     pub thread_id: ThreadId,
     pub timestamp: Instant,
     pub frame_id: Option<FrameId>,
+    pub path_id: Option<PathId>,
+    pub line: Option<Line>,
+    pub path: Option<String>,
 }
 
 /// Sink for proxy events. Later stages swap in a real writer-backed implementation.

codetracer-python-recorder/src/runtime/io_capture/events.rs

@@ -6,7 +6,6 @@ use std::sync::Arc;
 #[cfg_attr(not(test), allow(dead_code))]
 /// Controller that installs the proxies and restores the original streams.
 pub struct IoStreamProxies {
-    _sink: Arc<dyn ProxySink>,
     _stdout_proxy: Py<LineAwareStdout>,
     _stderr_proxy: Py<LineAwareStderr>,
     _stdin_proxy: Py<LineAwareStdin>,
@@ -24,19 +23,24 @@ impl IoStreamProxies {
         let stderr_original = sys.getattr("stderr")?.unbind();
         let stdin_original = sys.getattr("stdin")?.unbind();
 
-        let stdout_proxy =
-            Py::new(py, LineAwareStdout::new(stdout_original.clone_ref(py), sink.clone()))?;
-        let stderr_proxy =
-            Py::new(py, LineAwareStderr::new(stderr_original.clone_ref(py), sink.clone()))?;
-        let stdin_proxy =
-            Py::new(py, LineAwareStdin::new(stdin_original.clone_ref(py), sink.clone()))?;
+        let stdout_proxy = Py::new(
+            py,
+            LineAwareStdout::new(stdout_original.clone_ref(py), sink.clone()),
+        )?;
+        let stderr_proxy = Py::new(
+            py,
+            LineAwareStderr::new(stderr_original.clone_ref(py), sink.clone()),
+        )?;
+        let stdin_proxy = Py::new(
+            py,
+            LineAwareStdin::new(stdin_original.clone_ref(py), sink.clone()),
+        )?;
 
         sys.setattr("stdout", stdout_proxy.clone_ref(py))?;
         sys.setattr("stderr", stderr_proxy.clone_ref(py))?;
         sys.setattr("stdin", stdin_proxy.clone_ref(py))?;
 
         Ok(Self {
-            _sink: sink,
             _stdout_proxy: stdout_proxy,
             _stderr_proxy: stderr_proxy,
             _stdin_proxy: stdin_proxy,
@@ -58,7 +62,6 @@ impl IoStreamProxies {
         self.installed = false;
         Ok(())
     }
-
 }
 
 impl Drop for IoStreamProxies {

codetracer-python-recorder/src/runtime/io_capture/install.rs

@@ -1,13 +1,16 @@
 pub mod events;
-pub mod sink;
-pub mod proxies;
 pub mod install;
+pub mod mute;
+pub mod proxies;
+pub mod sink;
 
 #[allow(unused_imports)]
 pub use events::{IoOperation, IoStream, NullSink, ProxyEvent, ProxySink};
 #[allow(unused_imports)]
-pub use sink::{IoChunk, IoChunkConsumer, IoChunkFlags, IoEventSink};
+pub use install::IoStreamProxies;
+#[allow(unused_imports)]
+pub use mute::{is_io_capture_muted, ScopedMuteIoCapture};
 #[allow(unused_imports)]
 pub use proxies::{LineAwareStderr, LineAwareStdin, LineAwareStdout};
 #[allow(unused_imports)]
-pub use install::IoStreamProxies;
+pub use sink::{IoChunk, IoChunkConsumer, IoChunkFlags, IoEventSink};