[devin-ai-integration]

Summary

Adds SyncTokenCursor, a declarative incremental sync type for APIs that return an opaque, server-issued sync token in the response envelope (Ashby syncToken, Google/Todoist sync tokens, Box stream positions, Graph/Dataverse delta tokens). This PR adds capture and commit. It is the first of three stacked PRs: #1235 adds RESET_SYNC_TOKEN recovery and #1238 adds RequestPath tokens.

Part of https://github.com/airbytehq/airbyte-internal-issues/issues/17213:

• https://github.com/airbytehq/airbyte-internal-issues/issues/17213

Close on merge: no. Close #17213 once its acceptance criteria are checked off after the whole stack ships.

Merge gate

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

• Merge together with feat(declarative): add RESET_SYNC_TOKEN recovery for SyncTokenCursor (2/2) #1235. Without it, a rejected token is never recovered from.
• Settle the release order with feat(declarative): support RequestPath tokens and first-request checks in SyncTokenCursor #1238, whose new checks assume this cursor is unreleased.
• Record live validation against a real Ashby account, including whether a token covers records edited during the pass that issued it.
• Run the Google Calendar (feat(source-google-calendar): sync-token incremental for events via SyncTokenCursor (preview SDM) airbyte#89078) and Todoist (feat(source-todoist): incremental tasks and projects via Sync API SyncTokenCursor (preview SDM) airbyte#89077) drafts on a preview SDM. They inject the token through request_parameter and body_data.

Changes

• Schema / models: new SyncTokenCursor with five settings: token_path, token_option, completion_condition, initial_token, cursor_field. It is added to DeclarativeStream.incremental_sync.anyOf. The definition uses additionalProperties: true so that stream-level $parameters still validate.
• SyncTokenCursor (incremental/sync_token_cursor.py):
  • Unpartitioned state is {"state_type": "sync_token", "token": "<token>"}.
  • Keeps per-partition committed and pending token maps, keyed with PerPartitionKeySerializer.
  • A token becomes pending only on a terminal page that satisfies completion_condition and carries a valid string or int token. It is committed in close_partition, so state is queued after every record of the traversal.
  • Failed or partial traversals never change committed state.
  • Partition handling: 25,000-partition cap with no eviction. Tokens of vanished partitions are dropped only after the router enumerates successfully. Partitioned emission is throttled at 600 s, with a final unthrottled emission.
  • Foreign or mis-shaped state is ignored (INFO) and the stream bootstraps.
  • Log messages follow the issue's observability table.
• SyncTokenRequestOptionsProvider: sends the stored token, otherwise initial_token, otherwise nothing, through token_option. For partitioned streams it is wrapped in PerPartitionRequestOptionsProvider.
• SimpleRetriever: new page_observer hook (PageCompleted(stream_slice, response, pagination_exhausted)). It is called after a page's records are yielded and after the next page token is computed. It is bound only for SyncTokenCursor and is None (inert) for every other stream.
• DefaultStream: new source_defined_incremental flag. A token stream advertises [full_refresh, incremental], source_defined_cursor: true, is_resumable: true, and default_cursor_field: [] (or [cursor_field]). It never advertises null.
• Factory: cursor and provider wiring, plus every validation in the issue's table:
  • datetime-only keys
  • GroupingPartitionRouter
  • AsyncRetriever
  • lazy parents
  • incremental_dependency in either direction
  • a partition field named token
  • pagination_reset.limits
  • property chunking
  • a CustomRetriever warning

Example (Ashby):

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

Reproduction (before the change)

On main (v7.37.0), a manifest whose stream uses incremental_sync: {type: SyncTokenCursor, ...} fails ConcurrentDeclarativeSource(source_config=manifest) with jsonschema.exceptions.ValidationError: Validation against json schema defined in declarative_component_schema.yaml schema failed. incremental_sync only accepts DatetimeBasedCursor and IncrementingCountCursor.

Declarative-First Evaluation

This is a CDK feature, not a connector change. DatetimeBasedCursor and IncrementingCountCursor both derive state from record values, and no existing declarative component can capture a value from the response envelope and commit it only when the traversal completes. The new component is built-in and declarative, so adopters need no custom Python components.

Test Coverage

• unit_tests/sources/declarative/incremental/test_sync_token_cursor.py: unit tests for T-TOKEN, T-CAP, T-FOREIGN, T-SCHEMA, T-DISC and T-VALID, plus the cursor state rules.
• unit_tests/sources/declarative/incremental/test_sync_token_cursor_integration.py: end-to-end tests through ConcurrentDeclarativeSource.read with HttpMocker, asserting exact request bodies. Covers T-BOOT, T-INC, T-NOECHO, T-EMPTY, T-FAIL, T-BUILDER, T-INIT, T-PAR (3 partitions), T-RESUME and T-INERT.
• The new e2e tests fail on unmodified main (schema validation error) and pass with this change.
• Full unit suite: all tests pass except a few that also fail on unmodified main: two date-only parsing tests in file-based, and a schema-test fixture race under -n auto.

Breaking change

None. This is an additive, opt-in component. Existing cursors, streams and the retriever are unchanged when the cursor is not used: page_observer is None, and source_defined_incremental defaults to False.

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)

5 982 tests  +80   5 951 ✅ +80   6m 17s ⏱️ -4s
    1 suites ± 0      31 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit c51857f. ± Comparison against base commit 3e85723.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

5 985 tests  +80   5 954 ✅ +80   10m 56s ⏱️ -7s
    1 suites ± 0      31 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit c51857f. ± Comparison against base commit 3e85723.

