opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-25 21:48:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(agent): block unpublished agents in workflows (#39532)

Browse Source 
Co-authored-by: yyh <92089059+lyzno1@users.noreply.github.com>
This commit is contained in:
zyssyz123 2026-07-24 19:02:43 +08:00 committed by GitHub
parent a5703762c0
commit a877e1bd7e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 669 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

93
api/core/agent/publish_visibility.py Normal file
View File

@ -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)
)
)

30
api/core/workflow/nodes/agent_v2/binding_resolver.py
View File

@ -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 = (

37
api/core/workflow/nodes/agent_v2/validators.py
View File

@ -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

35
api/services/agent/composer_service.py
View File

@ -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

6
api/services/agent/roster_service.py
View File

@ -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())

6
api/services/agent/workflow_publish_service.py
View File

@ -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:

341
api/tests/unit_tests/core/agent/test_publish_visibility.py Normal file
View File

@ -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"] == []

96
api/tests/unit_tests/core/workflow/nodes/agent_v2/test_binding_resolver.py
View File

@ -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",

13
api/tests/unit_tests/core/workflow/nodes/agent_v2/test_validators.py
View File

@ -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"}]}

59
api/tests/unit_tests/services/agent/test_agent_services.py
View File

@ -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,

13
api/tests/unit_tests/services/agent/test_workflow_publish_service.py
View File

@ -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")