已撤回：网页监测改动（改为桌面 App 实现）

SOURCE-SHA256.json

@@ -2,7 +2,7 @@
   ".gitignore": "1364e59c28568273da17a19c4b4fb78dd86aaed90f6b7e71d7823585bba84707",
   ".openai/hosting.json": "8e1049aefe1f5e5853c8bd13a3e4e1b01f9251a452bfd8b5e384a260f4a551b7",
   "README.md": "f6cc6f56f7ca81b7a80cf5359a5849216d620fa5719910aee6a8e11289d289be",
-  "docs/OPERATIONS.md": "f3bf36698f26eb470016f5ed7b64fa174e8b9422a56375a0ccd9efae6b01ab29",
+  "docs/OPERATIONS.md": "985dbb23e3e1c84204be09cc51b786f6a5914267a3deed94ef1f7eb26399303b",
   "docs/typography-review.md": "57fc36226bcf1b0cff78acc9ce25c63a57e89ca739161f3c2e959f6ec35b4aa7",
   "downloads/GPU-Queue-Widget-1.1.2-macOS.zip": "77b1652f46675cf7f0b57616411df8d66124fbd2da499a270e1ad246c7902bff",
   "downloads/GPU-Queue-Widget-1.1.3-macOS.zip": "25d8a432737623a950a65115b1bb6b0f72bbfbbd86c783842a93a245a2ea4be1",
@@ -16,10 +16,10 @@
   "gpuqueue/cli.py": "d757987571b6552edc3c28fd11ffedb40116b2b7d2d5f6e118e0330314e4204e",
   "gpuqueue/control.py": "1412129e7f0f7c4f67db0c4c11f3e66f24cc1353f7bf67ad13b2ef86ae29de28",
   "gpuqueue/core.py": "a726498140ffc19076a461b9254ef3cb7ce7085119bd11997650f2068fef4c24",
-  "gpuqueue/platform.py": "1c4aad5ea1c2f41937ebaeafec03a5cc99ffee7d2dda706446d9919e49277f9d",
+  "gpuqueue/platform.py": "2e940049c5d58a03df556032ff1f11d433b0fce81c7288a854910539d4b22817",
   "gpuqueue/runner.py": "300c18ee4360bdf320f1a02c9a1eae21cca33d39347f4157b43e705f5177c304",
-  "gpuqueue/serverbox.py": "204470fb517ccd27609f5b2d1d68fa30d1a53f541a18d8e4412761fa8023817c",
-  "gpuqueue/service.py": "df60d5d835a3baef689dba4a80362088c07eea1e711ecc7244fda4ed504d9972",
+  "gpuqueue/serverbox.py": "247ca4fac527def799b4224fdf99506be4033467c27f413d139ab6b47b2afd1e",
+  "gpuqueue/service.py": "889854da765ca66a9487ae9fb7c01155b9f479f11321a88067531fba051b4677",
   "gpuqueue/stop_control.py": "10d2241a2882dbb97605f4fcb364c6f6fbe01451b8d90137d3aed1ff859433f7",
   "gpuqueue/web.py": "ae58b8225450409196de233a4e6901dc5811c5728d8ffcd9e3fdcce000345b06",
   "launchers/macos/Open-GPU-Queue.command": "b3a725d7435fe75c307f63bcb9a5c04ac578893c2ec67ab2f47f5e15f3611c6d",
@@ -50,16 +50,20 @@
   "tests/test_gpuqueue.py": "614df82d4652e8d529d5c030df852aabc766f9b03427bf2acf8d286a4e0cc316",
   "tests/test_launcher.py": "c336ce79267510501294ee1f73600b5b57ab8143469d75a8cf5d827e650e941d",
   "tests/test_local.py": "5721b2cf858b1fbaa4b56f479c0ee1490c1df63f1b8a27da4b632c92e908481d",
-  "tests/test_serverbox.py": "0716194806731cf4a11f33fb6fa8a163c6558404f336f8855c14188e98bb6cb3",
+  "tests/test_serverbox.py": "c8a7c0912a25263ab0b8e95545d689a38eba9fb70ffd81ff7c71eb5d76fea81d",
   "tests/test_stop.py": "39717a7e08eb2bf7c1e2b9b5416ed335bc71c0760210091fd3c2d70ffbf977cd",
   "tests/test_widget_autostart.py": "7819f157ca95931554a4aba9e4b0c195f99b10e21499b70e8eb9b0aa987bef40",
   "tests/test_widget_install.py": "bb0c80f1e2d8bb2fa685a51d8d5d9bebd05b4effc485e00d971618facda02b5c",
   "tests/widget-update.test.cjs": "7a3a3f5c9e7dd76c0e54968f09bbe48f9cdbf4ad959d46b46eddb3b8e2d48cb5",
   "tests/widget.test.cjs": "21dde14aa712b8f3c814354af8a1323e6c7f92af866169e78dd9b4095f501faf",
   "updates/gpu-queue-widget.json": "4ff3b46935ab67689c962cc22ceb3155960ff6de7572ff3435f15c481580a475",
