# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.Agent.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

import json
import os
import tempfile
import typing
from datetime import datetime
from unittest import mock
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from cxas_scrapi.migration.data_models import (
    DFCXAgentIR,
    IRAgent,
    IRBundle,
    IRMetadata,
    MigrationConfig,
    MigrationIR,
)
from cxas_scrapi.migration.service import MigrationService

# ---------------------------------------------------------------------------
# Shared fixtures for stage-method tests
# ---------------------------------------------------------------------------


def _make_ir(with_app: bool = True) -> MigrationIR:
    """Build a minimal MigrationIR with one agent."""
    return MigrationIR(
        metadata=IRMetadata(
            app_name="test-app",
            app_id="11111111-1111-1111-1111-111111111111",
            app_resource_name=(
                "projects/p/locations/us/apps/X" if with_app else None
            ),
        ),
        agents={
            "RootAgent": IRAgent(
                type="PLAYBOOK",
                display_name="Root Agent",
                instruction="<root/>",
                resource_name=(
                    "projects/p/locations/us/apps/X/agents/A"
                    if with_app
                    else None
                ),
            )
        },
    )


def _make_source_data() -> DFCXAgentIR:
    return DFCXAgentIR(
        name="projects/p/locations/us/agents/src",
        display_name="Test Source",
        default_language_code="en",
        playbooks=[
            {
                "name": "projects/p/locations/us/agents/src/playbooks/p1",
                "displayName": "Root Agent",
                "playbookType": "ROUTINE",
            }
        ],
        flows=[],
    )


def _make_bundle(with_app: bool = True) -> IRBundle:
    return IRBundle(
        config=MigrationConfig(
            project_id="test-project",
            target_name="test_target",
            model="gemini-2.5-flash-001",
            # Tests that don't explicitly cover the web gate opt out so
            # run_stage_1 doesn't spin up an HTTP server and block.
            web_confirm_grouping=False,
        ),
        source_agent_data=_make_source_data(),
        ir=_make_ir(with_app=with_app),
        app_url=(
            "https://ces.cloud.google.com/projects/p/locations/us/apps/X"
            if with_app
            else None
        ),
    )


def _make_service(ir: MigrationIR | None = None) -> MigrationService:
    """Build a MigrationService with heavy dependencies mocked out."""
    service = MigrationService(
        project_id="test-project",
        ps_apps_client=MagicMock(),
        ps_agents_client=MagicMock(),
        ps_tools_client=MagicMock(),
        ps_toolsets_client=MagicMock(),
        secret_manager_client=MagicMock(),
        cx_api_client=MagicMock(),
    )
    service.ir = ir if ir is not None else _make_ir()
    service.source_agent_data = _make_source_data()
    service._deploy_base_resources = AsyncMock()
    service._deploy_pending_agents = AsyncMock()
    service.topology_linker = MagicMock()
    return service


# ---------------------------------------------------------------------------
# run_migration end-to-end (pre-existing test, kept as-is)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_run_migration_success() -> None:
    # Mock external clients
    mock_ps_apps = MagicMock()
    mock_ps_agents = MagicMock()
    mock_ps_tools = MagicMock()
    mock_ps_toolsets = MagicMock()
    mock_secret_manager = MagicMock()
    mock_cx_api = MagicMock()

    service = MigrationService(
        project_id="test-project",
        ps_apps_client=mock_ps_apps,
        ps_agents_client=mock_ps_agents,
        ps_tools_client=mock_ps_tools,
        ps_toolsets_client=mock_ps_toolsets,
        secret_manager_client=mock_secret_manager,
        cx_api_client=mock_cx_api,
    )

    # Mock internal components
    service.exporter = MagicMock()
    service.exporter.fetch_full_agent_details.return_value = DFCXAgentIR(
        name="projects/p/locations/l/agents/a",
        display_name="Test Agent",
        default_language_code="en",
        playbooks=[],
        flows=[],
    )

    service.ai_augment = MagicMock()
    service.ai_augment.generate_agent_description = AsyncMock(
        return_value="Desc"
    )

    # Mock deploy methods
    service._deploy_base_resources = AsyncMock()
    service._deploy_pending_agents = AsyncMock()

    # Mock flow processing
    service._process_single_flow = AsyncMock()

    # Mock topology linker
    service.topology_linker = MagicMock()

    # Mock reporter to avoid creating report file during test
    service.reporter = MagicMock()

    with patch(
        "cxas_scrapi.migration.service.DFCXParameterExtractor.migrate_parameters"
    ) as mock_migrate:
        mock_migrate.return_value = ([], {})
        config = MigrationConfig(
            project_id="dummy-project",
            target_name="cxas-app",
            model="gemini-2.5-flash-001",
        )
        await service.run_migration(
            source_cx_agent_id="dfcx-123", config=config
        )

    # Verify sequence
    service.exporter.fetch_full_agent_details.assert_called_once_with(
        "dfcx-123", use_export=True
    )
    mock_migrate.assert_called_once()
    # Default config consolidates: base resources are pushed (Phase 1 fast
    # deploy + deferred update pass) but agents are NOT deployed here — the
    # agent push is deferred to Stage 1 to respect the CXAS 100-agent cap.
    assert service._deploy_base_resources.call_count == 2
    service._deploy_pending_agents.assert_not_called()
    service.topology_linker.link_and_finalize_topology.assert_called_once()


