Add agent scope observation and advice tools

README.md

@@ -145,6 +145,9 @@ WaveBench 主包长期预装 RTM2000/RTM2032、DS1104Z/DS1000Z、DG4000/DG4202
   - `run.schema`：返回 run plan schema
   - `run.check`：只解析并检查 `plans/*.toml` 下的 run plan，不连接仪器
   - `capture.inspect`：读取 `data/raw/` 下的离线采集包摘要
+  - `scope.observe`：默认只读连接配置中的示波器，返回 IDN、状态快照（若驱动支持）和高阻安全判断；支持 CH1-CH4 多通道观察；可传 `fetch_waveform=true` 读取一个或多个通道的当前波形摘要，但该模式可能改动示波器的波形传输源/模式，响应会标注 `read_only=false`。多通道 fetch 生成的 pairwise `relationships` 会标注 `same_acquisition=false`，相位等时序关系只作为建议线索
+  - `scope.advise`：基于 `scope.observe` 给出每通道时基/垂直档位/display 建议；只返回建议，不应用调整
+  - `doctor.config`：结构化返回配置中各仪器的只读可达性、IDN 和型号匹配检查结果
 - `/mcp` 与 `/call` 的 JSON 请求体有 1 MiB 上限；路径参数按工具限制在项目内固定目录
 
 ## 安全默认值

TODO.md

@@ -0,0 +1,48 @@
+# WaveBench AI-Agent Experience TODO
+
+目标：让 AI agent 能以低风险、结构化、可审计的方式观察实验台、理解已有产物、提出建议，并只通过显式 run plan 执行会改变仪器状态的动作。
+
+## P0: 安全只读上下文
+
+- [x] 增加 MCP 工具 `scope.observe`：读取配置中的示波器，返回 IDN、状态快照、通道高阻判断、当前波形摘要；不保存文件、不改仪器状态、不暴露 raw SCPI。
+- [x] `scope.observe` 支持 CH1-CH4 多通道观察；显式 `fetch_waveform=true` 且获取到两个以上通道时返回 pairwise relationships。
+- [x] 增加 MCP 工具 `doctor.config`：把现有 `doctor` 结果结构化返回给 agent，支持配置可达性和型号匹配判断。
+- [ ] 增加 MCP 工具 `net.discover`：结构化返回只读 LAN 仪器发现结果，默认限制网段大小和端口集合。
+- [ ] 所有 MCP 工具都标注 `read_only`、`mutates_instrument=false`、`raw_scpi=false` 等安全元数据。
+
+## P1: 离线产物理解
+
+- [ ] 增强 `capture.inspect`：可选返回 FFT 摘要、质量告警、推荐下一步采集参数，但不读取大数组进响应。
+- [ ] 增加 `capture.list`：列出 `data/raw` 下最近采集包，供 agent 找上下文。
+- [ ] 增加 `run.list` / `run.inspect`：列出和解释 `data/runs` 下的实验记录。
+- [ ] 给采集包生成 agent 友好的 `summary.md` 或结构化 `analysis.json`。
+
+## P2: Agent 建议层
+
+- [x] 增加 `scope.advise`：基于 `scope.observe` 的结果给出时基、垂直档位、触发、点数建议；只返回建议，不应用。
+  - [x] 根据实测或期望频率推荐每通道 focus time-range，默认约 10 个周期。
+  - [x] 根据实测 Vpp 推荐垂直档位，默认约占 5 格。
+  - [x] 识别 CH1/CH2 这类大频率跨度，建议分通道/分 profile 观察，避免单时基误判形状。
+- [x] 增加 `scope.observe.expectations`：把已知闭环信号作为结构化依据，例如 CH1 1 kHz/1 Vpp/50% 方波、CH2 50 kHz/1 Vpp/500 mVdc/30% 对称三角波，并返回逐项 pass/warn/fail。
+  - [x] 支持 frequency/Vpp/mean/duty/symmetry_percent 的基础断言。
+  - [x] 期望断言必须显式 `fetch_waveform=true`，避免 agent 在不知情时触发波形传输状态变化。
+  - [x] 返回总体 expectation status，并把 fail/warn 加入 agent hints。
+- [x] 增加多时基建议：当多个通道的频率跨度较大时，提醒 agent 不要用单个显示时基同时判断所有通道形状，应分通道或分 profile 采集。
+- [x] 多通道关系分析增加交点：对每对通道返回交点数量、采样返回点、交点时间/电压和相对斜率方向；交点过多时截断并给 warning。
+- [ ] 让 `scope.expect` 也参与多时基建议：即使当前频率估计低置信，也能利用用户给定的期望频率识别 CH1/CH2 这种 50x 频率跨度。
+- [ ] 增加 `plan.propose`：从自然语言目标或结构化目标生成保守 run plan 草案，默认写到 `plans/`，不执行。
+- [ ] 增加 `plan.explain`：解释某个 run plan 会读写哪些仪器、哪些步骤会改变输出状态、有哪些保护。
+
+## P3: 可控执行边界
+
+- [ ] MCP 继续保持默认只读；会改仪器状态的能力只通过显式 `run plan` 文件和人工确认入口暴露。
+- [x] 增加显式示波器显示控制 CLI：通道显示 on/off、focus 单通道、显式 autoscale；用于人类和 agent 做可审计调参。
+- [x] focus 动作只调整示波器显示/采集窗口，不改变信号源、电源或被测对象；执行后输出 mutation manifest。
+- [ ] 对所有会改状态的 plan 步骤生成 mutation manifest，便于 agent 在执行前向人类说明。
+- [ ] 增加跨进程仪器锁，避免 Windows/WSL/多个 agent 同时打开同一台仪器导致响应串线。
+
+## P4: 人类与 Agent 共用体验
+
+- [x] 增加 `scripts/wsl-run.ps1` 作为 Windows 到 WSL 的标准执行入口，并在 README 中记录。
+- [ ] 为 WSL/Windows 推荐环境增加 `doctor environment` 或 `env doctor`，检查 Python、PyVISA、WSL、网络可达性。
+- [ ] 增加最小可视化：采集包自动生成 waveform/FFT PNG，报告中可直接查看。

