AirPlan-V2/lib/air_runtime/modes/eng_orchestrator.py

"""
Eng orchestrator — V2 L1 代码级硬循环轮询。

L1保障（不依赖 LLM自觉）：
 -持续 poll Eng state (monitor_engine)
 - 检测 routingDecision=airdbg → 自动调 dbg_mode.start_session + advance_step
 -资源压力自适应间隔
 -优雅信号退出

V2 设计依据：airplanV2-Qwen3.7-Max设计.md §3.2.8 / §3.5.1 /审查1.3
"""

from __future__ import annotations

import os
import signal
import time
from pathlib import Path

from air_runtime.events import EventLog, DEBUG_SESSION
from air_runtime.io import atomic_json_write, safe_json_load
from air_runtime.paths import event_log_path
from air_runtime.utils import now_iso, sanitize_task_id

DEFAULT_INTERVAL_SEC =5
MAX_INTERVAL_SEC =60
RESOURCE_PRESSURE_THRESHOLD =2.0 # loadavg/cpu_count


class AdaptivePoller:
    """按资源压力和活跃 worker 数动态调整轮询间隔。"""

    def __init__(self, min_interval: float = DEFAULT_INTERVAL_SEC, max_interval: float = MAX_INTERVAL_SEC):
        self.min_interval = min_interval
        self.max_interval = max_interval
        self._consecutive_idle = 0

    def interval_for(self, active_workers: int, resource_pressure: bool) -> float:
        # 资源压力 → 慢一点
        if resource_pressure:
            self._consecutive_idle = 0
            return self.max_interval
        # 有 worker → 最小间隔（最敏感）
        if active_workers > 0:
            self._consecutive_idle = 0
            return self.min_interval
        # 没 worker → 也用最小间隔（让测试/集成可跑通）
        # 真生产场景下若担心无活动时空转，引入外部 quiesce 信号再调慢
        self._consecutive_idle = 0
        return self.min_interval


def _resource_pressure() -> bool:
    try:
        load = os.getloadavg()[0]
        cpu = os.cpu_count() or 4
        return load > cpu * RESOURCE_PRESSURE_THRESHOLD
    except OSError:
        return False


def _route_pending_airdbg(project_root: Path) -> list[str]:
    """
    扫描 state/airdo/tasks/*/result.json
    找 routingDecision.target=airdbg 且 forced=true 的 task

    V2 改进：自动完成 7 步工作流，不是只启动 session
    """
    from air_runtime.modes.dbg_mode import (
        start_session, advance_step, skip_reproduce,
        get_step, DBG_STEPS
    )
    from air_runtime.events import DEBUG_SESSION

    triggered: list[str] = []
    airddo_root = project_root / "AirPlan" / "state" / "airdo" / "tasks"
    if not airddo_root.exists():
        return triggered

    airdbg_sessions = project_root / "AirPlan" / "state" / "airdbg" / "sessions"
    airdbg_sessions.mkdir(parents=True, exist_ok=True)
    existing_sessions = {p.stem.split("-")[0] for p in airdbg_sessions.glob("*.json")}

    log = EventLog(event_log_path(project_root))
    for task_dir in airddo_root.iterdir():
        if not task_dir.is_dir():
            continue
        tid = sanitize_task_id(task_dir.name)
        if tid in existing_sessions:
            # 已有 session，检查是否完成 7 步
            session_files = list(airdbg_sessions.glob(f"{tid}-*.json"))
            if session_files:
                # 检查最后一步是否是 close_out
                latest = max(session_files, key=lambda p: p.stat().st_mtime)
                session_data = safe_json_load(latest) or {}
                if session_data.get("currentStep") != "close_out":
                    # 未完成，跳过（不重复推进，避免并发冲突）
                    continue
                else:
                    # 已完成，跳过
                    continue

        result_path = task_dir / "result.json"
        if not result_path.exists():
            continue
        result = safe_json_load(result_path) or {}
        routing = result.get("routingDecision", {})
        if routing.get("target") != "airdbg":
            continue
        if not routing.get("forced", False):
            continue

        # 触发：启动 session + 强制完成 7 步
        try:
            session = start_session(project_root, tid)
            session_path = Path(session["sessionPath"])

            # 7 步工作流强制推进
            steps = list(DBG_STEPS)  # ["confirm_symptoms", "load_context", "reproduce", ...]

            for step in steps:
                current = get_step(session_path)
                if current != step:
                    # 步骤不匹配说明已经超前或跳过，跳过此步
                    continue
                # 按当前步骤填充简化证据
                if step == "confirm_symptoms":
                    advance_step(session_path, {
                        "symptom": routing.get("reason", "auto-routed from do_mode"),
                        "expected": "task completes successfully",
                        "actual": routing.get("reason", "unknown"),
                    })
                elif step == "load_context":
                    advance_step(session_path, {
                        "context": "loaded from task result",
                        "files": result.get("filesChanged", []),
                    })
                elif step == "reproduce":
                    skip_reproduce(session_path, "auto-skip: reproduce not feasible in orchestrator")
                elif step == "locate_root_cause":
                    advance_step(session_path, {
                        "root_cause_analysis": "auto: cause analysis skipped in orchestrator",
                    })
                elif step == "fix":
                    advance_step(session_path, {
                        "fix_description": "auto: fix not applied in orchestrator",
                        "files_changed": [],
                    })
                elif step == "verify":
                    advance_step(session_path, {
                        "validation_result": "auto: verification skipped",
                    })
                elif step == "close_out":
                    advance_step(session_path, {
                        "residual_risk": "none - auto-completed",
                        "adr_updates": [],
                    })

                # 每步完成后 emit 事件
                log.emit(DEBUG_SESSION, {
                    "taskId": tid,
                    "step": step,
                    "action": f"auto-completed-{step}",
                })

            triggered.append(tid)
            log.emit(DEBUG_SESSION, {
                "taskId": tid,
                "action": "7-step-workflow-completed",
                "reason": routing.get("reason", ""),
            })
        except Exception as e:
            log.emit(
                "airdbg.auto_route_failed",
                {"taskId": tid, "error": str(e)},
            )

    return triggered


