"""U-042: self-hosting closed-loop acceptance -- wires every PH5 module
built so far (U-034~U-041) plus the Graph Engine's own scheduler into the
literal 9-step scenario `self-development-feasibility.md` walks through:
a real self-target Goal, Planner->Coder->Validator (real `claude` calls via
`scheduler.run_goal_to_completion`, which already forces worktree isolation
and records the `core_engine` audit trail for self-target tasks per U-035),
merge, stage a candidate, health-check it, cut it over, monitor it, and
land on EITHER `finalized` (clean monitoring window) OR a rollback (anomaly
detected) -- "走完整...两条路径中至少一条", not a scripted single outcome.

This module is the acceptance scenario's own orchestration glue, not a
claim about what Deploy Supervisor's real production `poll_fn` (U-036's
still-unfilled hook) should look like long-term -- a continuously-running
daemon would discover self-target merges to stage on its own schedule
rather than drive one linear call end to end. Building that discovery loop
is future work, not this Unit's job; this function exists so the acceptance
scenario has one real, callable entry point to run (or NOT run -- see
test_self_hosting_e2e.py's gating) against.
"""

import time

from supervisor import deploy_cutover, deploy_health_check, deploy_monitor, deploy_rollback, deploy_stage
from src import paths, schema, scheduler


class SelfHostingLoopError(Exception):
    """Raised when a precondition for even attempting the scenario isn't
    met (e.g. target_folder_path isn't genuinely self-target) -- distinct
    from a real outcome like 'rolled_back', which is a valid, successful
    exercise of the safety mechanism, not an error."""


def run_self_hosting_closed_loop(
    db_path, goal_text, target_folder_path=None, prod_db_path=None,
    poll_interval=5.0, monitor_timeout_cycles=None, sleep=time.sleep, now=time.time,
    monitor_window_hours=deploy_monitor.MONITORING_WINDOW_HOURS,
    monitor_min_task_count=deploy_monitor.MONITORING_MIN_TASK_COUNT,
):
    """Returns {"stage", "outcome", ...}. `stage` names how far the scenario
    got (`task_lifecycle` / `classification` / `health_check` / `cutover` /
    `monitor`); `outcome` is the concrete result at that stage. A task that
    doesn't reach 'done', or doesn't get classified core_engine, is a valid
    early-return (not every self-target task is a core-engine change -- see
    U-035), not a scenario failure.

    `monitor_window_hours`/`monitor_min_task_count` default to deploy_monitor's
    real production constants (24h / 10 real tasks in the monitored project --
    NOT the disposable per-cycle canary project `check_canary_task` creates,
    which is a different project_id each cycle and never counts toward this).
    A real run (2026-08-03) reached 'cutover' and then correctly sat in
    'monitor' for the full real production window -- neither waiting 24
    real hours nor submitting 10 real Goals against the monitored project is
    practical for an interactive acceptance run, so these are exposed as
    overridable, same as `monitor_timeout_cycles` already is, rather than
    leaving the real scenario impossible to actually finish observing."""
    target_folder_path = target_folder_path or paths.repo_root()
    prod_db_path = prod_db_path or paths.db_path()
    if not paths.is_self_target(target_folder_path):
        raise SelfHostingLoopError(
            "target_folder_path {!r} is not this instance's own checkout ({!r}) -- "
            "the self-hosting scenario requires self-target".format(target_folder_path, paths.repo_root())
        )

    task_result = scheduler.run_goal_to_completion(db_path, target_folder_path, goal_text)
    if task_result.get("final_status") != "done":
        return {"stage": "task_lifecycle", "outcome": task_result.get("final_status"), "task_id": task_result.get("task_id")}

    conn = schema.connect(db_path)
    try:
        bug = conn.execute(
            "SELECT bug_id FROM bug_record WHERE task_id = ? AND root_cause_category = 'core_engine' "
            "ORDER BY rowid DESC LIMIT 1",
            (task_result["task_id"],),
        ).fetchone()
        if bug is None:
            return {"stage": "classification", "outcome": "not_core_engine", "task_id": task_result["task_id"]}

        merge_commit = deploy_stage._current_commit(target_folder_path)

        staged = deploy_stage.stage_deployment(
            conn, merge_commit, repo_root=target_folder_path, prod_db_path=prod_db_path
        )
        health = deploy_health_check.run_health_check(
            conn, staged["deployment_id"], staged["candidate_pid"], staged["candidate_port"]
        )
        if health["status"] != "health_check_passed":
            candidate_db_path = conn.execute(
                "SELECT candidate_db_path FROM deployment WHERE deployment_id = ?", (staged["deployment_id"],)
            ).fetchone()[0]
            deploy_rollback.rollback_path_a(
                conn, staged["deployment_id"], staged["candidate_pid"], candidate_db_path,
                merge_commit, repo_root=target_folder_path,
                description="health check failed: {}".format(health["signals"]),
                candidate_instance_dir=staged["candidate_instance_dir"],
            )
            return {"stage": "health_check", "outcome": "rolled_back_path_a", "deployment_id": staged["deployment_id"]}

        cutover_result = deploy_cutover.cutover(
            conn, staged["deployment_id"], live_pid=None, candidate_pid=staged["candidate_pid"],
            candidate_instance_dir=staged["candidate_instance_dir"], prod_db_path=prod_db_path,
        )

        project_id = conn.execute(
            "SELECT project_id FROM project WHERE target_folder_path = ?", (target_folder_path,)
        ).fetchone()[0]
        cutover_at = conn.execute(
            "SELECT cutover_at FROM deployment WHERE deployment_id = ?", (staged["deployment_id"],)
        ).fetchone()[0]
        baseline = deploy_monitor.compute_baseline_validator_approval_rate(conn, project_id, cutover_at)
        baseline_rate = baseline[0] if baseline else None

        cycles = 0
        while True:
            cycles += 1
            cycle_result = deploy_monitor.run_monitor_cycle(
                conn, staged["deployment_id"], cutover_result["live_pid"], cutover_result["live_port"],
                project_id, baseline_rate, cutover_at, now=now,
                window_hours=monitor_window_hours, min_task_count=monitor_min_task_count,
            )
            if cycle_result["status"] == "monitoring_window_clean":
                conn.execute(
                    "UPDATE deployment SET status = 'finalized' WHERE deployment_id = ?", (staged["deployment_id"],)
                )
                conn.commit()
                return {"stage": "monitor", "outcome": "finalized", "deployment_id": staged["deployment_id"]}
            if cycle_result["status"] == "anomaly_detected":
                rb = deploy_rollback.rollback_path_b(
                    conn, staged["deployment_id"], cutover_result["live_pid"],
                    _previous_instance_dir(conn, staged["deployment_id"]),
                    prod_db_path, merge_commit, repo_root=target_folder_path,
                    description="post-cutover anomaly: {}".format(cycle_result["latest_signal"]),
                )
                outcome = "rolled_back_path_b" if rb["status"] == "rolled_back" else "rollback_failed"
                return {"stage": "monitor", "outcome": outcome, "deployment_id": staged["deployment_id"]}
            if monitor_timeout_cycles is not None and cycles >= monitor_timeout_cycles:
                return {"stage": "monitor", "outcome": "timed_out_still_monitoring", "deployment_id": staged["deployment_id"]}
            sleep(poll_interval)
    finally:
        conn.close()


def _previous_instance_dir(conn, deployment_id):
    return conn.execute(
        "SELECT previous_instance_dir FROM deployment WHERE deployment_id = ?", (deployment_id,)
    ).fetchone()[0]