@pytest.mark.asyncio
async def test_run_migration_no_consolidate_pushes_agents() -> None:
    """With no_consolidate=True, run_migration pushes the full 1:1 agent set
    immediately (legacy behavior)."""
    service = MigrationService(
        project_id="test-project",
        ps_apps_client=MagicMock(),
        ps_agents_client=MagicMock(),
        ps_tools_client=MagicMock(),
        ps_toolsets_client=MagicMock(),
        secret_manager_client=MagicMock(),
        cx_api_client=MagicMock(),
    )
    service.exporter = MagicMock()
    service.exporter.fetch_full_agent_details.return_value = DFCXAgentIR(
        name="projects/p/locations/l/agents/a",
        display_name="Test Agent",
        default_language_code="en",
        playbooks=[],
        flows=[],
    )
    service.ai_augment = MagicMock()
    service.ai_augment.generate_agent_description = AsyncMock(
        return_value="Desc"
    )
    service._deploy_base_resources = AsyncMock()
    service._deploy_pending_agents = AsyncMock()
    service._process_single_flow = AsyncMock()
    service.topology_linker = MagicMock()
    service.reporter = MagicMock()

    with (
        patch(
            "cxas_scrapi.migration.service.DFCXParameterExtractor."
            "migrate_parameters"
        ) as mock_migrate,
        patch("cxas_scrapi.migration.service.Versions"),
    ):
        mock_migrate.return_value = ([], {})
        config = MigrationConfig(
            project_id="dummy-project",
            target_name="cxas-app",
            model="gemini-2.5-flash-001",
            no_consolidate=True,
        )
        await service.run_migration(
            source_cx_agent_id="dfcx-123", config=config
        )

    service._deploy_pending_agents.assert_called_once()


@pytest.mark.asyncio
async def test_run_migration_early_utterance_harvesting() -> None:
    mock_ps_apps = MagicMock()
    mock_ps_agents = MagicMock()
    mock_ps_tools = MagicMock()
    mock_ps_toolsets = MagicMock()
    mock_secret_manager = MagicMock()
    mock_cx_api = MagicMock()

    service = MigrationService(
        project_id="test-project",
        ps_apps_client=mock_ps_apps,
        ps_agents_client=mock_ps_agents,
        ps_tools_client=mock_ps_tools,
        ps_toolsets_client=mock_ps_toolsets,
        secret_manager_client=mock_secret_manager,
        cx_api_client=mock_cx_api,
    )

    service.exporter = MagicMock()
    service.exporter.fetch_full_agent_details.return_value = DFCXAgentIR(
        name="projects/p/locations/l/agents/a",
        display_name="Test Agent",
        default_language_code="en",
        playbooks=[],
        flows=[],
    )

    service.ai_augment = MagicMock()
    service.ai_augment.generate_agent_description = AsyncMock(
        return_value="Desc"
    )

    service._deploy_base_resources = AsyncMock()
    service._deploy_pending_agents = AsyncMock()
    service._process_single_flow = AsyncMock()
    service.topology_linker = MagicMock()
    service.reporter = MagicMock()

    service.utterance_collector = MagicMock()
    service.utterance_collector.harvest_all.return_value = ["hello"]
    service.utterance_collector.classify_and_deduplicate = AsyncMock(
        return_value={
            "categorized_utterances": {"greeting_onboarding": ["hello"]},
            "rationales": {"hello": "Matched hello"},
        }
    )
    service.cuj_generator = MagicMock()
    service.cuj_generator.predict_cujs = AsyncMock(
        return_value=[
            {
                "graph_id": "core_sample_flow",
                "category": "Core CUJ",
                "description": "Sample path description",
                "nodes": [
                    {
                        "node_id": "node_001",
                        "type": "AGENT",
                        "speaker": "agent",
                        "mode": "VERBATIM",
                        "utterance": "hello",
                        "transitions": [],
                    }
                ],
            }
        ]
    )

    with patch(
        "cxas_scrapi.migration.service.DFCXParameterExtractor.migrate_parameters"
    ) as mock_migrate:
        mock_migrate.return_value = ([], {})
        config = MigrationConfig(
            project_id="dummy-project",
            target_name="cxas-app",
            model="gemini-2.5-flash-001",
            experimental_agent_xprs=True,
        )
        await service.run_migration(
            source_cx_agent_id="dfcx-123", config=config
        )

    service.utterance_collector.harvest_all.assert_called_once_with(
        service.source_agent_data
    )
    service.utterance_collector.classify_and_deduplicate.assert_awaited_once_with(
        ["hello"]
    )
    service.cuj_generator.predict_cujs.assert_awaited_once()
    assert service.ir.xprs_designer_data == {
        "raw": ["hello"],
        "raw_metadata": service.utterance_collector.raw_metadata,
        "categorized": {"greeting_onboarding": ["hello"]},
        "rationales": {"hello": "Matched hello"},
        "scenarios": [
            {
                "graph_id": "core_sample_flow",
                "category": "Core CUJ",
                "description": "Sample path description",
                "nodes": [
                    {
                        "node_id": "node_001",
                        "type": "AGENT",
                        "speaker": "agent",
                        "mode": "VERBATIM",
                        "utterance": "hello",
                        "transitions": [],
                    }
                ],
            }
        ],
    }


