code/self_healing_finetuning/eval/human_spot_check_email.py

from __future__ import annotations

import imaplib
import json
import os
import sys
import threading
import time
from email import message_from_bytes
from pathlib import Path
from typing import Any, TextIO

from eval.model_quality_gate import load_model_quality_thresholds
from n21.config import write_json
from n21.settings import SHFT_WORKSPACE_ROOT
from observability.audit_log import utc_now
from orchestrator.human_owner_decision import (
    _env_first,
    _extract_message_text,
    _owner_email,
    _send_email_or_outbox,
    _shcg_default,
)


VALID_REVIEW_DECISIONS = {"approve", "reject"}


def _response_root() -> Path:
    return SHFT_WORKSPACE_ROOT / "human_spot_check_reviews"


def _response_paths(run_id: str) -> list[Path]:
    return [
        SHFT_WORKSPACE_ROOT / "runs" / run_id / "eval" / "human_spot_check_response.json",
        _response_root() / "responses" / f"{run_id}.json",
    ]


def _normalize_decision(value: str) -> str | None:
    text = value.strip().lower()
    if text in VALID_REVIEW_DECISIONS:
        return text
    if text in {"approved", "pass", "passed", "yes", "y"}:
        return "approve"
    if text in {"rejected", "fail", "failed", "no", "n"}:
        return "reject"
    return None


def _find_decision(text: str) -> str | None:
    direct = _normalize_decision(text)
    if direct:
        return direct
    for line in text.replace("\r", "\n").split("\n"):
        decision = _normalize_decision(line)
        if decision:
            return decision
    return None


def _read_json_response(path: Path) -> dict[str, Any] | None:
    if not path.exists():
        return None
    try:
        payload = json.loads(path.read_text(encoding="utf-8-sig"))
    except (OSError, json.JSONDecodeError):
        return None
    raw_decision = payload.get("decision")
    if raw_decision is None and "approved" in payload:
        raw_decision = "approve" if payload.get("approved") is True else "reject"
    decision = _normalize_decision(str(raw_decision or payload.get("answer") or payload.get("instruction") or ""))
    if not decision:
        return None
    return {
        "source": "response_file",
        "decision": decision,
        "response_path": str(path),
        "raw_response": payload,
        "reviewer": payload.get("reviewer") or payload.get("owner") or payload.get("email"),
        "reviewed_samples": payload.get("reviewed_samples") or payload.get("sample_count"),
        "critical_failures": payload.get("critical_failures"),
        "received_at": utc_now(),
    }


def _poll_response_file(run_id: str) -> dict[str, Any] | None:
    for path in _response_paths(run_id):
        response = _read_json_response(path)
        if response:
            return response
    return None


def _poll_imap_response(run_id: str) -> dict[str, Any] | None:
    host = _env_first("SHFT_IMAP_HOST", "SHCG_INBOUND_IMAP_HOST", "SHCG_IMAP_HOST") or _shcg_default("imap.gmail.com")
    username = (
        _env_first(
            "SHFT_IMAP_USERNAME",
            "SHCG_INBOUND_IMAP_USERNAME",
            "SHCG_IMAP_USERNAME",
            "SHCG_ALERT_SMTP_USERNAME",
            "SHCG_SMTP_USERNAME",
            "SHCG_ALERT_SMTP_FROM",
            "SHCG_SMTP_FROM",
        )
        or _shcg_default("david.d.lin@linvest21.com")
    )
    password = _env_first("SHFT_IMAP_PASSWORD", "SHCG_INBOUND_IMAP_PASSWORD")
    if not (host and username and password):
        return None
    port = int(_env_first("SHFT_IMAP_PORT", "SHCG_INBOUND_IMAP_PORT", "SHCG_IMAP_PORT") or _shcg_default("993") or "993")
    mailbox = _env_first("SHFT_IMAP_MAILBOX", "SHCG_INBOUND_IMAP_MAILBOX") or "INBOX"
    try:
        with imaplib.IMAP4_SSL(host, port) as imap:
            imap.login(username, password)
            imap.select(mailbox, readonly=True)
            status, data = imap.search(None, "SUBJECT", f'"{run_id}"')
            if status != "OK" or not data or not data[0]:
                return None
            ids = data[0].split()[-20:]
            for message_id in reversed(ids):
                status, fetched = imap.fetch(message_id, "(RFC822)")
                if status != "OK":
                    continue
                for item in fetched:
                    if not isinstance(item, tuple):
                        continue
                    message = message_from_bytes(item[1])
                    subject = str(message.get("Subject") or "")
                    body = _extract_message_text(message)
                    decision = _find_decision(f"{subject}\n{body}")
                    if decision:
                        return {
                            "source": "imap_email",
                            "decision": decision,
                            "mailbox": mailbox,
                            "message_id": message_id.decode("ascii", errors="replace"),
                            "subject": subject,
                            "received_at": utc_now(),
                        }
    except Exception:
        return None
    return None


