user_id = await get_user()
if not user_id:
    return json.dumps({'status': 'error', 'message': '未登录'}, ensure_ascii=False)

rec_id = params_kw.get('id', '')
if not rec_id:
    return json.dumps({'status': 'error', 'message': '缺少id'}, ensure_ascii=False)

try:
    async with get_sor_context(request._run_ns, 'pipeline') as sor:
        rows = await sor.sqlExe("select host, port, user, ssh_key_path from sd_deploy_envs where id=${id}$", {'id': rec_id})
        if not rows:
            return json.dumps({'status': 'error', 'message': '环境记录不存在'}, ensure_ascii=False)
        env = rows[0] if isinstance(rows, list) else rows

    host = env.get('host', '')
    port = env.get('port', '22')
    ssh_user = env.get('user', '')
    ssh_key = env.get('ssh_key_path', '')

    import asyncio
    cmd = f"ssh -o StrictHostKeyChecking=no -o ConnectTimeout=5 -p {port}"
    if ssh_key:
        cmd += f" -i {ssh_key}"
    cmd += f" {ssh_user}@{host} echo ok"

    proc = await asyncio.create_subprocess_shell(cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
    stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=15)

    if proc.returncode == 0:
        return json.dumps({'status': 'ok', 'message': '连接成功'}, ensure_ascii=False)
    else:
        return json.dumps({'status': 'error', 'message': f'连接失败: {stderr.decode().strip()}'}, ensure_ascii=False)
except Exception as e:
    return json.dumps({'status': 'error', 'message': str(e)}, ensure_ascii=False)