diff --git a/openadapt_capture/recorder.py b/openadapt_capture/recorder.py index 2bd32f1..6e86e58 100644 --- a/openadapt_capture/recorder.py +++ b/openadapt_capture/recorder.py @@ -522,6 +522,7 @@ def __bool__(self): } NUM_MEMORY_STATS_TO_LOG = 3 STARTUP_WAIT_POLL_SECONDS = 0.1 +STARTUP_READY_TIMEOUT_SECONDS = 30.0 PRE_READY_TASK_JOIN_TIMEOUT_SECONDS = 2.0 TERMINAL_FRAME_SEAL_TIMEOUT_SECONDS = 10.0 @@ -547,6 +548,9 @@ def _wait_for_tasks_started( task_by_name: dict[str, Any], task_started_events: dict[str, Any], terminate_processing: Any, + task_errors: queue.Queue | None = None, + *, + timeout: float = STARTUP_READY_TIMEOUT_SECONDS, ) -> bool: """Wait for pipeline readiness while honoring shutdown and worker failure. @@ -555,6 +559,7 @@ def _wait_for_tasks_started( startup and signals the rest of the pipeline to stop. """ expected_starts = len(task_by_name) + deadline = time.monotonic() + timeout logger.info(f"{expected_starts=}") while True: @@ -566,6 +571,19 @@ def _wait_for_tasks_started( if not waiting_for: return True + remaining = deadline - time.monotonic() + if remaining <= 0: + detail = ", ".join(sorted(waiting_for)) + error = TimeoutError( + "recording startup did not retain its initial frame and start all " + f"required tasks within {timeout:.1f}s; unresolved readiness: {detail}" + ) + logger.error(str(error)) + if task_errors is not None: + task_errors.put(("startup_readiness", error)) + terminate_processing.set() + return False + stopped_before_ready = [ name for name in waiting_for @@ -578,7 +596,7 @@ def _wait_for_tasks_started( logger.info(f"Waiting for tasks to start: {waiting_for}") logger.info(f"Started tasks: {expected_starts - len(waiting_for)}/{expected_starts}") - terminate_processing.wait(STARTUP_WAIT_POLL_SECONDS) + terminate_processing.wait(min(STARTUP_WAIT_POLL_SECONDS, remaining)) def _join_tasks( @@ -714,6 +732,7 @@ def process_events( num_browser_events: multiprocessing.Value, num_video_events: multiprocessing.Value, producers_finished: threading.Event | None = None, + processing_aborted: threading.Event | None = None, ) -> None: """Process events from the event queue and write them to write queues. @@ -736,6 +755,8 @@ def process_events( producers_finished: Set after every event-journal producer has exited. When supplied, the processor drains the journal to empty after that boundary instead of racing the shared stop signal. + processing_aborted: Stop without publishing a completed journal after a + startup failure leaves a producer alive. """ utils.set_start_time(recording.timestamp) @@ -752,6 +773,8 @@ def process_events( started = False def processing_complete() -> bool: + if processing_aborted is not None and processing_aborted.is_set(): + return True if producers_finished is not None: return producers_finished.is_set() and event_q.empty() return terminate_processing.is_set() and event_q.empty() @@ -803,6 +826,29 @@ def bind_pending_actions( write_bound_action(action_event) pending_action_events.clear() + def retain_screen_frame(screen_event: Event) -> None: + """Queue one exact screen frame for durable pixel retention.""" + nonlocal prev_saved_screen_timestamp, prev_saved_screen_ordinal + process_event( + screen_event if config.RECORD_IMAGES else screen_event._replace(data=None), + screen_write_q, + write_screen_event, + recording, + perf_q, + ) + num_screen_events.value += 1 + prev_saved_screen_timestamp = screen_event.timestamp + prev_saved_screen_ordinal = screen_event.source_ordinal or 0 + if config.RECORD_VIDEO and not config.RECORD_FULL_VIDEO: + process_event( + screen_event._replace(type="screen/video"), + video_write_q, + write_video_event, + recording, + perf_q, + ) + num_video_events.value += 1 + while not processing_complete(): # Bounded get: a bare event_q.get() deadlocks shutdown when terminate # is set while the queue is empty and the readers have already exited @@ -873,17 +919,8 @@ def bind_pending_actions( perf_q, ) num_video_events.value += 1 + retain_screen_frame(event) if scoped_pair: - process_event( - event if config.RECORD_IMAGES else event._replace(data=None), - screen_write_q, - write_screen_event, - recording, - perf_q, - ) - num_screen_events.value += 1 - prev_saved_screen_timestamp = event.timestamp - prev_saved_screen_ordinal = event.source_ordinal or 0 assert prev_window_event is not None process_event( prev_window_event, @@ -895,17 +932,18 @@ def bind_pending_actions( num_window_events.value += 1 prev_saved_window_timestamp = prev_window_event.timestamp prev_saved_window_ordinal = prev_window_event.source_ordinal or 0 - if config.RECORD_VIDEO and not config.RECORD_FULL_VIDEO: - process_event( - event._replace(type="screen/video"), - video_write_q, - write_video_event, - recording, - perf_q, - ) - num_video_events.value += 1 elif event.type == "window": prev_window_event = event + process_event( + event, + window_write_q, + write_window_event, + recording, + perf_q, + ) + num_window_events.value += 1 + prev_saved_window_timestamp = event.timestamp + prev_saved_window_ordinal = event.source_ordinal or 0 elif event.type == "browser": if config.RECORD_BROWSER_EVENTS: process_event( @@ -918,16 +956,16 @@ def bind_pending_actions( num_browser_events.value += 1 elif event.type == "action": if prev_screen_event is None: - logger.warning("Discarding action that came before screen") - continue + raise WindowCaptureError("a native action arrived before its initial frame") else: event.data["screenshot_timestamp"] = prev_screen_event.timestamp event.data["screenshot_source_ordinal"] = prev_screen_event.source_ordinal if prev_window_event is None: if config.RECORD_WINDOW_DATA: - logger.warning("Discarding action that came before window") - continue + raise WindowCaptureError( + "a native action arrived before its configured window evidence" + ) # Window capture disabled — skip window timestamp requirement else: event.data["window_event_timestamp"] = prev_window_event.timestamp @@ -958,30 +996,7 @@ def bind_pending_actions( else prev_saved_screen_timestamp < prev_screen_event.timestamp ) if screen_is_new: - process_event( - ( - prev_screen_event - if config.RECORD_IMAGES - else prev_screen_event._replace(data=None) - ), - screen_write_q, - write_screen_event, - recording, - perf_q, - ) - num_screen_events.value += 1 - prev_saved_screen_timestamp = prev_screen_event.timestamp - prev_saved_screen_ordinal = prev_screen_event.source_ordinal or 0 - if config.RECORD_VIDEO and not config.RECORD_FULL_VIDEO: - prev_video_event = prev_screen_event._replace(type="screen/video") - process_event( - prev_video_event, - video_write_q, - write_video_event, - recording, - perf_q, - ) - num_video_events.value += 1 + retain_screen_frame(prev_screen_event) if prev_window_event is not None: window_is_new = ( prev_window_event.source_ordinal > prev_saved_window_ordinal @@ -1137,6 +1152,8 @@ def write_events( started_event: multiprocessing.Event, pre_callback: Callable[[float], dict] | None = None, post_callback: Callable[[dict], None] | None = None, + *, + ready_after_first_event: bool = False, ) -> None: """Write events of a specific type to the db using the provided write function. @@ -1154,6 +1171,8 @@ def write_events( timestamp as only argument, returns a state dict. post_callback: Optional function to call after main loop. Takes state dict as only argument, returns None. + ready_after_first_event: Delay the readiness signal until the first event + has been committed by ``write_fn``. """ utils.set_start_time(recording.timestamp) @@ -1184,7 +1203,7 @@ def write_events( # been processed for _ in range(num_processed): progress.update() - if not started: + if not started and not ready_after_first_event: started_event.set() started = True try: @@ -1194,6 +1213,9 @@ def write_events( assert event.type == event_type, (event_type, event) state = write_fn(session, recording, event, perf_q, **(state or {})) num_processed += 1 + if not started: + started_event.set() + started = True with num_events.get_lock(): if progress is not None: if progress.total < num_events.value: @@ -1635,6 +1657,7 @@ def read_screen_events( input_finished: threading.Event | None = None, input_frame_boundary: NativeInputFrameBoundary | None = None, terminal_frame_finished: threading.Event | None = None, + terminal_frame_cancelled: threading.Event | None = None, ) -> None: """Read screen events and add them to the event queue. @@ -1663,6 +1686,8 @@ def read_screen_events( input_frame_boundary: Active observer bridge for input-stable frames. terminal_frame_finished: Signals that the exact terminal frame sealed native input and entered the ordered journal. + terminal_frame_cancelled: Cancels terminal-frame coordination after a + startup failure that cannot produce a completed capture. """ if window_scope is not None and desktop_scope is not None: raise ValueError("screen reader cannot use both window and desktop scopes") @@ -1800,8 +1825,13 @@ def capture_one( t_end = time.perf_counter() _screen_timing.append((t_screenshot - t_start, t_end - t_start)) - if (window_scope is not None or desktop_scope is not None) and ( - terminal_frame_finished is not None + terminal_cancelled = ( + terminal_frame_cancelled is not None and terminal_frame_cancelled.is_set() + ) + if ( + (window_scope is not None or desktop_scope is not None) + and terminal_frame_finished is not None + and not terminal_cancelled ): timing = capture_one(seal_input=True) if timing is None: @@ -2064,6 +2094,7 @@ def read_input_events( finished_event: threading.Event | None = None, input_frame_boundary: NativeInputFrameBoundary | None = None, terminal_frame_finished: threading.Event | None = None, + terminal_frame_cancelled: threading.Event | None = None, ) -> None: """Read globally ordered keyboard and mouse events from one native observer.""" stop_sequences = [sequence for sequence in config.STOP_SEQUENCES if sequence] @@ -2202,12 +2233,30 @@ def deliver_observed(event: ObservedInput, reservation: object) -> None: finally: if started and observer is not None: terminal_error = None - if terminal_frame_finished is not None and not observer_failed: + terminal_cancelled = ( + terminal_frame_cancelled is not None + and terminal_frame_cancelled.is_set() + ) + if ( + terminal_frame_finished is not None + and not observer_failed + and not terminal_cancelled + ): terminal_timeout = max( 10.0, float(getattr(observer, "shutdown_timeout", 5.0)) * 2, ) - if not terminal_frame_finished.wait(timeout=terminal_timeout): + terminal_deadline = time.monotonic() + terminal_timeout + while not terminal_frame_finished.wait(timeout=0.1): + if ( + terminal_frame_cancelled is not None + and terminal_frame_cancelled.is_set() + ): + terminal_cancelled = True + break + if time.monotonic() >= terminal_deadline: + break + if not terminal_frame_finished.is_set() and not terminal_cancelled: terminal_error = InputObserverError( "the terminal frame did not seal before native input shutdown" ) @@ -2514,7 +2563,9 @@ def record( producers_finished = threading.Event() input_finished = threading.Event() terminal_frame_finished = threading.Event() + terminal_frame_cancelled = threading.Event() input_frame_boundary = NativeInputFrameBoundary() + processing_aborted = threading.Event() if window_scope is not None: # The preflight frame sizes the fixed stream. Capture again after the # recording clock starts, then publish pixels and geometry atomically @@ -2607,6 +2658,7 @@ def record( input_finished, input_frame_boundary, terminal_frame_finished, + terminal_frame_cancelled, ), terminate_processing, task_errors, @@ -2625,6 +2677,7 @@ def record( input_finished, input_frame_boundary, terminal_frame_finished, + terminal_frame_cancelled, ) input_event_reader = threading.Thread( target=_run_task_fail_loud, @@ -2668,6 +2721,7 @@ def record( num_browser_events, num_video_events, producers_finished, + processing_aborted, ) event_processor = threading.Thread( target=_run_task_fail_loud, @@ -2684,7 +2738,9 @@ def record( task_by_name["event_processor"] = event_processor screen_event_writer = multiprocessing.Process( - target=utils.WrapStdout(write_events), + target=utils.WrapStdout( + partial(write_events, ready_after_first_event=True) + ), args=( "screen", partial(write_screen_event, record_images=bool(config.RECORD_IMAGES)), @@ -2719,7 +2775,9 @@ def record( if config.RECORD_WINDOW_DATA or window_scope is not None: window_event_writer = multiprocessing.Process( - target=utils.WrapStdout(write_events), + target=utils.WrapStdout( + partial(write_events, ready_after_first_event=True) + ), args=( "window", write_window_event, @@ -2737,7 +2795,9 @@ def record( if config.RECORD_VIDEO: video_writer = multiprocessing.Process( - target=utils.WrapStdout(write_events), + target=utils.WrapStdout( + partial(write_events, ready_after_first_event=True) + ), args=( "screen/video", write_video_event, @@ -2821,6 +2881,7 @@ def record( task_by_name, task_started_events, terminate_processing, + task_errors, ) if startup_ready: for _ in range(5): @@ -2837,6 +2898,12 @@ def record( terminate_processing.set() else: logger.info("Tearing down recording after incomplete startup") + terminal_frame_cancelled.set() + input_frame_boundary.fail( + WindowCaptureError( + "recording startup ended before the native frame boundary was ready" + ) + ) terminate_processing.set() if status_pipe: @@ -2847,7 +2914,7 @@ def record( log_memory_usage(_tracker, performance_snapshots) pre_ready_timeout = None if startup_ready else PRE_READY_TASK_JOIN_TIMEOUT_SECONDS - _join_tasks( + lingering_tasks = _join_tasks( task_by_name, [ "window_event_reader", @@ -2858,9 +2925,23 @@ def record( timeout=pre_ready_timeout, ) - # The processor can now drain every completed reservation. No producer can - # append a later event after it observes an empty journal. - producers_finished.set() + journal_producers = { + "window_event_reader", + "input_event_reader", + "screen_event_reader", + } + lingering_producers = sorted(journal_producers.intersection(lingering_tasks)) + producer_shutdown_error = None + if lingering_producers: + producer_shutdown_error = RuntimeError( + "recording startup failed with live journal producers: " + + ", ".join(lingering_producers) + ) + processing_aborted.set() + else: + # The processor can now drain every completed reservation. No producer + # can append a later event after it observes an empty journal. + producers_finished.set() _join_tasks( task_by_name, ["event_processor"], @@ -2894,7 +2975,11 @@ def record( if not task_errors.empty(): task_name, task_error = task_errors.get_nowait() add_exception_note(task_error, f"recording task {task_name!r} failed") + if producer_shutdown_error is not None: + add_exception_note(task_error, str(producer_shutdown_error)) raise task_error + if producer_shutdown_error is not None: + raise producer_shutdown_error _raise_for_failed_processes(task_by_name) if window_scope is not None: window_scope.assert_current() diff --git a/tests/test_capture_terminal.py b/tests/test_capture_terminal.py index 24a945a..331d3ae 100644 --- a/tests/test_capture_terminal.py +++ b/tests/test_capture_terminal.py @@ -221,6 +221,7 @@ def _desktop_capture_directory( root: Path, *, frame_ordinals: tuple[int, ...] = (1, 3), + window_ordinals: tuple[int, ...] = (), action_ordinal: int | None = 2, before_binding_ordinal: int | None = None, after_binding_ordinal: int | None = None, @@ -280,6 +281,22 @@ def _desktop_capture_directory( ) session.add(screenshot) frames[ordinal] = screenshot + for ordinal in window_ordinals: + session.add( + WindowEvent( + recording_id=recording.id, + recording_timestamp=10.0, + timestamp=10.0 + ordinal, + source_ordinal=ordinal, + title="Fixture Window", + left=0, + top=0, + width=80, + height=60, + window_id="fixture-window", + state={}, + ) + ) if action_ordinal is not None: before_ordinal = before_binding_ordinal or frame_ordinals[0] before = frames[before_ordinal] @@ -318,11 +335,14 @@ def _desktop_capture_directory( event_counts={ "action": int(action_ordinal is not None), "screen": len(frame_ordinals), - "window": 0, + "window": len(window_ordinals), "browser": 0, "video": 0, }, - last_source_ordinal=max((*frame_ordinals, action_ordinal or 0)) or None, + last_source_ordinal=max( + (*frame_ordinals, *window_ordinals, action_ordinal or 0) + ) + or None, ) return capture_dir @@ -621,6 +641,39 @@ def test_verified_loader_requires_a_desktop_after_frame_for_every_action( CaptureSession.load_verified(capture_dir) +def test_verified_loader_accepts_a_complete_desktop_source_journal( + tmp_path, +) -> None: + capture_dir = _desktop_capture_directory( + tmp_path, + frame_ordinals=(1, 2, 4), + action_ordinal=3, + before_binding_ordinal=2, + ) + + with CaptureSession.load_verified(capture_dir) as capture: + assert [frame.source_ordinal for frame in capture.frames()] == [1, 2, 4] + action = capture.raw_events()[0] + assert action.source_ordinal == 3 + assert action.screenshot_source_ordinal == 2 + assert action.after_screenshot_source_ordinal == 4 + + +def test_verified_loader_accepts_desktop_window_evidence_without_an_action( + tmp_path, +) -> None: + capture_dir = _desktop_capture_directory( + tmp_path, + frame_ordinals=(1, 3), + window_ordinals=(2,), + action_ordinal=None, + ) + + with CaptureSession.load_verified(capture_dir) as capture: + assert [frame.source_ordinal for frame in capture.frames()] == [1, 3] + assert [event.source_ordinal for event in capture.window_events()] == [2] + + def test_verified_loader_rejects_a_skipped_desktop_nearest_before_frame( tmp_path, ) -> None: diff --git a/tests/test_desktop_capture.py b/tests/test_desktop_capture.py index c8b51d8..9f93988 100644 --- a/tests/test_desktop_capture.py +++ b/tests/test_desktop_capture.py @@ -214,6 +214,42 @@ def take_screenshot() -> Image.Image: journal.get_nowait() +def test_startup_failure_wakes_a_screen_reader_waiting_for_observer_attach() -> None: + boundary = NativeInputFrameBoundary() + terminate = threading.Event() + terminal_cancelled = threading.Event() + journal = OrderedEventJournal() + errors: list[BaseException] = [] + + def run_reader() -> None: + try: + read_screen_events( + journal, + terminate, + SimpleNamespace(timestamp=time.time()), + threading.Event(), + desktop_scope=_two_monitor_scope(), + input_frame_boundary=boundary, + terminal_frame_cancelled=terminal_cancelled, + ) + except BaseException as exc: + errors.append(exc) + + reader = threading.Thread(target=run_reader) + reader.start() + time.sleep(0.01) + + failure = RuntimeError("observer setup timed out") + terminal_cancelled.set() + boundary.fail(failure) + terminate.set() + reader.join(timeout=1) + + assert not reader.is_alive() + assert errors == [failure] + assert journal.empty() + + def test_desktop_terminal_frame_seals_input_before_journal_commit( monkeypatch, ) -> None: diff --git a/tests/test_input_observer.py b/tests/test_input_observer.py index ecba1eb..6732aa6 100644 --- a/tests/test_input_observer.py +++ b/tests/test_input_observer.py @@ -764,6 +764,60 @@ def create(callback, **kwargs): ] +def test_input_reader_skips_terminal_wait_after_startup_cancellation( + monkeypatch, +) -> None: + terminate = threading.Event() + terminate.set() + terminal_waiting = threading.Event() + + class TrackingEvent(threading.Event): + def wait(self, timeout=None): + terminal_waiting.set() + return super().wait(timeout) + + terminal_finished = TrackingEvent() + terminal_cancelled = threading.Event() + boundary = recorder_module.NativeInputFrameBoundary() + stopped = threading.Event() + + class FakeObserver: + def start(self) -> None: + return + + def stop(self) -> None: + stopped.set() + + monkeypatch.setattr( + recorder_module, + "create_input_observer", + lambda *_args, **_kwargs: FakeObserver(), + ) + + reader = threading.Thread( + target=recorder_module.read_input_events, + args=( + queue.Queue(), + terminate, + SimpleNamespace(timestamp=100.0), + threading.Event(), + ), + kwargs={ + "input_frame_boundary": boundary, + "terminal_frame_finished": terminal_finished, + "terminal_frame_cancelled": terminal_cancelled, + }, + ) + reader.start() + assert terminal_waiting.wait(timeout=1) + terminal_cancelled.set() + reader.join(timeout=1) + + assert not reader.is_alive() + assert stopped.is_set() + assert not terminal_finished.is_set() + + @pytest.mark.parametrize( ("detail", "pressed", "expected"), [ diff --git a/tests/test_performance.py b/tests/test_performance.py index bcf4ba0..06e1867 100644 --- a/tests/test_performance.py +++ b/tests/test_performance.py @@ -140,6 +140,15 @@ def capture_dir(tmp_path): class TestRecorderIntegration: """Integration tests that run the full recording pipeline.""" + def test_initial_frame_ready_without_input(self, capture_dir): + """Readiness requires one retained frame before any input action.""" + with Recorder(capture_dir, task_description="Initial frame test") as rec: + assert rec.wait_for_ready(timeout=120), "recorder failed to start" + assert rec.screen_count >= 1 + + with CaptureSession.load(capture_dir) as capture: + assert capture.frames(), "capture completed without an initial frame" + @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.""" diff --git a/tests/test_window_capture.py b/tests/test_window_capture.py index 2af98bf..796d507 100644 --- a/tests/test_window_capture.py +++ b/tests/test_window_capture.py @@ -459,6 +459,204 @@ def test_processor_binds_first_later_frame_as_the_exact_action_after(scope): assert action.data["after_window_geometry_generation"] == 1 +def test_processor_retains_initial_desktop_frame_without_an_action(): + from openadapt_capture.config import RecordingConfig, config_override + + image = Image.new("RGB", (4, 3), "blue") + journal = queue.Queue() + journal.put(Event(1.0, "screen", image, 1)) + queues = [queue.Queue() for _ in range(6)] + counters = [multiprocessing.Value("i", 0) for _ in range(5)] + producers_finished = threading.Event() + producers_finished.set() + + with config_override( + RecordingConfig( + capture_video=True, + capture_images=False, + capture_full_video=False, + capture_window_data=False, + ) + ): + recorder_module.process_events( + journal, + queues[0], + queues[1], + queues[2], + queues[3], + queues[4], + queues[5], + SimpleNamespace(timestamp=0.0), + threading.Event(), + threading.Event(), + *counters, + producers_finished, + ) + + retained = queues[0].get_nowait() + video = queues[4].get_nowait() + assert retained.source_ordinal == 1 + assert retained.data is None + assert video.source_ordinal == 1 + assert video.data is image + assert counters[0].value == 1 + assert counters[4].value == 1 + + +def test_processor_retains_the_contiguous_desktop_action_journal(): + from openadapt_capture.config import RecordingConfig, config_override + + images = [Image.new("RGB", (4, 3), color) for color in ("red", "green", "blue")] + journal = queue.Queue() + journal.put(Event(1.0, "screen", images[0], 1)) + journal.put(Event(2.0, "screen", images[1], 2)) + journal.put(Event(3.0, "action", {"name": "mouse.down"}, 3)) + journal.put(Event(4.0, "screen", images[2], 4)) + queues = [queue.Queue() for _ in range(6)] + counters = [multiprocessing.Value("i", 0) for _ in range(5)] + producers_finished = threading.Event() + producers_finished.set() + + with config_override( + RecordingConfig( + capture_video=True, + capture_images=False, + capture_full_video=False, + capture_window_data=False, + ) + ): + recorder_module.process_events( + journal, + queues[0], + queues[1], + queues[2], + queues[3], + queues[4], + queues[5], + SimpleNamespace(timestamp=0.0), + threading.Event(), + threading.Event(), + *counters, + producers_finished, + ) + + retained_ordinals = [queues[0].get_nowait().source_ordinal for _ in range(3)] + video_ordinals = [queues[4].get_nowait().source_ordinal for _ in range(3)] + action = queues[1].get_nowait() + assert retained_ordinals == [1, 2, 4] + assert video_ordinals == retained_ordinals + assert action.source_ordinal == 3 + assert action.data["screenshot_source_ordinal"] == 2 + assert action.data["after_screenshot_source_ordinal"] == 4 + assert counters[0].value == 3 + assert counters[1].value == 1 + assert counters[4].value == 3 + + +def test_processor_retains_desktop_window_events_without_an_action(): + from openadapt_capture.config import RecordingConfig, config_override + + image = Image.new("RGB", (4, 3), "blue") + journal = queue.Queue() + journal.put(Event(1.0, "screen", image, 1)) + journal.put(Event(2.0, "window", {"title": "Fixture"}, 2)) + journal.put(Event(3.0, "screen", image, 3)) + queues = [queue.Queue() for _ in range(6)] + counters = [multiprocessing.Value("i", 0) for _ in range(5)] + producers_finished = threading.Event() + producers_finished.set() + + with config_override( + RecordingConfig( + capture_video=False, + capture_images=True, + capture_window_data=True, + ) + ): + recorder_module.process_events( + journal, + queues[0], + queues[1], + queues[2], + queues[3], + queues[4], + queues[5], + SimpleNamespace(timestamp=0.0), + threading.Event(), + threading.Event(), + *counters, + producers_finished, + ) + + retained_ordinals = [queues[0].get_nowait().source_ordinal for _ in range(2)] + window = queues[2].get_nowait() + assert retained_ordinals == [1, 3] + assert window.source_ordinal == 2 + assert counters[0].value == 2 + assert counters[2].value == 1 + + +def test_processor_can_abort_without_claiming_that_producers_finished(): + queues = [queue.Queue() for _ in range(7)] + counters = [multiprocessing.Value("i", 0) for _ in range(5)] + producers_finished = threading.Event() + processing_aborted = threading.Event() + processing_aborted.set() + + recorder_module.process_events( + queues[0], + queues[1], + queues[2], + queues[3], + queues[4], + queues[5], + queues[6], + SimpleNamespace(timestamp=0.0), + threading.Event(), + threading.Event(), + *counters, + producers_finished, + processing_aborted, + ) + + assert not producers_finished.is_set() + assert all(counter.value == 0 for counter in counters) + + +def test_processor_fails_loud_when_an_action_precedes_configured_window_evidence(): + from openadapt_capture.config import RecordingConfig, config_override + + journal = queue.Queue() + journal.put(Event(1.0, "screen", Image.new("RGB", (4, 3), "blue"), 1)) + journal.put(Event(2.0, "action", {"name": "mouse.down"}, 2)) + queues = [queue.Queue() for _ in range(6)] + counters = [multiprocessing.Value("i", 0) for _ in range(5)] + producers_finished = threading.Event() + producers_finished.set() + + with config_override( + RecordingConfig( + capture_video=False, + capture_images=True, + capture_window_data=True, + ) + ), pytest.raises(WindowCaptureError, match="configured window evidence"): + recorder_module.process_events( + journal, + queues[0], + queues[1], + queues[2], + queues[3], + queues[4], + queues[5], + SimpleNamespace(timestamp=0.0), + threading.Event(), + threading.Event(), + *counters, + producers_finished, + ) + + def test_processor_refuses_a_native_action_without_a_terminal_after_frame(scope): image, _ = scope.capture_frame(publish=False) journal = queue.Queue() diff --git a/tests/test_writer_shutdown_drain.py b/tests/test_writer_shutdown_drain.py index f722250..fc7fd37 100644 --- a/tests/test_writer_shutdown_drain.py +++ b/tests/test_writer_shutdown_drain.py @@ -23,6 +23,7 @@ from __future__ import annotations import multiprocessing +import queue import threading import time from types import SimpleNamespace @@ -32,7 +33,11 @@ from openadapt_capture.db import create_db, crud from openadapt_capture.extensions import synchronized_queue as sq from openadapt_capture.recorder import Event as RecorderEvent -from openadapt_capture.recorder import write_events, write_window_event +from openadapt_capture.recorder import ( + _wait_for_tasks_started, + write_events, + write_window_event, +) NUM_EVENTS = 64 JOIN_TIMEOUT = 30.0 @@ -177,6 +182,70 @@ def test_writer_terminating_mid_stream_keeps_the_tail(tmp_path, perf_q): assert _count_window_events(db_path) == NUM_EVENTS +def test_writer_readiness_can_require_the_first_committed_event(tmp_path, perf_q): + recording, db_path = _make_recording(tmp_path) + write_q = sq.SynchronizedQueue() + write_q.put(_window_event(recording, 0)) + terminate = multiprocessing.Event() + ready = multiprocessing.Event() + write_started = threading.Event() + allow_commit = threading.Event() + + def gated_write_fn(session, rec, event, perf_queue): + write_started.set() + assert allow_commit.wait(timeout=JOIN_TIMEOUT) + write_window_event(session, rec, event, perf_queue) + + writer = threading.Thread( + target=write_events, + args=( + "window", + gated_write_fn, + write_q, + multiprocessing.Value("i", 1), + perf_q, + recording, + db_path, + terminate, + ready, + ), + kwargs={"ready_after_first_event": True}, + ) + writer.start() + + assert write_started.wait(timeout=JOIN_TIMEOUT) + assert not ready.is_set() + allow_commit.set() + assert ready.wait(timeout=JOIN_TIMEOUT) + + terminate.set() + writer.join(timeout=JOIN_TIMEOUT) + assert not writer.is_alive() + assert _count_window_events(db_path) == 1 + + +def test_startup_readiness_timeout_records_a_fail_loud_error(): + terminate = threading.Event() + task_errors: queue.Queue = queue.Queue() + ready = threading.Event() + task = SimpleNamespace(is_alive=lambda: True) + + startup_ready = _wait_for_tasks_started( + {"screen_event_writer": task}, + {"screen_event_writer": ready}, + terminate, + task_errors, + timeout=0.01, + ) + + assert startup_ready is False + assert terminate.is_set() + task_name, error = task_errors.get_nowait() + assert task_name == "startup_readiness" + assert isinstance(error, TimeoutError) + assert "retain its initial frame" in str(error) + + def test_producer_exits_before_release_and_the_tail_survives(tmp_path, perf_q): """The record() ordering contract: release writers only after the producer has fully exited; every queued event is then committed."""