dify/api/tests/unit_tests/controllers/console/agent/test_roster.py

from collections.abc import Generator
from datetime import datetime
from inspect import getsource, unwrap
from typing import Any, Self, cast
from unittest.mock import MagicMock

import pytest
from flask import Flask
from sqlalchemy.orm import Session
from werkzeug.exceptions import InternalServerError, NotFound

from controllers.console import console_ns
from controllers.console.agent import composer as composer_controller
from controllers.console.agent import roster as roster_controller
from controllers.console.agent.composer import (
    AgentComposerApi,
    AgentComposerCandidatesApi,
    WorkflowAgentComposerApi,
    WorkflowAgentComposerCandidatesApi,
    WorkflowAgentComposerCopyFromRosterApi,
    WorkflowAgentComposerImpactApi,
    WorkflowAgentComposerSaveToRosterApi,
    WorkflowAgentComposerValidateApi,
)
from controllers.console.agent.roster import (
    AgentAppCopyApi,
    AgentAppCopyPayload,
    AgentAppListApi,
    AgentBuildDraftApi,
    AgentBuildDraftApplyApi,
    AgentBuildDraftCheckoutApi,
    AgentBuildDraftCheckoutPayload,
    AgentDebugConversationRefreshApi,
    AgentInviteOptionsApi,
    AgentLogMessagesApi,
    AgentLogsApi,
    AgentLogSourcesApi,
    AgentPublishApi,
    AgentPublishPayload,
    AgentStatisticsQuery,
    AgentStatisticsSummaryApi,
)
from controllers.console.app import completion as completion_controller
from controllers.console.app.completion import AgentBuildChatFinalizeApi, AgentChatMessageApi, AgentChatMessageStopApi
from controllers.console.app.error import CompletionRequestError
from models.account import Account, TenantAccountRole
from models.agent import Agent, AgentConfigDraftType, AgentScope, AgentSource
from models.model import App, AppMode, IconType
from services.agent.observability_service import AgentLogQueryParams, AgentStatisticsQueryParams
from services.entities.agent_entities import (
    ComposerSavePayload,
    ComposerSaveStrategy,
    ComposerVariant,
    WorkflowAgentComposerQuery,
    WorkflowComposerCopyFromRosterPayload,
)
from tests.helpers.unit.model_factories import make_account
from tests.helpers.unit.rbac_fakes import RBACDomain


@pytest.fixture(autouse=True)
def rbac_services(monkeypatch: pytest.MonkeyPatch, rbac_domain: RBACDomain) -> None:
    monkeypatch.setattr(roster_controller, "application_services", lambda: rbac_domain)


def _version_response(version_id: str = "version-1") -> dict[str, object]:
    return {
        "id": version_id,
        "agent_id": "agent-1",
        "version": 1,
        "summary": None,
        "version_note": None,
        "created_by": "account-1",
        "created_at": None,
    }


def test_query_values_accepts_repeated_and_indexed_arrays() -> None:
    app = Flask(__name__)

    with app.test_request_context("/?sources=webapp:app-1&sources%5B1%5D=workflow:app-2&sources%5B0%5D=workflow:app-1"):
        assert roster_controller._query_values("sources", "source") == [
            "webapp:app-1",
            "workflow:app-1",
            "workflow:app-2",
        ]

    with app.test_request_context("/?source%5B0%5D=workflow:app-3"):
        assert roster_controller._query_values("sources", "source") == ["workflow:app-3"]


def _workflow_composer_response(**overrides: object) -> dict[str, object]:
    response: dict[str, object] = {
        "variant": "workflow",
        "agent": None,
        "active_config_snapshot": None,
        "binding": None,
        "soul_lock": {"locked": False, "can_unlock": False, "reason": "workflow_only_empty"},
        "agent_soul": {},
        "node_job": {},
        "effective_declared_outputs": [],
        "save_options": ["node_job_only"],
        "impact_summary": None,
        "app_id": "app-1",
        "workflow_id": "workflow-1",
        "node_id": "node-1",
    }
    response.update(overrides)
    return response


def _agent_app_composer_response() -> dict[str, object]:
    return {
        "variant": "agent_app",
        "agent": {
            "id": "agent-1",
            "name": "Analyst",
            "description": "",
            "scope": "roster",
            "status": "active",
            "active_config_snapshot_id": "version-1",
        },
        "active_config_snapshot": _version_response(),
        "active_config_is_published": True,
        "agent_soul": {},
        "save_options": ["save_to_current_version"],
    }


def _app_detail_obj(**overrides: object) -> App:
    data: dict[str, object] = {
        "id": "app-1",
        "tenant_id": "tenant-1",
        "name": "Iris",
        "description": "Agent app",
        "mode": AppMode.AGENT,
        "icon_type": "emoji",
        "icon": "robot",
        "icon_background": "#fff",
        "enable_site": False,
        "enable_api": False,
        "tracing": None,
        "use_icon_as_answer_icon": False,
        "created_by": "account-1",
        "created_at": datetime(2025, 1, 1),
        "updated_by": "account-1",
        "updated_at": datetime(2025, 1, 1),
        "max_active_requests": 0,
    }
    overrides.pop("bound_agent_id", None)
    data.update(overrides)
    data["icon_type"] = IconType(data["icon_type"])
    app = App()
    for key, value in data.items():
        setattr(app, key, value)
    return app


def _account(*, account_id: str = "account-1", privileged: bool = False, timezone: str | None = None) -> Account:
    return make_account(
        account_id=account_id,
        name="Agent Controller Tester",
        email=f"{account_id}@example.com",
        timezone=timezone,
        role=TenantAccountRole.OWNER if privileged else None,
    )


def _candidates_response(variant: str) -> dict[str, object]:
    return {
        "variant": variant,
        "allowed_node_job_candidates": {},
        "allowed_soul_candidates": {},
        "capabilities": {"human_roster_available": False},
    }


def test_agent_v2_console_routes_are_agent_id_first() -> None:
    paths = {route for item in console_ns.resources for route in item.urls}
    for route in (
        "/agent",
        "/agent/<uuid:agent_id>",
        "/agent/<uuid:agent_id>/composer",
        "/agent/<uuid:agent_id>/composer/validate",
        "/agent/<uuid:agent_id>/composer/candidates",
        "/agent/<uuid:agent_id>/features",
        "/agent/<uuid:agent_id>/copy",
        "/agent/<uuid:agent_id>/publish",
        "/agent/<uuid:agent_id>/build-draft/checkout",
        "/agent/<uuid:agent_id>/build-draft",
        "/agent/<uuid:agent_id>/build-draft/apply",
        "/agent/<uuid:agent_id>/referencing-workflows",
        "/agent/<uuid:agent_id>/sandbox/files",
        "/agent/<uuid:agent_id>/api-access",
        "/agent/<uuid:agent_id>/api-enable",
        "/agent/<uuid:agent_id>/api-keys",
        "/agent/<uuid:agent_id>/api-keys/<uuid:api_key_id>",
        "/agent/<uuid:agent_id>/debug-conversation/refresh",
        "/agent/<uuid:agent_id>/chat-messages",
        "/agent/<uuid:agent_id>/chat-messages/<string:task_id>/stop",
        "/agent/<uuid:agent_id>/feedbacks",
        "/agent/<uuid:agent_id>/chat-messages/<uuid:message_id>/suggested-questions",
        "/agent/<uuid:agent_id>/messages/<uuid:message_id>",
        "/agent/<uuid:agent_id>/logs",
        "/agent/<uuid:agent_id>/logs/<uuid:conversation_id>/messages",
        "/agent/<uuid:agent_id>/log-sources",
        "/agent/<uuid:agent_id>/statistics/summary",
        "/agent/<uuid:agent_id>/versions",
        "/agent/<uuid:agent_id>/versions/<uuid:version_id>",
        "/agent/<uuid:agent_id>/versions/<uuid:version_id>/restore",
        "/agent/invite-options",
    ):
        assert route in paths
    for route in (
        "/agents",
        "/agents/invite-options",
        "/agents/<uuid:agent_id>",
        "/agents/<uuid:agent_id>/versions",
        "/apps/<uuid:app_id>/agent-composer",
        "/apps/<uuid:app_id>/agent-composer/validate",
        "/apps/<uuid:app_id>/agent-composer/candidates",
        "/apps/<uuid:app_id>/agent-features",
        "/apps/<uuid:app_id>/agent-referencing-workflows",
        "/apps/<uuid:app_id>/agent-sandbox/files",
        "/apps/<uuid:agent_id>/api-access",
    ):
        assert route not in paths