-  "web/app.js": "9401da1be8229a99dc311c9e83df76d760a8eb553ac87269e25d87e2c3903ad0",
-  "web/index.html": "ad5cf456e1824c8ea9827c9c2db3e8741529779e68995d5074a766714c37f718",
-  "web/model.js": "55d7b7162a64eb2b142e6be6b5b4174c463a94b49d8525947e0563ba62896d4c",
-  "web/style.css": "e0e3d3335e69def00fe4ab990794bc2a8bf99d42767a37263a0a8f8a2a07c886",
-  "worker/api.js": "65ba1c1959a75f29eb03d4eaa0b4e1d8a776d3c6003985692d11417f8680dede"
+  "web/app.js": "58d19ef97f1d53c92e1c443bd9fbb6a9d6f9eb5f88fcd3ee5924e9841a9d86db",
+  "web/index.html": "1f50b7bdeadd7a9120480fe72dae3063ef4505c0ed19f8d907311e1f42b8b4cf",
+  "web/model.js": "aeec69ec79276ecd0c159f9f1c5f508ecfa90e5f3a227bd7bd269dccad5e6506",
+  "web/style.css": "7f73d024a00868bfb8839dee225ad0df7c33fdebe2ee672074332c90bf6f1957",
+  "worker/api.js": "65ba1c1959a75f29eb03d4eaa0b4e1d8a776d3c6003985692d11417f8680dede",
+  "tests/test_request_diagnostics.py": "8486cf3443efd0a15d6341c4e65378ed547b31b4a49a173df0a863f8f31dd2c4",
+  "gpuqueue/telemetry.py": "b142ffac5ebfbe5e98074595c94a67cf1f1dda597d90d87e09879bc891854f0c",
+  "tests/resources.test.js": "5c4150a62e891f2d59f6bb2d4c7f16a9da7b0f004655224a53f0d66d3beb5e94",
+  "tests/test_monitoring.py": "5bbca3ba9ec28269d4dadba01ff77c95ffd26d75a5ab53f8d1f2650318c1f676"
 }

docs/OPERATIONS.md

@@ -129,3 +129,13 @@ Mac 的 `serverbox-gpu-status` helper 需要支持此注册表协议。注册表
 - 停止任务不等于释放实例，也不能撤回已在途的 GPU 申请。进程退出证据与资源释放回执分开；平台主动释放接口仍未开放。
 
 测试使用隔离目录和 CPU 假进程验证进程树终止、无关进程保留、产物保留、版本/来源校验、队列竞争与停止确认，不用真实 GPU 任务做取消测试。
+
+## 持续 GPU 占用监测
+
+服务按申请 UUID 自动发现本人的资源，收到就绪通知后通过已有 `mf-gpu` 通道运行只读 `nvidia-smi` 采样。型号与容量取实际读数，支持 H100、A100、4090 等 NVIDIA 显卡及多卡混合显示，不按型号写死指标；不支持 AMD/Intel 专用驱动采集。
+
+成功采样后至少间隔 30 秒发出下一次探测；实际刷新还受平台队列与通知查询耗时影响。计算任务结束后继续监测。已知资源的探测失败按 30、60、120、240、300 秒退避，成功后清除错误。超时的未完成回执保留，迟到结果可恢复，不重复投放；提交回执不明时先按唯一探测标记核查本地规范文件。平台确认回收或失败后停止探测，未确认就绪的申请不运行 GPU 命令。
+
+网页每 5 秒刷新，逐卡显示型号、GPU UUID、利用率、显存（GiB）、温度与采样时间。超过 90 秒或资源断开时标记最后读数并排除实时汇总；服务器断线也不会把旧读数显示为实时。硬件不支持的指标显示“—”。进程占用查询不可用时仍展示硬件读数，但暂停复用该资源；更换 GPU UUID 后重新积累低占用历史，避免沿用旧卡记录判断并行。
+
+采样继续发布到 `~/.local/state/serverbox-gpu/registry.json`，保持 schema 1 与现有桌面客户端兼容。更新服务器端无需重新安装 Mac 客户端。新增回归见 `tests/test_monitoring.py` 和 `tests/resources.test.js`。