# ---------------------------------------------------------------------------

# persist_bundle
# ---------------------------------------------------------------------------


def test_persist_bundle_writes_file_and_appends_history() -> None:
    service = _make_service()
    bundle = _make_bundle()
    with tempfile.TemporaryDirectory() as td:
        path = os.path.join(td, "bundle.json")
        returned = service.persist_bundle(
            bundle, path, phase="stage_1", status="ok", notes="dedup done"
        )

    assert returned == path
    assert bundle.ir is service.ir
    assert len(bundle.stage_history) == 1
    entry = bundle.stage_history[0]
    assert entry.phase == "stage_1"
    assert entry.status == "ok"
    assert entry.notes == "dedup done"
    assert isinstance(entry.started_at, datetime)


def test_persist_bundle_without_phase_skips_history() -> None:
    service = _make_service()
    bundle = _make_bundle()
    with tempfile.TemporaryDirectory() as td:
        path = os.path.join(td, "bundle.json")
        service.persist_bundle(bundle, path)
    assert bundle.stage_history == []


# ---------------------------------------------------------------------------
# run_stage_1
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_run_stage_1_requires_bundle() -> None:
    service = _make_service()
    with pytest.raises(ValueError, match="requires bundle"):
        await service.run_stage_1(bundle=None)


@pytest.mark.asyncio
async def test_run_stage_1_creates_single_version_when_labels_set() -> None:
    service = _make_service()
    fake_versions_client = MagicMock()
    bundle = _make_bundle()
    fake_groupings = {
        "RootGroup": {
            "agents": ["Root Agent"],
            "journey": "main",
            "is_root": True,
        }
    }
    fake_consolidator = MagicMock()
    fake_consolidator.propose_groupings = AsyncMock(return_value=fake_groupings)
    fake_consolidator.consolidate = MagicMock(return_value=_make_ir())
    fake_consolidator.synthesize_instructions = AsyncMock(
        return_value={"RootGroup": "ok"}
    )

    with (
        patch(
            "cxas_scrapi.migration.stage_runner.run_stage_with_redeploy",
            new=AsyncMock(return_value=MagicMock(optimization_logs=[])),
        ),
        patch(
            "cxas_scrapi.migration.service.StructuralConsolidator",
            return_value=fake_consolidator,
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "detect_root_key",
            return_value="RootAgent",
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "validate_groupings"
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "persist_grouping"
        ),
        patch(
            "cxas_scrapi.migration.service.integrity_checks."
            "check_consolidation_integrity",
            return_value=([], []),
        ),
        patch(
            "cxas_scrapi.migration.service.topology_wirer.set_app_root_agent",
            return_value=(True, "set"),
        ),
        patch(
            "cxas_scrapi.migration.service.topology_wirer.delete_orphan_agents",
            return_value=(0, 0),
        ),
        patch(
            "cxas_scrapi.migration.service.Versions",
            return_value=fake_versions_client,
        ),
    ):
        await service.run_stage_1(
            bundle=bundle,
            version_label="0.0.3",
            dedup_version_label="0.0.2",
        )

    # Single-versioning: create_version is called once for consolidation!
    assert fake_versions_client.create_version.call_count == 1
    calls = fake_versions_client.create_version.call_args_list
    assert calls[0].kwargs["display_name"] == "0.0.3"
    assert "consolidation" in calls[0].kwargs["description"]


@pytest.mark.asyncio
async def test_run_stage_1_consolidate_runs_consolidator_persists_grouping() -> (  # noqa: E501
    None
):
    """End-to-end consolidation path with mocked consolidator + grouping
    callback returning the proposed groupings unchanged."""
    service = _make_service()
    bundle = _make_bundle()
    fake_groupings = {
        "RootGroup": {
            "agents": ["Root Agent"],
            "journey": "main",
            "is_root": True,
        }
    }

    consolidated_ir = _make_ir()

    fake_consolidator = MagicMock()
    fake_consolidator.propose_groupings = AsyncMock(return_value=fake_groupings)
    fake_consolidator.consolidate = MagicMock(return_value=consolidated_ir)
    fake_consolidator.synthesize_instructions = AsyncMock(
        return_value={"RootGroup": "ok"}
    )

    with (
        patch(
            "cxas_scrapi.migration.stage_runner.run_stage_with_redeploy",
            new=AsyncMock(return_value=MagicMock(optimization_logs=[])),
        ),
        patch(
            "cxas_scrapi.migration.service.StructuralConsolidator",
            return_value=fake_consolidator,
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "detect_root_key",
            return_value="RootAgent",
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "validate_groupings"
        ) as mock_validate,
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "persist_grouping"
        ) as mock_persist_grouping,
        patch(
            "cxas_scrapi.migration.service.integrity_checks."
            "check_consolidation_integrity",
            return_value=([], []),
        ),
        patch(
            "cxas_scrapi.migration.service.topology_wirer.set_app_root_agent",
            return_value=(True, "set"),
        ),
        patch(
            "cxas_scrapi.migration.service.topology_wirer.delete_orphan_agents",
            return_value=(0, 0),
        ),
        patch("cxas_scrapi.migration.service.Versions"),
    ):
        returned = await service.run_stage_1(
            bundle=bundle,
            version_label=None,
        )

    # persist_grouping was patched, so no file was written to disk.
    mock_persist_grouping.assert_called_once()

    assert returned == fake_groupings
    fake_consolidator.propose_groupings.assert_awaited_once()
    mock_validate.assert_called_once()
    fake_consolidator.consolidate.assert_called_once_with(fake_groupings)
    fake_consolidator.synthesize_instructions.assert_awaited_once()
    # Bundle mutated.
    assert bundle.grouping == fake_groupings
    # The pre-consolidation snapshot is used transiently for the integrity
    # check only and must NOT be persisted on the bundle — the raw 1:1
    # ("orphan") agents must never surface downstream.
    assert bundle.pre_consolidation_ir is None
    # Service IR replaced with consolidated output.
    assert service.ir is consolidated_ir
    # Update-pass deploys were called.
    service._deploy_base_resources.assert_awaited_once_with(is_update_pass=True)
    service._deploy_pending_agents.assert_awaited_once_with(is_update_pass=True)


