feat: add Python tracing API with Rust placeholders

Author: tzanko-matev

.agents/tasks/2025/08/14-1027-initial-python-api

@@ -1 +1,3 @@
-Implement the Python API described in the design document for the Rust-based module. Write tests. Don't actually implement tracing using `runtime_tracing` yet, just add placeholders.
+Implement the Python API described in the design document for the Rust-based module. Write tests. Don't actually implement tracing using `runtime_tracing` yet, just add placeholders.
+--- FOLLOW UP TASK ---
+Implement the Python API described in the design document for the Rust-based module. Write tests. Don't actually implement tracing using runtime_tracing yet, just add placeholders.

codetracer/__init__.py

@@ -0,0 +1,155 @@
+"""High-level tracing API built on a Rust backend.
+
+This module exposes a minimal interface for starting and stopping
+runtime traces. The heavy lifting is delegated to the
+`codetracer_python_recorder` Rust extension which will eventually hook
+into `runtime_tracing` and `sys.monitoring`.  For now the Rust side only
+maintains placeholder state and performs no actual tracing.
+"""
+from __future__ import annotations
+
+import contextlib
+import os
+from pathlib import Path
+from typing import Iterable, Iterator, Optional
+
+from codetracer_python_recorder import (
+    flush_tracing as _flush_backend,
+    is_tracing as _is_tracing_backend,
+    start_tracing as _start_backend,
+    stop_tracing as _stop_backend,
+)
+
+TRACE_BINARY: str = "binary"
+TRACE_JSON: str = "json"
+DEFAULT_FORMAT: str = TRACE_BINARY
+
+_active_session: Optional["TraceSession"] = None
+
+
+def _normalize_source_roots(source_roots: Iterable[os.PathLike | str] | None) -> Optional[list[str]]:
+    if source_roots is None:
+        return None
+    return [str(Path(p)) for p in source_roots]
+
+
+def start(
+    path: os.PathLike | str,
+    *,
+    format: str = DEFAULT_FORMAT,
+    capture_values: bool = True,
+    source_roots: Iterable[os.PathLike | str] | None = None,
+) -> "TraceSession":
+    """Start a global trace session.
+
+    Parameters mirror the design document.  The current implementation
+    merely records the active state on the Rust side and performs no
+    tracing.
+    """
+    global _active_session
+    if _is_tracing_backend():
+        raise RuntimeError("tracing already active")
+
+    trace_path = Path(path)
+    _start_backend(str(trace_path), format, capture_values, _normalize_source_roots(source_roots))
+    session = TraceSession(path=trace_path, format=format)
+    _active_session = session
+    return session
+
+
+def stop() -> None:
+    """Stop the active trace session if one is running."""
+    global _active_session
+    if not _is_tracing_backend():
+        return
+    _stop_backend()
+    _active_session = None
+
+
+def is_tracing() -> bool:
+    """Return ``True`` when a trace session is active."""
+    return _is_tracing_backend()
+
+
+def flush() -> None:
+    """Flush buffered trace data.
+
+    With the current placeholder implementation this is a no-op but the
+    function is provided to match the planned public API.
+    """
+    if _is_tracing_backend():
+        _flush_backend()
+
+
+@contextlib.contextmanager
+def trace(
+    path: os.PathLike | str,
+    *,
+    format: str = DEFAULT_FORMAT,
+    capture_values: bool = True,
+    source_roots: Iterable[os.PathLike | str] | None = None,
+) -> Iterator["TraceSession"]:
+    """Context manager helper for scoped tracing."""
+    session = start(
+        path,
+        format=format,
+        capture_values=capture_values,
+        source_roots=source_roots,
+    )
+    try:
+        yield session
+    finally:
+        session.stop()
+
+
+class TraceSession:
+    """Handle representing a live tracing session."""
+
+    path: Path
+    format: str
+
+    def __init__(self, path: Path, format: str) -> None:
+        self.path = path
+        self.format = format
+
+    def stop(self) -> None:
+        """Stop this trace session."""
+        if _active_session is self:
+            stop()
+
+    def flush(self) -> None:
+        """Flush buffered trace data for this session."""
+        flush()
+
+    def __enter__(self) -> "TraceSession":
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:  # pragma: no cover - thin wrapper
+        self.stop()
+
+
+def _auto_start_from_env() -> None:
+    path = os.getenv("CODETRACER_TRACE")
+    if not path:
+        return
+    fmt = os.getenv("CODETRACER_FORMAT", DEFAULT_FORMAT)
+    capture_env = os.getenv("CODETRACER_CAPTURE_VALUES")
+    capture = True
+    if capture_env is not None:
+        capture = capture_env.lower() not in {"0", "false", "no"}
+    start(path, format=fmt, capture_values=capture)
+
+
+_auto_start_from_env()
+
+__all__ = [
+    "TraceSession",
+    "DEFAULT_FORMAT",
+    "TRACE_BINARY",
+    "TRACE_JSON",
+    "start",
+    "stop",
+    "is_tracing",
+    "trace",
+    "flush",
+]

