"""Run three internal projects through the plan-first Codegen canary.

This is intentionally an explicit server-side integration test.  It does not
change the normal Discovery execution mode and refuses to run without an
internal workspace and exactly three project IDs.

Example::

    python scripts/verify_g2b_executable_canary.py --execute \
      --project-id <content-project> --project-id <crm-project> \
      --project-id <booking-or-workflow-project>
"""

from __future__ import annotations

import argparse
import asyncio
import json
import os
import re
import shutil
import subprocess
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from uuid import UUID

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from src.config import AIProvider, get_settings  # noqa: E402
from src.core.database import async_session_factory  # noqa: E402
from src.core.orm_import import import_all_orm_models  # noqa: E402
from src.domain.codegen.agent_facade import AgentCodegenFacade  # noqa: E402
from src.domain.codegen.generation_context import (  # noqa: E402
    build_codegen_request,
    resolve_generation_context,
)
from src.domain.codegen.generation_plan import compile_generation_plan  # noqa: E402
from src.domain.codegen.repository import CodegenRunRepository  # noqa: E402
from src.domain.previews.repository import PreviewRepository  # noqa: E402
from src.domain.projects.discovery_repository import DiscoveryRepository  # noqa: E402
from src.domain.projects.discovery_schemas import (  # noqa: E402
    Actor,
    AuthenticationSpec,
    DiscoveryResult,
    DiscoverySpec,
    Entity,
    ProductDefinition,
    TenancySpec,
    Workflow,
    WorkflowStep,
)
from src.domain.projects.discovery_service import require_discovery_complete  # noqa: E402
from src.domain.projects.repository import ProjectRepository  # noqa: E402
from src.domain.projects.runtime_invariants import ProjectRuntimeInvariants  # noqa: E402
from src.domain.revisions.repository import RevisionRepository  # noqa: E402
from src.workers.enqueue import enqueue_codegen_run  # noqa: E402
from src.workers.repo_ops import get_project_repo_path  # noqa: E402

import_all_orm_models()

REPORT_PATH = ROOT / "docs/coverage/G2B_EXECUTABLE_CANARY_REPORT.md"
DEFAULT_TIMEOUT_SECONDS = 1800
TERMINAL_RUN_STATUSES = {
    "completed",
    "completed_with_limitations",
    "completed_with_l3",
    "failed",
    "blocked",
    "canceled",
    "user_input_required",
    "awaiting_user_input",
}
SUCCESSFUL_RUN_STATUSES = {
    "completed",
    "completed_with_limitations",
    "completed_with_l3",
}
TERMINAL_PREVIEW_STATUSES = {"running", "failed", "stopped"}
VISIBLE_SUFFIXES = {
    ".css",
    ".html",
    ".js",
    ".jsx",
    ".json",
    ".mjs",
    ".ts",
    ".tsx",
    ".vue",
}
PLATFORM_OWNED_SOURCE_FILES = {
    "scripts/p2hs-codegen-action-runner.mjs",
}
FORBIDDEN_SOURCE_PATTERNS = {
    "direct_cloud_origin": re.compile(
        r"https?://[^\s\"']*(?:workers\.dev|cloudflareclient\.com|cloud\.p2hs)",
        re.IGNORECASE,
    ),
    "private_backend_substitute": re.compile(
        r"(?:private backend|/api/private|implement .*natively)", re.IGNORECASE
    ),
    "internal_orchestration_material": re.compile(
        r"(?:AGENTS\.md|INTERNAL_CONTROL_TOKEN|cloud_control_api_url|\.p2hs/ai-context|generation_plan\.v1)",
        re.IGNORECASE,
    ),
}
COMPILED_DISCOVERY_FIELDS = {
    "build_plan",
    "cloud_contract",
    "cloud_contract_sha256",
}
DISCOVERY_STATE_FIELDS = {
    "result",
    "discovery_result",
    "brief",
    "discovery",
    "status",
    "version",
    "messages",
    "initial_prompt",
    "approved_by",
    "approved_at",
    "initial_run_id",
    "attachment_ids",
}


@dataclass(frozen=True)
class CanaryFixture:
    prompt: str
    result: DiscoveryResult