@pytest.mark.asyncio
async def test_run_stage_1_consolidate_aborts_on_integrity_blocking() -> None:
    """When integrity_checks returns blocking errors,
    run_stage_1 raises RuntimeError.
    """
    service = _make_service()
    bundle = _make_bundle()
    fake_groupings = {"RootGroup": {"agents": ["Root Agent"], "is_root": True}}

    fake_consolidator = MagicMock()
    fake_consolidator.propose_groupings = AsyncMock(return_value=fake_groupings)
    fake_consolidator.consolidate = MagicMock(return_value=_make_ir())
    fake_consolidator.synthesize_instructions = AsyncMock(return_value={})

    with (
        patch(
            "cxas_scrapi.migration.stage_runner.run_stage_with_redeploy",
            new=AsyncMock(return_value=MagicMock(optimization_logs=[])),
        ),
        patch(
            "cxas_scrapi.migration.service.StructuralConsolidator",
            return_value=fake_consolidator,
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "detect_root_key",
            return_value="RootAgent",
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "validate_groupings"
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "persist_grouping"
        ),
        patch(
            "cxas_scrapi.migration.service.integrity_checks."
            "check_consolidation_integrity",
            return_value=(["BLOCKING: unknown tool foo"], []),
        ),
        pytest.raises(RuntimeError, match="blocking"),
    ):
        await service.run_stage_1(
            bundle=bundle,
            version_label=None,
        )

    # Deploy should NOT have run when we abort.
    service._deploy_base_resources.assert_not_awaited()


@pytest.mark.asyncio
async def test_run_stage_1_callback_returning_none_skips_consolidation() -> (
    None
):
    """When the grouping_callback returns None, the consolidation block
    is skipped.
    """
    service = _make_service()
    bundle = _make_bundle()

    fake_consolidator = MagicMock()
    fake_consolidator.propose_groupings = AsyncMock(
        return_value={"G": {"agents": ["Root Agent"], "is_root": True}}
    )

    # Callback receives kwargs (ir, groupings, consolidator, root_key,
    # dep_summary) — accept **_ so the test doesn't have to mirror the
    # full contract.
    async def reject_callback(**_: typing.Any) -> None:
        return None

    with (
        patch(
            "cxas_scrapi.migration.stage_runner.run_stage_with_redeploy",
            new=AsyncMock(return_value=MagicMock(optimization_logs=[])),
        ),
        patch(
            "cxas_scrapi.migration.service.StructuralConsolidator",
            return_value=fake_consolidator,
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "detect_root_key",
            return_value="RootAgent",
        ),
    ):
        result = await service.run_stage_1(
            bundle=bundle,
            grouping_callback=reject_callback,
            version_label=None,
        )

    assert result is None
    fake_consolidator.consolidate.assert_not_called()
    service._deploy_base_resources.assert_not_awaited()
    assert bundle.grouping is None
    assert bundle.pre_consolidation_ir is None


# ---------------------------------------------------------------------------
# run_stage_2
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_run_stage_2_creates_version_when_label_set() -> None:
    service = _make_service()
    fake_versions_client = MagicMock()

    with (
        patch(
            "cxas_scrapi.migration.stage_runner.run_stage_with_redeploy",
            new=AsyncMock(return_value=MagicMock(optimization_logs=[])),
        ),
        patch(
            "cxas_scrapi.migration.service.Versions",
            return_value=fake_versions_client,
        ),
    ):
        await service.run_stage_2(version_label="0.0.4")

    fake_versions_client.create_version.assert_called_once()
    assert (
        fake_versions_client.create_version.call_args.kwargs["display_name"]
        == "0.0.4"
    )