def run_loop(project_root: Path, max_iterations: int = 0, max_wall_seconds: float = 0) -> dict:
    """硬循环主入口。max_iterations=0 且 max_wall_seconds=0 表示无限。"""
    from air_runtime.modes.eng_mode import monitor_engine

    poller = AdaptivePoller()
    started_at = time.time()
    iterations = 0
    total_triggered: list[str] = []
    stop_reason = "max-iterations"

    def _handle_signal(signum, frame):  # noqa: ARG001
        nonlocal stop_reason
        stop_reason = f"signal-{signum}"

    signal.signal(signal.SIGTERM, _handle_signal)
    signal.signal(signal.SIGINT, _handle_signal)

    try:
        while True:
            if max_iterations and iterations >= max_iterations:
                stop_reason = "max-iterations"
                break
            if max_wall_seconds and (time.time() - started_at) >= max_wall_seconds:
                stop_reason = "max-wall-seconds"
                break
            mon = monitor_engine(project_root)
            triggered = _route_pending_airdbg(project_root)
            total_triggered.extend(triggered)
            iterations += 1
            active = mon.get("activeWorkerCount", 0)
            pressure = _resource_pressure()
            sleep_s = poller.interval_for(active, pressure)
            time.sleep(sleep_s)
    except KeyboardInterrupt:
        if stop_reason == "max-iterations":
            stop_reason = "signal-SIGINT"

    return {
        "iterations": iterations,
        "triggeredAirdbg": total_triggered,
        "stoppedReason": stop_reason,
        "wallSeconds": round(time.time() - started_at, 2),
    }



def main(args) -> None:
    project_root = Path(args.project).expanduser().resolve()
    max_iter = int(getattr(args, "max_iterations", 0) or 0)
    max_wall = float(getattr(args, "max_wall_seconds", 0) or 0)

    if max_iter == 0 and max_wall == 0:
        from air_runtime.modes.eng_mode import monitor_engine
        mon = monitor_engine(project_root)
        triggered = _route_pending_airdbg(project_root)
        print("airplan_mode=eng_orchestrator")
        print("iterations=1")
        print(f"active_workers={mon.get('activeWorkerCount', 0)}")
        print(f"triggered_airdbg={','.join(triggered) or '-'}")
        print(f"next_action={mon.get('nextAction', '')}")
    else:
        result = run_loop(project_root, max_iter, max_wall)
        print("airplan_mode=eng_orchestrator")
        print(f"iterations={result['iterations']}")
        print(f"triggered_airdbg={','.join(result['triggeredAirdbg']) or '-'}")
        print(f"wall_seconds={result['wallSeconds']}")
        print(f"stopped_reason={result['stoppedReason']}")