diff --git a/api/core/agent/publish_visibility.py b/api/core/agent/publish_visibility.py new file mode 100644 index 00000000000..5af035bb605 --- /dev/null +++ b/api/core/agent/publish_visibility.py @@ -0,0 +1,93 @@ +"""Publication visibility rules for calling roster Agents from Workflows. + +``Agent.active_config_is_published`` describes whether the editable shared +draft still matches the active snapshot. It is false both before the first +publish and after a published Agent receives new draft edits, so it must not be +used as a runtime availability flag. App-backed Agents are callable from a +Workflow only when the active snapshot has a revision created by a +publish-visible operation. Direct roster Agents are publish-visible by +construction and only need an active snapshot. +""" + +from sqlalchemy import and_, or_, select +from sqlalchemy.orm import Session +from sqlalchemy.sql.elements import ColumnElement + +from models.agent import Agent, AgentConfigRevision, AgentConfigRevisionOperation, AgentScope, AgentSource + +PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS = frozenset( + { + AgentConfigRevisionOperation.PUBLISH_DRAFT, + AgentConfigRevisionOperation.SAVE_CURRENT_VERSION, + AgentConfigRevisionOperation.SAVE_NEW_VERSION, + AgentConfigRevisionOperation.SAVE_NEW_AGENT, + AgentConfigRevisionOperation.SAVE_TO_ROSTER, + AgentConfigRevisionOperation.RESTORE_VERSION, + } +) + + +def workflow_callable_active_snapshot_filter() -> ColumnElement[bool]: + """Return the SQL predicate for an Agent with a Workflow-callable active snapshot. + + The caller remains responsible for tenant, roster scope, lifecycle status, + and model configuration filters. The correlated revision lookup makes the + predicate safe to compose into roster pagination queries. + """ + + app_backed_agent = or_( + Agent.source == AgentSource.AGENT_APP, + and_( + Agent.source == AgentSource.IMPORTED, + Agent.scope == AgentScope.ROSTER, + Agent.app_id.is_not(None), + ), + ) + publish_visible_revision_exists = ( + select(AgentConfigRevision.id) + .where( + AgentConfigRevision.tenant_id == Agent.tenant_id, + AgentConfigRevision.agent_id == Agent.id, + AgentConfigRevision.current_snapshot_id == Agent.active_config_snapshot_id, + AgentConfigRevision.operation.in_(PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS), + ) + .correlate(Agent) + .exists() + ) + return and_( + Agent.active_config_snapshot_id.is_not(None), + or_( + ~app_backed_agent, + publish_visible_revision_exists, + ), + ) + + +def agent_has_workflow_callable_active_snapshot(*, session: Session, agent: Agent) -> bool: + """Return whether ``agent`` has an active snapshot visible to Workflow. + + This object-level form is useful after ownership and lifecycle checks have + already loaded an Agent. It intentionally ignores dirty draft state so a + previously published snapshot keeps serving while later edits remain + unpublished. + """ + + if not agent.active_config_snapshot_id: + return False + is_app_backed = agent.source == AgentSource.AGENT_APP or ( + agent.source == AgentSource.IMPORTED and agent.scope == AgentScope.ROSTER and agent.app_id is not None + ) + if not is_app_backed: + return True + return bool( + session.scalar( + select(AgentConfigRevision.id) + .where( + AgentConfigRevision.tenant_id == agent.tenant_id, + AgentConfigRevision.agent_id == agent.id, + AgentConfigRevision.current_snapshot_id == agent.active_config_snapshot_id, + AgentConfigRevision.operation.in_(PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS), + ) + .limit(1) + ) + ) diff --git a/api/core/workflow/nodes/agent_v2/binding_resolver.py b/api/core/workflow/nodes/agent_v2/binding_resolver.py index 8dbabff2b8f..34812f670f7 100644 --- a/api/core/workflow/nodes/agent_v2/binding_resolver.py +++ b/api/core/workflow/nodes/agent_v2/binding_resolver.py @@ -4,8 +4,16 @@ from dataclasses import dataclass from sqlalchemy import select +from core.agent.publish_visibility import workflow_callable_active_snapshot_filter from core.db.session_factory import session_factory -from models.agent import Agent, AgentConfigSnapshot, AgentStatus, WorkflowAgentBindingType, WorkflowAgentNodeBinding +from models.agent import ( + Agent, + AgentConfigSnapshot, + AgentScope, + AgentStatus, + WorkflowAgentBindingType, + WorkflowAgentNodeBinding, +) class WorkflowAgentBindingError(Exception): @@ -24,7 +32,7 @@ class WorkflowAgentBindingBundle: class WorkflowAgentBindingResolver: - """Resolve the Agent binding owned by the current workflow id and node id.""" + """Resolve an owned binding without allowing unpublished roster snapshots to run.""" def resolve( self, @@ -53,18 +61,20 @@ class WorkflowAgentBindingResolver: if binding.agent_id is None: raise WorkflowAgentBindingError("agent_not_available", "Workflow Agent binding has no agent.") - agent = session.scalar( - select(Agent) - .where( - Agent.tenant_id == tenant_id, - Agent.id == binding.agent_id, - ) - .limit(1) + agent_stmt = select(Agent).where( + Agent.tenant_id == tenant_id, + Agent.id == binding.agent_id, ) + if binding.binding_type == WorkflowAgentBindingType.ROSTER_AGENT: + agent_stmt = agent_stmt.where( + Agent.scope == AgentScope.ROSTER, + workflow_callable_active_snapshot_filter(), + ) + agent = session.scalar(agent_stmt.limit(1)) if agent is None or agent.status == AgentStatus.ARCHIVED: raise WorkflowAgentBindingError( "agent_not_available", - f"Agent {binding.agent_id} is not available.", + f"Agent {binding.agent_id} is not available or has not been published.", ) snapshot_id = ( diff --git a/api/core/workflow/nodes/agent_v2/validators.py b/api/core/workflow/nodes/agent_v2/validators.py index 7e3969d8a16..6e56e28f829 100644 --- a/api/core/workflow/nodes/agent_v2/validators.py +++ b/api/core/workflow/nodes/agent_v2/validators.py @@ -6,9 +6,17 @@ from typing import Any from sqlalchemy import select from sqlalchemy.orm import Session +from core.agent.publish_visibility import workflow_callable_active_snapshot_filter from core.workflow.graph_topology import WorkflowGraphTopology from graphon.enums import BuiltinNodeTypes -from models.agent import Agent, AgentConfigSnapshot, AgentStatus, WorkflowAgentBindingType, WorkflowAgentNodeBinding +from models.agent import ( + Agent, + AgentConfigSnapshot, + AgentScope, + AgentStatus, + WorkflowAgentBindingType, + WorkflowAgentNodeBinding, +) from models.agent_config_entities import ( AgentFileRefConfig, AgentHumanContactConfig, @@ -117,21 +125,28 @@ class WorkflowAgentNodeValidator: binding: WorkflowAgentNodeBinding, topology: _WorkflowGraphTopology | None = None, ) -> None: + """Validate binding ownership, publication state, Agent Soul, and node-job references.""" + if binding.agent_id is None: raise WorkflowAgentNodeValidationError(f"Workflow Agent node {binding.node_id} is missing agent binding.") - agent = session.scalar( - select(Agent) - .where( - Agent.tenant_id == binding.tenant_id, - Agent.id == binding.agent_id, - ) - .limit(1) + agent_stmt = select(Agent).where( + Agent.tenant_id == binding.tenant_id, + Agent.id == binding.agent_id, ) - if agent is None or agent.status == AgentStatus.ARCHIVED: - raise WorkflowAgentNodeValidationError( - f"Workflow Agent node {binding.node_id} references an unavailable agent." + if binding.binding_type == WorkflowAgentBindingType.ROSTER_AGENT: + agent_stmt = agent_stmt.where( + Agent.scope == AgentScope.ROSTER, + workflow_callable_active_snapshot_filter(), ) + agent = session.scalar(agent_stmt.limit(1)) + if agent is None or agent.status == AgentStatus.ARCHIVED: + availability = ( + "an unavailable or unpublished roster agent" + if binding.binding_type == WorkflowAgentBindingType.ROSTER_AGENT + else "an unavailable agent" + ) + raise WorkflowAgentNodeValidationError(f"Workflow Agent node {binding.node_id} references {availability}.") snapshot_id = ( agent.active_config_snapshot_id diff --git a/api/services/agent/composer_service.py b/api/services/agent/composer_service.py index 48bb6d54a4f..96f0b8dde23 100644 --- a/api/services/agent/composer_service.py +++ b/api/services/agent/composer_service.py @@ -7,6 +7,7 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from sqlalchemy.sql.elements import ColumnElement +from core.agent.publish_visibility import agent_has_workflow_callable_active_snapshot from libs.helper import to_timestamp from models import Account from models.agent import ( @@ -271,6 +272,8 @@ class AgentComposerService: source_snapshot_id: str | None = None, idempotency_key: str | None = None, ) -> dict[str, Any]: + """Copy a callable roster Agent snapshot into a workflow-owned inline Agent.""" + workflow = cls._get_draft_workflow(session=session, tenant_id=tenant_id, app_id=app_id) binding = cls._require_binding( cls._get_workflow_binding(session=session, tenant_id=tenant_id, workflow_id=workflow.id, node_id=node_id) @@ -296,6 +299,8 @@ class AgentComposerService: source_agent = cls._require_agent(session=session, tenant_id=tenant_id, agent_id=source_agent_id) if source_agent.scope != AgentScope.ROSTER or source_agent.status != AgentStatus.ACTIVE: raise InvalidComposerConfigError("Source agent must be an active roster agent.") + if not agent_has_workflow_callable_active_snapshot(session=session, agent=source_agent): + raise InvalidComposerConfigError("Source agent must have a published config snapshot.") source_version = cls._require_version( session=session, tenant_id=tenant_id, @@ -529,39 +534,11 @@ class AgentComposerService: ) if not active_version: return False - if agent.source in APP_BACKED_AGENT_SOURCES and not cls._has_publish_visible_revision( - session=session, - tenant_id=tenant_id, - agent_id=agent.id, - snapshot_id=agent.active_config_snapshot_id, - ): + if not agent_has_workflow_callable_active_snapshot(session=session, agent=agent): return False return _agent_soul_config_json(agent_soul) == _agent_soul_config_json(active_version.config_snapshot_dict) - @classmethod - def _has_publish_visible_revision( - cls, *, session: Session, tenant_id: str, agent_id: str, snapshot_id: str - ) -> bool: - revisions = session.scalars( - select(AgentConfigRevision.operation).where( - AgentConfigRevision.tenant_id == tenant_id, - AgentConfigRevision.agent_id == agent_id, - AgentConfigRevision.current_snapshot_id == snapshot_id, - ) - ).all() - - return any( - operation - in { - AgentConfigRevisionOperation.PUBLISH_DRAFT, - AgentConfigRevisionOperation.SAVE_NEW_VERSION, - AgentConfigRevisionOperation.SAVE_TO_ROSTER, - AgentConfigRevisionOperation.RESTORE_VERSION, - } - for operation in revisions - ) - @classmethod def publish_agent_app_draft( cls, *, session: Session, tenant_id: str, agent_id: str, account_id: str, version_note: str | None = None diff --git a/api/services/agent/roster_service.py b/api/services/agent/roster_service.py index 8e8758890c9..c366f2af47b 100644 --- a/api/services/agent/roster_service.py +++ b/api/services/agent/roster_service.py @@ -6,6 +6,7 @@ from sqlalchemy.exc import IntegrityError from clients.agent_backend.session_cleanup import AgentBackendSessionCleanupPayload from constants.model_template import default_app_templates +from core.agent.publish_visibility import workflow_callable_active_snapshot_filter from core.app.apps.agent_app.session_store import AgentAppRuntimeSessionStore from core.app.entities.app_invoke_entities import InvokeFrom from libs.datetime_utils import naive_utc_now @@ -225,8 +226,11 @@ class AgentRosterService: def list_invite_options( self, *, tenant_id: str, page: int = 1, limit: int = 20, keyword: str | None = None, app_id: str | None = None ) -> dict[str, Any]: + """List active roster Agents whose published snapshot can be called by Workflow.""" + stmt = self._build_roster_agents_stmt(tenant_id=tenant_id, keyword=keyword).where( - Agent.active_config_has_model.is_(True) + Agent.active_config_has_model.is_(True), + workflow_callable_active_snapshot_filter(), ) total = self._session.scalar(select(func.count()).select_from(stmt.subquery())) or 0 agents = list(self._session.scalars(stmt.offset((page - 1) * limit).limit(limit)).all()) diff --git a/api/services/agent/workflow_publish_service.py b/api/services/agent/workflow_publish_service.py index 0997cfd2029..e74a93a899b 100644 --- a/api/services/agent/workflow_publish_service.py +++ b/api/services/agent/workflow_publish_service.py @@ -8,6 +8,7 @@ from pydantic import ValidationError from sqlalchemy import select from sqlalchemy.orm import Session +from core.agent.publish_visibility import workflow_callable_active_snapshot_filter from core.workflow.nodes.agent_v2.validators import WorkflowAgentNodeValidationError, WorkflowAgentNodeValidator from models.agent import ( Agent, @@ -447,6 +448,8 @@ class WorkflowAgentPublishService: node_id: str, agent_id: str, ) -> tuple[Agent, str]: + """Resolve an active roster Agent whose published snapshot is callable.""" + agent = session.scalar( select(Agent) .where( @@ -454,11 +457,12 @@ class WorkflowAgentPublishService: Agent.id == agent_id, Agent.scope == AgentScope.ROSTER, Agent.status == AgentStatus.ACTIVE, + workflow_callable_active_snapshot_filter(), ) .limit(1) ) if agent is None: - raise ValueError(f"Workflow Agent node {node_id} references an unavailable roster agent.") + raise ValueError(f"Workflow Agent node {node_id} references an unavailable or unpublished roster agent.") if agent.scope != AgentScope.ROSTER: raise ValueError(f"Workflow Agent node {node_id} roster_agent binding must reference a roster agent.") if not agent.active_config_snapshot_id: diff --git a/api/tests/unit_tests/core/agent/test_publish_visibility.py b/api/tests/unit_tests/core/agent/test_publish_visibility.py new file mode 100644 index 00000000000..7e6b0964c83 --- /dev/null +++ b/api/tests/unit_tests/core/agent/test_publish_visibility.py @@ -0,0 +1,341 @@ +from datetime import datetime + +import pytest +from sqlalchemy import select +from sqlalchemy.orm import Session + +from core.agent.publish_visibility import ( + PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS, + agent_has_workflow_callable_active_snapshot, + workflow_callable_active_snapshot_filter, +) +from models.agent import ( + Agent, + AgentConfigRevision, + AgentConfigRevisionOperation, + AgentConfigSnapshot, + AgentKind, + AgentScope, + AgentSource, + AgentStatus, + WorkflowAgentNodeBinding, +) +from models.agent_config_entities import AgentSoulConfig, AgentSoulModelConfig +from services.agent.roster_service import AgentRosterService + + +def _agent_soul() -> AgentSoulConfig: + return AgentSoulConfig( + model=AgentSoulModelConfig( + plugin_id="langgenius/openai", + model_provider="openai", + model="gpt-test", + ) + ) + + +def _add_agent( + session: Session, + *, + agent_id: str, + snapshot_id: str | None, + name: str, + source: AgentSource, + operation: AgentConfigRevisionOperation | None, + app_id: str | None = None, + scope: AgentScope = AgentScope.ROSTER, + status: AgentStatus = AgentStatus.ACTIVE, + has_model: bool | None = None, +) -> Agent: + agent = Agent( + id=agent_id, + tenant_id="tenant-1", + name=name, + description="", + agent_kind=AgentKind.DIFY_AGENT, + scope=scope, + source=source, + app_id=app_id, + status=status, + active_config_snapshot_id=snapshot_id, + active_config_has_model=snapshot_id is not None if has_model is None else has_model, + # A dirty draft must not hide an already published active snapshot. + active_config_is_published=False, + ) + session.add(agent) + if snapshot_id is None: + return agent + session.add( + AgentConfigSnapshot( + id=snapshot_id, + tenant_id="tenant-1", + agent_id=agent_id, + version=1, + config_snapshot=_agent_soul(), + ) + ) + if operation is not None: + session.add( + AgentConfigRevision( + id=f"revision-{agent_id}", + tenant_id="tenant-1", + agent_id=agent_id, + current_snapshot_id=snapshot_id, + revision=1, + operation=operation, + ) + ) + return agent + + +def test_publish_visible_operation_contract() -> None: + assert { + AgentConfigRevisionOperation.PUBLISH_DRAFT, + AgentConfigRevisionOperation.SAVE_CURRENT_VERSION, + AgentConfigRevisionOperation.SAVE_NEW_VERSION, + AgentConfigRevisionOperation.SAVE_NEW_AGENT, + AgentConfigRevisionOperation.SAVE_TO_ROSTER, + AgentConfigRevisionOperation.RESTORE_VERSION, + } == PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS + + +@pytest.mark.parametrize( + "sqlite_session", + [(Agent, AgentConfigSnapshot, AgentConfigRevision, WorkflowAgentNodeBinding)], + indirect=True, +) +def test_workflow_callable_filter_distinguishes_never_published_from_dirty_drafts( + sqlite_session: Session, +) -> None: + imported_draft = _add_agent( + sqlite_session, + agent_id="agent-imported-draft", + snapshot_id="snapshot-imported-draft", + name="Imported draft", + source=AgentSource.IMPORTED, + operation=AgentConfigRevisionOperation.IMPORT_PACKAGE, + app_id="app-imported-draft", + ) + app_draft = _add_agent( + sqlite_session, + agent_id="agent-app-draft", + snapshot_id="snapshot-app-draft", + name="App draft", + source=AgentSource.AGENT_APP, + operation=AgentConfigRevisionOperation.CREATE_VERSION, + ) + published_with_dirty_draft = _add_agent( + sqlite_session, + agent_id="agent-published-dirty", + snapshot_id="snapshot-published-dirty", + name="Published with dirty draft", + source=AgentSource.AGENT_APP, + operation=AgentConfigRevisionOperation.PUBLISH_DRAFT, + ) + saved_to_current_version = _add_agent( + sqlite_session, + agent_id="agent-saved-current", + snapshot_id="snapshot-saved-current", + name="Saved to current version", + source=AgentSource.AGENT_APP, + operation=AgentConfigRevisionOperation.SAVE_CURRENT_VERSION, + ) + saved_as_new_agent = _add_agent( + sqlite_session, + agent_id="agent-saved-new", + snapshot_id="snapshot-saved-new", + name="Saved as new agent", + source=AgentSource.AGENT_APP, + operation=AgentConfigRevisionOperation.SAVE_NEW_AGENT, + ) + saved_as_new_version = _add_agent( + sqlite_session, + agent_id="agent-saved-new-version", + snapshot_id="snapshot-saved-new-version", + name="Saved as new version", + source=AgentSource.AGENT_APP, + operation=AgentConfigRevisionOperation.SAVE_NEW_VERSION, + ) + saved_to_roster = _add_agent( + sqlite_session, + agent_id="agent-saved-to-roster", + snapshot_id="snapshot-saved-to-roster", + name="Saved to roster", + source=AgentSource.AGENT_APP, + operation=AgentConfigRevisionOperation.SAVE_TO_ROSTER, + ) + restored_version = _add_agent( + sqlite_session, + agent_id="agent-restored-version", + snapshot_id="snapshot-restored-version", + name="Restored version", + source=AgentSource.AGENT_APP, + operation=AgentConfigRevisionOperation.RESTORE_VERSION, + ) + direct_roster_agent = _add_agent( + sqlite_session, + agent_id="agent-direct-roster", + snapshot_id="snapshot-direct-roster", + name="Direct roster", + source=AgentSource.ROSTER, + operation=AgentConfigRevisionOperation.CREATE_VERSION, + ) + direct_imported_roster_agent = _add_agent( + sqlite_session, + agent_id="agent-direct-imported", + snapshot_id="snapshot-direct-imported", + name="Direct imported roster", + source=AgentSource.IMPORTED, + operation=AgentConfigRevisionOperation.IMPORT_PACKAGE, + ) + no_snapshot = _add_agent( + sqlite_session, + agent_id="agent-no-snapshot", + snapshot_id=None, + name="No snapshot", + source=AgentSource.ROSTER, + operation=None, + ) + published_without_model = _add_agent( + sqlite_session, + agent_id="agent-published-without-model", + snapshot_id="snapshot-published-without-model", + name="Published without model", + source=AgentSource.AGENT_APP, + operation=AgentConfigRevisionOperation.PUBLISH_DRAFT, + has_model=False, + ) + archived_agent = _add_agent( + sqlite_session, + agent_id="agent-archived", + snapshot_id="snapshot-archived", + name="Archived", + source=AgentSource.ROSTER, + operation=AgentConfigRevisionOperation.CREATE_VERSION, + status=AgentStatus.ARCHIVED, + ) + workflow_only_agent = _add_agent( + sqlite_session, + agent_id="agent-workflow-only", + snapshot_id="snapshot-workflow-only", + name="Workflow only", + source=AgentSource.WORKFLOW, + operation=AgentConfigRevisionOperation.CREATE_VERSION, + scope=AgentScope.WORKFLOW_ONLY, + ) + stale_published_snapshot = _add_agent( + sqlite_session, + agent_id="agent-stale-published-snapshot", + snapshot_id="snapshot-current-unpublished", + name="Stale published snapshot", + source=AgentSource.AGENT_APP, + operation=AgentConfigRevisionOperation.CREATE_VERSION, + ) + cross_owner_revision = _add_agent( + sqlite_session, + agent_id="agent-cross-owner-revision", + snapshot_id="snapshot-cross-owner-revision", + name="Cross owner revision", + source=AgentSource.AGENT_APP, + operation=None, + ) + sqlite_session.add( + AgentConfigSnapshot( + id="snapshot-old-published", + tenant_id="tenant-1", + agent_id=stale_published_snapshot.id, + version=2, + config_snapshot=_agent_soul(), + ) + ) + sqlite_session.add( + AgentConfigRevision( + id="revision-old-published", + tenant_id="tenant-1", + agent_id=stale_published_snapshot.id, + current_snapshot_id="snapshot-old-published", + revision=2, + operation=AgentConfigRevisionOperation.PUBLISH_DRAFT, + ) + ) + sqlite_session.add( + AgentConfigRevision( + id="revision-published-dirty-2", + tenant_id="tenant-1", + agent_id=published_with_dirty_draft.id, + current_snapshot_id=published_with_dirty_draft.active_config_snapshot_id, + revision=2, + operation=AgentConfigRevisionOperation.RESTORE_VERSION, + ) + ) + sqlite_session.add( + AgentConfigRevision( + id="revision-cross-owner", + tenant_id="tenant-other", + agent_id="agent-other", + current_snapshot_id=cross_owner_revision.active_config_snapshot_id, + revision=1, + operation=AgentConfigRevisionOperation.PUBLISH_DRAFT, + ) + ) + imported_draft.updated_at = datetime(2031, 7, 24, 12, 0, 0) + published_with_dirty_draft.updated_at = datetime(2030, 7, 24, 11, 0, 0) + sqlite_session.commit() + + callable_agent_ids = set( + sqlite_session.scalars(select(Agent.id).where(workflow_callable_active_snapshot_filter())).all() + ) + + assert callable_agent_ids == { + published_with_dirty_draft.id, + saved_to_current_version.id, + saved_as_new_agent.id, + saved_as_new_version.id, + saved_to_roster.id, + restored_version.id, + direct_roster_agent.id, + direct_imported_roster_agent.id, + published_without_model.id, + archived_agent.id, + workflow_only_agent.id, + } + assert agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=published_with_dirty_draft) + assert agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=direct_roster_agent) + assert agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=direct_imported_roster_agent) + assert not agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=imported_draft) + assert not agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=app_draft) + assert not agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=no_snapshot) + assert not agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=stale_published_snapshot) + assert not agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=cross_owner_revision) + + invite_options = AgentRosterService(sqlite_session).list_invite_options( + tenant_id="tenant-1", + page=1, + limit=20, + ) + + expected_invite_ids = callable_agent_ids - { + published_without_model.id, + archived_agent.id, + workflow_only_agent.id, + } + assert invite_options["total"] == len(expected_invite_ids) + assert {item["id"] for item in invite_options["data"]} == expected_invite_ids + + first_page = AgentRosterService(sqlite_session).list_invite_options( + tenant_id="tenant-1", + page=1, + limit=1, + ) + assert first_page["total"] == len(expected_invite_ids) + assert first_page["has_more"] is True + assert [item["id"] for item in first_page["data"]] == [published_with_dirty_draft.id] + + unpublished_keyword = AgentRosterService(sqlite_session).list_invite_options( + tenant_id="tenant-1", + page=1, + limit=20, + keyword="Imported draft", + ) + assert unpublished_keyword["total"] == 0 + assert unpublished_keyword["data"] == [] diff --git a/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_binding_resolver.py b/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_binding_resolver.py index a628c76b38d..86f3046d47f 100644 --- a/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_binding_resolver.py +++ b/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_binding_resolver.py @@ -1,10 +1,21 @@ import pytest +from sqlalchemy.orm import Session from core.workflow.nodes.agent_v2.binding_resolver import ( WorkflowAgentBindingError, WorkflowAgentBindingResolver, ) -from models.agent import Agent, AgentConfigSnapshot, AgentStatus, WorkflowAgentBindingType, WorkflowAgentNodeBinding +from models.agent import ( + Agent, + AgentConfigRevision, + AgentConfigRevisionOperation, + AgentConfigSnapshot, + AgentScope, + AgentSource, + AgentStatus, + WorkflowAgentBindingType, + WorkflowAgentNodeBinding, +) from models.agent_config_entities import AgentSoulConfig, AgentSoulModelConfig, WorkflowNodeJobConfig @@ -104,6 +115,89 @@ def test_binding_resolver_uses_active_snapshot_for_roster_agent(monkeypatch: pyt assert bundle.snapshot.id == "active-snapshot" +def test_binding_resolver_rejects_unpublished_roster_agent(monkeypatch: pytest.MonkeyPatch): + binding = _binding() + binding.binding_type = WorkflowAgentBindingType.ROSTER_AGENT + monkeypatch.setattr( + "core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session", + lambda: FakeSession([binding, None]), + ) + + with pytest.raises(WorkflowAgentBindingError) as exc_info: + WorkflowAgentBindingResolver().resolve(**_resolve()) + + assert exc_info.value.error_code == "agent_not_available" + assert "not been published" in str(exc_info.value) + + +@pytest.mark.parametrize( + "sqlite_session", + [(Agent, AgentConfigSnapshot, AgentConfigRevision, WorkflowAgentNodeBinding)], + indirect=True, +) +def test_binding_resolver_requires_publish_provenance_for_active_roster_snapshot( + monkeypatch: pytest.MonkeyPatch, + sqlite_session: Session, +) -> None: + binding = _binding() + binding.binding_type = WorkflowAgentBindingType.ROSTER_AGENT + binding.workflow_version = "draft" + agent = Agent( + id="agent-1", + tenant_id="tenant-1", + name="Imported Agent", + scope=AgentScope.ROSTER, + source=AgentSource.IMPORTED, + app_id="agent-app-1", + status=AgentStatus.ACTIVE, + active_config_snapshot_id="snapshot-1", + active_config_has_model=True, + # Dirty draft state must not hide a snapshot after it has publish provenance. + active_config_is_published=False, + ) + sqlite_session.add_all( + [ + binding, + agent, + _snapshot(), + AgentConfigRevision( + id="revision-import", + tenant_id="tenant-1", + agent_id="agent-1", + current_snapshot_id="snapshot-1", + revision=1, + operation=AgentConfigRevisionOperation.IMPORT_PACKAGE, + ), + ] + ) + sqlite_session.commit() + monkeypatch.setattr( + "core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session", + lambda: sqlite_session, + ) + + with pytest.raises(WorkflowAgentBindingError) as exc_info: + WorkflowAgentBindingResolver().resolve(**_resolve()) + assert exc_info.value.error_code == "agent_not_available" + + sqlite_session.add( + AgentConfigRevision( + id="revision-publish", + tenant_id="tenant-1", + agent_id="agent-1", + current_snapshot_id="snapshot-1", + revision=2, + operation=AgentConfigRevisionOperation.PUBLISH_DRAFT, + ) + ) + sqlite_session.commit() + + bundle = WorkflowAgentBindingResolver().resolve(**_resolve()) + + assert bundle.agent.id == agent.id + assert bundle.snapshot.id == "snapshot-1" + + def test_binding_resolver_raises_when_binding_missing(monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr( "core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session", diff --git a/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_validators.py b/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_validators.py index acf27ae2b03..c95f30b81c7 100644 --- a/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_validators.py +++ b/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_validators.py @@ -156,6 +156,19 @@ def test_publish_validation_uses_active_snapshot_for_roster_agent(): ) +def test_publish_validation_rejects_unpublished_roster_agent(): + binding = _binding(WorkflowNodeJobConfig()) + binding.binding_type = WorkflowAgentBindingType.ROSTER_AGENT + session = Mock() + session.scalar.side_effect = [binding, None] + + with pytest.raises(WorkflowAgentNodeValidationError, match="unpublished roster agent"): + WorkflowAgentNodeValidator.validate_published_workflow( + session=session, + workflow=_workflow(_graph([{"source": "start", "target": "agent-node"}])), + ) + + def test_publish_validation_rejects_non_upstream_previous_output_ref(): node_job = WorkflowNodeJobConfig.model_validate( {"previous_node_output_refs": [{"node_id": "later-node", "output": "text"}]} diff --git a/api/tests/unit_tests/services/agent/test_agent_services.py b/api/tests/unit_tests/services/agent/test_agent_services.py index 6ac297125ae..c1946968e67 100644 --- a/api/tests/unit_tests/services/agent/test_agent_services.py +++ b/api/tests/unit_tests/services/agent/test_agent_services.py @@ -592,6 +592,7 @@ def test_save_agent_app_composer_rejects_version_save_strategy(): def test_save_agent_app_composer_updates_normal_draft(monkeypatch: pytest.MonkeyPatch): agent = SimpleNamespace( id="agent-1", + tenant_id="tenant-1", source=AgentSource.AGENT_APP, active_config_snapshot_id="version-1", active_config_is_published=True, @@ -639,13 +640,14 @@ def test_save_agent_app_composer_keeps_published_when_draft_matches_active_snaps agent_soul = _agent_soul_with_model() agent = SimpleNamespace( id="agent-1", + tenant_id="tenant-1", source=AgentSource.AGENT_APP, active_config_snapshot_id="version-1", active_config_is_published=False, updated_by=None, ) active_version = SimpleNamespace(config_snapshot_dict=agent_soul.model_dump(mode="json")) - fake_session = FakeSession(scalar=[agent], scalars=[[AgentConfigRevisionOperation.PUBLISH_DRAFT]]) + fake_session = FakeSession(scalar=[agent, "publish-revision-1"]) session = fake_session monkeypatch.setattr(composer_service.ComposerConfigValidator, "validate_draft_save_payload", lambda payload: None) @@ -822,8 +824,7 @@ def test_agent_app_build_draft_checkout_and_apply_use_user_isolated_draft(monkey active_version = SimpleNamespace(config_snapshot_dict=build_draft.config_snapshot_dict) fake_session = FakeSession( - scalar=[agent, build_draft, normal_draft, active_version], - scalars=[[AgentConfigRevisionOperation.PUBLISH_DRAFT]], + scalar=[agent, build_draft, normal_draft, active_version, "publish-revision-1"], ) session = fake_session @@ -1612,7 +1613,7 @@ def test_node_job_only_rejects_inline_binding_pointing_to_roster_agent(monkeypat def test_copy_workflow_composer_from_roster_creates_inline_agent_and_preserves_node_job( monkeypatch: pytest.MonkeyPatch, ): - fake_session = FakeSession() + fake_session = FakeSession(scalar=["publish-revision-1"]) session = fake_session workflow = SimpleNamespace(id="workflow-1") node_job = WorkflowNodeJobConfig(workflow_prompt="keep this node task") @@ -1710,7 +1711,7 @@ def test_copy_workflow_composer_from_roster_creates_inline_agent_and_preserves_n def test_copy_workflow_composer_from_roster_rejects_stale_source_snapshot(monkeypatch: pytest.MonkeyPatch): - session = FakeSession() + session = FakeSession(scalar=["publish-revision-1"]) monkeypatch.setattr(AgentComposerService, "_get_draft_workflow", lambda **kwargs: SimpleNamespace(id="workflow-1")) monkeypatch.setattr( AgentComposerService, @@ -1758,6 +1759,49 @@ def test_copy_workflow_composer_from_roster_rejects_stale_source_snapshot(monkey ) +def test_copy_workflow_composer_from_roster_rejects_unpublished_source(monkeypatch: pytest.MonkeyPatch): + session = FakeSession() + binding = WorkflowAgentNodeBinding( + tenant_id="tenant-1", + app_id="app-1", + workflow_id="workflow-1", + workflow_version="draft", + node_id="node-1", + binding_type=WorkflowAgentBindingType.ROSTER_AGENT, + agent_id="roster-agent-1", + current_snapshot_id="roster-version-1", + node_job_config=WorkflowNodeJobConfig(), + ) + source_agent = Agent( + id="roster-agent-1", + tenant_id="tenant-1", + name="Unpublished import", + scope=AgentScope.ROSTER, + source=AgentSource.IMPORTED, + app_id="agent-app-1", + status=AgentStatus.ACTIVE, + active_config_snapshot_id="roster-version-1", + ) + monkeypatch.setattr(AgentComposerService, "_get_draft_workflow", lambda **kwargs: SimpleNamespace(id="workflow-1")) + monkeypatch.setattr(AgentComposerService, "_get_workflow_binding", lambda **kwargs: binding) + monkeypatch.setattr(AgentComposerService, "_require_agent", lambda **kwargs: source_agent) + require_version = MagicMock(side_effect=AssertionError("unpublished source must fail before loading snapshot")) + monkeypatch.setattr(AgentComposerService, "_require_version", require_version) + + with pytest.raises(InvalidComposerConfigError, match="published config snapshot"): + AgentComposerService.copy_workflow_composer_from_roster( + session=session, + tenant_id="tenant-1", + app_id="app-1", + node_id="node-1", + account_id="account-1", + source_agent_id="roster-agent-1", + ) + + require_version.assert_not_called() + assert session.flushes == 0 + + def test_copy_workflow_composer_from_roster_is_idempotent_when_already_inline(monkeypatch: pytest.MonkeyPatch): inline_binding = WorkflowAgentNodeBinding( tenant_id="tenant-1", @@ -2203,7 +2247,7 @@ def test_agent_app_draft_match_does_not_mark_create_version_as_published(monkeyp active_config_snapshot_id="snapshot-1", ) snapshot = SimpleNamespace(config_snapshot_dict=agent_soul) - fake_session = FakeSession(scalars=[[AgentConfigRevisionOperation.CREATE_VERSION]]) + fake_session = FakeSession() session = fake_session monkeypatch.setattr(AgentComposerService, "_get_version_if_present", lambda **kwargs: snapshot) @@ -2227,7 +2271,7 @@ def test_agent_app_draft_match_marks_publish_visible_revision_as_published(monke active_config_snapshot_id="snapshot-1", ) snapshot = SimpleNamespace(config_snapshot_dict=agent_soul) - fake_session = FakeSession(scalars=[[AgentConfigRevisionOperation.PUBLISH_DRAFT]]) + fake_session = FakeSession(scalar=["publish-revision-1"]) session = fake_session monkeypatch.setattr(AgentComposerService, "_get_version_if_present", lambda **kwargs: snapshot) @@ -5095,6 +5139,7 @@ def test_composer_save_rejects_missing_or_out_of_scope_knowledge_datasets(monkey def test_save_agent_composer_allows_incomplete_knowledge_draft(monkeypatch: pytest.MonkeyPatch): agent = SimpleNamespace( id="agent-1", + tenant_id="tenant-1", source=AgentSource.AGENT_APP, active_config_snapshot_id="version-1", active_config_is_published=True, diff --git a/api/tests/unit_tests/services/agent/test_workflow_publish_service.py b/api/tests/unit_tests/services/agent/test_workflow_publish_service.py index 4a928cf1a0a..ef687714717 100644 --- a/api/tests/unit_tests/services/agent/test_workflow_publish_service.py +++ b/api/tests/unit_tests/services/agent/test_workflow_publish_service.py @@ -196,6 +196,19 @@ def test_resolve_existing_inline_binding_agent_returns_valid_agent_or_none(monke ) +def test_resolve_roster_binding_rejects_unpublished_agent() -> None: + session = Mock() + session.scalar.return_value = None + + with pytest.raises(ValueError, match="unavailable or unpublished roster agent"): + WorkflowAgentPublishService._resolve_roster_agent_graph_binding( + session=session, + draft_workflow=_workflow(), + node_id="agent-node", + agent_id="agent-1", + ) + + def test_clone_inline_graph_binding_for_node_clones_source(monkeypatch) -> None: session = Mock() source_agent = SimpleNamespace(id="source-agent")