def test_agent_app_write_routes_do_not_reuse_app_billing_quota() -> None:
    for route_class in (AgentAppListApi, AgentAppCopyApi):
        assert '@cloud_edition_billing_resource_check("apps")' not in getsource(route_class)


def test_agent_list_reads_do_not_require_workspace_preview() -> None:
    for route_class in (AgentAppListApi, AgentInviteOptionsApi):
        assert "RBACPermission.AGENT_PREVIEW, Workspace()" not in getsource(route_class.get)


def test_agent_app_permission_keys_are_required_response_fields() -> None:
    assert roster_controller.AgentAppPartial.model_fields["permission_keys"].is_required()
    assert roster_controller.AgentAppDetailWithSite.model_fields["permission_keys"].is_required()


@pytest.fixture
def account_id() -> str:
    return "account-1"


def test_agent_app_create_payload_allows_optional_role() -> None:
    omitted = roster_controller.AgentAppCreatePayload.model_validate(
        {"name": "Iris", "description": "Agent app", "icon_type": "emoji", "icon": "robot"}
    )
    blank = roster_controller.AgentAppCreatePayload.model_validate(
        {"name": "Iris", "description": "Agent app", "role": "   ", "icon_type": "emoji", "icon": "robot"}
    )
    assert omitted.role is None
    assert blank.role == ""


def test_agent_app_copy_uses_agent_id_and_returns_agent_detail(
    app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str
) -> None:
    agent_id = "00000000-0000-0000-0000-000000000001"
    current_user = _account(account_id=account_id)
    copied_app = _app_detail_obj(id="copied-app", bound_agent_id="copied-agent")
    captured: dict[str, object] = {}

    class FakeRosterService:
        def duplicate_agent_app(self, **kwargs: object) -> object:
            captured.update(kwargs)
            return copied_app

    monkeypatch.setattr(roster_controller, "_agent_roster_service", lambda *_args: FakeRosterService())
    monkeypatch.setattr(
        roster_controller,
        "_serialize_agent_app_detail",
        lambda _session, app_model, **_kwargs: {"id": "copied-agent", "app_id": app_model.id, "name": app_model.name},
    )
    with app.test_request_context(
        "/console/api/agent/00000000-0000-0000-0000-000000000001/copy",
        json={
            "name": "Iris copy",
            "description": "Copied",
            "role": "Copied role",
            "icon_type": "emoji",
            "icon": "sparkles",
            "icon_background": "#fff",
        },
    ):
        copied, status = unwrap(AgentAppCopyApi.post)(
            AgentAppCopyApi(),
            AgentAppCopyPayload(
                name="Iris copy",
                description="Copied",
                role="Copied role",
                icon_type="emoji",
                icon="sparkles",
                icon_background="#fff",
            ),
            MagicMock(),
            "tenant-1",
            current_user,
            agent_id,
        )
    assert status == 201
    assert copied == {"id": "copied-agent", "app_id": "copied-app", "name": "Iris"}
    assert captured == {
        "tenant_id": "tenant-1",
        "agent_id": agent_id,
        "account": current_user,
        "name": "Iris copy",
        "description": "Copied",
        "role": "Copied role",
        "icon_type": "emoji",
        "icon": "sparkles",
        "icon_background": "#fff",
    }


def test_agent_debug_conversation_refresh_resets_build_for_current_user(
    app: Flask,
    monkeypatch: pytest.MonkeyPatch,
    account_id: str,
) -> None:
    agent_id = "00000000-0000-0000-0000-000000000001"
    captured: dict[str, object] = {}

    class FakeRosterService:
        def reset_build_conversation(self, **kwargs: object) -> str:
            captured.update(kwargs)
            return "new-debug-conversation-id"

    monkeypatch.setattr(roster_controller, "_agent_roster_service", lambda *_args: FakeRosterService())
    with app.test_request_context(
        "/console/api/agent/00000000-0000-0000-0000-000000000001/debug-conversation/refresh",
        method="POST",
    ):
        response = unwrap(AgentDebugConversationRefreshApi.post)(
            AgentDebugConversationRefreshApi(), MagicMock(), "tenant-1", _account(account_id=account_id), agent_id
        )
    assert response == {
        "debug_conversation_id": "new-debug-conversation-id",
        "debug_conversation_has_messages": False,
        "debug_conversation_message_count": 0,
    }
    assert captured == {
        "tenant_id": "tenant-1",
        "agent_id": agent_id,
        "account_id": account_id,
    }


