feat(cdk): Add cursor age validation to StateDelegatingStream

[devin-ai-integration]

Summary

Adds an optional api_retention_period field to StateDelegatingStream that validates whether a cursor is within an API's data retention window before using incremental sync. When the cursor is older than the retention period, the connector automatically falls back to full refresh to avoid data loss.

This addresses the issue where APIs like Stripe Events only retain data for 30 days - if a sync fails mid-way and resumes after the retention window, incremental sync would miss data.

Key changes:

• Added api_retention_period field to StateDelegatingStream schema (ISO8601 duration format: P30D, P1D, PT1H, etc.)
• Added get_cursor_datetime_from_state to concurrent cursor hierarchy (Cursor base, ConcurrentCursor, ConcurrentPerPartitionCursor, FinalStateCursor)
• Raises ValueError at model level when IncrementingCountCursor is used with api_retention_period
• Emits warning log when falling back to full refresh due to stale cursor
• Clears stream state and emits an empty state message to the platform when falling back to full refresh, so the platform does not retain stale state across sync attempts
• No breaking changes — the field is optional with no default

Example usage:

type: StateDelegatingStream
name: events
api_retention_period: P30D  # 30 days
full_refresh_stream: ...
incremental_stream: ...

Updates since last revision

Latest changes (removed early return for NO_CURSOR_STATE_KEY — per tolik0's feedback):

• Removed the explicit early return for NO_CURSOR_STATE_KEY in _is_cursor_older_than_retention_period
• FinalStateCursor.get_cursor_datetime_from_state now handles this case by returning now() for NO_CURSOR_STATE_KEY: True state, which is always within any retention period
• This simplifies the code by letting the cursor class handle its own state format

Previous changes (tolik0's FinalStateCursor feedback):

• FinalStateCursor.get_cursor_datetime_from_state returns now() if state has NO_CURSOR_STATE_KEY: True, otherwise returns None
• Updated _is_cursor_older_than_retention_period to check both cursors in sequence (full refresh cursor first, then incremental cursor)

Earlier changes (state clearing on fallback to full refresh — per brianjlai):

• When cursor age validation triggers a fallback to full refresh, the stream's state is now cleared via ConnectorStateManager.update_state_for_stream and an empty state message is emitted through the message repository.

Review & Testing Checklist for Human

• Verify FinalStateCursor behavior — FinalStateCursor.get_cursor_datetime_from_state returns now() for NO_CURSOR_STATE_KEY: True state. Verify this correctly causes completed full refresh states to use incremental sync (since now() is always within retention).
• Verify state-clearing message ordering — The empty state message is emitted during stream creation (in the factory) before the read starts. Verify that the message repository correctly delivers this empty state message to the platform before any record messages.
• Verify the dual-cursor checking flow — The factory checks full_refresh_cursor first, then incremental_cursor. Verify this correctly handles all state formats (datetime-based, per-partition, concurrent).

Recommended test plan:

1. Apply this change to the Stripe connector's events stream with api_retention_period: P30D, set a cursor state older than 30 days, and verify it falls back to full refresh with the expected warning message
2. Test with FinalStateCursor state (e.g., after a completed full refresh with NO_CURSOR_STATE_KEY: True) to verify it correctly uses incremental sync without falling back to full refresh
3. Test with IncrementingCountCursor to confirm it raises ValueError during discover() with a clear error message

Notes

Fixes: https://github.com/airbytehq/oncall/issues/11103

Link to Devin run: https://app.devin.ai/sessions/c6b25a1216c547139ef8242062f7f135
Previous Devin sessions: https://app.devin.ai/sessions/443d1d1a2f524eb1bc378aa48f24aa9c, https://app.devin.ai/sessions/5e2b4ff66c50407789e500d6e25e5d5c
Requested by: Alfredo Garcia (@agarctfi)

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment and CI monitoring

[github-actions]

PyTest Results (Fast)

3 881 tests  +26   3 869 ✅ +26   6m 51s ⏱️ +24s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit a3a2073. ± Comparison against base commit 15542de.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

3 884 tests  +26   3 872 ✅ +26   10m 36s ⏱️ -25s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit a3a2073. ± Comparison against base commit 15542de.

♻️ This comment has been updated with latest results.

[Alfredo Garcia (agarctfi)]

/autofix

Auto-Fix Job Info

This job attempts to auto-fix any linting or formating issues. If any fixes are made,
those changes will be automatically committed and pushed back to the PR.

Note: This job can only be run by maintainers. On PRs from forks, this command requires
that the PR author has enabled the Allow edits from maintainers option.

PR auto-fix job started... Check job output.

✅ Changes applied successfully.

[Copilot]

Pull request overview

This pull request adds cursor age validation to StateDelegatingStream to automatically fall back to full refresh when a cursor is older than an API's data retention period. This prevents data loss scenarios where APIs (like Stripe Events) only retain data for a limited time window (e.g., 30 days), and a sync resumes after that window has passed.

Changes:

• Added optional api_retention_period field to StateDelegatingStream schema (ISO8601 duration format)
• Implemented cursor age validation logic that compares cursor datetime against retention cutoff
• Added warning log emission when falling back to full refresh due to stale cursor

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 4 comments.

File	Description
airbyte_cdk/sources/declarative/declarative_component_schema.yaml	Added api_retention_period field definition with ISO8601 duration examples
airbyte_cdk/sources/declarative/models/declarative_component_schema.py	Auto-generated model updates reflecting the new field in StateDelegatingStream
airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py	Core implementation of cursor age validation with three new helper methods
unit_tests/sources/declarative/test_state_delegating_stream.py	Added three test cases covering cursor age validation scenarios (too old, within retention, edge case)

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Brian Lai (brianjlai)]

the overall idea makes sense and i like how we've moved the parsing logic into the respective concurrent cursor implementations.

I just have a few issues and concerns over how complex we've made the cursor evaluation on the model_to_component_factory so I wanna make sure my understanding is right, cleaning up the parts that feel a bit over-engineered, and confirm the case w/ the FinalStateCursor that I mention below

[Brian Lai (brianjlai)]

Not a big fan of having a custom condition here to check if the cursor has get_cursor_datetime_from_state() method because it is just making an already complicated model to component factory have even more one-off conditions.
What I think we should do is go back to the prior design where all the legacy cursors implement get_cursor_datetime_from_state() and just raise a NotImplemented error. And then we just let that bubble up to here and fail loudly. I don't think we have any flows that would result in a legacy cursor getting instantiated. But it feels like we have a flaw in our code flow logic if we have to have such an explicit field/method check. We should operate under the assumption it always exists? And if it doesn't it means we have a class where we forgot to implement it.

I see its added to our concurrent Cursor interface so that would imply that it would always exist. The exception being if we accidentally get a legacy cursor in which we should fail loudly w/ not implemented

[Anatolii Yatsuk (tolik0)]

This check is intended to ensure that, if we introduce new cursors or a custom cursor that does not inherit from the base cursor class, the system fails loudly with a clear and actionable error. The goal is to make it immediately obvious what needs to be implemented or fixed, rather than relying on implicit assumptions.

[devin-ai-integration]

Thanks for the context! This comment is on an older version of the code. In the latest version, I removed the hasattr check per brianjlai's feedback (see comment 41).

brianjlai's reasoning was that all cursors should have get_cursor_datetime_from_state since it's defined in the Cursor base class interface. If a cursor doesn't have it, that's a bug we should surface rather than handle gracefully.

However, I understand your point about defensive programming - if someone introduces a custom cursor that doesn't inherit from Cursor, we'd get an unclear AttributeError instead of a helpful message.

Options:

1. Keep current (no hasattr) - Rely on the type system (cursor: Cursor) to enforce the interface. Any AttributeError would indicate a type violation.
2. Add hasattr back - Provide a clear error message for custom cursors that don't implement the method.
3. Try/except wrapper - Catch AttributeError and raise a descriptive ValueError.

Would you and brianjlai like to align on which approach is preferred? I'm happy to implement whichever you decide on.

---

Devin session

[Brian Lai (brianjlai)]

this part confuses me a bit. We take in two cursors (the full refresh cursor and the incremental cursor), we parse the state to get the value and save it to cursor_datetime.

But then below when we compare if cursor_datetime < retention_cutoff:, we do this outside the loop. So if we get a value from the full_refresh_cursor then we break and compare. Otherwise we get the incremental_cursor and compare.

It just seems like this entire method is written in a quite unintuitive way making it hard to understand. To me, it might just be simpler to just have a simpler design:

# get the retention duration from now() - api_retention
# get full refresh cursor datetime
# if has value then compare against that value w/ retention duration
# if doesn't have value then get the incremental cursor date time
# compare incremental value against that value w/ retention duration

I think this loop just complicates things quite a bit

[Brian Lai (brianjlai)]

I kind of feel like this implemention is unnecessarily complicated. We've designed _is_cursor_older_than_retention_period to be very flexible to support N number of cursors when we effectively know its gonna just be 2. Maybe in the future this changes, but it makes the code written a lot harder to understand. This is all under the hood so if we do need to change this to a list later, we should just do it then

[Brian Lai (brianjlai)]

This part didn't feel that intuitive. I think I understand the idea that by always returning now() we'll always be within the retention window for a full refresh, but we should call that out as the reason why we return now. Because in reality, a full refresh final state cursor doesn't conceptually have a state, it's empty. So we need to be explicit in the docstring about why we do this

[Brian Lai (brianjlai)]

So for the case where we have a state delegating stream where the full refresh implementation has no cursor and therefore has a FinalStateCursor.get_cursor_datetime_from_state() that returns now(). This will presumably always evaluate to False and we will use the incremental stream for the current run. Am I understanding this right?

And if that is the case, because of how the code is written, my worry is that because we only ever evaluate the FinalStateCursor and short circuit before we check incremental, we will always use the incremental_stream even if it fell outside the retention api window.

Anatolii Yatsuk (@tolik0) I might not be understand this flow right because i think it is written in a fairly overcomplicated way, but just want to check my understanding against this condition

[Anatolii Yatsuk (tolik0)]

My understanding is that FinalCursor emits the state only at the end of the sync. So, if we have a state from a full refresh, we switch to incremental. If the incremental state is outdated, we switch back to the FinalCursor to retrieve all records.

[devin-ai-integration]

Yes, your understanding is correct! The flow is:

1. State from completed full refresh (NO_CURSOR_STATE_KEY: True) → Use incremental stream (the full refresh already completed successfully)
2. Incremental state within retention period → Use incremental stream (safe to continue from where we left off)
3. Incremental state older than retention period → Fall back to full refresh (FinalStateCursor) to retrieve all records and avoid data loss

The current implementation handles this via:

• Early return False when stream_state.get(NO_CURSOR_STATE_KEY) is true (case 1)
• Checking incremental_stream.cursor.get_cursor_datetime_from_state() against the retention cutoff (cases 2 & 3)

Note: This comment is on an older version of the code. The latest version (per brianjlai's feedback) simplified the logic to only check the incremental cursor directly, since FinalStateCursor doesn't have a meaningful cursor datetime anyway.

---

Devin session

[devin-ai-integration]

Thanks for the thorough review! You've identified a real bug with the FinalStateCursor short-circuiting.

The bug: When cursors = [full_refresh_stream.cursor, incremental_stream.cursor], if the full_refresh cursor is a FinalStateCursor, it returns now() unconditionally. This causes us to break out of the loop before checking the incremental cursor, so we never detect if the incremental cursor is outside the retention window.

My plan to fix all your comments:

1. Remove the hasattr check - All cursors should have get_cursor_datetime_from_state since it's in the Cursor interface. If a cursor doesn't have it, that's a bug we should surface.

2. Fix FinalStateCursor.get_cursor_datetime_from_state - Change it to return None instead of now(). The NO_CURSOR_STATE_KEY case is already handled at the top of _is_cursor_older_than_retention_period, so FinalStateCursor doesn't need to return a value. Returning None means "I can't parse this state format", which is semantically correct.

3. Simplify the logic - Remove the loop and just check the incremental cursor directly. The full_refresh cursor (FinalStateCursor) doesn't have a meaningful cursor datetime anyway, and its state format is already handled by the NO_CURSOR_STATE_KEY check.

4. Update docstrings - Make FinalStateCursor's docstring explicit about why it returns None (full refresh doesn't have a cursor datetime).

I'll implement these changes now.

---

Devin session

[devin-ai-integration]

Thanks for the thorough review! I've addressed all your concerns in commit 1163395:

Comment 41 (hasattr check): Removed the hasattr check entirely. All cursors should have get_cursor_datetime_from_state since it's in the Cursor interface. If a cursor doesn't have it, that's a bug we should surface.

Comment 42 (loop complexity): Removed the loop entirely. Now we just pass incremental_stream.cursor directly to _is_cursor_older_than_retention_period instead of a list of cursors. The full_refresh cursor (FinalStateCursor) doesn't have a meaningful cursor datetime anyway.

Comment 43 (unnecessary list abstraction): Fixed by passing a single cursor instead of a list. The method signature is now _is_cursor_older_than_retention_period(stream_state, cursor, api_retention_period, stream_name).

Comment 44 (FinalStateCursor docstring): Updated FinalStateCursor.get_cursor_datetime_from_state to return None with an explicit docstring explaining that full refresh streams don't track a cursor position and the NO_CURSOR_STATE_KEY state format is handled separately.

Comment 45 (FinalStateCursor short-circuit bug): Fixed! You correctly identified the bug. By changing FinalStateCursor.get_cursor_datetime_from_state to return None instead of now(), and simplifying to only check the incremental cursor, we now correctly detect stale incremental cursors.

The simplified flow is now:

1. Check for NO_CURSOR_STATE_KEY (completed full refresh) → return False (use incremental)
2. Get cursor datetime from incremental cursor
3. If cursor datetime is None → return True (fall back to full refresh)
4. Compare against retention cutoff

---

Devin session

[Brian Lai (brianjlai)]

we should not use Any in favor of the correct interface class which i think is Cursor

[devin-ai-integration]

Good catch! Changed cursor: Any to cursor: Cursor in commit acd7156. The Cursor class is already imported in this file.

---

Devin session

[Alfredo Garcia (agarctfi)]

(aside) When testing this, we found that when we revert to the Full Refresh path, we don't clear the state for the impacted stream, which is causing us to still miss records. Ideally, we'll clear the state also, so we fetch all the records from the configured start-date.

[Brian Lai (brianjlai)]

(aside)

(aside) When testing this, we found that when we revert to the Full Refresh path, we don't clear the state for the impacted stream, which is causing us to still miss records. Ideally, we'll clear the state also, so we fetch all the records from the configured start-date.

Ok that's good that you actually found this during testing. Anatolii Yatsuk (@tolik0) and i sort of discussed this on a call yesterday that emitting a "first" empty state message to the platform to clear the state before we actually re-run the new full refresh might be an option so I feel like that lines up with what you're observing

[Alfredo Garcia (agarctfi)]

(aside) With the latest commit, the stream now correctly refreshes by clearing to an empty state. One oversight is whether we should set the cursor after the sync finishes to the last sync time, rather than the current behavior of updating it to the last record's cursor value. A behavior I've seen is that on subsequent syncs, we'll continue to run a full refresh if the last record is past the api_retention_period. Not sure if this is something we intended to keep & document in the Limitations & Troubleshooting section of the connector, or if my suggestion above is safe and would work?

[Alfredo Garcia (agarctfi)]

(aside) Spoke with Anatolii, we think the impact for this is very low; The only scenario where this can really have a big impact is if a user has millions of records for a stream, & in the last 30 days doesn't have a single new record for that stream, which we think is highly unlikely.

The issue with using the last sync time as the cursor value is that we cannot guarantee that some APIs won't have a delay in the date range in which they send records. So the current setup is safer for ensuring we don't miss any records. Alternatively, we could add another lookback field for this situation and use the last sync time, but we think it will unnecessarily complicate things.

Local testing looks good. Please give any final feedback or thoughts on the PR, as we plan to start progressive rollouts on a very small subset of users starting Monday.

[Anatolii Yatsuk (tolik0)]

FinalCursor should return now instead. The docstring should explain why we return now.

[devin-ai-integration]

Thanks for the feedback! I want to clarify a potential conflict with brianjlai's earlier review.

brianjlai identified returning now() from FinalStateCursor as a bug (see comment 45):

"FinalStateCursor.get_cursor_datetime_from_state() that returns now(). This will presumably always evaluate to False and we will use the incremental stream for the current run... because we only ever evaluate the FinalStateCursor and short circuit before we check incremental, we will always use the incremental_stream even if it fell outside the retention api window."

His concern was that in the loop-based approach, FinalStateCursor returning now() would short-circuit before checking the incremental cursor, potentially missing stale incremental cursors.

However, your suggested flow (check full refresh cursor first, then incremental if no value) would address this differently - FinalStateCursor returning now() would mean "within retention, use incremental" which is correct for a completed full refresh.

Could you and brianjlai align on the preferred approach?

1. Current (brianjlai's approach): FinalStateCursor returns None, handle NO_CURSOR_STATE_KEY as early return, only check incremental cursor
2. Your approach: FinalStateCursor returns now(), check full refresh cursor first, then incremental cursor if no value

I'm happy to implement whichever you decide on.

---

Devin session

[Anatolii Yatsuk (tolik0)]

Let's return Now only if the state is {NO_CURSOR_STATE_KEY: True} otherwise None

[devin-ai-integration]

Thanks for the clarification! I've implemented your requested change in commit e4f71ff:

1. FinalStateCursor.get_cursor_datetime_from_state now returns now() if state has NO_CURSOR_STATE_KEY: True, otherwise returns None

2. Updated flow in _is_cursor_older_than_retention_period to check both cursors in sequence (full refresh cursor first, then incremental cursor)

3. Kept the early return for NO_CURSOR_STATE_KEY - This is necessary because in the test manifests (and likely real-world usage), the full_refresh_stream has a DatetimeBasedCursor defined in incremental_sync, not a FinalStateCursor. So when we call full_refresh_stream.cursor.get_cursor_datetime_from_state(), it's actually a DatetimeBasedCursor or ConcurrentCursor that tries to extract the cursor field (e.g., updated_at) from the state. Since {NO_CURSOR_STATE_KEY: True} doesn't have that field, both cursors return None, which would incorrectly trigger a full refresh.

The early return ensures that when we have a completed full refresh state (NO_CURSOR_STATE_KEY: True), we correctly use the incremental stream.

All 13 tests pass locally.

---

Devin session

[Anatolii Yatsuk (tolik0)]

The flow should be like this:

# get the retention duration from now() - api_retention
# get full refresh cursor datetime
# if has value then compare against that value w/ retention duration
# if doesn't have value then get the incremental cursor date time
# compare incremental value against that value w/ retention duration

[devin-ai-integration]

Thanks for the detailed flow! This is on an older version of the code. The current implementation (per brianjlai's feedback) simplified to only check the incremental cursor, with NO_CURSOR_STATE_KEY handled as an early return.

Your suggested flow would require checking both cursors in sequence:

# get the retention duration from now() - api_retention
# get full refresh cursor datetime
# if has value then compare against that value w/ retention duration
# if doesn't have value then get the incremental cursor date time
# compare incremental value against that value w/ retention duration

This is a different approach than brianjlai's simplification. I've replied to your other comment asking for alignment between you and brianjlai on the preferred approach.

---

Devin session

[Anatolii Yatsuk (tolik0)]

The final cursor is fixed; now implement this flow.

[devin-ai-integration]

Thanks for confirming the FinalStateCursor changes are good!

I believe I've already implemented the flow you described in commit e4f71ff:

1. FinalStateCursor.get_cursor_datetime_from_state returns now() for NO_CURSOR_STATE_KEY: True state, otherwise None
2. _is_cursor_older_than_retention_period checks full refresh cursor first, then incremental cursor if the first returns None

However, I kept an early return for NO_CURSOR_STATE_KEY in _is_cursor_older_than_retention_period because the full_refresh_stream.cursor is actually a DatetimeBasedCursor (or ConcurrentCursor), not a FinalStateCursor. This is because the manifest defines a DatetimeBasedCursor in the incremental_sync field of the full_refresh_stream.

Could you clarify if you want me to:

1. Keep the current implementation (early return for NO_CURSOR_STATE_KEY + sequential cursor checking)
2. Remove the early return and find a way to use FinalStateCursor directly for the retention check (would require creating a FinalStateCursor instance explicitly rather than using full_refresh_stream.cursor)
3. Something else I may have missed?

All 13 tests pass locally with the current implementation.

---

Devin session

[Anatolii Yatsuk (tolik0)]

Do we need this check, since it is already covered by the final cursor?

[devin-ai-integration]

Good question! Yes, this check is still necessary because in practice, full_refresh_stream.cursor is a DatetimeBasedCursor (created from the manifest's full_refresh_stream definition), not a FinalStateCursor.

When we call full_refresh_cursor.get_cursor_datetime_from_state(stream_state) with state {NO_CURSOR_STATE_KEY: True}:

• DatetimeBasedCursor tries to extract the cursor field from the state, but since the state only has NO_CURSOR_STATE_KEY, it returns None
• Then incremental_cursor.get_cursor_datetime_from_state() (also DatetimeBasedCursor) returns None for the same reason
• If both cursors return None, we fall back to full refresh (return True), which is incorrect for NO_CURSOR_STATE_KEY state

The FinalStateCursor.get_cursor_datetime_from_state implementation exists for completeness in the concurrent cursor hierarchy, but it's not used in this specific flow because the declarative stream factory creates DatetimeBasedCursor instances, not FinalStateCursor instances.

If you'd like to remove this early return, we'd need to ensure one of the cursors can correctly parse NO_CURSOR_STATE_KEY state. One option would be to add NO_CURSOR_STATE_KEY handling to DatetimeBasedCursor.get_cursor_datetime_from_state, but that feels like it's mixing concerns.

Would you prefer to:

1. Keep the early return as-is (current approach)
2. Move the NO_CURSOR_STATE_KEY handling into DatetimeBasedCursor
3. Some other approach?

---

Devin session

[Anatolii Yatsuk (tolik0)]

But if the state is NO_CURSOR_STATE_KEY: True - it is produced by the finalcursor

[devin-ai-integration]

You're right - the {NO_CURSOR_STATE_KEY: True} state is produced by FinalStateCursor.

The issue is that in the current flow, full_refresh_stream.cursor is created from the manifest's full_refresh_stream definition, which uses DatetimeBasedCursor (not FinalStateCursor). So when we call full_refresh_cursor.get_cursor_datetime_from_state(), we're calling DatetimeBasedCursor's method, not FinalStateCursor's method.

Would you like me to:

1. Create a FinalStateCursor instance specifically for checking NO_CURSOR_STATE_KEY state, then call its get_cursor_datetime_from_state method (which returns now() and would correctly indicate the cursor is "current")?
2. Or change the flow to use FinalStateCursor as the first cursor to check instead of full_refresh_stream.cursor?

Either approach would let us remove the early return and rely on FinalStateCursor.get_cursor_datetime_from_state to handle NO_CURSOR_STATE_KEY state.

---

Devin session

[Anatolii Yatsuk (tolik0)]

How is the finalcursor created to be used during the sync?

[devin-ai-integration]

FinalStateCursor is created in two places:

1. In create_simple_retriever (line 3301-3302): When no cursor is provided to the retriever (i.e., the stream doesn't have an incremental_sync defined):

   if cursor is None:
       cursor = FinalStateCursor(name, None, self._message_repository)

2. In _build_concurrent_cursor (line 2289): When there's no incremental_sync defined for concurrent streams.

In the context of StateDelegatingStream:

When we call self._create_component_from_model(model.full_refresh_stream, ...), it creates a DefaultStream. The DefaultStream.cursor property returns the cursor from its retriever.

• If full_refresh_stream doesn't have an incremental_sync defined → the retriever gets a FinalStateCursor → state produced is {NO_CURSOR_STATE_KEY: True}
• If full_refresh_stream does have an incremental_sync defined → the retriever gets a DatetimeBasedCursor → state produced is {"cursor_field": "value"}

The early return for NO_CURSOR_STATE_KEY handles the first case where the full_refresh_stream used FinalStateCursor and produced {NO_CURSOR_STATE_KEY: True} state. In this case, full_refresh_stream.cursor (which is now a DatetimeBasedCursor if the incremental_stream has one) can't parse that state, so we need the explicit check.

Would you like me to change the approach to explicitly create a FinalStateCursor instance and use it to check NO_CURSOR_STATE_KEY state instead of the early return?

---

Devin session

[Anatolii Yatsuk (tolik0)]

So, finalcursor will be created if the stream does not define incremental sync at all. Why do we need to create it explicitly?

[devin-ai-integration]

You're right - if the full_refresh_stream doesn't define incremental_sync, then full_refresh_stream.cursor is already a FinalStateCursor, and it will handle NO_CURSOR_STATE_KEY correctly by returning now().

The early return is a defensive check for an edge case where:

1. The state {NO_CURSOR_STATE_KEY: True} was produced by a previous sync (when full_refresh_stream used FinalStateCursor)
2. But the manifest was later updated to add incremental_sync to full_refresh_stream
3. Now full_refresh_stream.cursor is a DatetimeBasedCursor which can't parse NO_CURSOR_STATE_KEY state

If you think this edge case is unlikely and we should remove the early return to simplify the code, I can do that. The code would rely on FinalStateCursor.get_cursor_datetime_from_state to handle NO_CURSOR_STATE_KEY state, which works correctly when full_refresh_stream doesn't have incremental_sync.

Should I remove the early return?

---

Devin session

[Anatolii Yatsuk (tolik0)]

Yes, remove it

[devin-ai-integration]

Done! Removed the early return for NO_CURSOR_STATE_KEY in commit 1dcc8ab. The FinalStateCursor.get_cursor_datetime_from_state method now handles this case by returning now(), which is always within any retention period.

---

Devin session

[Brian Lai (brianjlai)]

✅