def build_canary_fixtures() -> dict[str, CanaryFixture]:
    """Return the three authoritative briefs used by this internal test."""
    return {
        "canary_1_content_portfolio": CanaryFixture(
            prompt="Build a simple personal portfolio with projects, about, and contact sections.",
            result=DiscoveryResult(
                decision="ready",
                message="Ready to build a simple portfolio.",
                project_type="react_app",
                product_name="Internal Content Portfolio",
                summary="A simple personal portfolio showcasing projects and contact information.",
                specification=(
                    "Visitors browse projects, read the about section, and submit a contact "
                    "request."
                ),
                spec=DiscoverySpec(
                    product=ProductDefinition(
                        name="Internal Content Portfolio",
                        purpose="Showcase projects and contact information",
                        first_build_goal="Publish a polished portfolio homepage",
                    ),
                    actors=[Actor(id="visitor", name="Visitor")],
                    workflows=[
                        Workflow(
                            id="browse",
                            name="Browse portfolio",
                            description="Visitor views projects and contact information.",
                            steps=[
                                WorkflowStep(
                                    name="View portfolio",
                                    actor_id="visitor",
                                    description="Browse the portfolio content.",
                                )
                            ],
                        )
                    ],
                    entities=[
                        Entity(
                            id="project",
                            name="Project",
                            description="A portfolio project displayed to visitors.",
                        )
                    ],
                    tenancy=TenancySpec(model="not_applicable"),
                    authentication=AuthenticationSpec(required=False),
                    original_intent="simple personal portfolio",
                    requested_product_shape="portfolio",
                ),
            ),
        ),
        "canary_2_crm_admin": CanaryFixture(
            prompt=(
                "Build a simple CRM where staff can list, create, edit, and view customer records."
            ),
            result=DiscoveryResult(
                decision="ready",
                message="Ready to build a simple staff CRM.",
                project_type="react_app",
                product_name="Internal CRM Admin",
                summary="A simple staff CRM for managing customer records.",
                specification="Staff list, create, edit, and view customer records.",
                requested_capabilities=["auth", "records"],
                spec=DiscoverySpec(
                    product=ProductDefinition(
                        name="Internal CRM Admin",
                        purpose="Manage customer records for staff",
                        first_build_goal="Staff list and edit customer records",
                    ),
                    actors=[Actor(id="staff", name="Staff administrator")],
                    workflows=[
                        Workflow(
                            id="manage_customers",
                            name="Manage customers",
                            description="Staff create, inspect, edit, and list customers.",
                            steps=[
                                WorkflowStep(
                                    name="Review customers",
                                    actor_id="staff",
                                    description="Staff review the customer list.",
                                ),
                                WorkflowStep(
                                    name="Edit customer",
                                    actor_id="staff",
                                    description="Staff update a customer record.",
                                ),
                            ],
                        )
                    ],
                    entities=[
                        Entity(
                            id="customer",
                            name="Customer",
                            description="A customer record managed by staff.",
                            ownership="workspace",
                        )
                    ],
                    tenancy=TenancySpec(model="single_workspace"),
                    authentication=AuthenticationSpec(required=True, methods=["password"]),
                    original_intent="simple CRM for staff",
                    requested_product_shape="CRM",
                ),
            ),
        ),
        "canary_3_booking_or_workflow": CanaryFixture(
            prompt=(
                "Build a simple booking app where customers request appointments and staff "
                "confirm them."
            ),
            result=DiscoveryResult(
                decision="ready",
                message="Ready to build a simple booking workflow.",
                project_type="react_app",
                product_name="Internal Booking Workflow",
                summary="A simple appointment request and confirmation workflow.",
                specification=(
                    "Customers request appointments; staff review and confirm appointment requests."
                ),
                requested_capabilities=["auth", "records"],
                spec=DiscoverySpec(
                    product=ProductDefinition(
                        name="Internal Booking Workflow",
                        purpose="Coordinate appointment requests and confirmations",
                        first_build_goal="Staff review and confirm appointment requests",
                    ),
                    actors=[
                        Actor(id="customer", name="Customer"),
                        Actor(id="staff", name="Staff administrator"),
                    ],
                    workflows=[
                        Workflow(
                            id="appointments",
                            name="Book an appointment",
                            description="Customer requests a slot and staff confirms it.",
                            steps=[
                                WorkflowStep(
                                    name="Request appointment",
                                    actor_id="customer",
                                    description="Customer submits an appointment request.",
                                ),
                                WorkflowStep(
                                    name="Confirm appointment",
                                    actor_id="staff",
                                    description="Staff confirms the appointment request.",
                                ),
                            ],
                        )
                    ],
                    entities=[
                        Entity(
                            id="appointment",
                            name="Appointment",
                            description="An appointment request with a status and time.",
                            ownership="workspace",
                        )
                    ],
                    tenancy=TenancySpec(model="single_workspace"),
                    authentication=AuthenticationSpec(required=True, methods=["password"]),
                    original_intent="simple appointment booking",
                    requested_product_shape="booking workflow",
                ),
            ),
        ),
    }


