's'

src/android_control.py

@@ -299,6 +299,14 @@ def _is_png(data: bytes) -> bool:
     return bool(data and len(data) > 8 and data.startswith(b"\x89PNG\r\n\x1a\n"))
 
 
+async def android_logcat_recent(serial: str, lines: int = 200) -> str:
+    """Dump recent logcat lines (logcat -d -t N), similar to device-console tooling."""
+    if not serial:
+        raise ValueError("Missing Android device serial")
+    n = max(1, min(int(lines), 3000))
+    return await _shell(serial, f"logcat -d -t {n}", timeout=60.0)
+
+
 async def android_screenshot(serial: str) -> bytes:
     if not serial:
         raise ValueError("Missing Android device serial")

src/control_plane.py

@@ -8,9 +8,11 @@ import shutil
 import socket
 import subprocess
 import sys
+import time
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Iterable
+from typing import Any, Iterable
+from urllib.error import URLError
 from urllib.request import urlopen
 
 BASE_DIR = Path(__file__).resolve().parent.parent
@@ -454,6 +456,65 @@ def load_stack_env() -> dict[str, str]:
     return values
 
 
+def _probe_tcp_port(host: str, port: int, timeout: float = 2.0) -> dict[str, Any]:
+    t0 = time.monotonic()
+    try:
+        with socket.create_connection((host, int(port)), timeout=timeout):
+            ms = (time.monotonic() - t0) * 1000
+            return {"ok": True, "reachable": True, "latency_ms": round(ms, 2), "error": ""}
+    except OSError as exc:
+        ms = (time.monotonic() - t0) * 1000
+        return {"ok": False, "reachable": False, "latency_ms": round(ms, 2), "error": str(exc)}
+
+
+def _probe_http_get(url: str, timeout: float = 2.5) -> dict[str, Any]:
+    t0 = time.monotonic()
+    try:
+        with urlopen(url, timeout=timeout) as resp:
+            code = int(resp.getcode() or 0)
+            ms = (time.monotonic() - t0) * 1000
+            return {
+                "ok": True,
+                "http_code": code,
+                "latency_ms": round(ms, 2),
+                "reachable": True,
+                "error": "",
+            }
+    except (URLError, OSError, ValueError) as exc:
+        ms = (time.monotonic() - t0) * 1000
+        return {
+            "ok": False,
+            "http_code": 0,
+            "latency_ms": round(ms, 2),
+            "reachable": False,
+            "error": str(exc),
+        }
+
+
+async def stack_stream_probes() -> dict[str, Any]:
+    """本机 SRS / Neko HTTP 端口探测（供 live.local 验收）。"""
+    env = load_stack_env()
+    srs_p = int(env.get("SRS_HTTP_PORT", "8080") or 8080)
+    neko_p = int(env.get("NEKO_HTTP_PORT", "9200") or 9200)
+    loop = asyncio.get_event_loop()
+
+    def run_srs() -> dict[str, Any]:
+        tcp = _probe_tcp_port("127.0.0.1", srs_p)
+        http = _probe_http_get(f"http://127.0.0.1:{srs_p}/", timeout=2.5)
+        return {"port": srs_p, "tcp": tcp, "http": http}
+
+    def run_neko() -> dict[str, Any]:
+        tcp = _probe_tcp_port("127.0.0.1", neko_p)
+        http = _probe_http_get(f"http://127.0.0.1:{neko_p}/", timeout=2.5)
+        return {"port": neko_p, "tcp": tcp, "http": http}
+
+    srs_d, neko_d = await asyncio.gather(
+        loop.run_in_executor(None, run_srs),
+        loop.run_in_executor(None, run_neko),
+    )
+    return {"srs": srs_d, "neko": neko_d}
+
+
 def stack_summary() -> dict:
     env = load_stack_env()
     hostname = env.get("HOSTNAME_ALIAS") or env.get("HOSTNAME") or socket.gethostname()

src/hub_routes.py

@@ -8,7 +8,7 @@ from fastapi import APIRouter
 from pydantic import BaseModel
 
 from src.android_control import adb_available, list_android_devices
-from src.control_plane import query_services, stack_summary, system_snapshot
+from src.control_plane import query_services, stack_stream_probes, stack_summary, system_snapshot
 from src.live_config import business_config, load_hub_config_snapshot, services_config_list
 from src.web_process_backend import WebProcessBackend
 
@@ -38,6 +38,7 @@ async def hub_dashboard() -> dict[str, Any]:
     else:
         devices = []
     stack = stack_summary()
+    probes = await stack_stream_probes()
 
     live_status: dict[str, Any] = {
         "processes": [],
@@ -68,6 +69,7 @@ async def hub_dashboard() -> dict[str, Any]:
     return {
         "snapshot": snap,
         "stack": stack,
+        "probes": probes,
         "services": services,
         "android": {
             "adb_available": adb_available(),