fix(agent): complete Agent deletion lifecycle

api/controllers/console/agent/roster.py

@@ -676,6 +676,7 @@ def put(
         return _serialize_agent_app_detail(session, updated, current_user=current_user)
 
     @console_ns.response(204, "Agent app deleted successfully")
+    @console_ns.response(409, "Agent is still referenced by an active workflow")
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
     @login_required

api/controllers/console/app/workflow.py

@@ -1320,7 +1320,7 @@ def post(self, current_user: Account, app_model: App):
 
             workflow_created_at = TimestampField().format(workflow.created_at)
 
-        binding_ids, home_snapshot_ids = WorkflowAgentRetirementService.retire_unowned(
+        binding_ids, home_snapshot_ids, purge_agent_ids = WorkflowAgentRetirementService.retire_unowned(
             tenant_id=app_model.tenant_id,
             agent_ids=retirement_candidates,
             account_id=current_user.id,
@@ -1329,6 +1329,7 @@ def post(self, current_user: Account, app_model: App):
             tenant_id=app_model.tenant_id,
             binding_ids=binding_ids,
             home_snapshot_ids=home_snapshot_ids,
+            purge_agent_ids=purge_agent_ids,
         )
         return {
             "result": "success",

api/controllers/console/snippets/snippet_workflow.py

@@ -310,7 +310,7 @@ def post(self, current_user: Account, snippet: CustomizedSnippet):
             except ValueError as e:
                 return {"message": str(e)}, 400
 
-        binding_ids, home_snapshot_ids = WorkflowAgentRetirementService.retire_unowned(
+        binding_ids, home_snapshot_ids, purge_agent_ids = WorkflowAgentRetirementService.retire_unowned(
             tenant_id=tenant_id,
             agent_ids=retirement_candidates,
             account_id=current_user.id,
@@ -319,6 +319,7 @@ def post(self, current_user: Account, snippet: CustomizedSnippet):
             tenant_id=tenant_id,
             binding_ids=binding_ids,
             home_snapshot_ids=home_snapshot_ids,
+            purge_agent_ids=purge_agent_ids,
         )
         return {
             "result": "success",

api/openapi/markdown/console-openapi.md

@@ -362,6 +362,7 @@ Check if activation token is valid
 | ---- | ----------- |
 | 204 | Agent app deleted successfully |
 | 403 | Insufficient permissions |
+| 409 | Agent is still referenced by an active workflow |
 
 ### [GET] /agent/{agent_id}
 #### Parameters

api/services/agent/composer_service.py

@@ -286,7 +286,7 @@ def save_workflow_composer(
             agent_id=binding.agent_id,
         )
         session.commit()
-        binding_ids, home_snapshot_ids = WorkflowAgentRetirementService.retire_unowned(
+        binding_ids, home_snapshot_ids, purge_agent_ids = WorkflowAgentRetirementService.retire_unowned(
             tenant_id=tenant_id,
             agent_ids=retirement_candidates,
             account_id=account_id,
@@ -295,6 +295,7 @@ def save_workflow_composer(
             tenant_id=tenant_id,
             binding_ids=binding_ids,
             home_snapshot_ids=home_snapshot_ids,
+            purge_agent_ids=purge_agent_ids,
         )
         return state
 

api/services/agent/deletion_service.py

@@ -0,0 +1,119 @@
+"""Hard-delete archived Agent aggregates after external resources are collected."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+from sqlalchemy import delete, select
+
+from core.db.session_factory import session_factory
+from models.agent import (
+    Agent,
+    AgentConfigDraft,
+    AgentConfigRevision,
+    AgentConfigSnapshot,
+    AgentDebugConversation,
+    AgentHomeSnapshot,
+    AgentStatus,
+    AgentWorkingResourceStatus,
+    AgentWorkspaceBinding,
+    WorkflowAgentNodeBinding,
+)
+from services.agent.retirement_service import WorkflowAgentRetirementService
+
+
+class AgentDeletionInvariantError(RuntimeError):
+    """An archived Agent no longer satisfies the hard-deletion contract."""
+
+
+class AgentDeletionService:
+    """Delete Agent aggregates and stale Workflow-owned binding soft references.
+
+    The aggregate includes Agent-owned configuration, debug, Home, and Workspace
+    Binding rows. Purge also removes invalid or historical
+    ``WorkflowAgentNodeBinding`` soft-reference rows that are owned by Workflows
+    rather than by the Agent aggregate.
+    """
+
+    @classmethod
+    def purge_archived_agents(cls, *, tenant_id: str, agent_ids: Iterable[str]) -> None:
+        """Idempotently hard-delete eligible archived Agent aggregates.
+
+        Missing targets are a no-op. Every stored target must be ``ARCHIVED``,
+        have no ACTIVE Workspace Binding or Home Snapshot, and have no effective
+        Workflow ownership. All dependent rows and Agents are deleted and
+        committed in one transaction; an exception before commit leaves the
+        transaction to roll back without a partial aggregate deletion.
+        """
+        candidates = tuple(sorted({agent_id for agent_id in agent_ids if agent_id}))
+        if not candidates:
+            return
+
+        with session_factory.create_session() as session:
+            agents = session.scalars(
+                select(Agent).where(Agent.tenant_id == tenant_id, Agent.id.in_(candidates)).with_for_update()
+            ).all()
+            if not agents:
+                return
+
+            stored_ids = [agent.id for agent in agents]
+            non_archived_ids = [agent.id for agent in agents if agent.status != AgentStatus.ARCHIVED]
+            if non_archived_ids:
+                raise AgentDeletionInvariantError(
+                    f"Agents must be ARCHIVED before deletion: {', '.join(non_archived_ids)}"
+                )
+
+            active_binding_id = session.scalar(
+                select(AgentWorkspaceBinding.id)
+                .where(
+                    AgentWorkspaceBinding.tenant_id == tenant_id,
+                    AgentWorkspaceBinding.agent_id.in_(stored_ids),
+                    AgentWorkspaceBinding.status == AgentWorkingResourceStatus.ACTIVE,
+                )
+                .limit(1)
+            )
+            if active_binding_id is not None:
+                raise AgentDeletionInvariantError(f"Agent aggregate still has ACTIVE Binding {active_binding_id}")
+
+            active_home_id = session.scalar(
+                select(AgentHomeSnapshot.id)
+                .where(
+                    AgentHomeSnapshot.tenant_id == tenant_id,
+                    AgentHomeSnapshot.agent_id.in_(stored_ids),
+                    AgentHomeSnapshot.status == AgentWorkingResourceStatus.ACTIVE,
+                )
+                .limit(1)
+            )
+            if active_home_id is not None:
+                raise AgentDeletionInvariantError(f"Agent aggregate still has ACTIVE Home Snapshot {active_home_id}")
+
+            effective_ids = WorkflowAgentRetirementService.effective_agent_ids(
+                session=session,
+                tenant_id=tenant_id,
+                agent_ids=stored_ids,
+            )
+            if effective_ids:
+                raise AgentDeletionInvariantError(
+                    f"Agents regained effective Workflow ownership: {', '.join(sorted(effective_ids))}"
+                )
+
+            for model in (
+                WorkflowAgentNodeBinding,
+                AgentDebugConversation,
+                AgentConfigRevision,
+                AgentConfigDraft,
+                AgentConfigSnapshot,
+                AgentHomeSnapshot,
+                AgentWorkspaceBinding,
+            ):
+                session.execute(
+                    delete(model).where(
+                        model.tenant_id == tenant_id,
+                        model.agent_id.in_(stored_ids),
+                    )
+                )
+            session.execute(delete(Agent).where(Agent.tenant_id == tenant_id, Agent.id.in_(stored_ids)))
+            session.commit()
+
+
+__all__ = ["AgentDeletionInvariantError", "AgentDeletionService"]

api/services/agent/errors.py

@@ -19,6 +19,15 @@ class AgentArchivedError(Conflict):
     description = "Archived agent cannot be modified."
 
 
+class AgentWorkflowReferenceConflictError(BaseHTTPException):
+    error_code = "agent_workflow_reference_conflict"
+    description = "Agent is still referenced by an active workflow."
+    code = 409
+
+    def __init__(self, reference_count: int):
+        super().__init__(description=f"Agent is still referenced by {reference_count} active workflow app(s).")
+
+
 class AgentVersionConflictError(Conflict):
     description = "Agent config version changed. Please reload and try again."
 

api/services/agent/home_snapshot_service.py

@@ -15,7 +15,6 @@
 from models.agent import (
     Agent,
     AgentConfigDraft,
-    AgentConfigSnapshot,
     AgentConfigVersionKind,
     AgentHomeSnapshot,
     AgentStatus,
@@ -108,13 +107,13 @@ def retire_all_for_agent(cls, *, session: Session, tenant_id: str, agent_id: str
             select(AgentHomeSnapshot).where(
                 AgentHomeSnapshot.tenant_id == tenant_id,
                 AgentHomeSnapshot.agent_id == agent_id,
-                AgentHomeSnapshot.status == AgentWorkingResourceStatus.ACTIVE,
             )
         ).all()
         now = naive_utc_now()
         for row in rows:
-            row.status = AgentWorkingResourceStatus.RETIRED
-            row.retired_at = now
+            if row.status == AgentWorkingResourceStatus.ACTIVE:
+                row.status = AgentWorkingResourceStatus.RETIRED
+                row.retired_at = now
         return [row.id for row in rows]
 
     @classmethod
@@ -129,13 +128,6 @@ def collect_retired_home_snapshot(cls, *, tenant_id: str, home_snapshot_id: str)
             )
             if snapshot is None:
                 return
-            referenced = session.scalar(
-                select(AgentConfigDraft.id).where(AgentConfigDraft.home_snapshot_id == home_snapshot_id).limit(1)
-            ) or session.scalar(
-                select(AgentConfigSnapshot.id).where(AgentConfigSnapshot.home_snapshot_id == home_snapshot_id).limit(1)
-            )
-            if referenced is not None:
-                return
             snapshot_ref = snapshot.snapshot_ref
         cls.delete(snapshot_ref=snapshot_ref)
         with session_factory.create_session() as session:

api/services/agent/retirement_service.py

@@ -37,12 +37,17 @@ def retire_unowned(
         tenant_id: str,
         agent_ids: Iterable[str],
         account_id: str | None,
-    ) -> tuple[list[str], list[str]]:
-        """Re-check ownership, archive orphans, and commit their resource retirement."""
+    ) -> tuple[list[str], list[str], list[str]]:
+        """Re-check ownership, archive orphans, and commit resource retirement.
+
+        Returns ``(binding_ids, home_snapshot_ids, purge_agent_ids)`` for the
+        resources and complete Agent aggregates made eligible by the committed
+        retirement transaction.
+        """
 
         candidates = tuple(sorted({agent_id for agent_id in agent_ids if agent_id}))
         if not candidates:
-            return [], []
+            return [], [], []
         retired_bindings: list[str] = []
         retired_snapshots: list[str] = []
         try:
@@ -58,17 +63,16 @@ def retire_unowned(
                         select(AgentWorkspaceBinding).where(
                             AgentWorkspaceBinding.tenant_id == tenant_id,
                             AgentWorkspaceBinding.agent_id == agent_id,
-                            AgentWorkspaceBinding.status == AgentWorkingResourceStatus.ACTIVE,
                         )
                     ).all()
                     for binding in bindings:
-                        binding_id = AgentWorkspaceService.retire_binding(
-                            session=session,
-                            tenant_id=tenant_id,
-                            binding_id=binding.id,
-                        )
-                        if binding_id is not None:
-                            retired_bindings.append(binding_id)
+                        if binding.status == AgentWorkingResourceStatus.ACTIVE:
+                            AgentWorkspaceService.retire_binding(
+                                session=session,
+                                tenant_id=tenant_id,
+                                binding_id=binding.id,
+                            )
+                        retired_bindings.append(binding.id)
                     retired_snapshots.extend(
                         AgentHomeSnapshotService.retire_all_for_agent(
                             session=session,
@@ -85,8 +89,8 @@ def retire_unowned(
                     "agent_ids": candidates,
                 },
             )
-            return [], []
-        return retired_bindings, retired_snapshots
+            raise
+        return retired_bindings, retired_snapshots, retired_agent_ids
 
     @classmethod
     def archive_unowned(
@@ -97,7 +101,7 @@ def archive_unowned(
         agent_ids: Iterable[str],
         account_id: str | None,
     ) -> list[str]:
-        """Archive active orphans and return every orphan eligible for Home cleanup."""
+        """Archive active orphans and return complete aggregate purge candidates."""
         candidates = tuple(sorted({agent_id for agent_id in agent_ids if agent_id}))
         if not candidates:
             return []
@@ -109,7 +113,7 @@ def archive_unowned(
                 Agent.status.in_((AgentStatus.ACTIVE, AgentStatus.ARCHIVED)),
             )
         ).all()
-        effective_agent_ids = cls._effective_agent_ids(
+        effective_agent_ids = cls.effective_agent_ids(
             session=session,
             tenant_id=tenant_id,
             agent_ids=[agent.id for agent in agents],
@@ -130,7 +134,7 @@ def archive_unowned(
         return cleanup_candidates
 
     @staticmethod
-    def _effective_agent_ids(
+    def effective_agent_ids(
         *,
         session: Session,
         tenant_id: str,