@pytest.mark.asyncio
async def test_run_stage_2_generate_unit_tests_writes_json(
    tmp_path: typing.Any,
) -> None:
    service = _make_service()
    out_path = str(tmp_path / "unit_tests.json")

    fake_test_case = MagicMock()
    fake_test_case.model_dump = MagicMock(return_value={"name": "tc1"})
    fake_gen = MagicMock()
    fake_gen.generate_tests_for_agent = MagicMock(return_value=[fake_test_case])

    with (
        patch(
            "cxas_scrapi.migration.stage_runner.run_stage_with_redeploy",
            new=AsyncMock(return_value=MagicMock(optimization_logs=[])),
        ),
        patch(
            "cxas_scrapi.migration.service.DeterministicEvalGenerator",
            return_value=fake_gen,
        ),
        patch("cxas_scrapi.migration.service.Versions"),
    ):
        await service.run_stage_2(
            version_label=None,
            generate_unit_tests=True,
            unit_tests_path=out_path,
        )

    assert os.path.exists(out_path)
    with open(out_path) as f:
        data = json.load(f)
    assert "RootAgent" in data
    assert data["RootAgent"][0]["name"] == "tc1"


@pytest.mark.asyncio
async def test_run_stage_2_run_lint_invokes_post_deploy_lint() -> None:
    service = _make_service()
    fake_lint = AsyncMock(return_value=(True, "lint passed"))

    with (
        patch(
            "cxas_scrapi.migration.stage_runner.run_stage_with_redeploy",
            new=AsyncMock(return_value=MagicMock(optimization_logs=[])),
        ),
        patch(
            "cxas_scrapi.migration.post_deploy_lint.run_post_deploy_lint",
            new=fake_lint,
        ),
        patch("cxas_scrapi.migration.service.Versions"),
    ):
        await service.run_stage_2(version_label=None, run_lint=True)

    fake_lint.assert_awaited_once()


# ---------------------------------------------------------------------------
# run_stage_3
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_run_stage_3_requires_grouping_on_bundle() -> None:
    service = _make_service()
    bundle = _make_bundle()  # bundle.grouping is None
    with pytest.raises(RuntimeError, match=r"bundle\.grouping"):
        await service.run_stage_3(bundle=bundle)


@pytest.mark.asyncio
async def test_run_stage_3_persists_bundle_on_success(
    tmp_path: typing.Any,
) -> None:
    service = _make_service()
    bundle = _make_bundle()
    bundle.grouping = {"RootGroup": {"agents": ["Root Agent"], "is_root": True}}
    bundle_path = str(tmp_path / "bundle.json")

    with (
        patch(
            "cxas_scrapi.migration.service.topology_wirer."
            "compute_group_children",
            return_value={"RootGroup": set()},
        ),
        patch(
            "cxas_scrapi.migration.service.topology_wirer.apply_topology",
            return_value=(1, 0, 0),
        ),
        patch(
            "cxas_scrapi.migration.service.topology_wirer.set_app_root_agent",
            return_value=(True, "ok"),
        ),
    ):
        updated, skipped, failed = await service.run_stage_3(
            bundle=bundle, persist_bundle_path=bundle_path
        )

    assert (updated, skipped, failed) == (1, 0, 0)
    assert os.path.exists(bundle_path)
    assert bundle.stage_history[-1].phase == "stage_3"
    assert bundle.stage_history[-1].status == "ok"


@pytest.mark.asyncio
async def test_run_stage_3_triggers_orphan_cleanup_correct_keep_resources() -> (
    None
):
    service = _make_service()
    service.ir.metadata.app_resource_name = "projects/p/locations/us/apps/X"
    service.ir.agents = {
        "RootAgent": IRAgent(
            type="PLAYBOOK",
            display_name="RootAgent",
            instruction="<x/>",
            resource_name="projects/p/locations/us/apps/X/agents/1",
        )
    }

    bundle = _make_bundle()
    bundle.grouping = {"RootGroup": {"agents": ["RootAgent"], "is_root": True}}

    with (
        patch(
            "cxas_scrapi.migration.service.topology_wirer."
            "compute_group_children",
            return_value={"RootGroup": set()},
        ),
        patch(
            "cxas_scrapi.migration.service.topology_wirer.apply_topology",
            return_value=(1, 0, 0),
        ),
        patch(
            "cxas_scrapi.migration.service.topology_wirer.set_app_root_agent",
            return_value=(True, "ok"),
        ),
        patch(
            "cxas_scrapi.migration.service.topology_wirer.delete_orphan_agents",
            return_value=(1, 0),
        ) as mock_delete,
    ):
        await service.run_stage_3(bundle=bundle)

    mock_delete.assert_called_once_with(
        "projects/p/locations/us/apps/X",
        keep_resources={"projects/p/locations/us/apps/X/agents/1"},
    )


