[devin-ai-integration]

Overview

👉 TL;DR: Lets low-code connectors sync incrementally from APIs that hand back a "sync token" at the end of a read, instead of filtering by a timestamp. The token is saved only after every record from that read has been emitted, and is sent back on the next sync to fetch only what changed. Specifically, this adds the SyncTokenCursor incremental type, SyncTokenRequestOptionsProvider, and a page_observer hook (PageCompleted) on SimpleRetriever.

This PR ships on its own for source-ashby, once the source-ashby adoption (airbytehq/airbyte#89675) drops its RESET_SYNC_TOKEN filter. The follow-ups are in draft: #1235 (RESET_SYNC_TOKEN recovery) and #1238 (RequestPath tokens and first-request checks).

Part of https://github.com/airbytehq/airbyte-internal-issues/issues/17213. Close #17213 only after the whole stack ships and its acceptance criteria are checked off.

Merge gate

Keep this PR in draft until all of these are done:

• Record live validation against a real Ashby account, including whether a token covers records edited during the pass that issued it.

Changes

• Schema and models: new SyncTokenCursor definition with token_path, token_option, completion_condition, initial_token and cursor_field, added to DeclarativeStream.incremental_sync.anyOf.
• SyncTokenCursor (incremental/sync_token_cursor.py):
  • Captures the token from the response envelope on the terminal page of a pass, holds it as pending, and commits it in close_partition, after the pass's records.
  • Failed or partial passes never change committed state.
  • State is {"state_type": "sync_token", "token": "..."}, or a states list per partition, capped at 25,000 partitions.
  • State from another cursor type is ignored with a WARN, and the stream bootstraps with a full read.
  • Malformed sync_token state fails the stream with a system_error that asks for a refresh, and is left in place.
• SyncTokenRequestOptionsProvider: sends the stored token, else initial_token, else nothing, through token_option. Wrapped in PerPartitionRequestOptionsProvider for partitioned streams.
• SimpleRetriever: new page_observer hook, called with PageCompleted(stream_slice, response, pagination_exhausted) after each page. Bound only for SyncTokenCursor; None everywhere else.
• DefaultStream: new source_defined_incremental flag, so token streams advertise incremental with a source-defined cursor. ConcurrentPerPartitionCursor gains a public partition_router property used by the same lookup.
• ModelToComponentFactory:
  • wires the cursor, the request options provider and the page observer
  • rejects unsupported combinations at build time, looking inside nested UnionPartitionRouter and GroupingPartitionRouter at any depth
  • rejects streaming decoders and api_retention_period

Example (Ashby):

incremental_sync:
  type: SyncTokenCursor
  token_path: ["syncToken"]
  token_option: {type: RequestOption, inject_into: body_json, field_name: syncToken}
  cursor_field: updatedAt

Recommended review order

Tests are left out. This PR doesn't change the entrypoint or the source (ConcurrentDeclarativeSource); the new type is reached through the schema and the factory.

1. Declarative component schema
   • declarative_component_schema.yaml#L1431-L1493: the SyncTokenCursor definition
   • declarative_component_schema.yaml#L1911: added to incremental_sync.anyOf
   • models/declarative_component_schema.py#L2976: generated model
2. Model to component factory (parsers/model_to_component_factory.py)
   • #L2185-L2200 and #L2318: request options provider and source_defined_incremental in create_default_stream
   • #L2322-L2463: _flatten_partition_router_models and _validate_sync_token_stream
   • #L2584-L2592 and #L2644-L2688: _build_concurrent_cursor branch and create_sync_token_cursor
   • #L4235-L4237: page_observer binding in create_simple_retriever
   • #L4852-L4857: api_retention_period rejection
3. Runtime components
   • incremental/sync_token_cursor.py: start at on_page_completed, then close_partition and _load_state
   • requesters/request_options/sync_token_request_options_provider.py: token injection
   • retrievers/simple_retriever.py#L84-L90 and #L562-L569: PageCompleted and the hook call site
4. Other
   • streams/concurrent/default_stream.py#L89-L93 and #L127: discovery flags and partition-router lookup
   • incremental/concurrent_partition_cursor.py#L201-L203: new partition_router property
   • incremental/__init__.py, requesters/request_options/__init__.py: exports

Breaking change

None. The component is additive and opt-in. When it isn't used, page_observer is None and source_defined_incremental is False, so existing cursors, streams and the retriever behave as before.

Risks

• A token is opaque, so the CDK can't check whether it covers records edited during the pass that issued it. That's API-specific and is a merge-gate item above.
• There is no automatic recovery from a rejected token until feat(declarative): add RESET_SYNC_TOKEN recovery for SyncTokenCursor (2/3) #1235. Map the API's expired or invalid token errors to FAIL with a refresh instruction, never to IGNORE: an ignored error without a new token keeps the stored token, so every later sync reads nothing, and an ignored 2xx response with a token at token_path commits that token.

Test plan

• unit_tests/sources/declarative/incremental/test_sync_token_cursor*.py: cursor state rules, schema, discovery and build-time validation, nested partition routers, plus end-to-end reads through ConcurrentDeclarativeSource.read with HttpMocker that assert exact request bodies and emitted state.
• The end-to-end tests fail on unmodified main (schema validation error) and pass with this change.
• poetry run pytest unit_tests/sources/declarative/incremental/ unit_tests/sources/declarative/parsers/test_model_to_component_factory.py: 509 passed. Ruff and MyPy pass.

Design notes: why a new cursor type

Show/Hide Content

DatetimeBasedCursor and IncrementingCountCursor both derive state from record values. No existing declarative component can capture a value from the response envelope and commit it only when the pass completes. SyncTokenCursor is built in and declarative, so adopting connectors need no custom Python components.

Link to Devin session: https://app.devin.ai/sessions/94b41efe75db43e78eb5a4f8d5f58484
Open in Devin Desktop: https://app.devin.ai/desktop/session/94b41efe75db43e78eb5a4f8d5f58484?variant=devin

[devin-ai-integration]

I'll fix CI failures and address comments from users with write access. I'll skip comments containing "(aside)".

• Disable automatic comment, CI, and merge conflict monitoring

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You can test this version of the CDK using the following:

# Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@devin/1792189821-opaque-token-cursor#egg=airbyte-python-cdk[dev]' --help

# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch devin/1792189821-opaque-token-cursor

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[github-actions]

PyTest Results (Fast)

6 075 tests  +121   6 044 ✅ +121   5m 52s ⏱️ +22s
    1 suites ±  0      31 💤 ±  0 
    1 files   ±  0       0 ❌ ±  0 

Results for commit e954929. ± Comparison against base commit d5dd299.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

6 078 tests   6 047 ✅  9m 47s ⏱️
    1 suites     31 💤
    1 files        0 ❌

Results for commit e954929.

♻️ This comment has been updated with latest results.

[devin-ai-integration]

👍 On it. I'm applying findings 1–12 on this branch as new commits on top of a merge of main (no rebase or force-push):

• the patch for findings 1, 2, 5 and 6, as written;
• both proposed test files;
• the T-FAIL, T-PAR and T-BUILDER test fixes (7);
• the four edge cases in 8;
• docstrings (9);
• the P3 cleanups (10–12).

No change for 13. After that I'll merge the updated branch into #1235. For finding 4, I'll change Resolves to Part of, retitle the PR, and add the merge gate to the description. The PR stays in draft.

Written by Devin

[Patrick Nilan (pnilan)]

🤖 This comment was generated by an AI Agent.

Review follow-up: malformed saved state starts a silent full read

Reviewed at 2332d29c. This came up while reviewing #1235, but the code is in this PR. It is the only new finding for #1234.

P1: malformed sync_token state is treated as "no state"

Where: _load_state, sync_token_cursor.py lines 377–420

Problem:

• When _load_state can't use the saved state, it logs one INFO line and treats the stream as never synced. The provider sends no token, the stream reads everything, and the bootstrap commits a new token over the old state.
• No RESET_SYNC_TOKEN filter is involved. Each of these produces one tokenless request, a COMPLETE stream and a new token:
  • {"state_type": "sync_token", "token": {"bad": "T1"}}
  • {"state_type": "sync_token", "token": 123}
  • {"state_type": "sync_token", "token": ""}
  • {"state_type": "sync_token", "token": null}
  • {"state_type": "sync_token", "states": []} on an unpartitioned stream
  • on a partitioned stream, an entry whose token isn't a non-empty string (only that partition reads in full)
• This cursor never writes any of these shapes, so each one means a bug or a hand edit. Today each becomes an unannounced full read. For Append connections, that also means a second copy of every record.

Requested change:

1. Fail on malformed state that claims to be this cursor's (state_type: sync_token). That covers a token that isn't a non-empty string, a shape that doesn't match the stream's partitioning, and a malformed partition entry.
   • Raise from stream_slices(), not __init__, so only this stream fails instead of the whole sync.
   • Use system_error, as _coerce_token does for a bad token_path value, with a message that tells the user to refresh the stream.
2. Don't overwrite the malformed state when the stream fails. _on_stream_is_done calls ensure_at_least_one_state_emitted() for failed streams too. Unless that changes, the bad state is replaced with {"state_type": "sync_token"} and the next sync bootstraps silently anyway. Skip the emission (or re-emit the incoming state unchanged) when loading failed.
3. Keep bootstrapping these:
   • no state, {}, and {"state_type": "sync_token"}, which ensure_at_least_one_state_emitted writes before the first commit;
   • "states": [] on a partitioned stream, and partitions with no entry yet;
   • state from another cursor type: no state_type, or a different value such as date-range. That is the planned migration path, for example Ashby applications moving off its datetime cursor. Other cursors write their own state_type values, so this cursor can't tell a foreign value from a corrupt one. Log it at WARN instead of INFO, because it costs a full read.
4. test_partition_state_load_skips_only_malformed_entries asserts the old skip-and-bootstrap behavior. Change it to expect the failure.

Regression tests to append to test_sync_token_cursor.py (11 fail at 2332d29c; the 4 empty-state controls pass)

@pytest.mark.parametrize(
    "state",
    [
        pytest.param({"state_type": "sync_token", "token": {"bad": "T1"}}, id="object-token"),
        pytest.param({"state_type": "sync_token", "token": 123}, id="integer-token"),
        pytest.param({"state_type": "sync_token", "token": ""}, id="empty-token"),
        pytest.param({"state_type": "sync_token", "token": None}, id="null-token"),
        pytest.param({"state_type": "sync_token", "states": []}, id="partitioned-shape"),
    ],
)
def test_malformed_sync_token_state_fails_the_stream_instead_of_reading_everything(state):
    cursor = _cursor(state=state)

    with pytest.raises(AirbyteTracedException):
        list(cursor.stream_slices())


@pytest.mark.parametrize(
    "state",
    [
        pytest.param({"state_type": "sync_token", "token": "T1"}, id="unpartitioned-shape"),
        pytest.param(
            {
                "state_type": "sync_token",
                "states": [{"partition": {"account": "a"}, "cursor": {"token": 7}}],
            },
            id="integer-token-entry",
        ),
    ],
)
def test_malformed_partitioned_sync_token_state_fails_the_stream(state):
    router = Router([StreamSlice(partition={"account": "a"}, cursor_slice={})])
    cursor = _cursor(state=state, router=router)

    with pytest.raises(AirbyteTracedException):
        list(cursor.stream_slices())


def test_malformed_sync_token_state_is_not_overwritten_when_the_stream_fails():
    state = {"state_type": "sync_token", "token": 123}
    cursor = _cursor(state=state)
    with pytest.raises(AirbyteTracedException):
        list(cursor.stream_slices())

    # ConcurrentReadProcessor._on_stream_is_done calls this for failed streams too.
    cursor.ensure_at_least_one_state_emitted()

    emitted = [
        message.state.stream.stream_state.__dict__
        for message in cursor._message_repository.consume_queue()
    ]
    assert emitted in ([], [state])


@pytest.mark.parametrize(
    "state",
    [
        pytest.param(None, id="no-state"),
        pytest.param({}, id="empty-state"),
        pytest.param({"state_type": "sync_token"}, id="nothing-committed-yet"),
    ],
)
def test_empty_state_still_bootstraps(state):
    cursor = _cursor(state=state)

    assert [stream_slice.cursor_slice for stream_slice in cursor.stream_slices()] == [{}]


def test_empty_partitioned_state_still_bootstraps():
    router = Router([StreamSlice(partition={"account": "a"}, cursor_slice={})])
    cursor = _cursor(state={"state_type": "sync_token", "states": []}, router=router)

    assert [stream_slice.cursor_slice for stream_slice in cursor.stream_slices()] == [{}]


@pytest.mark.parametrize(
    "state",
    [
        pytest.param({"updated_at": "2026-10-01T00:00:00Z"}, id="datetime-cursor"),
        pytest.param({"state_type": "date-range", "slices": []}, id="concurrent-datetime-cursor"),
        pytest.param({"state_type": "wrong_type", "token": "T1"}, id="unknown-state-type"),
    ],
)
def test_state_from_another_cursor_type_bootstraps_with_a_warning(state, caplog):
    caplog.set_level(logging.INFO)
    cursor = _cursor(state=state)

    assert [stream_slice.cursor_slice for stream_slice in cursor.stream_slices()] == [{}]
    assert any(record.levelno == logging.WARNING for record in caplog.records)

Reference patch (with it, all 15 tests above pass, as do 98 of the 99 existing sync-token tests; the one failure is the test in item 4. Ruff and mypy are clean.)

diff --git a/airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py b/airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py
index 83051a45e..81c2c4175 100644
--- a/airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py
+++ b/airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py
@@ -97,6 +97,7 @@ class SyncTokenCursor(Cursor):
         self._warned_no_token = False
         self._warned_cap = False
         self._partitions_over_cap: set[str] = set()
+        self._invalid_state_reason: Optional[str] = None
         self._load_state(stream_state or {})
 
     @property
@@ -131,6 +132,14 @@ class SyncTokenCursor(Cursor):
             }
 
     def stream_slices(self) -> Iterable[StreamSlice]:
