Fix concurrent capture worker rebuild after a reaped browser

code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_pipeline_and_manifest.py

@@ -108,6 +108,111 @@ def factory(_cfg):
     assert summary.count("stored") == 1  # retry on the fresh browser succeeded
 
 
+def test_concurrent_rebuild_survives_poisoned_thread_loop_state(tmp_path, make_fetcher):
+    """Regression: a SIGKILLed sync-Playwright browser leaves its asyncio loop
+    registered as *running* on the worker thread (teardown is thread-affine, so
+    ``_close_quietly``'s helper thread can't clear it). The rebuild must reset
+    that thread-local state — otherwise ``sync_playwright().start()`` raises
+    "Sync API inside the asyncio loop" and future.result() kills the whole run.
+    """
+    import asyncio
+    import threading
+
+    config = ArchiveConfig(s3_prefix="t", concurrency=1)
+    store = ContentStore(LocalBlobStore(tmp_path), config)
+    urls = ["https://a.test/x", "https://b.test/y"]
+    builds = {"n": 0}
+
+    class _SyncPlaywrightAlike:
+        """Mimics sync Playwright's thread behavior: __enter__ refuses if the
+        thread already reports a running loop, then registers its own; a killed
+        browser errors on fetch; teardown from a foreign thread fails the way a
+        greenlet does, leaving the loop state poisoned."""
+
+        name = "pwalike"
+
+        def __init__(self, dead: bool, inner):
+            self._dead = dead
+            self._inner = inner
+            self._thread = None
+
+        def __enter__(self):
+            try:
+                asyncio.get_running_loop()
+            except RuntimeError:
+                pass
+            else:  # what playwright's sync context manager raises
+                raise RuntimeError(
+                    "It looks like you are using Playwright Sync API inside "
+                    "the asyncio loop."
+                )
+            self._thread = threading.current_thread()
+            asyncio.events._set_running_loop(asyncio.new_event_loop())
+            return self
+
+        def fetch(self, url):
+            if self._dead:
+                raise RuntimeError("Target page, context or browser has been closed")
+            return self._inner.fetch(url)
+
+        def __exit__(self, *exc):
+            if threading.current_thread() is not self._thread:
+                raise RuntimeError("cannot switch to a different thread")
+            asyncio.events._set_running_loop(None)
+
+    def factory(_cfg):
+        builds["n"] += 1
+        inner = make_fetcher()
+        for u in urls:
+            inner.add(u)
+        return _SyncPlaywrightAlike(dead=builds["n"] == 1, inner=inner)
+
+    try:
+        summary = capture_urls_concurrent(urls, store, config, factory)
+    finally:
+        asyncio.events._set_running_loop(None)  # never leak into other tests
+
+    assert builds["n"] == 2  # death detected -> loop state reset -> rebuilt
+    assert summary.count("stored") == 2  # retried URL and the rest captured
+
+
+def test_concurrent_failed_rebuild_does_not_kill_the_run(tmp_path, make_fetcher):
+    """If the rebuild itself fails (e.g. a transient launch error), the run must
+    keep going: the URL keeps its error outcome and the next URL retries the
+    rebuild."""
+    from contextlib import contextmanager
+
+    config = ArchiveConfig(s3_prefix="t", concurrency=1)
+    store = ContentStore(LocalBlobStore(tmp_path), config)
+    urls = ["https://a.test/x", "https://b.test/y"]
+    builds = {"n": 0}
+
+    class _DeadBrowser:
+        name = "dead"
+
+        def fetch(self, url):
+            raise RuntimeError("Target page, context or browser has been closed")
+
+    @contextmanager
+    def factory(_cfg):
+        builds["n"] += 1
+        if builds["n"] == 1:
+            yield _DeadBrowser()  # first browser is dead
+        elif builds["n"] == 2:
+            raise RuntimeError("browser launch flaked")  # rebuild attempt fails
+        else:
+            fetcher = make_fetcher()
+            for u in urls:
+                fetcher.add(u)
+            yield fetcher  # second rebuild attempt works
+
+    summary = capture_urls_concurrent(urls, store, config, factory)
+
+    assert builds["n"] == 3  # dead -> failed rebuild -> successful rebuild
+    assert summary.count("error") == 1  # first URL kept its error outcome
+    assert summary.count("stored") == 1  # second URL captured after recovery
+
+
 class _BoomFetcher:
     """Raises an unexpected (non-FetchError) exception, like a bad screenshot."""
 

forecasting_tools/agents_and_tools/source_archive/fetchers/playwright_fetcher.py