# ---------------------------------------------------------------------------
# run_migration back-compat — refactored optimize_for_cxas branch
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_run_migration_optimize_for_cxas_calls_new_stage_methods() -> (
    None
):
    """run_migration with optimize_for_cxas=True should delegate to all three
    snake_case stage methods with correct standard version sequence parameters.
    """

    service = _make_service()

    # Replace the new stage methods so we can assert their invocations
    # without exercising the full optimizer pipeline.
    service.run_stage_1 = AsyncMock(return_value=None)
    service.run_stage_2 = AsyncMock(return_value=None)
    service.run_stage_3 = AsyncMock(return_value=(1, 0, 0))

    # Stub the source loader + reporter so run_migration can reach the
    # optimize_for_cxas branch without hitting external systems.
    service.exporter = MagicMock()
    service.exporter.fetch_full_agent_details.return_value = _make_source_data()
    service.ai_augment = MagicMock()
    service.ai_augment.generate_agent_description = AsyncMock(return_value="D")
    service._process_single_flow = AsyncMock()
    service.reporter = MagicMock()

    fake_versions_client = MagicMock()

    with (
        patch(
            "cxas_scrapi.migration.service.DFCXParameterExtractor."
            "migrate_parameters"
        ) as mock_migrate,
        patch(
            "cxas_scrapi.migration.service.Versions",
            return_value=fake_versions_client,
        ),
    ):
        mock_migrate.return_value = ([], {})
        config = MigrationConfig(
            project_id="test-project",
            target_name="cxas-app",
            model="gemini-2.5-flash-001",
            optimize_for_cxas=True,
        )
        await service.run_migration(source_cx_agent_id="dfcx-1", config=config)

    # Pre-opt Version 0.0.1 was created inline by run_migration.
    fake_versions_client.create_version.assert_called_once()
    assert (
        fake_versions_client.create_version.call_args.kwargs["display_name"]
        == "0.0.1"
    )

    # run_stage_1 was called with version_label="0.0.3"
    # and dedup_version_label="0.0.2".
    service.run_stage_1.assert_awaited_once_with(
        bundle=mock.ANY,
        version_label="0.0.3",
        dedup_version_label="0.0.2",
        persist_bundle_path=None,
    )
    # run_stage_2 was called with version_label="0.0.4".
    service.run_stage_2.assert_awaited_once_with(
        bundle=mock.ANY,
        version_label="0.0.4",
        persist_bundle_path=None,
    )
    # run_stage_3 was called with version_label="0.0.5".
    service.run_stage_3.assert_awaited_once_with(
        bundle=mock.ANY,
        mode="hub",
        version_label="0.0.5",
        persist_bundle_path=None,
    )


# ---------------------------------------------------------------------------
# Phase 4: web-review callback auto-install
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_run_stage_1_installs_web_review_when_configured() -> None:
    """When no callback is supplied and config.web_confirm_grouping is True,
    the service auto-installs grouping_web_review.web_review."""
    service = _make_service()
    bundle = _make_bundle()
    bundle.config.web_confirm_grouping = True
    bundle.config.optimize_for_cxas = True

    fake_consolidator = MagicMock()
    fake_consolidator.propose_groupings = AsyncMock(
        return_value={"G": {"agents": ["Root Agent"], "is_root": True}}
    )

    captured = {}

    async def fake_web_review(**kwargs: typing.Any) -> None:
        captured.update(kwargs)

    with (
        patch(
            "cxas_scrapi.migration.stage_runner.run_stage_with_redeploy",
            new=AsyncMock(return_value=MagicMock(optimization_logs=[])),
        ),
        patch(
            "cxas_scrapi.migration.service.StructuralConsolidator",
            return_value=fake_consolidator,
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "detect_root_key",
            return_value="RootAgent",
        ),
        patch(
            "cxas_scrapi.migration.grouping_web_review.web_review",
            new=fake_web_review,
        ),
        # Force the headless detector to return False so the install
        # path runs even under pytest (which has no TTY).
        patch(
            "cxas_scrapi.migration.service._is_headless_context",
            return_value=False,
        ),
    ):
        result = await service.run_stage_1(
            bundle=bundle,
            version_label=None,
        )

    # Aborted → result None, consolidate not called.
    assert result is None
    fake_consolidator.consolidate.assert_not_called()
    # And the auto-installed callback was actually invoked with the
    # standard kwargs.
    assert captured.get("ir") is service.ir
    assert "RootAgent" not in captured  # this is in groupings, not top-level
    assert captured.get("root_key") == "RootAgent"


@pytest.mark.asyncio
async def test_run_stage_1_skips_web_review_when_auto_confirm() -> None:
    """auto_confirm_grouping=True → no callback installed; Gemini proposal
    applied verbatim."""
    service = _make_service()
    bundle = _make_bundle()
    bundle.config.web_confirm_grouping = True
    bundle.config.auto_confirm_grouping = True
    bundle.config.optimize_for_cxas = True

    fake_consolidator = MagicMock()
    fake_consolidator.propose_groupings = AsyncMock(
        return_value={"G": {"agents": ["Root Agent"], "is_root": True}}
    )
    fake_consolidator.consolidate = MagicMock(return_value=service.ir)
    fake_consolidator.synthesize_instructions = AsyncMock(
        return_value={"G": "ok"}
    )

    called = {"n": 0}

    async def fake_web_review(**kwargs: typing.Any) -> None:  # noqa: ARG001
        called["n"] += 1

    with (
        patch(
            "cxas_scrapi.migration.stage_runner.run_stage_with_redeploy",
            new=AsyncMock(return_value=MagicMock(optimization_logs=[])),
        ),
        patch(
            "cxas_scrapi.migration.service.StructuralConsolidator",
            return_value=fake_consolidator,
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "detect_root_key",
            return_value="RootAgent",
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "validate_groupings",
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "persist_grouping",
        ),
        patch(
            "cxas_scrapi.migration.service.integrity_checks."
            "check_consolidation_integrity",
            return_value=(False, []),
        ),
        patch(
            "cxas_scrapi.migration.grouping_web_review.web_review",
            new=fake_web_review,
        ),
    ):
        await service.run_stage_1(
            bundle=bundle,
            version_label=None,
        )

    assert called["n"] == 0


