diff --git a/android/TradeBotMonitor/app/build.gradle.kts b/android/TradeBotMonitor/app/build.gradle.kts index 9950474..a9746ed 100644 --- a/android/TradeBotMonitor/app/build.gradle.kts +++ b/android/TradeBotMonitor/app/build.gradle.kts @@ -10,7 +10,7 @@ android { applicationId = "xyz.kusoft.tradebotmonitor" minSdk = 26 targetSdk = 36 - versionCode = 7 - versionName = "0.2.4" + versionCode = 8 + versionName = "0.2.5" } } diff --git a/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/MainActivity.kt b/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/MainActivity.kt index 86f60cb..3d431a0 100644 --- a/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/MainActivity.kt +++ b/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/MainActivity.kt @@ -575,12 +575,26 @@ class MainActivity : Activity() { private fun trainingComputerPanel(retrain: JSONObject): View = LinearLayout(this).apply { val coordination = retrain.optJSONObject("coordination") ?: JSONObject() - val agentOnline = coordination.optBoolean("agent_online", false) + val activeJob = coordination.optJSONObject("active_job") + val agentRecentlySeen = coordination.optBoolean( + "agent_recently_seen", + coordination.optBoolean("agent_online", false), + ) + val agentBusy = coordination.optBoolean( + "agent_busy", + activeJob?.optStringClean("status") == "running", + ) + val connectionText = when { + agentRecentlySeen -> "онлайн" + agentBusy -> "занят обучением" + else -> "нет связи" + } + val connectionColor = if (agentRecentlySeen || agentBusy) palette.green else palette.amber orientation = LinearLayout.VERTICAL addView(text("Компьютер обучения", 12f, Typeface.NORMAL, palette.muted)) addView(text(prefs.trainingComputerName, 18f, Typeface.BOLD, palette.green).top(dp(5))) addView(text(prefs.trainingComputerPath, 12f, Typeface.NORMAL, palette.muted).top(dp(4))) - addView(keyValueLine("Связь агента", if (agentOnline) "онлайн" else "нет связи", if (agentOnline) palette.green else palette.amber).top(dp(4))) + addView(keyValueLine("Связь агента", connectionText, connectionColor).top(dp(4))) addView(text("Бот ставит задания через tb.kusoft.xyz, а этот Windows-agent сам забирает их через интернет и возвращает результат.", 12f, Typeface.NORMAL, palette.muted).top(dp(8))) } diff --git a/crypto_spot_bot/training_coordination.py b/crypto_spot_bot/training_coordination.py index c58998f..0a78eda 100644 --- a/crypto_spot_bot/training_coordination.py +++ b/crypto_spot_bot/training_coordination.py @@ -206,9 +206,18 @@ class TrainingCoordinator: last_seen = _parse_time(str(worker.get("last_seen_at") or "")) active = self._active_job(state) latest = _latest_job(state) + recently_seen = bool(last_seen and datetime.now(UTC) - last_seen <= ONLINE_WINDOW) + agent_busy = bool( + active + and active.get("status") == "running" + and worker + and active.get("claimed_by") == worker.get("id") + ) return { "available": True, - "agent_online": bool(last_seen and datetime.now(UTC) - last_seen <= ONLINE_WINDOW), + "agent_online": recently_seen or agent_busy, + "agent_recently_seen": recently_seen, + "agent_busy": agent_busy, "worker": worker, "active_job": active, "latest_job": latest, diff --git a/tests/test_training_coordination.py b/tests/test_training_coordination.py index 60328fb..8be3fd5 100644 --- a/tests/test_training_coordination.py +++ b/tests/test_training_coordination.py @@ -2,6 +2,7 @@ from __future__ import annotations import base64 import hashlib +import json from crypto_spot_bot.training_coordination import TrainingCoordinator @@ -68,3 +69,20 @@ def test_training_coordinator_accepts_chunked_artifact_upload(tmp_path) -> None: assert part_2["complete"] is True assert (tmp_path / "lstm_forecaster.json").read_bytes() == payload assert coordinator.status()["latest_job"]["artifacts"][0]["sha256"] == sha256 + + +def test_running_claimed_job_keeps_agent_online_when_heartbeat_is_stale(tmp_path) -> None: + coordinator = TrainingCoordinator(tmp_path) + coordinator.request_retrain({"source": "android"}) + coordinator.claim({"worker_id": "win-1", "name": "DESKTOP-TMFDL0H"}) + + state_path = tmp_path / "training_coordination.json" + state = json.loads(state_path.read_text(encoding="utf-8")) + state["worker"]["last_seen_at"] = "2026-01-01T00:00:00+00:00" + state_path.write_text(json.dumps(state), encoding="utf-8") + + status = coordinator.status() + + assert status["agent_recently_seen"] is False + assert status["agent_busy"] is True + assert status["agent_online"] is True diff --git a/tools/windows_training_agent.py b/tools/windows_training_agent.py index b50140f..7ae300c 100644 --- a/tools/windows_training_agent.py +++ b/tools/windows_training_agent.py @@ -6,8 +6,10 @@ import hashlib import json import os import platform +import queue import subprocess import sys +import threading import time from datetime import datetime from pathlib import Path @@ -107,6 +109,13 @@ def run_retrain(args: argparse.Namespace, job_id: str, job: dict[str, Any], repo log(log_path, "Running retrain: " + " ".join(quote_for_log(part) for part in cmd)) report_progress(args, job_id, "running", "training", 8, "PyTorch retrain запущен") line_count = 0 + output_queue: queue.Queue[str] = queue.Queue() + + def read_output() -> None: + assert process.stdout is not None + for raw_line in process.stdout: + output_queue.put(raw_line.rstrip()) + with subprocess.Popen( cmd, cwd=str(repo_root), @@ -116,14 +125,32 @@ def run_retrain(args: argparse.Namespace, job_id: str, job: dict[str, Any], repo encoding="utf-8", errors="replace", ) as process: - assert process.stdout is not None - for line in process.stdout: - message = line.rstrip() - log(log_path, message) - line_count += 1 - if line_count == 1 or line_count % 12 == 0: - progress = min(70, 8 + line_count // 3) - report_progress(args, job_id, "running", "training", progress, message[-220:]) + reader = threading.Thread(target=read_output, name="training-output-reader", daemon=True) + reader.start() + last_report_at = 0.0 + last_message = "PyTorch retrain выполняется" + while True: + got_line = False + try: + message = output_queue.get(timeout=5) + got_line = True + log(log_path, message) + line_count += 1 + if message: + last_message = message[-220:] + except queue.Empty: + pass + + progress = min(70, 8 + line_count // 3) + now = time.monotonic() + if got_line or now - last_report_at >= 30: + safe_report_progress(args, job_id, "running", "training", progress, last_message, log_path) + last_report_at = now + + if process.poll() is not None and output_queue.empty(): + break + + reader.join(timeout=2) code = process.wait() if code != 0: raise RuntimeError(f"retrain failed with exit code {code}") @@ -173,6 +200,21 @@ def report_progress( ) +def safe_report_progress( + args: argparse.Namespace, + job_id: str, + status: str, + phase: str, + progress_percent: int, + message: str, + log_path: Path, +) -> None: + try: + report_progress(args, job_id, status, phase, progress_percent, message) + except Exception as exc: # noqa: BLE001 - keep the local training process alive. + log(log_path, f"Progress report failed: {exc}") + + def api_json(args: argparse.Namespace, path: str, payload: dict[str, Any], timeout: int = 30) -> dict[str, Any]: url = args.api_base_url.rstrip("/") + path body = json.dumps(payload, ensure_ascii=False).encode("utf-8")