♻️ This comment has been updated with latest results.

[Patrick Nilan (pnilan)]

🤖 This comment was generated by an AI Agent.

Review: fix before merge

Reviewed at 39add2c7; line numbers below refer to that commit.

Existing connectors are safe, and this is not a breaking change.

• Every change to shared code is gated on SyncTokenCursor or keeps the old default: page_observer stays None and source_defined_incremental stays False on every other stream.
• An offline build and discover of the monorepo's manifests shows zero differences between base and head across 5,192 streams.
• The schema only gains an anyOf branch that matches type: SyncTokenCursor.
• The core rule holds under concurrency. Tokens commit in close_partition, after the partition's records are queued. A stress run (20 rounds × 400 partitions × 16 threads, with failing and vanishing partitions) found no mismatches.

Four things need fixing before merge:

1. Streaming decoders break token capture in production, but pass in the Builder.
2. A StateDelegatingStream with api_retention_period crashes from the third sync.
3. The tests miss the injection paths that two of the three adopters use.
4. Merging this PR alone ships the cursor ahead of the plan's gates.

Five P2 items and four P3 items follow. At the end there is a patch for findings 1, 2, 5 and 6, plus 34 ready-to-use tests. With both applied, 614 tests pass.

---

P1: must fix

1. Streaming decoders break token capture in production; Builder test reads pass

Where: incremental/sync_token_cursor.py:166; the shared decoder instance at parsers/model_to_component_factory.py:3952.

Problem:

• The record selector decodes each page first. On the terminal page, the cursor decodes the same response again with the same decoder: for decoded in self._decoder.decode(page.response).
• A streaming decoder reads response.raw and then closes it (decoders/composite_raw_decoder.py:288-292), so the second read gets an empty body.
• The Connector Builder swaps these decoders for buffered ones, so the same manifest works there.

Reproduced at head. Without the cursor, every decoder below reads correctly in both modes.

Decoder	Production	Builder test read
JsonDecoder (control)	commits the token	commits the token
GzipDecoder → JsonDecoder	emits every record, then fails with Response JSON data failed to be parsed; state never advances	commits the token
JsonItemsDecoder	fails with IncompleteJSONError: premature EOF	never commits: it yields items, never the envelope
JsonlDecoder	ends COMPLETE but never commits, so every sync silently reads everything	—
IterableDecoder	fails with StreamConsumedError	—

The CDK already rejects the same hazard for CombinedExtractor in _reject_combined_extractor_over_streaming_decoder (:2737). Its docstring warns about manifests that "look correct in the Builder" and fail once published.

Fix:

• Reject the combination in create_sync_token_cursor, reusing that guard's predicate through a shared _decoder_streams_in_production helper. The predicate covers all five streaming decoders, including the buffered stand-ins the Builder builds.
• The patch below does this. test_streaming_decoder_is_rejected_when_the_stream_is_built covers each decoder in both modes.
• A larger alternative is to read the token from the body that the record selector has already decoded.

2. StateDelegatingStream with api_retention_period crashes once token state exists

Where: incremental/sync_token_cursor.py:43 (missing override); the call at parsers/model_to_component_factory.py:4867.

Problem:

• SyncTokenCursor doesn't override get_cursor_datetime_from_state.
• When FinalStateCursor can't parse the stored state, _is_cursor_older_than_retention_period calls that method on the incremental cursor. The base implementation raises (streams/concurrent/cursor.py:111).

Reproduced at head:

• Sync 1 succeeds as a full refresh.
• Sync 2 succeeds and commits T1.
• Sync 3 raises NotImplementedError: SyncTokenCursor does not implement get_cursor_datetime_from_state while the streams are built. The whole read fails with 0 records, unrelated streams included.

Builder test reads never carry state, so this would first appear on a connection's third sync. The pairing is natural for these APIs. Google Calendar is a named adopter, and source-google-calendar has a StateDelegatingStream with api_retention_period: P21D today (manifest.yaml:129). Its draft (airbytehq/airbyte#89078) removes the wrapper, so nothing hits this yet.

Fix: Reject the combination when the stream is built, next to the existing IncrementingCountCursor check (:4792). The patch does this, and test_api_retention_period_is_rejected_with_sync_token_cursor covers it.

If you'd rather support the combination, there are two alternatives:

• Return None, as the issue's spec says. This stops the crash. But StateDelegatingStream treats None as stale (:4869-4871), so every sync would quietly fall back to full refresh.
• Derive the datetime from committed_at, so token age counts against the retention period. That suits tokens that expire, but it is a deliberate change from the spec.

3. The tests miss the paths the adopters use

Where: requesters/request_options/sync_token_request_options_provider.py:58, :67, :76; retrievers/simple_retriever.py:563.

Problem: I mutation-tested the new branches. Of 34 single-line mutants, 23 survive all 44 of the PR's tests. The gaps that matter most:

