fix(watch): eager probe on resume so metrics appear immediately

After pausing clears Redis snapshots, resuming or adding pins now triggers
an immediate probe before list_slots returns, with frontend refresh fallback.

Co-authored-by: Cursor <cursoragent@cursor.com>

docs/audit/2026-06-14-watch-eager-probe-on-resume.md

@@ -0,0 +1,36 @@
+# 审计 — 监测开启后立即探针
+
+**Changelog**: `docs/changelog/2026-06-14-watch-eager-probe-on-resume.md`
+
+## 审计范围
+
+| 文件 | 变更 | 状态 |
+|------|------|------|
+| `watch_probe_runner.py` | 抽取 `_probe_server_with_pins`；`trigger_immediate_watch_probe` | ☑ |
+| `watch_service.py` | 加槽/替换/恢复监测后 eager probe | ☑ |
+| `useWatchPins.ts` | 恢复监测 3s/8s 兜底 refresh | ☑ |
+| `WatchSlotCard.vue` | 暂停底栏「已暂停」文案 | ☑ |
+| `test_watch_pins.py` | 恢复监测立即 metrics 用例 + noop mock | ☑ |
+
+## Step 3 规则扫描
+
+| H | 规则 | 结论 |
+|---|------|------|
+| H1 | 无静默吞错 | SAFE — `trigger_immediate_watch_probe` 异常 `logger.exception` |
+| H2 | 探针重复/竞态 | SAFE — 与 5s 循环共用 `_probe_server_with_pins`，仅多一次即时采样 |
+| H3 | 暂停清空 Redis | SAFE — 仍仅在 `monitoring_enabled=false` 时 `clear_watch_redis_snapshot` |
+| H4 | WS 推送 | SAFE — 即时探针同样 `publish_watch_metrics` |
+| H5 | 前端轮询 | SAFE — 仅 `isWatchMetricsPending` 时 refresh，不刷屏 |
+
+## Closure
+
+| H | 判定 | 依据 |
+|---|------|------|
+| H1–H5 | SAFE | 恢复监测空窗期根因已消除；测试 10/10；local_verify 26/26 |
+
+## DoD
+
+- [x] changelog ≥10 行
+- [x] `pytest tests/test_watch_pins.py`
+- [x] `bash scripts/local_verify.sh`
+- [ ] 生产：开启实时监测后数秒内见指标或探针错误态

docs/changelog/2026-06-14-watch-eager-probe-on-resume.md

@@ -0,0 +1,39 @@
+# 2026-06-14 — 监测开启后立即探针
+
+## 摘要
+
+开启实时监测或新加监测槽后，立即执行一次 5s 探针并写入 Redis，避免「等待服务器回传信息」长时间空白。
+
+## 动机
+
+用户反馈开启监测后「获取不到信息」。根因：暂停监测会 `clear_watch_redis_snapshot`，恢复后 API 返回 `metrics=null`，前端进入等待态，需等后台 5s 探针循环（最坏约 5–15s）；若 WebSocket 未及时推送，体验更差。
+
+## 变更
+
+- `server/background/watch_probe_runner.py`：抽取 `_probe_server_with_pins`；新增 `trigger_immediate_watch_probe(server_id)`
+- `server/application/services/watch_service.py`：加槽 / 替换槽 / 恢复监测后调用立即探针
+- `frontend/src/composables/useWatchPins.ts`：恢复监测后 3s、8s 兜底 `refreshPins`（仍 pending 时）
+- `frontend/src/components/watch/WatchSlotCard.vue`：暂停底栏文案「已暂停」替代误显示「已到期」
+- `tests/test_watch_pins.py`：恢复监测立即带回 metrics 用例
+
+## 涉及文件
+
+- `server/background/watch_probe_runner.py`
+- `server/application/services/watch_service.py`
+- `frontend/src/composables/useWatchPins.ts`
+- `frontend/src/components/watch/WatchSlotCard.vue`
+- `tests/test_watch_pins.py`
+
+## 迁移 / 重启
+
+- 需重启 API（后端逻辑变更）
+- 前端需 `vite build` 后部署静态资源
+
+## 验证
+
+```bash
+pytest tests/test_watch_pins.py -q
+bash scripts/local_verify.sh
+```
+
+生产：加槽 → 开启实时监测 → 应在数秒内看到 CPU/内存或明确探针错误态，而非长期空白。

