[devin-ai-integration]

Summary

Adds recovery for rejected or expired sync tokens, such as Ashby's sync_token_expired. This is PR 2 of 3 in the stack. It builds on SyncTokenCursor from #1234, which is now merged to main, and #1238 follows it.

Part of https://github.com/airbytehq/airbyte-internal-issues/issues/17213. Merging this PR alone does not close the issue.

Changes

• New ResponseAction.RESET_SYNC_TOKEN, with matching schema and model enum entries. The schema now documents what the action does and its limits:

  • it works only with SyncTokenCursor
  • it re-reads the partition at most once per attempt
  • it doesn't clear the destination
  • when the API requires the client to clear its copy, use FAIL instead

  Example filter:

  - type: HttpResponseFilter
    action: RESET_SYNC_TOKEN
    predicate: "{{ response.get('errorInfo', {}).get('code') in ['sync_token_expired', 'sync_token_invalid'] }}"

• HttpClient raises SyncTokenResetRequiredException for that action, the same way it handles RESET_PAGINATION. CompositeErrorHandler keeps the action instead of falling through to the next handler.

• SimpleRetriever catches the exception and calls SyncTokenCursor.restart_without_token. It then re-reads only the rejected partition from page 1, without the stored token: it sends initial_token, or no token if none is configured.

  • A WARN names the stream, the partition and the API's reason, and says that all of that partition's records are emitted again.
  • A sync-token-reset analytics message is emitted.

• Safeguards so a reset can't silently full-read on every sync:

  • A token committed by a reset re-read is saved with after_reset: true. If the API rejects that token on the next sync, the stream fails with a system_error that includes the API's reason. The next normal commit clears the flag.
  • If a reset re-read ends without a usable token, the stream fails with a system_error and keeps the committed token. That covers a failed completion_condition and a token that is missing, empty or null.
  • A second reset within the same read also fails with a system_error, as does a rejected read that sent no token.

• Build-time validation: the factory raises ValueError when RESET_SYNC_TOKEN appears on a stream whose incremental_sync isn't a SyncTokenCursor, or on a login requester, an AsyncRetriever requester or a file_uploader requester.

Declarative-First Evaluation

This is a CDK feature, not a connector change. RESET_PAGINATION restarts pagination but keeps the stored token, and no existing action drops it. The new action gives adopters a built-in declarative option.

Test Coverage

• End-to-end tests with HttpMocker that assert exact request bodies and emitted state:
  • reset on page 3, recovering with a new token marked after_reset
  • reset re-read context, including a record filter and transformation after a page-size reduction
  • second reset in one read
  • RESET_PAGINATION interplay
  • CompositeErrorHandler
  • reset re-read with a missing, empty or null token: fails and keeps T1
  • token committed after a reset and rejected on the next sync: fails without another tokenless request
  • reset followed by a normal sync: works
  • initial_token sent on the re-read
  • partitioned reset re-reads only the rejected partition
  • unrelated error during the re-read keeps the state
  • analytics message
• The test reset predicate is narrowed to success == false combined with a specific error code.
• Factory tests for the build-time rejection cases.
• Unit tests in test_http_client.py and test_composite_error_handler.py.

Breaking change

None. The action is opt-in, and existing actions and streams are unchanged. The after_reset state field is new, but the state format hasn't been released yet.

Not yet done

• The error-handling.md docs section lives in airbytehq/airbyte, so it will ship separately.
• Live recovery validation has not been re-run at this head. The only live run used an earlier preview with opaque_token state.

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-reset#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-reset

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 097 tests   6 066 ✅  6m 27s ⏱️
    1 suites     31 💤
    1 files        0 ❌

Results for commit acf0b95.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

6 100 tests  +22   6 069 ✅ +22   11m 41s ⏱️ + 3m 18s
    1 suites ± 0      31 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit acf0b95. ± Comparison against base commit 61ddc6a.

♻️ 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 7789d581.

What works: a reset never touches committed state. It re-reads only the rejected partition, from page 1 with a fresh paginator and record parser. It sends initial_token on the re-read, and fails on a second reset in the same attempt. Existing manifests are unaffected, so this is not a breaking change.

What needs fixing: what happens across syncs, and what the action exposes to connector authors. Keep RESET_SYNC_TOKEN only with the guardrails in 1 and 2, so a stream recovers once and then fails. Without them, adopters should map token errors to FAIL with a refresh instruction instead.

---

P1: must fix

1. A token that keeps getting rejected turns every sync into a silent full read

Where: simple_retriever.py lines 715–727 · sync_token_cursor.py lines 221–243

Problem:

• The only guard is per attempt: restart_without_token refuses a slice without a token. Nothing records that the previous sync already reset.
• If the token the re-read commits is also rejected, every sync resets, reads everything, and stays green. Causes include a bug that alters the token on the way out, or a token_path that captures a string that isn't the token.
• Reproduced: three consecutive syncs, three full reads, zero errors, one WARN each. State advances T2 → T3 → T4, so nothing looks stuck.

Requested change:

• Store "after_reset": true with a token that a reset re-read commits, and load it back in _load_state.
• If restart_without_token gets a slice whose committed token has that flag, raise a system_error instead of resetting again.
• The next normal commit clears the flag. The state format is unreleased, so adding the field costs nothing now.

2. Recovery reports success without a replacement token

Where: simple_retriever.py lines 715–727, together with on_page_completed and close_partition

Problem:

• Sometimes the re-read's last page has no token: it's missing, null or "". feat(declarative): add SyncTokenCursor for server-issued sync tokens #1234's normal rule then keeps the stored token, T1, which the API just rejected. The partition still finishes COMPLETE.
• Every later sync sends T1, is rejected, and reads everything again. Reproduced with all three forms over two consecutive syncs.
• Likely triggers: a mistake in token_path or completion_condition, or an API change that stops returning the token on a full pass.

Requested change: track the partitions that reset in this attempt. On a reset partition's last page, require a valid token. If there isn't one, or completion_condition is false, raise a system_error and keep T1. Leave #1234's tolerance unchanged for a normal token pass that ends without a new token.

Regression tests for 1 and 2, to append to test_sync_token_cursor_integration.py (4 fail at 7789d581; the control passes)

def _has_system_error(messages):
    return any(
        message.type == Type.TRACE
        and message.trace.error
        and message.trace.error.failure_type == FailureType.system_error
        for message in messages
    )


@pytest.mark.parametrize(
    "terminal",
    [
        pytest.param({"items": [{"id": "all-1"}]}, id="no-token"),
        pytest.param({"items": [{"id": "all-1"}], "syncToken": ""}, id="empty-token"),
        pytest.param({"items": [{"id": "all-1"}], "syncToken": None}, id="null-token"),
    ],
)
def test_reset_re_read_without_a_new_token_fails_and_keeps_the_committed_token(terminal):
    state = _incoming_state({"state_type": "sync_token", "token": "T1"})
    manifest = _manifest(error_handler=_reset_error_handler())
    with HttpMocker() as http_mocker:
        http_mocker.post(
            HttpRequest(_URL, body={"syncToken": "T1"}),
            HttpResponse(json.dumps({"success": False, "error": "sync_token_expired"})),
        )
        http_mocker.post(HttpRequest(_URL, body=None), HttpResponse(json.dumps(terminal)))
        source, catalog = _source_and_catalog(manifest, state)
        messages = _read_until_error(source, catalog, state)

    assert _state(messages)[-1].stream.stream_state.__dict__["token"] == "T1"
    assert _has_system_error(messages)


def test_token_committed_by_a_reset_that_is_rejected_again_fails_without_another_full_read():
    manifest = _manifest(error_handler=_reset_error_handler())
    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({"success": False, "error": "sync_token_invalid"})),
        )
        http_mocker.post(
            HttpRequest(_URL, body=None),
            HttpResponse(json.dumps({"items": [{"id": "all-1"}], "syncToken": "T2"})),
        )
        source, catalog = _source_and_catalog(manifest, state)
        first_sync = _read(source, catalog, state)
    assert _state(first_sync)[-1].stream.stream_state.__dict__["token"] == "T2"

    state = [_state(first_sync)[-1]]
    with HttpMocker() as http_mocker:
        http_mocker.post(
            HttpRequest(_URL, body={"syncToken": "T2"}),
            HttpResponse(json.dumps({"success": False, "error": "sync_token_invalid"})),
        )
        tokenless_request = HttpRequest(_URL, body=None)
        http_mocker.post(
            tokenless_request,
            HttpResponse(json.dumps({"items": [{"id": "all-1"}], "syncToken": "T3"})),
        )
        source, catalog = _source_and_catalog(manifest, state)
        second_sync = _read_until_error(source, catalog, state)
        http_mocker.assert_number_of_calls(tokenless_request, 0)

    assert not any(message.type == Type.RECORD for message in second_sync)
    assert _state(second_sync)[-1].stream.stream_state.__dict__["token"] == "T2"
    assert _has_system_error(second_sync)