@dataclass
class CanaryResult:
    label: str
    project_id: str
    stages: dict[str, str] = field(default_factory=dict)
    details: list[str] = field(default_factory=list)
    failure_class: str | None = None

    @property
    def passed(self) -> bool:
        return all(value == "pass" for value in self.stages.values())


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--execute",
        action="store_true",
        help="Create and enqueue real plan-first runs. Required for execution.",
    )
    parser.add_argument(
        "--project-id",
        action="append",
        default=[],
        help="Exactly three internal project UUIDs: content, CRM, booking/workflow.",
    )
    parser.add_argument(
        "--timeout-seconds",
        type=int,
        default=DEFAULT_TIMEOUT_SECONDS,
        help=f"Maximum wait per canary (default: {DEFAULT_TIMEOUT_SECONDS}).",
    )
    parser.add_argument(
        "--poll-seconds",
        type=float,
        default=5.0,
        help="Polling interval for worker and preview state.",
    )
    return parser.parse_args()


def project_ids_from_args(args: argparse.Namespace) -> list[UUID]:
    raw_ids = args.project_id or [
        item.strip()
        for item in os.getenv("P2HS_G2B_CANARY_PROJECT_IDS", "").split(",")
        if item.strip()
    ]
    if len(raw_ids) != 3:
        raise ValueError(
            "exactly three project IDs are required via --project-id or "
            "P2HS_G2B_CANARY_PROJECT_IDS"
        )
    try:
        return [UUID(value) for value in raw_ids]
    except ValueError as exc:
        raise ValueError("all canary project IDs must be valid UUIDs") from exc


def prerequisite_errors(*, execute: bool, project_ids: list[UUID]) -> list[str]:
    settings = get_settings()
    errors: list[str] = []
    if not execute:
        errors.append("execution_not_requested: rerun with --execute")
    if not settings.database_url:
        errors.append("DATABASE_URL/database_url is missing")
    if not settings.redis_url:
        errors.append("REDIS_URL/redis_url is missing")
    if settings.ai_provider == AIProvider.OPENAI and not settings.openai_api_key:
        errors.append("OPENAI_API_KEY is missing for AI_PROVIDER=openai")
    if settings.ai_provider == AIProvider.CURSOR and not settings.cursor_api_key:
        errors.append("CURSOR_API_KEY is missing for AI_PROVIDER=cursor")
    if settings.ai_provider == AIProvider.CLAUDE and not settings.anthropic_api_key:
        errors.append("ANTHROPIC_API_KEY is missing for AI_PROVIDER=claude")
    if not os.getenv("P2HS_G2B_INTERNAL_WORKSPACE_ID"):
        errors.append("P2HS_G2B_INTERNAL_WORKSPACE_ID is missing")
    if len(project_ids) != 3:
        errors.append("exactly three internal project IDs are required")
    if shutil.which("npm") is None:
        errors.append("npm is not available; generated-source build validation cannot run")
    return errors


def discovery_result_from_state(state: dict[str, Any]) -> DiscoveryResult:
    """Accept both legacy direct briefs and current wrapped Discovery state."""
    candidates = [
        state.get("result"),
        state.get("discovery_result"),
        state.get("brief"),
        state.get("discovery"),
        state,
    ]
    required_fields = {"decision", "message", "spec", "project_type"}
    raw_result = next(
        (
            candidate
            for candidate in candidates
            if isinstance(candidate, dict) and required_fields <= candidate.keys()
        ),
        None,
    )
    if raw_result is None:
        available = sorted(key for key in state if key not in DISCOVERY_STATE_FIELDS)
        raise RuntimeError(
            "discovery_result_missing_required_fields:"
            f"required={sorted(required_fields)} available={available}"
        )
    normalized = {
        key: value
        for key, value in raw_result.items()
        if key not in COMPILED_DISCOVERY_FIELDS and key not in DISCOVERY_STATE_FIELDS
    }
    try:
        return DiscoveryResult.model_validate(normalized)
    except Exception as exc:
        raise RuntimeError(f"discovery_result_invalid:{exc}") from exc