def _read_stdin_async(stdin: TextIO, result: dict[str, Any]) -> None:
    try:
        line = stdin.readline()
    except Exception as exc:  # pragma: no cover - defensive only
        result["error"] = str(exc)
        return
    decision = _normalize_decision(line or "")
    if decision:
        result.update({"source": "stdin", "decision": decision, "raw_response": line.strip(), "received_at": utc_now()})


def _review_summary(paired_eval: dict[str, Any]) -> str:
    candidate = paired_eval.get("candidate", {})
    improvement = paired_eval.get("improvement", {})
    return (
        f"Candidate aggregate: {candidate.get('aggregate')}\n"
        f"Candidate critical pass rate: {candidate.get('critical_pass_rate')}\n"
        f"Pairwise win rate: {improvement.get('pairwise_win_rate')}\n"
        f"Pairwise loss rate: {improvement.get('pairwise_loss_rate')}\n"
        f"Wins/ties/losses: {improvement.get('wins')}/{improvement.get('ties')}/{improvement.get('losses')}\n"
    )


def _human_report_from_decision(
    *,
    run_dir: Path,
    decision: dict[str, Any],
    reviewed_samples: int,
    email_record: dict[str, Any],
) -> dict[str, Any]:
    approved = decision.get("decision") == "approve"
    raw_critical = decision.get("critical_failures")
    try:
        critical_failures = int(raw_critical) if raw_critical is not None else (0 if approved else 1)
    except (TypeError, ValueError):
        critical_failures = 0 if approved else 1
    if approved and critical_failures != 0:
        approved = False
        decision = {**decision, "decision": "reject", "rejection_reason": "approval response reported critical failures"}
    report = {
        "schema_version": "shft_human_spot_check_report_v1",
        "run_id": run_dir.name,
        "sample_count": reviewed_samples,
        "reviewed_samples": reviewed_samples,
        "critical_failures": critical_failures,
        "approved": approved,
        "status": "approved" if approved else "rejected",
        "method": "operator_review_email_response_v1",
        "reviewer": decision.get("reviewer") or email_record.get("to"),
        "review_decision": decision,
        "email_request": {
            "to": email_record.get("to"),
            "subject": email_record.get("subject"),
            "delivery_status": email_record.get("delivery_status"),
            "outbox_path": email_record.get("outbox_path"),
        },
        "rationale": "Human spot-check approval was collected from an explicit operator response.",
        "created_at": utc_now(),
    }
    write_json(run_dir / "eval" / "human_spot_check_report.json", report)
    return report


