feat: Add STATE message handling for incremental sync support in universal source #987
Description
Summary
The universal source connector (SourcePyAirbyteUniversal) currently only forwards RECORD messages from the underlying PyAirbyte source. STATE messages are dropped, which means only full-refresh syncs are supported.
Context
In airbyte/cli/universal_connector/source.py, the read() method filters messages to only yield RECORD types. STATE messages require per-stream format conversion between airbyte_protocol.models and airbyte_cdk.models, which adds complexity.
Desired Behavior
- Forward STATE messages from the underlying source through the universal source connector
- Handle per-stream state format conversion between PyAirbyte's protocol models and the CDK models
- Enable incremental sync support for connectors wrapped by the universal source
References
- Related PR: feat(connectors): add smoke test source for destination regression testing #982
- File:
airbyte/cli/universal_connector/source.py(line 195)