test(api): type KnowledgeFS task tests for strict pyrefly

Make the shared facade fake return payloads checked against their real DTO
type, record failed-retrieval captures as typed calls, narrow nested JSON
metadata before indexing, and annotate the investigation task tests.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

api/tests/unit_tests/tasks/knowledge_fs_task_fakes.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 from collections.abc import Callable, Mapping
 from dataclasses import dataclass, field
+from typing import TypeVar, cast
 
 from celery.exceptions import Retry
 
@@ -25,6 +26,8 @@ from services.knowledge_fs.product_dto import (
     KnowledgeFSSourceWorkflowResponse,
 )
 
+_PayloadT = TypeVar("_PayloadT")
+
 
 @dataclass
 class RecordingKnowledgeFSMetrics:
@@ -85,9 +88,11 @@ class RecordingKnowledgeFSFacade:
     def call_count(self, method: str) -> int:
         return len(self.calls.get(method, []))
 
-    def payload(self, method: str) -> object:
-        """Return the ``payload`` keyword of the most recent call to ``method``."""
-        return self.calls[method][-1]["payload"]
+    def payload(self, method: str, payload_type: type[_PayloadT]) -> _PayloadT:
+        """Return the ``payload`` keyword of the most recent call to ``method``, checked against ``payload_type``."""
+        payload = self.calls[method][-1]["payload"]
+        assert isinstance(payload, payload_type), f"{method} payload is {type(payload).__name__}"
+        return payload
 
     def _respond(self, method: str, kwargs: dict[str, object]) -> object:
         self.calls.setdefault(method, []).append(kwargs)
@@ -139,6 +144,12 @@ class RecordingKnowledgeFSFacade:
         return self._respond("update_source_sync_policy", kwargs)
 
 
+def json_object(value: object) -> dict[str, object]:
+    """Narrow recorded JSON metadata, or an object nested in it, to a string-keyed dict."""
+    assert isinstance(value, dict), f"expected a JSON object, got {type(value).__name__}"
+    return cast(dict[str, object], value)
+
+
 @dataclass(frozen=True)
 class FacadeRuntime:
     """The ``KnowledgeFSRuntime`` member the source tasks read."""

api/tests/unit_tests/tasks/test_knowledge_fs_agent_investigation_tasks.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass, field
+from typing import Literal
 from uuid import uuid4
 
 import pytest
@@ -25,19 +26,20 @@ class _Gateway:
             raise self.error
 
 
-def report():
+def report() -> tuple[dict[str, object], str]:
     space = str(uuid4())
+    attempts: list[dict[str, object]] = []
     return {
         "investigation_id": str(uuid4()),
         "execution_context": {"tenant_id": "tenant", "agent_mode": "agent_app", "invoke_from": "published"},
         "bindings": [{"id": "docs", "name": "Docs", "control_space_id": space}],
         "query": "refund",
         "status": "completed",
-        "attempts": [],
+        "attempts": attempts,
     }, space
 
 
-def test_dataset_worker_reconstructs_report_for_each_delivery(monkeypatch):
+def test_dataset_worker_reconstructs_report_for_each_delivery(monkeypatch: pytest.MonkeyPatch) -> None:
     gateway = _Gateway()
     monkeypatch.setattr(module, "AgentKnowledgeGateway", lambda: gateway)
     payload, space = report()
@@ -51,7 +53,7 @@ def test_dataset_worker_reconstructs_report_for_each_delivery(monkeypatch):
     assert str(first[0].investigation_id) == payload["investigation_id"]
 
 
-def test_revoked_authority_stops_publication_and_transient_failures_retry(monkeypatch):
+def test_revoked_authority_stops_publication_and_transient_failures_retry(monkeypatch: pytest.MonkeyPatch) -> None:
     gateway = _Gateway(error=KnowledgeFsError("KNOWLEDGE_SCOPE_CHANGED", "revoked", 403))
     monkeypatch.setattr(module, "AgentKnowledgeGateway", lambda: gateway)
     payload, space = report()