def test_agent_publish_and_build_draft_routes_call_composer_service(
    app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str
) -> None:
    agent_id = "00000000-0000-0000-0000-000000000001"
    current_user = _account(account_id=account_id)
    captured: dict[str, dict[str, object]] = {}

    def publish_agent_app_draft(**kwargs: object) -> dict[str, object]:
        captured["publish"] = kwargs
        return {"result": "success", "active_config_snapshot_id": "version-1"}

    def checkout_agent_app_build_draft(**kwargs: object) -> dict[str, object]:
        captured["checkout"] = kwargs
        return {"variant": "agent_app", "draft": {"id": "build-draft-1"}, "agent_soul": {}}

    def load_agent_app_build_draft(**kwargs: object) -> dict[str, object]:
        captured["load"] = kwargs
        return {"variant": "agent_app", "draft": {"id": "build-draft-1"}, "agent_soul": {}}

    def save_agent_app_build_draft(**kwargs: object) -> dict[str, object]:
        captured["save"] = kwargs
        return {"variant": "agent_app", "draft": {"id": "build-draft-1"}, "agent_soul": {}}

    def apply_agent_app_build_draft(**kwargs: object) -> dict[str, object]:
        captured["apply"] = kwargs
        return {"result": "success", "draft": {"id": "draft-1"}}

    def discard_agent_app_build_draft(**kwargs: object) -> dict[str, object]:
        captured["discard"] = kwargs
        return {"result": "success"}

    monkeypatch.setattr(roster_controller.AgentComposerService, "publish_agent_app_draft", publish_agent_app_draft)
    monkeypatch.setattr(
        roster_controller.AgentComposerService, "checkout_agent_app_build_draft", checkout_agent_app_build_draft
    )
    monkeypatch.setattr(
        roster_controller.AgentComposerService, "load_agent_app_build_draft", load_agent_app_build_draft
    )
    monkeypatch.setattr(
        roster_controller.AgentComposerService, "save_agent_app_build_draft", save_agent_app_build_draft
    )
    monkeypatch.setattr(
        roster_controller.AgentComposerService, "apply_agent_app_build_draft", apply_agent_app_build_draft
    )
    monkeypatch.setattr(
        roster_controller.AgentComposerService, "discard_agent_app_build_draft", discard_agent_app_build_draft
    )
    with app.test_request_context(
        "/console/api/agent/00000000-0000-0000-0000-000000000001/publish", json={"version_note": "publish v1"}
    ):
        published = unwrap(AgentPublishApi.post)(
            AgentPublishApi(),
            AgentPublishPayload(version_note="publish v1"),
            MagicMock(),
            "tenant-1",
            current_user,
            agent_id,
        )
    assert published["active_config_snapshot_id"] == "version-1"
    captured["publish"].pop("session", None)
    assert captured["publish"] == {
        "tenant_id": "tenant-1",
        "agent_id": agent_id,
        "account_id": account_id,
        "version_note": "publish v1",
    }
    with app.test_request_context(
        "/console/api/agent/00000000-0000-0000-0000-000000000001/build-draft/checkout", json={"force": True}
    ):
        checked_out = unwrap(AgentBuildDraftCheckoutApi.post)(
            AgentBuildDraftCheckoutApi(),
            AgentBuildDraftCheckoutPayload(force=True),
            MagicMock(),
            "tenant-1",
            current_user,
            agent_id,
        )
    assert checked_out["draft"]["id"] == "build-draft-1"
    captured["checkout"].pop("session", None)
    assert captured["checkout"] == {
        "tenant_id": "tenant-1",
        "agent_id": agent_id,
        "account_id": account_id,
        "force": True,
    }
    with app.test_request_context("/console/api/agent/00000000-0000-0000-0000-000000000001/build-draft"):
        loaded = unwrap(AgentBuildDraftApi.get)(AgentBuildDraftApi(), MagicMock(), "tenant-1", current_user, agent_id)
    assert loaded["draft"]["id"] == "build-draft-1"
    captured["load"].pop("session", None)
    assert captured["load"] == {"tenant_id": "tenant-1", "agent_id": agent_id, "account_id": account_id}
    with app.test_request_context(
        "/console/api/agent/00000000-0000-0000-0000-000000000001/build-draft",
        json={"variant": "agent_app", "save_strategy": "save_to_current_version", "agent_soul": {}},
    ):
        saved = unwrap(AgentBuildDraftApi.put)(
            AgentBuildDraftApi(),
            ComposerSavePayload(
                variant=ComposerVariant.AGENT_APP,
                save_strategy=ComposerSaveStrategy.SAVE_TO_CURRENT_VERSION,
            ),
            MagicMock(),
            "tenant-1",
            current_user,
            agent_id,
        )
    assert saved["draft"]["id"] == "build-draft-1"
    assert captured["save"]["tenant_id"] == "tenant-1"
    assert captured["save"]["agent_id"] == agent_id
    assert captured["save"]["account_id"] == account_id
    assert cast(Any, captured["save"]["payload"]).variant == ComposerVariant.AGENT_APP
    with app.test_request_context(
        "/console/api/agent/00000000-0000-0000-0000-000000000001/build-draft/apply", method="POST"
    ):
        applied = unwrap(AgentBuildDraftApplyApi.post)(
            AgentBuildDraftApplyApi(), MagicMock(), "tenant-1", current_user, agent_id
        )
    assert applied == {"result": "success", "draft": {"id": "draft-1"}}
    captured["apply"].pop("session", None)
    assert captured["apply"] == {"tenant_id": "tenant-1", "agent_id": agent_id, "account_id": account_id}
    with app.test_request_context(
        "/console/api/agent/00000000-0000-0000-0000-000000000001/build-draft", method="DELETE"
    ):
        discarded = unwrap(AgentBuildDraftApi.delete)(
            AgentBuildDraftApi(), MagicMock(), "tenant-1", current_user, agent_id
        )
    assert discarded == {"result": "success"}
    captured["discard"].pop("session", None)
    assert captured["discard"] == {"tenant_id": "tenant-1", "agent_id": agent_id, "account_id": account_id}


