From f71eee2933aa7913851f5c58fcf802fa86cbefbc Mon Sep 17 00:00:00 2001 From: "caleb.evans" Date: Mon, 26 Oct 2026 19:30:17 +0000 Subject: [PATCH] feat(mcp): include job duration and end time in sync job results Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- airbyte/cloud/models.py | 10 ++++ airbyte/cloud/sync_results.py | 13 +++++ airbyte/mcp/cloud.py | 9 ++++ tests/unit_tests/test_cloud_api_util.py | 24 +++++++++ tests/unit_tests/test_mcp_cloud.py | 69 +++++++++++++++++++++++++ 5 files changed, 125 insertions(+) diff --git a/airbyte/cloud/models.py b/airbyte/cloud/models.py index 59caccd83..20b523167 100644 --- a/airbyte/cloud/models.py +++ b/airbyte/cloud/models.py @@ -57,6 +57,8 @@ class _JobResponseLike(Protocol): bytes_synced: int | None rows_synced: int | None start_time: str + duration: str | None + last_updated_at: str | None class ConnectionStatus(StrEnum): @@ -470,6 +472,12 @@ class CloudJobInfo(BaseModel): start_time: str """The job start time.""" + duration: str | None = None + """The job's ISO 8601 duration (e.g. `PT1M9S`), if available.""" + + last_updated_at: str | None = None + """When the job was last updated, if available. For a finished job, this is when it ended.""" + @classmethod def from_api_response(cls, job: _JobResponseLike) -> CloudJobInfo: """Create a public model from an internal API job response.""" @@ -479,6 +487,8 @@ def from_api_response(cls, job: _JobResponseLike) -> CloudJobInfo: bytes_synced=job.bytes_synced, rows_synced=job.rows_synced, start_time=job.start_time, + duration=job.duration, + last_updated_at=job.last_updated_at, ) diff --git a/airbyte/cloud/sync_results.py b/airbyte/cloud/sync_results.py index 3a15cb94f..48311823b 100644 --- a/airbyte/cloud/sync_results.py +++ b/airbyte/cloud/sync_results.py @@ -337,6 +337,19 @@ def start_time(self) -> datetime: return ab_datetime_parse(raw_start_time) raise + @property + def duration(self) -> str | None: + """The job's ISO 8601 duration (e.g. `PT1M9S`), if available.""" + return self._fetch_latest_job_info().duration + + @property + def last_updated_at(self) -> str | None: + """When the job was last updated, if available. + + For a finished job, this is when it ended. + """ + return self._fetch_latest_job_info().last_updated_at + def _fetch_job_with_attempts(self) -> dict[str, Any]: """Fetch job info with attempts from Config API using lazy loading pattern.""" if self._job_with_attempts_info is not None: diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index 3682dbc46..28787615e 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -535,6 +535,11 @@ class SyncJobResult(BaseModel): """Number of records synced in this job.""" start_time: str """ISO 8601 timestamp of when the job started.""" + duration: str | None = None + """ISO 8601 duration of the job (e.g. `PT1M9S` is 1 minute 9 seconds), or null if Cloud has + not reported one (e.g. the job is still running).""" + last_updated_at: str | None = None + """ISO 8601 timestamp of the job's last update. For a finished job, this is when it ended.""" job_url: str """URL to view the job in Airbyte Cloud.""" @@ -1130,6 +1135,8 @@ def get_cloud_sync_status( "bytes_synced": sync_result.bytes_synced, "records_synced": sync_result.records_synced, "start_time": sync_result.start_time.isoformat(), + "duration": sync_result.duration, + "last_updated_at": sync_result.last_updated_at, "job_url": sync_result.job_url, "attempts": [], } @@ -1232,6 +1239,8 @@ def list_cloud_sync_jobs( bytes_synced=sync_result.bytes_synced, records_synced=sync_result.records_synced, start_time=sync_result.start_time.isoformat(), + duration=sync_result.duration, + last_updated_at=sync_result.last_updated_at, job_url=sync_result.job_url, ) for sync_result in sync_results diff --git a/tests/unit_tests/test_cloud_api_util.py b/tests/unit_tests/test_cloud_api_util.py index f4907260b..c8f2ded4f 100644 --- a/tests/unit_tests/test_cloud_api_util.py +++ b/tests/unit_tests/test_cloud_api_util.py @@ -1573,6 +1573,30 @@ def test_sdk_decodes_queued_job_status() -> None: assert CloudJobInfo.from_api_response(job).status is JobStatusEnum.QUEUED +def test_cloud_job_info_preserves_duration_and_last_updated_at() -> None: + payload = {"data": [_raw_job_response(42, "succeeded")]} + payload["data"][0].update( + duration="PT1M9S", + lastUpdatedAt="2026-10-19T01:18:28Z", + ) + job = utils.unmarshal_json(json.dumps(payload), models.JobsResponse).data[0] + + job_info = CloudJobInfo.from_api_response(job) + + assert job_info.duration == "PT1M9S" + assert job_info.last_updated_at == "2026-10-19T01:18:28Z" + + +def test_cloud_job_info_defaults_missing_duration_and_last_updated_at_to_none() -> None: + payload = {"data": [_raw_job_response(42, "running")]} + job = utils.unmarshal_json(json.dumps(payload), models.JobsResponse).data[0] + + job_info = CloudJobInfo.from_api_response(job) + + assert job_info.duration is None + assert job_info.last_updated_at is None + + def test_get_job_logs_paginates_until_limit(monkeypatch: pytest.MonkeyPatch) -> None: """Verify job log pagination stops after collecting the requested limit.""" captured_requests: list[api.ListJobsRequest] = [] diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py index 51a7c44c4..dd99ab74a 100644 --- a/tests/unit_tests/test_mcp_cloud.py +++ b/tests/unit_tests/test_mcp_cloud.py @@ -72,6 +72,8 @@ class _SyncResultLike: start_time: datetime bytes_synced: int = 0 records_synced: int = 0 + duration: str | None = None + last_updated_at: str | None = None job_url: str = "https://cloud.airbyte.com/jobs" def get_job_status(self) -> JobStatusEnum: @@ -1061,6 +1063,71 @@ def _patch_sync_status_workspace( return workspace +def test_list_cloud_sync_jobs_includes_duration_and_last_updated_at( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Finished jobs include Cloud timing metadata; running jobs may omit it.""" + finished_job = _SyncResultLike( + job_id=42, + status=JobStatusEnum.SUCCEEDED, + start_time=datetime(2026, 10, 19, 1, 17, 19, tzinfo=timezone.utc), + duration="PT1M9S", + last_updated_at="2026-10-19T01:18:28Z", + ) + running_job = _SyncResultLike( + job_id=43, + status=JobStatusEnum.RUNNING, + start_time=datetime(2026, 10, 19, 1, 17, 19, tzinfo=timezone.utc), + ) + connection = SimpleNamespace( + get_previous_sync_logs=MagicMock(return_value=[finished_job, running_job]) + ) + _patch_sync_status_workspace(monkeypatch, connection) + + result = cloud_mcp.list_cloud_sync_jobs( + ctx=cast(Context, object()), + connection_id="connection-1", + workspace_id=None, + max_jobs=20, + from_tail=None, + job_type=None, + ) + + assert result.jobs[0].duration == "PT1M9S" + assert result.jobs[0].last_updated_at == "2026-10-19T01:18:28Z" + assert result.jobs[1].duration is None + assert result.jobs[1].last_updated_at is None + + +def test_get_cloud_sync_status_includes_duration_and_last_updated_at( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The status response includes the Cloud job duration and update timestamp.""" + sync_result = SimpleNamespace( + job_id=42, + get_job_status=MagicMock(return_value=JobStatusEnum.SUCCEEDED), + bytes_synced=123, + records_synced=456, + start_time=datetime(2026, 10, 19, 1, 17, 19, tzinfo=timezone.utc), + duration="PT1M9S", + last_updated_at="2026-10-19T01:18:28Z", + job_url="https://cloud.airbyte.com/jobs/42", + ) + connection = SimpleNamespace(get_sync_result=MagicMock(return_value=sync_result)) + _patch_sync_status_workspace(monkeypatch, connection) + + result = cloud_mcp.get_cloud_sync_status( + ctx=cast(Context, object()), + connection_id="connection-1", + job_id=42, + workspace_id=None, + include_attempts=False, + ) + + assert result["duration"] == "PT1M9S" + assert result["last_updated_at"] == "2026-10-19T01:18:28Z" + + @pytest.mark.parametrize( "failure_stage", [ @@ -1086,6 +1153,8 @@ def test_get_cloud_sync_status_surfaces_library_errors( bytes_synced=0, records_synced=0, start_time=datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc), + duration=None, + last_updated_at=None, job_url="https://cloud.airbyte.com/jobs", get_attempts=get_attempts, )