• Injection targets. Only body_json is tested. Google Calendar (feat(source-google-calendar): sync-token incremental for events via SyncTokenCursor (preview SDM) airbyte#89078) injects the token as request_parameter, and Todoist (feat(source-todoist): incremental tasks and projects via Sync API SyncTokenCursor (preview SDM) airbyte#89077) as body_data. Neither path has a test, and neither does header.
• initial_token with a stored token. A mutant where initial_token always wins survives. Todoist uses initial_token: "*", so that bug would re-read everything on every sync.
• The terminal-page rule. A mutant that always sets pagination_exhausted=True survives. No test has a mid-pass page that carries a token, followed by a terminal page without one. So "only a completed pass commits" is not tested end to end.
• Schema fields. completion_condition and cursor_field are never parsed from a manifest. The discovery test checks 2 of the 4 advertised fields.
• Defect paths. No test sets a decoder or wraps the stream in StateDelegatingStream (findings 1 and 2).

Fix: Add the 18 tests under "Proposed tests: coverage gaps" below. They pass at head. With them, 3 of the 34 mutants survive, and all three are equivalent to the original code.

4. Merging this PR alone ships the cursor ahead of the plan's gates

Problem:

• There is no live validation yet. airbytehq/airbyte-internal-issues#17213 requires it before merge, including the open question of whether Ashby's token covers records edited during the pass that issued it. The 2026-10-16 comment on that issue says live validation against a real Ashby account hasn't been done. The Todoist and Google Calendar drafts haven't run live either.
• Any merge can release it. A merge to main joins the open release draft. The next publish ships it to PyPI and to the SDM image, which Builder connectors pick up automatically.
• Without feat(declarative): add RESET_SYNC_TOKEN recovery for SyncTokenCursor (2/2) #1235 there is no recovery from a rejected token. Reproduced: after an HTTP 410 on a stored token, every later sync ends INCOMPLETE and replays the same token until someone resets state.
• feat(declarative): support RequestPath tokens and first-request checks in SyncTokenCursor #1238 assumes the cursor is unreleased. Its body says its new checks "only apply to SyncTokenCursor streams, which are unreleased (this stack)". If feat(declarative): add SyncTokenCursor for server-issued sync tokens #1234 ships first, those checks tighten a published contract and would need a cdk-migrations.md entry.
• Merging closes the design issue. Resolves https://github.com/airbytehq/airbyte-internal-issues/issues/17213 is a closing keyword. Merging PR 1 would close airbytehq/airbyte-internal-issues#17213 with its acceptance criteria unchecked.
• The title is out of date. It says "(1/2 …)", but the stack has three PRs, and the squash title becomes the release-note line.

Fix:

• Change Resolves to Part of in the body.
• Retitle to feat(declarative): add SyncTokenCursor for server-issued sync tokens.
• Add a merge gate to the body:
  • Merge together with feat(declarative): add RESET_SYNC_TOKEN recovery for SyncTokenCursor (2/2) #1235.
  • Settle feat(declarative): support RequestPath tokens and first-request checks in SyncTokenCursor #1238's release order first.
  • Record the live validation.
  • Run the Google Calendar and Todoist drafts on a preview SDM, since they use the untested paths from finding 3.
• Keep the PR in draft until those gates are met.
• Merge main. It is 11 commits ahead and merges cleanly.

---

P2: should fix

5. initial_token changes JSON type

Where: sync_token_request_options_provider.py:44

eval() literal-evaluates the value. With body_json injection:

• "0" is sent as 0
• "True" is sent as true
• "[]" is sent as []
• "None" sends nothing

The schema says type: string, and stored tokens are always strings.

Fix: eval(self.config, valid_types=(str,)). This is in the patch; test_initial_token_is_sent_as_a_string covers it.

6. cursor_field isn't interpolated

Where: sync_token_cursor.py:79; create_sync_token_cursor

DatetimeBasedCursor and IncrementingCountCursor both interpolate their cursor_field. A stream converted by changing only type therefore advertises default_cursor_field: ['{{ parameters.cursor_field }}']. Twelve cursor_field values across 6 monorepo connectors are templated today.

Fix: Evaluate it with InterpolatedString in the factory, and add interpolation_context: [config] to the schema field. Both are in the patch; test_cursor_field_is_interpolated covers them.

7. Some tests pass for weaker reasons than their names claim

Where: test_sync_token_cursor_integration.py:283 (T-FAIL), :314 (T-BUILDER), :378 (T-PAR)

• Unmatched requests go unnoticed. with HttpMocker() never checks that every matcher was called (airbyte_cdk/test/mock_http/mocker.py:50-56); that check only runs on the decorator path (:167). T-FAIL and T-PAR expect a failure, and an unmatched request also fails, so they can't tell the mocked 500 apart from a change in request shape. Use the @HttpMocker() decorator or assert_number_of_calls.
• T-BUILDER's page-limited half proves nothing. It stops on a page with no token, so "no commit" is guaranteed regardless. At head, a page-limited read does commit a token from a truncated page when completion_condition is unset. Put a token on the truncated page and set completion_condition; test_page_limited_read_does_not_commit_token_on_truncated_page does this.
• T-BUILDER's other half stops on a non-terminal page. Use a terminal page instead; test_early_exit_on_terminal_page_does_not_commit does this.
• Two behaviours have no test: the 600-second throttle and the partition-cap replacement. Both are in the coverage tests.

8. Validation edge cases

Where: model_to_component_factory.py:2356, :2380, :2617; sync_token_cursor.py:119, :173

Each case was verified at head.

• $parameters false positive. Stream-level $parameters: {datetime_format: …} propagates onto the cursor. The build then fails with SyncTokenCursor does not support datetime-only setting(s): datetime_format, although the cursor never set it. Skip keys that come from cursor_model.parameters.
• Reserved token key. The build-time check only covers a literal top-level ListPartitionRouter.cursor_field and a parent's partition_field. An interpolated or nested (UnionPartitionRouter) key fails mid-sync as a bare ValueError (sync_token_cursor.py:119), which users see as a system error. Raise AirbyteTracedException(failure_type=FailureType.config_error) instead. Then update test_partition_key_named_token_is_rejected_at_runtime in the coverage tests, which pins today's ValueError.
• CustomRetriever with a decoder key. create_sync_token_cursor passes the raw dict to _create_component_from_model (:2617-2619), which rejects it. The build crashes where the PR intends a warning (:2367). Build the decoder only when retriever_model.decoder is a model.
• Check order. The token is type-checked (:173) before completion_condition runs (:178). So a terminal page that fails the condition but has a non-string at token_path fails the partition. The spec checks the condition first: that page keeps the stored token and logs a WARN.

9. The new public API has no docstrings

SyncTokenCursor, SyncTokenRequestOptionsProvider and PageCompleted are exported, so they appear undocumented in the pdoc reference. page_observer is also missing from SimpleRetriever's Attributes: block.

Fix: Document:

• the two state shapes;
• the rule that tokens commit in close_partition;
• the 25,000-partition cap;
• that page_observer is None except on SyncTokenCursor streams.

P3: nice to have (4 items)

10. Redundant code

• default_stream.py:94-96 can never change anything, because the branch at :82-88 already sets both fields.
• The property-chunking check in create_simple_retriever (:4034-4041) repeats :2421-2425, which always runs first.
• not self.token_option (sync_token_request_options_provider.py:36) is always false, because the field is required.
• partition_key is computed again at sync_token_cursor.py:191.

11. Private attributes read across objects

• cursor._decoder (model_to_component_factory.py:3952)
• concurrent_cursor._partitioned (:2197)
• stream_slicer._partition_router (default_stream.py:132)

Add read-only properties, or build the decoder once in create_default_stream and pass it to both the cursor and the retriever.

12. Naming and headers

• The logs at sync_token_cursor.py:230, :244, :251, :291 and :313 still say "change token". Say "sync token".
• The new files say Copyright (c) 2025. Use 2026.

13. A generic validation error

With a third anyOf branch, an invalid DatetimeBasedCursor (for example, one missing datetime_format) now reports … is not valid under any of the given schemas. It used to report 'DatetimeBasedCursor' is not one of ['IncrementingCountCursor']. Nothing parses this message, and the old one was misleading anyway, so no change is needed.

---

CI

• Check: destination-motherduck fails on infrastructure. The GSM secret SECRET_DESTINATION_MOTHERDUCK__MOTHERDUCK__CREDS has no enabled version, and fix(declarative): copy the manifest once when propagating types and parameters #1246 and fix(low-code): build only the checked static streams during check #1247 failed the same way. Nothing in this diff needs to change for it.
• Validate PR title doesn't evaluate titles on draft PRs, so its pass means nothing yet.
• Test Connectors passing is not evidence for this component, because none of its six connectors uses it.

---

Patch and tests

The patch covers findings 1, 2, 5 and 6. I applied it to 39add2c7 together with the two test files below:

• 614 tests pass: the PR's 44, the 34 new ones, and the factory, decoder, request-options, CombinedExtractor and StateDelegatingStream suites. One decoder memory-usage test was left out, because it also crashes at head in my local environment.
• Ruff is clean, and mypy reports nothing that it doesn't also report at head.

Patch: findings 1, 2, 5 and 6

diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
index 70b85255d..a314b4400 100644
--- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
+++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
@@ -1487,6 +1487,7 @@ definitions:
           can order versions of a record when deduplicating. It is never used to filter
           records or to build state.
         type: string
+        interpolation_context: [config]
       $parameters:
         type: object
         additionalProperties: true
diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
index d9b0fa3e5..b26a03770 100644
--- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
+++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
@@ -2620,6 +2620,17 @@ class ModelToComponentFactory:
             if decoder_model
             else JsonDecoder(parameters={})
         )
