feat: support stream configuration during on-demand stream creation

[infiniteregrets]

Adds stream_config to AppendInput and as a keyword argument on read, read_session, append_session, and producer. It is JSON-encoded and sent as the s2-stream-config header; the server applies it over the basin's default stream configuration only when the append or read auto-creates the stream, and ignores it if the stream already exists. Sessions resend it on every connect. Same shape as the Rust SDK (stream_config), s2-sdk-go#385, and s2-sdk-typescript#358.

config = StreamConfig(retention_policy=3600)

await stream.append(AppendInput(records=[...], stream_config=config))
await stream.read(start=SeqNum(0), stream_config=config)
async with stream.append_session(stream_config=config) as session: ...
async with stream.read_session(start=SeqNum(0), stream_config=config) as session: ...

Server side: s2-streamstore/s2#718, s2-streamstore/s2-cloud#1767.

[greptile-apps]

Confidence Score: 5/5

The PR appears safe to merge; the new option is consistently propagated across supported operations and remains compatible with the shared SDK contract.

Summary

This PR adds optional stream configuration for streams created on demand through unary and session-based append/read operations.

• Encodes StreamConfig as the s2-stream-config request header.
• Propagates configuration through append sessions, read sessions, and producers, including reconnects.
• Adds end-to-end coverage for inheritance from basin defaults and behavior with existing streams.
• Documents the new API with an example.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart LR
    C[Caller supplies StreamConfig] --> E[Encode configuration as JSON]
    E --> H[s2-stream-config header]
    H --> U[Unary append or read]
    H --> S[Append or read session]
    S --> R[Resend header on reconnect]
    U --> A{Stream exists?}
    R --> A
    A -->|No| M[Merge with basin defaults and create stream]
    A -->|Yes| I[Ignore supplied configuration]

Loading

_{Reviews (2) · Last reviewed commit: "integration test instead of mock"}