def test_agent_observability_routes_resolve_app_from_agent_id(
    app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str
) -> None:
    agent_id = "00000000-0000-0000-0000-000000000001"
    app_model = _app_detail_obj(id="app-1")
    captured: dict[str, object] = {}

    class FakeObservabilityService:
        def list_logs(self, *, app: App, agent_id: str, params: AgentLogQueryParams) -> dict[str, object]:
            captured["logs"] = {"app": app, "agent_id": agent_id, "params": params}
            return {
                "data": [
                    {
                        "conversation_id": "conversation-1",
                        "id": "conversation-1",
                        "title": "Debug",
                        "end_user_id": "end-user-1",
                        "message_count": 2,
                        "user_rate": None,
                        "operation_rate": None,
                        "unread": True,
                        "source": {
                            "id": "webapp:app-1",
                            "type": "webapp",
                            "app_id": "app-1",
                            "app_name": "Iris",
                            "app_icon_type": "emoji",
                            "app_icon": "robot",
                            "app_icon_background": "#fff",
                            "workflow_id": None,
                            "workflow_version": None,
                            "node_id": None,
                        },
                        "status": "success",
                        "created_at": 1,
                        "updated_at": 2,
                    }
                ],
                "page": 2,
                "limit": 5,
                "total": 6,
                "has_more": False,
            }

        def list_log_messages(
            self, *, app: App, agent_id: str, conversation_id: str, params: AgentLogQueryParams
        ) -> dict[str, object]:
            captured["messages"] = {
                "app": app,
                "agent_id": agent_id,
                "conversation_id": conversation_id,
                "params": params,
            }
            return {
                "data": [
                    {
                        "id": "message-1",
                        "message_id": "message-1",
                        "conversation_id": "conversation-1",
                        "query": "hello",
                        "answer": "hi",
                        "status": "success",
                        "error": None,
                        "from_end_user_id": None,
                        "from_account_id": account_id,
                        "message_tokens": 1,
                        "answer_tokens": 2,
                        "total_tokens": 3,
                        "total_price": "0",
                        "currency": "USD",
                        "latency": 1.2,
                        "created_at": 1,
                        "updated_at": 2,
                    }
                ],
                "page": 1,
                "limit": 20,
                "total": 1,
                "has_more": False,
            }

        def list_log_sources(self, *, app: App, agent_id: str) -> dict[str, object]:
            captured["sources"] = {"app": app, "agent_id": agent_id}
            return {
                "data": [
                    {
                        "id": "webapp:app-1",
                        "type": "webapp",
                        "app_id": "app-1",
                        "app_name": "Iris",
                        "app_icon_type": "emoji",
                        "app_icon": "robot",
                        "app_icon_background": "#fff",
                        "workflow_id": None,
                        "workflow_version": None,
                        "node_id": None,
                    }
                ],
                "groups": [{"type": "webapp", "label": "WEBAPP", "sources": []}],
            }

        def get_statistics_summary(
            self, *, app: App, agent_id: str, params: AgentStatisticsQueryParams
        ) -> dict[str, object]:
            captured["statistics"] = {"app": app, "agent_id": agent_id, "params": params}
            return {
                "source": "all",
                "summary": {
                    "total_messages": 1,
                    "total_conversations": 1,
                    "total_end_users": 1,
                    "total_tokens": 3,
                    "total_price": "0",
                    "currency": "USD",
                    "average_session_interactions": 1,
                    "average_response_time": 1200,
                    "tokens_per_second": 2,
                    "user_satisfaction_rate": 100,
                },
                "charts": {
                    "daily_messages": [{"date": "2026-06-17", "message_count": 1}],
                    "daily_conversations": [{"date": "2026-06-17", "conversation_count": 1}],
                    "daily_end_users": [{"date": "2026-06-17", "terminal_count": 1}],
                    "token_usage": [{"date": "2026-06-17", "token_count": 3, "total_price": "0", "currency": "USD"}],
                    "average_session_interactions": [{"date": "2026-06-17", "interactions": 1}],
                    "average_response_time": [{"date": "2026-06-17", "latency": 1200}],
                    "tokens_per_second": [{"date": "2026-06-17", "tps": 2}],
                    "user_satisfaction_rate": [{"date": "2026-06-17", "rate": 100}],
                },
            }

    monkeypatch.setattr(roster_controller, "_resolve_agent_runtime_app_model", lambda _session, **kwargs: app_model)
    monkeypatch.setattr(roster_controller, "_agent_observability_service", lambda *_args: FakeObservabilityService())
    account = _account(account_id=account_id, timezone="UTC")
    with app.test_request_context(
        "/console/api/agent/00000000-0000-0000-0000-000000000001/logs?page=2&limit=5&keyword=hello&statuses=success&statuses=failed&sources=webapp:app-1&sources=workflow:app-2:workflow-1:v1:node-1&sort_by=created_at&sort_order=asc"
    ):
        logs = unwrap(AgentLogsApi.get)(AgentLogsApi(), MagicMock(), "tenant-1", account, agent_id)
    assert logs["data"][0]["id"] == "conversation-1"
    assert logs["data"][0]["source"]["id"] == "webapp:app-1"
    logs_call = cast(dict[str, object], captured["logs"])
    assert logs_call["app"] is app_model
    assert logs_call["agent_id"] == agent_id
    logs_params = cast(Any, logs_call["params"])
    assert logs_params.page == 2
    assert logs_params.limit == 5
    assert logs_params.keyword == "hello"
    assert logs_params.statuses == ("success", "failed")
    assert logs_params.sources == ("webapp:app-1", "workflow:app-2:workflow-1:v1:node-1")
    assert logs_params.sort_by == "created_at"
    assert logs_params.sort_order == "asc"
    with app.test_request_context(
        "/console/api/agent/00000000-0000-0000-0000-000000000001/logs/00000000-0000-0000-0000-000000000002/messages"
    ):
        messages = unwrap(AgentLogMessagesApi.get)(
            AgentLogMessagesApi(), MagicMock(), "tenant-1", account, agent_id, "00000000-0000-0000-0000-000000000002"
        )
    assert messages["data"][0]["id"] == "message-1"
    messages_call = cast(dict[str, object], captured["messages"])
    assert messages_call["app"] is app_model
    assert messages_call["agent_id"] == agent_id
    assert messages_call["conversation_id"] == "00000000-0000-0000-0000-000000000002"
    messages_params = cast(Any, messages_call["params"])
    assert messages_params.sources == ()
    assert messages_params.statuses == ()
    with app.test_request_context("/console/api/agent/00000000-0000-0000-0000-000000000001/log-sources"):
        sources = unwrap(AgentLogSourcesApi.get)(AgentLogSourcesApi(), MagicMock(), "tenant-1", account, agent_id)
    assert sources["data"][0]["id"] == "webapp:app-1"
    sources_call = cast(dict[str, object], captured["sources"])
    assert sources_call["app"] is app_model
    assert sources_call["agent_id"] == agent_id
    with app.test_request_context(
        "/console/api/agent/00000000-0000-0000-0000-000000000001/statistics/summary?source=api"
    ):
        statistics = unwrap(AgentStatisticsSummaryApi.get)(
            AgentStatisticsSummaryApi(), AgentStatisticsQuery(source="api"), MagicMock(), "tenant-1", account, agent_id
        )
    assert statistics["summary"]["total_messages"] == 1
    stats_call = cast(dict[str, object], captured["statistics"])
    assert stats_call["app"] is app_model
    assert stats_call["agent_id"] == agent_id
    stats_params = cast(Any, stats_call["params"])
    assert stats_params.source == "api"
    assert stats_params.timezone == "UTC"


def test_workflow_composer_get_put_validate_candidates_impact_and_save(
    app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str
) -> None:
    app_model = _app_detail_obj(id="app-1")
    payload = {
        "variant": ComposerVariant.WORKFLOW.value,
        "save_strategy": ComposerSaveStrategy.NODE_JOB_ONLY.value,
        "binding": {"binding_type": "roster_agent", "current_snapshot_id": "version-1"},
    }
    captured_load: dict[str, object] = {}
    monkeypatch.setattr(
        composer_controller.AgentComposerService,
        "load_workflow_composer",
        lambda **kwargs: captured_load.update(kwargs) or _workflow_composer_response(node_id=kwargs["node_id"]),
    )
    monkeypatch.setattr(
        composer_controller.AgentComposerService,
        "save_workflow_composer",
        lambda **kwargs: _workflow_composer_response(save_options=[kwargs["payload"].save_strategy.value]),
    )
    monkeypatch.setattr(composer_controller.ComposerConfigValidator, "validate_publish_payload", lambda payload: None)
    monkeypatch.setattr(
        composer_controller.AgentComposerService,
        "get_workflow_candidates",
        lambda **kwargs: _candidates_response("workflow"),
    )
    monkeypatch.setattr(
        composer_controller.AgentComposerService,
        "calculate_impact",
        lambda **kwargs: {
            "current_snapshot_id": kwargs["current_snapshot_id"],
            "workflow_node_count": 1,
            "bindings": [],
        },
    )
    with app.test_request_context("?snapshot_id=preview-version"):
        workflow_state = unwrap(WorkflowAgentComposerApi.get)(
            WorkflowAgentComposerApi(),
            WorkflowAgentComposerQuery(snapshot_id="preview-version"),
            MagicMock(),
            "tenant-1",
            account_id,
            app_model,
            "node-1",
        )
    assert workflow_state["node_id"] == "node-1"
    assert captured_load["account_id"] == account_id
    assert captured_load["snapshot_id"] == "preview-version"
    composer_save_payload = ComposerSavePayload(
        variant=ComposerVariant.WORKFLOW,
        save_strategy=ComposerSaveStrategy.NODE_JOB_ONLY,
        binding={"binding_type": "roster_agent", "current_snapshot_id": "version-1"},
    )
    with app.test_request_context(json=payload):
        saved_state = unwrap(WorkflowAgentComposerApi.put)(
            WorkflowAgentComposerApi(),
            composer_save_payload,
            MagicMock(),
            "tenant-1",
            account_id,
            app_model,
            "node-1",
        )
        assert saved_state["save_options"] == ["node_job_only"]
        assert unwrap(WorkflowAgentComposerValidateApi.post)(
            WorkflowAgentComposerValidateApi(),
            composer_save_payload,
            MagicMock(),
            "tenant-1",
            app_model,
            "node-1",
        ) == {"result": "success", "errors": [], "warnings": [], "knowledge_retrieval_placeholder": []}
    assert (
        unwrap(WorkflowAgentComposerCandidatesApi.get)(
            WorkflowAgentComposerCandidatesApi(), MagicMock(), "tenant-1", account_id, app_model, "node-1"
        )["variant"]
        == "workflow"
    )
    with app.test_request_context(json=payload):
        assert unwrap(WorkflowAgentComposerImpactApi.post)(
            WorkflowAgentComposerImpactApi(),
            composer_save_payload,
            MagicMock(),
            "tenant-1",
            app_model,
            "node-1",
        ) == {"current_snapshot_id": "version-1", "workflow_node_count": 1, "bindings": []}
        assert unwrap(WorkflowAgentComposerSaveToRosterApi.post)(
            WorkflowAgentComposerSaveToRosterApi(),
            composer_save_payload,
            MagicMock(),
            "tenant-1",
            account_id,
            app_model,
            "node-1",
        )["save_options"] == ["node_job_only"]


