Function api_start_selfcheck_run

src/web/routes/selfcheck.py:236–306

(request: StartSelfCheckRequest)

@router.post("/runs")
def api_start_selfcheck_run(request: StartSelfCheckRequest):
    mode = _normalize_mode(request.mode)
    source = str(request.source or "manual").strip().lower() or "manual"

    if has_running_selfcheck_run():
        running = _build_running_run_payload()
        raise HTTPException(
            status_code=409,
            detail={
                "message": "已有运行中的自检任务，请稍后再试",
                "running_run": running,
            },
        )

    if request.run_async:
        latest = _start_selfcheck_background(mode, source)
        return {
            "success": True,
            "message": "自检任务已创建并开始执行",
            "run": latest,
        }

    run = create_selfcheck_run(mode=mode, source=source)
    run_id = int(run["id"])
    task_id = _selfcheck_task_id(run_id)
    task_manager.register_domain_task(
        domain="selfcheck",
        task_id=task_id,
        task_type="selfcheck_run",
        payload={"run_id": run_id, "mode": mode, "source": source},
        progress={"completed": 0, "total": 0},
        max_retries=3,
    )
    acquired, running, quota = task_manager.try_acquire_domain_slot("selfcheck", task_id)
    if not acquired:
        reason = f"并发配额已满（running={running}, quota={quota}）"
        task_manager.update_domain_task(
            "selfcheck",
            task_id,
            status="failed",
            finished_at=datetime.utcnow().isoformat(),
            message=reason,
            error=reason,
        )
        raise HTTPException(status_code=429, detail=reason)
    try:
        result = execute_selfcheck_run(
            run_id,
            mode=mode,
            source=source,
            cancel_checker=lambda: task_manager.is_domain_task_cancel_requested("selfcheck", task_id),
        )
        status_text = str(result.get("status") or "").strip().lower()
        mapped_status = status_text if status_text in {"completed", "failed", "cancelled"} else "completed"
        task_manager.update_domain_task(
            "selfcheck",
            task_id,
            status=mapped_status,