# SPDX-FileCopyrightText: 2025 Weibo, Inc.
#
# SPDX-License-Identifier: Apache-2.0

"""
API integration tests for OpenAPI v1/responses endpoints.
"""

import asyncio
import json
from datetime import datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session

from app.api.endpoints.openapi_responses import (
    _create_non_streaming_response_unified,
    _filter_current_assistant_turn,
    _task_to_response_object,
)
from app.models.kind import Kind
from app.models.subtask import SenderType, Subtask, SubtaskRole, SubtaskStatus
from app.models.task import TaskResource
from app.models.user import User


@pytest.fixture
def test_team(test_db: Session, test_user: User) -> Kind:
    """Create a test Team Kind."""
    team_json = {
        "apiVersion": "agent.wecode.io/v1",
        "kind": "Team",
        "metadata": {"name": "test-team", "namespace": "default"},
        "spec": {
            "collaborationModel": "sequential",
            "members": [
                {
                    "botRef": {"name": "test-bot", "namespace": "default"},
                    "role": "worker",
                }
            ],
        },
    }
    team = Kind(
        user_id=test_user.id,
        kind="Team",
        name="test-team",
        namespace="default",
        json=team_json,
        is_active=True,
    )
    test_db.add(team)
    test_db.commit()
    test_db.refresh(team)
    return team


@pytest.fixture
def test_bot(test_db: Session, test_user: User) -> Kind:
    """Create a test Bot Kind."""
    bot_json = {
        "apiVersion": "agent.wecode.io/v1",
        "kind": "Bot",
        "metadata": {"name": "test-bot", "namespace": "default"},
        "spec": {
            "shellRef": {"name": "chat-shell", "namespace": "default"},
            "ghostRef": {"name": "test-ghost", "namespace": "default"},
            "modelRef": {"name": "gpt-4", "namespace": "default"},
        },
    }
    bot = Kind(
        user_id=test_user.id,
        kind="Bot",
        name="test-bot",
        namespace="default",
        json=bot_json,
        is_active=True,
    )
    test_db.add(bot)
    test_db.commit()
    test_db.refresh(bot)
    return bot


@pytest.fixture
def test_shell(test_db: Session, test_user: User) -> Kind:
    """Create a test Shell Kind (Chat Shell type)."""
    shell_json = {
        "apiVersion": "agent.wecode.io/v1",
        "kind": "Shell",
        "metadata": {"name": "chat-shell", "namespace": "default"},
        "spec": {"shellType": "Chat"},
    }
    shell = Kind(
        user_id=test_user.id,
        kind="Shell",
        name="chat-shell",
        namespace="default",
        json=shell_json,
        is_active=True,
    )
    test_db.add(shell)
    test_db.commit()
    test_db.refresh(shell)
    return shell


@pytest.fixture
def test_public_shell(test_db: Session) -> Kind:
    """Create a public Shell Kind (user_id=0)."""
    shell_json = {
        "apiVersion": "agent.wecode.io/v1",
        "kind": "Shell",
        "metadata": {"name": "chat-shell", "namespace": "default"},
        "spec": {"shellType": "Chat"},
    }
    shell = Kind(
        user_id=0,
        kind="Shell",
        name="chat-shell",
        namespace="default",
        json=shell_json,
        is_active=True,
    )
    test_db.add(shell)
    test_db.commit()
    test_db.refresh(shell)
    return shell


@pytest.fixture
def test_model(test_db: Session, test_user: User) -> Kind:
    """Create a test Model Kind."""
    model_json = {
        "apiVersion": "agent.wecode.io/v1",
        "kind": "Model",
        "metadata": {"name": "gpt-4", "namespace": "default"},
        "spec": {"provider": "openai", "modelName": "gpt-4"},
    }
    model = Kind(
        user_id=test_user.id,
        kind="Model",
        name="gpt-4",
        namespace="default",
        json=model_json,
        is_active=True,
    )
    test_db.add(model)
    test_db.commit()
    test_db.refresh(model)
    return model


@pytest.fixture
def test_task(test_db: Session, test_user: User, test_team: Kind) -> TaskResource:
    """Create a test Task Kind."""
    task_json = {
        "apiVersion": "agent.wecode.io/v1",
        "kind": "Task",
        "metadata": {
            "name": "task-test",
            "namespace": "default",
            "labels": {
                "type": "online",
                "taskType": "chat",
                "source": "chat_shell",
            },
        },
        "spec": {
            "title": "Test task",
            "prompt": "Hello",
            "teamRef": {"name": "test-team", "namespace": "default"},
            "workspaceRef": {"name": "workspace-1", "namespace": "default"},
        },
        "status": {
            "status": "COMPLETED",
            "progress": 100,
            "result": {"value": "Hello! How can I help you?"},
            "errorMessage": "",
            "createdAt": datetime.now().isoformat(),
            "updatedAt": datetime.now().isoformat(),
            "completedAt": datetime.now().isoformat(),
        },
    }
    # Don't set explicit id, let the database assign it
    task = TaskResource(
        user_id=test_user.id,
        kind="Task",
        name="task-test",
        namespace="default",
        json=task_json,
        is_active=True,
    )
    test_db.add(task)
    test_db.commit()
    test_db.refresh(task)
    return task


