diff --git a/api/services/agent/home_snapshot_service.py b/api/services/agent/home_snapshot_service.py index 937c2506441..722c0f53a9a 100644 --- a/api/services/agent/home_snapshot_service.py +++ b/api/services/agent/home_snapshot_service.py @@ -2,8 +2,6 @@ from __future__ import annotations -import logging - from dify_agent.client import Client, DifyAgentNotFoundError from dify_agent.protocol import CreateHomeSnapshotFromBindingRequest from sqlalchemy import select @@ -26,8 +24,6 @@ from models.agent import ( from services.agent.errors import AgentBuildSandboxNotFoundError from services.agent.workspace_service import AgentWorkspaceService, WorkspaceOwnerScope -logger = logging.getLogger(__name__) - class AgentHomeSnapshotUnavailableError(RuntimeError): """The requested owner-scoped Home Snapshot cannot be used.""" @@ -122,16 +118,6 @@ class AgentHomeSnapshotService: @classmethod def collect_retired_home_snapshot(cls, *, tenant_id: str, home_snapshot_id: str) -> None: - try: - cls._collect_retired_home_snapshot(tenant_id=tenant_id, home_snapshot_id=home_snapshot_id) - except Exception: - logger.exception( - "Failed to collect retired Agent Home Snapshot", - extra={"tenant_id": tenant_id, "home_snapshot_id": home_snapshot_id}, - ) - - @classmethod - def _collect_retired_home_snapshot(cls, *, tenant_id: str, home_snapshot_id: str) -> None: with session_factory.create_session() as session: snapshot = session.scalar( select(AgentHomeSnapshot).where( @@ -150,14 +136,7 @@ class AgentHomeSnapshotService: if referenced is not None: return snapshot_ref = snapshot.snapshot_ref - try: - cls.delete(snapshot_ref=snapshot_ref) - except Exception: - logger.exception( - "Failed to collect retired Agent Home Snapshot", - extra={"tenant_id": tenant_id, "home_snapshot_id": home_snapshot_id}, - ) - return + cls.delete(snapshot_ref=snapshot_ref) with session_factory.create_session() as session: snapshot = session.scalar( select(AgentHomeSnapshot).where( diff --git a/api/services/agent/workspace_service.py b/api/services/agent/workspace_service.py index 574054238c0..5053d9294bf 100644 --- a/api/services/agent/workspace_service.py +++ b/api/services/agent/workspace_service.py @@ -298,16 +298,6 @@ class AgentWorkspaceService: @classmethod def collect_retired_binding(cls, *, tenant_id: str, binding_id: str) -> None: - try: - cls._collect_retired_binding(tenant_id=tenant_id, binding_id=binding_id) - except Exception: - logger.exception( - "Failed to collect retired Agent Workspace Binding", - extra={"tenant_id": tenant_id, "binding_id": binding_id}, - ) - - @classmethod - def _collect_retired_binding(cls, *, tenant_id: str, binding_id: str) -> None: with session_factory.create_session() as session: binding = session.scalar( select(AgentWorkspaceBinding).where( @@ -332,20 +322,13 @@ class AgentWorkspaceService: if workspace_id is not None: cls.collect_retired_workspace(tenant_id=tenant_id, workspace_id=workspace_id) return - try: - with cls._client() as client: - client.destroy_execution_binding_sync( - DestroyExecutionBindingRequest( - binding_ref=backend_binding_ref, - destroy_workspace=False, - ) + with cls._client() as client: + client.destroy_execution_binding_sync( + DestroyExecutionBindingRequest( + binding_ref=backend_binding_ref, + destroy_workspace=False, ) - except Exception: - logger.exception( - "Failed to collect retired Agent Workspace Binding", - extra={"tenant_id": tenant_id, "binding_id": binding_id}, ) - return with session_factory.create_session() as session: binding = session.scalar( select(AgentWorkspaceBinding).where( @@ -360,16 +343,6 @@ class AgentWorkspaceService: @classmethod def collect_retired_workspace(cls, *, tenant_id: str, workspace_id: str) -> None: - try: - cls._collect_retired_workspace(tenant_id=tenant_id, workspace_id=workspace_id) - except Exception: - logger.exception( - "Failed to collect retired Agent Workspace", - extra={"tenant_id": tenant_id, "workspace_id": workspace_id}, - ) - - @classmethod - def _collect_retired_workspace(cls, *, tenant_id: str, workspace_id: str) -> None: with session_factory.create_session() as session: workspace = session.scalar( select(AgentWorkspace).where( @@ -400,21 +373,14 @@ class AgentWorkspaceService: workspace_ref = workspace.backend_workspace_ref binding_ref = anchor.backend_binding_ref anchor_id = anchor.id - try: - with cls._client() as client: - client.destroy_execution_binding_sync( - DestroyExecutionBindingRequest( - binding_ref=binding_ref, - workspace_ref=workspace_ref, - destroy_workspace=True, - ) + with cls._client() as client: + client.destroy_execution_binding_sync( + DestroyExecutionBindingRequest( + binding_ref=binding_ref, + workspace_ref=workspace_ref, + destroy_workspace=True, ) - except Exception: - logger.exception( - "Failed to collect retired Agent Workspace", - extra={"tenant_id": tenant_id, "workspace_id": workspace_id, "binding_id": anchor_id}, ) - return with session_factory.create_session() as session: stored_workspace = session.scalar( select(AgentWorkspace).where( diff --git a/api/tasks/collect_agent_resources_task.py b/api/tasks/collect_agent_resources_task.py index 2b84cc01b79..bfec87b7bf3 100644 --- a/api/tasks/collect_agent_resources_task.py +++ b/api/tasks/collect_agent_resources_task.py @@ -45,6 +45,7 @@ def collect_agent_resources( "resource_id": resource_id, }, ) + raise def enqueue_agent_resource_collection( diff --git a/api/tests/unit_tests/services/agent/test_home_snapshot_service.py b/api/tests/unit_tests/services/agent/test_home_snapshot_service.py index 2f7e7aceccc..abe565aaa94 100644 --- a/api/tests/unit_tests/services/agent/test_home_snapshot_service.py +++ b/api/tests/unit_tests/services/agent/test_home_snapshot_service.py @@ -129,27 +129,23 @@ def test_build_apply_fails_fast_without_source_binding() -> None: ) -def test_home_snapshot_collection_database_failure_is_best_effort(monkeypatch: pytest.MonkeyPatch) -> None: +def test_home_snapshot_collection_database_failure_propagates(monkeypatch: pytest.MonkeyPatch) -> None: context = MagicMock() session = context.__enter__.return_value - session.scalar.side_effect = RuntimeError("database unavailable") - log_exception = MagicMock() + error = RuntimeError("database unavailable") + session.scalar.side_effect = error monkeypatch.setattr( "services.agent.home_snapshot_service.session_factory.create_session", lambda: context, ) - monkeypatch.setattr("services.agent.home_snapshot_service.logger.exception", log_exception) - AgentHomeSnapshotService.collect_retired_home_snapshot( - tenant_id="tenant-1", - home_snapshot_id="home-1", - ) + with pytest.raises(RuntimeError) as exc_info: + AgentHomeSnapshotService.collect_retired_home_snapshot( + tenant_id="tenant-1", + home_snapshot_id="home-1", + ) - session.scalar.assert_called_once() - log_exception.assert_called_once_with( - "Failed to collect retired Agent Home Snapshot", - extra={"tenant_id": "tenant-1", "home_snapshot_id": "home-1"}, - ) + assert exc_info.value is error @pytest.mark.parametrize( @@ -157,7 +153,7 @@ def test_home_snapshot_collection_database_failure_is_best_effort(monkeypatch: p [(AgentHomeSnapshot, AgentConfigDraft, AgentConfigSnapshot)], indirect=True, ) -def test_home_snapshot_collection_final_delete_failure_is_best_effort( +def test_home_snapshot_collection_backend_failure_propagates_and_preserves_retired_snapshot( monkeypatch: pytest.MonkeyPatch, sqlite_session: Session ) -> None: snapshot = AgentHomeSnapshot( @@ -169,7 +165,45 @@ def test_home_snapshot_collection_final_delete_failure_is_best_effort( ) sqlite_session.add(snapshot) sqlite_session.commit() - commit = MagicMock(side_effect=RuntimeError("database unavailable")) + error = RuntimeError("Agent backend unavailable") + delete = MagicMock(side_effect=error) + monkeypatch.setattr( + "services.agent.home_snapshot_service.session_factory.create_session", + lambda: nullcontext(sqlite_session), + ) + monkeypatch.setattr(AgentHomeSnapshotService, "delete", delete) + + with pytest.raises(RuntimeError) as exc_info: + AgentHomeSnapshotService.collect_retired_home_snapshot( + tenant_id="tenant-1", + home_snapshot_id=snapshot.id, + ) + + assert exc_info.value is error + stored_snapshot = sqlite_session.get(AgentHomeSnapshot, snapshot.id) + assert stored_snapshot is not None + assert stored_snapshot.status is AgentWorkingResourceStatus.RETIRED + + +@pytest.mark.parametrize( + "sqlite_session", + [(AgentHomeSnapshot, AgentConfigDraft, AgentConfigSnapshot)], + indirect=True, +) +def test_home_snapshot_collection_final_delete_failure_propagates( + monkeypatch: pytest.MonkeyPatch, sqlite_session: Session +) -> None: + snapshot = AgentHomeSnapshot( + id="home-1", + tenant_id="tenant-1", + agent_id="agent-1", + snapshot_ref="snapshot-ref-1", + status=AgentWorkingResourceStatus.RETIRED, + ) + sqlite_session.add(snapshot) + sqlite_session.commit() + error = RuntimeError("database unavailable") + commit = MagicMock(side_effect=error) delete = MagicMock() monkeypatch.setattr( "services.agent.home_snapshot_service.session_factory.create_session", @@ -178,9 +212,11 @@ def test_home_snapshot_collection_final_delete_failure_is_best_effort( monkeypatch.setattr(sqlite_session, "commit", commit) monkeypatch.setattr(AgentHomeSnapshotService, "delete", delete) - AgentHomeSnapshotService.collect_retired_home_snapshot( - tenant_id="tenant-1", - home_snapshot_id="home-1", - ) + with pytest.raises(RuntimeError) as exc_info: + AgentHomeSnapshotService.collect_retired_home_snapshot( + tenant_id="tenant-1", + home_snapshot_id="home-1", + ) + assert exc_info.value is error delete.assert_called_once_with(snapshot_ref="snapshot-ref-1") diff --git a/api/tests/unit_tests/services/agent/test_workspace_service.py b/api/tests/unit_tests/services/agent/test_workspace_service.py index ade845e33f7..d678a92106d 100644 --- a/api/tests/unit_tests/services/agent/test_workspace_service.py +++ b/api/tests/unit_tests/services/agent/test_workspace_service.py @@ -431,45 +431,91 @@ def test_collect_workspace_destroys_workspace_then_remaining_bindings( assert sqlite_session.get(AgentWorkspaceBinding, remaining.id) is None -def test_binding_collection_database_failure_is_best_effort(monkeypatch: pytest.MonkeyPatch) -> None: +def test_binding_collection_database_failure_propagates(monkeypatch: pytest.MonkeyPatch) -> None: context = MagicMock() session = context.__enter__.return_value - session.scalar.side_effect = RuntimeError("database unavailable") - log_exception = MagicMock() + error = RuntimeError("database unavailable") + session.scalar.side_effect = error monkeypatch.setattr("services.agent.workspace_service.session_factory.create_session", lambda: context) - monkeypatch.setattr("services.agent.workspace_service.logger.exception", log_exception) - AgentWorkspaceService.collect_retired_binding(tenant_id="tenant-1", binding_id="binding-1") + with pytest.raises(RuntimeError) as exc_info: + AgentWorkspaceService.collect_retired_binding(tenant_id="tenant-1", binding_id="binding-1") - session.scalar.assert_called_once() - log_exception.assert_called_once_with( - "Failed to collect retired Agent Workspace Binding", - extra={"tenant_id": "tenant-1", "binding_id": "binding-1"}, - ) + assert exc_info.value is error @pytest.mark.parametrize("sqlite_session", [(AgentWorkspace, AgentWorkspaceBinding)], indirect=True) -def test_workspace_collection_final_delete_failure_is_best_effort( +def test_binding_collection_backend_failure_propagates_and_preserves_retired_binding( + monkeypatch: pytest.MonkeyPatch, sqlite_session: Session +) -> None: + workspace = _workspace() + binding = _binding(status=AgentWorkingResourceStatus.RETIRED) + sqlite_session.add_all([workspace, binding]) + sqlite_session.commit() + error = RuntimeError("Agent backend unavailable") + client = MagicMock() + client.destroy_execution_binding_sync.side_effect = error + monkeypatch.setattr( + "services.agent.workspace_service.session_factory.create_session", lambda: nullcontext(sqlite_session) + ) + monkeypatch.setattr(AgentWorkspaceService, "_client", lambda: nullcontext(client)) + + with pytest.raises(RuntimeError) as exc_info: + AgentWorkspaceService.collect_retired_binding(tenant_id="tenant-1", binding_id=binding.id) + + assert exc_info.value is error + stored_binding = sqlite_session.get(AgentWorkspaceBinding, binding.id) + assert stored_binding is not None + assert stored_binding.status is AgentWorkingResourceStatus.RETIRED + + +@pytest.mark.parametrize("sqlite_session", [(AgentWorkspace, AgentWorkspaceBinding)], indirect=True) +def test_workspace_collection_backend_failure_propagates_and_preserves_retired_resources( monkeypatch: pytest.MonkeyPatch, sqlite_session: Session ) -> None: workspace = _workspace(status=AgentWorkingResourceStatus.RETIRED) anchor = _binding(status=AgentWorkingResourceStatus.RETIRED) sqlite_session.add_all([workspace, anchor]) sqlite_session.commit() - commit = MagicMock(side_effect=RuntimeError("database unavailable")) + error = RuntimeError("Agent backend unavailable") + client = MagicMock() + client.destroy_execution_binding_sync.side_effect = error + monkeypatch.setattr( + "services.agent.workspace_service.session_factory.create_session", lambda: nullcontext(sqlite_session) + ) + monkeypatch.setattr(AgentWorkspaceService, "_client", lambda: nullcontext(client)) + + with pytest.raises(RuntimeError) as exc_info: + AgentWorkspaceService.collect_retired_workspace(tenant_id="tenant-1", workspace_id=workspace.id) + + assert exc_info.value is error + stored_workspace = sqlite_session.get(AgentWorkspace, workspace.id) + stored_anchor = sqlite_session.get(AgentWorkspaceBinding, anchor.id) + assert stored_workspace is not None + assert stored_workspace.status is AgentWorkingResourceStatus.RETIRED + assert stored_anchor is not None + assert stored_anchor.status is AgentWorkingResourceStatus.RETIRED + + +@pytest.mark.parametrize("sqlite_session", [(AgentWorkspace, AgentWorkspaceBinding)], indirect=True) +def test_workspace_collection_final_delete_failure_propagates( + monkeypatch: pytest.MonkeyPatch, sqlite_session: Session +) -> None: + workspace = _workspace(status=AgentWorkingResourceStatus.RETIRED) + anchor = _binding(status=AgentWorkingResourceStatus.RETIRED) + sqlite_session.add_all([workspace, anchor]) + sqlite_session.commit() + error = RuntimeError("database unavailable") + commit = MagicMock(side_effect=error) client = MagicMock() - log_exception = MagicMock() monkeypatch.setattr( "services.agent.workspace_service.session_factory.create_session", lambda: nullcontext(sqlite_session) ) monkeypatch.setattr(sqlite_session, "commit", commit) monkeypatch.setattr(AgentWorkspaceService, "_client", lambda: nullcontext(client)) - monkeypatch.setattr("services.agent.workspace_service.logger.exception", log_exception) - AgentWorkspaceService.collect_retired_workspace(tenant_id="tenant-1", workspace_id="workspace-1") + with pytest.raises(RuntimeError) as exc_info: + AgentWorkspaceService.collect_retired_workspace(tenant_id="tenant-1", workspace_id="workspace-1") + assert exc_info.value is error client.destroy_execution_binding_sync.assert_called_once() - log_exception.assert_called_once_with( - "Failed to collect retired Agent Workspace", - extra={"tenant_id": "tenant-1", "workspace_id": "workspace-1"}, - ) diff --git a/api/tests/unit_tests/tasks/test_collect_agent_resources_task.py b/api/tests/unit_tests/tasks/test_collect_agent_resources_task.py index 48fe8ff2def..eb751061f17 100644 --- a/api/tests/unit_tests/tasks/test_collect_agent_resources_task.py +++ b/api/tests/unit_tests/tasks/test_collect_agent_resources_task.py @@ -39,14 +39,13 @@ def test_enqueue_deduplicates_ids_and_skips_empty_input(monkeypatch: pytest.Monk ) -def test_collection_continues_in_workspace_binding_snapshot_order(monkeypatch: pytest.MonkeyPatch) -> None: +def test_collection_runs_in_workspace_binding_snapshot_order(monkeypatch: pytest.MonkeyPatch) -> None: calls: list[str] = [] - - def collect_workspace(**_kwargs: object) -> None: - calls.append("workspace") - raise RuntimeError("workspace failed") - - monkeypatch.setattr(AgentWorkspaceService, "collect_retired_workspace", collect_workspace) + monkeypatch.setattr( + AgentWorkspaceService, + "collect_retired_workspace", + lambda **_kwargs: calls.append("workspace"), + ) monkeypatch.setattr( AgentWorkspaceService, "collect_retired_binding", @@ -68,6 +67,48 @@ def test_collection_continues_in_workspace_binding_snapshot_order(monkeypatch: p assert calls == ["workspace", "binding", "home"] +def test_collection_failure_propagates_and_stops_task(monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[str] = [] + error = RuntimeError("workspace failed") + log_exception = MagicMock() + + def collect_workspace(**_kwargs: object) -> None: + calls.append("workspace") + raise error + + monkeypatch.setattr(AgentWorkspaceService, "collect_retired_workspace", collect_workspace) + monkeypatch.setattr( + AgentWorkspaceService, + "collect_retired_binding", + lambda **_kwargs: calls.append("binding"), + ) + monkeypatch.setattr( + AgentHomeSnapshotService, + "collect_retired_home_snapshot", + lambda **_kwargs: calls.append("home"), + ) + monkeypatch.setattr("tasks.collect_agent_resources_task.logger.exception", log_exception) + + with pytest.raises(RuntimeError) as exc_info: + collect_agent_resources.run( + tenant_id="tenant-1", + workspace_ids=["workspace-1"], + binding_ids=["binding-1"], + home_snapshot_ids=["home-1"], + ) + + assert exc_info.value is error + assert calls == ["workspace"] + log_exception.assert_called_once_with( + "Failed to collect retired Agent resource", + extra={ + "tenant_id": "tenant-1", + "resource_type": "workspace", + "resource_id": "workspace-1", + }, + ) + + def test_enqueue_failure_is_best_effort(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr( collect_agent_resources, diff --git a/dify-agent/docs/dify-agent/concepts/runtime-resources/index.md b/dify-agent/docs/dify-agent/concepts/runtime-resources/index.md index 0745f1f9fce..7b3e02e9483 100644 --- a/dify-agent/docs/dify-agent/concepts/runtime-resources/index.md +++ b/dify-agent/docs/dify-agent/concepts/runtime-resources/index.md @@ -115,9 +115,11 @@ Retirement is a database transition from `ACTIVE` to `RETIRED`. It prevents new product use without performing network I/O inside the caller's transaction. Product lifecycle paths commit this transition synchronously. After the transaction commits, one Celery task asks Dify Agent to destroy the physical -resources. A successful collector deletes the corresponding ledger row; a -failed collector logs the failure and leaves the RETIRED row available for a -future retry or reconciler. +resources. A successful collector deletes the corresponding ledger row. If a +collector raises, the task logs the tenant, resource type, and resource ID, +re-raises the exception so collection stops and Celery records the task as +failed, and leaves the RETIRED row intact. No automatic retry or reconciliation +is performed. The unified `collect_agent_resources` task is registered on normal Celery workers and explicitly uses the existing `retention` queue. Standard workers