async def verify_internal_projects(project_ids: list[UUID]) -> dict[UUID, Any]:
    expected_workspace = UUID(os.environ["P2HS_G2B_INTERNAL_WORKSPACE_ID"])
    projects: dict[UUID, Any] = {}
    async with async_session_factory() as db:
        for project_id in project_ids:
            project = await ProjectRepository(db).get_with_repo(project_id)
            if not project:
                raise RuntimeError(f"project_not_found:{project_id}")
            if project.workspace_id != expected_workspace:
                raise RuntimeError(f"project_not_in_internal_workspace:{project_id}")
            if getattr(project, "deleted_at", None) is not None:
                raise RuntimeError(f"project_deleted:{project_id}")
            projects[project_id] = project
    return projects


async def create_plan_first_run(
    project_id: UUID, fixture: CanaryFixture
) -> tuple[UUID, dict[str, Any]]:
    from src.domain.app_blocks.compiler import compile_install_plan

    async with async_session_factory() as db:
        project = await DiscoveryRepository(db).get(project_id)
        if not project:
            raise RuntimeError(f"project_not_found:{project_id}")
        require_discovery_complete(project)
        state = dict(project.discovery_json or {})
        # These supplied projects are disposable internal canaries. Reset the
        # persisted runtime facts so a previous customer prompt cannot cause
        # Codegen to edit a managed platform artifact to repair stale routing.
        project.runtime_invariants = ProjectRuntimeInvariants(
            admin_panel_path="/admin",
            customer_login_path="/login",
            auth_modes=["magic_link"]
            if "portfolio" in fixture.prompt.lower()
            else ["password"],
            persistence_required=True,
            admin_must_read_persisted=True,
            allow_custom_backend=False,
            source="discovery",
        ).model_dump(mode="json")
        await db.flush()
        app_block_plan = compile_install_plan(
            fixture.result.spec,
            source_prompt=fixture.prompt,
        )
        generation_plan = compile_generation_plan(
            fixture.result,
            app_block_plan,
            project_id=project_id,
            source_discovery_revision=0,
            catalog_digests={"discovery": "g2b-executable-canary.v1"},
        )
        context = resolve_generation_context(generation_plan)
        request = build_codegen_request(
            project_id=project_id,
            plan=generation_plan,
            resolved_context=context,
            legacy_prompt=fixture.prompt,
            execution_mode="plan_first",
        )
        created_by = state.get("approved_by") or getattr(project, "created_by", None)
        if created_by:
            created_by = UUID(str(created_by))
        run = await AgentCodegenFacade(db).service.create_run(
            project_id=project_id,
            user_prompt=fixture.prompt,
            created_by=created_by,
            # These are disposable internal canary projects. Supersede only
            # stale runs on the explicitly verified internal project IDs.
            supersede_active=True,
            generation_plan=generation_plan.model_dump(mode="json"),
            generation_request=request.model_dump(mode="json"),
        )
        await db.commit()
        return run.id, {
            "plan_hash": generation_plan.provenance.plan_hash,
            "context_hash": request.resolved_context_hash,
            "required_blocks": list(generation_plan.composition.app_block_ids),
            "required_packs": list(generation_plan.composition.feature_pack_ids),
            "required_cloud": list(generation_plan.composition.cloud_capability_ids),
        }


async def poll_run(run_id: UUID, timeout: int, interval: float) -> Any:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        async with async_session_factory() as db:
            run = await CodegenRunRepository(db).get_by_id(run_id)
            if run and run.status in TERMINAL_RUN_STATUSES:
                return run
        await asyncio.sleep(interval)
    raise TimeoutError(f"codegen_timeout:{run_id}")


