test(workflow): cover async persistence patch paths

Add workflow execution task coverage and async repository validation coverage for the mixed persistence implementation so Codecov patch coverage includes the changed persistence paths.

api/tests/unit_tests/core/repositories/test_sqlalchemy_workflow_execution_repository.py

@@ -240,6 +240,27 @@ class TestSQLAlchemyWorkflowExecutionRepository:
         cached_model = repo._execution_cache[sample_workflow_execution.id_]
         assert cached_model.id == sample_workflow_execution.id_
 
+    def test_save_rejects_existing_run_from_other_tenant(
+        self, mock_session_factory, mock_account, sample_workflow_execution
+    ):
+        repo = SQLAlchemyWorkflowExecutionRepository(
+            session_factory=mock_session_factory,
+            user=mock_account,
+            app_id="test_app",
+            triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
+        )
+        existing_run = WorkflowRun()
+        existing_run.id = sample_workflow_execution.id_
+        existing_run.tenant_id = "other-tenant"
+        session = mock_session_factory.return_value.__enter__.return_value
+        session.get.return_value = existing_run
+
+        with pytest.raises(ValueError, match="Unauthorized access to workflow run"):
+            repo.save(sample_workflow_execution)
+
+        session.merge.assert_not_called()
+        session.commit.assert_not_called()
+
     @patch("core.repositories.sqlalchemy_workflow_execution_repository.save_workflow_execution_task")
     def test_save_queues_celery_task_when_async_persistence_enabled(
         self, mock_task, mock_session_factory, mock_account, sample_workflow_execution
@@ -264,6 +285,33 @@ class TestSQLAlchemyWorkflowExecutionRepository:
         session = mock_session_factory.return_value.__enter__.return_value
         session.merge.assert_not_called()
 
+    @patch("core.repositories.sqlalchemy_workflow_execution_repository.save_workflow_execution_task")
+    def test_queue_async_save_requires_context(
+        self, mock_task, mock_session_factory, mock_account, sample_workflow_execution
+    ):
+        repo = SQLAlchemyWorkflowExecutionRepository(
+            session_factory=mock_session_factory,
+            user=mock_account,
+            app_id="test_app",
+            triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
+        )
+
+        repo._triggered_from = None
+        with pytest.raises(ValueError, match="triggered_from is required"):
+            repo._queue_async_save(sample_workflow_execution)
+
+        repo._triggered_from = WorkflowRunTriggeredFrom.APP_RUN
+        repo._creator_user_id = None
+        with pytest.raises(ValueError, match="created_by is required"):
+            repo._queue_async_save(sample_workflow_execution)
+
+        repo._creator_user_id = "user-id"
+        repo._creator_user_role = None
+        with pytest.raises(ValueError, match="created_by_role is required"):
+            repo._queue_async_save(sample_workflow_execution)
+
+        mock_task.delay.assert_not_called()
+
     def test_save_uses_execution_started_at_when_record_does_not_exist(
         self, mock_session_factory, mock_account, sample_workflow_execution
     ):

api/tests/unit_tests/core/repositories/test_sqlalchemy_workflow_node_execution_repository.py

@@ -661,6 +661,62 @@ def test_save_execution_data_queues_celery_task_when_async_persistence_enabled(
     assert call_args["creator_user_id"] == "user"
 
 
+def test_queue_async_save_requires_context(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "core.repositories.sqlalchemy_workflow_node_execution_repository.FileService",
+        lambda *_: SimpleNamespace(upload_file=Mock()),
+    )
+    repo = SQLAlchemyWorkflowNodeExecutionRepository(
+        session_factory=Mock(spec=sessionmaker),
+        user=_mock_account(),
+        app_id="app",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+    )
+    execution = _execution()
+
+    repo._triggered_from = None
+    with pytest.raises(ValueError, match="triggered_from is required"):
+        repo._queue_async_save(execution)
+
+    repo._triggered_from = WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN
+    repo._creator_user_id = None
+    with pytest.raises(ValueError, match="created_by is required"):
+        repo._queue_async_save(execution)
+
+    repo._creator_user_id = "user"
+    repo._creator_user_role = None
+    with pytest.raises(ValueError, match="created_by_role is required"):
+        repo._queue_async_save(execution)
+
+
+def test_queue_async_save_execution_data_requires_context(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "core.repositories.sqlalchemy_workflow_node_execution_repository.FileService",
+        lambda *_: SimpleNamespace(upload_file=Mock()),
+    )
+    repo = SQLAlchemyWorkflowNodeExecutionRepository(
+        session_factory=Mock(spec=sessionmaker),
+        user=_mock_account(),
+        app_id="app",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+    )
+    execution = _execution()
+
+    repo._triggered_from = None
+    with pytest.raises(ValueError, match="triggered_from is required"):
+        repo._queue_async_save_execution_data(execution)
+
+    repo._triggered_from = WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN
+    repo._creator_user_id = None
+    with pytest.raises(ValueError, match="created_by is required"):
+        repo._queue_async_save_execution_data(execution)
+
+    repo._creator_user_id = "user"
+    repo._creator_user_role = None
+    with pytest.raises(ValueError, match="created_by_role is required"):
+        repo._queue_async_save_execution_data(execution)
+
+
 def test_save_retries_duplicate_and_logs_non_duplicate(
     monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
 ) -> None:

api/tests/unit_tests/tasks/test_workflow_execution_tasks.py

@@ -0,0 +1,78 @@
+from datetime import UTC, datetime, timedelta
+
+from graphon.entities import WorkflowExecution
+from graphon.enums import WorkflowExecutionStatus, WorkflowType
+from models import CreatorUserRole, WorkflowRun
+from models.enums import WorkflowRunTriggeredFrom
+from tasks.workflow_execution_tasks import (
+    _calculate_elapsed_time,
+    _create_workflow_run_from_execution,
+    _update_workflow_run_from_execution,
+)
+
+
+def _execution(
+    *,
+    elapsed_time: float = 3.5,
+    exceptions_count: int = 2,
+    finished_at: datetime | None = None,
+) -> WorkflowExecution:
+    started_at = datetime(2026, 1, 1, 12, 0, 0)
+    return WorkflowExecution(
+        id_="workflow-run-id",
+        workflow_id="workflow-id",
+        workflow_type=WorkflowType.WORKFLOW,
+        workflow_version="1.0",
+        graph={"nodes": [], "edges": []},
+        inputs={"input": "value"},
+        outputs={"output": "value"},
+        status=WorkflowExecutionStatus.SUCCEEDED,
+        error_message="",
+        elapsed_time=elapsed_time,
+        total_tokens=100,
+        total_steps=5,
+        exceptions_count=exceptions_count,
+        started_at=started_at,
+        finished_at=finished_at,
+    )
+
+
+def test_create_workflow_run_calculates_elapsed_time_and_exceptions_count() -> None:
+    execution = _execution(finished_at=datetime(2026, 1, 1, 12, 0, 12), exceptions_count=3)
+
+    workflow_run = _create_workflow_run_from_execution(
+        execution=execution,
+        tenant_id="tenant-id",
+        app_id="app-id",
+        triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
+        creator_user_id="user-id",
+        creator_user_role=CreatorUserRole.ACCOUNT,
+    )
+
+    assert workflow_run.elapsed_time == 12.0
+    assert workflow_run.exceptions_count == 3
+
+
+def test_update_workflow_run_calculates_elapsed_time_and_exceptions_count() -> None:
+    workflow_run = WorkflowRun()
+    execution = _execution(finished_at=datetime(2026, 1, 1, 12, 0, 8), exceptions_count=4)
+
+    _update_workflow_run_from_execution(workflow_run, execution)
+
+    assert workflow_run.elapsed_time == 8.0
+    assert workflow_run.exceptions_count == 4
+
+
+def test_calculate_elapsed_time_uses_runtime_elapsed_time_until_finished() -> None:
+    execution = _execution(finished_at=None)
+    execution.started_at = datetime.now(UTC).replace(tzinfo=None) - timedelta(seconds=4)
+
+    elapsed_time = _calculate_elapsed_time(execution)
+
+    assert 3.9 <= elapsed_time <= 5.0
+
+
+def test_calculate_elapsed_time_clamps_negative_duration_to_zero() -> None:
+    execution = _execution(finished_at=datetime(2026, 1, 1, 11, 59, 59))
+
+    assert _calculate_elapsed_time(execution) == 0.0

api/tests/unit_tests/tasks/test_workflow_node_execution_tasks.py

@@ -1,3 +1,4 @@
+from collections.abc import Mapping
 from datetime import UTC, datetime
 from unittest.mock import Mock, patch
 
@@ -17,7 +18,13 @@ from tasks.workflow_node_execution_tasks import (
 )
 
 
-def _execution() -> WorkflowNodeExecution:
+def _execution(
+    *,
+    metadata: Mapping[WorkflowNodeExecutionMetadataKey, object] | None = None,
+) -> WorkflowNodeExecution:
+    if metadata is None:
+        metadata = {WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: 10}
+
     return WorkflowNodeExecution(
         id="exec-id",
         node_execution_id="node-exec-id",
@@ -31,7 +38,7 @@ def _execution() -> WorkflowNodeExecution:
         process_data={"process": "value"},
         outputs={"output": "value"},
         status=WorkflowNodeExecutionStatus.SUCCEEDED,
-        metadata={WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: 10},
+        metadata=metadata,
         created_at=datetime.now(UTC).replace(tzinfo=None),
         finished_at=datetime.now(UTC).replace(tzinfo=None),
     )
@@ -53,6 +60,19 @@ def test_create_node_execution_persists_metadata_without_data_payloads() -> None
     assert db_model.execution_metadata == '{"total_tokens": 10}'
 
 
+def test_create_node_execution_defaults_empty_metadata() -> None:
+    db_model = _create_node_execution_from_domain(
+        execution=_execution(metadata={}),
+        tenant_id="tenant-id",
+        app_id="app-id",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+        creator_user_id="user-id",
+        creator_user_role=CreatorUserRole.ACCOUNT,
+    )
+
+    assert db_model.execution_metadata == "{}"
+
+
 def test_update_node_execution_metadata_preserves_data_payloads() -> None:
     db_model = WorkflowNodeExecutionModel()
     db_model.inputs = '{"old_input": true}'
@@ -67,6 +87,14 @@ def test_update_node_execution_metadata_preserves_data_payloads() -> None:
     assert db_model.status == WorkflowNodeExecutionStatus.SUCCEEDED
 
 
+def test_update_node_execution_metadata_defaults_empty_metadata() -> None:
+    db_model = WorkflowNodeExecutionModel()
+
+    _update_node_execution_metadata(db_model, _execution(metadata={}))
+
+    assert db_model.execution_metadata == "{}"
+
+
 @patch("tasks.workflow_node_execution_tasks._create_sqlalchemy_repository")
 def test_save_workflow_node_execution_data_task_uses_sqlalchemy_repository(mock_create_repository: Mock) -> None:
     repository = Mock()
@@ -96,6 +124,33 @@ def test_save_workflow_node_execution_data_task_uses_sqlalchemy_repository(mock_
     assert saved_data_execution.model_dump() == execution.model_dump()
 
 
+@patch("tasks.workflow_node_execution_tasks._create_sqlalchemy_repository")
+def test_save_workflow_node_execution_data_task_retries_on_failure(mock_create_repository: Mock) -> None:
+    mock_create_repository.side_effect = RuntimeError("db unavailable")
+    execution = _execution()
+
+    with (
+        patch.object(
+            save_workflow_node_execution_data_task,
+            "retry",
+            side_effect=RuntimeError("retry requested"),
+        ) as retry,
+        pytest.raises(RuntimeError, match="retry requested"),
+    ):
+        save_workflow_node_execution_data_task.run(
+            execution_data=execution.model_dump(),
+            tenant_id="tenant-id",
+            app_id="app-id",
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
+            creator_user_id="user-id",
+            creator_user_role=CreatorUserRole.ACCOUNT.value,
+        )
+
+    retry.assert_called_once()
+    assert isinstance(retry.call_args.kwargs["exc"], RuntimeError)
+    assert retry.call_args.kwargs["countdown"] == 60
+
+
 @patch("tasks.workflow_node_execution_tasks.session_factory.create_session")
 def test_save_workflow_node_execution_task_creates_metadata_record(mock_create_session: Mock) -> None:
     session = _TaskSession(existing_execution=None)