Bug - `incremental_dependency: true` triggers full refresh on child streams due to context loss in SubstreamPartitionRouter (Concurrent CDK)

### [Bug] `incremental_dependency: true` triggers full refresh on child streams due to context loss in SubstreamPartitionRouter (Concurrent CDK)

*Note: This investigation stems from [Issue #61567](https://github.com/airbytehq/airbyte/issues/61567) already reported in the main Airbyte repository. To identify the root cause, I leveraged GitHub Copilot and Gemini Pro to analyze the current architecture of the `airbyte-python-cdk` repository. The findings point to a specific architectural gap causing the bug.*

#### Problem Summary
There is a severe performance degradation in the Concurrent CDK affecting any connector built with the Declarative CDK / Connector Builder that uses hierarchical relationships with `incremental_dependency: true`.

- **Initial Sync (Sync 1):** Works correctly. The `streamState` is persisted with the parent's cursor checkpoints.
- **Subsequent Syncs (Sync 2+):** The child stream loses its state context, forcing the parent stream to ignore its saved cursor and restart data extraction from the beginning (`start_date`).
- **Impact:** Delta syncs are 100x to 1000x slower. This affects production connectors such as `source-gmail`, `source-mixpanel`, `source-instagram`, `source-typeform`, and custom implementations (e.g., `source-dolibarr`).

#### Root Cause Analysis (Code Investigation)
The issue lies in how the concurrent engine fails to propagate the state context within the `SubstreamPartitionRouter`.

**Target File:** `airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py`

**1. Context loss in `cursor_slice` (Line ~214):**
Inside the `stream_slices` generator, when iterating over partitions (`partition.read()`), the `StreamSlice` is built and yielded with a hardcoded empty dictionary for the cursor:
```python
yield StreamSlice(
partition={
partition_field: partition_value,
"parent_slice": parent_partition or {},
},
cursor_slice={}, # <-- Parent's cursor context is lost here
extra_fields=extracted_extra_fields,
)
```
By yielding an empty `cursor_slice`, downstream components that rely on this slice to maintain the parent's checkpoint fail to associate the exact cursor bounds.

**2. Brittle state migration (`_migrate_child_state_to_parent_state` - Line ~261):**
The CDK attempts to mitigate this by rebuilding the parent state. However, the logic assumes very simple structures (strictly looking for a `"state"` key) and fails to rebuild the nested `parent_state` envelope required by child streams when multiple cursors or complex dates are involved.

**State Structure Breakdown:**
The state JSON collapses from its required nested format to an unusable flat structure:
*Required:* `{"parent_state": {"parent_stream_name": {"date_modification": "2024-08-01T00:00:00Z"}}}`
*Generated (Broken):* `{"date_modification": "2024-08-01T00:00:00Z"}`

---

#### Proposed Core CDK Solution
To resolve this natively in the concurrent framework, `stream_slices` and `read_parent_stream` should be modified to optionally accept `stream_state`. The parent state should be extracted inline and injected into the `cursor_slice` rather than forcing an empty dictionary. Additionally, `_migrate_child_state_to_parent_state` needs to be strengthened to respect nested `parent_state` structures.

---

#### Temporary Connector-Level Workaround (Proof of Concept)
While a core solution is evaluated, it is possible to bypass this limitation in affected connectors by implementing a custom `StateMigration` that rebuilds the JSON before the CDK parses it. This makes the workaround idempotent and works without breaking the concurrent engine:

**1. `components.py`**
```python
from dataclasses import dataclass
from typing import Any, Mapping
from airbyte_cdk.sources.declarative.state_migration import StateMigration

@dataclass
class IncrementalSyncStateMigration(StateMigration):
def should_migrate(self, stream_state: Mapping[str, Any]) -> bool:
return "parent_state" not in stream_state and stream_state and len(stream_state) > 0

def migrate(self, stream_state: Mapping[str, Any]) -> Mapping[str, Any]:
if not stream_state:
return stream_state

migrated_parent_state = {}
for key, value in stream_state.items():
if key != "states" and isinstance(value, dict) and any(
cursor in value for cursor in ["date_modification", "updated_at", "timestamp", "tms"]
):
migrated_parent_state[key] = value

return {
"parent_state": migrated_parent_state,
"states": [],
"lookback_window": 0,
}
```
*(Integrated by registering the `state_migration` class directly in the `definitions` block of `manifest.yaml`).*

**Steps to reproduce and verify:**
1. Create a connector with a parent and child stream using `incremental_dependency: true`.
2. Run the initial sync and save the `STATE` output.
3. Inject the saved `STATE` into a second run (`docker-compose run connector read config.json catalog.json <<< "$STATE"`).
4. **Expected result:** Fast delta sync.
5. **Actual result:** Logs show re-extraction of all parent records.

I would highly appreciate it if the maintainers could review this state propagation architecture. Official connectors (like `source-intercom` and `source-jira`) have already had to patch this locally. Does fixing `cursor_slice` in the main router align with the vision for the concurrent CDK?

Start with airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py, especially stream_slices, read_parent_stream, and _migrate_child_state_to_parent_state, then reproduce a parent/child sync with incremental_dependency: true. Compare the saved STATE between initial and subsequent runs; done means the child retains the parent cursor context and the second sync performs a delta instead of re-extracting all parent records.

由索引模型根據 Issue 內容生成。