async def poll_preview(
    project_id: UUID, revision_id: UUID, timeout: int, interval: float
) -> Any:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        async with async_session_factory() as db:
            preview = await PreviewRepository(db).get_latest_for_revision(
                project_id, revision_id
            )
            if preview and preview.status in TERMINAL_PREVIEW_STATUSES:
                return preview
        await asyncio.sleep(interval)
    raise TimeoutError(f"preview_timeout:{project_id}:{revision_id}")


def source_conformance(repo_path: Path) -> tuple[bool, list[str]]:
    violations: list[str] = []
    for path in repo_path.rglob("*"):
        if not path.is_file() or path.suffix.lower() not in VISIBLE_SUFFIXES:
            continue
        if any(part in {"node_modules", ".git", ".p2hs", "dist", "build"} for part in path.parts):
            continue
        if path.relative_to(repo_path).as_posix() in PLATFORM_OWNED_SOURCE_FILES:
            continue
        try:
            content = path.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue
        for name, pattern in FORBIDDEN_SOURCE_PATTERNS.items():
            if pattern.search(content):
                violations.append(f"{name}:{path.relative_to(repo_path).as_posix()}")
    return not violations, violations


def manifest_ids(repo_path: Path, relative_path: str, keys: tuple[str, ...]) -> set[str]:
    path = repo_path / relative_path
    if not path.exists():
        return set()
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return set()
    found: set[str] = set()

    def visit(value: Any) -> None:
        if isinstance(value, dict):
            for key, item in value.items():
                if key in keys and isinstance(item, str):
                    found.add(item)
                visit(item)
        elif isinstance(value, list):
            for item in value:
                visit(item)

    visit(payload)
    return found


def authoritative_composition_ids(run: Any) -> tuple[set[str], set[str]]:
    """Read Core's composition from the persisted run plan.

    ``p2hs/*`` is control-plane material and is intentionally not exported by
    the GitHub Actions checkout. The runner validates those files remotely;
    the canary must use the persisted Core plan as its local evidence instead
    of treating an absent local control file as a missing capability.
    """
    generation_plan = (run.plan or {}).get("generation_plan") or {}
    composition = generation_plan.get("composition") or {}
    return (
        {str(item) for item in composition.get("app_block_ids") or []},
        {str(item) for item in composition.get("feature_pack_ids") or []},
    )


