#!/usr/bin/env python3
"""Nightly advance orchestrator.

Executes the nightly pipeline against the main repo:

  Phase 0  preflight    daemon health, reentry lock, dev tip freeze
  Phase 1  plan         scripts/nightly/plan.py ordering
  Phase 2  advance      per-task headless cursor-agent runs (bounded)
  Phase 3  integrate    daemon ff-only via queue enqueue (already done inside
                        phase 2 per task); wait for queue drain
  Phase 4  dev-service  rebuild dev env when tip moved; /health /ready
  Phase 5  cleanup      verify + remove finished worktrees (normal remove
                        only; refusals leave cleanup=pending, never --force)
  Phase 6  report       .doc/test-reports/<date>-nightly-advance-test-report.md
                        draft written to run dir (committed by report task)

Every phase records structured state into the run dir (data/nightly/<run>/),
so a crashed run is inspectable and never double-drives: a live lock file
with a healthy heartbeat aborts the new invocation.

Exit codes: 0 success, 1 partial failures, 2 usage, 3 fatal infrastructure.
"""

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

DEFAULT_REPO = "/home/weichen/projects/Agent-System"
# data/ is root-owned (docker mounts); .tmp/ is gitignored and user-writable.
RUN_ROOT = ".tmp/nightly"
LOCK_REL = f"{RUN_ROOT}/.lock"
# Standing task identity for the perpetual automation (env leases survive the
# delivery task closeout). Override with NIGHTLY_TASK_ID if retasked.
STANDING_TASK = os.environ.get("NIGHTLY_TASK_ID", "nightly-advance-automation")
HEARTBEAT_STALE_SECONDS = 3600
REPORT_SLUG = "nightly-advance"

PROMPT_ADVANCE = """你是 Agent-System 夜间自动化执行器。任务：推进 worktree {worktree} 中的分支任务 {task_id} 至完成。

硬边界（违反即失败）：
1. 只在 {worktree} 内工作，不碰其他 worktree 或主目录 dev。
2. 任何 git 写操作（add/commit）必须包在 `python3 ../../scripts/agentctl.py with --task {task_id} --resource path:<路径> --resource git:index -- bash -lc '<命令>'` 内。
3. 读取该 worktree 下的任务文档（.doc/ops/short-term/ 或 .doc/audit/ 下与任务 id 匹配的文件）了解剩余工作与验收标准。
4. 按任务文档推进实现/修复/文档工作；禁止 push、禁止晋升 test/main、禁止操作其他环境。
5. 按验证契约选择验收：工程工具/文档类 → `python3 -B -m unittest discover -s scripts/tests -p 'test_*.py'`；产品代码改动 → 本 worktree 内可行的真实检查（go build / go test 相关包）。
6. 完成后：任务状态五字段填入任务文档（review_state/integration_state=validated_cddidate_ready/deployment/cleanup/remote_sync），不要自行 integrate。
7. 所有命令真实执行并保留输出；禁止用 SKIP/stub 冒充通过。

完成时输出 JSON 摘要（最后一行）：{{"task_id":"{task_id}","result":"advanced|done|blocked|failed","summary":"<一句话>","evidence":"<相对路径>"}}"""