@@ -67,7 +69,9 @@ def test_revoked_authority_stops_publication_and_transient_failures_retry(monkey
 
 
 @pytest.mark.parametrize("status_code", [400, 403, 409, 413, 422])
-def test_permanent_remote_rejections_do_not_retry(monkeypatch, status_code):
+def test_permanent_remote_rejections_do_not_retry(
+    monkeypatch: pytest.MonkeyPatch, status_code: Literal[400, 403, 409, 413, 422]
+) -> None:
     gateway = _Gateway(error=KnowledgeFSProductRequestRejectedError(status_code=status_code))
     monkeypatch.setattr(module, "AgentKnowledgeGateway", lambda: gateway)
     task = module.capture_agent_knowledge_investigation_task
@@ -82,7 +86,9 @@ def test_permanent_remote_rejections_do_not_retry(monkeypatch, status_code):
 
 
 @pytest.mark.parametrize(("retries", "countdown"), [(0, 30), (1, 60), (2, 120)])
-def test_rate_limited_report_retries_with_backoff(monkeypatch, retries, countdown):
+def test_rate_limited_report_retries_with_backoff(
+    monkeypatch: pytest.MonkeyPatch, retries: int, countdown: int
+) -> None:
     error = KnowledgeFSProductRequestRejectedError(status_code=429)
     gateway = _Gateway(error=error)
     monkeypatch.setattr(module, "AgentKnowledgeGateway", lambda: gateway)
@@ -98,7 +104,7 @@ def test_rate_limited_report_retries_with_backoff(monkeypatch, retries, countdow
     assert retry.calls == [{"exc": error, "countdown": countdown}]
 
 
-def test_malformed_queued_report_is_rejected_before_gateway_publication(monkeypatch):
+def test_malformed_queued_report_is_rejected_before_gateway_publication(monkeypatch: pytest.MonkeyPatch) -> None:
     gateway = _Gateway()
     monkeypatch.setattr(module, "AgentKnowledgeGateway", lambda: gateway)
     task = module.capture_agent_knowledge_investigation_task

api/tests/unit_tests/tasks/test_knowledge_fs_failed_retrieval_tasks.py

@@ -1,14 +1,19 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Literal, TypedDict
+from typing import Literal
 
 import pytest
 from celery.app.task import Context
 from celery.exceptions import Retry
+from typing_extensions import TypedDict
 
+from core.knowledge_fs.resource import KnowledgeResourceRef
 from services.knowledge_fs.app_admission_service import KnowledgeFSAppAdmissionError
-from services.knowledge_fs.product_dto import KnowledgeFSWorkflowFailedRetrievalCaptureResponse
+from services.knowledge_fs.product_dto import (
+    KnowledgeFSWorkflowFailedRetrievalCapturePayload,
+    KnowledgeFSWorkflowFailedRetrievalCaptureResponse,
+)
 from services.knowledge_fs.product_remote import KnowledgeFSProductRequestRejectedError
 from tasks import knowledge_fs_failed_retrieval_tasks as task_module
 from tests.unit_tests.tasks.knowledge_fs_task_fakes import RecordingRetry
@@ -17,7 +22,7 @@ from tests.unit_tests.tasks.task_options import task_options
 EVENT_ID = "019fac9f-bfb0-75ee-9af5-252ebafbac1e"
 
 
-class FailedRetrievalTaskArguments(TypedDict):
+class FailedRetrievalTaskArguments(TypedDict, closed=True):
     event_id: str
     tenant_id: str
     app_id: str
@@ -39,13 +44,28 @@ def _task_kwargs() -> FailedRetrievalTaskArguments:
     }
 
 
+@dataclass(frozen=True)
+class _CaptureCall:
+    tenant_id: str
+    app_id: str
+    resource: KnowledgeResourceRef
+    payload: KnowledgeFSWorkflowFailedRetrievalCapturePayload
+
+
 @dataclass
 class _AppCapabilities:
     outcome: KnowledgeFSWorkflowFailedRetrievalCaptureResponse | Exception
-    calls: list[dict[str, object]] = field(default_factory=list)
+    calls: list[_CaptureCall] = field(default_factory=list)
 
-    def capture_workflow_failed_retrieval(self, **kwargs: object) -> KnowledgeFSWorkflowFailedRetrievalCaptureResponse:
-        self.calls.append(kwargs)
+    def capture_workflow_failed_retrieval(
+        self,
+        *,
+        tenant_id: str,
+        app_id: str,
+        resource: KnowledgeResourceRef,
+        payload: KnowledgeFSWorkflowFailedRetrievalCapturePayload,
+    ) -> KnowledgeFSWorkflowFailedRetrievalCaptureResponse:
+        self.calls.append(_CaptureCall(tenant_id=tenant_id, app_id=app_id, resource=resource, payload=payload))
         if isinstance(self.outcome, Exception):
             raise self.outcome
         return self.outcome
@@ -91,12 +111,12 @@ def test_failed_retrieval_task_runs_on_dataset_queue_and_reauthorizes_app(
     assert task_options(task_module.capture_workflow_failed_retrieval_task)["queue"] == "dataset"
     assert len(capability.calls) == 1
     call = capability.calls[0]
-    assert call["tenant_id"] == "tenant-1"
-    assert call["app_id"] == "app-1"
-    assert call["resource"].control_space_id == "control-1"
-    assert str(call["payload"].event_id) == EVENT_ID
-    assert call["payload"].query == "missing answer"
-    assert call["payload"].retrieval_trace_id == "retrieval-trace-1"
+    assert call.tenant_id == "tenant-1"
+    assert call.app_id == "app-1"
+    assert call.resource.control_space_id == "control-1"
+    assert str(call.payload.event_id) == EVENT_ID
+    assert call.payload.query == "missing answer"
+    assert call.payload.retrieval_trace_id == "retrieval-trace-1"
 
 
 def test_enqueue_is_idempotency_aware_and_never_propagates_broker_failure(monkeypatch: pytest.MonkeyPatch) -> None:

api/tests/unit_tests/tasks/test_knowledge_fs_initial_source_tasks.py

@@ -10,9 +10,15 @@ from models.account import Account
 from models.knowledge_fs import KnowledgeFSControlSpace, KnowledgeFSControlSpaceState
 from models.oauth import DatasourceProvider
 from services.knowledge_fs.product_dto import (
+    KnowledgeFSCrawlImportPayload,
     KnowledgeFSInitialOnlineDocumentSourcePayload,
     KnowledgeFSInitialOnlineDriveSourcePayload,
     KnowledgeFSInitialWebsiteSourcePayload,
+    KnowledgeFSSourceConnectionCreatePayload,
+    KnowledgeFSSourceCreatePayload,
+    KnowledgeFSSourceSyncPolicyPayload,
+    KnowledgeFSSourceUpdatePayload,
+    KnowledgeFSSourceWorkflowImportPayload,
     knowledge_fs_initial_preview_configuration_fingerprint,
 )
 from services.knowledge_fs.product_remote import (
@@ -31,6 +37,7 @@ from tasks.knowledge_fs_initial_source_tasks import (
 from tests.unit_tests.tasks.knowledge_fs_task_fakes import (
     FacadeRuntime,
     RecordingKnowledgeFSFacade,
+    json_object,
     source_connection_list_response,
     source_connection_response,
     source_list_response,
@@ -286,7 +293,7 @@ def test_initial_website_source_import_recrawls_exact_selection_and_configures_d
 
     assert _start(facade, _payload()) == "workflow-1"
 
-    source_payload = facade.payload("create_source")
+    source_payload = facade.payload("create_source", KnowledgeFSSourceCreatePayload)
     assert source_payload.status == "disabled"
     assert source_payload.connection_id == "connection-1"
     assert source_payload.metadata["datasourceParameterMode"] == "exact"
@@ -299,16 +306,17 @@ def test_initial_website_source_import_recrawls_exact_selection_and_configures_d
         "limit": 25,
         "url": "https://docs.dify.ai",
     }
-    import_payload = facade.payload("import_selected_source_crawl")
+    import_payload = facade.payload("import_selected_source_crawl", KnowledgeFSCrawlImportPayload)
     assert import_payload.source_urls == [
         "https://docs.dify.ai/a",
         "https://docs.dify.ai/b",
     ]
-    source_update_payload = facade.payload("update_source")
+    source_update_payload = facade.payload("update_source", KnowledgeFSSourceUpdatePayload)
+    source_update_metadata = json_object(source_update_payload.metadata)
     assert source_update_payload.expected_version == 3
-    assert source_update_payload.metadata == {"preview": False}
+    assert source_update_metadata == {"preview": False}
     assert source_update_payload.status == "active"
-    sync_payload = facade.payload("update_source_sync_policy")
+    sync_payload = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload)
     assert sync_payload.mode == "interval"
     assert sync_payload.enabled is True
     assert sync_payload.expected_revision == 0
@@ -334,7 +342,7 @@ def test_initial_website_source_import_preserves_distinct_urls_with_the_same_can
 
     assert _start(facade, payload) == "workflow-1"
 
-    import_payload = facade.payload("import_selected_source_crawl")
+    import_payload = facade.payload("import_selected_source_crawl", KnowledgeFSCrawlImportPayload)
     assert import_payload.source_urls == [
         "https://docs.dify.ai/a/",
         "https://DOCS.dify.ai:443/a#section",
@@ -346,7 +354,7 @@ def test_initial_website_source_import_preserves_legacy_crawl_option_projection(
 
     assert _start(facade, _legacy_payload()) == "workflow-1"
 
-    source_payload = facade.payload("create_source")
+    source_payload = facade.payload("create_source", KnowledgeFSSourceCreatePayload)
     assert "datasourceParameterMode" not in source_payload.metadata
     assert "parameters" not in source_payload.metadata
     assert source_payload.metadata["crawlOptions"] == {
@@ -436,7 +444,7 @@ def test_initial_connector_source_import_uses_exact_binding_and_selection(
         )
 
     assert result == "connector-workflow-1"
-    connection_payload = facade.payload("create_source_connection")
+    connection_payload = facade.payload("create_source_connection", KnowledgeFSSourceConnectionCreatePayload)
     assert connection_payload.provider_id == provider_id
     assert connection_payload.configuration == {
         "credentialId": credential_id,
@@ -445,17 +453,17 @@ def test_initial_connector_source_import_uses_exact_binding_and_selection(
         "provider": payload.provider,
         "providerKind": "online-document" if workflow_kind == "online-document-import" else "online-drive",
     }
-    source_payload = facade.payload("create_source")
+    source_payload = facade.payload("create_source", KnowledgeFSSourceCreatePayload)
     assert source_payload.connection_id == "connector-1"
     assert source_payload.name == expected_source_name
     assert source_payload.type == "connector"
     assert source_payload.uri == "connector://connector-1"
     assert source_payload.metadata["providerId"] == provider_id
     assert source_payload.metadata["providerName"] == expected_provider_name
-    import_payload = facade.payload("import_source_workflow").root
+    import_payload = facade.payload("import_source_workflow", KnowledgeFSSourceWorkflowImportPayload).root
     assert import_payload.kind == workflow_kind
     assert import_payload.items == payload.selection
-    sync_payload = facade.payload("update_source_sync_policy")
+    sync_payload = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload)
     assert sync_payload.mode == ("manual" if workflow_kind == "online-drive-import" else "interval")
 
 
@@ -488,12 +496,13 @@ def test_initial_website_source_import_reuses_source_across_pages_and_preserves_
     assert _start(facade, _payload()) == "failed-workflow"
     assert facade.call_count("create_source") == 0
     assert facade.call_count("list_source_connections") == 0
-    source_update_payload = facade.payload("update_source")
+    source_update_payload = facade.payload("update_source", KnowledgeFSSourceUpdatePayload)
+    source_update_metadata = json_object(source_update_payload.metadata)
     assert source_update_payload.status == "disabled"
-    assert "parameters" not in source_update_payload.metadata
-    assert source_update_payload.metadata["preview"] is False
-    assert source_update_payload.metadata["initialImport"]["state"] == "failed"
-    assert source_update_payload.metadata["initialImport"]["syncPolicy"] == {
+    assert "parameters" not in source_update_metadata
+    assert source_update_metadata["preview"] is False
+    assert json_object(source_update_metadata["initialImport"])["state"] == "failed"
+    assert json_object(source_update_metadata["initialImport"])["syncPolicy"] == {
         "customIntervalSeconds": None,
         "enabled": True,
         "mode": "interval",
@@ -522,12 +531,13 @@ def test_initial_website_source_import_exposes_preview_consume_rejection() -> No
     with pytest.raises(KnowledgeFSProductRequestRejectedError):
         _start(facade, payload)
 
-    source_update = facade.payload("update_source")
+    source_update = facade.payload("update_source", KnowledgeFSSourceUpdatePayload)
+    source_update_metadata = json_object(source_update.metadata)
     assert source_update.status == "disabled"
-    assert source_update.metadata["preview"] is False
-    assert source_update.metadata["initialImport"]["state"] == "failed"
-    assert source_update.metadata["initialImport"]["errorCode"] == "SOURCE_WORKFLOW_FAILED"
-    assert source_update.metadata["initialImport"]["syncPolicy"] == {
+    assert source_update_metadata["preview"] is False
+    assert json_object(source_update_metadata["initialImport"])["state"] == "failed"
+    assert json_object(source_update_metadata["initialImport"])["errorCode"] == "SOURCE_WORKFLOW_FAILED"
+    assert json_object(source_update_metadata["initialImport"])["syncPolicy"] == {
         "customIntervalSeconds": None,
         "enabled": True,
         "mode": "interval",
@@ -567,7 +577,7 @@ def test_initial_website_source_import_configures_remaining_sync_modes(
     facade.script("get_source_sync_policy", sync_policy_response(revision=7))
 
     assert _start(facade, _payload(sync_policy)) == "workflow-1"
-    sync_payload = facade.payload("update_source_sync_policy")
+    sync_payload = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload)
     assert sync_payload.mode == expected_mode
     assert sync_payload.enabled is expected_enabled
     assert sync_payload.custom_interval_seconds == expected_custom_interval_seconds
@@ -592,7 +602,7 @@ def test_initial_website_source_import_does_not_recommit_active_source_after_ret
     assert _start(facade, _payload()) == "workflow-1"
 
     assert facade.call_count("update_source") == 0
-    sync_payload = facade.payload("update_source_sync_policy")
+    sync_payload = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload)
     assert sync_payload.expected_source_version == 4
 
 
@@ -688,7 +698,7 @@ def test_initial_website_source_import_creates_missing_firecrawl_connection() ->
 
     assert _start(facade, _payload()) == "workflow-1"
 
-    connection_payload = facade.payload("create_source_connection")
+    connection_payload = facade.payload("create_source_connection", KnowledgeFSSourceConnectionCreatePayload)
     assert connection_payload.auth_kind == "endpoint"
     assert connection_payload.configuration == {
         "credentialId": "firecrawl-credential-1",
@@ -699,7 +709,7 @@ def test_initial_website_source_import_creates_missing_firecrawl_connection() ->
     }
     assert connection_payload.credentials == {}
     assert connection_payload.provider_id == "plugin-daemon-website"
-    source_payload = facade.payload("create_source")
+    source_payload = facade.payload("create_source", KnowledgeFSSourceCreatePayload)
     assert source_payload.connection_id == "connection-created"
 
 
@@ -754,7 +764,8 @@ def test_initial_website_source_import_polls_existing_workflow_without_recreatin
     assert facade.call_count("list_source_connections") == 0
     assert facade.call_count("create_source") == 0
     assert facade.call_count("import_selected_source_crawl") == 0
-    source_update_payload = facade.payload("update_source")
+    source_update_payload = facade.payload("update_source", KnowledgeFSSourceUpdatePayload)
+    source_update_metadata = json_object(source_update_payload.metadata)
     assert source_update_payload.status == "active"
 
 
@@ -788,10 +799,11 @@ def test_initial_website_source_import_exposes_failed_source_without_activating_
 
     assert _start(facade, _payload(), workflow_id="workflow-1") == "workflow-1"
 
-    source_update_payload = facade.payload("update_source")
+    source_update_payload = facade.payload("update_source", KnowledgeFSSourceUpdatePayload)
+    source_update_metadata = json_object(source_update_payload.metadata)
     assert source_update_payload.status == "disabled"
-    assert source_update_payload.metadata["preview"] is False
-    initial_import = source_update_payload.metadata["initialImport"]
+    assert source_update_metadata["preview"] is False
+    initial_import = json_object(source_update_metadata["initialImport"])
     assert initial_import["errorCode"] == "SOURCE_DOCUMENT_MATERIALIZATION_FAILED"
     assert initial_import["errorMessage"] == "Source document materialization failed"
     assert initial_import["state"] == "failed"
@@ -803,7 +815,9 @@ def test_initial_website_source_import_exposes_failed_source_without_activating_
         "enabled": True,
         "mode": "interval",
     }
-    assert len(initial_import["configurationFingerprint"]) == 64
+    fingerprint = initial_import["configurationFingerprint"]
+    assert isinstance(fingerprint, str)
+    assert len(fingerprint) == 64
     assert facade.call_count("update_source_sync_policy") == 0
 
 

api/tests/unit_tests/tasks/test_knowledge_fs_source_import_tasks.py

@@ -4,7 +4,12 @@ from unittest.mock import patch
 import pytest
 from celery.exceptions import Retry
 
-from services.knowledge_fs.product_dto import KnowledgeFSSourceResponse, KnowledgeFSSourceWorkflowResponse
+from services.knowledge_fs.product_dto import (
+    KnowledgeFSSourceResponse,
+    KnowledgeFSSourceSyncPolicyPayload,
+    KnowledgeFSSourceUpdatePayload,
+    KnowledgeFSSourceWorkflowResponse,
+)
 from services.knowledge_fs.product_remote import (
     KnowledgeFSProductRequestRejectedError,
     KnowledgeFSProductResourceNotFoundError,
@@ -18,6 +23,7 @@ from tests.unit_tests.tasks.knowledge_fs_task_fakes import (
     FacadeRuntime,
     RecordingKnowledgeFSFacade,
     RecordingRetry,
+    json_object,
     source_response,
     source_workflow_response,
 )
@@ -102,16 +108,17 @@ def test_finalize_source_import_activates_source_then_applies_policy_for_new_ver
 
     assert _run(facade) == "import-1"
 
-    policy = facade.payload("update_source_sync_policy")
+    policy = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload)
     assert policy.enabled is False
     assert policy.mode == "manual"
     assert policy.expected_revision == 0
     assert policy.expected_source_version == 5
-    update = facade.payload("update_source")
+    update = facade.payload("update_source", KnowledgeFSSourceUpdatePayload)
+    update_metadata = json_object(update.metadata)
     assert update.status == "active"
-    assert "parameters" not in update.metadata
-    assert update.metadata["pendingImport"] is None
-    assert update.metadata["lastImport"]["state"] == "completed"
+    assert "parameters" not in update_metadata
+    assert update_metadata["pendingImport"] is None
+    assert json_object(update_metadata["lastImport"])["state"] == "completed"
 
 
 def test_finalize_source_import_retries_policy_after_source_activation() -> None:
@@ -137,7 +144,7 @@ def test_finalize_source_import_retries_policy_after_source_activation() -> None
     assert _run(facade) == "import-1"
 
     assert facade.call_count("update_source") == 0
-    policy = facade.payload("update_source_sync_policy")
+    policy = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload)
     assert policy.expected_source_version == 5
 
 
@@ -184,11 +191,13 @@ def test_finalize_source_import_recovers_a_stale_failed_marker_after_workflow_co
 
     assert _run(facade) == "import-1"
 
-    update = facade.payload("update_source")
+    update = facade.payload("update_source", KnowledgeFSSourceUpdatePayload)
+
+    update_metadata = json_object(update.metadata)
     assert update.status == "active"
-    assert update.metadata["lastImport"]["state"] == "completed"
-    assert update.metadata["pendingImport"] is None
-    policy = facade.payload("update_source_sync_policy")
+    assert json_object(update_metadata["lastImport"])["state"] == "completed"
+    assert update_metadata["pendingImport"] is None
+    policy = facade.payload("update_source_sync_policy", KnowledgeFSSourceSyncPolicyPayload)
     assert policy.expected_source_version == 6
 
 
@@ -206,10 +215,12 @@ def test_finalize_source_import_persists_failure_on_visible_source() -> None:
 
     assert _run(facade) == "import-1"
 
-    update = facade.payload("update_source")
+    update = facade.payload("update_source", KnowledgeFSSourceUpdatePayload)
+
+    update_metadata = json_object(update.metadata)
     assert update.status == "error"
-    assert "parameters" not in update.metadata
-    assert update.metadata["lastImport"] == {
+    assert "parameters" not in update_metadata
+    assert update_metadata["lastImport"] == {
         "errorCode": "PROVIDER_FAILED",
         "errorMessage": "provider failed",
         "kind": "crawl-preview-selection",
@@ -218,7 +229,7 @@ def test_finalize_source_import_persists_failure_on_visible_source() -> None:
         "syncPolicy": {"enabled": False, "mode": "manual"},
         "workflowId": "import-1",
     }
-    assert update.metadata["pendingImport"] is None
+    assert update_metadata["pendingImport"] is None
     assert facade.call_count("update_source_sync_policy") == 0