Show live training status and internal GPT logs

app.py

@@ -15,6 +15,7 @@ import os
 import shutil
 import subprocess
 import sys
+import threading
 import time
 from dataclasses import dataclass
 from datetime import datetime
@@ -66,7 +67,7 @@ MODEL_PATTERNS = [
     "*.safetensors",
     "*.model",
 ]
-UPSTREAM_PATCH_VERSION = "2026-05-25-worker0-v2"
+UPSTREAM_PATCH_VERSION = "2026-05-25-worker0-v3"
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
 log = logging.getLogger(__name__)
@@ -102,6 +103,10 @@ class VersionContext:
     prep_live_log: Path
     sovits_live_log: Path
     gpt_live_log: Path
+    prep_status_path: Path
+    sovits_status_path: Path
+    gpt_status_path: Path
+    gpt_train_log: Path
     sovits_ckpt_dir: Path
     gpt_log_dir: Path
     sovits_output_dir: Path
@@ -172,6 +177,10 @@ def get_version_context(version: str):
         prep_live_log=exp_dir / "_live_prepare.log",
         sovits_live_log=exp_dir / "_live_sovits.log",
         gpt_live_log=exp_dir / "_live_gpt.log",
+        prep_status_path=exp_dir / "_status_prepare.json",
+        sovits_status_path=exp_dir / "_status_sovits.json",
+        gpt_status_path=exp_dir / "_status_gpt.json",
+        gpt_train_log=exp_dir / "train.log",
         sovits_ckpt_dir=exp_dir / f"logs_s2_{version}",
         gpt_log_dir=exp_dir / f"logs_s1_{version}",
         sovits_output_dir=OUTPUT_ROOT / f"SoVITS_weights_{version}",
@@ -222,11 +231,102 @@ def clear_live_log(path: Path):
     path.write_text("", encoding="utf-8")
 
 
-def read_live_log(path: Path):
+def clear_status_file(path: Path):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if path.exists():
+        path.unlink()
+
+
+def read_text_tail(path: Path, max_lines=200):
     if not path.exists():
         return ""
     lines = path.read_text(encoding="utf-8", errors="ignore").splitlines()
-    return "\n".join(lines[-200:])
+    return "\n".join(lines[-max_lines:])
+
+
+def read_live_log(path: Path):
+    return read_text_tail(path, max_lines=200)
+
+
+def write_status_file(path: Path, payload):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp_path = path.with_suffix(path.suffix + ".tmp")
+    tmp_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+    tmp_path.replace(path)
+
+
+def read_status_file(path: Path):
+    if not path.exists():
+        return None
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except Exception:
+        return None
+
+
+def now_iso():
+    return datetime.now().isoformat(timespec="seconds")
+
+
+def process_metrics(pid: int, previous=None):
+    stat_path = Path(f"/proc/{pid}/stat")
+    statm_path = Path(f"/proc/{pid}/statm")
+    if not stat_path.exists() or not statm_path.exists():
+        return {"alive": False, "cpu_percent": 0.0, "rss_mb": 0.0, "cpu_time_seconds": 0.0}
+    stat_fields = stat_path.read_text(encoding="utf-8").split()
+    cpu_ticks = int(stat_fields[13]) + int(stat_fields[14])
+    clock_ticks = os.sysconf(os.sysconf_names["SC_CLK_TCK"])
+    cpu_time_seconds = cpu_ticks / float(clock_ticks)
+    rss_pages = int(statm_path.read_text(encoding="utf-8").split()[1])
+    page_size = os.sysconf("SC_PAGE_SIZE")
+    rss_mb = rss_pages * page_size / (1024 * 1024)
+    cpu_percent = 0.0
+    if previous is not None:
+        prev_ts, prev_cpu_seconds = previous
+        elapsed = max(time.time() - prev_ts, 1e-6)
+        cpu_percent = max(0.0, ((cpu_time_seconds - prev_cpu_seconds) / elapsed) * 100.0)
+    return {
+        "alive": True,
+        "cpu_percent": round(cpu_percent, 2),
+        "rss_mb": round(rss_mb, 2),
+        "cpu_time_seconds": round(cpu_time_seconds, 2),
+    }
+
+
+def status_summary(status):
+    if not status:
+        return ""
+    parts = [
+        f"阶段: {status.get('state', 'unknown')}",
+        f"更新时间: {status.get('updated_at', '-')}",
+    ]
+    if status.get("pid") is not None:
+        parts.append(f"pid={status['pid']}")
+    if status.get("cpu_percent") is not None:
+        parts.append(f"cpu={status['cpu_percent']}%")
+    if status.get("rss_mb") is not None:
+        parts.append(f"rss={status['rss_mb']} MB")
+    if status.get("silent_for_seconds") is not None:
+        parts.append(f"静默={status['silent_for_seconds']}s")
+    if status.get("last_output_line"):
+        parts.append(f"最后输出: {status['last_output_line'][:160]}")
+    return "[状态]\n" + " | ".join(parts)
+
+
+def combined_log_text(primary_path: Path, status_path: Path | None = None, extra_paths=None):
+    sections = []
+    status = read_status_file(status_path) if status_path else None
+    summary = status_summary(status)
+    if summary:
+        sections.append(summary)
+    primary = read_text_tail(primary_path, max_lines=200)
+    if primary:
+        sections.append(primary)
+    for label, path in extra_paths or []:
+        content = read_text_tail(path, max_lines=120)
+        if content:
+            sections.append(f"[{label}]\n{content}")
+    return "\n\n".join(section for section in sections if section)
 
 
 def push(logs, message, live_path: Path | None = None):