PROMPT_INTEGRATE = """你是 Agent-System 夜间自动化执行器。任务：把集成候选 {task_id} 合入 dev（通过治理门禁）。

步骤：
1. 在 {worktree} 内检查 `git status --porcelain` 是否干净、`git rev-parse HEAD` 是否为冻结候选。
2. 若树不干净：停止并输出 blocked（不要 stash/reset）。
3. 若干净：构造候选收口——执行 `python3 ../../scripts/agentctl.py validate --task {task_id} -- python3 -B -m unittest discover -s scripts/tests -p 'test_*.py'`（治理类变更）；产品变更按任务文档已冻结的验收命令。
4. validate 通过后：`python3 ../../scripts/agentctl.py queue enqueue --task {task_id} --expected-dev {expected_dev} --head $(git rev-parse HEAD)`。daemon 会 ff-only 合入。
5. 轮询 `queue list` 至空或该 task 消失（最多 60s）。
6. 禁止手动 git merge dev、禁止 push、禁止动 test/main。

陈旧候选恢复（仅当 validate 因 TASK_CLOSED 拒绝，或候选第一父不等于当前 dev）：
- 先判断功能内容是否已经在 dev 上（`git merge-base --is-ancestor <候选head或其功能分支> dev`，或对比 dev log）。内容已在 dev → 输出 result=merged, summary 说明由哪个提交带入，不重建。
- 内容不在 dev 且任务已 cancelled/done → 输出 blocked 并说明需要人工重开任务；禁止复活关闭的任务。
- 内容不在 dev 且任务仍 active 但基线前移 → 在 worktree 内用 `python3 ../../scripts/agentctl.py integrate prepare --task <基础功能task_id> --expected-dev <当前dev> --integrate-id {task_id}-r<N+1>` 从新 dev 重建候选（N 为现有最大轮次），对返回的新 worktree 重复步骤 1-5。

完成时输出 JSON 摘要（最后一行）：{{"task_id":"{task_id}","result":"enqueued|merged|blocked|failed","summary":"<一句话>","new_dev_tip":"<sha 或 null>"}}"""

PROMPT_REPORT = """你是 Agent-System 夜间自动化日报执行器。任务：把 {date} 夜间运行的报告草稿正式化并归档。

步骤：
1. 读 {run_dir}/report-draft.md、advance.json、env.json、cleanup.json、state.json。
2. 在当前 worktree（报告任务工作区）把草稿改写为正式报告 .doc/test-reports/{date}-nightly-advance-test-report.md：
   - 按 test-reports-standard 模板补全七大板块；引用 run 目录真实数据（不得虚构）。
   - 每个任务的 result/耗时/证据日志相对路径如实填入「实际测试结果与判定」。
   - blocked/failed 任务在「测试结论与后续」列出后续建议（人工裁决点）。
3. 用治理租约提交（禁止裸 git）：
   python3 ../../scripts/agentctl.py with --task {report_task_id} --resource path:.doc --resource git:index -- bash -lc 'git add .doc/test-reports/{date}-nightly-advance-test-report.md && git commit -m "docs(test): nightly advance report {date}"'
4. 不 integrate、不 push；提交留在报告任务分支即可。

完成时输出 JSON 摘要（最后一行）：{{"date":"{date}","result":"archived|failed","report":".doc/test-reports/{date}-nightly-advance-test-report.md","commit":"<sha 或 null>"}}"""


def sh(args: list[str], cwd: Path, env: dict | None = None, timeout: int = 600) -> tuple[int, str]:
    """Run a subprocess; returns (rc, combined output). Never raises."""
    e = dict(os.environ)
    if env:
        e.update(env)
    try:
        p = subprocess.run(
            args, cwd=str(cwd), capture_output=True, text=True, timeout=timeout, env=e
        )
        out = p.stdout + ("\n--stderr--\n" + p.stderr if p.stderr else "")
        return p.returncode, out
    except subprocess.TimeoutExpired as exc:
        return 124, f"TIMEOUT after {timeout}s\n{(exc.stdout or '')[-800:]}"
    except Exception as exc:  # noqa: BLE001 — orchestrator must not die
        return 125, f"ERROR {exc}"


def agentctl(repo: Path, *args: str, timeout: int = 600) -> tuple[int, dict | str]:
    rc, out = sh([sys.executable, "-B", str(repo / "scripts/agentctl.py"), *args], repo, timeout=timeout)
    clean = "\n".join(ln for ln in out.splitlines() if not ln.startswith("[notice]"))
    try:
        return rc, json.loads(clean.strip() or "{}")
    except json.JSONDecodeError:
        return rc, out


def write_json(path: Path, data) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def now_utc() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


# ---------------------------------------------------------------- phase 0