def test_workflow_composer_get_uses_write_transaction() -> None:
    assert "@with_session\n    @get_app_model" in getsource(WorkflowAgentComposerApi)


def test_build_draft_apply_leaves_transaction_ownership_to_service() -> None:
    assert "@with_session(write=False)\n    def post" in getsource(AgentBuildDraftApplyApi)


def test_workflow_composer_copy_from_roster(app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str) -> None:
    app_model = _app_detail_obj(id="app-1")
    captured: dict[str, object] = {}

    def fake_copy_from_roster(**kwargs: object) -> dict[str, object]:
        captured.update(kwargs)
        return _workflow_composer_response(
            binding={
                "id": "binding-1",
                "binding_type": "inline_agent",
                "agent_id": "inline-agent-1",
                "current_snapshot_id": "inline-version-1",
                "workflow_id": "workflow-1",
                "node_id": kwargs["node_id"],
            },
            agent={
                "id": "inline-agent-1",
                "name": "Nadia",
                "description": "",
                "scope": "workflow_only",
                "status": "active",
            },
            active_config_snapshot={"id": "inline-version-1", "version": 1},
        )

    monkeypatch.setattr(
        composer_controller.AgentComposerService, "copy_workflow_composer_from_roster", fake_copy_from_roster
    )
    with app.test_request_context(
        json={
            "source_agent_id": "roster-agent-1",
            "source_snapshot_id": "roster-version-1",
            "idempotency_key": "copy-1",
        }
    ):
        result = unwrap(WorkflowAgentComposerCopyFromRosterApi.post)(
            WorkflowAgentComposerCopyFromRosterApi(),
            WorkflowComposerCopyFromRosterPayload(
                source_agent_id="roster-agent-1",
                source_snapshot_id="roster-version-1",
                idempotency_key="copy-1",
            ),
            MagicMock(),
            "tenant-1",
            account_id,
            app_model,
            "node-1",
        )
    assert result["binding"]["binding_type"] == "inline_agent"
    captured.pop("session", None)
    assert captured == {
        "tenant_id": "tenant-1",
        "app_id": "app-1",
        "node_id": "node-1",
        "account_id": account_id,
        "source_agent_id": "roster-agent-1",
        "source_snapshot_id": "roster-version-1",
        "idempotency_key": "copy-1",
    }


def test_workflow_impact_returns_empty_without_version(app: Flask) -> None:
    payload = {"variant": ComposerVariant.WORKFLOW.value, "save_strategy": ComposerSaveStrategy.NODE_JOB_ONLY.value}
    with app.test_request_context(json=payload):
        result = unwrap(WorkflowAgentComposerImpactApi.post)(
            WorkflowAgentComposerImpactApi(),
            ComposerSavePayload(
                variant=ComposerVariant.WORKFLOW,
                save_strategy=ComposerSaveStrategy.NODE_JOB_ONLY,
            ),
            MagicMock(),
            "tenant-1",
            _app_detail_obj(id="app-1"),
            "node-1",
        )
    assert result == {"current_snapshot_id": None, "workflow_node_count": 0, "bindings": []}


def test_agent_composer_routes_resolve_app_from_agent_id(
    app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str
) -> None:
    agent_id = "00000000-0000-0000-0000-000000000001"
    captured: dict[str, object] = {}
    payload = {
        "variant": ComposerVariant.AGENT_APP.value,
        "save_strategy": ComposerSaveStrategy.SAVE_TO_CURRENT_VERSION.value,
        "agent_soul": {"prompt": {"system_prompt": "x"}},
    }

    def load_agent_composer(**kwargs: object) -> dict[str, object]:
        captured["load"] = kwargs
        return _agent_app_composer_response()

    def save_agent_composer(**kwargs: object) -> dict[str, object]:
        captured["save"] = kwargs
        return _agent_app_composer_response()

    def get_agent_app_candidates(**kwargs: object) -> dict[str, object]:
        captured["candidates"] = kwargs
        return _candidates_response("agent_app")

    monkeypatch.setattr(composer_controller.AgentComposerService, "load_agent_composer", load_agent_composer)
    monkeypatch.setattr(composer_controller.AgentComposerService, "save_agent_composer", save_agent_composer)
    monkeypatch.setattr(composer_controller.ComposerConfigValidator, "validate_publish_payload", lambda payload: None)
    monkeypatch.setattr(composer_controller.AgentComposerService, "get_agent_app_candidates", get_agent_app_candidates)
    composer = unwrap(AgentComposerApi.get)(AgentComposerApi(), MagicMock(), "tenant-1", agent_id)
    assert composer["variant"] == "agent_app"
    assert composer["active_config_is_published"] is True
    assert cast(dict[str, object], captured["load"])["agent_id"] == agent_id
    composer_save_payload = ComposerSavePayload(
        variant=ComposerVariant.AGENT_APP,
        save_strategy=ComposerSaveStrategy.SAVE_TO_CURRENT_VERSION,
        agent_soul={"prompt": {"system_prompt": "x"}},
    )
    with app.test_request_context(json=payload):
        saved_composer = unwrap(AgentComposerApi.put)(
            AgentComposerApi(), composer_save_payload, MagicMock(), "tenant-1", account_id, agent_id
        )
        assert saved_composer["variant"] == "agent_app"
        assert saved_composer["active_config_is_published"] is True
        assert cast(dict[str, object], captured["save"])["agent_id"] == agent_id
    candidates = unwrap(AgentComposerCandidatesApi.get)(
        AgentComposerCandidatesApi(), MagicMock(), "tenant-1", account_id, agent_id
    )
    assert candidates["variant"] == "agent_app"
    assert cast(dict[str, object], captured["candidates"])["agent_id"] == agent_id


def test_agent_composer_get_uses_read_only_session() -> None:
    assert "@with_session(write=False)\n    def get" in getsource(AgentComposerApi)