frontend/src/components/watch/WatchSlotCard.vue

@@ -264,7 +264,7 @@ const showSwap = computed(() => (props.slot.metrics?.swap_total_gb ?? 0) > 0)
       @click.stop
     >
       <span class="text-caption text-medium-emphasis">
-        剩余 {{ formatRemaining(slot.remaining_sec) }} · {{ watchTtlShortLabel(slotTtlMinutes) }}
+        已暂停 · {{ watchTtlShortLabel(slotTtlMinutes) }} · Agent 60s 上报
       </span>
       <v-btn size="x-small" variant="text" @click.stop="emit('history', slot)">历史</v-btn>
     </div>

frontend/src/composables/useWatchPins.ts

@@ -6,6 +6,7 @@ import { http, ApiError } from '@/api'
 import { buildWebSocketUrl } from '@/utils/wsUrl'
 import { useAuthStore } from '@/stores/auth'
 import { formatApiError } from '@/utils/apiError'
+import { isWatchMetricsPending } from '@/utils/watchFormat'
 
 import type { WatchTtlMinutes } from '@/constants/watchTtl'
 
@@ -219,6 +220,21 @@ async function setSlotMonitoring(slotIndex: number, monitoringEnabled: boolean)
     { monitoring_enabled: monitoringEnabled },
   )
   applySlots(data)
+  if (monitoringEnabled) {
+    scheduleMetricsRefresh(slotIndex)
+  }
+}
+
+function scheduleMetricsRefresh(slotIndex: number) {
+  for (const delayMs of [3000, 8000]) {
+    setTimeout(() => {
+      const slot = slots.value.find((s) => s.slot_index === slotIndex)
+      if (!slot || slot.empty || slot.monitoring_enabled === false) return
+      if (isWatchMetricsPending(slot.metrics)) {
+        void refreshPins()
+      }
+    }, delayMs)
+  }
 }
 
 async function setSlotTtl(slotIndex: number, ttlMinutes: WatchTtlMinutes) {

server/application/services/watch_service.py

@@ -13,6 +13,7 @@ from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl
 from server.infrastructure.database.server_repo import ServerRepositoryImpl
 from server.infrastructure.database.watch_repo import WatchRepositoryImpl
 from server.infrastructure.redis.client import get_redis
+from server.background.watch_probe_runner import trigger_immediate_watch_probe
 from server.utils.watch_state import clear_watch_redis_snapshot
 from server.utils.watch_metrics import (
     WATCH_MAX_SLOTS,
@@ -80,6 +81,11 @@ async def _restore_idle_agent_watch(watch_repo: WatchRepositoryImpl, server_id:
         await clear_watch_redis_snapshot(server_id)
 
 
+async def _eager_watch_probe(server_id: int) -> None:
+    """Fill Redis + WS immediately when monitoring starts (avoid empty-metrics gap after resume)."""
+    await trigger_immediate_watch_probe(server_id)
+
+
 async def _finalize_due_pins(watch_repo: WatchRepositoryImpl, db: AsyncSession) -> None:
     """Pause TTL-expired pins on read/write paths (probe loop also runs this every 5s)."""
     expired_servers = await watch_repo.expire_due_pins()
@@ -193,6 +199,7 @@ class WatchService:
                 ip_address=ip_address,
             )
             await self.db.commit()
+            await _eager_watch_probe(server_id)
             return await self.list_slots(admin.id)
 
         empty = await self.watch_repo.first_empty_slot(admin.id)
@@ -211,6 +218,7 @@ class WatchService:
                 ip_address=ip_address,
             )
             await self.db.commit()
+            await _eager_watch_probe(server_id)
             return await self.list_slots(admin.id)
 
         if replace_slot is not None:
@@ -228,6 +236,7 @@ class WatchService:
                 ip_address=ip_address,
             )
             await self.db.commit()