gpuqueue/platform.py

@@ -1,8 +1,6 @@
 """Personal Markflow adapter. No action is called until browser approval exists."""
-import csv
 import datetime as dt
 import json
-import math
 import os
 from pathlib import Path
 import re
@@ -24,20 +22,7 @@ def reserve_mb(gpu, parallel=False):
     # margin applies only when admitting another task onto an occupied GPU.
     return max(2048, gpu['total_mb'] * .1) if parallel else 2048
 
-PROBE_SCRIPT = '''import subprocess,json,time,socket,sys
-def query(fields,kind='gpu'):
- p=subprocess.run(['nvidia-smi','--query-'+kind+'='+fields,'--format=csv,noheader,nounits'],capture_output=True,text=True,timeout=20)
- if p.returncode:raise RuntimeError('nvidia-smi query failed')
- return p.stdout.strip()
-try:
- gpus=query('index,uuid,name,memory.total,memory.used,utilization.gpu,temperature.gpu')
- at=time.time()
- apps=query('gpu_uuid,pid','compute-apps')
-except subprocess.TimeoutExpired:
- print('GPUQ_PROBE_TIMEOUT=1',flush=True)
- sys.exit(124)
-print('GPUQ_TELEMETRY='+json.dumps({'gpus':gpus,'apps':apps,'at':at,'host':socket.gethostname()}))
-'''
+from .telemetry import PROBE_SCRIPT, parse_sample
 
 class PlatformError(RuntimeError):
     pass
@@ -151,54 +136,88 @@ def submit_job(self, request_id, argv, cwd):
             raise PlatformError('GPU 任务回执不完整，已停止自动重试')
         return value['job']
 
+    def probe_failure(self, resource, message, timed_out=False):
+        failures = resource.get('probe_failures', 0)+1
+        resource.update(status='unknown', error=message, probe_failures=failures,
+                        probe_retry_at=time.time()+min(300, 30*2**min(failures-1, 4)))
+        if timed_out:
+            resource['probe_timeout'] = {'at': now(), 'job': (resource.get('probe') or {}).get('job')}
+
     def refresh_probe(self, resource):
-        if resource['status'] not in ['ready', 'online']:
+        if resource['status'] in ['pending', 'closed', 'failed']:
+            return
+        # Retry only known allocations, never ambiguous application receipts.
+        if resource['status'] == 'unknown' and not (resource.get('telemetry') or
+                resource.get('ready_message') or resource.get('ready_notice_consumed') or resource.get('probe') or resource.get('probe_submission')):
             return
+        submission = resource.get('probe_submission')
+        if submission:
+            for specification in self.job_folder(resource['id']).glob('*.json'):
+                if read_json(specification, {}).get('argv') == submission['argv']:
+                    resource['probe'] = {'job': specification.stem, 'at': submission['at']}
+                    resource.pop('probe_submission', None)
+                    break
+            if resource.get('probe_submission'):
+                return  # Unknown submit outcome: reconcile, never duplicate it.
         probe = resource.get('probe')
         if probe:
             folder = self.job_folder(resource['id']) / probe['job']
             status = read_json(folder/'status.json', {})
             if status.get('state') == 'finished' and status.get('exit_code') == 0:
-                output = (folder/'output.log').read_text(errors='replace')
-                line = next((line for line in output.splitlines() if line.startswith('GPUQ_TELEMETRY=')), None)
-                if not line:
-                    raise PlatformError('GPU 检查已结束但没有有效遥测')
-                payload = json.loads(line.split('=', 1)[1])
-                gpus = []
-                for row in csv.reader(payload['gpus'].splitlines(), skipinitialspace=True):
-                    if len(row) not in [6, 7]:
-                        raise PlatformError('GPU 遥测字段不完整')
-                    temperature = row[6].strip() if len(row) == 7 else 'N/A'
-                    gpus.append({'index': int(row[0]), 'uuid': row[1], 'name': row[2], 'total_mb': float(row[3]), 'used_mb': float(row[4]), 'utilization': float(row[5]), 'temperature_c': None if temperature in ['N/A', '[N/A]', 'Not Supported', '[Not Supported]'] else float(temperature)})
-                apps = [row for row in csv.reader(payload['apps'].splitlines(), skipinitialspace=True) if len(row) == 2]
-                stamp = float(payload.get('at', status.get('at', (folder/'output.log').stat().st_mtime)))
-                if not math.isfinite(stamp) or stamp <= 0 or stamp > time.time()+5:
-                    raise PlatformError('GPU 遥测采样时间无效')
-                resource.update(status='online', telemetry={'gpus': gpus, 'apps': apps, 'at': stamp, 'host': payload.get('host', '')}, probe=None, probe_checked=time.time())
-                resource.setdefault('samples', []).append({'at': stamp, 'utilization': max((g['utilization'] for g in gpus), default=100)})
-                resource['samples'] = [sample for sample in resource['samples'] if stamp-sample['at'] < 420]
+                try:
+                    output = (folder/'output.log').read_text(errors='replace')
+                    line = next((line for line in output.splitlines() if line.startswith('GPUQ_TELEMETRY=')), None)
+                    if not line:
+                        raise ValueError('GPU 检查已结束但没有有效遥测')
+                    payload = json.loads(line.split('=', 1)[1])
+                    telemetry = parse_sample(payload, (folder/'output.log').stat().st_mtime)
+                    if telemetry['at'] <= resource.get('telemetry', {}).get('at', 0):
+                        raise ValueError('GPU 遥测采样时间未前进')
+                except (ValueError, KeyError, TypeError, OSError) as error:
+                    self.probe_failure(resource, 'GPU 遥测无效：'+str(error))
+                    resource['probe'] = None
+                    return
+                previous = {g['uuid'] for g in resource.get('telemetry', {}).get('gpus', [])}
+                if previous != {g['uuid'] for g in telemetry['gpus']}:
+                    resource['samples'] = []
+                resource.update(status='online', telemetry=telemetry, probe=None,
+                                probe_checked=time.time(), probe_failures=0)
+                for key in ['error', 'probe_timeout', 'probe_retry_at']:
+                    resource.pop(key, None)
+                utilization = max((g['utilization'] if g['utilization'] is not None else 100
+                                   for g in telemetry['gpus']), default=100)
+                resource.setdefault('samples', []).append({'at': telemetry['at'], 'utilization': utilization})
+                resource['samples'] = [s for s in resource['samples'] if telemetry['at']-s['at'] < 420]
                 return
             if status.get('state') in ['failed', 'finished']:
                 output = folder/'output.log'
                 timed_out = (status.get('exit_code') == 124 and output.is_file() and
                              'GPUQ_PROBE_TIMEOUT=1' in output.read_text(errors='replace').splitlines())
-                resource.update(status='unknown', probe=None,
-                                error='GPU 检查超时；旧实例是否回收尚未确认' if timed_out else 'GPU 只读检查失败，无法确认资源可用')
-                if timed_out:
-                    resource['probe_timeout'] = {'at': now(), 'job': probe['job']}
+                self.probe_failure(resource, 'GPU 检查超时；旧实例是否回收尚未确认' if timed_out
+                                   else 'GPU 只读检查失败，稍后重试', timed_out)
+                resource['probe'] = None
                 return
-            if time.time()-probe['at'] > 100:
-                # Only remove a diagnostic that has not been claimed.
-                specification = self.job_folder(resource['id'])/(probe['job']+'.json')
-                if not status and specification.exists():
-                    specification.unlink()
-                resource.update(status='unknown', probe=None,
-                                probe_timeout={'at': now(), 'job': probe['job']},
-                                error='GPU 检查超时；旧实例是否回收尚未确认')
+            if time.time()-probe['at'] > 100 and resource['status'] != 'unknown':
+                # Keep the receipt: a slow or disconnected worker can complete
+                # it later. Never queue duplicate probes behind an unclaimed one.
+                self.probe_failure(resource, 'GPU 检查超时；旧实例是否回收尚未确认', True)
+            return
+        if time.time() < resource.get('probe_retry_at', 0):
             return
         if time.time()-resource.get('probe_checked', 0) >= 30:
-            job = self.submit_job(resource['id'], ['/usr/bin/python3', '-c', PROBE_SCRIPT], self.home)
+            argv = ['/usr/bin/python3', '-c', PROBE_SCRIPT+'\n# gpuq-probe '+str(uuid.uuid4())]
+            resource['probe_submission'] = {'argv': argv, 'at': time.time()}
+            self.persist()
+            try:
+                job = self.submit_job(resource['id'], argv, self.home)
+            except (PlatformError, OSError, subprocess.SubprocessError) as error:
+                if isinstance(error, OSError):
+                    resource.pop('probe_submission', None)  # Child never started.
+                self.probe_failure(resource, 'GPU 只读检查暂不可用：'+str(error))
+                self.persist()
+                raise
             resource['probe'] = {'job': job, 'at': time.time()}
+            resource.pop('probe_submission', None)
             self.persist()
 
     def refresh(self, active=False):
@@ -211,13 +230,20 @@ def refresh(self, active=False):
             for task in self.store.tasks():
                 replaceable.update(r['id'] for r in self.replacement_candidates(task))
                 replaceable.update(task['internal'].get('timeout_replacement', {}).get('prior_requests', []))
+            # Collect all outstanding receipts before potentially slow notice I/O.
+            for resource in list(self.resources.values()):
+                if resource.get('probe'):
+                    try:
+                        self.refresh_probe(resource)
+                    except (PlatformError, OSError, subprocess.SubprocessError, ValueError) as error:
+                        errors.append(error)
             for resource in list(self.resources.values()):
                 # A failed notification query must not prevent reading a probe
                 # already completed in shared HOME, or skip other allocations.
                 for refresh in (self.refresh_notice, self.refresh_probe):
                     try:
                         refresh(resource)
-                    except (PlatformError, OSError, subprocess.SubprocessError) as error:
+                    except (PlatformError, OSError, subprocess.SubprocessError, ValueError) as error:
                         if refresh == self.refresh_notice and resource['id'] in replaceable and self.probe_timed_out(resource):
                             # An unreachable historical notice must not veto the
                             # user's timeout policy or block a newer allocation.
@@ -243,7 +269,7 @@ def request(self, task):
         required = task['doc']['resources']
         # A known smaller allocation may be insufficient. Unknown or busy usable
         # allocations must be reconciled rather than treated as absent.
-        if any(r['status'] != 'online' or time.time()-r.get('telemetry', {}).get('at', 0)>90 or len([g for g in r.get('telemetry', {}).get('gpus', []) if g['total_mb'] >= memory_mb(required)+reserve_mb(g)]) >= required['gpus'] and r.get('type') == required['type'] for r in unresolved):
+        if any(r['status'] != 'online' or time.time()-r.get('telemetry', {}).get('at', 0)>90 or len([g for g in r.get('telemetry', {}).get('gpus', []) if g.get('total_mb') is not None and g['total_mb'] >= memory_mb(required)+reserve_mb(g)]) >= required['gpus'] and r.get('type') == required['type'] for r in unresolved):
             return None
         if read_json(self.store.root/'tasks'/task['id']/'run/stop.json',{}).get('digest') == task['digest']:
             return None
@@ -305,7 +331,7 @@ def select(self, task, active):
             return None, '默认顺序执行；未满足安全并行条件'
         for resource in self.resources.values():
             telemetry = resource.get('telemetry', {})
-            if resource['status'] != 'online' or time.time()-telemetry.get('at', 0) > 90 or resource.get('type') != required['type']:
+            if resource['status'] != 'online' or time.time()-telemetry.get('at', 0) > 90 or resource.get('type') != required['type'] or telemetry.get('apps_known') is False:
                 continue
             if active:
                 samples = [s for s in resource.get('samples', []) if time.time()-s['at'] < 330]
@@ -321,7 +347,7 @@ def select(self, task, active):
                 managed_pids.update(str(s['pid']) for s in status.get('steps', []) if s.get('state') == 'running' and s.get('pid'))
             occupied = {row[0] for row in telemetry.get('apps', []) if row[1] not in managed_pids}
             required_mb = memory_mb(required)
-            available = [g for g in telemetry.get('gpus', []) if g['uuid'] not in occupied and g['total_mb']-g['used_mb'] >= required_mb+reserve_mb(g, parallel=bool(active))]
+            available = [g for g in telemetry.get('gpus', []) if g.get('total_mb') is not None and g.get('used_mb') is not None and g['uuid'] not in occupied and g['total_mb']-g['used_mb'] >= required_mb+reserve_mb(g, parallel=bool(active))]
             available.sort(key=lambda gpu: gpu['total_mb']-gpu['used_mb'], reverse=True)
             if len(available) >= required['gpus']:
                 return (resource, [g['uuid'] for g in available[:required['gpus']]]), ''