@@ -236,7 +336,7 @@ def push(logs, message, live_path: Path | None = None):
     return "\n".join(logs[-200:])
 
 
-def run_cmd(command, cwd=None, env=None):
+def run_cmd(command, cwd=None, env=None, status_path: Path | None = None, live_path: Path | None = None, status_label=None):
     proc = subprocess.Popen(
         command,
         cwd=str(cwd) if cwd else None,
@@ -246,14 +346,82 @@ def run_cmd(command, cwd=None, env=None):
         text=True,
         bufsize=1,
     )
-    yield f"$ {' '.join(command)}"
+    command_text = " ".join(command)
+    shared_state = {
+        "started_at": now_iso(),
+        "last_output_at_ts": time.time(),
+        "last_output_line": f"$ {command_text}",
+        "last_heartbeat_at_ts": 0.0,
+    }
+    stop_event = threading.Event()
+
+    def write_runtime_status(state, exit_code=None):
+        if status_path is None:
+            return
+        payload = {
+            "label": status_label or "任务",
+            "state": state,
+            "pid": proc.pid,
+            "command": command_text,
+            "started_at": shared_state["started_at"],
+            "updated_at": now_iso(),
+            "last_output_at": datetime.fromtimestamp(shared_state["last_output_at_ts"]).isoformat(timespec="seconds"),
+            "last_output_line": shared_state["last_output_line"],
+            "silent_for_seconds": round(max(time.time() - shared_state["last_output_at_ts"], 0.0), 1),
+            "exit_code": exit_code,
+        }
+        payload.update(process_metrics(proc.pid))
+        write_status_file(status_path, payload)
+
+    def monitor_process():
+        previous = None
+        while not stop_event.is_set() and proc.poll() is None:
+            metrics = process_metrics(proc.pid, previous)
+            previous = (time.time(), metrics["cpu_time_seconds"])
+            if status_path is not None:
+                payload = {
+                    "label": status_label or "任务",
+                    "state": "running",
+                    "pid": proc.pid,
+                    "command": command_text,
+                    "started_at": shared_state["started_at"],
+                    "updated_at": now_iso(),
+                    "last_output_at": datetime.fromtimestamp(shared_state["last_output_at_ts"]).isoformat(timespec="seconds"),
+                    "last_output_line": shared_state["last_output_line"],
+                    "silent_for_seconds": round(max(time.time() - shared_state["last_output_at_ts"], 0.0), 1),
+                    "exit_code": None,
+                }
+                payload.update(metrics)
+                write_status_file(status_path, payload)
+            silent_for = time.time() - shared_state["last_output_at_ts"]
+            if live_path is not None and silent_for >= 30 and time.time() - shared_state["last_heartbeat_at_ts"] >= 30:
+                append_live_log(
+                    live_path,
+                    (
+                        f"[状态] {status_label or '任务'} 仍在运行 | pid={proc.pid} | "
+                        f"cpu={metrics['cpu_percent']}% | rss={metrics['rss_mb']} MB | 静默 {int(silent_for)}s"
+                    ),
+                )
+                shared_state["last_heartbeat_at_ts"] = time.time()
+            stop_event.wait(10)
+
+    write_runtime_status("starting")
+    monitor_thread = threading.Thread(target=monitor_process, daemon=True)
+    monitor_thread.start()
+    yield f"$ {command_text}"
     for raw in proc.stdout:
         line = raw.rstrip()
         if line:
+            shared_state["last_output_at_ts"] = time.time()
+            shared_state["last_output_line"] = line
+            write_runtime_status("running")
             yield line
     code = proc.wait()
+    stop_event.set()
+    monitor_thread.join(timeout=1)
+    write_runtime_status("completed" if code == 0 else "failed", exit_code=code)
     if code != 0:
-        raise RuntimeError(f"命令失败 (exit={code}): {' '.join(command)}")
+        raise RuntimeError(f"命令失败 (exit={code}): {command_text}")
 
 
 def python_script_command(script_path, *script_args, file_system_sharing=False):
@@ -387,8 +555,12 @@ def managed_single_paths(ctx: VersionContext):
         ("文本特征总表", ctx.text_path),
         ("语义特征总表", ctx.semantic_path),
         ("预处理实时日志", ctx.prep_live_log),
+        ("预处理状态", ctx.prep_status_path),
         ("SoVITS 实时日志", ctx.sovits_live_log),
+        ("SoVITS 状态", ctx.sovits_status_path),
         ("GPT 实时日志", ctx.gpt_live_log),
+        ("GPT 状态", ctx.gpt_status_path),
+        ("GPT train.log", ctx.gpt_train_log),
     ]
 
 
@@ -486,8 +658,12 @@ def allowed_delete_roots(ctx: VersionContext):
         ctx.text_path,
         ctx.semantic_path,
         ctx.prep_live_log,
+        ctx.prep_status_path,
         ctx.sovits_live_log,
+        ctx.sovits_status_path,
         ctx.gpt_live_log,
+        ctx.gpt_status_path,
+        ctx.gpt_train_log,
     ]
 
 
@@ -525,6 +701,7 @@ def cleanup_preprocess(version=DEFAULT_VERSION, selected_path=None):
     ctx = get_version_context(version)
     reset_preprocess_outputs(ctx)
     clear_live_log(ctx.prep_live_log)
+    clear_status_file(ctx.prep_status_path)
     status = f"已清理 {ctx.spec.version} 的预处理产物。"
     return manager_action_result(version, status, selected_path)
 
@@ -550,6 +727,9 @@ def cleanup_live_logs(version=DEFAULT_VERSION, selected_path=None):
     clear_live_log(ctx.prep_live_log)
     clear_live_log(ctx.sovits_live_log)
     clear_live_log(ctx.gpt_live_log)
+    clear_status_file(ctx.prep_status_path)
+    clear_status_file(ctx.sovits_status_path)
+    clear_status_file(ctx.gpt_status_path)
     status = f"已清理 {ctx.spec.version} 的实时日志。"
     return manager_action_result(version, status, selected_path)
 