+        if self._invalid_state_reason is not None:
+            message = (
+                f"Stream {self._stream_name}: the saved sync token state is malformed "
+                f"({self._invalid_state_reason}). Refresh the stream to start a new full sync."
+            )
+            raise AirbyteTracedException(
+                message=message, internal_message=message, failure_type=FailureType.system_error
+            )
         seen_partitions: set[str] = set()
         for stream_slice in self._partition_router.stream_slices():
             partition = dict(stream_slice.partition or {})
@@ -256,7 +265,10 @@ class SyncTokenCursor(Cursor):
 
     def _log_foreign_state(self) -> None:
         self._state_is_foreign = True
-        self._log_bootstrap("foreign state")
+        logger.warning(
+            "Stream %s: the saved state belongs to another cursor type; reading all records once.",
+            self._stream_name,
+        )
 
     def _log_replay(self, partition: Mapping[str, Any], committed: Mapping[str, Any]) -> None:
         age = self._human_age(committed.get("committed_at"))
@@ -357,6 +369,8 @@ class SyncTokenCursor(Cursor):
         self._state_emitted = True
 
     def ensure_at_least_one_state_emitted(self) -> None:
+        if self._invalid_state_reason is not None:
+            return  # leave the malformed state in place for diagnosis
         if self._partitioned or not self._state_emitted:
             self._emit_state_message()
         if self._partitions_over_cap and not self._warned_cap:
@@ -384,13 +398,13 @@ class SyncTokenCursor(Cursor):
         if self._partitioned:
             states = stream_state.get("states")
             if not isinstance(states, list) or "token" in stream_state:
-                self._log_foreign_state()
+                self._invalid_state_reason = "unpartitioned shape on a partitioned stream"
                 return
             committed: dict[str, dict[str, Any]] = {}
             for state in states:
                 if not isinstance(state, Mapping):
-                    self._log_bootstrap("foreign state")
-                    continue
+                    self._invalid_state_reason = "a partition entry is not an object"
+                    return
                 partition = state.get("partition")
                 cursor = state.get("cursor")
                 if (
@@ -404,11 +418,12 @@ class SyncTokenCursor(Cursor):
                         "committed_at": cursor.get("committed_at", ""),
                     }
                 else:
-                    self._log_bootstrap("foreign state")
+                    self._invalid_state_reason = "a partition entry has no valid token"
+                    return
             self._committed.update(committed)
         else:
             if "states" in stream_state:
-                self._log_foreign_state()
+                self._invalid_state_reason = "partitioned shape on an unpartitioned stream"
                 return
             token = stream_state.get("token")
             if isinstance(token, str) and token:
@@ -416,5 +431,5 @@ class SyncTokenCursor(Cursor):
                     "token": token,
                     "committed_at": stream_state.get("committed_at", ""),
                 }
