diff --git a/api/core/workflow/nodes/agent_v2/agent_node.py b/api/core/workflow/nodes/agent_v2/agent_node.py index c822e5d3ab1..86c6f993221 100644 --- a/api/core/workflow/nodes/agent_v2/agent_node.py +++ b/api/core/workflow/nodes/agent_v2/agent_node.py @@ -178,6 +178,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]): workflow_run_id=workflow_run_id, node_id=self._node_id, node_execution_id=self.execution_id, + conversation_id=conversation_id, ) bundle = self._binding_resolver.resolve( tenant_id=dify_ctx.tenant_id, @@ -223,6 +224,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]): workflow_agent_binding_id=bundle.binding.id, agent_id=bundle.agent.id, agent_config_snapshot_id=bundle.snapshot.id, + conversation_id=conversation_id, ) node_job = WorkflowNodeJobConfig.model_validate(bundle.binding.node_job_config_dict) diff --git a/api/core/workflow/nodes/agent_v2/session_store.py b/api/core/workflow/nodes/agent_v2/session_store.py index 134ccc0a62a..7ea3450deb5 100644 --- a/api/core/workflow/nodes/agent_v2/session_store.py +++ b/api/core/workflow/nodes/agent_v2/session_store.py @@ -29,6 +29,41 @@ _CALLER_VISIBILITY_ATTEMPTS = 60 _CALLER_VISIBILITY_INTERVAL_SECONDS = 0.05 +def resolve_workflow_agent_workspace_owner_scope( + *, + tenant_id: str, + app_id: str, + conversation_id: str | None, + workflow_run_id: str | None, + node_id: str, + workflow_agent_binding_id: str, + node_execution_id: str | None = None, +) -> WorkspaceOwnerScope: + """Choose the Workspace owner for a workflow Agent participant. + + Chatflow runs carry ``conversation_id`` and should persist Agent memory across + turns the same way standalone Agent Apps do. Pure workflow runs keep the + per-run ``WORKFLOW_RUN`` scope. + """ + + owner_scope_key = f"{node_id}:{workflow_agent_binding_id}" + if conversation_id: + return WorkspaceOwnerScope( + tenant_id=tenant_id, + app_id=app_id, + owner_type=AgentWorkspaceOwnerType.CONVERSATION, + owner_id=conversation_id, + owner_scope_key=owner_scope_key, + ) + return WorkspaceOwnerScope( + tenant_id=tenant_id, + app_id=app_id, + owner_type=AgentWorkspaceOwnerType.WORKFLOW_RUN, + owner_id=workflow_run_id or node_execution_id or "", + owner_scope_key=owner_scope_key, + ) + + @dataclass(frozen=True, slots=True) class WorkflowAgentSessionScope: tenant_id: str @@ -40,15 +75,18 @@ class WorkflowAgentSessionScope: workflow_agent_binding_id: str agent_id: str agent_config_snapshot_id: str + conversation_id: str | None = None @property def workspace_owner(self) -> WorkspaceOwnerScope: - return WorkspaceOwnerScope( + return resolve_workflow_agent_workspace_owner_scope( tenant_id=self.tenant_id, app_id=self.app_id, - owner_type=AgentWorkspaceOwnerType.WORKFLOW_RUN, - owner_id=self.workflow_run_id or self.node_execution_id, - owner_scope_key=f"{self.node_id}:{self.workflow_agent_binding_id}", + conversation_id=self.conversation_id, + workflow_run_id=self.workflow_run_id, + node_id=self.node_id, + workflow_agent_binding_id=self.workflow_agent_binding_id, + node_execution_id=self.node_execution_id, ) @@ -75,6 +113,7 @@ class WorkflowAgentWorkspaceStore: workflow_run_id: str | None, node_id: str, node_execution_id: str, + conversation_id: str | None = None, ) -> WorkflowAgentSessionScope | None: """Return the generation pinned by an existing node execution participant.""" @@ -97,12 +136,14 @@ class WorkflowAgentWorkspaceStore: workflow_agent_binding_id = process_data.get("workflow_agent_binding_id") if not isinstance(workflow_agent_binding_id, str): raise AgentWorkspaceNotFoundError("Workflow node execution caller identity is missing") - owner_scope = WorkspaceOwnerScope( + owner_scope = resolve_workflow_agent_workspace_owner_scope( tenant_id=tenant_id, app_id=app_id, - owner_type=AgentWorkspaceOwnerType.WORKFLOW_RUN, - owner_id=workflow_run_id or node_execution_id, - owner_scope_key=f"{node_id}:{workflow_agent_binding_id}", + conversation_id=conversation_id, + workflow_run_id=workflow_run_id, + node_id=node_id, + workflow_agent_binding_id=workflow_agent_binding_id, + node_execution_id=node_execution_id, ) binding = AgentWorkspaceService.get_active_binding( session=session, @@ -122,6 +163,7 @@ class WorkflowAgentWorkspaceStore: workflow_agent_binding_id=workflow_agent_binding_id, agent_id=binding.agent_id, agent_config_snapshot_id=binding.agent_config_version_id, + conversation_id=conversation_id, ) def load_or_create_node_execution_session( @@ -140,13 +182,10 @@ class WorkflowAgentWorkspaceStore: binding_id = execution.agent_workspace_binding_id if binding_id is None: - binding = AgentWorkspaceService.create_binding( + binding = self._resolve_or_create_binding( session=session, - scope=scope.workspace_owner, - agent_id=scope.agent_id, - base_home_snapshot_id=home_snapshot_id, - agent_config_version_id=scope.agent_config_snapshot_id, - agent_config_version_kind=AgentConfigVersionKind.SNAPSHOT, + scope=scope, + home_snapshot_id=home_snapshot_id, ) execution.agent_workspace_binding_id = binding.id execution.process_data = json.dumps( @@ -175,7 +214,7 @@ class WorkflowAgentWorkspaceStore: agent_config_version_id=scope.agent_config_snapshot_id, agent_config_version_kind=AgentConfigVersionKind.SNAPSHOT, ) - return self._stored(scope, binding) + return self._stored(session, scope, binding) def save_active_snapshot( self, @@ -267,7 +306,44 @@ class WorkflowAgentWorkspaceStore: raise AgentWorkspaceNotFoundError("Workflow node execution caller is unavailable") @staticmethod - def _stored(scope: WorkflowAgentSessionScope, binding: AgentWorkspaceBinding) -> StoredWorkflowAgentSession: + def _resolve_or_create_binding( + *, + session: Session, + scope: WorkflowAgentSessionScope, + home_snapshot_id: str | None, + ) -> AgentWorkspaceBinding: + """Reuse a conversation-scoped participant or allocate a new one.""" + + if scope.conversation_id is not None: + existing_binding = AgentWorkspaceService.resolve_active_binding_for_scope( + session=session, + scope=scope.workspace_owner, + agent_id=scope.agent_id, + ) + if existing_binding is not None: + AgentWorkspaceService.validate_binding_generation( + existing_binding, + base_home_snapshot_id=home_snapshot_id, + agent_config_version_id=scope.agent_config_snapshot_id, + agent_config_version_kind=AgentConfigVersionKind.SNAPSHOT, + ) + return existing_binding + + return AgentWorkspaceService.create_binding( + session=session, + scope=scope.workspace_owner, + agent_id=scope.agent_id, + base_home_snapshot_id=home_snapshot_id, + agent_config_version_id=scope.agent_config_snapshot_id, + agent_config_version_kind=AgentConfigVersionKind.SNAPSHOT, + ) + + @staticmethod + def _stored( + session: Session, + scope: WorkflowAgentSessionScope, + binding: AgentWorkspaceBinding, + ) -> StoredWorkflowAgentSession: snapshot = ( CompositorSessionSnapshot.model_validate_json(binding.session_snapshot) if binding.session_snapshot @@ -284,4 +360,9 @@ class WorkflowAgentWorkspaceStore: ) -__all__ = ["StoredWorkflowAgentSession", "WorkflowAgentSessionScope", "WorkflowAgentWorkspaceStore"] +__all__ = [ + "StoredWorkflowAgentSession", + "WorkflowAgentSessionScope", + "WorkflowAgentWorkspaceStore", + "resolve_workflow_agent_workspace_owner_scope", +] diff --git a/api/services/agent/workspace_service.py b/api/services/agent/workspace_service.py index a3ffeea0064..fee8348b66f 100644 --- a/api/services/agent/workspace_service.py +++ b/api/services/agent/workspace_service.py @@ -104,6 +104,39 @@ class AgentWorkspaceService: ) ) + @classmethod + def resolve_active_binding_for_scope( + cls, + *, + session: Session, + scope: WorkspaceOwnerScope, + agent_id: str, + ) -> AgentWorkspaceBinding | None: + """Return the ACTIVE participant for a stable Workspace owner scope.""" + + return session.scalar( + select(AgentWorkspaceBinding) + .join( + AgentWorkspace, + (AgentWorkspace.tenant_id == AgentWorkspaceBinding.tenant_id) + & (AgentWorkspace.id == AgentWorkspaceBinding.workspace_id), + ) + .where( + AgentWorkspaceBinding.tenant_id == scope.tenant_id, + AgentWorkspaceBinding.app_id == scope.app_id, + AgentWorkspaceBinding.agent_id == agent_id, + AgentWorkspaceBinding.status == AgentWorkingResourceStatus.ACTIVE, + AgentWorkspace.tenant_id == scope.tenant_id, + AgentWorkspace.app_id == scope.app_id, + AgentWorkspace.owner_type == scope.owner_type, + AgentWorkspace.owner_id == scope.owner_id, + AgentWorkspace.owner_scope_key == scope.owner_scope_key, + AgentWorkspace.status == AgentWorkingResourceStatus.ACTIVE, + ) + .order_by(AgentWorkspaceBinding.created_at.desc()) + .limit(1) + ) + @classmethod def create_binding( cls, @@ -297,6 +330,37 @@ class AgentWorkspaceService: retired.append(workspace_id) return retired + @classmethod + def retire_all_for_conversation( + cls, + *, + session: Session, + tenant_id: str, + app_id: str, + conversation_id: str, + ) -> list[str]: + """Retire all ACTIVE conversation-owned Workspaces for one Chatflow conversation.""" + + workspaces = session.scalars( + select(AgentWorkspace).where( + AgentWorkspace.tenant_id == tenant_id, + AgentWorkspace.app_id == app_id, + AgentWorkspace.owner_type == AgentWorkspaceOwnerType.CONVERSATION, + AgentWorkspace.owner_id == conversation_id, + AgentWorkspace.status == AgentWorkingResourceStatus.ACTIVE, + ) + ).all() + retired: list[str] = [] + for workspace in workspaces: + workspace_id = cls.retire_workspace( + session=session, + tenant_id=tenant_id, + workspace_id=workspace.id, + ) + if workspace_id is not None: + retired.append(workspace_id) + return retired + @classmethod def collect_retired_binding(cls, *, tenant_id: str, binding_id: str) -> None: with session_factory.create_session() as session: diff --git a/api/services/agent_app_sandbox_service.py b/api/services/agent_app_sandbox_service.py index f9b0149abf1..02deb6f0023 100644 --- a/api/services/agent_app_sandbox_service.py +++ b/api/services/agent_app_sandbox_service.py @@ -25,6 +25,7 @@ from clients.agent_backend.factory import create_agent_backend_client from configs import dify_config from core.db.session_factory import session_factory from core.tools.signature import bind_file_uri +from core.workflow.nodes.agent_v2.session_store import resolve_workflow_agent_workspace_owner_scope from models.agent import ( Agent, AgentConfigDraft, @@ -32,7 +33,7 @@ from models.agent import ( AgentWorkspaceBinding, AgentWorkspaceOwnerType, ) -from models.model import App, AppMode, Conversation +from models.model import App, AppMode, Conversation, Message from models.workflow import WorkflowNodeExecutionModel from services.agent.roster_service import AgentRosterService from services.agent.workspace_service import AgentWorkspaceService, WorkspaceOwnerScope @@ -406,16 +407,26 @@ class WorkflowAgentSandboxService: "this Workflow Agent node execution has no active Workspace Binding", status_code=404, ) + conversation_id = session.scalar( + select(Message.conversation_id) + .where( + Message.app_id == app_id, + Message.workflow_run_id == workflow_run_id, + ) + .limit(1) + ) binding = AgentWorkspaceService.get_active_binding( session=session, tenant_id=tenant_id, binding_id=execution.agent_workspace_binding_id, - expected_owner_scope=WorkspaceOwnerScope( + expected_owner_scope=resolve_workflow_agent_workspace_owner_scope( tenant_id=tenant_id, app_id=app_id, - owner_type=AgentWorkspaceOwnerType.WORKFLOW_RUN, - owner_id=workflow_run_id, - owner_scope_key=f"{node_id}:{workflow_agent_binding_id}", + conversation_id=conversation_id, + workflow_run_id=workflow_run_id, + node_id=node_id, + workflow_agent_binding_id=workflow_agent_binding_id, + node_execution_id=node_execution_id, ), ) if binding is None: diff --git a/api/services/conversation_service.py b/api/services/conversation_service.py index 68e8d3c3e2c..de73a7cf32f 100644 --- a/api/services/conversation_service.py +++ b/api/services/conversation_service.py @@ -201,7 +201,6 @@ class ConversationService: """ conversation = cls.get_conversation(app_model, conversation_id, user, session=session) binding_id = conversation.agent_workspace_binding_id - retired_binding_id: str | None = None if binding_id is not None: owner_scope = WorkspaceOwnerScope( tenant_id=app_model.tenant_id, @@ -224,23 +223,21 @@ class ConversationService: app_model.name, conversation_id, ) - if binding_id is not None: - retired_binding_id = AgentWorkspaceService.retire_binding( - session=session, - tenant_id=app_model.tenant_id, - binding_id=binding_id, - ) - if retired_binding_id is None: - raise AgentWorkspaceNotFoundError("Conversation participant Binding is unavailable") + retired_workspace_ids = AgentWorkspaceService.retire_all_for_conversation( + session=session, + tenant_id=app_model.tenant_id, + app_id=app_model.id, + conversation_id=conversation.id, + ) conversation.is_deleted = True session.commit() except Exception: session.rollback() raise - if retired_binding_id is not None: + if retired_workspace_ids: enqueue_agent_resource_collection( tenant_id=app_model.tenant_id, - binding_ids=(retired_binding_id,), + workspace_ids=retired_workspace_ids, ) try: delete_conversation_related_data.delay(conversation.id) diff --git a/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_session_store.py b/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_session_store.py index a1de053a81f..7f88c8eccad 100644 --- a/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_session_store.py +++ b/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_session_store.py @@ -6,10 +6,16 @@ from unittest.mock import MagicMock import pytest from agenton.compositor import CompositorSessionSnapshot +from agenton.compositor.schemas import LayerSessionSnapshot +from agenton.layers.base import LifecycleState from sqlalchemy import Engine, event, func, select from sqlalchemy.orm import Session, sessionmaker -from core.workflow.nodes.agent_v2.session_store import WorkflowAgentSessionScope, WorkflowAgentWorkspaceStore +from core.workflow.nodes.agent_v2.session_store import ( + WorkflowAgentSessionScope, + WorkflowAgentWorkspaceStore, + resolve_workflow_agent_workspace_owner_scope, +) from graphon.enums import WorkflowNodeExecutionStatus from models.agent import ( AgentConfigVersionKind, @@ -41,16 +47,18 @@ def _scope() -> WorkflowAgentSessionScope: def _execution_row( *, + execution_id: str = "execution-1", + workflow_run_id: str = "run-1", binding_id: str | None = None, process_data: dict[str, object] | None = None, ) -> WorkflowNodeExecutionModel: return WorkflowNodeExecutionModel( - id="execution-1", + id=execution_id, tenant_id="tenant-1", app_id="app-1", workflow_id="workflow-1", triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN, - workflow_run_id="run-1", + workflow_run_id=workflow_run_id, index=1, predecessor_node_id=None, node_execution_id="node-execution-1", @@ -158,6 +166,101 @@ def test_scope_uses_node_and_workflow_binding_as_workspace_subscope() -> None: assert owner.owner_scope_key == "node-1:workflow-binding-1" +def test_resolve_workspace_owner_scope_uses_conversation_for_chatflow() -> None: + owner = resolve_workflow_agent_workspace_owner_scope( + tenant_id="tenant-1", + app_id="app-1", + conversation_id="conversation-1", + workflow_run_id="run-1", + node_id="node-1", + workflow_agent_binding_id="workflow-binding-1", + ) + assert owner.owner_type is AgentWorkspaceOwnerType.CONVERSATION + assert owner.owner_id == "conversation-1" + assert owner.owner_scope_key == "node-1:workflow-binding-1" + + +def test_resolve_workspace_owner_scope_keeps_workflow_run_for_pure_workflow() -> None: + owner = resolve_workflow_agent_workspace_owner_scope( + tenant_id="tenant-1", + app_id="app-1", + conversation_id=None, + workflow_run_id="run-1", + node_id="node-1", + workflow_agent_binding_id="workflow-binding-1", + node_execution_id="execution-1", + ) + assert owner.owner_type is AgentWorkspaceOwnerType.WORKFLOW_RUN + assert owner.owner_id == "run-1" + + +def test_conversation_scope_reuses_existing_binding_on_later_turn( + monkeypatch: pytest.MonkeyPatch, + sqlite_session: Session, +) -> None: + prior_snapshot = CompositorSessionSnapshot( + layers=[ + LayerSessionSnapshot( + name="history", + lifecycle_state=LifecycleState.SUSPENDED, + runtime_state={"turn": 1}, + ) + ] + ) + workspace = AgentWorkspace( + id="workspace-conversation", + tenant_id="tenant-1", + app_id="app-1", + owner_type=AgentWorkspaceOwnerType.CONVERSATION, + owner_id="conversation-1", + owner_scope_key="node-1:workflow-binding-1", + backend_workspace_ref="workspace-1-ref", + status=AgentWorkingResourceStatus.ACTIVE, + active_guard=1, + ) + prior_binding = AgentWorkspaceBinding( + id="binding-prior", + tenant_id="tenant-1", + app_id="app-1", + workspace_id=workspace.id, + agent_id="agent-1", + base_home_snapshot_id="home-1", + agent_config_version_id="config-1", + agent_config_version_kind=AgentConfigVersionKind.SNAPSHOT, + backend_binding_ref="backend-binding-prior", + status=AgentWorkingResourceStatus.ACTIVE, + session_snapshot=prior_snapshot.model_dump_json(), + ) + execution = _execution_row( + execution_id="execution-2", + workflow_run_id="run-2", + process_data={"workflow_agent_binding_id": "workflow-binding-1"}, + ) + sqlite_session.add_all([_home_snapshot(), workspace, prior_binding, execution]) + sqlite_session.commit() + client = _install_backend_client(monkeypatch) + + scope = WorkflowAgentSessionScope( + tenant_id="tenant-1", + app_id="app-1", + workflow_id="workflow-1", + workflow_run_id="run-2", + node_id="node-1", + node_execution_id="execution-2", + workflow_agent_binding_id="workflow-binding-1", + agent_id="agent-1", + agent_config_snapshot_id="config-1", + conversation_id="conversation-1", + ) + stored = WorkflowAgentWorkspaceStore().load_or_create_node_execution_session(scope, home_snapshot_id="home-1") + + assert stored.workspace_id == workspace.id + assert stored.binding_id == prior_binding.id + assert stored.session_snapshot == prior_snapshot + client.create_execution_binding_sync.assert_not_called() + assert sqlite_session.scalar(select(func.count()).select_from(AgentWorkspaceBinding)) == 1 + + def test_load_existing_scope_reads_the_generation_from_the_persisted_binding(sqlite_session: Session) -> None: sqlite_session.add_all( [ @@ -417,3 +520,49 @@ def test_retire_workflow_run_returns_existing_retired_workspace(sqlite_session: sqlite_session.expire(workspace) assert workspace.status is AgentWorkingResourceStatus.RETIRED assert workspace_ids == [workspace.id] + + +def test_workflow_scope_creates_binding_without_existing_workspace_ref( + monkeypatch: pytest.MonkeyPatch, + sqlite_session: Session, +) -> None: + execution = _execution_row(process_data={"workflow_agent_binding_id": "workflow-binding-1"}) + sqlite_session.add_all([_home_snapshot(), execution]) + sqlite_session.commit() + client = _install_backend_client(monkeypatch) + + WorkflowAgentWorkspaceStore().load_or_create_node_execution_session(_scope(), home_snapshot_id="home-1") + + request = client.create_execution_binding_sync.call_args.args[0] + assert request.existing_workspace_ref is None + + +def test_conversation_scope_first_turn_creates_binding_without_existing_workspace_ref( + monkeypatch: pytest.MonkeyPatch, + sqlite_session: Session, +) -> None: + execution = _execution_row( + execution_id="execution-2", + workflow_run_id="run-2", + process_data={"workflow_agent_binding_id": "workflow-binding-1"}, + ) + sqlite_session.add_all([_home_snapshot(), execution]) + sqlite_session.commit() + client = _install_backend_client(monkeypatch) + + scope = WorkflowAgentSessionScope( + tenant_id="tenant-1", + app_id="app-1", + workflow_id="workflow-1", + workflow_run_id="run-2", + node_id="node-1", + node_execution_id="execution-2", + workflow_agent_binding_id="workflow-binding-1", + agent_id="agent-1", + agent_config_snapshot_id="config-1", + conversation_id="conversation-1", + ) + WorkflowAgentWorkspaceStore().load_or_create_node_execution_session(scope, home_snapshot_id="home-1") + + request = client.create_execution_binding_sync.call_args.args[0] + assert request.existing_workspace_ref is None 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 2a239c6e726..cfa7d56a708 100644 --- a/api/tests/unit_tests/services/agent/test_workspace_service.py +++ b/api/tests/unit_tests/services/agent/test_workspace_service.py @@ -49,6 +49,7 @@ def _workspace( app_id: str = "app-1", owner_type: AgentWorkspaceOwnerType = AgentWorkspaceOwnerType.CONVERSATION, owner_id: str = "conversation-1", + owner_scope_key: str = "root", status: AgentWorkingResourceStatus = AgentWorkingResourceStatus.ACTIVE, updated_at: datetime | None = None, backend_workspace_ref: str = "workspace-ref", @@ -59,7 +60,7 @@ def _workspace( app_id=app_id, owner_type=owner_type, owner_id=owner_id, - owner_scope_key="root", + owner_scope_key=owner_scope_key, backend_workspace_ref=backend_workspace_ref, status=status, active_guard=1 if status is AgentWorkingResourceStatus.ACTIVE else None, @@ -329,6 +330,62 @@ def test_retire_workspace_retires_all_active_bindings(sqlite_session: Session) - assert all(binding.retired_at == workspace.retired_at for binding in bindings) +def test_resolve_active_binding_for_scope_returns_matching_participant(sqlite_session: Session) -> None: + workspace = _workspace(owner_scope_key="node-1:workflow-binding-1") + binding = _binding(binding_id="binding-chatflow", workspace_id=workspace.id) + sqlite_session.add_all([workspace, binding]) + sqlite_session.commit() + + resolved = AgentWorkspaceService.resolve_active_binding_for_scope( + session=sqlite_session, + scope=WorkspaceOwnerScope( + tenant_id="tenant-1", + app_id="app-1", + owner_type=AgentWorkspaceOwnerType.CONVERSATION, + owner_id="conversation-1", + owner_scope_key="node-1:workflow-binding-1", + ), + agent_id="agent-1", + ) + + assert resolved is not None + assert resolved.id == binding.id + + +def test_retire_all_for_conversation_retires_only_matching_conversation_workspaces( + sqlite_session: Session, +) -> None: + matching = _workspace( + workspace_id="workspace-chatflow", + owner_scope_key="node-1:workflow-binding-1", + ) + other_conversation = _workspace( + workspace_id="workspace-other-conversation", + owner_id="conversation-2", + owner_scope_key="node-2:workflow-binding-2", + ) + workflow_run = _workspace( + workspace_id="workspace-workflow-run", + owner_type=AgentWorkspaceOwnerType.WORKFLOW_RUN, + owner_id="run-1", + owner_scope_key="node-1:workflow-binding-1", + ) + sqlite_session.add_all([matching, other_conversation, workflow_run]) + sqlite_session.commit() + + retired_ids = AgentWorkspaceService.retire_all_for_conversation( + session=sqlite_session, + tenant_id="tenant-1", + app_id="app-1", + conversation_id="conversation-1", + ) + + assert retired_ids == [matching.id] + assert matching.status is AgentWorkingResourceStatus.RETIRED + assert other_conversation.status is AgentWorkingResourceStatus.ACTIVE + assert workflow_run.status is AgentWorkingResourceStatus.ACTIVE + + def test_retire_all_for_app_retires_only_active_workspaces_for_that_app(sqlite_session: Session) -> None: active = _workspace(workspace_id="workspace-active", owner_id="conversation-active") already_retired = _workspace( diff --git a/api/tests/unit_tests/services/test_conversation_service.py b/api/tests/unit_tests/services/test_conversation_service.py index 0f2d94a877f..43868748761 100644 --- a/api/tests/unit_tests/services/test_conversation_service.py +++ b/api/tests/unit_tests/services/test_conversation_service.py @@ -178,10 +178,10 @@ def test_delete_retires_then_commits_before_enqueue(monkeypatch: pytest.MonkeyPa sqlite_session.flush() events: list[str] = [] get_binding = MagicMock(return_value=_workspace_binding("conversation-binding-1")) - retire_binding = MagicMock(side_effect=lambda **_kwargs: events.append("retire") or "conversation-binding-1") + retire_conversation = MagicMock(side_effect=lambda **_kwargs: events.append("retire") or ["workspace-1"]) monkeypatch.setattr(ConversationService, "get_conversation", MagicMock(return_value=conversation)) monkeypatch.setattr(AgentWorkspaceService, "get_active_binding", get_binding) - monkeypatch.setattr(AgentWorkspaceService, "retire_binding", retire_binding) + monkeypatch.setattr(AgentWorkspaceService, "retire_all_for_conversation", retire_conversation) event.listen(sqlite_session, "after_commit", lambda _session: events.append("commit")) monkeypatch.setattr( conversation_service, @@ -196,7 +196,7 @@ def test_delete_retires_then_commits_before_enqueue(monkeypatch: pytest.MonkeyPa assert events == ["retire", "commit", "enqueue"] assert conversation.is_deleted is True assert get_binding.call_args.kwargs["binding_id"] == "conversation-binding-1" - assert retire_binding.call_args.kwargs["binding_id"] == "conversation-binding-1" + assert retire_conversation.call_args.kwargs["conversation_id"] == conversation.id delete_related.assert_called_once_with(conversation.id) @@ -219,7 +219,7 @@ def test_delete_commit_failure_does_not_enqueue(monkeypatch: pytest.MonkeyPatch, "get_active_binding", MagicMock(return_value=_workspace_binding("binding-1")), ) - monkeypatch.setattr(AgentWorkspaceService, "retire_binding", MagicMock(return_value="binding-1")) + monkeypatch.setattr(AgentWorkspaceService, "retire_all_for_conversation", MagicMock(return_value=["workspace-1"])) enqueue_collection = MagicMock() delete_related = MagicMock() monkeypatch.setattr(conversation_service, "enqueue_agent_resource_collection", enqueue_collection) diff --git a/docker/envs/core-services/web.env.example b/docker/envs/core-services/web.env.example index bafb965778a..342e7f08e7f 100644 --- a/docker/envs/core-services/web.env.example +++ b/docker/envs/core-services/web.env.example @@ -32,6 +32,7 @@ ENABLE_WEBSITE_WATERCRAWL=true NEXT_PUBLIC_ENABLE_SINGLE_DOLLAR_LATEX=false NEXT_PUBLIC_ENABLE_FEATURE_PREVIEW=true NEXT_PUBLIC_ENABLE_AGENT_V2=true +NEXT_PUBLIC_ENABLE_AGENT_V2_IN_CHATFLOW=false NEXT_PUBLIC_COOKIE_DOMAIN= NEXT_PUBLIC_BATCH_CONCURRENCY=5 CSP_WHITELIST= diff --git a/web/.env.example b/web/.env.example index 4433c3282b4..d2fff4dfae4 100644 --- a/web/.env.example +++ b/web/.env.example @@ -107,6 +107,10 @@ NEXT_PUBLIC_ENABLE_FEATURE_PREVIEW=true # Enable Agent v2 frontend entry points. NEXT_PUBLIC_ENABLE_AGENT_V2=true +# Surface the Agent v2 node inside Chatflow (advanced-chat) apps. +# Requires NEXT_PUBLIC_ENABLE_AGENT_V2. Disabled by default. +NEXT_PUBLIC_ENABLE_AGENT_V2_IN_CHATFLOW=false + # The maximum number of tree node depth for workflow NEXT_PUBLIC_MAX_TREE_DEPTH=50 diff --git a/web/app/components/workflow-app/hooks/__tests__/use-available-nodes-meta-data.spec.ts b/web/app/components/workflow-app/hooks/__tests__/use-available-nodes-meta-data.spec.ts index 6ded926c890..8d105f403b7 100644 --- a/web/app/components/workflow-app/hooks/__tests__/use-available-nodes-meta-data.spec.ts +++ b/web/app/components/workflow-app/hooks/__tests__/use-available-nodes-meta-data.spec.ts @@ -4,6 +4,7 @@ import { useAvailableNodesMetaData } from '../use-available-nodes-meta-data' const mockUseIsChatMode = vi.fn() const mockIsAgentV2Enabled = vi.hoisted(() => vi.fn(() => true)) +const mockIsAgentV2InChatflowEnabled = vi.hoisted(() => vi.fn(() => false)) vi.mock('../use-is-chat-mode', () => ({ useIsChatMode: () => mockUseIsChatMode(), @@ -11,6 +12,7 @@ vi.mock('../use-is-chat-mode', () => ({ vi.mock('@/features/agent-v2/feature-flag', () => ({ isAgentV2Enabled: () => mockIsAgentV2Enabled(), + isAgentV2InChatflowEnabled: () => mockIsAgentV2InChatflowEnabled(), })) vi.mock('@/context/i18n', () => ({ @@ -21,6 +23,7 @@ describe('useAvailableNodesMetaData', () => { beforeEach(() => { vi.clearAllMocks() mockIsAgentV2Enabled.mockReturnValue(true) + mockIsAgentV2InChatflowEnabled.mockReturnValue(false) }) it('should include chat-specific nodes and make the start node undeletable in chat mode', () => { @@ -40,7 +43,7 @@ describe('useAvailableNodesMetaData', () => { ) }) - it('should expose only legacy Agent in chat mode while retaining Agent v2 metadata', () => { + it('should expose only legacy Agent in chat mode by default even when Agent v2 is enabled', () => { mockUseIsChatMode.mockReturnValue(true) const { result } = renderHook(() => useAvailableNodesMetaData()) @@ -52,6 +55,33 @@ describe('useAvailableNodesMetaData', () => { expect(result.current.nodesMap?.[BlockEnum.AgentV2]).toBeDefined() }) + it('should expose Agent v2 in chat mode when the Chatflow gate is enabled', () => { + mockUseIsChatMode.mockReturnValue(true) + mockIsAgentV2InChatflowEnabled.mockReturnValue(true) + + const { result } = renderHook(() => useAvailableNodesMetaData()) + const nodeTypes = result.current.nodes.map((node) => node.metaData.type) + + expect(nodeTypes).toContain(BlockEnum.AgentV2) + expect(nodeTypes).not.toContain(BlockEnum.Agent) + expect(result.current.nodesMap?.[BlockEnum.AgentV2]).toBeDefined() + expect(result.current.nodesMap?.[BlockEnum.Agent]).toBeDefined() + }) + + it('should expose only legacy Agent in chat mode when Agent v2 is disabled', () => { + mockUseIsChatMode.mockReturnValue(true) + mockIsAgentV2Enabled.mockReturnValue(false) + mockIsAgentV2InChatflowEnabled.mockReturnValue(true) + + const { result } = renderHook(() => useAvailableNodesMetaData()) + const nodeTypes = result.current.nodes.map((node) => node.metaData.type) + + expect(nodeTypes).toContain(BlockEnum.Agent) + expect(nodeTypes).not.toContain(BlockEnum.AgentV2) + expect(result.current.nodesMap?.[BlockEnum.Agent]).toBeDefined() + expect(result.current.nodesMap?.[BlockEnum.AgentV2]).toBeDefined() + }) + it('should include workflow-specific trigger and end nodes outside chat mode', () => { mockUseIsChatMode.mockReturnValue(false) diff --git a/web/app/components/workflow-app/hooks/use-available-nodes-meta-data.ts b/web/app/components/workflow-app/hooks/use-available-nodes-meta-data.ts index efa4a219858..b1027598994 100644 --- a/web/app/components/workflow-app/hooks/use-available-nodes-meta-data.ts +++ b/web/app/components/workflow-app/hooks/use-available-nodes-meta-data.ts @@ -13,7 +13,7 @@ import TriggerScheduleDefault from '@/app/components/workflow/nodes/trigger-sche import TriggerWebhookDefault from '@/app/components/workflow/nodes/trigger-webhook/default' import { BlockEnum } from '@/app/components/workflow/types' import { useDocLink } from '@/context/i18n' -import { isAgentV2Enabled } from '@/features/agent-v2/feature-flag' +import { isAgentV2Enabled, isAgentV2InChatflowEnabled } from '@/features/agent-v2/feature-flag' import { isProductlessDocPathWithAnchor } from '@/types/doc-paths' import { useIsChatMode } from './use-is-chat-mode' @@ -29,7 +29,9 @@ export const useAvailableNodesMetaData = () => { const isChatMode = useIsChatMode() const docLink = useDocLink() const agentV2Enabled = isAgentV2Enabled() - const shouldUseAgentV2 = agentV2Enabled && !isChatMode + // Chatflow (advanced-chat) keeps Agent v2 hidden by default; opt in via + // NEXT_PUBLIC_ENABLE_AGENT_V2_IN_CHATFLOW. Pure workflows are unaffected. + const shouldUseAgentV2 = agentV2Enabled && (!isChatMode || isAgentV2InChatflowEnabled()) const startNodeMetaData = useMemo( () => ({ diff --git a/web/docker/entrypoint.sh b/web/docker/entrypoint.sh index 4c0918f3043..dc1dd2e35c9 100755 --- a/web/docker/entrypoint.sh +++ b/web/docker/entrypoint.sh @@ -47,6 +47,7 @@ export NEXT_PUBLIC_ENABLE_WEBSITE_WATERCRAWL=${ENABLE_WEBSITE_WATERCRAWL:-true} export NEXT_PUBLIC_ENABLE_SINGLE_DOLLAR_LATEX=${NEXT_PUBLIC_ENABLE_SINGLE_DOLLAR_LATEX:-false} export NEXT_PUBLIC_ENABLE_FEATURE_PREVIEW=${NEXT_PUBLIC_ENABLE_FEATURE_PREVIEW:-true} export NEXT_PUBLIC_ENABLE_AGENT_V2=${NEXT_PUBLIC_ENABLE_AGENT_V2:-${ENABLE_AGENT_V2:-false}} +export NEXT_PUBLIC_ENABLE_AGENT_V2_IN_CHATFLOW=${NEXT_PUBLIC_ENABLE_AGENT_V2_IN_CHATFLOW:-${ENABLE_AGENT_V2_IN_CHATFLOW:-false}} export NEXT_PUBLIC_LOOP_NODE_MAX_COUNT=${LOOP_NODE_MAX_COUNT} export NEXT_PUBLIC_MAX_PARALLEL_LIMIT=${MAX_PARALLEL_LIMIT} export NEXT_PUBLIC_MAX_ITERATIONS_NUM=${MAX_ITERATIONS_NUM} diff --git a/web/env.ts b/web/env.ts index a36abbd9943..6d755eb70ae 100644 --- a/web/env.ts +++ b/web/env.ts @@ -72,6 +72,12 @@ const clientSchema = { NEXT_PUBLIC_DEPLOY_ENV: z.enum(['DEVELOPMENT', 'PRODUCTION', 'TESTING']).optional(), NEXT_PUBLIC_DISABLE_UPLOAD_IMAGE_AS_ICON: coercedBoolean.default(false), NEXT_PUBLIC_ENABLE_AGENT_V2: coercedBoolean.default(false), + /** + * Surface the Agent v2 node inside Chatflow (advanced-chat) apps. + * Requires NEXT_PUBLIC_ENABLE_AGENT_V2. Disabled by default so Chatflow keeps + * showing only the legacy Agent node until this is explicitly turned on. + */ + NEXT_PUBLIC_ENABLE_AGENT_V2_IN_CHATFLOW: coercedBoolean.default(false), /** * Enable preview features that are still in development. * Currently gates the `/create` and `/refine` slash commands in the @@ -234,6 +240,9 @@ export const env = createEnv({ NEXT_PUBLIC_ENABLE_AGENT_V2: isServer ? process.env.NEXT_PUBLIC_ENABLE_AGENT_V2 : getRuntimeEnvFromBody('enableAgentV2'), + NEXT_PUBLIC_ENABLE_AGENT_V2_IN_CHATFLOW: isServer + ? process.env.NEXT_PUBLIC_ENABLE_AGENT_V2_IN_CHATFLOW + : getRuntimeEnvFromBody('enableAgentV2InChatflow'), NEXT_PUBLIC_ENABLE_FEATURE_PREVIEW: isServer ? process.env.NEXT_PUBLIC_ENABLE_FEATURE_PREVIEW : getRuntimeEnvFromBody('enableFeaturePreview'), diff --git a/web/features/agent-v2/feature-flag.ts b/web/features/agent-v2/feature-flag.ts index 3b7e5589dc1..70361de7aca 100644 --- a/web/features/agent-v2/feature-flag.ts +++ b/web/features/agent-v2/feature-flag.ts @@ -1,3 +1,5 @@ import { env } from '@/env' export const isAgentV2Enabled = () => env.NEXT_PUBLIC_ENABLE_AGENT_V2 + +export const isAgentV2InChatflowEnabled = () => env.NEXT_PUBLIC_ENABLE_AGENT_V2_IN_CHATFLOW