def preflight(repo: Path, run_dir: Path) -> tuple[bool, list[str], dict]:
    """Reentry lock + daemon health + dev tip freeze. Returns (ok, errors, ctx)."""
    errors: list[str] = []
    ctx: dict = {"started_at": now_utc()}

    # lock: atomic create; stale after HEARTBEAT_STALE_SECONDS without update
    lock = repo / LOCK_REL
    lock.parent.mkdir(parents=True, exist_ok=True)
    try:
        fd = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
        os.write(fd, json.dumps({"pid": os.getpid(), "started": ctx["started_at"]}).encode())
        os.close(fd)
    except FileExistsError:
        age = time.time() - lock.stat().st_mtime
        if age > HEARTBEAT_STALE_SECONDS:
            lock.unlink()
            fd = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
            os.write(fd, json.dumps({"pid": os.getpid(), "took_over_after_stale": age}).encode())
            os.close(fd)
        else:
            return False, [f"another nightly run is active (lock age {age:.0f}s < {HEARTBEAT_STALE_SECONDS}s)"], ctx

    rc, daemon = agentctl(repo, "daemon", "status")
    if rc != 0 or (isinstance(daemon, dict) and daemon.get("status") != "ok"):
        return False, [f"daemon not healthy: {str(daemon)[:200]}"], ctx
    ctx["daemon"] = "ok"

    rc, tip = sh(["git", "-C", str(repo), "rev-parse", "HEAD"], repo)
    if rc != 0:
        return False, ["cannot resolve dev tip"], ctx
    ctx["dev_tip_start"] = tip.strip()
    return True, errors, ctx


def touch_heartbeat(repo: Path) -> None:
    lock = repo / LOCK_REL
    try:
        os.utime(lock, None)
    except OSError:
        pass


# ---------------------------------------------------------------- phase 1

def build_plan(repo: Path, run_dir: Path) -> dict:
    # Locate plan.py next to this file first; fall back to <repo>/scripts so
    # the orchestrator works both from a task worktree and from merged dev.
    here = Path(__file__).resolve().parent
    plan_py = here / "plan.py"
    if not plan_py.exists():
        plan_py = repo / "scripts/nightly/plan.py"
    out_file = run_dir / "plan.json"
    rc, out = sh(
        [sys.executable, "-B", str(plan_py), "--repo", str(repo), "--out", str(out_file)],
        repo,
    )
    if rc != 0 or not out_file.exists():
        return {"error": out[-500:], "entries": []}
    return json.loads(out_file.read_text(encoding="utf-8"))


# ---------------------------------------------------------------- phase 2

def run_agent(prompt: str, worktree: Path, timeout_s: int, run_dir: Path, tag: str) -> dict:
    """Headless cursor-agent run inside a worktree; full log kept in run_dir."""
    log = run_dir / f"agent-{tag}.log"
    cmd = [
        "cursor-agent", "--print", "--trust",
        "--workspace", str(worktree),
        prompt,
    ]
    env = {"PATH": os.environ.get("PATH", "") + ":/usr/bin:/home/linuxbrew/.linuxbrew/bin"}
    start = time.monotonic()
    rc, out = sh(cmd, cwd=worktree, env=env, timeout=timeout_s)
    dur = time.monotonic() - start
    log.write_text(out, encoding="utf-8", errors="replace")
    summary = {"rc": rc, "duration_s": round(dur, 1), "log": str(log)}
    # parse last JSON line the agent was asked to emit
    for line in reversed(out.strip().splitlines()):
        line = line.strip().strip("`")
        if line.startswith("{") and line.endswith("}"):
            try:
                summary["agent"] = json.loads(line)
                break
            except json.JSONDecodeError:
                continue
    return summary


# ---------------------------------------------------------------- phase 4

def dev_tip(repo: Path) -> str | None:
    rc, out = sh(["git", "-C", str(repo), "rev-parse", "HEAD"], repo)
    return out.strip() if rc == 0 else None


