Stage 5

README.md: 
codetracer-python-recorder/src/runtime/io_capture/fd_mirror/unix.rs: 
design-docs/io-capture-line-proxy-implementation-plan.status.md: 

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

Author: tzanko-matev

README.md

@@ -50,12 +50,34 @@ Pass `--codetracer-json-errors` (or set the policy via `configure_policy(json_er
 
 ### IO capture configuration
 
-Line-aware stdout/stderr capture proxies are now enabled by default. Control them through the policy layer:
+Line-aware capture (see [ADR 0008](design-docs/adr/0008-line-aware-io-capture.md)) installs `LineAwareStdout`, `LineAwareStderr`, and `LineAwareStdin` proxies so every chunk carries `{path_id, line, frame_id}` metadata. The proxies forward writes immediately to keep TTY behaviour unchanged and the batching sink emits newline/flush/step-delimited chunks. When the FD mirror fallback observes bytes that bypassed the proxies, the resulting `IoChunk` carries the `mirror` flag so downstream tooling can highlight native writers separately. Recorder logs and telemetry use `ScopedMuteIoCapture` to avoid recursive capture.
+
+Control the feature through the policy layer:
 
 - CLI: `python -m codetracer_python_recorder --io-capture=off script.py` disables capture, while `--io-capture=proxies+fd` also mirrors raw file-descriptor writes.
 - Python: `configure_policy(io_capture_line_proxies=False)` toggles proxies, and `configure_policy(io_capture_fd_fallback=True)` enables the FD fallback.
 - Environment: set `CODETRACER_CAPTURE_IO=off`, `proxies`, or `proxies+fd` (`,` is also accepted) to match the CLI and Python helpers.
 
+Manual smoke check: `python -m codetracer_python_recorder examples/stdout_script.py` should report the proxied output while leaving the console live.
+
+#### Troubleshooting replaced stdout/stderr
+
+Third-party tooling occasionally replaces `sys.stdout` / `sys.stderr` after the proxies install. When that happens, IO metadata stops updating and the recorder falls back to passthrough behaviour. You can verify the binding at runtime:
+
+```python
+import sys
+from codetracer_python_recorder.runtime import LineAwareStdout, LineAwareStderr
+
+print(type(sys.stdout).__name__, isinstance(sys.stdout, LineAwareStdout))
+print(type(sys.stderr).__name__, isinstance(sys.stderr, LineAwareStderr))
+```
+
+Both `isinstance` checks should return `True`. If they do not:
+
+1. Re-run `configure_policy(io_capture_line_proxies=True)` (or restart tracing) to reinstall the proxies before the other tool mutates the streams.
+2. Fall back to FD mirroring by enabling `CODETRACER_CAPTURE_IO=proxies+fd` so native writes still reach the ledger-backed mirror.
+3. As a last resort, disable IO capture (`--io-capture=off`) and rely on console output while investigating the conflicting integration.
+
 ### Migration checklist for downstream tools
 
 - Catch `RecorderError` (or a subclass) instead of `RuntimeError`.

codetracer-python-recorder/src/runtime/io_capture/fd_mirror/unix.rs

@@ -7,7 +7,7 @@ use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use std::thread;
-use std::time::Instant;
+use std::time::{Duration, Instant};
 
 #[derive(Debug)]
 pub struct FdMirrorError {
@@ -212,6 +212,9 @@ struct StreamMirror {
 }
 
 impl StreamMirror {
+    const SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(250);
+    const SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_millis(10);
+
     fn start(
         stream: IoStream,
         ledger: Arc<Ledger>,
@@ -289,14 +292,88 @@ impl StreamMirror {
             warn!("failed to restore fd {} after mirroring", self.target_fd);
         }
         if let Some(join) = self.join.take() {
-            if let Err(err) = join.join() {
-                warn!("mirror thread join failed: {err:?}");
+            if wait_for_join(&join, Self::SHUTDOWN_TIMEOUT, Self::SHUTDOWN_POLL_INTERVAL) {
+                if let Err(err) = join.join() {
+                    warn!("mirror thread join failed: {err:?}");
+                }
+            } else {
+                warn!(
+                    "mirror thread on fd {} did not stop within {:?}; detaching background reader",
+                    self.target_fd,
+                    Self::SHUTDOWN_TIMEOUT
+                );
+                drop(join);
             }
         }
         self.ledger.clear();
     }
 }
 
+fn wait_for_join(
+    handle: &thread::JoinHandle<()>,
+    timeout: Duration,
+    poll_interval: Duration,
+) -> bool {
+    if handle.is_finished() {
+        return true;
+    }
+    if timeout.is_zero() {
+        return false;
+    }
+    let mut remaining = timeout;
+    loop {
+        if handle.is_finished() {
+            return true;
+        }
+        let sleep_for = std::cmp::min(poll_interval, remaining);
+        thread::sleep(sleep_for);
+        if handle.is_finished() {
+            return true;
+        }
+        if remaining <= sleep_for {
+            return false;
+        }
+        remaining -= sleep_for;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn wait_for_join_succeeds_for_completed_thread() {
+        let handle = thread::spawn(|| {});
+        assert!(wait_for_join(
+            &handle,
+            Duration::from_millis(100),
+            Duration::from_millis(5)
+        ));
+        handle.join().expect("thread join should succeed");
+    }
+
+    #[test]
+    fn wait_for_join_times_out_for_blocked_thread() {
+        let shutdown = Arc::new(AtomicBool::new(false));
+        let worker_shutdown = shutdown.clone();
+        let handle = thread::spawn(move || {
+            while !worker_shutdown.load(Ordering::Relaxed) {
+                thread::sleep(Duration::from_millis(5));
+            }
+        });
+
+        assert!(
+            !wait_for_join(&handle, Duration::from_millis(15), Duration::from_millis(5)),
+            "wait_for_join should time out on a stuck thread"
+        );
+
+        shutdown.store(true, Ordering::Relaxed);
+        handle
+            .join()
+            .expect("thread join should succeed after shutdown signal");
+    }
+}
+
 impl Drop for StreamMirror {
     fn drop(&mut self) {
         self.shutdown();

design-docs/io-capture-line-proxy-implementation-plan.status.md

@@ -17,9 +17,12 @@
 - ✅ **Stage 2 – Implement IoEventSink and batching:** Added the `IoChunk` model plus a `IoEventSink` batching layer that groups stdout/stderr writes per thread, flushes on newline, explicit `flush()`, step boundaries, and 5 ms gaps, and emits stdin reads immediately. Updated the proxies to surface flush events and introduced focused unit tests that cover batching, timer splits, step flushes, and stdin capture. `just test` runs the sink tests alongside the existing proxy coverage.
 - ✅ **Stage 3 – Wire proxies into lifecycle:** `RuntimeTracer::install_io_capture` now instantiates the sink, installs the proxies behind the policy flag, and drains/flushed buffered chunks at step and finish boundaries. `IoChunk` records path IDs, frame IDs, and thread IDs sourced from the `LineSnapshotStore`, with a Python stack fallback filling metadata when monitoring snapshots are not yet available. Metadata emitted by `RecordEvent` now includes `path_id`, `line`, and `frame_id` for stdout/stderr chunks, and the Stage 3 integration test passes end-to-end.
 - 🔄 **Stage 4 – Optional FD mirror:** Implemented the shared ledger (`runtime::io_capture::fd_mirror`), plumbed optional `MirrorLedgers`/`FdMirrorController` through `IoEventSink` and `RuntimeTracer::install_io_capture`, and added runtime tests that assert `os.write` payloads are captured only when `io_capture_fd_fallback` is enabled. Next actions: tighten metadata/telemetry (expose mirror stats, warn when descriptor duplication fails) and stress-test concurrent native writers.
-- ⏳ **Stage 5 – Hardening and docs:** Not started.
+- 🔄 **Stage 5 – Hardening and docs:** Kickoff 2025-10-15. Focus areas: add teardown timeouts for the FD mirror threads, expand README coverage (include ADR 0008 link plus troubleshooting steps for replaced `sys.stdout`), and capture manual/CI verification notes.
+  - ✅ Added FD mirror shutdown timeout with a polling helper to detach stuck reader threads plus unit coverage.
+  - ✅ Documented README guidance (ADR link, mirror flag notes, troubleshooting for replaced `sys.stdout`) and recorded the manual smoke command.
+  - ✅ Verification: `just dev test` (Linux) passes with new mirror timeout tests; Windows CI regression run still queued.
 
 ## Next Steps
 1. Stage 4: finish FD mirror wiring — add mirror-only tests (`os.write` / mixed stdout/stderr), surface user-facing warnings on setup failure, and document the new `mirror` flag in chunk metadata.
-2. Stage 5 hardening: tighten logging guards (`ScopedMuteIoCapture`), expand integration coverage (policy disabled paths, multi-threaded writers), and document configuration knobs for IO capture.
+2. Stage 5 follow-up: monitor the queued Windows CI regression run and flip ADR 0008 to Accepted once cross-platform verification lands.
 3. Evaluate performance impact of the Python stack fallback and gate it behind monitoring snapshots once `sys.monitoring` integration fully drives the snapshot store.