@pytest.fixture
def test_subtasks(
    test_db: Session, test_user: User, test_task: TaskResource, test_team: Kind
) -> list:
    """Create test subtasks for a task."""
    user_subtask = Subtask(
        user_id=test_user.id,
        task_id=test_task.id,
        team_id=test_team.id,
        title="User message",
        bot_ids=[1],
        role=SubtaskRole.USER,
        executor_namespace="",
        executor_name="",
        prompt="Hello",
        status=SubtaskStatus.COMPLETED,
        progress=100,
        message_id=1,
        parent_id=0,
        error_message="",
        completed_at=datetime.now(),
        result=None,
        sender_type=SenderType.USER,
        sender_user_id=test_user.id,
    )
    test_db.add(user_subtask)

    assistant_subtask = Subtask(
        user_id=test_user.id,
        task_id=test_task.id,
        team_id=test_team.id,
        title="Assistant response",
        bot_ids=[1],
        role=SubtaskRole.ASSISTANT,
        executor_namespace="",
        executor_name="",
        prompt="",
        status=SubtaskStatus.COMPLETED,
        progress=100,
        message_id=2,
        parent_id=1,
        error_message="",
        completed_at=datetime.now(),
        result={"value": "Hello! How can I help you?"},
        sender_type=SenderType.TEAM,
        sender_user_id=0,
    )
    test_db.add(assistant_subtask)
    test_db.commit()
    test_db.refresh(user_subtask)
    test_db.refresh(assistant_subtask)
    return [user_subtask, assistant_subtask]


def _assistant_subtask_with_pending_form(
    *,
    subtask_id: int,
    task_id: int,
    user_id: int,
    team_id: int,
    ask_id: str,
) -> Subtask:
    return Subtask(
        id=subtask_id,
        user_id=user_id,
        task_id=task_id,
        team_id=team_id,
        title="Assistant response",
        bot_ids=[1],
        role=SubtaskRole.ASSISTANT,
        executor_namespace="",
        executor_name="",
        prompt="",
        status=SubtaskStatus.COMPLETED,
        progress=100,
        message_id=2,
        parent_id=1,
        error_message="",
        completed_at=datetime.now(),
        result={
            "blocks": [
                {
                    "id": f"call_mcp_{subtask_id}",
                    "type": "tool",
                    "tool_use_id": f"call_mcp_{subtask_id}",
                    "tool_name": "interactive_form_question",
                    "tool_input": {"ask_id": ask_id},
                    "tool_output": {
                        "pending_user_input": True,
                        "pending_user_input_payload": {
                            "type": "interactive_form_question",
                            "ask_id": ask_id,
                            "submit_mode": "new_response",
                            "submit_format": "markdown_message",
                        },
                    },
                    "tool_protocol": "mcp_call",
                    "server_label": "interactive-form",
                    "status": "done",
                }
            ]
        },
        sender_type=SenderType.TEAM,
        sender_user_id=0,
    )