def update_dev_service(repo: Path, run_dir: Path) -> dict:
    """Rebuild dev env only when dev tip moved; verify /health and /ready."""
    tip = dev_tip(repo)
    result: dict = {"dev_tip": tip}
    if tip is None:
        result["status"] = "failed"
        result["error"] = "cannot resolve dev tip"
        return result
    start = tip == run_dir_state(run_dir).get("dev_tip_start")
    if start:
        result["status"] = "not_required"
        result["reason"] = "dev tip unchanged tonight"
        return result
    # env:dev write goes through agentctl with on the standing nightly task
    # lease. The daemon itself never touches env:*, so we drive env-service.sh
    # here.
    rc, out = sh(
        [sys.executable, "-B", str(repo / "scripts/agentctl.py"), "with",
         "--task", STANDING_TASK,
         "--resource", "env:dev", "--cwd", str(repo),
         "--", str(repo / "scripts/env-service.sh"), "dev", "rebuild"],
        repo, timeout=3600,
    )
    log = run_dir / "env-rebuild.log"
    log.write_text(out, encoding="utf-8", errors="replace")
    result["rebuild_rc"] = rc
    result["rebuild_log"] = str(log)
    if rc != 0:
        result["status"] = "failed"
        return result
    ok_h, body_h = http_get("http://127.0.0.1:8080/health")
    ok_r, body_r = http_get("http://127.0.0.1:8080/ready")
    result["health"] = {"ok": ok_h, "body": (body_h or "")[:300]}
    result["ready"] = {"ok": ok_r, "body": (body_r or "")[:300]}
    result["status"] = "ok" if (ok_h and ok_r) else "unhealthy"
    return result


def http_get(url: str, retries: int = 6, delay: float = 10.0) -> tuple[bool, str | None]:
    for i in range(retries):
        rc, out = sh(["curl", "-fsS", "--max-time", "10", url], Path("/tmp"), timeout=30)
        if rc == 0:
            return True, out
        time.sleep(delay if i else 2)
    return False, None


def run_dir_state(run_dir: Path) -> dict:
    f = run_dir / "state.json"
    if f.exists():
        try:
            return json.loads(f.read_text(encoding="utf-8"))
        except json.JSONDecodeError:
            return {}
    return {}


# ---------------------------------------------------------------- phase 5

def cleanup_pass(repo: Path, run_dir: Path, plan: dict) -> dict:
    """Verify + remove finished worktrees. Normal remove only; refusals
    leave cleanup=pending recorded in the report — never --force."""
    results = []
    for entry in plan.get("entries", []):
        if entry.get("phase") != "cleanup_verify":
            continue
        wt = Path(entry["worktree"])
        task_id = entry["task_id"]
        checks: dict = {"task_id": task_id, "worktree": str(wt)}
        # 1) task closed (not active in registry)
        rc, st = agentctl(repo, "status")
        reg = {}
        if isinstance(st, dict):
            raw = st.get("tasks") or {}
            if isinstance(raw, list):
                reg = {t.get("id", ""): t for t in raw if isinstance(t, dict)}
            elif isinstance(raw, dict):
                reg = raw
        tstate = reg.get(task_id, {}).get("state", "absent")
        checks["registry_state"] = tstate
        # 2) branch head is dev ancestor (content already on dev)
        rc, out = sh(["git", "-C", str(repo), "merge-base", "--is-ancestor",
                      entry.get("branch", ""), "dev"], repo)
        checks["ancestor_of_dev"] = rc == 0
        # 3) worktree clean of tracked/untracked (ignored ok)
        rc, out = sh(["git", "-C", str(wt), "status", "--porcelain"], wt)
        lines = [ln for ln in out.splitlines() if ln.strip() and not ln.startswith("!!")]
        checks["clean"] = not lines
        checks["dirty_sample"] = lines[:10]
        if tstate not in {"done", "cancelled", "absent"}:
            checks["verdict"] = "skip_task_active"
        elif not checks["ancestor_of_dev"]:
            checks["verdict"] = "skip_not_on_dev"
        elif not checks["clean"]:
            checks["verdict"] = "skip_dirty"
        else:
            rc, out = sh(
                [str(repo / "scripts/agent-worktree.sh"), "remove",
                 entry["branch"], "--delete-branch"],
                repo, timeout=600,
            )
            checks["remove_rc"] = rc
            checks["remove_out"] = out[-400:]
            checks["verdict"] = "removed" if rc == 0 else "remove_refused_keep_pending"
        results.append(checks)
    write_json(run_dir / "cleanup.json", {"results": results})
    return {"results": results}