-            elif token is not None:
-                self._log_foreign_state()
+            elif "token" in stream_state:
+                self._invalid_state_reason = "the token is not a non-empty string"

[devin-ai-integration]

Re: review follow-up on malformed saved state

☑️ Resolved in 4050495. Malformed state_type: sync_token state now fails the affected stream from stream_slices() with a system_error that tells the user to refresh. ensure_at_least_one_state_emitted() no longer overwrites that state.

• I applied your patch and all 15 regression tests. test_partition_state_load_skips_only_malformed_entries now expects the failure.
• I also added an end-to-end test. Incoming state {"state_type": "sync_token", "token": 123} ends with a system_error trace, sends no HTTP requests, and leaves the emitted state equal to the incoming state.
• One deviation: {"state_type": "sync_token"} with neither token nor states still bootstraps, on both partitioned and unpartitioned streams. That is the state an unpartitioned stream writes before its first commit, and a connector may later add a partition router. A parametrized test covers it.
• Foreign state (a different or missing state_type) now logs at WARN.
• All required checks pass. The only failure is the optional destination-motherduck check, which fails on a missing CI secret.

---

Posted by Devin, as requested by Patrick Nilan (@pnilan).

[Baz (bazarnov)]

Approving, with one condition under Before merge. SyncTokenCursor is opt-in and inert for every existing stream, and I found nothing else blocking.