+        if self._decoder_streams_in_production(decoder):
+            raise AirbyteTracedException(
+                message=f"Stream {stream_name}: SyncTokenCursor is not supported with a streaming decoder.",
+                internal_message=(
+                    f"Stream {stream_name} uses {type(decoder).__name__}, which streams the response "
+                    f"and can only read it once. SyncTokenCursor reads the token from the response "
+                    f"after the record selector has consumed it. Use a non-streaming decoder such "
+                    f"as JsonDecoder or XmlDecoder."
+                ),
+                failure_type=FailureType.config_error,
+            )
         return SyncTokenCursor(
             stream_name=stream_name,
             stream_namespace=None,
@@ -2632,7 +2643,13 @@ class ModelToComponentFactory:
             config=config or {},
             parameters=model.parameters or {},
             completion_condition=model.completion_condition,
-            cursor_field=model.cursor_field,
+            cursor_field=(
+                InterpolatedString.create(
+                    model.cursor_field, parameters=model.parameters or {}
+                ).eval(config=config)
+                if model.cursor_field
+                else None
+            ),
         )
 
     def create_default_error_handler(
@@ -2734,6 +2751,20 @@ class ModelToComponentFactory:
             decoder.parser, (CsvParser, JsonLineParser, JsonItemsParser, GzipParser)
         )
 
