feat!: introduce ReadSession with caught-up state tracking

[infiniteregrets]

No description provided.

[greptile-apps]

Greptile Summary

This PR introduces ReadSession, a new first-class async iterator that wraps a stream read session and exposes is_caught_up() / caught_up() for tracking when the consumer has processed all records up to the stream's reported tail. The existing S2Stream.read_session async-generator is replaced by a synchronous factory returning a ReadSession object, and ignore_command_records filtering is promoted from the low-level transport layer into ReadSession._handle_event so that command records are accounted for in caught-up calculations before being filtered for delivery.

• run_read_session in _s2s/_read_session.py now yields a typed discriminated union (_ReadSessionBatch | _ReadSessionHeartbeat | _ReadSessionRetrying) instead of raw ReadBatch values, and hard-raises S2ClientError on an empty batch with no tail.
• ReadSession synchronises the background _run task with the consumer via a maxsize=1 delivery queue and a per-delivery asyncio.Event; cancellation re-propagation follows the current.cancelling() pattern to avoid swallowing outer-task CancelledError during close().
• The caught_up_tail invariant—set only when last_record.seq_num + 1 == tail.seq_num—matches the Go SDK's readBatchCaughtUp semantics.

Confidence Score: 5/5

Safe to merge — the delivery queue + event synchronisation is correct, cancellation re-propagates cleanly, and the caught-up semantics match the Go SDK.

The core ReadSession logic is well-structured: the maxsize-1 queue gates exactly one in-flight delivery at a time, _finish drains and resets it atomically (no await points), and the current.cancelling() guard correctly re-raises outer-task cancellation. Caught-up state transitions align with the Go SDK reference implementation.

No files require special attention.

Important Files Changed

Filename	Overview
src/s2_sdk/_read_session.py	New ReadSession class implementing async iterator with caught-up state tracking via delivery queue + asyncio.Event; cancellation and error propagation are well-handled.
src/s2_sdk/_s2s/_read_session.py	run_read_session refactored to yield typed _ReadSessionEvent discriminated union instead of ReadBatch; ignore_command_records moved up to ReadSession layer; empty-batch-without-tail now raises a hard error.
src/s2_sdk/_ops.py	read_session changed from async generator to sync factory returning ReadSession; @Fallible decorator correctly switches to sync_wrapper path.
tests/test_read_session.py	New unit test file covering caught-up transitions, heartbeats, command-record filtering, retry semantics, cancellation, and error propagation.
tests/test_correctness.py	Updated to use async context manager pattern; added correctness test for is_caught_up() after tail delivery.
src/s2_sdk/_client.py	Minor cleanup: try/except CancelledError replaced with contextlib.suppress in ConnectionPool.close().

_{Reviews (3): Last reviewed commit: "return -> get" | Re-trigger Greptile}

[quettabit]

@greptileai pls review again