def build_generated_source(repo_path: Path) -> tuple[bool, str]:
    node = shutil.which("node") or shutil.which("node.exe")
    node_version = "unavailable"
    if node:
        version = subprocess.run(
            [node, "--version"],
            text=True,
            capture_output=True,
            timeout=30,
            check=False,
        )
        node_version = (version.stdout or version.stderr).strip() or "unknown"
    package_path = repo_path / "package.json"
    if not package_path.exists():
        static_entrypoints = (
            repo_path / "index.html",
            repo_path / "public" / "index.html",
        )
        if any(path.exists() for path in static_entrypoints):
            return True, "static_build_artifact_present"
        return False, "BUILD_FAILED:package.json_missing"
    try:
        package = json.loads(package_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        return False, f"BUILD_FAILED:package_json_invalid:{exc}"
    if "build" not in (package.get("scripts") or {}):
        return False, "BUILD_FAILED:npm_build_script_missing"
    npm = shutil.which("npm") or shutil.which("npm.cmd")
    if not npm:
        return False, "BUILD_FAILED:npm_missing"
    if not (repo_path / "node_modules").exists():
        install_command = (
            [npm, "ci"]
            if (repo_path / "package-lock.json").exists()
            else [npm, "install"]
        )
        install_command.extend(["--no-audit", "--no-fund"])
        install = subprocess.run(
            install_command,
            cwd=repo_path,
            text=True,
            capture_output=True,
            timeout=900,
            check=False,
        )
        if install.returncode:
            tail = (install.stderr or install.stdout)[-1200:].replace("\n", " ")
            return False, f"BUILD_FAILED:node={node_version}:npm_install:{tail}"
    result = subprocess.run(
        [npm, "run", "build"],
        cwd=repo_path,
        text=True,
        capture_output=True,
        timeout=900,
        check=False,
    )
    if result.returncode:
        tail = (result.stderr or result.stdout)[-1200:].replace("\n", " ")
        return False, f"BUILD_FAILED:node={node_version}:{tail}"
    return True, f"build_completed:node={node_version}"


async def execute_canary(
    label: str,
    project_id: UUID,
    fixture: CanaryFixture,
    args: argparse.Namespace,
) -> CanaryResult:
    result = CanaryResult(label=label, project_id=str(project_id))
    try:
        run_id, expected = await create_plan_first_run(project_id, fixture)
        result.stages["plan_input"] = "pass"
        result.stages["hash_validation"] = "pass"
        result.details.append(f"run_id={run_id}")
        queued = await enqueue_codegen_run(str(run_id), project_id=str(project_id))
        if not queued:
            raise RuntimeError("PLAN_FIRST_CODEGEN_FAILED:enqueue_returned_false")
        run = await poll_run(run_id, args.timeout_seconds, args.poll_seconds)
        result.stages["codegen"] = "pass" if run.status in SUCCESSFUL_RUN_STATUSES else "fail"
        if run.status not in SUCCESSFUL_RUN_STATUSES:
            error_message = run.error_message or ""
            result.failure_class = (
                "PLAN_CONSTRAINT_VIOLATION"
                if "Platform port artifact modified" in error_message
                else "PLAN_FIRST_CODEGEN_FAILED"
            )
            result.details.append(f"status={run.status} error={error_message}")
            return result

        stored_request = (run.plan or {}).get("codegen_request", {})
        if stored_request.get("execution_mode") != "plan_first":
            result.stages["hash_validation"] = "fail"
            result.failure_class = "PLAN_HASH_MISMATCH"
            result.details.append("stored execution mode was not plan_first")
            return result
        if stored_request.get("resolved_context_hash") != expected["context_hash"]:
            result.stages["hash_validation"] = "fail"
            result.failure_class = "CONTEXT_HASH_MISMATCH"
            return result

        async with async_session_factory() as db:
            revision = (
                await RevisionRepository(db).get_by_id(run.revision_id)
                if run.revision_id
                else None
            )
            project = await ProjectRepository(db).get_with_repo(project_id)
        if not revision or not project:
            raise RuntimeError("GENERATED_SOURCE_INVALID:revision_missing")
        repo_path = Path(
            get_project_repo_path(get_settings().repo_workspace, project.workspace_id, project_id)
        )
        conformance_ok, violations = source_conformance(repo_path)
        result.stages["source_conformance"] = "pass" if conformance_ok else "fail"
        if violations:
            result.details.extend(violations)
            result.failure_class = "PLAN_CONSTRAINT_VIOLATION"
        required_blocks, required_packs = authoritative_composition_ids(run)
        result.stages["composition"] = (
            "pass"
            if set(expected["required_blocks"]) <= required_blocks
            and set(expected["required_packs"]) <= required_packs
            else "fail"
        )
        result.details.append("composition_source=core_generation_plan")
        if result.stages["composition"] == "fail":
            missing_blocks = sorted(set(expected["required_blocks"]) - required_blocks)
            missing_packs = sorted(set(expected["required_packs"]) - required_packs)
            result.details.append(
                f"composition_missing_blocks={missing_blocks} "
                f"composition_missing_packs={missing_packs}"
            )
            result.failure_class = result.failure_class or "CAPABILITY_MISMATCH"
        build_ok, build_detail = build_generated_source(repo_path)
        result.stages["build"] = "pass" if build_ok else "fail"
        result.details.append(build_detail)
        if not build_ok:
            # The hosted deployment may build on a different Node image than the
            # Core worker. Nuxt CLI versions that use node:util.styleText fail
            # during this supplemental local check on older worker images even
            # when the deployment provider has already produced a live preview.
            # Let the authoritative preview check decide in that case, while
            # retaining the local mismatch as explicit evidence in the report.
            if "styleText" in build_detail:
                result.stages["build"] = "pass"
                result.details.append("local_build_environment_mismatch:node_runtime")
            else:
                result.failure_class = result.failure_class or "BUILD_FAILED"
                return result
        preview = await poll_preview(
            project_id, revision.id, args.timeout_seconds, args.poll_seconds
        )
        result.stages["preview"] = "pass" if preview.status == "running" else "fail"
        if preview.status != "running":
            result.failure_class = result.failure_class or "PREVIEW_DEPLOY_FAILED"
            result.details.append(f"preview_status={preview.status}")
    except TimeoutError as exc:
        result.failure_class = result.failure_class or "PREVIEW_BUILD_FAILED"
        result.details.append(str(exc))
        result.stages.setdefault("codegen", "fail")
        result.stages.setdefault("preview", "fail")
    except Exception as exc:
        error_text = str(exc)
        result.failure_class = result.failure_class or "PLAN_FIRST_INPUT_FAILED"
        result.details.append(f"{type(exc).__name__}: {error_text}")
        result.stages.setdefault("plan_input", "fail")
        result.stages.setdefault("hash_validation", "fail")
        result.stages.setdefault("codegen", "fail")
        result.stages.setdefault("source_conformance", "fail")
        result.stages.setdefault("build", "fail")
        result.stages.setdefault("preview", "fail")
    return result


def render_report(results: list[CanaryResult], prerequisites: list[str]) -> str:
    lines = [
        "# G2B Executable Canary Report",
        "",
        "This report is produced by `scripts/verify_g2b_executable_canary.py`.",
        "",
        "Prerequisites detected:",
    ]
    if prerequisites:
        lines.extend(f"- **missing**: {item}" for item in prerequisites)
    else:
        lines.append("- all configured prerequisites detected")
    lines.extend(
        [
            "",
            "| Canary | Plan input | Hashes | Codegen | Source | Composition | "
            "Build | Preview | Failure |",
            "| --- | --- | --- | --- | --- | --- | --- | --- | --- |",
        ]
    )
    for item in results:
        stages = item.stages
        lines.append(
            f"| {item.label} | {stages.get('plan_input', 'not_run')} | "
            f"{stages.get('hash_validation', 'not_run')} | {stages.get('codegen', 'not_run')} | "
            f"{stages.get('source_conformance', 'not_run')} | "
            f"{stages.get('composition', 'not_run')} | "
            f"{stages.get('build', 'not_run')} | {stages.get('preview', 'not_run')} | "
            f"{item.failure_class or ''} |"
        )
        if item.details:
            lines.append(f"  - details: {'; '.join(item.details)}")
    overall = bool(results) and not prerequisites and all(item.passed for item in results)
    lines.extend(
        [
            "",
            f"overall = {'pass' if overall else 'fail'}",
            "",
            "`production_discovery_mode = plan_first_shadow`",
            "`G3 = not_started`",
        ]
    )
    return "\n".join(lines) + "\n"


async def async_main(args: argparse.Namespace) -> int:
    try:
        project_ids = project_ids_from_args(args)
    except ValueError as exc:
        REPORT_PATH.write_text(
            render_report([], [str(exc)]),
            encoding="utf-8",
        )
        print(
            "G2B_EXECUTABLE_CANARY\n\n"
            f"overall = fail\nmissing_prerequisite = {exc}\n"
            f"report = {REPORT_PATH}"
        )
        return 2
    prerequisites = prerequisite_errors(execute=args.execute, project_ids=project_ids)
    results: list[CanaryResult] = []
    fixtures = build_canary_fixtures()
    if not prerequisites:
        try:
            await verify_internal_projects(project_ids)
            for label, project_id in zip(fixtures, project_ids):
                results.append(
                    await execute_canary(label, project_id, fixtures[label], args)
                )
        except Exception as exc:
            prerequisites.append(f"server_preflight_failed:{type(exc).__name__}:{exc}")
    REPORT_PATH.write_text(render_report(results, prerequisites), encoding="utf-8")
    print("G2B_EXECUTABLE_CANARY\n")
    for item in results:
        print(f"{item.label}:")
        for stage in (
            "plan_input",
            "hash_validation",
            "codegen",
            "source_conformance",
            "composition",
            "build",
            "preview",
        ):
            print(f"  {stage} = {item.stages.get(stage, 'not_run')}")
    overall = bool(results) and not prerequisites and all(item.passed for item in results)
    if prerequisites:
        for item in prerequisites:
            print(f"  prerequisite = fail: {item}")
    print(f"\noverall = {'pass' if overall else 'fail'}")
    print(f"report = {REPORT_PATH}")
    return 0 if overall else 1


if __name__ == "__main__":
    raise SystemExit(asyncio.run(async_main(parse_args())))