doc/project/WaveBench_HTTP_MCP_只读接口.md

@@ -51,16 +51,21 @@ MCP notification 请求没有 `id` 时返回空响应；普通请求返回 JSON-
 - `run.schema`：返回 run plan schema 文本和结构化 schema 行。
 - `run.check`：参数 `{"plan": "plans/<name>.toml"}`，只解析并检查 `plans/*.toml` 下的 run plan，不连接仪器。
 - `capture.inspect`：参数 `{"path": "data/raw/<capture_dir>"}`，读取 `data/raw/` 下的离线采集包摘要。
+- `scope.observe`：参数 `{"channels": [1, 2], "fetch_waveform": false}`，只读连接配置中的示波器，返回 IDN、状态快照（若驱动支持）和高阻安全判断。支持 CH1-CH4 多通道观察。显式传 `fetch_waveform=true` 时，会读取一个或多个通道的当前波形摘要；当至少两个通道读取成功时，返回 `relationships`，包含频率比、Vpp/均值关系、公共时间窗、归一化相关性、估计延迟、同频相位和通道间交点。该工具不保存文件，不暴露 raw SCPI，但抓波形可能改动示波器的波形传输源/模式。
+  - 可选 `expectations`：例如 `{"1": {"frequency_hz": 1000, "vpp_v": 1.0, "duty_percent": 50}, "2": {"frequency_hz": 50000, "vpp_v": 1.0, "mean_v": 0.5, "symmetry_percent": 30}}`。使用 expectations 必须同时传 `fetch_waveform=true`。
+- `scope.advise`：参数同 `scope.observe`，另有 `target_cycles` 和 `target_vertical_divisions`。它基于当前观察和可选 expectations 给出 `scope focus` / `scope display` 建议、每通道推荐时基窗口和垂直档位；只返回建议，绝不应用调整。若传 `fetch_waveform=true`，同样可能改动示波器的波形传输源/模式。
+- `doctor.config`：参数 `{"timeout_ms": 1000}`，结构化返回配置中各仪器的只读可达性、IDN 和型号匹配检查结果；不执行网段发现。
 
 ## 安全边界
 
 - 默认只监听 `127.0.0.1`。
 - 拒绝监听 `0.0.0.0`。
 - `/mcp`、`/tools` 和 `/call` 强制 Bearer token。
-- 当前工具全部只读。
+- 当前工具不提供 raw SCPI，不应用显示/输出/采集建议；`scope.observe` / `scope.advise` 在 `fetch_waveform=true` 时会显式标注可能的波形传输状态影响。
 - 不提供 raw SCPI。
 - 不提供 power/source output on/off。
 - 不提供 run 执行工具。
 - `run.check` 只允许项目内 `plans/*.toml`。
 - `capture.inspect` 只允许项目内 `data/raw/` 离线采集包。
+- `doctor.config` 只检查当前配置中的资源，不扫描网段。
 - `/mcp` 和 `/call` 的 JSON 请求体有 1 MiB 上限。

