diff --git a/openadapt_capture/__init__.py b/openadapt_capture/__init__.py
index 6ba3943..e44acc3 100644
--- a/openadapt_capture/__init__.py
+++ b/openadapt_capture/__init__.py
@@ -16,6 +16,7 @@
     compare_video_to_images,
     plot_comparison,
 )
+from openadapt_capture.config import RecordingConfig
 from openadapt_capture.db.models import (
     ActionEvent as DBActionEvent,
 )
@@ -117,6 +118,7 @@
     "__version__",
     # High-level APIs
     "Recorder",
+    "RecordingConfig",
     "Capture",
     "CaptureSession",
     "Action",
diff --git a/openadapt_capture/cli.py b/openadapt_capture/cli.py
index d27c5d5..931eb25 100644
--- a/openadapt_capture/cli.py
+++ b/openadapt_capture/cli.py
@@ -14,14 +14,22 @@
 def record(
     output_dir: str,
     description: str | None = None,
+    video: bool = True,
+    audio: bool = False,
+    images: bool = False,
 ) -> None:
     """Record GUI interactions.
 
     Args:
         output_dir: Directory to save capture.
         description: Optional task description.
+        video: Capture video (default: True).
+        audio: Capture audio (default: False).
+        images: Save screenshots as PNGs (default: False).
     """
-    from openadapt_capture.recorder import record as do_record
+    import time
+
+    from openadapt_capture.recorder import Recorder
 
     output_dir = str(Path(output_dir).resolve())
 