# ---------------------------------------------------------------- phase 6

def write_report(repo: Path, run_dir: Path, ctx: dict, plan: dict, advance: list, envs: dict, cleanup: dict) -> Path:
    date = ctx["started_at"][:10]
    lines = []
    lines.append(f"# 夜间推进自动化测试报告 {date}")
    lines.append("")
    lines.append("## 一、 测试背景")
    lines.append("每日 03:10 定时推进 .worktrees 分支任务、合入 dev、更新 dev 服务并清理已收口目录的自动化链路。")
    lines.append("")
    lines.append("## 二、 测试目标")
    lines.append("验证夜间流水线可无人值守执行：计划构建 → 任务推进 → dev 合入 → 服务更新 → 完成目录清理 → 报告落盘。")
    lines.append("")
    lines.append("## 三、 测试方案与环境")
    lines.append(f"- 主仓：{repo}（dev）")
    lines.append("- 触发：cron 10 3 * * * → scripts/nightly-advance.sh")
    lines.append(f"- 执行器：cursor-agent 无头模式；agentctl daemon（coordinator.service）")
    lines.append(f"- run 目录：{run_dir}")
    lines.append("")
    lines.append("## 四、 预期结果")
    lines.append("活跃任务按计划推进/合入或带原因 blocked；dev tip 变化时服务重建并通过 /health /ready；完成目录经无内容丢失验证后删除；每日报告归档 .doc/test-reports/。")
    lines.append("")
    lines.append("## 五、 实际测试结果与判定")
    lines.append(f"- 开始：{ctx.get('started_at')}；dev 起点 {ctx.get('dev_tip_start','?')[:12]}")
    tip = ctx.get("dev_tip_end") or dev_tip(repo)
    lines.append(f"- 结束：{now_utc()}；dev 终点 {(tip or '?')[:12]}")
    moved = tip != ctx.get("dev_tip_start")
    lines.append(f"- dev 变更：{'是' if moved else '否'}")
    lines.append("")
    lines.append("### 任务推进明细")
    lines.append("| task | phase | result | 摘要 | 证据 |")
    lines.append("|---|---|---|---|---|")
    for a in advance:
        ag = a.get("agent") or {}
        ev = ag.get("evidence") or a.get("log", "")
        lines.append(f"| {a.get('task_id')} | {a.get('phase')} | {ag.get('result', 'no_summary')} | {(ag.get('summary') or '')[:80]} | {ev} |")
    lines.append("")
    lines.append("### dev 服务")
    lines.append(f"状态：{envs.get('status')}")
    if envs.get("health"):
        lines.append(f"- /health ok={envs['health'].get('ok')} body={envs['health'].get('body','')[:120]}")
        lines.append(f"- /ready ok={envs['ready'].get('ok')} body={envs['ready'].get('body','')[:120]}")
    lines.append("")
    lines.append("### 清理（已合入并验证无内容丢失后删除）")
    lines.append("| worktree | 判定 | 原因 |")
    lines.append("|---|---|---|")
    for c in cleanup.get("results", []):
        lines.append(f"| {Path(c['worktree']).name} | {c['verdict']} | {c.get('remove_out','')[:100].replace('|','/')} |")
    lines.append("")
    lines.append("## 六、 自动化测试套件汇总")
    lines.append(f"- plan 构造：{plan.get('count', 0)} 项")
    failed = [a for a in advance if (a.get("agent") or {}).get("result") in {"failed", "blocked"}]
    lines.append(f"- 任务推进：{len(advance)} 项，失败/受阻 {len(failed)} 项")
    lines.append("")
    lines.append("## 七、 测试结论与后续")
    overall = "PASS" if not failed and envs.get("status") in {"ok", "not_required"} else "PARTIAL"
    lines.append(f"整体判定：{overall}（细节见 run 目录原始日志）")
    lines.append("")
    report_path = run_dir / "report-draft.md"
    report_path.write_text("\n".join(lines), encoding="utf-8")
    return report_path


# ---------------------------------------------------------------- main