Verified

• Fleet: built and discovered all 540 declarative manifests from airbytehq/airbyte master (5,448 streams, with and without incoming state) on main and on this branch. Catalogs, errors and warnings are identical.
• Runtime: a token commits only after its partition's records (stress runs with failing and vanishing partitions and a bounded queue). Malformed and foreign state behave as described end to end, and the token reaches requests through all four inject_into targets, including on partitioned streams.
• source-ashby without #1235: with its RESET_SYNC_TOKEN filter removed, the #89675 manifest builds on this branch. Mocked reads cover a cold read, a warm read, an expired token (the stream fails and keeps its stored token) and the interview_plans parent.
• Platform contract: source_defined_cursor: true with default_cursor_field: [] is accepted for Incremental Append and Append + Deduped (cursor validation, default sync modes). Bulk CDK destinations order by emittedAt when the cursor is empty (DestinationStream.kt).
• CI on e954929: all checks pass except Check: destination-motherduck, which fails at secret fetch (SECRET_DESTINATION_MOTHERDUCK__MOTHERDUCK__CREDS has no enabled version), as on other CDK PRs. It is unrelated to this change.
• Breaking change: none. The schema only gains an anyOf branch. Every behavior change in shared code is gated on SyncTokenCursor (page_observer stays None, source_defined_incremental stays False), and the other shared edits are refactors with no behavior change, which the fleet run confirms.

