Repository navigation
fix(cloud): keep PyAirbyte guidance on run_cloud_sync 409s and report the running job ID - #1405
Conversation
Co-Authored-By: AJ Steers <[email protected]>
Co-Authored-By: AJ Steers <[email protected]>
Co-Authored-By: AJ Steers <[email protected]>
Co-Authored-By: AJ Steers <[email protected]>
|
I'll fix CI failures and address comments from users with write access. I'll skip comments containing "(aside)".
|
👋 Greetings, Airbyte Team Member!Here are some helpful tips and reminders for your convenience. 💡 Show Tips and TricksTesting This PyAirbyte VersionYou can test this version of PyAirbyte using the following: # Run PyAirbyte CLI from this branch:
uvx --from 'git+https://github.com/airbytehq/PyAirbyte.git@devin/1792200236-run-sync-409-guidance' pyairbyte --help
# Install PyAirbyte from this branch for development:
pip install 'git+https://github.com/airbytehq/PyAirbyte.git@devin/1792200236-run-sync-409-guidance'PR Slash CommandsAirbyte Maintainers can execute the following slash commands on your PR:
📚 Show Repo GuidanceHelpful ResourcesCommunity SupportQuestions? Join the #pyairbyte channel in our Slack workspace. |
Code coverage overviewLanguages: Python Python / code-coverage/pytest-no-creds
View coverage of the 10 most impacted files.
Python / code-coverage/pytest
View coverage of the 10 most impacted files.
Python / code-coverage/pytest-unit
Line coverage data for the release-candidate/v1 branch is not yet available. View coverage of the 10 most covered files.
ℹ️ Calculations represent line coverage. Results may change as calculations are updated. Updated |
There was a problem hiding this comment.
🟡 Changes recommended
Global formatting exceeds the clarified recovery scope, diagnostic failures can mask conflicts, and the exception’s structured job ID remains unset.
3 open issues
What changed in this PR
Improves Cloud sync-conflict recovery by preserving local guidance and identifying an already-running job.
Changes:
- Adds running-job lookup and recovery instructions for duplicate sync requests.
- Prioritizes local guidance in agent-facing Cloud errors.
- Expands conflict tests and sync-tool documentation.
| File | Description |
|---|---|
| tests/unit_tests/test_mcp_tool_errors.py | Tests guidance preservation and active-sync formatting. |
| tests/unit_tests/test_cloud_connections.py | Tests running, completed, missing, and disabled-job scenarios. |
| airbyte/mcp/cloud.py | Documents conflict recovery and timeout behavior. |
| airbyte/mcp/_error_handling.py | Changes error formatting and registers active-sync errors. |
| airbyte/exceptions.py | Broadens the active-sync exception docstring. |
| airbyte/cloud/connections.py | Detects running syncs and supplies recovery guidance. |
Review effort: Balanced
|
|
||
| job_id = sync_result.job_id | ||
| return AirbyteConnectionSyncActiveError( | ||
| connection_id=self.connection_id, |
There was a problem hiding this comment.
Valid. If the exception path stays in the rework, I'll pass job_id=job_id and assert it in the test.
There was a problem hiding this comment.
🙋 Human Input Needed: should the already-running case still raise an exception? I've proposed that run_cloud_sync return a normal "already running" result with the job ID. That would make this point moot. If the exception stays, I'll pass job_id=job_id and assert error.job_id in the test. This reply replaces my earlier one, which didn't follow the reply format.
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| active_sync_error = self._already_running_sync_error(ex) | ||
| if active_sync_error is not None: | ||
| raise active_sync_error from ex |
There was a problem hiding this comment.
🟡 Running conflicts bypass sync error handlers
When a sync is already running, run_sync raises an error outside the AirbyteConnectionSyncError hierarchy. Existing callers catching sync errors no longer handle that conflict.
Learn more
Previously a 409 conflict from run_sync propagated as AirbyteConnectionSyncError. The new running-job branch replaces it with AirbyteConnectionSyncActiveError, whose superclass is AirbyteCloudConnectionError. That class does not inherit from AirbyteConnectionSyncError. Code that catches sync errors to handle failed starts now misses the active-sync case.
Example: try: connection.run_sync(); except AirbyteConnectionSyncError: recover() used to enter recover() on a running-sync 409. With a running job detected, the new active exception bypasses it.
Recommended fix: Preserve the AirbyteConnectionSyncError catch contract for failed sync starts, either by adjusting exception inheritance with consideration for the pre-existing HTTP 423 state-update use, or by providing a compatible exception type for this path.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
🙋 Human Input Needed: should the already-running case still raise an exception? The break is real: AirbyteConnectionSyncActiveError subclasses AirbyteCloudConnectionError, not AirbyteConnectionSyncError, so except AirbyteConnectionSyncError no longer catches a running-sync 409. I've proposed that run_cloud_sync return a normal "already running" result instead of raising, which removes the break. If the exception stays, I'll keep raising AirbyteConnectionSyncError and put the job ID in its message and context.
Overview
👉 TL;DR: When Airbyte Cloud refuses to start a sync with a conflict error, agents now get PyAirbyte's explanation and next steps. Before, they only got Cloud's terse "State conflict" text and retried.
Specifically,
format_user_facing_error/agent_error_textkeep a local error's message +guidanceahead of the Cloud detail, andCloudConnection.run_sync()turns409 try-again-laterintoAirbyteConnectionSyncActiveErrorcarrying the running job's ID.Closes #1403
Changes
airbyte/mcp/_error_handling.py: when the chain holds a Cloud problem and a localAirbyteLibErrorwithguidance, the agent text becomes<local message> <local guidance> Airbyte Cloud response: <cloud detail>. The suffix is omitted when the Cloud detail is already in the local text. Local errors without guidance still show only the Cloud text (+ Cloud guidance), as before. Span trace messages are unchanged.CloudConnection.run_sync(): on409 try-again-laterfor an enabled connection, it looks up the newest sync job. If that job is still running, it raisesAirbyteConnectionSyncActiveErrorwith the job ID, the job URL, and next steps: monitor withget_sync_result/get_cloud_sync_status, cancel withcancel_sync/cancel_cloud_synconly if a restart is wanted, and don't retry. If no running job is found, the original error is re-raised.AirbyteConnectionSyncActiveError: docstring widened from the HTTP 423 state-update case to "a sync is currently running"; added toMCP_TOOL_USER_FACING_ERRORS.run_cloud_syncdocstring: covers the disabled-connection 409, the already-running 409, and what to do after await=Truetimeout.Review Spotlight
Reviewers with limited time, please review first:
_error_handling.py#L200-L224: new local-guidance-first ruleconnections.py#L400-L447: already-running detection + running-job lookupRisks
guidance: agents now see the local text first. One refactor(mcp): relay Cloud problem detail instead of the fixed error manifest #1368 test case (nestedcarrier) changes its expected output for this reason.get_previous_sync_logs(limit=1)), and only after Cloud has already rejected the sync withtry-again-later.Test plan
tests/unit_tests/test_mcp_tool_errors.py: disabled-connection chain, duplicate-message suppression, active-sync formatting.tests/unit_tests/test_cloud_connections.py: try-again-later with a running job, with no job, and the disabled connection still taking priority.ruff check,ruff format --check, docstring lint, andpyrefly check: all pass.Link to Devin session: https://app.devin.ai/sessions/1fd07730c0494565ba33fec3854b26d2
Open in Devin Desktop: https://app.devin.ai/desktop/session/1fd07730c0494565ba33fec3854b26d2?variant=devin
Requested by: Aaron ("AJ") Steers (@aaronsteers)