+    def _decoder_streams_in_production(self, decoder: Decoder) -> bool:
+        """Can this decoder read a response body only once in production?
+
+        True for decoders that stream (`is_stream_response()`), and, in the Connector Builder, for
+        the buffered stand-ins it builds for them (`_is_decoder_downgraded_by_connector_builder`).
+        """
+        inner_decoder = (
+            decoder.decoder if isinstance(decoder, PaginationDecoderDecorator) else decoder
+        )
+        return inner_decoder.is_stream_response() or (
+            self._emit_connector_builder_messages
+            and self._is_decoder_downgraded_by_connector_builder(inner_decoder)
+        )
+
     def _reject_combined_extractor_over_streaming_decoder(self, decoder: Optional[Decoder]) -> None:
         """Refuse to build a `CombinedExtractor` whose response body can only be read once.
 
@@ -2747,17 +2778,11 @@ class ModelToComponentFactory:
         read can be replayed, which would let a manifest look correct in the Builder and lose
         records once published. The downgraded decoders are therefore rejected in the Builder too.
         """
-        if decoder is None:
+        if decoder is None or not self._decoder_streams_in_production(decoder):
             return
         inner_decoder = (
             decoder.decoder if isinstance(decoder, PaginationDecoderDecorator) else decoder
         )
-        streams_in_production = inner_decoder.is_stream_response() or (
-            self._emit_connector_builder_messages
-            and self._is_decoder_downgraded_by_connector_builder(inner_decoder)
-        )
-        if not streams_in_production:
-            return
         raise AirbyteTracedException(
             message="CombinedExtractor is not supported with a streaming decoder.",
             internal_message=(
@@ -4795,6 +4820,12 @@ class ModelToComponentFactory:
                         f"with api_retention_period. IncrementingCountCursor does not use datetime-based "
                         f"cursors, so cursor age validation cannot be performed."
                     )
+                if isinstance(stream_model.incremental_sync, SyncTokenCursorModel):
+                    raise ValueError(
+                        f"Stream '{model.name}' uses SyncTokenCursor which is not supported "
+                        f"with api_retention_period. A sync token has no datetime cursor, so "
+                        f"cursor age validation cannot be performed."
+                    )
 
         stream_state = self._connector_state_manager.get_stream_state(model.name, None)
 
diff --git a/airbyte_cdk/sources/declarative/requesters/request_options/sync_token_request_options_provider.py b/airbyte_cdk/sources/declarative/requesters/request_options/sync_token_request_options_provider.py
index e621b8861..6bc7311f6 100644
--- a/airbyte_cdk/sources/declarative/requesters/request_options/sync_token_request_options_provider.py
+++ b/airbyte_cdk/sources/declarative/requesters/request_options/sync_token_request_options_provider.py
@@ -41,7 +41,7 @@ class SyncTokenRequestOptionsProvider(RequestOptionsProvider):
             else None
         )
         if token is None and self._initial_token:
-            token = self._initial_token.eval(self.config)
+            token = self._initial_token.eval(self.config, valid_types=(str,))
         if token is None:
             return {}
         result: dict[str, Any] = {}

Proposed tests: coverage gaps (unit_tests/sources/declarative/incremental/test_sync_token_cursor_coverage.py; 18 tests, pass at head)

#
# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
#

import copy
import json
from unittest.mock import MagicMock

import pytest

from airbyte_cdk.models import SyncMode
from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
    ConcurrentDeclarativeSource,
)
from airbyte_cdk.sources.declarative.partition_routers.substream_partition_router import (
    SubstreamPartitionRouter,
)
from airbyte_cdk.sources.declarative.retrievers.simple_retriever import PageCompleted
from airbyte_cdk.sources.types import StreamSlice
from airbyte_cdk.test.mock_http import HttpMocker, HttpRequest, HttpResponse
from unit_tests.sources.declarative.incremental.test_sync_token_cursor import (
    Router,
    _cursor,
    _partition,
    _response,
)
from unit_tests.sources.declarative.incremental.test_sync_token_cursor_integration import (
    _URL,
    _incoming_state,
    _manifest,
    _read,
    _source_and_catalog,
    _state,
)

NO_RETRY = {"type": "DefaultErrorHandler", "max_retries": 0, "backoff_strategies": []}
TOKEN_CURSOR = {
    "type": "SyncTokenCursor",
    "token_path": ["syncToken"],
    "token_option": {
        "type": "RequestOption",
        "field_name": "syncToken",
        "inject_into": "body_json",
    },
}
DATETIME_CURSOR = {
    "type": "DatetimeBasedCursor",
    "cursor_field": "created_at",
    "start_datetime": "2020-01-01T00:00:00Z",
    "datetime_format": "%Y-%m-%dT%H:%M:%SZ",
}


def _token(messages):
    return _state(messages)[-1].stream.stream_state.__dict__.get("token")


# A mid-pass page's token must not commit when the terminal page carries none.
def test_intermediate_page_token_is_not_committed_when_terminal_page_has_none():
    state = _incoming_state({"state_type": "sync_token", "token": "T1"})
    with HttpMocker() as http_mocker:
        http_mocker.post(
            HttpRequest(_URL, body={"syncToken": "T1"}),
            HttpResponse(
                json.dumps({"items": [{"id": "1"}], "nextCursor": "p2", "syncToken": "MID"})
            ),
        )
        http_mocker.post(
            HttpRequest(_URL, body={"syncToken": "T1", "nextCursor": "p2"}),
            HttpResponse(json.dumps({"items": [{"id": "2"}]})),
        )
        source, catalog = _source_and_catalog(_manifest(error_handler=NO_RETRY), state)
        messages = _read(source, catalog, state)
    assert _token(messages) == "T1"