@pytest.mark.asyncio
async def test_run_stage_1_boots_server_early_and_passes_to_web_review() -> (
    None
):
    """Verify that the service boots the review server early and reuses
    its context during consolidation.
    """
    from unittest.mock import AsyncMock, MagicMock, patch  # noqa: PLC0415

    service = _make_service()
    service._ensure_analysis_builder("test_agent")

    bundle = _make_bundle()
    bundle.config.web_confirm_grouping = True
    bundle.config.auto_confirm_grouping = False
    bundle.config.web_confirm_host = "127.0.0.1"
    bundle.config.web_confirm_port = 0
    bundle.config.web_confirm_timeout_s = 1800
    bundle.config.target_name = "test_agent"

    fake_consolidator = MagicMock()
    fake_consolidator.propose_groupings = AsyncMock(
        return_value={"GroupG": {"agents": ["Root Agent"], "is_root": True}}
    )
    fake_consolidator.synthesize_instructions = AsyncMock(return_value={})

    fake_context = MagicMock()

    boot_mock = AsyncMock(return_value=fake_context)
    review_mock = AsyncMock(
        return_value={"GroupG": {"agents": ["Root Agent"], "is_root": True}}
    )

    with (
        patch(
            "cxas_scrapi.migration.stage_runner.run_stage_1",
            new=AsyncMock(return_value=MagicMock(optimization_logs=[])),
        ),
        patch(
            "cxas_scrapi.migration.service.StructuralConsolidator",
            return_value=fake_consolidator,
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "detect_root_key",
            return_value="RootAgent",
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "validate_groupings",
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "persist_grouping",
        ),
        patch(
            "cxas_scrapi.migration.service.integrity_checks."
            "check_consolidation_integrity",
            return_value=(False, []),
        ),
        patch(
            "cxas_scrapi.migration.grouping_web_review.boot_review_server",
            new=boot_mock,
        ),
        patch(
            "cxas_scrapi.migration.grouping_web_review.web_review",
            new=review_mock,
        ),
        patch(
            "cxas_scrapi.migration.service._is_headless_context",
            return_value=False,
        ),
    ):
        await service.run_stage_1(
            bundle=bundle,
            version_label=None,
        )

    # 1. Verify boot_review_server was called early (once)
    boot_mock.assert_called_once()

    # 2. Verify web_review was called with active_context=fake_context
    review_mock.assert_called_once()
    _, kwargs = review_mock.call_args
    assert kwargs.get("active_context") is fake_context


@pytest.mark.asyncio
async def test_run_stage_1_auto_confirms_in_headless_context() -> None:
    """When web_confirm_grouping is True but the runtime has no TTY
    (CI etc.), the install path must be skipped and Gemini's proposal
    auto-applied so the migration doesn't hang on the 30-min timeout."""
    service = _make_service()
    bundle = _make_bundle()
    bundle.config.web_confirm_grouping = True
    bundle.config.auto_confirm_grouping = False
    bundle.config.optimize_for_cxas = True

    fake_consolidator = MagicMock()
    fake_consolidator.propose_groupings = AsyncMock(
        return_value={"GroupG": {"agents": ["Root Agent"], "is_root": True}}
    )
    fake_consolidator.consolidate = MagicMock(return_value=service.ir)
    fake_consolidator.synthesize_instructions = AsyncMock(
        return_value={"GroupG": "ok"}
    )

    called = {"n": 0}

    async def fake_web_review(**kwargs: typing.Any) -> None:  # noqa: ARG001
        called["n"] += 1

    with (
        patch(
            "cxas_scrapi.migration.stage_runner.run_stage_with_redeploy",
            new=AsyncMock(return_value=MagicMock(optimization_logs=[])),
        ),
        patch(
            "cxas_scrapi.migration.service.StructuralConsolidator",
            return_value=fake_consolidator,
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "detect_root_key",
            return_value="RootAgent",
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "validate_groupings",
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "persist_grouping",
        ),
        patch(
            "cxas_scrapi.migration.service.integrity_checks."
            "check_consolidation_integrity",
            return_value=(False, []),
        ),
        patch(
            "cxas_scrapi.migration.grouping_web_review.web_review",
            new=fake_web_review,
        ),
        patch(
            "cxas_scrapi.migration.service._is_headless_context",
            return_value=True,
        ),
    ):
        await service.run_stage_1(
            bundle=bundle,
            version_label=None,
        )

    assert called["n"] == 0