def test_a_normal_reset_followed_by_a_normal_sync_still_works():
    manifest = _manifest(error_handler=_reset_error_handler())
    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({"success": False, "error": "sync_token_expired"})),
        )
        http_mocker.post(
            HttpRequest(_URL, body=None),
            HttpResponse(json.dumps({"items": [{"id": "all-1"}], "syncToken": "T2"})),
        )
        source, catalog = _source_and_catalog(manifest, state)
        first_sync = _read(source, catalog, state)

    state = [_state(first_sync)[-1]]
    with HttpMocker() as http_mocker:
        http_mocker.post(
            HttpRequest(_URL, body={"syncToken": "T2"}),
            HttpResponse(json.dumps({"items": [{"id": "changed-1"}], "syncToken": "T3"})),
        )
        source, catalog = _source_and_catalog(manifest, state)
        second_sync = _read(source, catalog, state)

    assert [m.record.data["id"] for m in second_sync if m.type == Type.RECORD] == ["changed-1"]
    assert _state(second_sync)[-1].stream.stream_state.__dict__["token"] == "T3"

Reference patch for 1 and 2 (with it, all 5 tests above and all 127 existing sync-token and composite-handler tests pass. 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 abdd67228..c7712f284 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._reset_partitions: set[str] = set()
         self._load_state(stream_state or {})
 
     @property
@@ -201,12 +202,17 @@ class SyncTokenCursor(Cursor):
             )
         else:
             complete = True
-        if not complete:
-            self._warn_missing_token()
-            return
-
-        token = self._coerce_token(token_value, path)
+        token = self._coerce_token(token_value, path) if complete else None
         if token is None:
+            if partition_key in self._reset_partitions:
+                message = (
+                    f"Stream {self._stream_name}: the full re-read after a sync token reset ended "
+                    "without a new token, so the next sync would be rejected the same way. "
+                    "Check token_path and completion_condition."
+                )
+                raise AirbyteTracedException(
+                    message=message, internal_message=message, failure_type=FailureType.system_error
+                )
             self._warn_missing_token()
             return
 
@@ -217,6 +223,8 @@ class SyncTokenCursor(Cursor):
                 .isoformat(timespec="milliseconds")
                 .replace("+00:00", "Z"),
             }