def request_human_spot_check_approval(
    *,
    run_dir: Path,
    release_id: str,
    paired_eval: dict[str, Any],
    timeout_seconds: int | None = None,
    stdin: TextIO | None = None,
    stdout: TextIO | None = None,
    stderr: TextIO | None = None,
) -> dict[str, Any]:
    thresholds = load_model_quality_thresholds().get("human_spot_check", {})
    reviewed_samples = int(thresholds.get("min_reviewed_samples", 10))
    owner = _owner_email()
    timeout = int(timeout_seconds if timeout_seconds is not None else os.environ.get("SHFT_HUMAN_REVIEW_TIMEOUT_SECONDS", "1800"))
    poll_interval = float(os.environ.get("SHFT_HUMAN_REVIEW_POLL_SECONDS", "10"))
    stdin = stdin or sys.stdin
    stdout = stdout or sys.stdout
    stderr = stderr or sys.stderr
    predictions_path = run_dir / "eval" / "paired_predictions.jsonl"
    response_paths = _response_paths(run_dir.name)
    subject = f"SHFT human spot-check approval required: {release_id} {run_dir.name}"
    body = (
        "SHFT requires explicit human spot-check approval before model-quality certification can pass.\n"
        f"Reviewer: {owner}\n"
        f"Run: {run_dir.name}\n"
        f"Release: {release_id}\n"
        f"Required reviewed samples: {reviewed_samples}\n"
        f"Predictions to review: {predictions_path}\n\n"
        f"{_review_summary(paired_eval)}\n"
        "Reply with exactly one word on its own line: approve or reject.\n"
        "Approve only if at least the required samples were reviewed and there are zero critical failures.\n"
        f"Response file options: {', '.join(str(path) for path in response_paths)}\n"
        "Response file JSON example: {\"decision\":\"approve\",\"reviewed_samples\":10,\"critical_failures\":0,\"reviewer\":\"your name\"}\n"
    )
    email_record = _send_email_or_outbox(to_email=owner, subject=subject, body=body, run_id=run_dir.name)
    request_artifact = {
        "schema_version": "shft_human_spot_check_email_request_v1",
        "run_id": run_dir.name,
        "release_id": release_id,
        "owner_email": owner,
        "subject": subject,
        "body": body,
        "response_paths": [str(path) for path in response_paths],
        "email": email_record,
        "timeout_seconds": timeout,
        "created_at": utc_now(),
    }
    write_json(run_dir / "eval" / "human_spot_check_email_request.json", request_artifact)
    delivered = bool(email_record.get("delivered"))
    line = (
        "[SHFT HUMAN REVIEW EMAIL] "
        f"DELIVERED={delivered} STATUS={email_record.get('delivery_status')} "
        f"TO={owner} OUTBOX={email_record.get('outbox_path')}"
    )
    if not delivered:
        line += " ACTION=CHECK_OUTBOX_OR_CONFIGURE_SMTP_IMAP"
    print(line, file=stdout)
    print(line, file=stderr)
    print("[SHFT HUMAN REVIEW ASK]", file=stdout)
    print(body, file=stdout)
    print("[SHFT HUMAN REVIEW ASK]", file=stderr)
    print(body, file=stderr)
    stdout.flush()
    stderr.flush()

    stdin_result: dict[str, Any] = {}
    if getattr(stdin, "isatty", lambda: False)() or os.environ.get("SHFT_HUMAN_REVIEW_READ_STDIN", "true").lower() == "true":
        thread = threading.Thread(target=_read_stdin_async, args=(stdin, stdin_result), daemon=True)
        thread.start()

    deadline = time.time() + max(0, timeout)
    decision: dict[str, Any] | None = None
    while time.time() <= deadline:
        if stdin_result.get("decision"):
            decision = dict(stdin_result)
            break
        file_decision = _poll_response_file(run_dir.name)
        if file_decision:
            decision = file_decision
            break
        imap_decision = _poll_imap_response(run_dir.name)
        if imap_decision:
            decision = imap_decision
            break
        if timeout == 0:
            break
        time.sleep(poll_interval)

    if decision is None:
        decision = {"source": None, "decision": "pending", "received_at": None, "timeout_seconds": timeout}
        report = {
            "schema_version": "shft_human_spot_check_report_v1",
            "run_id": run_dir.name,
            "sample_count": reviewed_samples,
            "reviewed_samples": reviewed_samples,
            "critical_failures": 1,
            "approved": False,
            "status": "pending_human_review",
            "method": "operator_review_email_response_v1",
            "reviewer": owner,
            "review_decision": decision,
            "email_request": {
                "to": email_record.get("to"),
                "subject": email_record.get("subject"),
                "delivery_status": email_record.get("delivery_status"),
                "outbox_path": email_record.get("outbox_path"),
            },
            "rationale": "No explicit human approval response was received before timeout.",
            "created_at": utc_now(),
        }
        write_json(run_dir / "eval" / "human_spot_check_report.json", report)
    else:
        report = _human_report_from_decision(
            run_dir=run_dir,
            decision=decision,
            reviewed_samples=reviewed_samples,
            email_record=email_record,
        )

    artifact = {
        "schema_version": "shft_human_spot_check_approval_v1",
        "run_id": run_dir.name,
        "release_id": release_id,
        "owner_email": owner,
        "email": email_record,
        "decision": decision,
        "human_spot_check_report": report,
        "response_paths": [str(path) for path in response_paths],
        "ok": report.get("approved") is True and int(report.get("critical_failures") or 0) == 0,
        "created_at": utc_now(),
    }
    write_json(run_dir / "eval" / "human_spot_check_approval.json", artifact)
    write_json(_response_root() / f"{run_dir.name}_approval.json", artifact)
    return artifact