def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--repo", default=DEFAULT_REPO)
    ap.add_argument("--max-tasks", type=int, default=0)
    ap.add_argument("--task-timeout", type=int, default=3600)
    ap.add_argument("--skip-agent", action="store_true", help="dry: plan+cleanup only")
    ap.add_argument("--only-report", action="store_true", help="dry: reuse existing run dir state")
    args = ap.parse_args()

    repo = Path(args.repo).resolve()
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    run_dir = repo / RUN_ROOT / stamp
    run_dir.mkdir(parents=True, exist_ok=True)

    ok, errors, ctx = preflight(repo, run_dir)
    write_json(run_dir / "state.json", ctx)
    if not ok:
        for e in errors:
            print(f"FATAL: {e}", file=sys.stderr)
        return 3
    touch_heartbeat(repo)

    plan = build_plan(repo, run_dir)
    ctx["plan_count"] = plan.get("count", 0)
    write_json(run_dir / "state.json", ctx)

    advance: list[dict] = []
    if not args.skip_agent:
        entries = plan.get("entries", [])
        if args.max_tasks > 0:
            entries = entries[: args.max_tasks]
        for i, entry in enumerate(entries):
            if entry.get("phase") == "cleanup_verify":
                continue  # handled in phase 5
            touch_heartbeat(repo)
            tag = f"{i:02d}-{entry['task_id']}"
            worktree = Path(entry["worktree"])
            prompt = (
                PROMPT_INTEGRATE.format(
                    task_id=entry["task_id"], worktree=worktree,
                    expected_dev=ctx["dev_tip_start"],
                )
                if entry.get("phase") == "integrate_closeout"
                else PROMPT_ADVANCE.format(task_id=entry["task_id"], worktree=worktree)
            )
            res = run_agent(prompt, worktree, args.task_timeout, run_dir, tag)
            res["task_id"] = entry["task_id"]
            res["phase"] = entry.get("phase")
            res["branch"] = entry.get("branch")
            res["owner"] = entry.get("owner")
            advance.append(res)
            write_json(run_dir / "advance.json", advance)

    # phase 3: wait for daemon to drain the queue
    for _ in range(30):
        rc, q = agentctl(repo, "queue", "list")
        if isinstance(q, dict) and q.get("items") == []:
            break
        touch_heartbeat(repo)
        time.sleep(10)

    ctx["dev_tip_end"] = dev_tip(repo)
    write_json(run_dir / "state.json", ctx)

    # phase 4
    envs = update_dev_service(repo, run_dir)
    write_json(run_dir / "env.json", envs)

    # phase 5
    cleanup = cleanup_pass(repo, run_dir, plan)
    ctx["dev_tip_final"] = dev_tip(repo)
    write_json(run_dir / "state.json", ctx)

    # phase 6
    report = write_report(repo, run_dir, ctx, plan, advance, envs, cleanup)

    # phase 6.5: formalize the daily report into .doc/test-reports/ via a
    # headless agent inside the report worktree (lease-wrapped commit).
    report_archived = None
    if not args.skip_agent:
        report_task_wt = Path(os.environ.get(
            "NIGHTLY_REPORT_WORKTREE",
            str(repo / ".worktrees/nightly-advance-automation-2026-09-28"),
        ))
        if report_task_wt.exists():
            date = ctx["started_at"][:10]
            report_task_id = os.environ.get("NIGHTLY_REPORT_TASK", STANDING_TASK)
            prompt = PROMPT_REPORT.format(
                date=date, run_dir=run_dir, report_task_id=report_task_id
            )
            res = run_agent(
                prompt, report_task_wt, min(args.task_timeout, 1800), run_dir, "report"
            )
            report_archived = res
            write_json(run_dir / "report-archive.json", res)

    print(json.dumps({
        "run_dir": str(run_dir),
        "report_draft": str(report),
        "report_archived": bool((report_archived or {}).get("agent")),
        "plan_count": plan.get("count", 0),
        "advanced": len(advance),
        "env_status": envs.get("status"),
    }, ensure_ascii=False))
    (repo / LOCK_REL).unlink(missing_ok=True)
    return 0


if __name__ == "__main__":
    sys.exit(main())