@pytest.mark.parametrize(
    "inject_into,request_kwargs",
    [
        ("request_parameter", {"query_params": {"syncToken": "T1"}}),
        ("header", {"headers": {"syncToken": "T1"}}),
        ("body_data", {"body": "syncToken=T1"}),
    ],
)
def test_stored_token_is_sent_through_each_injection_target(inject_into, request_kwargs):
    manifest = _manifest(error_handler=NO_RETRY)
    manifest["streams"][0]["incremental_sync"]["token_option"]["inject_into"] = inject_into
    state = _incoming_state({"state_type": "sync_token", "token": "T1"})
    with HttpMocker() as http_mocker:
        http_mocker.post(
            HttpRequest(_URL, **request_kwargs),
            HttpResponse(json.dumps({"items": [{"id": "1"}], "syncToken": "T2"})),
        )
        source, catalog = _source_and_catalog(manifest, state)
        messages = _read(source, catalog, state)
    assert _token(messages) == "T2"


def test_initial_token_is_not_sent_when_a_token_is_stored():
    state = _incoming_state({"state_type": "sync_token", "token": "T1"})
    with HttpMocker() as http_mocker:
        http_mocker.post(
            HttpRequest(_URL, body={"syncToken": "T1"}),
            HttpResponse(json.dumps({"items": [], "syncToken": "T2"})),
        )
        source, catalog = _source_and_catalog(
            _manifest(initial_token="*", error_handler=NO_RETRY), state
        )
        messages = _read(source, catalog, state)
    assert _token(messages) == "T2"


# Partitioned state is emitted at most every 600 s, plus one final unthrottled emission.
def test_partitioned_state_emission_is_throttled(mocker):
    router = Router([StreamSlice(partition={"account": a}, cursor_slice={}) for a in "abc"])
    cursor = _cursor(router=router)
    clock = mocker.patch("time.time")
    for now, stream_slice in zip((1000, 1100, 1700), router.slices):
        clock.return_value = now
        token = stream_slice.partition["account"].upper()
        cursor.on_page_completed(PageCompleted(stream_slice, _response({"syncToken": token}), True))
        cursor.close_partition(_partition(stream_slice.partition))
    assert len(list(cursor._message_repository.consume_queue())) == 2
    cursor.ensure_at_least_one_state_emitted()
    (final,) = list(cursor._message_repository.consume_queue())
    states = final.state.stream.stream_state.__dict__["states"]
    assert {(s["partition"]["account"], s["cursor"]["token"]) for s in states} == {
        ("a", "A"),
        ("b", "B"),
        ("c", "C"),
    }


def test_stored_partition_replaces_its_own_token_at_the_partition_cap():
    router = Router([StreamSlice(partition={"account": a}, cursor_slice={}) for a in "ab"])
    cursor = _cursor(
        router=router,
        max_partitions=1,
        state={
            "state_type": "sync_token",
            "states": [
                {
                    "partition": {"account": "a"},
                    "cursor": {"token": "A0", "committed_at": "2026-01-01T00:00:00Z"},
                }
            ],
        },
    )
    list(cursor.stream_slices())
    for stream_slice, token in zip(router.slices, ("A1", "B1")):
        cursor.on_page_completed(PageCompleted(stream_slice, _response({"syncToken": token}), True))
        cursor.close_partition(_partition(stream_slice.partition))
    assert [(s["partition"]["account"], s["cursor"]["token"]) for s in cursor.state["states"]] == [
        ("a", "A1")
    ]


def test_partition_key_named_token_is_rejected_at_runtime():
    cursor = _cursor(router=Router([StreamSlice(partition={"token": "x"}, cursor_slice={})]))
    with pytest.raises(ValueError, match="reserved key 'token'"):
        list(cursor.stream_slices())


def _parent_child_manifest(*, parent_sync=None, child_sync=None, incremental_dependency=False):
    manifest = _manifest()
    child = manifest["streams"][0]
    parent = copy.deepcopy(child)
    parent["name"] = "accounts"
    parent["retriever"]["requester"]["path"] = "/accounts"
    if parent_sync is None:
        parent.pop("incremental_sync")
    else:
        parent["incremental_sync"] = parent_sync
    if child_sync is not None:
        child["incremental_sync"] = child_sync
    child["retriever"]["partition_router"] = {
        "type": "SubstreamPartitionRouter",
        "parent_stream_configs": [
            {
                "type": "ParentStreamConfig",
                "stream": parent,
                "parent_key": "id",
                "partition_field": "account",
                "incremental_dependency": incremental_dependency,
            }
        ],
    }
    manifest["streams"] = [parent, child]
    return manifest


def test_get_partition_router_returns_substream_router_for_token_child():
    source = ConcurrentDeclarativeSource(
        source_config=_parent_child_manifest(), config={}, catalog=None
    )
    child = {s.name: s for s in source.streams(config={})}["events"]
    assert isinstance(child.get_partition_router(), SubstreamPartitionRouter)


@pytest.mark.parametrize("cursor_field,expected", [(None, []), ("updatedAt", ["updatedAt"])])
def test_discovery_advertises_the_documented_shape(cursor_field, expected):
    manifest = _manifest()
    if cursor_field:
        manifest["streams"][0]["incremental_sync"]["cursor_field"] = cursor_field
    stream = (
        ConcurrentDeclarativeSource(source_config=manifest, config={}, catalog=None)
        .discover(logger=MagicMock(), config={})
        .streams[0]
    )
    assert stream.supported_sync_modes == [SyncMode.full_refresh, SyncMode.incremental]
    assert stream.source_defined_cursor is True
    assert stream.is_resumable is True
    assert stream.default_cursor_field == expected


