fix(sdk): close replaced stream iterators

DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md

@@ -6812,3 +6812,66 @@ untouched.
   desired-image pin still differs from the already accepted running Python-role
   image and requires a separate operator packet if those roles are to be
   recreated; it is not part of this consumer closure.
+
+#### C.8 Post-Merge Long-Soak Stream Session Recovery Closure
+
+**2026-08-21 status: `SOURCE PASS / IMMUTABLE RUNTIME ACCEPTANCE PENDING`:**
+
+- Source branches merged cleanly into Data Layer `origin/dev` commit `6b6b345`
+  and Trading System `origin/dev` commit `be80256`; both merge trees are byte
+  identical to their tested feature heads. No `dev -> main` release is allowed
+  by this fact alone.
+- Read-only inspection after approximately eight hours of runtime found the
+  Trading System V2 consumer repeatedly receiving `DATA_STALE`, then
+  `RATE_LIMITED: consumer request quota is exhausted` for Binance USD-M and OKX
+  Swap. The shared quota reached 602-604 requests against its unchanged
+  600/minute limit. `market_data_service` remained running but grew to about
+  212 MiB and no longer held stable long-lived TRADE sessions.
+- The source-owned SDK session replaces its current transport iterator during
+  retry and cursor replacement, while `warmup_then_stream` closes only the
+  iterator created at initial entry. A replacement iterator can therefore lose
+  cleanup ownership. Immediate bounded SDK retries then amplify a stream fault
+  into quota pressure. Raising quota/freshness, dropping events or weakening
+  Trading System validation is forbidden.
+- Approved hotfix scope is limited to explicit iterator ownership in
+  `WarmupStreamSession`: close the current iterator before replacement, close
+  the current iterator on context exit, make close idempotent and retain the
+  last acknowledged cursor. Public V2 models, protobuf, endpoint, provider,
+  source policy, freshness, quota and V1 contracts stay frozen.
+- Required gates are deterministic retry/cursor-replacement/context-exit close
+  tests; no duplicate/gap and no acknowledgment before consumer commit; the
+  existing stream SDK/transport/security suite; full network-off Python and
+  locked Rust regressions; deterministic `qdl_sdk==2.0.0` rebuild and consumer
+  repin; then an immutable runtime test with bounded real Binance/OKX streams,
+  request rate below quota, stable memory, fresh cache and V1 fallback intact.
+- Runtime rollback remains the already exercised Trading System service-only
+  `V2_PRIMARY -> V1` route. Source verification does not authorize a container
+  recreation, Redis mutation, authority CAS, DNSE promotion, alpha startup or
+  `dev -> main` release.
+
+**2026-08-21 source hotfix result: `PASS`:**
+
+- `WarmupStreamSession` now owns exactly one current iterator. Cursor expiry and
+  retry close the old iterator before replacement; terminal errors close it
+  before propagating; context exit closes the current replacement and repeated
+  `aclose()` is idempotent. Cursor restoration and acknowledgment ordering are
+  unchanged.
+- Deterministic recovery regression proves three generations (expired cursor,
+  transient reconnect and final live iterator) are each closed exactly once and
+  that an additional close is a no-op. The complete stream SDK suite passed
+  15/15, including real gRPC handoff, standby failover, slow-consumer recovery,
+  signed cursor scope and bar revision behavior.
+- Full network-off Data Layer Python regression passed 556 tests with six
+  explicit environment skips. Production SDK lint and `git diff --check`
+  passed. Rust source is byte-identical to merged `dev`; no Rust authority,
+  canonical or provider behavior changed in this SDK-only slice.
+- Two independent SDK builds were byte-identical at SHA-256
+  `6c1e374153756d1918be03c7efeac2d36c68ef235e46f12035ee59afa462a19a`;
+  source digest is
+  `1535f7f5cfb50050dc300a3b65471508ca9e10d4f3bcff0d9a9a9108cc23737e`.
+  The corresponding release manifest and SBOM are the only artifacts eligible
+  for the Trading System repin.
+- No running container, quota key, cursor, Kafka record, Redis projection,
+  PostgreSQL row, V1 route or authority state was changed. Runtime acceptance
+  remains blocked on an immutable image/recreation packet and a bounded
+  real-provider soak.

qdl_sdk/client.py

@@ -160,14 +160,18 @@ def __init__(
         self._reconnect_attempts = 0
         self._telemetry = telemetry
         self.state_restored = state_restored
+        self._closed = False
 
     def __aiter__(self):
         return self
 
     async def __anext__(self) -> StreamEvent | ControlEvent:
+        if self._closed or self._events is None:
+            raise StopAsyncIteration
         try:
             event = await self._events.__anext__()
         except CursorExpiredError:
+            await self._close_events()
             self.warmup = await self._fresh_snapshot()
             self._last_seen_offset = self.warmup.watermark_offset
             self._events = self._subscribe(
@@ -180,6 +184,7 @@ async def __anext__(self) -> StreamEvent | ControlEvent:
                 self.warmup,
             )
         except DataLayerError as error:
+            await self._close_events()
             if not error.retryable or self._reconnect_attempts >= self._max_reconnect_attempts:
                 raise
             self._reconnect_attempts += 1
@@ -227,6 +232,19 @@ def acknowledge(self, event: StreamEvent) -> None:
                 cursor_offset=event.logical_offset,
             )
 
+    async def aclose(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        await self._close_events()
+
+    async def _close_events(self) -> None:
+        events = self._events
+        self._events = None
+        close = getattr(events, "aclose", None)
+        if close is not None:
+            await close()
+
     def _subscribe(self, token: str):
         return self._stream_transport.subscribe(
             self.requirement,
@@ -482,9 +500,7 @@ async def warmup_then_stream(
         try:
             yield session
         finally:
-            close = getattr(events, "aclose", None)
-            if close is not None:
-                await close()
+            await session.aclose()
 
     async def close(self) -> None:
         await self.stream_transport.close()

tests/test_fund_phase5_stream_sdk.py

@@ -244,6 +244,7 @@ def load(self, requirement, *, consumer_id):
 class ScriptedIterator:
     def __init__(self, values):
         self.values = list(values)
+        self.close_calls = 0
 
     def __aiter__(self):
         return self
@@ -257,18 +258,21 @@ async def __anext__(self):
         return value
 
     async def aclose(self):
-        return None
+        self.close_calls += 1
 
 
 class ScriptedStreamTransport:
     def __init__(self, scripts):
         self.scripts = list(scripts)
         self.tokens = []
+        self.iterators = []
 
     def subscribe(self, requirement, **kwargs):
         del requirement
         self.tokens.append(kwargs["cursor_token"])
-        return ScriptedIterator(self.scripts.pop(0))
+        iterator = ScriptedIterator(self.scripts.pop(0))
+        self.iterators.append(iterator)
+        return iterator
 
     async def close(self):
         return None
@@ -761,6 +765,15 @@ async def test_cursor_expiration_rebuilds_snapshot_and_transient_error_reconnect
             stream.tokens,
             ["snapshot-token", "snapshot-token", "token-1"],
         )
+        self.assertEqual(
+            [iterator.close_calls for iterator in stream.iterators],
+            [1, 1, 1],
+        )
+        await session.aclose()
+        self.assertEqual(
+            [iterator.close_calls for iterator in stream.iterators],
+            [1, 1, 1],
+        )
 
     async def test_warmup_applies_realtime_quality_only_to_tail_watermark(self):
         requirement = DataRequirement(