fix: missing input exhaustion check before append session completion

src/s2_sdk/_s2s/_append_session.py

@@ -280,6 +280,10 @@ async def _run_attempt(
             )
         if advised_reconnect.is_set():
             return _AttemptOutcome.RECONNECT_ADVISED
+        if not session_state.inputs_exhausted:
+            raise S2ClientError(
+                "Append session response stream closed before the input source was exhausted"
+            )
         return _AttemptOutcome.COMPLETE
 
 

tests/test_append_session.py

@@ -45,7 +45,7 @@ def decoded_messages(stream: AsyncIterator[bytes]) -> AsyncIterator[Message]:
 
     attempts = [
         ((_ack(0, reconnect_advised=True), _ack(1)), 2),
-        ((), 0),
+        ((), None),
     ]
     responses: list[_Response] = []
 
@@ -59,8 +59,12 @@ async def streaming_request(
         ) -> AsyncGenerator[_Response, None]:
             messages, inputs_to_consume = attempts.pop(0)
             if content is not None:
-                for _ in range(inputs_to_consume):
-                    await content.__anext__()
+                if inputs_to_consume is None:
+                    async for _ in content:
+                        pass
+                else:
+                    for _ in range(inputs_to_consume):
+                        await content.__anext__()
             response = _Response(messages)
             responses.append(response)
             try: