refactor(ogc): keep snapshot semantics for ChunkInterrupted partial state

Revert the live-view change (item 3 of #346): restore exc.partial_frame /
exc.partial_response as raise-time snapshots rather than properties that
delegate to exc.call.

Snapshot is the better contract for an exception — a record of the failure
moment, not a live handle:

- In a resume loop each interruption stays a faithful record of what it saw.
  Live-view aliased every exception to the one shared call, so after a second
  failure exc1.partial_frame and exc2.partial_frame returned the same (later)
  state.
- After a successful resume, live-view's .partial_frame returned the COMPLETE
  result — a field named "partial" that no longer was.
- Exception payloads shouldn't mutate after you catch them (cf.
  CalledProcessError.output).

Removing the delegation also drops the properties and the _pickled_* /
__getstate__ snapshot machinery it required: partial_frame/partial_response
are plain instance attributes again, pickled by the base __getstate__
(net -21 lines in interruptions.py).

No behavior change versus released main — this PR is unmerged, so live-view
never shipped. #346 is now a true no-behavioral-change refactor.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: thodson-usgs

dataretrieval/ogc/interruptions.py

@@ -55,14 +55,14 @@ class ChunkInterrupted(DataRetrievalError):
     total_chunks : int
         Total sub-requests in the plan.
     partial_frame : pandas.DataFrame
-        Live view of work completed so far — delegates to
-        ``call.partial_frame``. Advances as ``call.resume()`` makes
-        further progress. Returns an empty ``DataFrame`` when
-        ``call`` is ``None`` (degraded/unpickled state).
+        Combined frame of work completed by the moment this exception
+        was raised. Snapshot at raise time — does NOT advance on a
+        later ``call.resume()`` (use ``exc.call.partial_frame`` for
+        the live view).
     partial_response : httpx.Response or None