@pytest.mark.asyncio
async def test_run_stage_1_skips_early_server_boot_in_headless_context() -> (
    None
):
    """The early review-server boot in run_stage_1 must never run in a
    headless context (CI, non-TTY), or pytest runs spawn real HTTP
    servers and browser tabs."""
    service = _make_service()
    bundle = _make_bundle()
    bundle.config.web_confirm_grouping = True
    bundle.config.auto_confirm_grouping = False
    bundle.config.optimize_for_cxas = True

    fake_consolidator = MagicMock()
    fake_consolidator.propose_groupings = AsyncMock(
        return_value={"GroupG": {"agents": ["Root Agent"], "is_root": True}}
    )
    fake_consolidator.consolidate = MagicMock(return_value=service.ir)
    fake_consolidator.synthesize_instructions = AsyncMock(
        return_value={"GroupG": "ok"}
    )

    boot_mock = AsyncMock()

    with (
        patch(
            "cxas_scrapi.migration.stage_runner.run_stage_with_redeploy",
            new=AsyncMock(return_value=MagicMock(optimization_logs=[])),
        ),
        patch(
            "cxas_scrapi.migration.service.StructuralConsolidator",
            return_value=fake_consolidator,
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "detect_root_key",
            return_value="RootAgent",
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "validate_groupings",
        ),
        patch(
            "cxas_scrapi.migration.service.structural_consolidator."
            "persist_grouping",
        ),
        patch(
            "cxas_scrapi.migration.service.integrity_checks."
            "check_consolidation_integrity",
            return_value=(False, []),
        ),
        patch(
            "cxas_scrapi.migration.grouping_web_review.boot_review_server",
            new=boot_mock,
        ),
        patch(
            "cxas_scrapi.migration.service._is_headless_context",
            return_value=True,
        ),
    ):
        await service.run_stage_1(
            bundle=bundle,
            version_label=None,
        )

    boot_mock.assert_not_called()


def _make_run_migration_service() -> MigrationService:
    """Build a MigrationService with everything run_migration touches
    mocked out, mirroring test_run_migration_success."""
    service = MigrationService(
        project_id="test-project",
        ps_apps_client=MagicMock(),
        ps_agents_client=MagicMock(),
        ps_tools_client=MagicMock(),
        ps_toolsets_client=MagicMock(),
        secret_manager_client=MagicMock(),
        cx_api_client=MagicMock(),
    )
    service.exporter = MagicMock()
    service.exporter.fetch_full_agent_details.return_value = DFCXAgentIR(
        name="projects/p/locations/l/agents/a",
        display_name="Test Agent",
        default_language_code="en",
        playbooks=[],
        flows=[],
    )
    service.ai_augment = MagicMock()
    service.ai_augment.generate_agent_description = AsyncMock(
        return_value="Desc"
    )
    service._deploy_base_resources = AsyncMock()
    service._deploy_pending_agents = AsyncMock()
    service._process_single_flow = AsyncMock()
    service.topology_linker = MagicMock()
    service.reporter = MagicMock()
    return service


@pytest.mark.asyncio
async def test_run_migration_skips_early_server_boot_in_headless_context() -> (
    None
):
    """run_migration with the default config (web_confirm_grouping=True)
    must skip the early review-server boot in headless contexts, so
    plain `pytest` runs never spawn servers or browser windows."""
    service = _make_run_migration_service()

    boot_mock = AsyncMock()

    with (
        patch(
            "cxas_scrapi.migration.service.DFCXParameterExtractor."
            "migrate_parameters",
            return_value=([], {}),
        ),
        patch(
            "cxas_scrapi.migration.grouping_web_review.boot_review_server",
            new=boot_mock,
        ),
        patch(
            "cxas_scrapi.migration.service._is_headless_context",
            return_value=True,
        ),
    ):
        # Rely on the config defaults, which leave the web gate enabled.
        config = MigrationConfig(
            project_id="dummy-project",
            target_name="cxas-app",
            model="gemini-2.5-flash-001",
        )
        await service.run_migration(
            source_cx_agent_id="dfcx-123", config=config
        )

    boot_mock.assert_not_called()


@pytest.mark.asyncio
async def test_run_migration_boots_server_early_when_interactive() -> None:
    """Complement to the headless regression test: with a TTY available
    and the web gate enabled, run_migration still boots the review
    server early (once)."""
    service = _make_run_migration_service()

    boot_mock = AsyncMock(return_value=MagicMock())

    with (
        patch(
            "cxas_scrapi.migration.service.DFCXParameterExtractor."
            "migrate_parameters",
            return_value=([], {}),
        ),
        patch(
            "cxas_scrapi.migration.grouping_web_review.boot_review_server",
            new=boot_mock,
        ),
        patch(
            "cxas_scrapi.migration.grouping_web_review.web_review",
            new=AsyncMock(return_value=None),
        ),
        patch(
            "cxas_scrapi.migration.service._is_headless_context",
            return_value=False,
        ),
    ):
        config = MigrationConfig(
            project_id="dummy-project",
            target_name="cxas-app",
            model="gemini-2.5-flash-001",
        )
        await service.run_migration(
            source_cx_agent_id="dfcx-123", config=config
        )

    boot_mock.assert_called_once()