+            await _eager_watch_probe(server_id)
             return await self.list_slots(admin.id)
 
         raise SlotsFullError()
@@ -276,6 +285,7 @@ class WatchService:
             ip_address=ip_address,
         )
         await self.db.commit()
+        await _eager_watch_probe(server_id)
         return await self.list_slots(admin.id)
 
     async def remove_slot(
@@ -340,6 +350,8 @@ class WatchService:
         await self.db.commit()
         if not monitoring_enabled:
             await _restore_idle_agent_watch(self.watch_repo, pin.server_id)
+        else:
+            await _eager_watch_probe(pin.server_id)
         return await self.list_slots(admin.id)
 
     async def set_slot_ttl(

server/background/watch_probe_runner.py

@@ -175,6 +175,129 @@ async def probe_server_metrics(
     return sample
 
 
+async def _probe_server_with_pins(
+    *,
+    watch_repo: WatchRepositoryImpl,
+    server_repo: ServerRepositoryImpl,
+    redis,
+    server_id: int,
+    pin_rows: list[dict[str, Any]],
+    now: datetime,
+    now_ts: float,
+    fetch_processes: bool,
+) -> None:
+    """Probe one server, persist Redis snapshot, WS push, and DB records for all its pins."""
+    server = await server_repo.get_by_id(server_id)
+    if not server:
+        sample = failure_sample(status="offline", source="ssh", error="服务器不存在")
+        server_name = f"server-{server_id}"
+    else:
+        server_name = server.name or f"server-{server_id}"
+        heartbeat = await redis.hgetall(f"{REDIS_HEARTBEAT_PREFIX}{server_id}")
+        sample = await probe_server_metrics(
+            server,
+            heartbeat or None,
+            fetch_processes=fetch_processes,
+        )
+
+    prev = await _load_rate_state(redis, server_id)
+    new_state = apply_rates_to_sample(sample, prev, now_ts)
+    await _save_rate_state(redis, server_id, new_state)
+
+    live = sample.to_live_dict(server_id=server_id)
+    live["ts"] = now.isoformat()
+    await _append_live_point(redis, server_id, live)
+    if sample.processes_json:
+        await _save_processes(redis, server_id, sample.processes_json)
+
+    if sample.probe_status == "ok" and server:
+        try:
+            await process_watch_probe_alerts(
+                server_id=server_id,
+                server_name=server_name,
+                sample=sample,
+            )
+        except Exception:
+            logger.debug("watch alert processing failed", exc_info=True)
+
+    admin_ids = [r["admin_id"] for r in pin_rows]
+    try:
+        from server.api.websocket import publish_watch_metrics
+
+        await publish_watch_metrics(
+            {
+                "type": "watch_metrics",
+                "server_id": server_id,
+                "admin_ids": admin_ids,
+                "metrics": live,
+            }
+        )
+    except Exception:
+        logger.debug("watch WS publish failed", exc_info=True)
+
+    for row in pin_rows:
+        await watch_repo.insert_probe_record(
+            session_id=row["session_id"],
+            server_id=server_id,
+            admin_id=row["admin_id"],
+            recorded_at=now,
+            source=sample.source,
+            probe_status=sample.probe_status,
+            duration_ms=sample.duration_ms,
+            is_online=sample.is_online,
+            cpu_pct=sample.cpu_pct,
+            mem_pct=sample.mem_pct,
+            disk_pct=sample.disk_pct,
+            disk_mount=sample.disk_mount,
+            load_1=sample.load_1,
+            load_5=sample.load_5,
+            load_15=sample.load_15,
+            net_bytes_sent=sample.net_bytes_sent,
+            net_bytes_recv=sample.net_bytes_recv,
+            disk_read_bytes=sample.disk_read_bytes,
+            disk_write_bytes=sample.disk_write_bytes,
+            net_up_bps=sample.net_up_bps,
+            net_down_bps=sample.net_down_bps,
+            disk_read_bps=sample.disk_read_bps,
+            disk_write_bps=sample.disk_write_bps,
+            processes_json=sample.processes_json,
+            error=sample.error,
+        )
+
+
+async def trigger_immediate_watch_probe(server_id: int) -> None:
+    """Eager probe after pin add or monitoring resume — fills Redis before list_slots."""
+    try:
+        async with AsyncSessionLocal() as db:
+            watch_repo = WatchRepositoryImpl(db)
+            pin_rows = [
+                p
+                for p in await watch_repo.list_active_pins_all(monitoring_only=True)
+                if p["server_id"] == server_id
+            ]
+            if not pin_rows:
+                return
+            server_repo = ServerRepositoryImpl(db)
+            redis = get_redis()
+            now = _utc_naive_now()
+            now_ts = time.time()
+            _process_cycle[server_id] += 1
+            fetch_procs = _process_cycle[server_id] % 2 == 0
+            await _probe_server_with_pins(
+                watch_repo=watch_repo,
+                server_repo=server_repo,
+                redis=redis,
+                server_id=server_id,
+                pin_rows=pin_rows,
+                now=now,
+                now_ts=now_ts,
+                fetch_processes=fetch_procs,
+            )
+            await db.commit()
+    except Exception:
+        logger.exception("immediate watch probe failed for server %s", server_id)
+
+
 async def _run_probe_cycle() -> None:
     async with AsyncSessionLocal() as db:
         watch_repo = WatchRepositoryImpl(db)
@@ -197,84 +320,18 @@ async def _run_probe_cycle() -> None:
         now_ts = time.time()
 
         for server_id, pin_rows in by_server.items():
-            server = await server_repo.get_by_id(server_id)
-            if not server:
-                sample = failure_sample(status="offline", source="ssh", error="服务器不存在")
-                server_name = f"server-{server_id}"
-            else:
-                server_name = server.name or f"server-{server_id}"
-                heartbeat = await redis.hgetall(f"{REDIS_HEARTBEAT_PREFIX}{server_id}")
-                _process_cycle[server_id] += 1
-                fetch_procs = _process_cycle[server_id] % 2 == 0
-                sample = await probe_server_metrics(
-                    server,
-                    heartbeat or None,
-                    fetch_processes=fetch_procs,
-                )
-
-            prev = await _load_rate_state(redis, server_id)
-            new_state = apply_rates_to_sample(sample, prev, now_ts)
-            await _save_rate_state(redis, server_id, new_state)
-
-            live = sample.to_live_dict(server_id=server_id)
-            live["ts"] = now.isoformat()
-            await _append_live_point(redis, server_id, live)
-            if sample.processes_json:
-                await _save_processes(redis, server_id, sample.processes_json)
-
-            if sample.probe_status == "ok" and server:
-                try:
-                    await process_watch_probe_alerts(
-                        server_id=server_id,
-                        server_name=server_name,
-                        sample=sample,
-                    )
-                except Exception:
-                    logger.debug("watch alert processing failed", exc_info=True)
-
-            admin_ids = [r["admin_id"] for r in pin_rows]
-            try:
-                from server.api.websocket import publish_watch_metrics
-
-                await publish_watch_metrics(
-                    {
-                        "type": "watch_metrics",
-                        "server_id": server_id,
-                        "admin_ids": admin_ids,
-                        "metrics": live,
-                    }
-                )
-            except Exception:
-                logger.debug("watch WS publish failed", exc_info=True)
-
-            for row in pin_rows:
-                await watch_repo.insert_probe_record(
-                    session_id=row["session_id"],
-                    server_id=server_id,
-                    admin_id=row["admin_id"],
-                    recorded_at=now,
-                    source=sample.source,
-                    probe_status=sample.probe_status,
-                    duration_ms=sample.duration_ms,
-                    is_online=sample.is_online,
-                    cpu_pct=sample.cpu_pct,
-                    mem_pct=sample.mem_pct,
-                    disk_pct=sample.disk_pct,
-                    disk_mount=sample.disk_mount,
-                    load_1=sample.load_1,
-                    load_5=sample.load_5,
-                    load_15=sample.load_15,
-                    net_bytes_sent=sample.net_bytes_sent,
-                    net_bytes_recv=sample.net_bytes_recv,
-                    disk_read_bytes=sample.disk_read_bytes,
-                    disk_write_bytes=sample.disk_write_bytes,
-                    net_up_bps=sample.net_up_bps,
-                    net_down_bps=sample.net_down_bps,
-                    disk_read_bps=sample.disk_read_bps,
-                    disk_write_bps=sample.disk_write_bps,
-                    processes_json=sample.processes_json,
-                    error=sample.error,
-                )
+            _process_cycle[server_id] += 1
+            fetch_procs = _process_cycle[server_id] % 2 == 0
+            await _probe_server_with_pins(
+                watch_repo=watch_repo,
+                server_repo=server_repo,
+                redis=redis,
+                server_id=server_id,
+                pin_rows=pin_rows,
+                now=now,
+                now_ts=now_ts,
+                fetch_processes=fetch_procs,
+            )
 
         await db.commit()
 

tests/test_watch_pins.py

@@ -20,6 +20,14 @@ def mock_watch_redis(monkeypatch):
     mock_redis.delete = AsyncMock(return_value=1)
     monkeypatch.setattr("server.application.services.watch_service.get_redis", lambda: mock_redis)
     monkeypatch.setattr("server.utils.watch_state.get_redis", lambda: mock_redis)
+
+    async def noop_immediate_probe(_server_id: int) -> None:
+        return None
+
+    monkeypatch.setattr(
+        "server.application.services.watch_service.trigger_immediate_watch_probe",
+        noop_immediate_probe,
+    )
     return mock_redis
 
 
@@ -122,6 +130,44 @@ async def test_watch_pin_monitoring_toggle(db_session, test_admin, mock_watch_re
     assert resumed["slots"][0]["remaining_sec"] > 0
 
 
+@pytest.mark.asyncio
+async def test_watch_resume_eager_probe_fills_metrics(
+    db_session, test_admin, mock_watch_redis, monkeypatch,
+):
+    """开启监测后立即探针，list_slots 应带上 metrics（不等 5s 周期）。"""
+    import json
+
+    from server.application.services.watch_service import WatchService
+
+    server = Server(name="watch-eager", domain="we.example.com", port=22, agent_version="1.0")
+    db_session.add(server)
+    await db_session.commit()
+
+    async def fake_immediate_probe(server_id: int) -> None:
+        live = json.dumps({"probe_status": "ok", "cpu_pct": 9.5, "server_id": server_id})
+        mock_watch_redis.lindex = AsyncMock(return_value=live)
+
+    monkeypatch.setattr(
+        "server.application.services.watch_service.trigger_immediate_watch_probe",
+        fake_immediate_probe,
+    )
+
+    svc = WatchService(db_session)
+    admin = test_admin["admin"]
+    await svc.pin_server(admin=admin, server_id=server.id)
+    await svc.set_slot_monitoring(admin=admin, slot_index=0, monitoring_enabled=False)
+
+    resumed = await svc.set_slot_monitoring(
+        admin=admin,
+        slot_index=0,
+        monitoring_enabled=True,
+    )
+    metrics = resumed["slots"][0]["metrics"]
+    assert metrics is not None
+    assert metrics["probe_status"] == "ok"
+    assert metrics["cpu_pct"] == 9.5
+
+
 @pytest.mark.asyncio
 async def test_watch_pin_ttl_minutes(db_session, test_admin, mock_watch_redis):
     from server.application.services.watch_service import WatchService