fix(cloud): keep PyAirbyte guidance on run_cloud_sync 409s and report the running job ID

airbyte/cloud/connections.py

@@ -9,6 +9,7 @@
 
 from airbyte._direct_connectors import api_util as agents_api_util
 from airbyte._util import api_util
+from airbyte._util.cloud_errors import parse_cloud_error
 from airbyte.cloud._connection_catalog import (
     _denormalize_catalog_to_api,
     _is_protocol_catalog_format,
@@ -35,6 +36,7 @@
 from airbyte.cloud.sync_results import SyncResult
 from airbyte.exceptions import (
     AirbyteCloudError,
+    AirbyteConnectionSyncActiveError,
     AirbyteConnectionSyncError,
     AirbyteLibInputError,
     AirbyteMissingResourceError,
@@ -375,6 +377,10 @@ def run_sync(
                     ),
                     context={"connection_id": self.connection_id},
                 ) from ex
+            if ex.context and ex.context.get("status_code") == HTTPStatus.CONFLICT:
+                active_sync_error = self._already_running_sync_error(ex)
+                if active_sync_error is not None:
+                    raise active_sync_error from ex
             raise
         sync_result = SyncResult(
             workspace=self.workspace,
@@ -391,6 +397,55 @@ def run_sync(
 
         return sync_result
 
+    def _already_running_sync_error(
+        self,
+        error: AirbyteConnectionSyncError,
+    ) -> AirbyteConnectionSyncActiveError | None:
+        """Return actionable guidance when Cloud rejects a duplicate running sync."""
+        sdk_error = error.__cause__
+        if not isinstance(sdk_error, api_util.SDKError):
+            return None
+
+        problem = parse_cloud_error(
+            sdk_error.status_code,
+            api_util.error_response_body(sdk_error),
+        )
+        if problem.slug != "try-again-later":
+            return None
+
+        sync_results = self.get_previous_sync_logs(
+            limit=1,
+            job_type=JobTypeEnum.SYNC,
+        )
+        sync_result = sync_results[0] if sync_results else None
+        if sync_result is None or sync_result.is_job_complete():
+            return None
+
+        job_id = sync_result.job_id
+        return AirbyteConnectionSyncActiveError(
+            connection_id=self.connection_id,
+            message=(
+                f"A sync is already running for connection '{self.connection_id}' "
+                f"(job {job_id})."
+            ),
+            guidance=(
+                "Don't start another sync. Monitor the running job with "
+                f"`connection.get_sync_result(job_id={job_id})` or the "
+                "`get_cloud_sync_status` MCP tool "
+                f"(`connection_id='{self.connection_id}', job_id={job_id}`). "
+                "Only if the user wants to restart it, cancel it first with "
+                f"`connection.cancel_sync(job_id={job_id})` or the "
+                "`cancel_cloud_sync` MCP tool "
+                f"(`connection_id='{self.connection_id}', job_id={job_id}`), "
+                f"then run the sync again. Job URL: {sync_result.job_url}."
+            ),
+            context={
+                "connection_id": self.connection_id,
+                "job_id": job_id,
+                "job_url": sync_result.job_url,
+            },
+        )
+
     def _get_latest_cancellable_sync_job_id(self) -> int:
         """Get the latest cancellable sync job ID."""
         sync_results = self.get_previous_sync_logs(

airbyte/exceptions.py

@@ -698,7 +698,7 @@ class AirbyteConnectionSyncError(AirbyteCloudConnectionError):
 
 @dataclass
 class AirbyteConnectionSyncActiveError(AirbyteCloudConnectionError):
-    """State update rejected because a sync is currently running (HTTP 423)."""
+    """Rejected because a sync is currently running on this connection."""
 
 
 @dataclass

airbyte/mcp/_error_handling.py

@@ -1,5 +1,5 @@
 # Copyright (c) 2026 Airbyte, Inc., all rights reserved.
-"""PyAirbyte-specific handling and trace classification of MCP tool errors."""
+"""Format MCP tool errors with Cloud details and local recovery guidance."""
 
 from __future__ import annotations
 
@@ -28,6 +28,8 @@
 from airbyte.exceptions import (
     AirbyteAgentsUnavailableError,
     AirbyteCloudApiError,
+    AirbyteCloudError,
+    AirbyteConnectionSyncActiveError,
     AirbyteConnectionSyncError,
     AirbyteConnectionSyncTimeoutError,
     AirbyteConnectorInUseError,
@@ -90,6 +92,7 @@
     AirbyteDestructiveToolsDisabledError,
     AirbyteExternalAccessDisabledError,
     AirbyteAgentsUnavailableError,
+    AirbyteConnectionSyncActiveError,
     AirbyteConnectorNotReadyError,
     AirbyteConnectorInUseError,
     AirbyteMissingResourceError,
@@ -194,10 +197,38 @@ def _cloud_problem_text(problem: CloudErrorInfo) -> str:
     return f"{problem.message} {guidance}".rstrip() if guidance else problem.message
 
 
-def format_user_facing_error(error: BaseException) -> str:
-    """Return the error message followed by its guidance, when present."""
-    if (problem := _cloud_error_info(error)) is not None:
+def _format_cloud_error_with_local_guidance(error: BaseException) -> str | None:
+    """Keep local recovery guidance alongside a selected Cloud problem message."""
+    problem = _cloud_error_info(error)
+    if problem is None:
+        return None
+
+    local_error = next(
+        (
+            cause
+            for cause in _chain(error)
+            if isinstance(cause, AirbyteLibError)
+            and not isinstance(cause, AirbyteCloudApiError)
+            and not (
+                isinstance(cause, AirbyteCloudError) and cause.get_message() == problem.message
+            )
+        ),
+        None,
+    )
+    if local_error is None or not local_error.guidance:
         return _cloud_problem_text(problem)
+
+    local_message = local_error.get_message()
+    local_text = f"{local_message} {local_error.guidance}"
+    if problem.message == local_message or problem.message in local_text:
+        return local_text
+    return f"{local_text} Airbyte Cloud response: {problem.message}"
+
+
+def format_user_facing_error(error: BaseException) -> str:
+    """Format local error guidance and Cloud problem details without debug context."""
+    if text := _format_cloud_error_with_local_guidance(error):
+        return text
     if not isinstance(error, AirbyteLibError):
         return str(error)
     text = error.get_message()
@@ -241,9 +272,9 @@ def mcp_tool_error_reason(error: BaseException) -> str | None:
 
 
 def agent_error_text(error: ToolError) -> str:
-    """Return the text an agent sees for a tool error before redaction."""
-    if (problem := _cloud_error_info(error)) is not None:
-        return _cloud_problem_text(problem)
+    """Return local guidance and Cloud details visible to an agent before redaction."""
+    if text := _format_cloud_error_with_local_guidance(error):
+        return text
     cause = error.__cause__
     if isinstance(cause, AirbyteLibError):
         return format_user_facing_error(cause)

airbyte/mcp/cloud.py

@@ -1006,7 +1006,14 @@ def run_cloud_sync(
         ),
     ],
 ) -> str:
-    """Run a sync job on Airbyte Cloud."""
+    """Run a sync job on Airbyte Cloud.
+
+    A disabled connection returns 409; re-enable it with `update_cloud_connection`
+    (`enabled=True`) first. If a sync is already running, the error includes its job ID:
+    monitor it with `get_cloud_sync_status` instead of retrying, or call `cancel_cloud_sync`
+    first if a restart is really wanted. After a `wait=True` timeout, the sync keeps running;
+    check `get_cloud_sync_status` rather than starting another sync.
+    """
     workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)
     connection = workspace.get_connection(connection_id=connection_id)
     sync_result = connection.run_sync(wait=wait, wait_timeout=wait_timeout)

tests/unit_tests/test_cloud_connections.py

@@ -21,6 +21,7 @@
 from airbyte.cloud.workspaces import CloudWorkspace
 from airbyte.exceptions import (
     AirbyteCloudApiError,
+    AirbyteConnectionSyncActiveError,
     AirbyteConnectionSyncError,
     AirbyteCloudError,
     AirbyteMissingResourceError,
@@ -826,26 +827,92 @@ def _connection_info_with_status(status: str) -> MagicMock:
     return info
 
 
-def test_run_sync_conflict_on_disabled_connection_raises_input_error(
+@pytest.mark.parametrize(
+    ("scenario", "connection_status"),
+    [
+        pytest.param("running-job", "active", id="running-job"),
+        pytest.param("no-jobs", "active", id="no-jobs"),
+        pytest.param("completed-job", "active", id="completed-job"),
+        pytest.param("disabled", "inactive", id="disabled-connection"),
+    ],
+)
+def test_run_sync_handles_try_again_later_conflicts(
     monkeypatch: pytest.MonkeyPatch,
+    scenario: str,
+    connection_status: str,
 ) -> None:
-    """Verify a 409 on an inactive connection surfaces a disabled-connection hint."""
+    """Explain known running-sync conflicts and preserve unrelated errors."""
     connection = _connection()
+    body = (
+        '{"type":"https://reference.airbyte.com/reference/errors#try-again-later",'
+        '"title":"try-again-later",'
+        '"detail":"A sync is already running for: connection-id"}'
+    )
+    sdk_error = SDKError("Status 409", 409, body, MagicMock())
+    sync_error = _sync_error(409)
+    sync_error.__cause__ = sdk_error
 
     def run_connection(**kwargs: object) -> None:
         """Raise the wrapped 409 sync error."""
         _ = kwargs
-        raise _sync_error(409)
+        raise sync_error
 
     monkeypatch.setattr(api_util, "run_connection", run_connection)
-    fetch_mock = MagicMock(return_value=_connection_info_with_status("inactive"))
+    fetch_mock = MagicMock(return_value=_connection_info_with_status(connection_status))
     monkeypatch.setattr(connection, "_fetch_connection_info", fetch_mock)
 
-    with pytest.raises(AirbyteLibInputError) as exc_info:
-        connection.run_sync()
+    if scenario == "running-job":
+        sync_result = MagicMock(
+            job_id=42,
+            job_url="https://cloud.airbyte.com/jobs/42",
+            is_job_complete=MagicMock(return_value=False),
+        )
+        get_logs = MagicMock(return_value=[sync_result])
+        monkeypatch.setattr(connection, "get_previous_sync_logs", get_logs)
+
+        with pytest.raises(AirbyteConnectionSyncActiveError) as exc_info:
+            connection.run_sync()
+
+        assert "job 42" in exc_info.value.get_message()
+        assert exc_info.value.context == {
+            "connection_id": "connection-id",
+            "job_id": 42,
+            "job_url": "https://cloud.airbyte.com/jobs/42",
+        }
+        assert "`get_cloud_sync_status`" in exc_info.value.guidance
+        get_logs.assert_called_once_with(limit=1, job_type=JobTypeEnum.SYNC)
+        assert exc_info.value.__cause__ is sync_error
+    elif scenario in {"no-jobs", "completed-job"}:
+        sync_results = (
+            []
+            if scenario == "no-jobs"
+            else [
+                MagicMock(
+                    job_id=41,
+                    is_job_complete=MagicMock(return_value=True),
+                )
+            ]
+        )
+        monkeypatch.setattr(
+            connection,
+            "get_previous_sync_logs",
+            MagicMock(return_value=sync_results),
+        )
+
+        with pytest.raises(AirbyteConnectionSyncError) as exc_info:
+            connection.run_sync()
+
+        assert exc_info.value is sync_error
+    else:
+        get_logs = MagicMock()
+        monkeypatch.setattr(connection, "get_previous_sync_logs", get_logs)
+
+        with pytest.raises(AirbyteLibInputError) as exc_info:
+            connection.run_sync()
 
-    assert "disabled" in exc_info.value.message
-    assert "enabled=True" in exc_info.value.guidance
+        assert "disabled" in exc_info.value.message
+        assert "enabled=True" in exc_info.value.guidance
+        get_logs.assert_not_called()
     fetch_mock.assert_called_once_with(force_refresh=True)