def test_agent_composer_get_accepts_missing_draft(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
    payload = _agent_app_composer_response()
    payload["draft"] = None
    payload["agent_soul"] = {"prompt": {"system_prompt": "read-only snapshot"}}
    monkeypatch.setattr(composer_controller.AgentComposerService, "load_agent_composer", lambda **_kwargs: payload)
    with app.test_request_context():
        result = unwrap(AgentComposerApi.get)(AgentComposerApi(), MagicMock(), "tenant-1", "agent-1")
    assert result["draft"] is None
    assert result["agent_soul"]["prompt"]["system_prompt"] == "read-only snapshot"


def test_agent_chat_generate_and_stop_routes_resolve_app_from_agent_id(
    app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str, unbound_session: Session
) -> None:
    agent_id = "00000000-0000-0000-0000-000000000001"
    app_model = _app_detail_obj(id="app-1", tenant_id="tenant-1", mode=AppMode.AGENT)
    captured: dict[str, object] = {}

    def resolve_agent_app_model(**kwargs: object) -> object:
        captured["stop_resolve"] = kwargs
        return app_model

    def resolve_agent_app_model_with_session(_self, **kwargs: object) -> object:
        captured["resolve_session"] = _self._session
        captured["resolve"] = kwargs
        return app_model

    def create_chat_message(**kwargs: object) -> dict[str, object]:
        captured["create"] = kwargs
        return {"result": "generated"}

    def stop_chat_message(**kwargs: object) -> tuple[dict[str, object], int]:
        captured["stop"] = kwargs
        return ({"result": "success"}, 200)

    monkeypatch.setattr(completion_controller, "resolve_agent_runtime_app_model", resolve_agent_app_model)
    monkeypatch.setattr(
        completion_controller.AgentRosterService,
        "get_agent_runtime_app_model",
        resolve_agent_app_model_with_session,
    )
    monkeypatch.setattr(completion_controller, "_create_chat_message", create_chat_message)
    monkeypatch.setattr(completion_controller, "_stop_chat_message", stop_chat_message)
    with app.test_request_context(json={"inputs": {}, "query": "hello"}):
        assert unwrap(AgentChatMessageApi.post)(
            AgentChatMessageApi(), unbound_session, "tenant-1", _account(account_id=account_id), agent_id
        ) == {"result": "generated"}
    assert cast(dict[str, object], captured["resolve"]) == {"tenant_id": "tenant-1", "agent_id": agent_id}
    assert captured["resolve_session"] is unbound_session
    create_call = cast(dict[str, object], captured["create"])
    assert create_call["session"] is unbound_session
    assert create_call["app_model"] is app_model
    assert cast(Account, create_call["current_user"]).id == account_id
    assert unwrap(AgentChatMessageStopApi.post)(
        AgentChatMessageStopApi(), unbound_session, "tenant-1", account_id, agent_id, "task-1"
    ) == ({"result": "success"}, 200)
    assert captured["stop_resolve"] == {
        "session": unbound_session,
        "tenant_id": "tenant-1",
        "agent_id": agent_id,
    }
    stop_call = cast(dict[str, object], captured["stop"])
    assert stop_call == {"current_user_id": account_id, "app_model": app_model, "task_id": "task-1"}


def test_agent_chat_stream_preflight_raises_first_error_event() -> None:
    class ClosableStream:
        def __init__(self) -> None:
            self.closed = False
            self._chunks = iter(
                [
                    "event: ping\n\n",
                    (
                        'data: {"event":"error","message":"Incorrect API key provided",'
                        '"code":"completion_request_error","status":400}\n\n'
                    ),
                ]
            )

        def __iter__(self) -> Self:
            return self

        def __next__(self) -> str:
            return next(self._chunks)

        def close(self) -> None:
            self.closed = True

    stream = ClosableStream()
    with pytest.raises(CompletionRequestError) as exc_info:
        completion_controller._raise_agent_stream_error_before_response(stream)
    assert exc_info.value.description is not None
    assert "Incorrect API key provided" in exc_info.value.description
    assert stream.closed is True


def test_agent_chat_stream_preflight_preserves_first_normal_event() -> None:
    stream = iter(
        ["event: ping\n\n", 'data: {"event":"message","answer":"hello"}\n\n', 'data: {"event":"message_end"}\n\n']
    )
    wrapped = completion_controller._raise_agent_stream_error_before_response(stream)
    assert list(wrapped) == [
        "event: ping\n\n",
        'data: {"event":"message","answer":"hello"}\n\n',
        'data: {"event":"message_end"}\n\n',
    ]


def test_agent_build_chat_finalize_route_resolves_app_from_agent_id(
    app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str, unbound_session: Session
) -> None:
    agent_id = "00000000-0000-0000-0000-000000000001"
    app_model = _app_detail_obj(id="app-1", tenant_id="tenant-1", mode=AppMode.AGENT)
    captured: dict[str, object] = {}

    def resolve_agent_app_model(**kwargs: object) -> object:
        captured["resolve"] = kwargs
        return app_model

    def create_finalization_message(**kwargs: object) -> dict[str, object]:
        captured["finalize"] = kwargs
        return {"result": "generated"}

    monkeypatch.setattr(
        completion_controller.AgentRosterService,
        "get_agent_runtime_app_model",
        lambda _self, **kwargs: resolve_agent_app_model(**kwargs),
    )
    monkeypatch.setattr(completion_controller, "_create_build_chat_finalization_message", create_finalization_message)
    with app.test_request_context():
        assert unwrap(AgentBuildChatFinalizeApi.post)(
            AgentBuildChatFinalizeApi(), unbound_session, "tenant-1", _account(account_id=account_id), agent_id
        ) == {"result": "generated"}
    assert cast(dict[str, object], captured["resolve"]) == {"tenant_id": "tenant-1", "agent_id": agent_id}
    finalize_call = cast(dict[str, object], captured["finalize"])
    assert finalize_call["session"] is unbound_session
    assert finalize_call["app_model"] is app_model
    assert finalize_call["current_tenant_id"] == "tenant-1"
    assert finalize_call["agent_id"] == agent_id
    assert cast(Account, finalize_call["current_user"]).id == account_id


def test_build_chat_finalization_helper_forces_debug_build_and_push_prompt(
    app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str, unbound_session: Session
) -> None:
    app_model = _app_detail_obj(id="app-1", tenant_id="tenant-1", mode=AppMode.AGENT)
    captured: dict[str, object] = {}

    def resolve_debug_conversation(**kwargs: object) -> str:
        captured["resolve_debug_conversation"] = kwargs
        return "debug-conversation-1"

    def generate(**kwargs: object) -> object:
        captured["generate"] = kwargs
        return iter(
            ["event: ping\n\n", 'data: {"event":"message","answer":"working"}\n\n', 'data: {"event":"message_end"}\n\n']
        )

    monkeypatch.setattr(
        completion_controller, "_resolve_current_user_agent_debug_conversation_id", resolve_debug_conversation
    )
    monkeypatch.setattr(completion_controller.AppGenerateService, "generate", generate)
    with app.test_request_context(headers={"X-Trace-Id": "trace-1"}):
        result = completion_controller._create_build_chat_finalization_message(
            current_tenant_id="tenant-1",
            current_user=_account(account_id=account_id),
            app_model=app_model,
            agent_id="agent-1",
            session=unbound_session,
        )
    assert result == ({"result": "success"}, 200)
    resolve_call = cast(dict[str, object], captured["resolve_debug_conversation"])
    assert resolve_call.pop("current_user") is not None
    assert resolve_call == {
        "session": unbound_session,
        "current_tenant_id": "tenant-1",
        "app_model": app_model,
        "agent_id": "agent-1",
        "draft_type": AgentConfigDraftType.DEBUG_BUILD,
    }
    generate_call = cast(dict[str, object], captured["generate"])
    assert generate_call["app_model"] is app_model
    assert generate_call["streaming"] is True
    args = cast(dict[str, object], generate_call["args"])
    assert args["draft_type"] == "debug_build"
    assert args["response_mode"] == "streaming"
    assert args["conversation_id"] == "debug-conversation-1"
    assert args["inputs"] == {}
    assert args["auto_generate_name"] is False
    assert "_agent_runtime_exit_intent" not in args
    assert args["external_trace_id"] == "trace-1"


def test_drain_streaming_generate_response_returns_on_message_end() -> None:
    closed: list[bool] = []

    def generate_response() -> Generator[str]:
        try:
            yield "event: ping\n\n"
            yield 'data: {"event":"message","answer":"working"}\n\n'
            yield 'data: {"event":"message_end","message_id":"msg-1"}\n\n'
        finally:
            closed.append(True)

    # Retain the generator so garbage collection cannot satisfy the cleanup assertion.
    response = generate_response()
    assert completion_controller._drain_streaming_generate_response(response) is None
    assert closed == [True]


def test_drain_streaming_generate_response_maps_error_event() -> None:
    response = (chunk for chunk in ['data: {"event":"error","message":"backend failed"}\n\n'])
    with pytest.raises(CompletionRequestError, match="backend failed"):
        completion_controller._drain_streaming_generate_response(response)


def test_drain_streaming_generate_response_raises_when_stream_ends_early() -> None:
    response = (chunk for chunk in ['data: {"event":"message","answer":"working"}\n\n'])
    with pytest.raises(CompletionRequestError, match="did not complete"):
        completion_controller._drain_streaming_generate_response(response)


@pytest.mark.parametrize(
    ("payload_extra", "expected_draft_type", "expected_start_new"),
    [
        ({}, AgentConfigDraftType.DRAFT, True),
        ({"conversation_id": None}, AgentConfigDraftType.DRAFT, True),
        ({"conversation_id": ""}, AgentConfigDraftType.DRAFT, True),
        (
            {"conversation_id": "00000000-0000-0000-0000-000000000001"},
            AgentConfigDraftType.DRAFT,
            False,
        ),
        ({"draft_type": "debug_build"}, AgentConfigDraftType.DEBUG_BUILD, False),
        (
            {
                "draft_type": "debug_build",
                "conversation_id": "00000000-0000-0000-0000-000000000001",
            },
            AgentConfigDraftType.DEBUG_BUILD,
            False,
        ),
    ],
)
def test_agent_chat_helper_resolves_scoped_conversation_and_forces_streaming(
    app: Flask,
    monkeypatch: pytest.MonkeyPatch,
    account_id: str,
    payload_extra: dict[str, str | None],
    expected_draft_type: AgentConfigDraftType,
    expected_start_new: bool,
    unbound_session: Session,
) -> None:
    app_model = _app_detail_obj(id="app-1", tenant_id="tenant-1", mode=AppMode.AGENT)
    current_user = _account(account_id=account_id)
    captured: dict[str, object] = {}

    def generate(**kwargs: object) -> dict[str, object]:
        captured.update(kwargs)
        return {"answer": "ok"}

    def resolve_debug_conversation(**kwargs: object) -> str:
        captured["resolve_debug_conversation"] = kwargs
        return "00000000-0000-0000-0000-000000000001"

    monkeypatch.setattr(completion_controller.AppGenerateService, "generate", generate)
    monkeypatch.setattr(
        completion_controller,
        "_resolve_current_user_agent_debug_conversation_id",
        resolve_debug_conversation,
    )
    monkeypatch.setattr(
        completion_controller.helper, "compact_generate_response", lambda response: {"response": response}
    )
    with app.test_request_context(
        json={"inputs": {}, "query": "hello", "response_mode": "streaming", **payload_extra},
        headers={"X-Trace-Id": "trace-1"},
    ):
        result = completion_controller._create_chat_message(
            current_user=current_user, app_model=app_model, session=unbound_session
        )
    assert result == {"response": {"answer": "ok"}}
    assert captured["app_model"] is app_model
    assert captured["user"] is current_user
    assert captured["streaming"] is True
    args = cast(dict[str, object], captured["args"])
    assert args["response_mode"] == "streaming"
    assert args["conversation_id"] == "00000000-0000-0000-0000-000000000001"
    assert args["auto_generate_name"] is False
    assert args["external_trace_id"] == "trace-1"
    resolve_call = cast(dict[str, object], captured["resolve_debug_conversation"])
    assert resolve_call["draft_type"] == expected_draft_type
    assert resolve_call["start_new"] is expected_start_new


def test_agent_chat_helper_ignores_private_exit_intent_payload_key(
    app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str, unbound_session: Session
) -> None:
    app_model = _app_detail_obj(id="app-1", tenant_id="tenant-1", mode=AppMode.AGENT)
    current_user = _account(account_id=account_id)
    captured: dict[str, object] = {}

    def generate(**kwargs: object) -> dict[str, object]:
        captured.update(kwargs)
        return {"answer": "ok"}

    monkeypatch.setattr(completion_controller.AppGenerateService, "generate", generate)
    monkeypatch.setattr(
        completion_controller,
        "_resolve_current_user_agent_debug_conversation_id",
        lambda **kwargs: "debug-conversation-1",
    )
    monkeypatch.setattr(
        completion_controller.helper,
        "compact_generate_response",
        lambda response: {"response": response},
    )

    with app.test_request_context(
        json={
            "inputs": {},
            "query": "hello",
            "response_mode": "streaming",
            "_agent_runtime_exit_intent": "delete",
        }
    ):
        result = completion_controller._create_chat_message(
            current_user=current_user,
            app_model=app_model,
            session=unbound_session,
        )

    assert result == {"response": {"answer": "ok"}}
    assert captured["streaming"] is True
    args = cast(dict[str, object], captured["args"])
    assert args["response_mode"] == "streaming"
    assert args["conversation_id"] == "debug-conversation-1"
    assert "_agent_runtime_exit_intent" not in args


@pytest.mark.parametrize(
    ("payload_extra", "expected_draft_type"),
    [
        ({}, AgentConfigDraftType.DRAFT),
        ({"draft_type": "debug_build"}, AgentConfigDraftType.DEBUG_BUILD),
    ],
)
def test_agent_chat_helper_rejects_foreign_debug_conversation_before_generation(
    app: Flask,
    monkeypatch: pytest.MonkeyPatch,
    account_id: str,
    payload_extra: dict[str, str],
    expected_draft_type: AgentConfigDraftType,
    unbound_session: Session,
) -> None:
    app_model = _app_detail_obj(id="app-1", tenant_id="tenant-1", mode=AppMode.AGENT)
    generate = MagicMock()
    resolve_debug_conversation = MagicMock(return_value="owned-conversation")
    monkeypatch.setattr(completion_controller.AppGenerateService, "generate", generate)
    monkeypatch.setattr(
        completion_controller,
        "_resolve_current_user_agent_debug_conversation_id",
        resolve_debug_conversation,
    )
    with app.test_request_context(
        json={
            "inputs": {},
            "query": "hello",
            "response_mode": "streaming",
            "conversation_id": "00000000-0000-0000-0000-000000000001",
            **payload_extra,
        }
    ):
        with pytest.raises(NotFound):
            completion_controller._create_chat_message(
                current_tenant_id="tenant-1",
                current_user=_account(account_id=account_id),
                app_model=app_model,
                agent_id="agent-1",
                session=unbound_session,
            )

    resolve_debug_conversation.assert_called_once()
    resolve_call = resolve_debug_conversation.call_args.kwargs
    assert resolve_call["draft_type"] == expected_draft_type
    assert resolve_call["start_new"] is False
    generate.assert_not_called()


def test_resolve_current_user_agent_debug_conversation_uses_agent_or_backing_app(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    calls: list[dict[str, object]] = []

    class FakeRosterService:
        def __init__(self, session: object) -> None:
            calls.append({"session": session})

        def get_or_create_build_conversation(self, **kwargs: object) -> str:
            calls.append({"get_build": kwargs})
            return f"debug-{kwargs['agent_id']}"

        def rotate_preview_conversation(self, **kwargs: object) -> str:
            calls.append({"rotate_preview": kwargs})
            return f"new-{kwargs['agent_id']}"

        def get_current_preview_conversation(self, **kwargs: object) -> str:
            calls.append({"get_preview": kwargs})
            return f"preview-{kwargs['agent_id']}"

        def get_app_backing_agent(self, **kwargs: object) -> object:
            calls.append({"get_app_backing_agent": kwargs})
            return Agent(
                tenant_id="tenant-1",
                name="Agent",
                scope=AgentScope.ROSTER,
                source=AgentSource.AGENT_APP,
                id="backing-agent",
            )

    monkeypatch.setattr(completion_controller, "AgentRosterService", FakeRosterService)
    explicit_id = completion_controller._resolve_current_user_agent_debug_conversation_id(
        session="session-1",  # type: ignore[arg-type]
        current_tenant_id="tenant-1",
        current_user=_account(),
        app_model=_app_detail_obj(id="app-1"),
        agent_id="agent-1",
        draft_type=AgentConfigDraftType.DRAFT,
        start_new=True,
    )
    fallback_id = completion_controller._resolve_current_user_agent_debug_conversation_id(
        session="session-1",  # type: ignore[arg-type]
        current_tenant_id="tenant-1",
        current_user=_account(),
        app_model=_app_detail_obj(id="app-1"),
        agent_id=None,
        draft_type=AgentConfigDraftType.DEBUG_BUILD,
    )
    fallback_preview_id = completion_controller._resolve_current_user_agent_debug_conversation_id(
        session="session-1",  # type: ignore[arg-type]
        current_tenant_id="tenant-1",
        current_user=_account(),
        app_model=_app_detail_obj(id="app-1"),
        agent_id=None,
        draft_type=AgentConfigDraftType.DRAFT,
        start_new=True,
    )
    current_preview_id = completion_controller._resolve_current_user_agent_debug_conversation_id(
        session="session-1",  # type: ignore[arg-type]
        current_tenant_id="tenant-1",
        current_user=_account(),
        app_model=_app_detail_obj(id="app-1"),
        agent_id="agent-1",
        draft_type=AgentConfigDraftType.DRAFT,
    )
    assert explicit_id == "new-agent-1"
    assert fallback_id == "debug-backing-agent"
    assert fallback_preview_id == "new-backing-agent"
    assert current_preview_id == "preview-agent-1"
    assert calls[1] == {
        "rotate_preview": {
            "tenant_id": "tenant-1",
            "agent_id": "agent-1",
            "account_id": "account-1",
        }
    }
    assert calls[3] == {"get_app_backing_agent": {"tenant_id": "tenant-1", "app_id": "app-1"}}
    assert calls[4] == {
        "get_build": {
            "tenant_id": "tenant-1",
            "agent_id": "backing-agent",
            "account_id": "account-1",
        }
    }
    assert calls[6] == {"get_app_backing_agent": {"tenant_id": "tenant-1", "app_id": "app-1"}}
    assert calls[7] == {
        "rotate_preview": {
            "tenant_id": "tenant-1",
            "agent_id": "backing-agent",
            "account_id": "account-1",
        }
    }
    assert calls[9] == {
        "get_preview": {
            "tenant_id": "tenant-1",
            "agent_id": "agent-1",
            "account_id": "account-1",
        }
    }


@pytest.mark.parametrize(
    ("error", "expected"),
    [
        (completion_controller.services.errors.conversation.ConversationNotExistsError(), NotFound),
        (
            completion_controller.services.errors.conversation.ConversationCompletedError(),
            completion_controller.ConversationCompletedError,
        ),
        (
            completion_controller.services.errors.app_model_config.AppModelConfigBrokenError(),
            completion_controller.AppUnavailableError,
        ),
        (
            completion_controller.ProviderTokenNotInitError("not initialized"),
            completion_controller.ProviderNotInitializeError,
        ),
        (completion_controller.QuotaExceededError(), completion_controller.ProviderQuotaExceededError),
        (
            completion_controller.ModelCurrentlyNotSupportError(),
            completion_controller.ProviderModelCurrentlyNotSupportError,
        ),
        (completion_controller.InvokeRateLimitError("rate limited"), completion_controller.InvokeRateLimitHttpError),
        (completion_controller.InvokeError("invoke failed"), completion_controller.CompletionRequestError),
        (RuntimeError("unexpected"), InternalServerError),
    ],
)
def test_agent_chat_helper_maps_generation_errors(
    app: Flask,
    monkeypatch: pytest.MonkeyPatch,
    error: Exception,
    expected: type[Exception],
    unbound_session: Session,
) -> None:
    app_model = _app_detail_obj(id="app-1", mode=AppMode.CHAT)
    monkeypatch.setattr(completion_controller.AppGenerateService, "generate", lambda **_: (_ for _ in ()).throw(error))
    with app.test_request_context(json={"inputs": {}, "query": "hello"}):
        with pytest.raises(expected):
            completion_controller._create_chat_message(
                current_user=_account(), app_model=app_model, session=unbound_session
            )


def test_dify_tool_candidate_response_keeps_granularity_fields() -> None:
    """Both selection granularities must survive the fields-layer model —
    the frontend needs granularity/tools_count to render the Tools menu."""
    from fields.agent_fields import AgentComposerDifyToolCandidateResponse

    provider_entry = AgentComposerDifyToolCandidateResponse.model_validate(
        {
            "id": "duckduckgo/*",
            "granularity": "provider",
            "name": "DuckDuckGo",
            "provider": "duckduckgo",
            "plugin_id": "langgenius/duckduckgo",
            "tools_count": 2,
        }
    ).model_dump(exclude_none=True)
    assert provider_entry["granularity"] == "provider"
    assert provider_entry["tools_count"] == 2