src/wavebench/data/expectations.py

@@ -0,0 +1,282 @@
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+
+from wavebench.instruments.models import WaveformData
+
+
+def evaluate_waveform_expectation(
+    waveform: WaveformData,
+    expectation: dict[str, Any],
+) -> dict[str, Any]:
+    summary = waveform.summary(
+        expected_frequency_hz=_optional_positive_float(expectation, "frequency_hz"),
+        frequency_tolerance_ratio=float(expectation.get("frequency_tolerance_ratio", 0.05)),
+    )
+    checks: list[dict[str, Any]] = []
+    _check_frequency(summary, expectation, checks)
+    _check_vpp(summary, expectation, checks)
+    _check_mean(summary, expectation, checks)
+    _check_duty(summary, expectation, checks)
+    _check_symmetry(waveform, expectation, checks)
+    statuses = {check["status"] for check in checks}
+    if "fail" in statuses:
+        status = "fail"
+    elif "warn" in statuses:
+        status = "warn"
+    else:
+        status = "pass"
+    return {
+        "status": status,
+        "channel": waveform.channel,
+        "label": expectation.get("label"),
+        "shape": expectation.get("shape"),
+        "checks": checks,
+    }
+
+
+def expectation_summary(results: dict[int, dict[str, Any]]) -> dict[str, Any]:
+    statuses = {result["status"] for result in results.values()}
+    if "fail" in statuses:
+        status = "fail"
+    elif "warn" in statuses:
+        status = "warn"
+    else:
+        status = "pass" if results else "skipped"
+    return {
+        "status": status,
+        "channels": {str(channel): result["status"] for channel, result in sorted(results.items())},
+    }
+
+
+def estimate_triangle_symmetry_percent(waveform: WaveformData) -> float | None:
+    times = waveform.times_s
+    values = np.asarray(waveform.voltages_v, dtype=np.float64)
+    if times.size != values.size or values.size < 8:
+        return None
+    span = float(np.max(values) - np.min(values))
+    if span <= 1e-12:
+        return None
+    centered = values - float(np.mean(values))
+    diffs = np.diff(centered)
+    if diffs.size < 3:
+        return None
+    signs = np.sign(diffs)
+    for index in range(1, signs.size):
+        if signs[index] == 0:
+            signs[index] = signs[index - 1]
+    maxima = [
+        index
+        for index in range(1, values.size - 1)
+        if signs[index - 1] > 0 and signs[index] < 0
+    ]
+    minima = [
+        index
+        for index in range(1, values.size - 1)
+        if signs[index - 1] < 0 and signs[index] > 0
+    ]
+    fractions: list[float] = []
+    for left_min, right_min in zip(minima, minima[1:]):
+        if right_min <= left_min:
+            continue
+        peaks = [index for index in maxima if left_min < index < right_min]
+        if not peaks:
+            continue
+        peak = max(peaks, key=lambda index: values[index])
+        period = float(times[right_min] - times[left_min])
+        if period <= 0:
+            continue
+        fractions.append(float((times[peak] - times[left_min]) / period * 100.0))
+    if not fractions:
+        return None
+    return float(np.median(np.asarray(fractions, dtype=np.float64)))
+
+
+def _check_frequency(
+    summary: dict[str, Any],
+    expectation: dict[str, Any],
+    checks: list[dict[str, Any]],
+) -> None:
+    expected = _optional_positive_float(expectation, "frequency_hz")
+    if expected is None:
+        return
+    actual = summary.get("frequency_estimate_hz")
+    tolerance = float(expectation.get("frequency_tolerance_ratio", 0.05))
+    low_confidence = any(
+        str(item).startswith("low_cycle_count")
+        for item in summary.get("quality_warnings", [])
+    )
+    if not isinstance(actual, (int, float)) or actual <= 0:
+        checks.append(_check("frequency_hz", "warn", expected, actual, "frequency unavailable"))
+        return
+    error_ratio = abs(float(actual) - expected) / expected
+    if low_confidence:
+        checks.append(
+            _check(
+                "frequency_hz",
+                "warn",
+                expected,
+                float(actual),
+                "frequency low confidence because waveform contains too few cycles",
+                error_ratio=error_ratio,
+                tolerance_ratio=tolerance,
+            )
+        )
+        return
+    checks.append(
+        _check(
+            "frequency_hz",
+            "pass" if error_ratio <= tolerance else "fail",
+            expected,
+            float(actual),
+            "ok" if error_ratio <= tolerance else "frequency out of tolerance",
+            error_ratio=error_ratio,
+            tolerance_ratio=tolerance,
+        )
+    )
+
+
+def _check_vpp(
+    summary: dict[str, Any],
+    expectation: dict[str, Any],
+    checks: list[dict[str, Any]],
+) -> None:
+    expected = _optional_positive_float(expectation, "vpp_v")
+    if expected is None:
+        return
+    actual = summary.get("voltage_vpp_v")
+    tolerance = float(expectation.get("vpp_tolerance_ratio", 0.10))
+    if not isinstance(actual, (int, float)):
+        checks.append(_check("vpp_v", "warn", expected, actual, "Vpp unavailable"))
+        return
+    error_ratio = abs(float(actual) - expected) / expected
+    checks.append(
+        _check(
+            "vpp_v",
+            "pass" if error_ratio <= tolerance else "fail",
+            expected,
+            float(actual),
+            "ok" if error_ratio <= tolerance else "Vpp out of tolerance",
+            error_ratio=error_ratio,
+            tolerance_ratio=tolerance,
+        )
+    )
+
+
+def _check_mean(
+    summary: dict[str, Any],
+    expectation: dict[str, Any],
+    checks: list[dict[str, Any]],
+) -> None:
+    expected = _optional_float(expectation, "mean_v")
+    if expected is None:
+        expected = _optional_float(expectation, "offset_v")
+    if expected is None:
+        return
+    actual = summary.get("voltage_mean_v")
+    tolerance = float(expectation.get("mean_tolerance_v", 0.05))
+    if not isinstance(actual, (int, float)):
+        checks.append(_check("mean_v", "warn", expected, actual, "mean unavailable"))
+        return
+    error = abs(float(actual) - expected)
+    checks.append(
+        _check(
+            "mean_v",
+            "pass" if error <= tolerance else "fail",
+            expected,
+            float(actual),
+            "ok" if error <= tolerance else "mean out of tolerance",
+            error_abs=error,
+            tolerance_abs=tolerance,
+        )
+    )
+
+
+def _check_duty(
+    summary: dict[str, Any],
+    expectation: dict[str, Any],
+    checks: list[dict[str, Any]],
+) -> None:
+    expected = _optional_float(expectation, "duty_cycle")
+    if expected is None and "duty_percent" in expectation:
+        expected = _optional_float(expectation, "duty_percent")
+        if expected is not None:
+            expected /= 100.0
+    if expected is None:
+        return
+    actual = summary.get("duty_cycle")
+    tolerance = float(expectation.get("duty_tolerance", 0.05))
+    if not isinstance(actual, (int, float)):
+        checks.append(_check("duty_cycle", "warn", expected, actual, "duty unavailable"))
+        return
+    error = abs(float(actual) - expected)
+    checks.append(
+        _check(
+            "duty_cycle",
+            "pass" if error <= tolerance else "fail",
+            expected,
+            float(actual),
+            "ok" if error <= tolerance else "duty out of tolerance",
+            error_abs=error,
+            tolerance_abs=tolerance,
+        )
+    )
+
+
+def _check_symmetry(
+    waveform: WaveformData,
+    expectation: dict[str, Any],
+    checks: list[dict[str, Any]],
+) -> None:
+    expected = _optional_float(expectation, "symmetry_percent")
+    if expected is None:
+        return
+    actual = estimate_triangle_symmetry_percent(waveform)
+    tolerance = float(expectation.get("symmetry_tolerance_percent", 5.0))
+    if actual is None:
+        checks.append(_check("symmetry_percent", "warn", expected, actual, "symmetry unavailable"))
+        return
+    error = abs(actual - expected)
+    checks.append(
+        _check(
+            "symmetry_percent",
+            "pass" if error <= tolerance else "fail",
+            expected,
+            actual,
+            "ok" if error <= tolerance else "symmetry out of tolerance",
+            error_abs=error,
+            tolerance_abs=tolerance,
+        )
+    )
+
+
+def _check(name: str, status: str, expected: Any, actual: Any, message: str, **extra: Any) -> dict[str, Any]:
+    return {
+        "metric": name,
+        "status": status,
+        "expected": expected,
+        "actual": actual,
+        "message": message,
+        **extra,
+    }
+
+
+def _optional_float(data: dict[str, Any], name: str) -> float | None:
+    if name not in data or data[name] is None:
+        return None
+    value = data[name]
+    if isinstance(value, bool):
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _optional_positive_float(data: dict[str, Any], name: str) -> float | None:
+    value = _optional_float(data, name)
+    if value is None or value <= 0:
+        return None
+    return value