@@ -149,11 +149,24 @@ def __enter__(self) -> "PlaywrightFetcher":
         return self
 
     def __exit__(self, *exc) -> None:
-        if self._browser is not None:
-            self._browser.close()
+        # close() raises when the browser process is already gone (crashed or
+        # killed by the pipeline's reaper). Still attempt stop(): it tears down
+        # the driver and un-registers the sync API's event loop from this
+        # thread — without that, the next sync_playwright().start() here fails
+        # with "Sync API inside the asyncio loop".
+        try:
+            if self._browser is not None:
+                self._browser.close()
+        except Exception as e:
+            logger.info("browser close failed (already dead?): %s", e)
+        finally:
             self._browser = None
-        if self._playwright is not None:
-            self._playwright.stop()
+        try:
+            if self._playwright is not None:
+                self._playwright.stop()
+        except Exception as e:
+            logger.info("playwright stop failed: %s", e)
+        finally:
             self._playwright = None
 
     def _settle(self, page) -> None:

forecasting_tools/agents_and_tools/source_archive/pipeline.py

@@ -134,6 +134,43 @@ def _close() -> None:
     done.wait(timeout_s)
 
 
+def _running_loop():
+    import asyncio
+
+    try:
+        return asyncio.get_running_loop()
+    except RuntimeError:
+        return None
+
+
+def _restore_thread_loop_state(baseline) -> None:
+    """Un-poison a worker thread's asyncio state after abandoning a dead
+    sync-Playwright instance.
+
+    Sync Playwright drives its asyncio loop *on the calling thread* (via a
+    greenlet), so while an instance is alive the thread is marked as being
+    inside a running event loop. A clean ``stop()`` clears that mark — but a
+    SIGKILLed browser can't be closed cleanly: ``close()``/``stop()`` fail or
+    hang, and both are thread-affine, so :func:`_close_quietly`'s helper thread
+    can't reach them either. The abandoned loop then stays registered as
+    running, and the next ``sync_playwright().start()`` on this thread refuses
+    with "Playwright Sync API inside the asyncio loop", killing the rebuild.
+
+    Resetting the thread-local marker to its pre-fetcher ``baseline`` lets the
+    rebuilt fetcher start a fresh loop. The old loop object is leaked on
+    purpose (its browser processes are swept by the reaper); that is the price
+    of recovering without touching thread-affine Playwright internals.
+    """
+    import asyncio
+
+    if _running_loop() is baseline:
+        return
+    try:
+        asyncio.events._set_running_loop(baseline)
+    except Exception:
+        pass
+
+
 def _reap_browser_descendants() -> None:
     """Best-effort: kill automation Chromium descending from this process. Used
     both to recover a wedged worker (kill its browser so the blocked sync call
@@ -219,6 +256,7 @@ def supervisor() -> None:
 
     def work(idx: int, shard: list[str]) -> list[CaptureOutcome]:
         outcomes: list[CaptureOutcome] = []
+        baseline_loop = _running_loop()  # almost always None; see _restore_...
         cm = fetcher_factory(config)
         pipeline = CapturePipeline(cm.__enter__(), store)
         try:
@@ -231,16 +269,31 @@ def work(idx: int, shard: list[str]) -> list[CaptureOutcome]:
                         "browser died; rebuilding worker %d, retrying %s", idx, url
                     )
                     _close_quietly(cm)
-                    cm = fetcher_factory(config)
-                    pipeline = CapturePipeline(cm.__enter__(), store)
-                    with hb_lock:
-                        heartbeats[idx] = time.monotonic()
-                    outcome = pipeline.capture_url(url)  # one retry on a fresh browser
+                    _restore_thread_loop_state(baseline_loop)
+                    try:
+                        cm = fetcher_factory(config)
+                        pipeline = CapturePipeline(cm.__enter__(), store)
+                    except Exception as e:
+                        # A failed rebuild must never kill the run: keep this
+                        # URL's error outcome and move on — the still-dead
+                        # pipeline makes the next URL error with a dead-browser
+                        # reason, which retries the rebuild.
+                        logger.warning(
+                            "worker %d rebuild failed (%s); keeping error outcome",
+                            idx,
+                            e,
+                        )
+                    else:
+                        with hb_lock:
+                            heartbeats[idx] = time.monotonic()
+                        # one retry on a fresh browser
+                        outcome = pipeline.capture_url(url)
                 outcomes.append(outcome)
                 with hb_lock:
                     heartbeats[idx] = None
         finally:
             _close_quietly(cm)
+            _restore_thread_loop_state(baseline_loop)
         return outcomes
 
     supervisor_thread = threading.Thread(target=supervisor, daemon=True)