@pytest.mark.api
class TestOpenAPIResponsesCreate:
    """Test POST /api/v1/responses endpoint."""

    def test_create_response_invalid_model_format(
        self, test_client: TestClient, test_api_key
    ):
        """Test create response fails with invalid model format."""
        response = test_client.post(
            "/api/v1/responses",
            headers={"X-API-Key": test_api_key[0]},
            json={"model": "invalid-model-format", "input": "Hello"},
        )

        assert response.status_code == 400
        assert "Invalid model format" in response.json()["detail"]

    def test_create_response_team_not_found(
        self, test_client: TestClient, test_api_key
    ):
        """Test create response fails when team not found."""
        response = test_client.post(
            "/api/v1/responses",
            headers={"X-API-Key": test_api_key[0]},
            json={"model": "default#nonexistent-team", "input": "Hello"},
        )

        assert response.status_code == 404
        assert "not found" in response.json()["detail"]

    def test_create_response_invalid_previous_response_id_format(
        self, test_client: TestClient, test_api_key, test_team: Kind, test_bot: Kind
    ):
        """Document current behavior for previous_response_id without resp_ prefix."""
        from app.schemas.openapi_response import ResponseObject

        mock_response = ResponseObject(
            id="resp_123",
            created_at=int(datetime.now().timestamp()),
            status="completed",
            model="default#test-team",
            output=[],
        )

        with patch(
            "app.api.endpoints.openapi_responses._create_non_streaming_response_unified",
            return_value=mock_response,
        ) as mock_create_sync:
            response = test_client.post(
                "/api/v1/responses",
                headers={"X-API-Key": test_api_key[0]},
                json={
                    "model": "default#test-team",
                    "input": "Hello",
                    "previous_response_id": "not-a-resp-id",
                },
            )

        assert response.status_code == 200
        mock_create_sync.assert_called_once()

    def test_create_response_invalid_previous_response_id_number(
        self, test_client: TestClient, test_api_key, test_team: Kind, test_bot: Kind
    ):
        """Test create response fails with invalid previous_response_id number."""
        response = test_client.post(
            "/api/v1/responses",
            headers={"X-API-Key": test_api_key[0]},
            json={
                "model": "default#test-team",
                "input": "Hello",
                "previous_response_id": "resp_abc",
            },
        )

        assert response.status_code == 400
        assert "Invalid previous_response_id format" in response.json()["detail"]

    def test_create_response_previous_response_not_found(
        self, test_client: TestClient, test_api_key, test_team: Kind, test_bot: Kind
    ):
        """Test create response fails when previous response not found."""
        response = test_client.post(
            "/api/v1/responses",
            headers={"X-API-Key": test_api_key[0]},
            json={
                "model": "default#test-team",
                "input": "Hello",
                "previous_response_id": "resp_99999",
            },
        )

        assert response.status_code == 404
        assert "Previous response" in response.json()["detail"]

    def test_create_response_model_not_found(
        self, test_client: TestClient, test_api_key, test_team: Kind, test_bot: Kind
    ):
        """Test create response fails when specified model not found."""
        response = test_client.post(
            "/api/v1/responses",
            headers={"X-API-Key": test_api_key[0]},
            json={"model": "default#test-team#nonexistent-model", "input": "Hello"},
        )

        assert response.status_code == 400
        assert "not found" in response.json()["detail"]

    def test_create_response_without_auth(self, test_client: TestClient):
        """Test create response fails without authentication."""
        response = test_client.post(
            "/api/v1/responses",
            json={"model": "default#test-team", "input": "Hello"},
        )

        assert response.status_code == 401

    @patch("app.api.endpoints.openapi_responses._create_non_streaming_response_unified")
    def test_create_response_sync_success(
        self,
        mock_create_sync,
        test_client: TestClient,
        test_api_key,
        test_team: Kind,
        test_bot: Kind,
        test_model: Kind,
        test_public_shell: Kind,
    ):
        """Test successful synchronous response creation."""
        from app.schemas.openapi_response import (
            OutputMessage,
            OutputTextContent,
            ResponseObject,
        )

        mock_response = ResponseObject(
            id="resp_123",
            created_at=int(datetime.now().timestamp()),
            status="completed",
            model="default#test-team",
            output=[
                OutputMessage(
                    id="msg_1",
                    status="completed",
                    role="assistant",
                    content=[OutputTextContent(text="Hello! How can I help you?")],
                )
            ],
        )
        mock_create_sync.return_value = mock_response

        response = test_client.post(
            "/api/v1/responses",
            headers={"X-API-Key": test_api_key[0]},
            json={"model": "default#test-team", "input": "Hello", "stream": False},
        )

        assert response.status_code == 200
        data = response.json()
        assert data["id"] == "resp_123"
        assert data["status"] == "completed"
        assert len(data["output"]) == 1

    @pytest.mark.asyncio
    async def test_create_non_streaming_response_unified_sets_pending_user_input(
        self,
        test_db: Session,
        test_user: User,
        test_team: Kind,
    ):
        """Queued non-streaming responses should expose pending interactive form state."""
        setup = SimpleNamespace(
            task=SimpleNamespace(id=101, json={"metadata": {"labels": {}}}),
            task_id=101,
            user_subtask=SimpleNamespace(id=321),
            assistant_subtask=SimpleNamespace(id=654),
        )
        execution_request = SimpleNamespace(task_id=101, subtask_id=654)
        assistant_subtask = _assistant_subtask_with_pending_form(
            subtask_id=654,
            task_id=101,
            user_id=test_user.id,
            team_id=test_team.id,
            ask_id="ask_654",
        )
        query_db = MagicMock()
        query_db.query.return_value.filter.return_value.order_by.return_value.all.return_value = [
            assistant_subtask
        ]

        with (
            patch(
                "app.services.openapi.chat_session.setup_chat_session",
                return_value=setup,
            ),
            patch(
                "app.services.chat.trigger.unified.build_execution_request",
                new=AsyncMock(return_value=execution_request),
            ),
            patch(
                "app.services.execution.execution_dispatcher.supports_streaming",
                return_value=False,
            ),
            patch(
                "app.services.execution.execution_dispatcher.dispatch",
                new=AsyncMock(),
            ),
            patch(
                "app.db.session.SessionLocal",
                return_value=query_db,
            ),
        ):
            response = await _create_non_streaming_response_unified(
                db=test_db,
                user=test_user,
                team=test_team,
                model_info={"namespace": "default", "team_name": "test-team"},
                request_body=SimpleNamespace(
                    model="default#test-team",
                    input="hello",
                    previous_response_id=None,
                    stream=False,
                    tools=None,
                    background=False,
                    reasoning=None,
                    attachment_ids=None,
                    wegent_options=None,
                ),
                input_text="hello",
                tool_settings={},
            )

        assert response.status == "queued"
        assert response.pending_user_input is True
        assert response.pending_user_input_payload["ask_id"] == "ask_654"
        assert response.output[0].type == "mcp_call"
        assert response.output[0].output["pending_user_input"] is True

    def test_filter_current_assistant_turn_only_returns_active_subtask(
        self,
        test_subtasks: list,
    ):
        current_assistant = next(
            subtask
            for subtask in test_subtasks
            if subtask.role == SubtaskRole.ASSISTANT
        )
        previous_assistant = Subtask(
            user_id=current_assistant.user_id,
            task_id=current_assistant.task_id,
            team_id=current_assistant.team_id,
            title="Previous assistant",
            bot_ids=current_assistant.bot_ids,
            role=SubtaskRole.ASSISTANT,
            executor_namespace="",
            executor_name="",
            prompt="Earlier reply",
            status=SubtaskStatus.COMPLETED,
            progress=100,
            message_id=current_assistant.message_id - 1,
            parent_id=0,
            error_message="",
            completed_at=datetime.now(),
            result={"value": "Earlier reply"},
            sender_type=SenderType.TEAM,
            sender_user_id=0,
        )

        filtered = _filter_current_assistant_turn(
            [previous_assistant, current_assistant],
            current_assistant.id,
        )

        assert filtered == [current_assistant]

    @patch("app.api.endpoints.openapi_responses._create_streaming_response_unified")
    def test_create_response_streaming_not_supported_for_executor(
        self,
        mock_create_streaming,
        test_client: TestClient,
        test_api_key,
        test_team: Kind,
        test_bot: Kind,
        test_model: Kind,
        test_public_shell: Kind,
    ):
        """Test streaming returns error when not supported by shell type.

        Note: With the unified trigger architecture, streaming support is determined
        by ExecutionRouter based on shell_type. For non-Chat Shell types, the
        dispatch_sse_stream will raise NotImplementedError.
        """
        from fastapi import HTTPException

        # Mock the streaming function to raise NotImplementedError (simulating non-SSE mode)
        mock_create_streaming.side_effect = HTTPException(
            status_code=400,
            detail="Streaming is only supported for Chat Shell type teams",
        )

        response = test_client.post(
            "/api/v1/responses",
            headers={"X-API-Key": test_api_key[0]},
            json={"model": "default#test-team", "input": "Hello", "stream": True},
        )

        assert response.status_code == 400
        assert "Streaming is only supported" in response.json()["detail"]

    @pytest.mark.asyncio
    async def test_streaming_unsupported_returns_response_failed_event(
        self,
        test_db: Session,
        test_user: User,
        test_team: Kind,
    ):
        """Non-SSE shells use callback streaming via Redis. When Redis subscribe
        fails, the stream should return a response.failed event."""
        from app.api.endpoints.openapi_responses import (
            _create_streaming_response_unified,
        )
        from app.schemas.openapi_response import ResponseCreateInput

        setup = SimpleNamespace(
            task=SimpleNamespace(id=101, json={"metadata": {"labels": {}}}),
            task_id=101,
            user_subtask=SimpleNamespace(id=321),
            assistant_subtask=SimpleNamespace(id=654),
        )
        execution_request = SimpleNamespace(task_id=101, subtask_id=654)

        with (
            patch(
                "app.services.openapi.chat_session.setup_chat_session",
                return_value=setup,
            ),
            patch(
                "app.services.chat.trigger.unified.build_execution_request",
                new=AsyncMock(return_value=execution_request),
            ),
            patch(
                "app.services.execution.execution_dispatcher.supports_streaming",
                return_value=False,
            ),
            patch(
                "app.services.chat.storage.session_manager.register_stream",
                new=AsyncMock(return_value=asyncio.Event()),
            ),
            patch(
                "app.services.chat.storage.session_manager.subscribe_callback_channel",
                new=AsyncMock(return_value=(None, None)),
            ),
            patch(
                "app.services.chat.storage.session_manager.unregister_stream",
                new=AsyncMock(),
            ),
            patch(
                "app.services.chat.storage.session_manager.delete_streaming_content",
                new=AsyncMock(),
            ),
            patch(
                "app.api.endpoints.openapi_responses.collect_completed_result",
                new=AsyncMock(return_value={"value": ""}),
            ),
            patch(
                "app.api.endpoints.openapi_responses.persist_completed_result",
                new=AsyncMock(),
            ),
        ):
            response = await _create_streaming_response_unified(
                db=test_db,
                user=test_user,
                team=test_team,
                model_info={"namespace": "default", "team_name": "test-team"},
                request_body=ResponseCreateInput(
                    model="default#test-team",
                    input="hello",
                    stream=True,
                ),
                input_text="hello",
                tool_settings={},
            )
            body = []
            async for chunk in response.body_iterator:
                body.append(chunk.decode() if isinstance(chunk, bytes) else chunk)

        payload = "".join(body)
        events = []
        for chunk in payload.split("\n\n"):
            chunk = chunk.strip()
            if not chunk.startswith("data: "):
                continue
            events.append(json.loads(chunk.removeprefix("data: ").strip()))

        failed_event = next(
            event for event in events if event["type"] == "response.failed"
        )
        assert failed_event["response"]["status"] == "failed"
        assert failed_event["response"]["error"]["code"] == "stream_error"

    def test_create_response_with_wegent_tools(
        self, test_client: TestClient, test_api_key
    ):
        """Test create response with Wegent tools parameter."""
        response = test_client.post(
            "/api/v1/responses",
            headers={"X-API-Key": test_api_key[0]},
            json={
                "model": "default#nonexistent-team",
                "input": "Hello",
                "tools": [{"type": "wegent_deep_thinking"}],
            },
        )

        # Should fail at team lookup, but tools should be parsed
        assert response.status_code == 404

    def test_create_response_with_list_input(
        self, test_client: TestClient, test_api_key
    ):
        """Test create response with list input format."""
        response = test_client.post(
            "/api/v1/responses",
            headers={"X-API-Key": test_api_key[0]},
            json={
                "model": "default#nonexistent-team",
                "input": [{"role": "user", "content": "Hello"}],
            },
        )

        # Should fail at team lookup, but input parsing should work
        assert response.status_code == 404

    def test_create_response_with_wegent_source_header(
        self, test_client: TestClient, test_api_key
    ):
        """Test create response with wegent-source header."""
        response = test_client.post(
            "/api/v1/responses",
            headers={
                "X-API-Key": test_api_key[0],
                "wegent-source": "test-source",
            },
            json={"model": "default#nonexistent-team", "input": "Hello"},
        )

        # Should fail at team lookup, but header should be accepted
        assert response.status_code == 404

    @pytest.mark.asyncio
    async def test_non_streaming_build_request_passes_user_subtask_id(
        self,
        test_db: Session,
        test_user: User,
        test_team: Kind,
    ):
        """OpenAPI follow-up should propagate user_subtask_id like WebSocket chat."""
        from app.api.endpoints.openapi_responses import (
            _create_non_streaming_response_unified,
        )
        from app.schemas.openapi_response import ResponseCreateInput

        setup = SimpleNamespace(
            task=SimpleNamespace(id=101, json={"metadata": {"labels": {}}}),
            task_id=101,
            user_subtask=SimpleNamespace(id=321),
            assistant_subtask=SimpleNamespace(id=654),
        )
        execution_request = SimpleNamespace(task_id=101, subtask_id=654)
        query_db = MagicMock()
        query_db.query.return_value.filter.return_value.order_by.return_value.all.return_value = (
            []
        )

        with (
            patch(
                "app.services.openapi.chat_session.setup_chat_session",
                return_value=setup,
            ),
            patch(
                "app.services.chat.trigger.unified.build_execution_request",
                new=AsyncMock(return_value=execution_request),
            ) as mock_build_execution_request,
            patch(
                "app.services.execution.execution_dispatcher.supports_streaming",
                return_value=False,
            ),
            patch(
                "app.services.execution.execution_dispatcher.dispatch", new=AsyncMock()
            ),
            patch(
                "app.db.session.SessionLocal",
                return_value=query_db,
            ),
        ):
            response = await _create_non_streaming_response_unified(
                db=test_db,
                user=test_user,
                team=test_team,
                model_info={"namespace": "default", "team_name": "test-team"},
                request_body=ResponseCreateInput(
                    model="default#test-team",
                    input="follow-up question",
                ),
                input_text="follow-up question",
                tool_settings={},
            )

        assert response.status == "queued"
        query_db.close.assert_called()
        assert mock_build_execution_request.await_args.kwargs["user_subtask_id"] == 321
        assert mock_build_execution_request.await_args.kwargs["message"] == (
            "follow-up question"
        )

    @pytest.mark.asyncio
    async def test_non_streaming_dispatch_failure_persists_failed_status(
        self,
        test_db: Session,
        test_user: User,
        test_team: Kind,
    ):
        """Dispatch failures before terminal events should still persist FAILED."""
        from fastapi import HTTPException

        from app.api.endpoints.openapi_responses import (
            _create_non_streaming_response_unified,
        )
        from app.schemas.openapi_response import ResponseCreateInput

        setup = SimpleNamespace(
            task=SimpleNamespace(id=101, json={"metadata": {"labels": {}}}),
            task_id=101,
            user_subtask=SimpleNamespace(id=321),
            assistant_subtask=SimpleNamespace(id=654),
        )
        execution_request = SimpleNamespace(task_id=101, subtask_id=654)
        mock_emitter = MagicMock()
        mock_emitter.collect = AsyncMock(return_value=("", None))

        with (
            patch(
                "app.services.openapi.chat_session.setup_chat_session",
                return_value=setup,
            ),
            patch(
                "app.services.chat.trigger.unified.build_execution_request",
                new=AsyncMock(return_value=execution_request),
            ),
            patch(
                "app.services.execution.execution_dispatcher.supports_streaming",
                return_value=True,
            ),
            patch(
                "app.services.execution.execution_dispatcher.dispatch",
                new=AsyncMock(side_effect=RuntimeError("dispatcher boom")),
            ),
            patch(
                "app.services.execution.emitters.SSEResultEmitter",
                return_value=mock_emitter,
            ),
            patch(
                "app.api.endpoints.openapi_responses.collect_completed_result",
                new=AsyncMock(return_value={"value": ""}),
            ) as mock_collect_result,
            patch(
                "app.api.endpoints.openapi_responses.persist_completed_result",
                new=AsyncMock(),
            ) as mock_persist_result,
        ):
            with pytest.raises(HTTPException) as exc_info:
                await _create_non_streaming_response_unified(
                    db=test_db,
                    user=test_user,
                    team=test_team,
                    model_info={"namespace": "default", "team_name": "test-team"},
                    request_body=ResponseCreateInput(
                        model="default#test-team",
                        input="follow-up question",
                    ),
                    input_text="follow-up question",
                    tool_settings={},
                )

        assert exc_info.value.status_code == 500
        mock_collect_result.assert_awaited_once_with(
            654,
            status="FAILED",
            error_message="dispatcher boom",
            error_code=None,
        )
        mock_persist_result.assert_awaited_once_with(
            subtask_id=654,
            task_id=101,
            status="FAILED",
            result={"value": ""},
            error="dispatcher boom",
        )

    def test_setup_chat_session_reuses_previous_executor_for_follow_up(
        self,
        test_db: Session,
        test_user: User,
        test_team: Kind,
        test_bot: Kind,
        test_model: Kind,
        test_public_shell: Kind,
        test_task: TaskResource,
    ):
        """Follow-up assistant subtask should inherit previous executor metadata."""
        from app.services.openapi.chat_session import setup_chat_session

        previous_user = Subtask(
            user_id=test_user.id,
            task_id=test_task.id,
            team_id=test_team.id,
            title="Previous user message",
            bot_ids=[test_bot.id],
            role=SubtaskRole.USER,
            executor_namespace="",
            executor_name="",
            prompt="Draw a cat",
            status=SubtaskStatus.COMPLETED,
            progress=100,
            message_id=1,
            parent_id=0,
            error_message="",
            completed_at=datetime.now(),
            result=None,
            sender_type=SenderType.USER,
            sender_user_id=test_user.id,
        )
        previous_assistant = Subtask(
            user_id=test_user.id,
            task_id=test_task.id,
            team_id=test_team.id,
            title="Previous assistant response",
            bot_ids=[test_bot.id],
            role=SubtaskRole.ASSISTANT,
            executor_namespace="default",
            executor_name="executor-123",
            prompt="",
            status=SubtaskStatus.COMPLETED,
            progress=100,
            message_id=2,
            parent_id=1,
            error_message="",
            completed_at=datetime.now(),
            result={"value": "Here is the image"},
            sender_type=SenderType.TEAM,
            sender_user_id=0,
        )
        test_db.add(previous_user)
        test_db.add(previous_assistant)
        test_db.commit()

        setup = setup_chat_session(
            db=test_db,
            user=test_user,
            team=test_team,
            model_info={"namespace": "default", "team_name": "test-team"},
            input_text="Make it blue",
            tool_settings={},
            task_id=test_task.id,
        )

        assert setup.assistant_subtask.executor_name == "executor-123"
        assert setup.assistant_subtask.executor_namespace == "default"


