宜博 / lemongraph

"""c_deploy_supervisor_monitor (03-C §3.15 / 04-P §4.6 / design doc §6):
post-cutover monitoring -- four signals sampled periodically while
`deployment.status='monitoring'`, called repeatedly by Deploy Supervisor's
own poll loop (U-036's `poll_fn`), not a one-shot call (03-C's own text:
"由 Deploy Supervisor 自身的调度循环周期性调用").

`run_monitor_cycle()` is ONE sampling pass: process_alive + http_health +
validator_approval_rate every cycle, canary_task only when the caller says
`run_canary=True` (real scheduling of "every CANARY_INTERVAL" is the
caller's job -- this module just runs whatever it's told to run once,
matching `c_deploy_supervisor_health_check`'s own "idempotent, doesn't
itself decide cadence" framing).

**`validator_approval_rate` operationalization (documented interpretation,
not literally spelled out anywhere in 03-C/04-P/design doc -- there's no
`event_type='validator_approved'` or similar in 01-M's execution_event enum,
so "approved 占比" has to be derived from what IS actually persisted)**: a
task counts as Validator-approved if it reached `task_status_changed(
to_status='validating')` and NO `bug_record(source='validator')` was ever
created for it afterward (Validator's real judgment: PASS -> straight to
`close_task_done`, no bug_record at all; FAIL/INCONCLUSIVE -> `org_level_
loop.run_fix_loop` always creates one via `root_cause_trace.judge_root_cause`
-> `bug_record.create_bug_record`). This is a real, always-true structural
fact about how this codebase's Validator path already works (scheduler.py's
`finish_task_lifecycle`), not a guess -- verified against that code before
writing this, not merely assumed.

**Crash-count / consecutive-fail windows are approximated by sample COUNT,
not wall-clock TIME (documented simplification)**: `MAX_CRASH_COUNT`/
`CRASH_WINDOW` and `MAX_CONSECUTIVE_HTTP_FAIL` are defined in 05 §5.9 as
count-within-a-time-window, but this module counts among the most recent N
persisted `deploy_health_signal` rows (N = CRASH_WINDOW /
MONITOR_SAMPLE_INTERVAL) instead of re-deriving a true rolling time window
per sample -- correct as long as sampling stays roughly interval-paced (the
whole point of a periodic monitor loop); revisit only if sampling cadence
ever becomes genuinely irregular.
"""

import json
import time
import uuid
from datetime import datetime, timezone

from supervisor import deploy_health_check, deploy_supervisor

MONITOR_SAMPLE_INTERVAL = 5 * 60
CANARY_INTERVAL = 60 * 60
MAX_CRASH_COUNT = 2
CRASH_WINDOW = 10 * 60
MAX_CONSECUTIVE_HTTP_FAIL = 3
VALIDATOR_RATE_DEVIATION_THRESHOLD = 0.30  # fraction; 05 §5.9's "±30 个百分点"
MONITORING_WINDOW_HOURS = 24
MONITORING_MIN_TASK_COUNT = 10


def _parse_sqlite_timestamp(text):
    """Same convention as work_node_run.py's own helper of the same name --
    duplicated (not imported) since it's a 2-line stdlib parse and this
    module has no other reason to depend on work_node_run.py."""
    return datetime.strptime(text, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc).timestamp()