Pushed in e954929, on top of a main merge

• Removed the hand-added interpolation_context from the generated SyncTokenCursor model. The generator only emits deprecated and deprecation_message (generator).
• Added a read test for a token stream used as a parent without incremental_dependency, the shape source-ashby uses for interview_stages (#89675), and more malformed partitioned-state cases. Each new case fails on a mutant of the line it guards.
• Dropped two tests whose names said malformed shapes are ignored. The malformed-state tests cover the same shapes, expect the failure and check that nothing is loaded.
• Updated the description for the standalone release.

Before merge

• Patrick Nilan (@pnilan), the merge gate is checked, but the only live-validation record I found (#89675) says a warm read that returns a changed record was not produced. Please link the mid-pass edit result on airbytehq/airbyte-internal-issues#17213. If edits made during a pass are missed, #17213 calls for the overlap option before release.

Notes for follow-ups (aside), no changes needed in this PR

• With #1235 deferred, a rejected token is not recovered automatically, and Ashby tokens are valid for 14 days (Ashby docs). The Ashby adoption has to drop RESET_SYNC_TOKEN, which this schema does not accept (#89675), along with the docs and AGENTS.md lines that promise a full re-read on these errors. The existing success_false filter then fails the stream with a system_error and no refresh instruction (manifest). A dedicated FAIL filter with failure_type: config_error for sync_token_expired, sync_token_invalid and incremental_sync_too_large that asks for a refresh would be clearer. Never IGNORE them: the stored token is kept, so every later sync reads nothing.
• #1238 says its new checks apply to an unreleased cursor. Once this ships, they tighten a published contract and need a breaking-change look.
• incremental-syncs.md documents only DatetimeBasedCursor so far (the docs step in #17213).

[Patrick Nilan (pnilan)]

/prerelease

Prerelease Job Info

This job triggers the publish workflow with default arguments to create a prerelease.

Prerelease job started... Check job output.

❌ Failed to trigger prerelease workflow.
Prerelease Job Info

This job triggers the publish workflow with default arguments to create a prerelease.

Prerelease job started... Check job output.

✅ Prerelease workflow triggered successfully.

View the publish workflow run: https://github.com/airbytehq/airbyte-python-cdk/actions/runs/40907784392

[Patrick Nilan (pnilan)]

Sync-token follow-up read: ✅ the API accepted all 10 injected tokens and returned only the deltas. I seeded test 1's final committed state into a single-version READ of the same connection on the #89675 build, about 3 hours after test 1. The read exited cleanly, all 12 streams reached COMPLETE, and there were no WARN/ERROR logs, HTTP 4xx responses or token rejections.

Tokens used: all 10 sync-token streams logged replaying the sync token committed 2 hours 58 minutes ago.
Records: 440 in total, against 82,945 on the first read. The sync-token streams returned only changed rows: candidates 24 (vs 78,668), interview_schedules 8 (vs 2,229), users 2 (vs 233), and 0 for candidate_tags, custom_fields, departments, feedback_form_definitions, interviews, locations and offers. The two full-refresh streams returned everything, as expected: sources 381 and archive_reasons 25.
New state: every sync-token stream committed a new token, committed_at 19:34 UTC, including the streams with 0 records.

I couldn't see the outgoing syncToken query parameter in the logs. The evidence is the connector's replay log lines plus the drop in record counts. Separately, the connector logs no stored sync token … reading all records for streams that aren't in the catalog (e.g. approvals, interview_plans). That's log noise only; those streams weren't read.

Overall: test 1 showed record parity on the first sync and the expected state and catalog changes. This follow-up shows the tokens are accepted on the next sync. Test 2 is still inconclusive because it timed out at 180 minutes during the control 2.0.1 read. I'm not doing anything more on this unless you want test 2 rerun with the heavy applications streams excluded.

Confirmed working. Can't inject state into regression tests so Devin ran a test read with output cursors from previous regression test run.