diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 96d4c7e..c672339 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -36,6 +36,73 @@ jobs: run: uv run pytest tests/ -v --ignore=tests/test_browser_bridge.py --timeout=120 timeout-minutes: 10 + # Windows is the primary live-recording platform: this job runs the full + # unit suite AND the live Recorder integration tests (tests/test_performance.py, + # marked 'slow'), which start the real multiprocessing pipeline (pynput + # listeners -> writer processes -> per-capture SQLite + video) and verify + # clean startup, bounded shutdown/memory, and db creation. The three + # listener-dependent tests (roundtrip/reuse/throughput) skip here via + # OPENADAPT_CI_NO_INPUT_INJECTION: hosted runners execute jobs in a + # non-interactive session, so injected input never reaches pynput's hooks; + # run those on an interactive Windows desktop. These tests skip everywhere + # except Windows, so this job is the org's live-recording CI proof. + test-windows: + runs-on: windows-latest + steps: + - uses: actions/checkout@v4 + + - name: Install uv + uses: astral-sh/setup-uv@v4 + with: + version: "latest" + + - name: Set up Python + run: uv python install 3.12 + + - name: Install dependencies + run: uv sync --extra dev + + - name: Run unit tests + run: uv run pytest tests/ -v --ignore=tests/test_browser_bridge.py -m "not slow" --timeout=120 + timeout-minutes: 10 + + - name: Run live recorder integration tests + env: + # Non-interactive session: SendInput never reaches pynput's + # low-level hooks, so the event-capture tests skip (see + # tests/test_performance.py); pipeline start/stop/db/memory + # tests still run the live recorder for real. + OPENADAPT_CI_NO_INPUT_INJECTION: "1" + run: uv run pytest tests/test_performance.py -v -m slow --timeout=300 + timeout-minutes: 20 + + # macOS coverage: runs the unit suite (event processing, storage, video + # encode/decode via PyAV, headless-import invariant, Recorder API surface — + # CGWindowList-backed screen capture works on GitHub macOS runners). The + # 'slow' live-recording tests stay skipped here: pynput input injection + # requires Accessibility permissions a hosted runner cannot grant, and the + # multiprocessing 'spawn' writer path is not yet validated on macOS (see + # tests/test_performance.py). + test-macos: + runs-on: macos-latest + steps: + - uses: actions/checkout@v4 + + - name: Install uv + uses: astral-sh/setup-uv@v4 + with: + version: "latest" + + - name: Set up Python + run: uv python install 3.12 + + - name: Install dependencies + run: uv sync --extra dev + + - name: Run unit tests + run: uv run pytest tests/ -v --ignore=tests/test_browser_bridge.py -m "not slow" --timeout=120 + timeout-minutes: 10 + lint: runs-on: ubuntu-latest steps: diff --git a/openadapt_capture/capture.py b/openadapt_capture/capture.py index 73b6576..1cd00b9 100644 --- a/openadapt_capture/capture.py +++ b/openadapt_capture/capture.py @@ -435,14 +435,24 @@ def load(cls, capture_dir: str | Path) -> "CaptureSession": from openadapt_capture.db.models import Recording session = get_session_for_path(str(db_path)) + + def _discard_session() -> None: + # Close the session AND dispose its engine: the pool otherwise + # keeps the SQLite file handle open, which on Windows leaves + # recording.db locked (WinError 32 when deleting the directory). + bind = session.get_bind() + session.close() + if bind is not None: + bind.dispose() + try: recording = session.query(Recording).first() except Exception: - session.close() + _discard_session() raise if recording is None: - session.close() + _discard_session() raise FileNotFoundError(f"Invalid capture (no recording found): {capture_dir}") return cls(capture_dir, session, recording) @@ -627,9 +637,18 @@ def get_frame_at(self, timestamp: float, tolerance: float = 0.5) -> "Image" | No return None def close(self) -> None: - """Close the capture and release resources.""" + """Close the capture and release resources. + + Disposes the session's engine as well: SQLAlchemy's connection pool + keeps the SQLite file handle open after ``Session.close()``, which on + Windows leaves ``recording.db`` locked — deleting the capture + directory would fail with WinError 32 (sharing violation). + """ if self._session is not None: + bind = self._session.get_bind() self._session.close() + if bind is not None: + bind.dispose() self._session = None def __enter__(self) -> "CaptureSession": diff --git a/openadapt_capture/db/__init__.py b/openadapt_capture/db/__init__.py index 0884a73..96fda3f 100644 --- a/openadapt_capture/db/__init__.py +++ b/openadapt_capture/db/__init__.py @@ -153,9 +153,16 @@ def get_session_for_path(db_path: str, echo: bool = False): """ db_url = f"sqlite:///{db_path}" engine = get_engine(db_url, echo=echo) - # Older recording.db files may predate columns the models now expect; - # add any missing ones so loading them does not fail with 'no such - # column'. Safe/no-op when the schema is already current. - migrate_missing_columns(engine) - Session = get_session_maker(engine) - return Session() + try: + # Older recording.db files may predate columns the models now expect; + # add any missing ones so loading them does not fail with 'no such + # column'. Safe/no-op when the schema is already current. + migrate_missing_columns(engine) + Session = get_session_maker(engine) + return Session() + except Exception: + # A corrupt/unreadable db raises above; dispose the engine so its + # pooled connection does not keep the file handle open (on Windows + # a lingering handle makes the capture directory undeletable). + engine.dispose() + raise diff --git a/openadapt_capture/recorder.py b/openadapt_capture/recorder.py index 284fc99..0e5335a 100644 --- a/openadapt_capture/recorder.py +++ b/openadapt_capture/recorder.py @@ -1,7 +1,18 @@ -"""Script for creating Recordings. - -Copied from legacy OpenAdapt record.py. Only import paths changed + -adaptation for per-capture databases. +"""OpenAdapt's original, hand-built recorder — the live recording engine. + +This is the production recorder carried forward from OpenAdapt's record.py: +the most battle-tested recording code in the org, refined against years of +real desktop workflows. It is the recorder behind OpenAdapt's desktop path +(openadapt-flow ``record --backend windows|rdp`` wraps ``Recorder``). + +Architecture: a multiprocessing pipeline. pynput listeners feed raw +mouse/keyboard events into synchronized queues; dedicated writer processes +persist action events, screenshots, video frames, and (optionally) audio and +window state into a per-capture SQLite database plus time-aligned media files. +Adapted from the original for per-capture databases. Importing this module +must never touch the display (no screenshot at module scope — enforced by +tests/test_headless_import.py); pynput itself may be unavailable headless, in +which case the package ``__init__`` degrades ``Recorder`` to ``None``. Usage: @@ -252,7 +263,14 @@ def process_events( prev_saved_window_timestamp = 0 started = False while not terminate_processing.is_set() or not event_q.empty(): - event = event_q.get() + # Bounded get: a bare event_q.get() deadlocks shutdown when terminate + # is set while the queue is empty and the readers have already exited + # (nobody left to feed an event, so the loop condition is never + # re-checked and join_tasks() hangs forever on this thread). + try: + event = event_q.get(timeout=1) + except queue.Empty: + continue if not started: started_event.set() started = True diff --git a/pyproject.toml b/pyproject.toml index 823d8d5..723df67 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,9 +22,9 @@ classifiers = [ dependencies = [ "pydantic>=2.0.0", - "pynput>=1.7.0", - "av>=10.0.0", - "Pillow>=9.0.0", + "pynput>=1.7.6", + "av>=12.0.0", + "Pillow>=10.1.0", "fire>=0.7.1", "mss>=6.0.0", "sounddevice>=0.5.3", @@ -33,14 +33,14 @@ dependencies = [ "pydantic-settings>=2.12.0", "openai>=2.11.0", "websockets>=12.0", - # Legacy recording dependencies (matching OpenAdapt record.py) + # Recording-pipeline dependencies (carried forward from OpenAdapt record.py) "sqlalchemy>=2.0.0", "alembic>=1.0.0", "loguru>=0.7.0", "psutil>=5.0.0", "pympler>=1.0.0", "tqdm>=4.0.0", - "numpy>=1.20.0", + "numpy>=1.26.0", "oa-atomacos>=3.2.0; sys_platform == 'darwin'", ] diff --git a/tests/test_highlevel.py b/tests/test_highlevel.py index 27641f4..4669110 100644 --- a/tests/test_highlevel.py +++ b/tests/test_highlevel.py @@ -19,11 +19,25 @@ Recorder = None +# Sessions/engines created by _create_test_recording, released by the +# temp_capture_dir teardown BEFORE the TemporaryDirectory is removed. Both +# must be released: crud.insert_recording ends with session.refresh(), which +# leaves the connection checked out until the session closes, and the engine +# pool keeps the SQLite file handle open until dispose(). On Windows an open +# recording.db makes the rmtree fail with WinError 32 (sharing violation). +_HELPER_SESSIONS = [] +_HELPER_ENGINES = [] + + @pytest.fixture def temp_capture_dir(): """Create a temporary directory for captures.""" with tempfile.TemporaryDirectory() as tmpdir: yield tmpdir + while _HELPER_SESSIONS: + _HELPER_SESSIONS.pop().close() + while _HELPER_ENGINES: + _HELPER_ENGINES.pop().dispose() def _create_test_recording(capture_dir, task_description="Test task"): @@ -34,7 +48,9 @@ def _create_test_recording(capture_dir, task_description="Test task"): os.makedirs(capture_dir, exist_ok=True) db_path = os.path.join(capture_dir, "recording.db") engine, Session = create_db(db_path) + _HELPER_ENGINES.append(engine) session = Session() + _HELPER_SESSIONS.append(session) timestamp = time.time() recording_data = { @@ -355,8 +371,10 @@ def test_session_leak_on_no_recording(self, temp_capture_dir): capture_path = str(Path(temp_capture_dir) / "capture") os.makedirs(capture_path, exist_ok=True) db_path = os.path.join(capture_path, "recording.db") - # Create DB with tables but no recording row - create_db(db_path) + # Create DB with tables but no recording row (dispose the setup + # engine so only Capture.load's own handle is under test) + engine, _ = create_db(db_path) + engine.dispose() with pytest.raises(FileNotFoundError, match="no recording found"): Capture.load(capture_path) @@ -513,6 +531,7 @@ def test_pixel_ratio_round_trips_through_model(self, temp_capture_dir): }, ) session.close() + engine.dispose() # The column is real: it is present in the on-disk schema. con = sqlite3.connect(db_path) diff --git a/tests/test_performance.py b/tests/test_performance.py index fb57bac..dfd54ba 100644 --- a/tests/test_performance.py +++ b/tests/test_performance.py @@ -7,12 +7,12 @@ Marked as 'slow' — skip with: pytest -m "not slow" Run only these: pytest -m slow -v -NOTE: The legacy recorder uses multiprocessing.Process for writer tasks. -On macOS (Python "spawn" start method), writer processes may fail to start -because each child re-imports modules and triggers side effects like -take_screenshot(). These tests are designed for Windows (the primary -recording platform) and will skip on macOS/Linux if the recorder -cannot start all processes within a timeout. +NOTE: The recorder uses multiprocessing.Process for writer tasks. On macOS +(Python "spawn" start method) writer processes historically failed to start +because each child re-imported modules with display side effects; imports are +side-effect free since 0.5.4, but the spawn path on macOS/Linux is not yet +validated end to end. These tests target Windows (the primary recording +platform) and are exercised in CI on windows-latest. """ import os @@ -33,14 +33,30 @@ except ImportError: Recorder = None -# Skip on non-Windows platforms where the legacy recorder has known issues +# Skip on non-Windows platforms where the live pipeline is not yet validated _SKIP_REASON = ( - "Legacy recorder uses multiprocessing.Process which requires Windows " - "or fork-safe environment. On macOS/Linux with 'spawn' start method, " - "writer processes may fail to start." + "Live recorder integration tests target Windows (the primary recording " + "platform, exercised in CI on windows-latest). The multiprocessing " + "'spawn' writer path on macOS/Linux is not yet validated end to end; " + "on GitHub macOS runners pynput input injection also needs Accessibility " + "permissions that cannot be granted." ) _ON_WINDOWS = sys.platform == "win32" +# GitHub-hosted Windows runners execute jobs in a non-interactive session: +# SendInput-injected events never reach the low-level hooks pynput uses, so +# listener-dependent tests capture zero events there. The CI workflow sets +# this flag; the live-pipeline tests that do not depend on captured input +# (startup/shutdown, db creation, bounded memory) still run for real. Run the +# full set on an interactive Windows desktop (developer machine or an +# interactive self-hosted runner). +_NO_INPUT_INJECTION = os.environ.get("OPENADAPT_CI_NO_INPUT_INJECTION") == "1" +_INJECTION_SKIP_REASON = ( + "OPENADAPT_CI_NO_INPUT_INJECTION=1: injected input does not reach pynput " + "hooks in a non-interactive session, so event-capture assertions cannot " + "hold (hosted CI runner limitation, not a recorder bug)" +) + # --------------------------------------------------------------------------- # Helpers @@ -122,6 +138,7 @@ def capture_dir(tmp_path): class TestRecorderIntegration: """Integration tests that run the full recording pipeline.""" + @pytest.mark.skipif(_NO_INPUT_INJECTION, reason=_INJECTION_SKIP_REASON) def test_record_and_load_roundtrip(self, capture_dir): """Record synthetic input, stop, reload, and verify events round-trip.""" duration = 3 # seconds @@ -129,9 +146,10 @@ def test_record_and_load_roundtrip(self, capture_dir): input_stop = threading.Event() cycles = [0] - with Recorder(capture_dir, task_description="Integration test"): - # Give recorder a moment to start listeners - time.sleep(1) + with Recorder(capture_dir, task_description="Integration test") as rec: + # Wait until listeners + writers are actually up (cold CI runners + # can take several seconds; a fixed sleep races the startup). + assert rec.wait_for_ready(timeout=120), "recorder failed to start" # Generate synthetic input in background thread def run_input(): @@ -170,6 +188,7 @@ def run_input(): capture.close() + @pytest.mark.skipif(_NO_INPUT_INJECTION, reason=_INJECTION_SKIP_REASON) def test_recorder_reuse(self, tmp_path): """Test that Recorder can be used twice in the same process. @@ -179,8 +198,8 @@ def test_recorder_reuse(self, tmp_path): d = str(tmp_path / f"capture_{i}") input_stop = threading.Event() - with Recorder(d, task_description=f"Reuse test {i}"): - time.sleep(1) + with Recorder(d, task_description=f"Reuse test {i}") as rec: + assert rec.wait_for_ready(timeout=120), "recorder failed to start" def run_input(): _generate_synthetic_input(1, input_stop) @@ -203,8 +222,8 @@ def test_shutdown_time(self, capture_dir): duration = 2 input_stop = threading.Event() - with Recorder(capture_dir, task_description="Shutdown test"): - time.sleep(0.5) + with Recorder(capture_dir, task_description="Shutdown test") as rec: + assert rec.wait_for_ready(timeout=120), "recorder failed to start" def run_input(): _generate_synthetic_input(duration, input_stop) @@ -239,8 +258,8 @@ def test_memory_bounded(self, capture_dir): ) mem_thread.start() - with Recorder(capture_dir, task_description="Memory test"): - time.sleep(0.5) + with Recorder(capture_dir, task_description="Memory test") as rec: + assert rec.wait_for_ready(timeout=120), "recorder failed to start" def run_input(): _generate_synthetic_input(duration, input_stop) @@ -269,8 +288,8 @@ def test_db_file_created(self, capture_dir): """Test that recording.db is created in the capture directory.""" input_stop = threading.Event() - with Recorder(capture_dir, task_description="DB test"): - time.sleep(0.5) + with Recorder(capture_dir, task_description="DB test") as rec: + assert rec.wait_for_ready(timeout=120), "recorder failed to start" def run_input(): _generate_synthetic_input(1, input_stop) @@ -285,14 +304,15 @@ def run_input(): assert db_path.exists(), f"recording.db not found in {capture_dir}" assert db_path.stat().st_size > 0, "recording.db is empty" + @pytest.mark.skipif(_NO_INPUT_INJECTION, reason=_INJECTION_SKIP_REASON) def test_event_throughput(self, capture_dir): """Test that event capture rate is reasonable.""" duration = 3 input_stop = threading.Event() cycles = [0] - with Recorder(capture_dir, task_description="Throughput test"): - time.sleep(0.5) + with Recorder(capture_dir, task_description="Throughput test") as rec: + assert rec.wait_for_ready(timeout=120), "recorder failed to start" def run_input(): cycles[0] = _generate_synthetic_input(duration, input_stop)