def _format_sqlite_timestamp(epoch_seconds):
    return datetime.fromtimestamp(epoch_seconds, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S")


def compute_validator_approval_rate(conn, project_id, since_text, until_text=None):
    """Returns (rate: float in [0,1], sample_size: int), or None if no task
    reached 'validating' in the window (not enough data to mean anything)."""
    query = (
        "SELECT task_id, payload FROM execution_event "
        "WHERE project_id = ? AND event_type = 'task_status_changed' AND created_at >= ?"
    )
    params = [project_id, since_text]
    if until_text is not None:
        query += " AND created_at < ?"
        params.append(until_text)
    rows = conn.execute(query, params).fetchall()

    task_ids = set()
    for task_id, payload_json in rows:
        payload = json.loads(payload_json)
        if payload.get("to_status") == "validating":
            task_ids.add(task_id)
    if not task_ids:
        return None

    approved = sum(
        1 for task_id in task_ids
        if conn.execute(
            "SELECT 1 FROM bug_record WHERE task_id = ? AND source = 'validator' LIMIT 1", (task_id,)
        ).fetchone() is None
    )
    return approved / len(task_ids), len(task_ids)


def compute_baseline_validator_approval_rate(conn, project_id, before_text, lookback_hours=MONITORING_WINDOW_HOURS):
    """Historical rate over the `lookback_hours` immediately preceding
    `before_text` (typically `cutover_at`) -- "与该 deployment 之前的历史基线
    比较" (design doc §6), computed from the same persisted data rather than
    requiring a separately-stored baseline field. Returns None (same as
    compute_validator_approval_rate) if there's no prior data -- callers
    should treat a None baseline as "can't judge deviation yet", not as 0."""
    since_text = _format_sqlite_timestamp(_parse_sqlite_timestamp(before_text) - lookback_hours * 3600)
    return compute_validator_approval_rate(conn, project_id, since_text, until_text=before_text)


def _record_signal(conn, deployment_id, signal):
    conn.execute(
        "INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status, detail) "
        "VALUES (?, ?, CURRENT_TIMESTAMP, ?, ?, ?)",
        (str(uuid.uuid4()), deployment_id, signal["signal_type"], signal["status"], signal["detail"]),
    )
    conn.commit()


def _recent_statuses(conn, deployment_id, signal_type, limit):
    rows = conn.execute(
        "SELECT status FROM deploy_health_signal WHERE deployment_id = ? AND signal_type = ? "
        "ORDER BY rowid DESC LIMIT ?",
        (deployment_id, signal_type, limit),
    ).fetchall()
    return [r[0] for r in rows]


def _sample_process_alive(pid):
    alive = not deploy_supervisor.reap_if_exited(pid)
    return {"signal_type": "process_alive", "status": "ok" if alive else "fail", "detail": None}


def _sample_validator_approval_rate(conn, project_id, baseline_rate, window_start_text, threshold):
    if baseline_rate is None:
        return {
            "signal_type": "validator_approval_rate", "status": "ok",
            "detail": "no baseline yet -- insufficient history to judge deviation",
        }
    result = compute_validator_approval_rate(conn, project_id, window_start_text)
    if result is None:
        return {
            "signal_type": "validator_approval_rate", "status": "ok",
            "detail": "no validated tasks in window yet",
        }
    rate, sample_size = result
    deviation = abs(rate - baseline_rate)
    status = "fail" if deviation > threshold else "ok"
    return {
        "signal_type": "validator_approval_rate", "status": status,
        "detail": "rate={:.2f} n={} baseline={:.2f} deviation={:.2f}".format(rate, sample_size, baseline_rate, deviation),
    }


def _real_task_count_since(conn, project_id, since_text):
    return conn.execute(
        "SELECT COUNT(*) FROM task WHERE project_id = ? AND created_at >= ?", (project_id, since_text)
    ).fetchone()[0]


def run_monitor_cycle(
    conn, deployment_id, pid, port, project_id, baseline_validator_rate, window_start_text,
    run_canary=True, crash_window_samples=None, consecutive_http_fail_limit=MAX_CONSECUTIVE_HTTP_FAIL,
    validator_threshold=VALIDATOR_RATE_DEVIATION_THRESHOLD, min_task_count=MONITORING_MIN_TASK_COUNT,
    window_hours=MONITORING_WINDOW_HOURS, now=time.time,
):
    """One sampling pass. Returns {status, latest_signal}. `window_start_text`
    is `deployment.cutover_at` (a sqlite TEXT timestamp) -- used both for
    the elapsed-time-since-cutover math and as the SQL lower bound for
    "real tasks since the window opened"/"validator judgments since the
    window opened". Stops sampling further signals the moment one comes
    back fail (same fail-closed, don't-pile-on-more-checks pattern as
    `c_deploy_supervisor_health_check`) and transitions
    `deployment.status='anomaly_detected'` immediately -- 04-P's own state
    machine has no intermediate "still checking" state distinct from
    'monitoring' itself.
    """
    if crash_window_samples is None:
        crash_window_samples = max(1, CRASH_WINDOW // MONITOR_SAMPLE_INTERVAL)

    process_signal = _sample_process_alive(pid)
    _record_signal(conn, deployment_id, process_signal)
    latest_signal = process_signal
    anomaly = _recent_statuses(conn, deployment_id, "process_alive", crash_window_samples).count("fail") >= MAX_CRASH_COUNT

    if not anomaly:
        http_signal = deploy_health_check.check_http_health(port)
        _record_signal(conn, deployment_id, http_signal)
        latest_signal = http_signal
        recent_http = _recent_statuses(conn, deployment_id, "http_health", consecutive_http_fail_limit)
        anomaly = len(recent_http) >= consecutive_http_fail_limit and all(s == "fail" for s in recent_http)

    if not anomaly and run_canary:
        canary_signal = deploy_health_check.check_canary_task(port)
        _record_signal(conn, deployment_id, canary_signal)
        latest_signal = canary_signal
        anomaly = canary_signal["status"] == "fail"  # single failure -- no grey area (design doc §6)

    if not anomaly:
        rate_signal = _sample_validator_approval_rate(
            conn, project_id, baseline_validator_rate, window_start_text, validator_threshold
        )
        _record_signal(conn, deployment_id, rate_signal)
        latest_signal = rate_signal
        anomaly = rate_signal["status"] == "fail"

    if anomaly:
        conn.execute(
            "UPDATE deployment SET status = 'anomaly_detected', rollback_reason = ? WHERE deployment_id = ?",
            (json.dumps(latest_signal), deployment_id),
        )
        conn.commit()
        return {"status": "anomaly_detected", "latest_signal": latest_signal}

    elapsed_hours = (now() - _parse_sqlite_timestamp(window_start_text)) / 3600.0
    real_task_count = _real_task_count_since(conn, project_id, window_start_text)
    if elapsed_hours >= window_hours or real_task_count >= min_task_count:
        conn.execute(
            "UPDATE deployment SET status = 'monitoring_window_clean', monitoring_closed_at = CURRENT_TIMESTAMP "
            "WHERE deployment_id = ?",
            (deployment_id,),
        )
        conn.commit()
        return {"status": "monitoring_window_clean", "latest_signal": latest_signal}

    conn.execute("UPDATE deployment SET status = 'monitoring' WHERE deployment_id = ?", (deployment_id,))
    conn.commit()
    return {"status": "monitoring", "latest_signal": latest_signal}