+            if partition_key in self._reset_partitions:
+                self._pending[partition_key]["after_reset"] = True
 
     def restart_without_token(
         self, stream_slice: StreamSlice, reason: Optional[str] = None
@@ -234,6 +242,17 @@ class SyncTokenCursor(Cursor):
 
         partition_key = self._serializer.to_partition_key(stream_slice.partition or {})
         with self._lock:
+            if (self._committed.get(partition_key) or {}).get("after_reset"):
+                message = (
+                    f"Stream {self._stream_name}: the API rejected the sync token that the previous "
+                    f"sync committed after a full re-read ({reason or 'no reason given'}). Two "
+                    "rejections in a row usually mean token_path, token_option or the "
+                    "RESET_SYNC_TOKEN predicate is wrong. Refresh the stream after fixing it."
+                )
+                raise AirbyteTracedException(
+                    message=message, internal_message=message, failure_type=FailureType.system_error
+                )
+            self._reset_partitions.add(partition_key)
             self._pending.pop(partition_key, None)
 
         return StreamSlice(
@@ -426,6 +445,7 @@ class SyncTokenCursor(Cursor):
                     committed[self._serializer.to_partition_key(partition)] = {
                         "token": cursor["token"],
                         "committed_at": cursor.get("committed_at", ""),
+                        **({"after_reset": True} if cursor.get("after_reset") else {}),
                     }
                 else:
                     self._log_bootstrap("foreign state")
@@ -439,6 +459,7 @@ class SyncTokenCursor(Cursor):
                 self._committed["{}"] = {
                     "token": token,
                     "committed_at": stream_state.get("committed_at", ""),
+                    **({"after_reset": True} if stream_state.get("after_reset") else {}),
                 }
             elif token is not None:
                 self._log_foreign_state()

---

P2

3. The only signal is one WARN line, and it describes the wrong scope

Where: simple_retriever.py lines 721–724

• A reset is the most expensive thing a token stream can do. Emit a structured signal next to the WARN so resets can be counted per connection, for example create_analytics_message("sync-token-reset", ...), as file-based sources do (file_based_source.py:441).
• The WARN says "reading this stream in full", but only the rejected partition is re-read. Name the partition when there is one.
• "N records read earlier in this attempt may be emitted again" undercounts. The re-read emits every record in the partition again.

4. RESET_SYNC_TOKEN has no description

Where: declarative_component_schema.yaml:2889

The value has no description anywhere. In the Builder it reads like something as cheap as RESET_PAGINATION. Extend the HttpResponseFilter.action description, for example:

RESET_SYNC_TOKEN works only on streams whose incremental_sync is a SyncTokenCursor. It drops the partition's stored token and re-reads the partition from page 1 without it, at most once per attempt. The destination is not cleared: append connections receive the partition's records again, and records the API deleted or hid while the token was invalid stay in the destination. When the API requires the client to clear its copy (for example, Google Calendar's HTTP 410), use FAIL with an instruction to refresh the stream instead.

Also add the error-handling.md section that airbytehq/airbyte-internal-issues#17213 planned.

5. No build-time validation, unlike REDUCE_PAGE_SIZE and SPLIT_REQUEST_WINDOW

Where: model_to_component_factory.py lines 4238–4240 · sync_token_reset_exception.py lines 11–14

• The factory rejects the sibling actions wherever it can see they can't be honored, using _uses_reduce_page_size_action and _uses_split_request_window_action. It also calls _reject_*_action on SessionTokenAuthenticator login requesters, AsyncRetriever requesters and file uploaders.
• RESET_SYNC_TOKEN has none of these checks.
  • On a non-token stream, a manifest with it builds and passes the Builder, then fails in production with a config_error the first time the API returns that error.
  • On a login or async requester, SyncTokenResetRequiredException would escape unclassified, because it is a plain Exception.
• Requested: add _uses_reset_sync_token_action, raising ValueError when the stream's incremental_sync isn't a SyncTokenCursor, and call it at the existing _reject_*_action call sites. test_reset_action_on_non_token_stream_is_not_supported then becomes a build-time check.

---

P3

6. _reset_error_handler uses {{ response.get('success') == false }} (test_sync_token_cursor_integration.py:236). That matches every success: false envelope, not just token errors, and adopters copy tests. Narrow it to {{ response.get('success') == false and response.get('error') in ['sync_token_expired', 'sync_token_invalid'] }}.
7. These cases have no tests yet. All three already behave correctly at this head:
   • initial_token on the reset re-read (the T-INIT row in the design issue);
   • a partitioned reset, where only the rejected partition is re-read;
   • a re-read that fails with an unrelated error, where state must stay T1.
8. restart_without_token never uses its reason parameter; use it in the error from 1. Also, self._pending.pop(...) can never find an entry, because a pending token is set only on the last page.
9. Housekeeping:
   • The title says "(2/2)", but feat(declarative): support RequestPath tokens and first-request checks in SyncTokenCursor #1238 makes this 2 of 3.
   • The description says "Resolves … Close on merge: yes", while feat(declarative): add SyncTokenCursor for server-issued sync tokens #1234 says "Part of" and closes airbytehq/airbyte-internal-issues#17213 only after the whole stack ships.
   • The only live run of the recovery path was the prove-fix on fix(source-ashby): reset rejected candidates sync token with RESET_SYNC_TOKEN (preview SDM) airbyte#89056 (2026-10-17). It used an earlier build (opaque_token state, preview SDM 7.37.0.post5). Rerun its Recovery step on the current head before release.

[devin-ai-integration]

Re: review: fix before merge

☑️ Resolved in c192670 (merges #1234 at 4050495), except for the two items under "Not done yet" below.

• P1, after_reset: a token committed by a reset re-read is saved with after_reset: true, in both unpartitioned and partitioned state. If that token is rejected on the next sync, the stream fails with a system_error that includes reason, and no second tokenless read happens. The next normal commit clears the flag. A non-boolean after_reset counts as malformed state, following feat(declarative): add SyncTokenCursor for server-issued sync tokens #1234's rule.
• P1, replacement token required: a reset re-read whose last page fails completion_condition, or returns no token, an empty token or a null token, raises a system_error. T1 stays committed. Non-reset passes still only log a warning.
• P2, telemetry and warning: a sync-token-reset analytics message is emitted. The WARN now names the partition and says all of that partition's records are emitted again.
• P2, schema docs: your RESET_SYNC_TOKEN paragraph is in HttpResponseFilter.action and in the generated model.
• P2, build-time validation: _uses_reset_sync_token_action and _reject_reset_sync_token_action are wired up. ValueError is raised for a non-SyncTokenCursor stream, a login requester, AsyncRetriever requesters and file_uploader. The Ashby, Todoist and Google Calendar draft manifests still build.
• P3: I narrowed the sample predicate, used reason, and removed the dead _pending.pop. All five of your tests are in, plus tests for initial_token, a partitioned reset, an unrelated error during the re-read, the factory rejections, and the analytics message.
• PR text: the title is now (2/3), and the description says Part of #17213 and no longer claims it closes the issue.
• All required checks pass. The only failure is the optional destination-motherduck check, which fails on a missing CI secret.

Not done yet:

• The error-handling.md section lives in airbytehq/airbyte, so it will ship as a separate docs PR.
• Live recovery validation at this head still needs a new preview image and an Ashby run.

---

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