-        Live view of the raw aggregate response — delegates to
-        ``call.partial_response``. ``None`` when ``call`` is ``None``
-        or nothing has completed yet. (Raw, not finalized — use
+        Raw aggregate response covering the completed sub-requests at
+        raise time; ``None`` if nothing had completed yet. Same snapshot
+        semantics as ``partial_frame``. (Raw, not finalized — use
         ``exc.call.resume()`` for the finalized ``(df, metadata)`` result.)
 
     Examples
@@ -118,51 +118,30 @@ def __init__(
         self.total_chunks = total_chunks
         self.call = call
         self.retry_after = retry_after
-
-    @property
-    def partial_frame(self) -> pd.DataFrame:
-        """Live view of work completed so far — delegates to ``call.partial_frame``.
-
-        Returns an empty ``DataFrame`` when ``call`` is ``None`` (degraded /
-        unpickled state), unless a pickled snapshot was restored during
-        unpickling.
-        """
-        if self.call is not None:
-            return self.call.partial_frame
-        # Degraded / unpickled: return the snapshot captured at pickle time,
-        # or an empty frame if no snapshot exists (hand-constructed exc).
-        return getattr(self, "_pickled_partial_frame", pd.DataFrame())
-
-    @property
-    def partial_response(self) -> httpx.Response | None:
-        """Live view of the raw aggregate response.
-
-        Delegates to ``call.partial_response``.
-        Returns ``None`` when ``call`` is ``None`` (degraded / unpickled state)
-        and no pickled snapshot exists.
-        """
-        if self.call is not None:
-            return self.call.partial_response
-        return getattr(self, "_pickled_partial_response", None)
+        # Snapshot partial state at raise time so the exception stays a stable
+        # record of the failure moment: ``exc.partial_frame`` /
+        # ``.partial_response`` do NOT advance on a later ``call.resume()``
+        # (that live view is on ``call.partial_frame`` / ``.partial_response``).
+        # This keeps each interruption in a resume loop a faithful record of
+        # what it saw, rather than every exception aliasing the shared call's
+        # advancing state. ``.copy()`` guards the single-chunk fast path, where
+        # the combined frame may be returned verbatim.
+        if call is None:
+            self.partial_frame: pd.DataFrame = pd.DataFrame()
+            self.partial_response: httpx.Response | None = None
+        else:
+            self.partial_frame = call.partial_frame.copy()
+            self.partial_response = call.partial_response
 
     def __getstate__(self) -> dict[str, Any]:
         # Drop the live ChunkedCall before pickling: its ``.fetch`` is an
         # undecorated module function pickle can't reference by name, so the
         # interruption can't cross a process boundary with ``.call`` attached.
-        # The degraded ``call=None`` form keeps the counts, retry hint, and
-        # partial frame / response; only ``.resume()`` is lost (cross-process
-        # resume was never possible anyway).
-        #
-        # Since ``partial_frame`` and ``partial_response`` are now live
-        # properties (delegating to ``self.call``), snapshot them here into
-        # private ``_pickled_*`` keys. The base ``__setstate__`` restores
-        # those keys via ``self.__dict__.update``, and the properties fall
-        # back to them (through ``getattr``) once ``call`` is ``None`` — so
-        # no ``__setstate__`` override is needed.
-        state = {**super().__getstate__(), "call": None}
-        state["_pickled_partial_frame"] = self.partial_frame.copy()
-        state["_pickled_partial_response"] = self.partial_response
-        return state
+        # The degraded ``call=None`` form keeps the counts, retry hint, and the
+        # snapshotted partial frame / response — plain instance attributes the
+        # base ``__getstate__`` already pickles; only ``.resume()`` is lost
+        # (cross-process resume was never possible anyway).
+        return {**super().__getstate__(), "call": None}
 
 
 class QuotaExhausted(ChunkInterrupted):

tests/waterdata_chunking_test.py

@@ -956,11 +956,13 @@ async def fetch(args):
     assert err.partial_response is not None
 
 
-def test_partial_frame_is_live_view_across_resume():
-    """``exc.partial_frame`` is now a live view delegating to
-    ``exc.call.partial_frame``. Calling ``exc.call.resume()`` advances the
-    underlying ``ChunkedCall`` and ``exc.partial_frame`` reflects that — it
-    is no longer a frozen snapshot."""
+def test_partial_frame_snapshot_stable_across_resume():
+    """``exc.partial_frame`` / ``exc.partial_response`` snapshot the
+    state at raise time. Calling ``exc.call.resume()`` advances the
+    underlying ``ChunkedCall`` but must NOT mutate the snapshot on
+    the exception — otherwise a diagnostic that reads
+    ``exc.partial_frame`` after a resume sees post-resume state under
+    a name that promises pre-resume state."""
     state = {"i": 0, "blow_up": True}
 
     async def fetch(args):
@@ -977,24 +979,27 @@ async def fetch(args):
     with pytest.raises(QuotaExhausted) as excinfo:
         decorated({"sites": ["S1" * 10, "S2" * 10, "S3" * 10, "S4" * 10, "S5" * 10]})
     err = excinfo.value
-    pre_resume_rows = len(err.partial_frame)
-    assert pre_resume_rows > 0  # two chunks worth of data captured
+    snapshot_rows = len(err.partial_frame)
+    assert snapshot_rows > 0  # two chunks worth of data captured
 
-    # Resume; the live view on .call grows — and so does exc.partial_frame.
+    # Resume; the live view on .call grows.
     state["blow_up"] = False
     err.call.resume()
-    assert len(err.call.partial_frame) > pre_resume_rows
+    assert len(err.call.partial_frame) > snapshot_rows
 
-    # The exception's partial_frame IS the live view — it advances too.
-    assert len(err.partial_frame) == len(err.call.partial_frame)
+    # The exception's snapshot must NOT advance.
+    assert len(err.partial_frame) == snapshot_rows
 
 
-def test_partial_frame_is_live_view_when_single_chunk():
-    """``partial_frame`` is a live view delegating to ``call.partial_frame``.
-    When a single chunk completes, ``partial_frame`` still reflects the
-    current state of ``call.partial_frame`` (which recomputes on each
-    access). This verifies the live-view semantics even on the single-
-    chunk fast path."""
+def test_partial_frame_snapshot_is_a_copy_when_single_chunk():
+    """``_combine_chunk_frames`` returns ``non_empty[0]`` verbatim on
+    its single-frame fast path. ``ChunkInterrupted.__init__`` must
+    therefore defensively ``.copy()`` so an in-place mutation of the
+    underlying chunk frame (e.g. user diagnostic code adding a
+    column on the live view) doesn't leak through the snapshot.
+    Companion to ``test_partial_frame_snapshot_stable_across_resume``,
+    which uses ≥2 completed chunks and so goes through
+    ``pd.concat`` (which already produces a fresh frame)."""
     state = {"i": 0, "blow_up": True}
 
     async def fetch(args):
@@ -1017,11 +1022,12 @@ async def fetch(args):
     err = excinfo.value
     assert err.completed_chunks == 1
 
-    # The live view delegates to call.partial_frame; verify it's
-    # accessible and reflects the call's current state.
-    assert not err.partial_frame.empty
-    assert list(err.partial_frame.columns) == ["sites"]
-    assert err.partial_frame.equals(err.call.partial_frame)
+    snapshot_cols = list(err.partial_frame.columns)
+    # Mutate the underlying chunk in place — the snapshot must NOT
+    # reflect the mutation.
+    err.call._chunks[0][0]["extra"] = 0
+    assert list(err.partial_frame.columns) == snapshot_cols
+    assert "extra" not in err.partial_frame.columns
 
 
 def test_combine_chunk_responses_returns_independent_headers():