@@ -739,6 +919,16 @@ class Text2SemanticDataModule(LightningDataModule):
 """,
         encoding="utf-8",
     )
+    s1_train = GPT_SOVITS_DIR / "GPT_SoVITS" / "s1_train.py"
+    s1_train_content = s1_train.read_text(encoding="utf-8")
+    trainer_anchor = "        callbacks=[ckpt_callback],\n        use_distributed_sampler=False,"
+    if "log_every_n_steps=1" not in s1_train_content and trainer_anchor in s1_train_content:
+        s1_train_content = s1_train_content.replace(
+            trainer_anchor,
+            "        callbacks=[ckpt_callback],\n        log_every_n_steps=1,\n        use_distributed_sampler=False,",
+            1,
+        )
+        s1_train.write_text(s1_train_content, encoding="utf-8")
     patch_marker.write_text(f"{UPSTREAM_PATCH_VERSION}\n", encoding="utf-8")
 
 
@@ -1021,6 +1211,7 @@ def check_environment(version=DEFAULT_VERSION):
     logs = []
     ctx = get_version_context(version)
     clear_live_log(ctx.prep_live_log)
+    clear_status_file(ctx.prep_status_path)
     try:
         final = None
         final = yield from setup_environment_steps(logs, version, ctx.prep_live_log)
@@ -1035,6 +1226,7 @@ def download_dataset(version=DEFAULT_VERSION):
     logs = []
     ctx = get_version_context(version)
     clear_live_log(ctx.prep_live_log)
+    clear_status_file(ctx.prep_status_path)
     try:
         final = None
         final = yield from download_dataset_steps(logs, version, ctx.prep_live_log)
@@ -1049,6 +1241,7 @@ def prepare_data(version=DEFAULT_VERSION):
     logs = []
     ctx = get_version_context(version)
     clear_live_log(ctx.prep_live_log)
+    clear_status_file(ctx.prep_status_path)
     try:
         final = yield from prepare_data_steps(logs, version, ctx.prep_live_log)
         yield push(logs, final, ctx.prep_live_log)
@@ -1061,6 +1254,7 @@ def start_training(version=DEFAULT_VERSION, epochs=2, batch_size=1, save_every_e
     logs = []
     ctx = get_version_context(version)
     clear_live_log(ctx.sovits_live_log)
+    clear_status_file(ctx.sovits_status_path)
     try:
         yield push(logs, f"当前版本：{ctx.spec.version}（{ctx.spec.stability}）", ctx.sovits_live_log), None
         for update in setup_environment_steps(logs, version, ctx.sovits_live_log):
@@ -1081,6 +1275,9 @@ def start_training(version=DEFAULT_VERSION, epochs=2, batch_size=1, save_every_e
             ),
             cwd=GPT_SOVITS_DIR,
             env=env,
+            status_path=ctx.sovits_status_path,
+            live_path=ctx.sovits_live_log,
+            status_label=f"{ctx.spec.version} SoVITS",
         ):
             yield push(logs, line, ctx.sovits_live_log), None
         latest = latest_file(ctx.sovits_output_dir, ".pth")
@@ -1096,6 +1293,7 @@ def start_gpt_training(version=DEFAULT_VERSION, epochs=1, batch_size=1, save_eve
     logs = []
     ctx = get_version_context(version)
     clear_live_log(ctx.gpt_live_log)
+    clear_status_file(ctx.gpt_status_path)
     try:
         yield push(logs, f"当前版本：{ctx.spec.version}（{ctx.spec.stability}）", ctx.gpt_live_log), None
         for update in setup_environment_steps(logs, version, ctx.gpt_live_log):
@@ -1118,6 +1316,9 @@ def start_gpt_training(version=DEFAULT_VERSION, epochs=1, batch_size=1, save_eve
             ),
             cwd=GPT_SOVITS_DIR,
             env=env,
+            status_path=ctx.gpt_status_path,
+            live_path=ctx.gpt_live_log,
+            status_label=f"{ctx.spec.version} GPT",
         ):
             yield push(logs, line, ctx.gpt_live_log), None
         latest = latest_file(ctx.gpt_output_dir, ".ckpt")
@@ -1136,9 +1337,13 @@ def refresh_outputs(version=DEFAULT_VERSION):
 def live_logs(version=DEFAULT_VERSION):
     ctx = get_version_context(version)
     return (
-        read_live_log(ctx.prep_live_log),
-        read_live_log(ctx.sovits_live_log),
-        read_live_log(ctx.gpt_live_log),
+        combined_log_text(ctx.prep_live_log, ctx.prep_status_path),
+        combined_log_text(ctx.sovits_live_log, ctx.sovits_status_path),
+        combined_log_text(
+            ctx.gpt_live_log,
+            ctx.gpt_status_path,
+            extra_paths=[("train.log", ctx.gpt_train_log)],
+        ),
     )
 
 
@@ -1205,6 +1410,11 @@ def agent_status(version=DEFAULT_VERSION):
             "sovits": [file_record(path) for path in sovits_files[:10]],
             "gpt": [file_record(path) for path in gpt_files[:10]],
         },
+        "tasks": {
+            "prepare": read_status_file(ctx.prep_status_path),
+            "sovits": read_status_file(ctx.sovits_status_path),
+            "gpt": read_status_file(ctx.gpt_status_path),
+        },
     }