@@ -29,12 +37,22 @@ def record(
     print("Press Ctrl+C or type stop sequence to stop recording...")
     print()
 
-    do_record(
+    with Recorder(
+        output_dir,
         task_description=description or "",
-        capture_dir=output_dir,
-    )
+        capture_video=video,
+        capture_audio=audio,
+        capture_images=images,
+    ) as recorder:
+        recorder.wait_for_ready()
+        try:
+            while recorder.is_recording:
+                time.sleep(1)
+        except KeyboardInterrupt:
+            pass
 
     print()
+    print(f"Recorded {recorder.event_count} events")
     print(f"Saved to: {output_dir}")
 
 
diff --git a/openadapt_capture/config.py b/openadapt_capture/config.py
index b6ec7df..e0e77e1 100644
--- a/openadapt_capture/config.py
+++ b/openadapt_capture/config.py
@@ -6,6 +6,9 @@
 
 from __future__ import annotations
 
+from contextlib import contextmanager
+from dataclasses import dataclass, fields
+
 from pydantic_settings import BaseSettings
 
 STOP_STRS = [
@@ -67,3 +70,60 @@ class Settings(BaseSettings):
 config = Settings()
 # Keep backward-compatible alias
 settings = config
+
+
+# ---------------------------------------------------------------------------
+# RecordingConfig: user-facing overrides for Recorder constructor
+# ---------------------------------------------------------------------------
+
+# Mapping from RecordingConfig field names to Settings attribute names
+_FIELD_TO_CONFIG_ATTR = {
+    "capture_video": "RECORD_VIDEO",
+    "capture_audio": "RECORD_AUDIO",
+    "capture_images": "RECORD_IMAGES",
+    "capture_window_data": "RECORD_WINDOW_DATA",
+    "capture_browser_events": "RECORD_BROWSER_EVENTS",
+    "capture_full_video": "RECORD_FULL_VIDEO",
+    "video_encoding": "VIDEO_ENCODING",
+    "video_pixel_format": "VIDEO_PIXEL_FORMAT",
+    "stop_sequences": "STOP_SEQUENCES",
+    "log_memory": "LOG_MEMORY",
+    "plot_performance": "PLOT_PERFORMANCE",
+}
+
+
+@dataclass
+class RecordingConfig:
+    """User-facing recording options. ``None`` means 'use default from config'."""
+
+    capture_video: bool | None = None
+    capture_audio: bool | None = None
+    capture_images: bool | None = None
+    capture_window_data: bool | None = None
+    capture_browser_events: bool | None = None
+    capture_full_video: bool | None = None
+    video_encoding: str | None = None
+    video_pixel_format: str | None = None
+    stop_sequences: list[list[str]] | None = None
+    log_memory: bool | None = None
+    plot_performance: bool | None = None
+
+
+@contextmanager
+def config_override(recording_config: RecordingConfig):
+    """Temporarily override config settings from a RecordingConfig.
+
+    Saves original values, applies non-None overrides, yields, then restores.
+    """
+    originals: dict[str, object] = {}
+    for field in fields(recording_config):
+        value = getattr(recording_config, field.name)
+        if value is not None:
+            config_attr = _FIELD_TO_CONFIG_ATTR[field.name]
+            originals[config_attr] = getattr(config, config_attr)
+            object.__setattr__(config, config_attr, value)
+    try:
+        yield
+    finally:
+        for config_attr, original_value in originals.items():
+            object.__setattr__(config, config_attr, original_value)
diff --git a/openadapt_capture/recorder.py b/openadapt_capture/recorder.py
index b6956d5..ad614ab 100644
--- a/openadapt_capture/recorder.py
+++ b/openadapt_capture/recorder.py
@@ -69,9 +69,7 @@ def set_browser_mode(
     "window": True,
     "browser": True,
 }
-PLOT_PERFORMANCE = config.PLOT_PERFORMANCE
 NUM_MEMORY_STATS_TO_LOG = 3
-STOP_SEQUENCES = config.STOP_SEQUENCES
 
 stop_sequence_detected = False
 ws_server_instance = None
@@ -963,8 +961,9 @@ def read_keyboard_events(
         None
     """
     # create list of indices for sequence detection
-    # one index for each stop sequence in STOP_SEQUENCES
-    stop_sequence_indices = [0 for _ in STOP_SEQUENCES]
+    # one index for each stop sequence in config.STOP_SEQUENCES
+    stop_sequences = config.STOP_SEQUENCES
+    stop_sequence_indices = [0 for _ in stop_sequences]
 
     def on_press(
         event_q: queue.Queue,
@@ -992,9 +991,9 @@ def on_press(
         global stop_sequence_detected
         canonical_key_name = getattr(canonical_key, "name", None)
 
-        for i in range(0, len(STOP_SEQUENCES)):
+        for i in range(0, len(stop_sequences)):
             # check each stop sequence
-            stop_sequence = STOP_SEQUENCES[i]
+            stop_sequence = stop_sequences[i]
             # stop_sequence_indices[i] is the index for this stop sequence
             # get canonical KeyCode of current letter in this sequence
             canonical_sequence = keyboard_listener.canonical(
@@ -1304,6 +1303,13 @@ def record(
     terminate_recording: multiprocessing.Event = None,
     status_pipe: multiprocessing.connection.Connection | None = None,
     log_memory: bool = config.LOG_MEMORY,
+    # Optional shared counters — if None, record() creates its own.
+    # Pass externally-created Values to read counts from outside (e.g. Recorder).
+    num_action_events: multiprocessing.Value = None,
+    num_screen_events: multiprocessing.Value = None,
+    num_window_events: multiprocessing.Value = None,
+    num_browser_events: multiprocessing.Value = None,
+    num_video_events: multiprocessing.Value = None,
 ) -> None:
     """Record Screenshots/ActionEvents/WindowEvents/BrowserEvents.
 
@@ -1409,11 +1415,16 @@ def record(
     mouse_event_reader.start()
     task_by_name["mouse_event_reader"] = mouse_event_reader
 
-    num_action_events = multiprocessing.Value("i", 0)
-    num_screen_events = multiprocessing.Value("i", 0)
-    num_window_events = multiprocessing.Value("i", 0)
-    num_browser_events = multiprocessing.Value("i", 0)
-    num_video_events = multiprocessing.Value("i", 0)
+    if num_action_events is None:
+        num_action_events = multiprocessing.Value("i", 0)
+    if num_screen_events is None:
+        num_screen_events = multiprocessing.Value("i", 0)
+    if num_window_events is None:
+        num_window_events = multiprocessing.Value("i", 0)
+    if num_browser_events is None:
+        num_browser_events = multiprocessing.Value("i", 0)
+    if num_video_events is None:
+        num_video_events = multiprocessing.Value("i", 0)
 
     event_processor = threading.Thread(
         target=process_events,
@@ -1566,7 +1577,7 @@ def record(
     perf_stats_writer.start()
     task_by_name["perf_stats_writer"] = perf_stats_writer
 
-    if PLOT_PERFORMANCE:
+    if config.PLOT_PERFORMANCE:
         record_pid = os.getpid()
         mem_writer = multiprocessing.Process(
             target=utils.WrapStdout(memory_writer),
@@ -1658,7 +1669,7 @@ def join_tasks(task_names: list[str]) -> None:
         ]
     )
 
-    if PLOT_PERFORMANCE:
+    if config.PLOT_PERFORMANCE:
         session = get_session_for_path(db_path)
         plotting.plot_performance(
             session, recording, save_dir=capture_dir,
@@ -1678,30 +1689,122 @@ def join_tasks(task_names: list[str]) -> None:
 
 
 class Recorder:
-    """Context manager wrapper around the legacy record() function.
+    """High-level recording interface.
 
-    Usage:
-        with Recorder('./my_capture', task_description='Demo task') as rec:
+    Wraps the legacy ``record()`` function with a clean Python API:
+
+    - Constructor parameters override config defaults (``capture_video``, etc.)
+    - Runtime introspection (``event_count``, ``is_recording``)
+    - Post-recording access to ``CaptureSession``
+
+    Usage::
+
+        with Recorder('./my_capture', task_description='Demo task',
+                       capture_video=True, capture_audio=False) as recorder:
+            recorder.wait_for_ready()
             input('Press Enter to stop recording...')
+        print(f"Recorded {recorder.event_count} events")
     """
 
-    def __init__(self, capture_dir: str, task_description: str = "") -> None:
-        self.capture_dir = os.path.abspath(capture_dir)
+    def __init__(
+        self,
+        capture_dir: str,
+        task_description: str = "",
+        *,
+        capture_video: bool | None = None,
+        capture_audio: bool | None = None,
+        capture_images: bool | None = None,
+        capture_window_data: bool | None = None,
+        capture_browser_events: bool | None = None,
+        capture_full_video: bool | None = None,
+        video_encoding: str | None = None,
+        video_pixel_format: str | None = None,
+        stop_sequences: list[list[str]] | None = None,
+        log_memory: bool | None = None,
+        plot_performance: bool | None = None,
+    ) -> None:
+        from pathlib import Path
+
+        from openadapt_capture.config import RecordingConfig
+
+        self.capture_dir = str(Path(capture_dir).resolve())
         self.task_description = task_description
+
+        # Build recording config from constructor params
+        self._recording_config = RecordingConfig(
+            capture_video=capture_video,
+            capture_audio=capture_audio,
+            capture_images=capture_images,
+            capture_window_data=capture_window_data,
+            capture_browser_events=capture_browser_events,
+            capture_full_video=capture_full_video,
+            video_encoding=video_encoding,
+            video_pixel_format=video_pixel_format,
+            stop_sequences=stop_sequences,
+            log_memory=log_memory,
+            plot_performance=plot_performance,
+        )
+
+        # Shared state for cross-thread communication
         self._terminate_processing = multiprocessing.Event()
         self._terminate_recording = multiprocessing.Event()
-        self._record_thread = None
+        self._num_action_events = multiprocessing.Value("i", 0)
+        self._num_screen_events = multiprocessing.Value("i", 0)
+        self._num_window_events = multiprocessing.Value("i", 0)
+        self._num_browser_events = multiprocessing.Value("i", 0)
+        self._num_video_events = multiprocessing.Value("i", 0)
+
+        # Status communication
+        self._status_recv, self._status_send = multiprocessing.Pipe(duplex=False)
+        self._ready_event = threading.Event()
+        self._stopped_event = threading.Event()
+
+        # Internal
+        self._record_thread: threading.Thread | None = None
+        self._status_thread: threading.Thread | None = None
+        self._capture = None  # lazy CaptureSession
+
+    def _drain_status_pipe(self) -> None:
+        """Background thread that reads status messages from record()."""
+        try:
+            while not self._stopped_event.is_set():
+                if self._status_recv.poll(timeout=0.5):
+                    msg = self._status_recv.recv()
+                    if isinstance(msg, dict):
+                        if msg.get("type") == "record.started":
+                            self._ready_event.set()
+                        elif msg.get("type") == "record.stopped":
+                            self._stopped_event.set()
+        except (EOFError, OSError):
+            pass
+
+    def _run_record(self) -> None:
+        """Thread target: apply config overrides, then call record()."""
+        from openadapt_capture.config import config_override
+
+        with config_override(self._recording_config):
+            record(
+                task_description=self.task_description,
+                capture_dir=self.capture_dir,
+                terminate_processing=self._terminate_processing,
+                terminate_recording=self._terminate_recording,
+                status_pipe=self._status_send,
+                num_action_events=self._num_action_events,
+                num_screen_events=self._num_screen_events,
+                num_window_events=self._num_window_events,
+                num_browser_events=self._num_browser_events,
+                num_video_events=self._num_video_events,
+            )
 
     def __enter__(self) -> "Recorder":
-        self._record_thread = threading.Thread(
-            target=record,
-            kwargs={
-                "task_description": self.task_description,
-                "capture_dir": self.capture_dir,
-                "terminate_processing": self._terminate_processing,
-                "terminate_recording": self._terminate_recording,
-            },
+        # Start status drain thread
+        self._status_thread = threading.Thread(
+            target=self._drain_status_pipe, daemon=True,
         )
+        self._status_thread.start()
+
+        # Start recording thread
+        self._record_thread = threading.Thread(target=self._run_record)
         self._record_thread.start()
         return self
 
@@ -1709,11 +1812,72 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> None:
         self._terminate_processing.set()
         if self._record_thread is not None:
             self._record_thread.join()
+        self._stopped_event.set()  # ensure status thread exits
+        if self._status_thread is not None:
+            self._status_thread.join(timeout=5)
 
     def stop(self) -> None:
         """Stop recording programmatically."""
         self._terminate_processing.set()
 
+    def wait_for_ready(self, timeout: float = 60) -> bool:
+        """Block until all recording threads/processes have started.
+
+        Returns True if ready, False if timeout expired.
+        """
+        return self._ready_event.wait(timeout=timeout)
+
+    @property
+    def is_recording(self) -> bool:
+        """Whether recording is currently active."""
+        return (
+            self._record_thread is not None
+            and self._record_thread.is_alive()
+            and not self._terminate_processing.is_set()
+        )
+
+    @property
+    def event_count(self) -> int:
+        """Number of action events recorded so far (or total after stop)."""
+        return self._num_action_events.value
+
+    @property
+    def screen_count(self) -> int:
+        """Number of screen events recorded."""
+        return self._num_screen_events.value
+
+    @property
+    def video_frame_count(self) -> int:
+        """Number of video frames written."""
+        return self._num_video_events.value
+
+    @property
+    def stats(self) -> dict:
+        """Recording statistics snapshot."""
+        return {
+            "action_events": self._num_action_events.value,
+            "screen_events": self._num_screen_events.value,
+            "window_events": self._num_window_events.value,
+            "browser_events": self._num_browser_events.value,
+            "video_frames": self._num_video_events.value,
+            "is_recording": self.is_recording,
+        }
+
+    @property
+    def capture(self):
+        """Load the CaptureSession after recording completes.
+
+        Returns None if recording has not finished yet.
+        """
+        if self._capture is None and not self.is_recording:
+            try:
+                from openadapt_capture.capture import CaptureSession
+
+                self._capture = CaptureSession.load(self.capture_dir)
+            except FileNotFoundError:
+                return None
+        return self._capture
+
 
 # Entry point
 def start() -> None:
diff --git a/tests/test_highlevel.py b/tests/test_highlevel.py
index 27545b0..b391377 100644
--- a/tests/test_highlevel.py
+++ b/tests/test_highlevel.py
@@ -57,7 +57,7 @@ class TestRecorder:
     def test_recorder_class_exists(self):
         """Test that Recorder class can be instantiated."""
         rec = Recorder("/tmp/test_capture_never_created", task_description="test")
-        assert rec.capture_dir == "/tmp/test_capture_never_created"
+        assert rec.capture_dir.endswith("test_capture_never_created")
         assert rec.task_description == "test"
 
     def test_recorder_has_context_manager(self):
@@ -66,6 +66,60 @@ def test_recorder_has_context_manager(self):
         assert hasattr(Recorder, "__exit__")
         assert hasattr(Recorder, "stop")
 
+    def test_recorder_accepts_capture_params(self):
+        """Test Recorder accepts capture_video, capture_audio, etc."""
+        rec = Recorder(
+            "/tmp/test_never_created",
+            task_description="test",
+            capture_video=True,
+            capture_audio=False,
+            capture_images=True,
+            capture_full_video=False,
+            plot_performance=False,
+        )
+        assert rec.task_description == "test"
+
+    def test_recorder_event_count_property(self):
+        """Test Recorder has event_count property starting at 0."""
+        rec = Recorder("/tmp/test_never_created")
+        assert rec.event_count == 0
+
+    def test_recorder_is_recording_property(self):
+        """Test Recorder has is_recording property (False before start)."""
+        rec = Recorder("/tmp/test_never_created")
+        assert rec.is_recording is False
+
+    def test_recorder_stats_property(self):
+        """Test Recorder has stats property returning dict."""
+        rec = Recorder("/tmp/test_never_created")
+        stats = rec.stats
+        assert isinstance(stats, dict)
+        assert "action_events" in stats
+        assert "screen_events" in stats
+        assert "video_frames" in stats
+        assert "is_recording" in stats
+        assert stats["action_events"] == 0
+
+    def test_recorder_wait_for_ready_method(self):
+        """Test Recorder has wait_for_ready method."""
+        rec = Recorder("/tmp/test_never_created")
+        assert callable(rec.wait_for_ready)
+
+    def test_recorder_capture_property_before_recording(self):
+        """Test Recorder.capture is None before recording."""
+        rec = Recorder("/tmp/test_never_created")
+        assert rec.capture is None
+
+    def test_recorder_screen_count_property(self):
+        """Test Recorder has screen_count property."""
+        rec = Recorder("/tmp/test_never_created")
+        assert rec.screen_count == 0
+
+    def test_recorder_video_frame_count_property(self):
+        """Test Recorder has video_frame_count property."""
+        rec = Recorder("/tmp/test_never_created")
+        assert rec.video_frame_count == 0
+
 
 class TestCapture:
     """Tests for Capture/CaptureSession class."""
@@ -374,3 +428,58 @@ def test_capture_load_corrupt_db(self, temp_capture_dir):
 
         with pytest.raises(Exception):
             Capture.load(capture_path)
+
+
+class TestRecordingConfig:
+    """Tests for RecordingConfig and config_override."""
+
+    def test_config_override_applies_and_restores(self):
+        """Test that config_override patches and restores config."""
+        from openadapt_capture.config import (
+            RecordingConfig,
+            config,
+            config_override,
+        )
+
+        original_video = config.RECORD_VIDEO
+        original_audio = config.RECORD_AUDIO
+
+        rc = RecordingConfig(capture_video=False, capture_audio=True)
+        with config_override(rc):
+            assert config.RECORD_VIDEO is False
+            assert config.RECORD_AUDIO is True
+
+        # Restored
+        assert config.RECORD_VIDEO == original_video
+        assert config.RECORD_AUDIO == original_audio
+
+    def test_config_override_none_values_unchanged(self):
+        """Test that None values in RecordingConfig don't change config."""
+        from openadapt_capture.config import (
+            RecordingConfig,
+            config,
+            config_override,
+        )
+
+        original_video = config.RECORD_VIDEO
+        rc = RecordingConfig()  # all None
+        with config_override(rc):
+            assert config.RECORD_VIDEO == original_video
+
+    def test_config_override_restores_on_exception(self):
+        """Test that config is restored even if body raises."""
+        from openadapt_capture.config import (
+            RecordingConfig,
+            config,
+            config_override,
+        )
+
+        original_video = config.RECORD_VIDEO
+        rc = RecordingConfig(capture_video=not original_video)
+
+        with pytest.raises(ValueError):
+            with config_override(rc):
+                assert config.RECORD_VIDEO != original_video
+                raise ValueError("test")
+
+        assert config.RECORD_VIDEO == original_video