@pytest.mark.api
class TestOpenAPIResponsesGet:
    """Test GET /api/v1/responses/{response_id} endpoint."""

    def test_get_response_invalid_format(self, test_client: TestClient, test_api_key):
        """Test get response fails with invalid response_id format."""
        response = test_client.get(
            "/api/v1/responses/invalid_id",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 400
        assert "Invalid response_id format" in response.json()["detail"]

    def test_get_response_invalid_number(self, test_client: TestClient, test_api_key):
        """Test get response fails with invalid response_id number."""
        response = test_client.get(
            "/api/v1/responses/resp_abc",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 400
        assert "Invalid response_id format" in response.json()["detail"]

    def test_get_response_not_found(self, test_client: TestClient, test_api_key):
        """Test get response fails when response not found."""
        response = test_client.get(
            "/api/v1/responses/resp_99999",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 404
        assert "not found" in response.json()["detail"]

    def test_get_response_without_auth(self, test_client: TestClient):
        """Test get response fails without authentication."""
        response = test_client.get("/api/v1/responses/resp_1")

        assert response.status_code == 401

    def test_get_response_success(
        self,
        test_client: TestClient,
        test_api_key,
        test_task: TaskResource,
        test_subtasks: list,
    ):
        """Test successful response retrieval."""
        response = test_client.get(
            f"/api/v1/responses/resp_{test_task.id}",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 200
        data = response.json()
        assert data["id"] == f"resp_{test_task.id}"
        assert data["status"] == "completed"
        assert len(data["output"]) == 1
        assert data["output"][0]["type"] == "message"
        assert data["output"][0]["role"] == "assistant"

    def test_task_to_response_object_sets_pending_user_input_from_subtasks(
        self,
        test_user: User,
        test_team: Kind,
    ):
        """GET /v1/responses/{id} should expose pending interactive form state."""
        task_dict = {
            "id": 42,
            "status": "COMPLETED",
            "created_at": datetime.now(),
        }
        assistant_subtask = _assistant_subtask_with_pending_form(
            subtask_id=204,
            task_id=42,
            user_id=test_user.id,
            team_id=test_team.id,
            ask_id="ask_204",
        )

        response = _task_to_response_object(
            task_dict,
            "default#test-team",
            subtasks=[assistant_subtask],
        )

        assert response.status == "completed"
        assert response.pending_user_input is True
        assert response.pending_user_input_payload["ask_id"] == "ask_204"
        assert response.output[0].type == "mcp_call"
        assert response.output[0].output["pending_user_input"] is True

    def test_get_response_exposes_pending_user_input(
        self,
        test_client: TestClient,
        test_api_key,
        test_db: Session,
        test_user: User,
        test_team: Kind,
        test_task: TaskResource,
    ):
        """GET endpoint should serialize pending interactive form state."""
        assistant_subtask = _assistant_subtask_with_pending_form(
            subtask_id=304,
            task_id=test_task.id,
            user_id=test_user.id,
            team_id=test_team.id,
            ask_id="ask_304",
        )
        assistant_subtask.message_id = 1
        test_db.add(assistant_subtask)
        test_db.commit()

        response = test_client.get(
            f"/api/v1/responses/resp_{test_task.id}",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 200
        data = response.json()
        assert data["pending_user_input"] is True
        assert data["pending_user_input_payload"]["ask_id"] == "ask_304"
        assert data["output"][0]["type"] == "mcp_call"
        assert data["output"][0]["output"]["pending_user_input"] is True

    def test_task_to_response_object_ignores_stale_pending_state_from_earlier_turn(
        self,
        test_user: User,
        test_team: Kind,
    ):
        """GET reconstruction should only inspect the latest assistant turn for pending state."""
        task_dict = {
            "id": 43,
            "status": "COMPLETED",
            "created_at": datetime.now(),
        }
        stale_assistant = _assistant_subtask_with_pending_form(
            subtask_id=205,
            task_id=43,
            user_id=test_user.id,
            team_id=test_team.id,
            ask_id="ask_205",
        )
        current_assistant = Subtask(
            user_id=test_user.id,
            task_id=43,
            team_id=test_team.id,
            title="Assistant response",
            bot_ids=[1],
            role=SubtaskRole.ASSISTANT,
            executor_namespace="",
            executor_name="",
            prompt="",
            status=SubtaskStatus.COMPLETED,
            progress=100,
            message_id=3,
            parent_id=1,
            error_message="",
            completed_at=datetime.now(),
            result={"value": "done"},
            sender_type=SenderType.TEAM,
            sender_user_id=0,
        )

        response = _task_to_response_object(
            task_dict,
            "default#test-team",
            subtasks=[stale_assistant, current_assistant],
        )

        assert response.status == "completed"
        assert response.pending_user_input is None
        assert response.pending_user_input_payload is None

    def test_get_response_ignores_stale_pending_state_from_earlier_turn(
        self,
        test_client: TestClient,
        test_api_key,
        test_db: Session,
        test_user: User,
        test_team: Kind,
        test_task: TaskResource,
    ):
        """GET endpoint should compute top-level pending state from the latest assistant turn only."""
        stale_assistant = _assistant_subtask_with_pending_form(
            subtask_id=305,
            task_id=test_task.id,
            user_id=test_user.id,
            team_id=test_team.id,
            ask_id="ask_305",
        )
        stale_assistant.message_id = 1
        current_assistant = Subtask(
            user_id=test_user.id,
            task_id=test_task.id,
            team_id=test_team.id,
            title="Assistant response",
            bot_ids=[1],
            role=SubtaskRole.ASSISTANT,
            executor_namespace="",
            executor_name="",
            prompt="",
            status=SubtaskStatus.COMPLETED,
            progress=100,
            message_id=2,
            parent_id=1,
            error_message="",
            completed_at=datetime.now(),
            result={"value": "done"},
            sender_type=SenderType.TEAM,
            sender_user_id=0,
        )
        test_db.add(stale_assistant)
        test_db.add(current_assistant)
        test_db.commit()

        response = test_client.get(
            f"/api/v1/responses/resp_{test_task.id}",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 200
        data = response.json()
        assert data["pending_user_input"] is None
        assert data["pending_user_input_payload"] is None

    def test_get_response_user_isolation(
        self,
        test_client: TestClient,
        test_admin_api_key,
        test_task: TaskResource,
    ):
        """Test users cannot access other users' responses."""
        response = test_client.get(
            f"/api/v1/responses/resp_{test_task.id}",
            headers={"X-API-Key": test_admin_api_key[0]},
        )

        assert response.status_code == 404


@pytest.mark.api
class TestOpenAPIResponsesCancel:
    """Test POST /api/v1/responses/{response_id}/cancel endpoint."""

    def test_cancel_response_invalid_format(
        self, test_client: TestClient, test_api_key
    ):
        """Test cancel response fails with invalid response_id format."""
        response = test_client.post(
            "/api/v1/responses/invalid_id/cancel",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 400
        assert "Invalid response_id format" in response.json()["detail"]

    def test_cancel_response_invalid_number(
        self, test_client: TestClient, test_api_key
    ):
        """Test cancel response fails with invalid response_id number."""
        response = test_client.post(
            "/api/v1/responses/resp_abc/cancel",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 400
        assert "Invalid response_id format" in response.json()["detail"]

    def test_cancel_response_not_found(self, test_client: TestClient, test_api_key):
        """Test cancel response fails when response not found."""
        response = test_client.post(
            "/api/v1/responses/resp_99999/cancel",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 404
        assert "not found" in response.json()["detail"]

    def test_cancel_response_without_auth(self, test_client: TestClient):
        """Test cancel response fails without authentication."""
        response = test_client.post("/api/v1/responses/resp_1/cancel")

        assert response.status_code == 401

    @patch("app.services.chat.storage.session_manager")
    def test_cancel_response_chat_shell_success(
        self,
        mock_session_manager,
        test_client: TestClient,
        test_api_key,
        test_db: Session,
        test_user: User,
        test_team: Kind,
    ):
        """Test successful cancel for Chat Shell task."""
        # Create a running task
        task_json = {
            "apiVersion": "agent.wecode.io/v1",
            "kind": "Task",
            "metadata": {
                "name": "task-running",
                "namespace": "default",
                "labels": {"source": "chat_shell"},
            },
            "spec": {
                "title": "Running task",
                "prompt": "Hello",
                "teamRef": {"name": "test-team", "namespace": "default"},
                "workspaceRef": {"name": "workspace-1", "namespace": "default"},
            },
            "status": {
                "status": "RUNNING",
                "progress": 50,
                "result": None,
                "errorMessage": "",
                "createdAt": datetime.now().isoformat(),
                "updatedAt": datetime.now().isoformat(),
            },
        }
        task = TaskResource(
            user_id=test_user.id,
            kind="Task",
            name="task-running",
            namespace="default",
            json=task_json,
            is_active=True,
        )
        test_db.add(task)
        test_db.commit()
        test_db.refresh(task)

        # Create running subtask
        running_subtask = Subtask(
            user_id=test_user.id,
            task_id=task.id,
            team_id=test_team.id,
            title="Assistant response",
            bot_ids=[1],
            role=SubtaskRole.ASSISTANT,
            executor_namespace="",
            executor_name="",
            prompt="",
            status=SubtaskStatus.RUNNING,
            progress=50,
            message_id=1,
            parent_id=0,
            error_message="",
            completed_at=datetime(1970, 1, 1, 0, 0, 0),
            result=None,
            sender_type=SenderType.TEAM,
            sender_user_id=0,
        )
        test_db.add(running_subtask)
        test_db.commit()
        test_db.refresh(running_subtask)

        # Mock session_manager methods
        mock_session_manager.get_streaming_content = AsyncMock(
            return_value="Partial content"
        )
        mock_session_manager.cancel_stream = AsyncMock()

        response = test_client.post(
            f"/api/v1/responses/resp_{task.id}/cancel",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 200
        data = response.json()
        assert data["id"] == f"resp_{task.id}"


@pytest.mark.api
class TestOpenAPIResponsesDelete:
    """Test DELETE /api/v1/responses/{response_id} endpoint."""

    def test_delete_response_invalid_format(
        self, test_client: TestClient, test_api_key
    ):
        """Test delete response fails with invalid response_id format."""
        response = test_client.delete(
            "/api/v1/responses/invalid_id",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 400
        assert "Invalid response_id format" in response.json()["detail"]

    def test_delete_response_invalid_number(
        self, test_client: TestClient, test_api_key
    ):
        """Test delete response fails with invalid response_id number."""
        response = test_client.delete(
            "/api/v1/responses/resp_abc",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 400
        assert "Invalid response_id format" in response.json()["detail"]

    def test_delete_response_not_found(self, test_client: TestClient, test_api_key):
        """Test delete response fails when response not found."""
        response = test_client.delete(
            "/api/v1/responses/resp_99999",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 404
        assert "not found" in response.json()["detail"]

    def test_delete_response_without_auth(self, test_client: TestClient):
        """Test delete response fails without authentication."""
        response = test_client.delete("/api/v1/responses/resp_1")

        assert response.status_code == 401

    def test_delete_response_success(
        self,
        test_client: TestClient,
        test_api_key,
        test_db: Session,
        test_user: User,
    ):
        """Test successful response deletion."""
        # Create a task to delete
        task_json = {
            "apiVersion": "agent.wecode.io/v1",
            "kind": "Task",
            "metadata": {
                "name": "task-to-delete",
                "namespace": "default",
                "labels": {"source": "api"},
            },
            "spec": {
                "title": "Task to delete",
                "prompt": "Hello",
                "teamRef": {"name": "test-team", "namespace": "default"},
                "workspaceRef": {"name": "workspace-1", "namespace": "default"},
            },
            "status": {
                "status": "COMPLETED",
                "progress": 100,
                "result": {"value": "Done"},
                "errorMessage": "",
                "createdAt": datetime.now().isoformat(),
                "updatedAt": datetime.now().isoformat(),
                "completedAt": datetime.now().isoformat(),
            },
        }
        task = TaskResource(
            user_id=test_user.id,
            kind="Task",
            name="task-to-delete",
            namespace="default",
            json=task_json,
            is_active=True,
        )
        test_db.add(task)
        test_db.commit()
        test_db.refresh(task)

        task_id = task.id

        response = test_client.delete(
            f"/api/v1/responses/resp_{task_id}",
            headers={"X-API-Key": test_api_key[0]},
        )

        assert response.status_code == 200
        data = response.json()
        assert data["id"] == f"resp_{task_id}"
        assert data["deleted"] is True

        # Verify task is deleted (should return 404)
        get_response = test_client.get(
            f"/api/v1/responses/resp_{task_id}",
            headers={"X-API-Key": test_api_key[0]},
        )
        assert get_response.status_code == 404

    def test_delete_response_user_isolation(
        self,
        test_client: TestClient,
        test_admin_api_key,
        test_task: TaskResource,
    ):
        """Test user isolation on delete.

        Note: Current implementation of delete_task in task_kinds_service
        does NOT enforce user isolation - it only checks if the task exists.
        This test documents the current behavior. Consider adding user_id
        filtering to delete_task for proper isolation.
        """
        # Current behavior: any authenticated user can delete any task
        # This test documents actual behavior, not ideal behavior
        response = test_client.delete(
            f"/api/v1/responses/resp_{test_task.id}",
            headers={"X-API-Key": test_admin_api_key[0]},
        )

        # Task can be deleted by any user (current implementation)
        # Ideally this should return 404 for non-owner
        assert response.status_code == 200


@pytest.mark.api
class TestOpenAPIResponsesHelpers:
    """Test helper functions used in openapi_responses."""

    def test_parse_model_string_two_parts(self):
        """Test parsing model string with namespace and team name."""
        from app.services.openapi.helpers import parse_model_string

        result = parse_model_string("default#my-team")
        assert result["namespace"] == "default"
        assert result["team_name"] == "my-team"
        assert result["model_id"] is None

    def test_parse_model_string_three_parts(self):
        """Test parsing model string with namespace, team name, and model id."""
        from app.services.openapi.helpers import parse_model_string

        result = parse_model_string("default#my-team#gpt-4")
        assert result["namespace"] == "default"
        assert result["team_name"] == "my-team"
        assert result["model_id"] == "gpt-4"

    def test_parse_model_string_invalid(self):
        """Test parsing invalid model string raises exception."""
        from app.services.openapi.helpers import parse_model_string

        with pytest.raises(Exception):
            parse_model_string("invalid-format")

    def test_extract_input_text_string(self):
        """Test extracting input text from string."""
        from app.services.openapi.helpers import extract_input_text

        result = extract_input_text("Hello, world!")
        assert result == "Hello, world!"

    def test_extract_input_text_list(self):
        """Test extracting input text from list."""
        from app.schemas.openapi_response import InputItem
        from app.services.openapi.helpers import extract_input_text

        input_list = [
            InputItem(role="user", content="First message"),
            InputItem(role="assistant", content="Response"),
            InputItem(role="user", content="Second message"),
        ]
        result = extract_input_text(input_list)
        assert result == "Second message"

    def test_parse_wegent_tools_empty(self):
        """Test parsing empty tools list."""
        from app.services.openapi.helpers import parse_wegent_tools

        result = parse_wegent_tools(None)
        assert result["enable_chat_bot"] is False

    def test_parse_wegent_tools_chat_bot(self):
        """Test parsing tools with chat bot enabled."""
        from app.schemas.openapi_response import WegentTool
        from app.services.openapi.helpers import parse_wegent_tools

        tools = [WegentTool(type="wegent_chat_bot")]
        result = parse_wegent_tools(tools)
        assert result["enable_chat_bot"] is True

    def test_wegent_status_to_openai_status(self):
        """Test status conversion from Wegent to OpenAI format."""
        from app.services.openapi.helpers import wegent_status_to_openai_status

        assert wegent_status_to_openai_status("PENDING") == "queued"
        assert wegent_status_to_openai_status("RUNNING") == "in_progress"
        assert wegent_status_to_openai_status("COMPLETED") == "completed"
        assert wegent_status_to_openai_status("FAILED") == "failed"
        assert wegent_status_to_openai_status("CANCELLED") == "cancelled"
        assert wegent_status_to_openai_status("UNKNOWN") == "incomplete"

    def test_subtask_status_to_message_status(self):
        """Test subtask status to message status conversion."""
        from app.services.openapi.helpers import subtask_status_to_message_status

        assert subtask_status_to_message_status("PENDING") == "in_progress"
        assert subtask_status_to_message_status("RUNNING") == "in_progress"
        assert subtask_status_to_message_status("COMPLETED") == "completed"
        assert subtask_status_to_message_status("FAILED") == "incomplete"