crates/codetracer-python-recorder/src/lib.rs

@@ -1,17 +1,58 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use pyo3::exceptions::PyRuntimeError;
 use pyo3::prelude::*;
 
-/// codetracer_python_recorder
-///
-/// Minimal placeholder for the Rust-backed recorder. This exposes a trivial
-/// function to verify the module builds and imports successfully.
+/// Global flag tracking whether tracing is active.
+static ACTIVE: AtomicBool = AtomicBool::new(false);
+
+/// Start tracing. Placeholder implementation that simply flips the
+/// global active flag and ignores all parameters.
+#[pyfunction]
+fn start_tracing(
+    _path: &str,
+    _format: &str,
+    _capture_values: bool,
+    _source_roots: Option<Vec<String>>,
+) -> PyResult<()> {
+    if ACTIVE.swap(true, Ordering::SeqCst) {
+        return Err(PyRuntimeError::new_err("tracing already active"));
+    }
+    Ok(())
+}
+
+/// Stop tracing by resetting the global flag.
+#[pyfunction]
+fn stop_tracing() -> PyResult<()> {
+    ACTIVE.store(false, Ordering::SeqCst);
+    Ok(())
+}
+
+/// Query whether tracing is currently active.
+#[pyfunction]
+fn is_tracing() -> PyResult<bool> {
+    Ok(ACTIVE.load(Ordering::SeqCst))
+}
+
+/// Flush buffered trace data. No-op placeholder for now.
+#[pyfunction]
+fn flush_tracing() -> PyResult<()> {
+    Ok(())
+}
+
+/// Trivial function kept for smoke tests verifying the module builds.
 #[pyfunction]
 fn hello() -> PyResult<String> {
     Ok("Hello from codetracer-python-recorder (Rust)".to_string())
 }
 
+/// Python module definition.
 #[pymodule]
 fn codetracer_python_recorder(_py: Python<'_>, m: Bound<'_, PyModule>) -> PyResult<()> {
-    let hello_fn = wrap_pyfunction!(hello, &m)?;
-    m.add_function(hello_fn)?;
+    m.add_function(wrap_pyfunction!(start_tracing, &m)?)?;
+    m.add_function(wrap_pyfunction!(stop_tracing, &m)?)?;
+    m.add_function(wrap_pyfunction!(is_tracing, &m)?)?;
+    m.add_function(wrap_pyfunction!(flush_tracing, &m)?)?;
+    m.add_function(wrap_pyfunction!(hello, &m)?)?;
     Ok(())
 }

tests/test_codetracer_api.py

@@ -0,0 +1,46 @@
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+import codetracer
+
+
+class TracingApiTests(unittest.TestCase):
+    def setUp(self) -> None:  # ensure clean state before each test
+        codetracer.stop()
+
+    def test_start_stop_and_status(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            trace_path = Path(tmpdir) / "trace.bin"
+            session = codetracer.start(trace_path)
+            self.assertTrue(codetracer.is_tracing())
+            self.assertIsInstance(session, codetracer.TraceSession)
+            self.assertEqual(session.path, trace_path)
+            self.assertEqual(session.format, codetracer.DEFAULT_FORMAT)
+            codetracer.flush()  # should not raise
+            session.flush()  # same
+            session.stop()
+            self.assertFalse(codetracer.is_tracing())
+
+    def test_context_manager(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            trace_path = Path(tmpdir) / "trace.bin"
+            with codetracer.trace(trace_path) as session:
+                self.assertTrue(codetracer.is_tracing())
+                self.assertIsInstance(session, codetracer.TraceSession)
+            self.assertFalse(codetracer.is_tracing())
+
+    def test_environment_auto_start(self) -> None:
+        script = "import codetracer, sys; sys.stdout.write(str(codetracer.is_tracing()))"
+        with tempfile.TemporaryDirectory() as tmpdir:
+            env = os.environ.copy()
+            env["CODETRACER_TRACE"] = str(Path(tmpdir) / "trace.bin")
+            out = subprocess.check_output([sys.executable, "-c", script], env=env)
+            self.assertEqual(out.decode(), "True")
+
+
+if __name__ == "__main__":
+    unittest.main()