def test_manifest_completion_condition_blocks_commit():
    state = _incoming_state({"state_type": "sync_token", "token": "T1"})
    with HttpMocker() as http_mocker:
        http_mocker.post(
            HttpRequest(_URL, body={"syncToken": "T1"}),
            HttpResponse(json.dumps({"items": [{"id": "1"}], "syncToken": "T2", "success": False})),
        )
        source, catalog = _source_and_catalog(
            _manifest(
                completion_condition="{{ response.get('success', true) }}", error_handler=NO_RETRY
            ),
            state,
        )
        messages = _read(source, catalog, state)
    assert _token(messages) == "T1"


# A page-limited read stops on a page that carries a token; completion_condition must block it.
def test_page_limited_read_does_not_commit_token_on_truncated_page():
    with HttpMocker() as http_mocker:
        http_mocker.post(
            HttpRequest(_URL, body=None),
            HttpResponse(
                json.dumps(
                    {
                        "items": [{"id": "1"}],
                        "nextCursor": "p2",
                        "syncToken": "MID",
                        "moreDataAvailable": True,
                    }
                )
            ),
        )
        source, catalog = _source_and_catalog(
            _manifest(completion_condition="{{ not response.get('moreDataAvailable') }}"),
            limit_pages_fetched_per_slice=1,
        )
        messages = _read(source, catalog)
    assert not any("token" in m.stream.stream_state.__dict__ for m in _state(messages))


def test_early_exit_on_terminal_page_does_not_commit():
    with HttpMocker() as http_mocker:
        http_mocker.post(
            HttpRequest(_URL, body=None),
            HttpResponse(
                json.dumps({"items": [{"id": "1"}, {"id": "2"}], "syncToken": "must-not-commit"})
            ),
        )
        source, _ = _source_and_catalog(_manifest())
        stream = source.streams(config={})[0]
        partition = next(iter(stream.generate_partitions()))
        record_iterator = iter(partition.read())
        assert next(record_iterator).data == {"id": "1"}
        record_iterator.close()
        stream.cursor.close_partition(partition)
    assert "token" not in stream.cursor.state


def test_xml_decoder_reaches_records_and_token():
    manifest = _manifest(error_handler=NO_RETRY)
    stream = manifest["streams"][0]
    stream["retriever"]["decoder"] = {"type": "XmlDecoder"}
    stream["retriever"]["record_selector"]["extractor"]["field_path"] = ["root", "items"]
    stream["retriever"]["paginator"] = {"type": "NoPagination"}
    stream["incremental_sync"]["token_path"] = ["root", "syncToken"]
    with HttpMocker() as http_mocker:
        http_mocker.post(
            HttpRequest(_URL, body=None),
            HttpResponse("<root><items><id>1</id></items><syncToken>X1</syncToken></root>"),
        )
        source, catalog = _source_and_catalog(manifest)
        messages = _read(source, catalog)
    assert any(m.type.value == "RECORD" for m in messages)
    assert _token(messages) == "X1"


def test_datetime_child_cannot_depend_on_token_parent():
    manifest = _parent_child_manifest(
        parent_sync=TOKEN_CURSOR, child_sync=DATETIME_CURSOR, incremental_dependency=True
    )
    source = ConcurrentDeclarativeSource(source_config=manifest, config={}, catalog=None)
    with pytest.raises(ValueError, match="SyncTokenCursor parent"):
        source.discover(logger=MagicMock(), config={})


def test_datetime_child_cannot_depend_on_state_delegating_token_parent():
    manifest = _parent_child_manifest(
        parent_sync=TOKEN_CURSOR, child_sync=DATETIME_CURSOR, incremental_dependency=True
    )
    parent_config = manifest["streams"][1]["retriever"]["partition_router"][
        "parent_stream_configs"
    ][0]
    full_refresh = copy.deepcopy(parent_config["stream"])
    full_refresh.pop("incremental_sync")
    parent_config["stream"] = {
        "type": "StateDelegatingStream",
        "name": "accounts",
        "full_refresh_stream": full_refresh,
        "incremental_stream": parent_config["stream"],
    }
    source = ConcurrentDeclarativeSource(source_config=manifest, config={}, catalog=None)
    with pytest.raises(ValueError, match="SyncTokenCursor parent"):
        source.discover(logger=MagicMock(), config={})


def test_manifest_property_chunking_is_rejected():
    manifest = _manifest()
    manifest["streams"][0]["retriever"]["requester"]["request_parameters"] = {
        "properties": {
            "type": "QueryProperties",
            "property_list": ["a", "b"],
            "property_chunking": {
                "type": "PropertyChunking",
                "property_limit_type": "property_count",
                "property_limit": 1,
            },
        }
    }
    source = ConcurrentDeclarativeSource(source_config=manifest, config={}, catalog=None)
    with pytest.raises(ValueError, match="query property chunking"):
        source.discover(logger=MagicMock(), config={})

Proposed tests: the fixes (unit_tests/sources/declarative/incremental/test_sync_token_cursor_fixes.py; 16 tests, fail at head, pass with the patch)

#
# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
#

import copy
import json
from unittest.mock import MagicMock

import pytest

from airbyte_cdk.models import FailureType
from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
    ConcurrentDeclarativeSource,
)
from airbyte_cdk.test.mock_http import HttpMocker, HttpRequest, HttpResponse
from airbyte_cdk.utils.traced_exception import AirbyteTracedException
from unit_tests.sources.declarative.incremental.test_sync_token_cursor_integration import (
    _URL,
    _manifest,
    _read,
    _source_and_catalog,
    _state,
)

