From 4d46f339c12774f91eb09fd5172d626d1d46ace0 Mon Sep 17 00:00:00 2001 From: Chris Alfano Date: Wed, 9 Sep 2026 12:51:46 -0400 Subject: [PATCH] fix(recorder): read every list page and stop losing the transcript race Two interception bugs made the CLI silently lossy. GetRecordingList is paginated at 10 items per page, and the web app issues one request per page. The interceptor resolved on the *first* response, so only the newest page was ever seen. Recordings past the first page could not be listed, and since `transcript`, `download`, and `info` all resolve a recording's audio_id through that same list, they failed those recordings with "Recording not found". Adding an 11th recording was enough to make a previously working recording unreachable. `_intercept_grpc_pages` now accumulates every page and merges them, deduped by id. The GetTranscription "wait for largest payload" path resolved exactly 3s after listener registration, whether or not anything had arrived; if the page issued the call later (cold profile, slow network) the future was never resolved and the caller timed out with a misleading "may not have a transcript yet". The settle window is now measured from the last arrival rather than from registration, so it waits for the payload instead of racing it. Verified 8/8 consecutive successes on recordings that previously failed ~40% of the time, including a 41-minute one. List and transcript deadlines go 30s -> 60s to accommodate the settle window, and the transcript timeout message no longer asserts a cause it cannot know. Co-Authored-By: Claude Opus 5 --- recorder_cli/recorder.py | 135 ++++++++++++++++++++++++++++++++------- 1 file changed, 111 insertions(+), 24 deletions(-) diff --git a/recorder_cli/recorder.py b/recorder_cli/recorder.py index 38346ce..b8666ac 100644 --- a/recorder_cli/recorder.py +++ b/recorder_cli/recorder.py @@ -46,37 +46,123 @@ async def _read() -> None: page.on("response", on_response) future.add_done_callback(lambda _: page.remove_listener("response", on_response)) else: - # Collect all responses and return the largest + # Collect all responses and return the largest. + # The endpoint sends a small payload first, then a larger full one, so we + # resolve only after a quiet SETTLE_SECONDS gap with no new arrivals -- + # measured from the LAST arrival, not from registration. A fixed delay + # measured from registration silently loses the race whenever the page + # takes longer than that to issue the call (cold profile, slow network), + # leaving the future unresolved until the caller's own timeout fires. responses = [] - + last_arrival: float | None = None + SETTLE_SECONDS = 3.0 + POLL_SECONDS = 0.25 + def on_response(response: object) -> None: if endpoint in response.url: # type: ignore[attr-defined] async def _read() -> None: + nonlocal last_arrival try: data = await response.json() # type: ignore[attr-defined] responses.append(data) + last_arrival = loop.time() except Exception: pass asyncio.ensure_future(_read()) - + page.on("response", on_response) - - # Set up cleanup - after a delay, pick the largest response - async def _resolve_largest(): - # Wait a bit for all responses to arrive - await asyncio.sleep(3) - if responses and not future.done(): - # Return the largest response (by JSON string length) - import json - largest = max(responses, key=lambda r: len(json.dumps(r))) - future.set_result(largest) - page.remove_listener("response", on_response) - + + async def _resolve_largest() -> None: + # Poll until the payloads stop arriving. Never gives up on its own -- + # the caller's asyncio.wait_for supplies the deadline. + while not future.done(): + await asyncio.sleep(POLL_SECONDS) + if last_arrival is None: + continue + if loop.time() - last_arrival < SETTLE_SECONDS: + continue + if responses and not future.done(): + import json + largest = max(responses, key=lambda r: len(json.dumps(r))) + future.set_result(largest) + page.remove_listener("response", on_response) + return + asyncio.ensure_future(_resolve_largest()) return future +def _intercept_grpc_pages(page, endpoint: str, settle_seconds: float = 3.0) -> asyncio.Future: + """ + Collect EVERY response for a paginated gRPC endpoint and merge their item lists. + + The recorder web app requests its recording list one page at a time (10 per + page), so intercepting a single response silently yields only the newest page. + Anything older is invisible -- it cannot be listed, and because the other + commands resolve a recording's audio_id through this same list, it also cannot + have its transcript, audio, or metadata fetched ("Recording not found"). + + Resolves with the merged ``[[items...], token]`` shape the parsers already + expect, deduped by recording id and keeping first-seen (newest-first) order. + """ + loop = asyncio.get_event_loop() + future: asyncio.Future = loop.create_future() + + merged: list = [] + seen: set = set() + token = None + last_arrival: float | None = None + + def on_response(response: object) -> None: + if endpoint not in response.url: # type: ignore[attr-defined] + return + + async def _read() -> None: + nonlocal last_arrival, token + try: + data = await response.json() # type: ignore[attr-defined] + items = data[0] + except Exception: + return + if not isinstance(items, list): + return + for item in items: + try: + rid = item[0] + except (IndexError, TypeError): + continue + if rid in seen: + continue + seen.add(rid) + merged.append(item) + if len(data) > 1: + token = data[1] + last_arrival = loop.time() + + asyncio.ensure_future(_read()) + + page.on("response", on_response) + + async def _resolve_when_settled() -> None: + # Pages trickle in; resolve once none have arrived for settle_seconds. + # The caller's asyncio.wait_for supplies the overall deadline. + while not future.done(): + await asyncio.sleep(0.25) + if last_arrival is None: + continue + if loop.time() - last_arrival < settle_seconds: + continue + if not future.done(): + future.set_result([merged, token]) + page.remove_listener("response", on_response) + return + + asyncio.ensure_future(_resolve_when_settled()) + + return future + + class RecorderClient: """Client for recorder.google.com operations.""" @@ -86,10 +172,10 @@ async def list_recordings(self) -> list[Recording]: context = await create_context(p) page = await context.new_page() try: - future = _intercept_grpc(page, "GetRecordingList") + future = _intercept_grpc_pages(page, "GetRecordingList") await page.goto(RECORDER_URL) try: - data = await asyncio.wait_for(asyncio.shield(future), timeout=30) + data = await asyncio.wait_for(asyncio.shield(future), timeout=60) except asyncio.TimeoutError: raise TimeoutError( "Recorder did not load. Session may be expired. Run: recorder login" @@ -125,10 +211,10 @@ async def get_transcript(self, recording_id: str) -> Transcript: page = await context.new_page() try: # Step 1: get recording list to find audio_id - list_future = _intercept_grpc(page, "GetRecordingList") + list_future = _intercept_grpc_pages(page, "GetRecordingList") await page.goto(RECORDER_URL) try: - list_data = await asyncio.wait_for(asyncio.shield(list_future), timeout=30) + list_data = await asyncio.wait_for(asyncio.shield(list_future), timeout=60) except asyncio.TimeoutError: raise TimeoutError( "Recorder did not load. Session may be expired. Run: recorder login" @@ -147,11 +233,12 @@ async def get_transcript(self, recording_id: str) -> Transcript: trans_future = _intercept_grpc(page, "GetTranscription", wait_for_largest=True) await page.goto(f"{RECORDER_URL}/{audio_id}") try: - data = await asyncio.wait_for(asyncio.shield(trans_future), timeout=30) + data = await asyncio.wait_for(asyncio.shield(trans_future), timeout=60) except asyncio.TimeoutError: raise TimeoutError( - f"Transcript not available for {recording_id}. " - "The recording may not have a transcript yet." + f"No GetTranscription payload for {recording_id} within 60s. " + "The recording may have no transcript yet, or the page did not " + "finish loading. Retrying usually succeeds." ) return self._parse_transcript(recording_id, data) finally: @@ -164,10 +251,10 @@ async def download_audio(self, recording_id: str, output_path: Path) -> Path: page = await context.new_page() try: # Get recording list to find audio_id and title - list_future = _intercept_grpc(page, "GetRecordingList") + list_future = _intercept_grpc_pages(page, "GetRecordingList") await page.goto(RECORDER_URL) try: - list_data = await asyncio.wait_for(asyncio.shield(list_future), timeout=30) + list_data = await asyncio.wait_for(asyncio.shield(list_future), timeout=60) except asyncio.TimeoutError: raise TimeoutError( "Recorder did not load. Session may be expired. Run: recorder login"