Add deterministic batch runtime and harden GFS caching

[kirankarnam]

Earth2Studio Pull Request

Description

Adds a deterministic batching API for executing compatible forecast requests sequentially while reusing model and data-source resources. The runtime supports injected model loaders for callers such as PhysicsNeMo-Serve, preserves per request results, continues after individual failures, and publishes successful Zarr outputs atomically.

Also hardens GFS caching by:

• Including the path, byte offset, and byte length in cache identity.
• Validating cached and downloaded range sizes.
• Coalescing concurrent downloads safely across event loops.
• Publishing cache files atomically.
• Invalidating corrupt GRIB data without deleting valid data when a modifier fails.

Adds unit tests, API documentation, Sphinx registration, and changelog entries.

Checklist

• I am familiar with the Contributing Guidelines.
• New or existing tests cover these changes.
• The documentation is up to date with these changes.
• The CHANGELOG.md is up to date with these changes.
• An issue is linked to this pull request.
• Assess and address Greptile feedback (AI code review bot for guidance; use discretion, addressing all feedback is not required).

Dependencies

[greptile-apps]

Greptile Summary

This PR adds a DeterministicBatchRuntime for sequentially executing forecast requests with shared model and data-source resources, and hardens GFS caching with a wider cache key (path + offset + length), size validation, atomic tmp\u2192final promotion, and loop-safe download coalescing via a per-loop task registry.

• GFS cache: hash now covers byte_length, preventing range collisions; corrupted GRIB files are invalidated on open failure while valid files are preserved on modifier failure; concurrent same-loop downloads are coalesced via asyncio.shield over a shared asyncio.Task.
• Batched runtime: DeterministicBatchRuntime exposes injection points for model loader, data factory, and runner; each request writes to a staging path and atomically renames to the final path on success; failures are isolated per-request with staging cleanup.
• Tests: focused unit tests cover all new GFS cache behaviors and the batch runtime\u2019s resource-reuse, failure-isolation, and partial-state-rollback contracts.

Confidence Score: 3/5

The GFS cache hardening is well-designed overall, but _cache_file_is_valid catches only FileNotFoundError from stat(), so a PermissionError on an existing cache file would propagate through the entire fetch path rather than falling back to a fresh download.

_cache_file_is_valid catches only FileNotFoundError when calling stat(), meaning a PermissionError or I/O error on an existing cache file propagates as an unexpected exception instead of gracefully retrying. The download helper also splits a write across two asyncio.to_thread calls with the file handle held in the event loop, creating a non-standard ownership boundary during cancellation.

earth2studio/data/gfs.py — specifically _cache_file_is_valid (exception scope) and _download_remote_file (write + replace threading pattern).

Important Files Changed

Filename	Overview
earth2studio/data/gfs.py	Significant cache hardening with new hash, atomic writes, and loop-safe coalescing. Exception handling in _cache_file_is_valid is too narrow and old-scheme cache files accumulate.
earth2studio/batched_workflows.py	New deterministic batch runtime with clean injection points, atomic staging promotion, per-request failure isolation, and correct partial-state rollback.
test/data/test_gfs.py	Comprehensive unit tests for all new GFS cache behaviors.
test/test_batched_workflows.py	Good coverage of batch runtime resource reuse, failure isolation, and partial-state rollback.
docs/superpowers/specs/2026-07-15-gfs-cache-integrity-design.md	Internal design spec committed under docs/; unusual artifact for a public OSS repository.
docs/modules/workflows.rst	Sphinx registration for the new batched_workflows module looks correct.
CHANGELOG.md	Changelog entries added correctly for both the new batching API and GFS cache fixes.

Comments Outside Diff (1)

1. docs/superpowers/specs/2026-07-15-gfs-cache-integrity-design.md, line 1-74 (link)

   Internal design spec committed to the public repo

   docs/superpowers/specs/ looks like an internal AI-assisted design artifact rather than user-facing documentation. Committing it to a public OSS repo under docs/ may confuse external contributors. If useful as a record of intent, a dev/ or docs/dev/ directory would be a more appropriate home.

   Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

_{Reviews (1): Last reviewed commit: "remove design doc" | Re-trigger Greptile}

[greptile-apps]

You're right on all three points. Looking at the code:

1. _load_default_data_source() returns GFS() — that's just object construction, no network I/O.
2. The actual GFS fetch happens inside _run_default_deterministic_forecast → deterministic(...), which is called after self._model = model and self._data = data are committed. So a transient download failure goes through the except block in run(), not through _ensure_loaded, and the cached state is preserved for the next call.
3. PyTorch's CUDA allocator cache is a pool of reusable blocks — it doesn't represent wasted memory.

My original comment assumed GFS downloads could fail during data_factory() and leave model memory unrecoverable, but the actual execution order makes that scenario impossible. I'll withdraw the concern.