NO_RETRY = {"type": "DefaultErrorHandler", "max_retries": 0, "backoff_strategies": []}
STREAMING_DECODERS = [
    {"type": "GzipDecoder", "decoder": {"type": "JsonDecoder"}},
    {"type": "JsonlDecoder"},
    {"type": "JsonItemsDecoder", "items_path": "items.item"},
    {"type": "CsvDecoder"},
    {"type": "IterableDecoder"},
]


@pytest.mark.parametrize(
    "emit_connector_builder_messages", [False, True], ids=["platform", "builder"]
)
@pytest.mark.parametrize("decoder", STREAMING_DECODERS, ids=lambda decoder: decoder["type"])
def test_streaming_decoder_is_rejected_when_the_stream_is_built(
    decoder, emit_connector_builder_messages
):
    manifest = _manifest()
    manifest["streams"][0]["retriever"]["decoder"] = decoder
    manifest["streams"][0]["retriever"]["paginator"] = {"type": "NoPagination"}
    source = ConcurrentDeclarativeSource(
        source_config=manifest,
        config={},
        catalog=None,
        emit_connector_builder_messages=emit_connector_builder_messages,
    )
    with pytest.raises(
        AirbyteTracedException, match="SyncTokenCursor is not supported with a streaming decoder"
    ) as error:
        source.discover(logger=MagicMock(), config={})
    assert error.value.failure_type == FailureType.config_error


def test_api_retention_period_is_rejected_with_sync_token_cursor():
    incremental_stream = _manifest()["streams"][0]
    full_refresh_stream = copy.deepcopy(incremental_stream)
    full_refresh_stream.pop("incremental_sync")
    manifest = _manifest()
    manifest["streams"] = [
        {
            "type": "StateDelegatingStream",
            "name": "events",
            "api_retention_period": "P21D",
            "full_refresh_stream": full_refresh_stream,
            "incremental_stream": incremental_stream,
        }
    ]
    source = ConcurrentDeclarativeSource(source_config=manifest, config={}, catalog=None)
    with pytest.raises(
        ValueError, match="SyncTokenCursor which is not supported with api_retention_period"
    ):
        source.discover(logger=MagicMock(), config={})


@pytest.mark.parametrize("initial_token", ["0", "True", "None", "[]"])
def test_initial_token_is_sent_as_a_string(initial_token):
    with HttpMocker() as http_mocker:
        http_mocker.post(
            HttpRequest(_URL, body={"syncToken": initial_token}),
            HttpResponse(json.dumps({"items": [], "syncToken": "T1"})),
        )
        source, catalog = _source_and_catalog(
            _manifest(initial_token=initial_token, error_handler=NO_RETRY)
        )
        messages = _read(source, catalog)
    assert _state(messages)[-1].stream.stream_state.__dict__.get("token") == "T1"


def test_cursor_field_is_interpolated():
    manifest = _manifest()
    stream = manifest["streams"][0]
    stream["$parameters"] = {"cursor_field": "updatedAt"}
    stream["incremental_sync"]["cursor_field"] = "{{ parameters.cursor_field }}"
    discovered = (
        ConcurrentDeclarativeSource(source_config=manifest, config={}, catalog=None)
        .discover(logger=MagicMock(), config={})
        .streams[0]
    )
    assert discovered.default_cursor_field == ["updatedAt"]

[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

[devin-ai-integration]

☑️ Resolved in 8ee8478 and c51857f, on top of a merge of main. This is the follow-up to the review in comment 6369869413; findings 1–12 are addressed.

• 1, 2, 5, 6: your patch is applied as written. That covers the shared _decoder_streams_in_production helper, rejecting a streaming decoder, rejecting api_retention_period, initial_token with valid_types=(str,), and an interpolated cursor_field with interpolation_context: [config].

• 3: test_sync_token_cursor_coverage.py (18 tests) and test_sync_token_cursor_fixes.py (16 tests) are added as written.

• 7: T-FAIL and T-PAR now assert that every mocked request was made. The weak halves of T-BUILDER are replaced by the stronger tests.

• 8:

  • Keys that come from stream $parameters no longer trigger the datetime-only check.
  • A partition key named token now raises a config_error.
  • A CustomRetriever whose decoder is a dict now gets the warning instead of a crash.
  • completion_condition is evaluated before the token's type is checked.

  Each change has a test.

• 9: added docstrings for SyncTokenCursor, SyncTokenRequestOptionsProvider and PageCompleted, and documented SimpleRetriever.page_observer.

• 10–12:

  • Removed the redundant code: the default_stream cursor block, the token_option falsy guard, and the duplicate property-chunking check.
  • Cross-object private reads now go through read-only properties: decoder, partitioned and partition_router.
  • Log messages say "sync token" instead of "change token".
  • The copyright year in the new files is now 2026.

• 13: no change.

• 4: the title is now feat(declarative): add SyncTokenCursor for server-issued sync tokens. The description says Part of #17213 and Close on merge: no, and adds a merge-gate checklist covering feat(declarative): add RESET_SYNC_TOKEN recovery for SyncTokenCursor (2/2) #1235, the feat(declarative): support RequestPath tokens and first-request checks in SyncTokenCursor #1238 release order, live Ashby validation, and preview SDM runs for Calendar and Todoist. The PR stays in draft.

Locally, 771 tests pass across the incremental, factory, decoder, request-option, retriever and state-delegating suites; ruff and mypy are clean. The same changes are merged into #1235, now at c15226fb.

Written by Devin