From 3030406d34a5cf262b869527fde480acb95212bc Mon Sep 17 00:00:00 2001 From: JD Bothma Date: Mon, 3 Aug 2026 17:35:16 +0100 Subject: [PATCH 1/3] Read snapshot artifacts via async fsspec API Reading artifacts with the synchronous fs.open() crashed on the gs:// backend with "RuntimeError: got Future attached to a different loop". Pravda binds gcsfs's aiohttp session to the pipeline event loop by awaiting its async API during capture; the subsequent sync fs.open() dispatched the same cached filesystem instance's coroutine to fsspec's background loop, mixing loops on one session. Wrap synchronous backends in AsyncFileSystemWrapper (as Pravda's Storage.from_url does) and read via `await fs._cat_file()`, keeping all storage I/O on the running loop. Local (sync) backends are unaffected; this only manifested on remote async storage in the cluster. Co-Authored-By: Claude Opus 4.8 --- kolkhoz/capture.py | 19 ++++++++++++++++--- kolkhoz/cli.py | 6 +++--- 2 files changed, 19 insertions(+), 6 deletions(-) diff --git a/kolkhoz/capture.py b/kolkhoz/capture.py index 53dea56..56adef9 100644 --- a/kolkhoz/capture.py +++ b/kolkhoz/capture.py @@ -18,6 +18,7 @@ import os import fsspec +from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper from PIL import Image from pravda import Pravda, PravdaConfig, Snapshot @@ -46,12 +47,22 @@ def storage_filesystem(settings: PravdaSettings): Pravda resolves each snapshot's ``prefix`` against this same base path, so opening ``/`` on this filesystem locates the artifact for both local paths and remote (``gs://``/``s3://``) URLs. + + Synchronous backends (e.g. the local filesystem) are wrapped so reads use + the same async API as remote ones. This mirrors Pravda's own + ``Storage.from_url`` and, crucially, keeps every artifact read on the + running event loop: reading via the async API avoids fsspec's sync bridge, + which would otherwise drive the shared (async, e.g. ``gcsfs``) instance from + its background loop while Pravda has bound its session to this loop — + raising "got Future attached to a different loop". """ fs, _ = fsspec.core.url_to_fs(settings.storage_base_path) + if not fs.async_impl: + fs = AsyncFileSystemWrapper(fs) return fs -def read_artifact(fs, snapshot: Snapshot, filename: str | None) -> bytes: +async def read_artifact(fs, snapshot: Snapshot, filename: str | None) -> bytes: """Read a snapshot artifact blob from the shared storage backend. ``snapshot.prefix`` is the backend-resolved directory (base path plus the @@ -59,14 +70,16 @@ def read_artifact(fs, snapshot: Snapshot, filename: str | None) -> bytes: content-addressed name Pravda stored. Both are required for a stored artifact: a missing one is a malformed snapshot, so this fails loud rather than returning empty bytes. + + Reads through the async API on the current event loop; see + ``storage_filesystem`` for why the sync ``fs.open`` path is avoided. """ if snapshot.prefix is None: raise ValueError(f"snapshot {snapshot.id} has no storage prefix") if filename is None: raise ValueError(f"snapshot {snapshot.id} has no artifact filename") path = os.path.join(snapshot.prefix, filename) - with fs.open(path, "rb") as fh: - return fh.read() + return await fs._cat_file(path) def is_blank(blob: bytes) -> bool: diff --git a/kolkhoz/cli.py b/kolkhoz/cli.py index cefd048..7f9cd0e 100644 --- a/kolkhoz/cli.py +++ b/kolkhoz/cli.py @@ -43,10 +43,10 @@ async def extract_snapshot( client: OpenAI, ) -> list[dict]: """Extract flattened holder observations from one captured snapshot.""" - text = read_artifact(fs, snapshot, snapshot.plaintext).decode( + text = (await read_artifact(fs, snapshot, snapshot.plaintext)).decode( "utf-8", errors="replace" ) - html = read_artifact(fs, snapshot, snapshot.rendered_html).decode( + html = (await read_artifact(fs, snapshot, snapshot.rendered_html)).decode( "utf-8", errors="replace" ) @@ -56,7 +56,7 @@ async def extract_snapshot( if reason is not None: log.info(" → %s → including screenshot", reason) if snapshot.screenshot is not None: - blob = read_artifact(fs, snapshot, snapshot.screenshot) + blob = await read_artifact(fs, snapshot, snapshot.screenshot) if not is_blank(blob): screenshot_blob = blob metadata = metadata_from_html(snapshot.url, html) From de24058a454786c85f222d2db1c3c99a20e351f2 Mon Sep 17 00:00:00 2001 From: JD Bothma Date: Tue, 4 Aug 2026 12:41:18 +0100 Subject: [PATCH 2/3] Write outputs via async fsspec API too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The read-side fix moved the pipeline past capture and extraction, but it then crashed identically ("got Future attached to a different loop") in write_outputs, which used the synchronous fs.makedirs()/fs.open() on the gs:// backend from inside the pipeline event loop. Give write_outputs the same treatment as read_artifact: wrap sync backends and write via `await fs._makedirs()` / `await fs._pipe_file()`, keeping all storage I/O on the running loop. load_inputs is unaffected — it runs before asyncio.run(), outside the pipeline loop. Co-Authored-By: Claude Opus 4.8 --- kolkhoz/cli.py | 2 +- kolkhoz/export.py | 26 +++++++++++++++++++------- 2 files changed, 20 insertions(+), 8 deletions(-) diff --git a/kolkhoz/cli.py b/kolkhoz/cli.py index 7f9cd0e..a9343a1 100644 --- a/kolkhoz/cli.py +++ b/kolkhoz/cli.py @@ -158,7 +158,7 @@ async def _run_pipeline( if holders: hits += 1 - write_outputs(groups, config.paths) + await write_outputs(groups, config.paths) log.info("extraction: %d hit, %d miss", hits, extracted - hits) diff --git a/kolkhoz/export.py b/kolkhoz/export.py index 3a792e2..977e891 100644 --- a/kolkhoz/export.py +++ b/kolkhoz/export.py @@ -6,6 +6,7 @@ from datetime import datetime import fsspec +from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper from pravda import Snapshot from kolkhoz.config import PathsConfig @@ -56,19 +57,30 @@ def holder_to_record( return {name: record[name] for name in EXPORT_FIELDS} -def write_outputs(groups: dict[str, list[dict]], paths: PathsConfig) -> None: - """Write exactly the records produced by the current run.""" +async def write_outputs(groups: dict[str, list[dict]], paths: PathsConfig) -> None: + """Write exactly the records produced by the current run. + + Runs inside the pipeline's event loop, so it uses the async fsspec API on + that loop rather than the synchronous ``fs.makedirs``/``fs.open`` bridge; + see ``kolkhoz.capture.storage_filesystem`` for why mixing the sync bridge + with Pravda's async use of the shared (e.g. ``gcsfs``) instance raises + "got Future attached to a different loop". Synchronous backends (local + filesystem) are wrapped so the same async calls work there too. + """ fs, base = fsspec.core.url_to_fs(paths.output_base_path) + if not fs.async_impl: + fs = AsyncFileSystemWrapper(fs) date = datetime.now().strftime(EXPORT_DATE_FORMAT) total = 0 for dataset, records in groups.items(): out_dir = os.path.join(base, dataset) out_file = os.path.join(out_dir, f"{date}.jsonl") - fs.makedirs(out_dir, exist_ok=True) - with fs.open(out_file, "wb") as fh: - for record in records: - fh.write(json.dumps(record, ensure_ascii=False).encode("utf-8")) - fh.write(b"\n") + await fs._makedirs(out_dir, exist_ok=True) + payload = b"".join( + json.dumps(record, ensure_ascii=False).encode("utf-8") + b"\n" + for record in records + ) + await fs._pipe_file(out_file, payload) total += len(records) log.info("wrote %d record(s) → %s", len(records), out_file) log.info("wrote %d record(s) across %d dataset(s)", total, len(groups)) From 303ad4ee925a111736584e3acd50ffdb34b2f0b8 Mon Sep 17 00:00:00 2001 From: JD Bothma Date: Tue, 4 Aug 2026 15:09:36 +0100 Subject: [PATCH 3/3] Trim verbosity --- kolkhoz/export.py | 11 ++--------- 1 file changed, 2 insertions(+), 9 deletions(-) diff --git a/kolkhoz/export.py b/kolkhoz/export.py index 977e891..b990950 100644 --- a/kolkhoz/export.py +++ b/kolkhoz/export.py @@ -58,17 +58,10 @@ def holder_to_record( async def write_outputs(groups: dict[str, list[dict]], paths: PathsConfig) -> None: - """Write exactly the records produced by the current run. - - Runs inside the pipeline's event loop, so it uses the async fsspec API on - that loop rather than the synchronous ``fs.makedirs``/``fs.open`` bridge; - see ``kolkhoz.capture.storage_filesystem`` for why mixing the sync bridge - with Pravda's async use of the shared (e.g. ``gcsfs``) instance raises - "got Future attached to a different loop". Synchronous backends (local - filesystem) are wrapped so the same async calls work there too. - """ + """Write exactly the records produced by the current run.""" fs, base = fsspec.core.url_to_fs(paths.output_base_path) if not fs.async_impl: + # Consistently use async API on pipeline's event loop fs = AsyncFileSystemWrapper(fs) date = datetime.now().strftime(EXPORT_DATE_FORMAT) total = 0