From 63370213e00c1b3b4e99d945a359b5142662d99c Mon Sep 17 00:00:00 2001
From: Nept Epslion <11226778+Nept-Epslion@user.noreply.gitee.com>
Date: Tue, 6 Oct 2026 00:18:59 +0800
Subject: [PATCH] fix(scope): hold observation leases and fail closed before
 capture

---
 docs/development/contributing.md          |   2 +
 docs/development/scope-observation.md     |  85 +++
 docs/how-to/serve-mcp.md                  |  15 +-
 docs/reference/cli.md                     |  35 +-
 mkdocs.yml                                |   1 +
 src/wavebench/data/relationships.py       |  11 +-
 src/wavebench/instruments/factory.py      |  13 +-
 src/wavebench/services/agent_advise.py    |  57 +-
 src/wavebench/services/agent_observe.py   | 171 ++++--
 src/wavebench/services/scope_service.py   | 184 ++++++-
 tests/test_agent_advise.py                |  94 ++++
 tests/test_agent_observe.py               |  36 +-
 tests/test_scope_observation_contracts.py | 616 ++++++++++++++++++++++
 tests/test_scope_observe_cli.py           |  28 +-
 tests/test_waveform_binary_factory.py     |  12 +-
 tests/test_waveform_relationships.py      |  37 +-
 16 files changed, 1273 insertions(+), 124 deletions(-)
 create mode 100644 docs/development/scope-observation.md
 create mode 100644 tests/test_scope_observation_contracts.py

diff --git a/docs/development/contributing.md b/docs/development/contributing.md
index 68e53d21..3b62b6f3 100644
--- a/docs/development/contributing.md
+++ b/docs/development/contributing.md
@@ -13,3 +13,5 @@ WaveBench 的代码、文档、schema 和测试在同一仓库中维护。提交
 按[测试说明](testing.md)选择检查：局部修改先做聚焦验证，文档或 Skill 修改检查相关内容，跨模块或合并评估执行集成检查。远端 CI 仍按仓库 workflow 执行，不要求每个编辑步骤重跑全量测试。
 
 新增或修改用户可见行为时，更新唯一 canonical Reference，并用[文档工作流](documentation.md)进行 scoped review。插件专用流程见[插件开发](plugin-development.md)。
+
+尚未发布的示波器观察入口及执行验收合同见[开发线示波器观察](scope-observation.md)。
diff --git a/docs/development/scope-observation.md b/docs/development/scope-observation.md
new file mode 100644
index 00000000..2b7447fc
--- /dev/null
+++ b/docs/development/scope-observation.md
@@ -0,0 +1,85 @@
+# 开发线示波器观察
+
+> 状态：`Proposed / Future`，开发线已实现、尚未正式发布。
+> 本页面向实现与测试人员，记录开发线的执行合同；不表示已安装的正式版本提供这些入口。
+
+## 实现入口与职责
+
+CLI 的 `scope observe` 和 MCP 的 `scope.observe` / `scope.advise` 由
+`services/agent_observe.py`、`services/agent_advise.py`、`ScopeService` 与 `mcp_http.py`
+共同实现。CLI 参数以本开发线的 `cli_parser.py` 和离线 `--help` 为准。
+Core 表达观察、访问约束、会话和报告语义；型号通道、端接与传输差异由 descriptor 和 driver
+提供，不在观察层维护型号范围表。
+
+MCP 的 `doctor.config` 同属未发布入口，调用 `doctor_records()` 并返回结构化检查记录。
+它不读取波形或执行建议；实际查询合同以 doctor 与 driver 实现为准。
+
+## 观察与建议
+
+默认 `scope observe` 只获取身份、可用状态与输入耦合，不读取波形。观察路径收紧有效访问
+权限，原配置为 `disabled` 时在打开连接前拒绝。只有具备纯查询合同的状态才进入结果；缺少字段或
+capability 时保留不可用原因，不把部分状态当作完整快照。
+
+MCP `scope.advise` 依据可用状态和调用方的 `expected_frequencies_hz` 提供显示或时基建议，
+不会执行建议。无可用档位、测量值或期望频率时返回逐通道 `advice_unavailable`，表示无法
+评估当前设置。查询失败不能生成「无需调整」的判断。调用方提供的频率只标为 configured，
+不能当作实测证据；带低周期告警的测量频率不能直接用于时基建议。
+
+生成的 focus 建议遵循现有 CLI 格式：`--vertical-scale CHANNEL=V_PER_DIV`；
+隐藏其它通道使用 `--hide-others`。建议的执行仍须经过命令自身的 access、capability 和
+运行时安全检查。
+
+## 显式波形报告与预检
+
+`scope observe --fetch-waveform` 会操作仪器，可能停止采集、启用通道显示、改变波形传输
+source/mode/format/points，以及按配置消费错误队列。它不恢复原来的运行状态。MCP 不提供
+这一读取路径。
+
+`--target-cycles` 与 `--target-vertical-divisions` 必须是有限正数；expectation 的字段名、
+类型、有限性与取值范围由 `data/expectations.py` 校验。这些输入在创建仪器服务前拒绝。
+型号相关的通道支持验证可能需要连接后的纯查询预检；全部请求通道必须在第一次采集写入
+前完成验证。它不是「任意型号输入错误都在打开会话前拒绝」的保证。
+
+离线 access、capability 或波形配置预检失败时不构造 driver 或 transport。观察路径的
+factory 构造阶段对所有 descriptor 启用 Core I/O 锁，禁止通过 context transport 查询或
+写入；构造与声明校验完成后才释放。需要在构造阶段进行设备初始化的旧插件不能使用这一
+入口，应将设备操作移入明确的执行方法。该约束只保护 Core transport，不提供 Python 沙箱。
+
+报告在受控持续会话和独占资源租约内执行。最终高阻确认与波形读取共用会话和租约，避免
+遵守同一资源锁的其它进程在两者之间改变输入设置；资源锁不能阻止前面板或外部软件操作。
+不确定 I/O 或会话健康故障使剩余采集停止，并记录中止原因；不自动重连继续写入。
+
+持续会话不证明波形来自同一次 acquisition。报告仍将 `same_acquisition` 标为 false，
+跳过相位、相关性、延迟和交点，保留同步无关摘要。正式同步时序分析应使用
+`scope capture --synchronized` 产物和既有 `analysis.pair` 同步证明合同。
+`data.relationships` 的默认值同样不认定同步；其显式断言仅供已确认来源的内部分析。
+
+## 期望检查与输出
+
+`--expect <file.toml>` 需要显式波形报告；示例格式为：
+
+```toml
+[channels.1]
+frequency_hz = 1000
+frequency_tolerance_ratio = 0.05
+vpp_v = 3.3
+duty_percent = 50
+```
+
+汇总保留全部待验收通道。无法取得或评估波形时通道为 `unavailable`；全部不可用时总体为
+`unavailable`，有可用验收但不完整时为 `partial`，已确认的 `fail` 优先。
+无可执行期望指标时为 `skipped`。报告成功返回不等于所有期望通过，调用方须检查验收汇总
+以及 warnings 和逐通道结果。
+
+## 维护与验收
+
+聚焦验证覆盖生命周期关闭与借用、跨进程租约竞争、非法后续通道零采集写入、会话故障后的
+剩余通道中止、禁用配置及离线失败零构造、legacy factory I/O 拦截、真实 guard 的观察零写，
+以及缺证据建议与同步默认值。测试只使用 fake
+transport 和合成信号，不连接设备；实机 evidence 属于单独授权的插件验证。
+
+发布时依据实际 tag 和实现将正式可用行为转入 CLI Reference 和 MCP How-to，并保留唯一
+事实来源。本页不提前承诺发布版本、插件支持范围或硬件验收结论。
+
+相关合同见[安全模型](../concepts/safety-model.md)、[会话与恢复](../concepts/sessions-and-recovery.md)、
+[插件模型](../concepts/plugin-model.md)和[测试说明](testing.md)。
diff --git a/docs/how-to/serve-mcp.md b/docs/how-to/serve-mcp.md
index 02f9437f..f7f04cdd 100644
--- a/docs/how-to/serve-mcp.md
+++ b/docs/how-to/serve-mcp.md
@@ -1,6 +1,6 @@
 # 启动只读 MCP 服务
 
-WaveBench HTTP MCP 服务提供本机或受控网络中的离线信息和只读仪器观察。它不提供 raw SCPI、输出控制或 run 执行。
+WaveBench HTTP MCP 服务提供本机或受控网络中的离线信息和 run plan 检查。它不提供 raw SCPI、输出控制或 run 执行。
 
 ## 启动服务
 
@@ -20,23 +20,16 @@ python -m wavebench mcp serve \
 | `GET /tools` | Bearer token | 列出只读工具。 |
 | `POST /call`、`POST /mcp` | Bearer token | 调用 MCP 工具或 JSON-RPC 方法。 |
 
-请求体上限为 1 MiB。所有工具都是只读的，不会改变仪器状态，也不会读取波形。工具的权威列表和元数据以 `GET /tools` 返回为准：
+请求体上限为 1 MiB。已发布的工具不连接仪器；安装版本的实际工具列表以 `GET /tools` 返回为准：
 
 | 工具 | 行为 |
 | --- | --- |
 | `run.schema` | 返回 run plan schema。 |
 | `run.check` | 只接受项目内的 `plans/*.toml`，离线解析并校验 run plan。 |
 | `capture.inspect` | 只读取项目内的离线采集包摘要。 |
-| `doctor.config` | 对配置中的仪器执行只读 doctor 检查，返回结构化记录。 |
-| `scope.observe` | 读取示波器身份、每通道状态快照和输入耦合安全；不读取波形。 |
-| `scope.advise` | 基于只读状态快照和调用方给出的 `expected_frequencies_hz` 建议显示/采集参数；不应用建议。 |
 
-`doctor.config`、`scope.observe` 和 `scope.advise` 是实验性工具：开发线已实现并有离线测试，但尚未随正式版本发布，
-支持范围不作承诺。
-
-`scope.observe` 和 `scope.advise` 不读取示波器波形。波形摘要、期望值检查和跨通道关系分析需要
-先显式读取波形（属于会改变仪器状态的写路径），请在操作者明确执行
-[`scope observe --fetch-waveform`](../reference/cli.md) 时进行，不通过 MCP 暴露。
+开发线新增的 `doctor.config`、`scope.observe` 和 `scope.advise` 尚未随正式版本发布，
+不构成本页的功能承诺。其实施边界见[开发线示波器观察](../development/scope-observation.md)。
 
 ## Verification
 
diff --git a/docs/reference/cli.md b/docs/reference/cli.md
index c5362125..0f027a55 100644
--- a/docs/reference/cli.md
+++ b/docs/reference/cli.md
@@ -18,40 +18,13 @@ python -m wavebench --json <domain> <command> ...
 | --- | --- | --- |
 | 离线、只读本地文件 | `run schema`、`run check`、`run intent`、`run compare`、`run resume`、`capture inspect`、`capability explain`、`lock status` | 不连接仪器。 |
 | 离线且可能写本地文件 | `run template --output`、`run report`、`run report-index` | 不连接仪器，但会创建模板、报告或索引文件。 |
-| 连接读取或预检 | `doctor`、状态／身份查询、`scope observe`（不带 `--fetch-waveform`）、`run verify` | 会访问配置的仪器，不应改变实验设置。 |
-| 可能改变状态或触发采集 | 输出和 setter、`scope auto`／`scope capture`、`scope observe --fetch-waveform`、非 fake TUI、`run plan` | 可能写入仪器、触发采集或切换输出。 |
+| 连接读取或预检 | `doctor`、状态／身份查询、`run verify` | 会访问配置的仪器，不应改变实验设置。 |
+| 可能改变状态或触发采集 | 输出和 setter、`scope auto`／`scope capture`、非 fake TUI、`run plan` | 可能写入仪器、触发采集或切换输出。 |
 
 `scope fetch` 读取已有波形，但仍是仪器 I/O；不要把它当作离线命令。每次硬件操作前确认接线、输入阻抗、输出状态和安全限制。WaveBench 不会自动执行 `*RST`，也不会因设置电压、幅度或频率而自动开启输出。
 
-`scope observe` 是实验性命令：开发线已实现并有离线测试，但尚未随正式版本发布，兼容性和支持范围不作承诺。
-它默认只读：只查询身份、通道状态快照和输入耦合安全，不读取波形。
-`scope observe --fetch-waveform` 是显式写路径，可能停止正在运行的采集、修改波形传输
-source/mode/format/points 并打开通道显示；它逐通道读取波形，因此多通道结果不保证来自同一次
-acquisition，此时跨通道的相位、相关性、延迟和交点不会被计算（`correlation`、`intersections`
-返回 `skipped`，相位为 `null`）。需要驱动可证明的同一次采集时，使用
-`scope capture --synchronized`（通道和输出格式要求见其 `--help`）。
-
-期望值检查通过 `--expect <file.toml>` 提供，需要 `--fetch-waveform`：
-
-```toml
-[channels.1]
-frequency_hz = 1000
-frequency_tolerance_ratio = 0.05
-vpp_v = 3.3
-duty_percent = 50
-```
-
-字段名、类型、有限性和取值范围由实现严格校验（`src/wavebench/data/expectations.py` 的 `validate_expectation()`）；
-拼错字段名会直接报错，不会被静默忽略。上面的 TOML 只示范格式，字段全集以该实现为准。任何输入错误都在
-打开仪器会话之前被拒绝，因此不会产生仪器写入。
-
-`--target-cycles` 和 `--target-vertical-divisions` 必须为有限正数，并在加载配置或创建仪器服务
-之前校验。生成的 focus 建议使用 `--vertical-scale CHANNEL=V_PER_DIV`；隐藏其他通道的参数为
-`--hide-others`。建议不会自动执行。
-
-期望值汇总的 `channels` 保留所有待验收通道。波形读取、安全检查或期望值计算失败的通道标记为
-`unavailable`；没有可用检查结果时汇总为 `unavailable`，部分通道已有 `pass`／`warn` 结果时为
-`partial`。已确认的 `fail` 仍优先返回 `fail`。未提供期望值或期望值没有可执行指标时保持 `skipped`。
+`scope observe` 尚未随正式版本发布，不属于本页的已发布命令承诺。开发线的参数、输出和
+副作用合同见[开发线示波器观察](../development/scope-observation.md)。
 
 ## JSON 输出与退出码
 
diff --git a/mkdocs.yml b/mkdocs.yml
index dd6489d9..03c5801b 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -70,6 +70,7 @@ nav:
   - 开发:
       - 贡献: development/contributing.md
       - 测试: development/testing.md
+      - 开发线示波器观察: development/scope-observation.md
       - 文档工作流: development/documentation.md
       - 插件开发: development/plugin-development.md
       - 新增仪器驱动: development/instrument-drivers.md
diff --git a/src/wavebench/data/relationships.py b/src/wavebench/data/relationships.py
index 943a92cb..4aed22fd 100644
--- a/src/wavebench/data/relationships.py
+++ b/src/wavebench/data/relationships.py
@@ -12,7 +12,7 @@
 def analyze_waveform_relationships(
     waveforms: dict[int, WaveformData],
     *,
-    same_acquisition: bool = True,
+    same_acquisition: bool = False,
     max_correlation_points: int = 4096,
     max_intersections: int = 64,
 ) -> list[dict[str, Any]]:
@@ -34,10 +34,17 @@ def analyze_waveform_pair(
     left: WaveformData,
     right: WaveformData,
     *,
-    same_acquisition: bool = True,
+    same_acquisition: bool = False,
     max_correlation_points: int = 4096,
     max_intersections: int = 64,
 ) -> dict[str, Any]:
+    """Summarize waveforms; timing requires an explicit shared-acquisition assertion.
+
+    This helper does not validate capture provenance. Real capture timing analysis
+    must use the validated synchronization contract in ``pair_analysis``.
+    """
+    if not isinstance(same_acquisition, bool):
+        raise ValueError("same_acquisition must be a boolean")
     left_summary = left.summary()
     right_summary = right.summary()
     warnings: list[str] = []
diff --git a/src/wavebench/instruments/factory.py b/src/wavebench/instruments/factory.py
index 6cd26a2a..87a9735d 100644
--- a/src/wavebench/instruments/factory.py
+++ b/src/wavebench/instruments/factory.py
@@ -58,7 +58,16 @@ def open_instrument_driver(
     serial_config: DmmConfig | None = None,
     access: AccessMode = "read_write",
     lease: ResourceLease | None = None,
+    force_deferred_io: bool = False,
 ) -> OpenedInstrument:
+    """Construct and validate a driver, optionally deferring all guarded I/O.
+
+    Observation callers use force_deferred_io for legacy descriptors as well as
+    V2 descriptors. The latch is released only after factory validation succeeds.
+    """
+
+    if not isinstance(force_deferred_io, bool):
+        raise ConfigError("force_deferred_io must be bool")
     normalized_access = normalize_access_mode(access, "access")
     if lease is not None and lease.fingerprint != resource_fingerprint(resource, lease.lock_id):
         raise ConfigError("resource lease does not match configured instrument resource")
@@ -82,7 +91,9 @@ def open_instrument_driver(
     strict_v2_capability_opt_in = bool(
         set(descriptor.capabilities) & SCOPE_STRICT_V2_CAPABILITIES
     )
-    construction_latched = bounded_binary_profile_opt_in or strict_v2_capability_opt_in
+    construction_latched = (
+        force_deferred_io or bounded_binary_profile_opt_in or strict_v2_capability_opt_in
+    )
     backend = _select_backend(configured_backend, descriptor.backends)
     _validate_resource_scheme(resource, descriptor.resource_schemes)
     try:
diff --git a/src/wavebench/services/agent_advise.py b/src/wavebench/services/agent_advise.py
index 176c2467..84d65fc1 100644
--- a/src/wavebench/services/agent_advise.py
+++ b/src/wavebench/services/agent_advise.py
@@ -128,6 +128,7 @@ def _recommendations(
         channel = channel_section.get("channel")
         if not isinstance(channel, int):
             continue
+        recommendation_count = len(recommendations)
         summary = _waveform_summary(channel_section)
         snapshot = _scope_status_data(channel_section)
         frequency_hz, source, confidence, withheld_reason = _frequency_for_advice(
@@ -152,7 +153,8 @@ def _recommendations(
             "time_range_s": time_range,
             "vertical_scale_v_per_div": vertical_scale,
         }
-        if snapshot and snapshot.get("channel", {}).get("enabled") is False:
+        snapshot_channel = None if snapshot is None else snapshot.get("channel")
+        if isinstance(snapshot_channel, dict) and snapshot_channel.get("enabled") is False:
             recommendations.append(
                 _command_recommendation(
                     "display_on",
@@ -201,6 +203,8 @@ def _recommendations(
                     },
                 )
             )
+        if len(recommendations) == recommendation_count:
+            recommendations.append(_unavailable_advice(channel))
     span = _frequency_span(channel_profiles)
     if span is not None and span["ratio_high_over_low"] > 10.0:
         recommendations.append(
@@ -223,19 +227,27 @@ def _recommendations(
             }
         )
     if not recommendations:
-        recommendations.append(
-            {
-                "id": "no_adjustment_needed",
-                "priority": "low",
-                "action": "keep_current_scope_settings",
-                "reason": "No obvious display or acquisition-window issue was found.",
-                "mutates_instrument_if_applied": False,
-                "raw_scpi": False,
-            }
-        )
+        recommendations.append(_unavailable_advice(None))
     return recommendations
 
 
+def _unavailable_advice(channel: int | None) -> dict[str, Any]:
+    recommendation: dict[str, Any] = {
+        "id": "advice_unavailable",
+        "priority": "normal",
+        "action": "obtain_scope_evidence",
+        "reason": (
+            "No usable display settings, waveform metrics or expected frequency are available; "
+            "the current settings could not be assessed."
+        ),
+        "mutates_instrument_if_applied": False,
+        "raw_scpi": False,
+    }
+    if channel is not None:
+        recommendation["channel"] = channel
+    return recommendation
+
+
 def _waveform_summary(channel_section: dict[str, Any]) -> dict[str, Any] | None:
     waveform = channel_section.get("waveform", {})
     if waveform.get("status") != "ok":
@@ -246,6 +258,8 @@ def _waveform_summary(channel_section: dict[str, Any]) -> dict[str, Any] | None:
 
 def _scope_status_data(channel_section: dict[str, Any]) -> dict[str, Any] | None:
     status = channel_section.get("scope_status", {})
+    if status.get("status") not in {"ok", "partial"}:
+        return None
     data = status.get("data")
     return data if isinstance(data, dict) else None
 
@@ -279,7 +293,10 @@ def _summary_frequency(summary: dict[str, Any] | None) -> float | None:
     if summary is None:
         return None
     value = summary.get("frequency_estimate_hz")
-    if not isinstance(value, (int, float)) or isinstance(value, bool) or value <= 0:
+    if (
+        not isinstance(value, (int, float)) or isinstance(value, bool)
+        or not math.isfinite(value) or value <= 0
+    ):
         return None
     return float(value)
 
@@ -304,12 +321,20 @@ def _recommended_vertical_scale(
     target_vertical_divisions: float,
 ) -> float | None:
     vpp = None if summary is None else summary.get("voltage_vpp_v")
-    if isinstance(vpp, (int, float)) and not isinstance(vpp, bool) and vpp > 0:
+    if (
+        isinstance(vpp, (int, float)) and not isinstance(vpp, bool)
+        and math.isfinite(vpp) and vpp > 0
+    ):
         return float(vpp) / target_vertical_divisions
     scale = None
     if snapshot is not None:
-        scale = snapshot.get("channel", {}).get("scale_v_per_div")
-    if isinstance(scale, (int, float)) and not isinstance(scale, bool) and scale > 0:
+        snapshot_channel = snapshot.get("channel")
+        if isinstance(snapshot_channel, dict):
+            scale = snapshot_channel.get("scale_v_per_div")
+    if (
+        isinstance(scale, (int, float)) and not isinstance(scale, bool)
+        and math.isfinite(scale) and scale > 0
+    ):
         return float(scale)
     return None
 
@@ -420,6 +445,8 @@ def _agent_hints(
     recommendations: list[dict[str, Any]],
 ) -> list[str]:
     hints = list(observation.get("agent_hints", []))
+    if any(item["id"] == "advice_unavailable" for item in recommendations):
+        hints.append("advise: insufficient evidence is not a recommendation to keep current settings")
     if any(item["id"] == "separate_timebase_profiles" for item in recommendations):
         hints.append("advise: run focus/observe per channel when frequencies differ greatly")
     if any(item["id"] == "timebase_advice_withheld" for item in recommendations):
diff --git a/src/wavebench/services/agent_observe.py b/src/wavebench/services/agent_observe.py
index a5a1b6d4..c0a3b796 100644
--- a/src/wavebench/services/agent_observe.py
+++ b/src/wavebench/services/agent_observe.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from dataclasses import asdict
+from dataclasses import dataclass, replace
 from pathlib import Path
 from typing import Any
 
@@ -11,10 +11,13 @@
     validate_expectation,
 )
 from wavebench.data.relationships import analyze_waveform_relationships
-from wavebench.errors import ConfigError, WaveBenchError
+from wavebench.errors import (
+    ConfigError, ConnectionError, DataError, SessionHealthError, TransportIOError, WaveBenchError,
+)
 from wavebench.instruments.models import WaveformData
 from wavebench.logging import CommandLogger
 from wavebench.services.scope_service import ScopeService
+from wavebench.transport.session import SessionHealth
 
 # 读取波形可能造成的仪器状态影响。读取前不恢复原采集状态，调用方必须先确认。
 _WAVEFORM_STATE_EFFECTS = [
@@ -22,9 +25,31 @@
     "waveform transfer source/mode/format/points may be changed",
     "some drivers may enable the requested analog channel display before fetching",
     "the previous acquisition run state is not restored",
+    "the instrument error queue may be consumed when check_errors is enabled",
 ]
 
 
+@dataclass
+class _ObservationExecution:
+    service: ScopeService
+    stop_reason: str | None = None
+    mutation_block_reason: str | None = None
+
+    def io_failed(self, exc: Exception) -> None:
+        state = self.service.session_state
+        if (
+            isinstance(exc, (SessionHealthError, TransportIOError, ConnectionError, OSError))
+            or (state is not None and state.health is not SessionHealth.HEALTHY)
+            or not isinstance(exc, (ConfigError, DataError))
+        ):
+            self.stop_reason = f"instrument I/O stopped after {type(exc).__name__}: {exc}"
+
+    def check_health(self) -> None:
+        state = self.service.session_state
+        if state is not None and state.health is not SessionHealth.HEALTHY:
+            self.stop_reason = f"instrument I/O stopped because session health is {state.health.value}"
+
+
 def scope_observe_payload(
     *,
     config_path: str | Path,
@@ -33,9 +58,10 @@ def scope_observe_payload(
     allow_50ohm: bool = False,
     resource: str | None = None,
 ) -> dict[str, Any]:
-    """严格只读的示波器观察：IDN、每通道快照和高阻安全判断。
+    """通过受限查询执行身份、可用 V2 快照和高阻安全判断。
 
-    该函数不读取波形，也不发送任何会改变仪器状态的命令，因此可以安全地通过 MCP 暴露。
+    该函数不读取波形，观察阶段限制为文本查询；V2 快照以 descriptor 的 pure-read 合同为准。
+    插件属于可信本地代码，guard 的 I/O 限制不替代插件查询语义的合同验证。
     需要波形、期望值检查或多通道关系时，请使用 ``scope_waveform_report_payload``
     （对应显式 CLI 命令 ``wavebench scope observe --fetch-waveform``）。
     """
@@ -62,7 +88,7 @@ def scope_waveform_report_payload(
     """显式读取波形并给出摘要、期望值检查和多通道关系。
 
     读取波形属于写操作：可能停止正在运行的采集、修改波形传输参数并打开通道显示。
-    所有输入（通道、期望值）都在任何仪器 I/O 之前完成校验，非法输入不会产生任何仪器写入。
+    所有通道和采集参数都在采集写入之前完成预检；型号相关通道验证可能需要只读查询。
     """
     normalized_expectations = _normalize_expectations(expectations)
     return _build_observation(
@@ -89,6 +115,8 @@ def _build_observation(
     config = load_config(config_path)
     if resource:
         config = config.with_resource(resource)
+    if not fetch_waveform and config.scope.access == "read_write":
+        config = replace(config, scope=replace(config.scope, access="read_only"))
     observed_channels = _scope_channels(
         channel=channel,
         channels=channels,
@@ -107,21 +135,53 @@ def _build_observation(
     warnings: list[str] = []
     fetched_waveforms: dict[int, WaveformData] = {}
     expectation_results: dict[int, dict[str, Any]] = {}
+    execution = _ObservationExecution(service)
+    channel_sections: list[dict[str, Any]] = []
+    access_validation = _attempt(
+        service.validate_observation_access, warnings=warnings, name="observation_access",
+    )
+    if access_validation["status"] != "ok":
+        execution.stop_reason = "observation access/capability validation failed before session open"
+    if fetch_waveform:
+        validation = _attempt(
+            service.validate_observation_fetch, warnings=warnings, name="waveform_validation",
+        )
+        if validation["status"] != "ok":
+            execution.stop_reason = "waveform configuration/access/capability validation failed before session open"
 
-    sections["identity"] = _attempt(lambda: {"idn": service.idn()}, warnings=warnings, name="identity")
-    channel_sections = [
-        _observe_channel(
-            service,
-            observed_channel,
-            fetch_waveform=fetch_waveform,
-            allow_50ohm=allow_50ohm,
-            warnings=warnings,
-            fetched_waveforms=fetched_waveforms,
-            expectations=expectations or {},
-            expectation_results=expectation_results,
+    def collect() -> None:
+        with service.session_context(observation=True):
+            sections["identity"] = _attempt(
+                lambda: {"idn": service.observation_identity()},
+                warnings=warnings, name="identity", execution=execution, io=True,
+            )
+            if fetch_waveform and execution.mutation_block_reason is None:
+                preflight = _attempt(
+                    lambda: service.preflight_observation_fetch(
+                        observed_channels, allow_50ohm=allow_50ohm,
+                    ),
+                    warnings=warnings, name="waveform_preflight", execution=execution, io=True,
+                )
+                if preflight["status"] != "ok":
+                    execution.mutation_block_reason = "all-channel waveform preflight failed"
+            for observed_channel in observed_channels:
+                channel_sections.append(observe_channel(observed_channel))
+
+    def observe_channel(observed_channel: int) -> dict[str, Any]:
+        return _observe_channel(
+            service, observed_channel, fetch_waveform=fetch_waveform,
+            allow_50ohm=allow_50ohm, warnings=warnings,
+            fetched_waveforms=fetched_waveforms, expectations=expectations or {},
+            expectation_results=expectation_results, execution=execution,
         )
-        for observed_channel in observed_channels
-    ]
+
+    lifecycle = _attempt(collect, warnings=warnings, name="session", execution=execution, io=True)
+    if lifecycle["status"] != "ok":
+        execution.stop_reason = "observation session unavailable or close failed"
+        sections["session"] = lifecycle
+        sections.setdefault("identity", lifecycle)
+        for observed_channel in observed_channels[len(channel_sections):]:
+            channel_sections.append(observe_channel(observed_channel))
     first_channel = channel_sections[0]
     sections["scope_status"] = first_channel["scope_status"]
     sections["coupling"] = first_channel["coupling"]
@@ -153,16 +213,19 @@ def _build_observation(
         "warnings": warnings,
     }
     if fetch_waveform:
-        # 每个通道各自打开 session，波形不保证来自同一次 acquisition；跨采集的时序关系不成立。
+        # 共享 session/lease 不能证明各通道来自同一次 acquisition。
         payload["waveform_source"] = {
             "same_acquisition": False,
             "reason": "channels are fetched channel-by-channel, not in one acquisition",
         }
-        payload["relationships"] = (
-            analyze_waveform_relationships(fetched_waveforms, same_acquisition=False)
-            if len(fetched_waveforms) >= 2
-            else []
-        )
+        payload["relationships"] = []
+        if len(fetched_waveforms) >= 2:
+            relationships = _attempt(
+                lambda: analyze_waveform_relationships(fetched_waveforms, same_acquisition=False),
+                warnings=warnings, name="relationships",
+            )
+            if relationships["status"] == "ok":
+                payload["relationships"] = relationships["data"]
         payload["expectations"] = expectation_summary(expectation_results)
     payload["agent_hints"] = _agent_hints(
         sections,
@@ -172,6 +235,7 @@ def _build_observation(
         expectation_results=expectation_results,
         fetch_waveform=fetch_waveform,
     )
+    payload["status"] = "ok" if not warnings else "partial"
     return payload
 
 
@@ -217,18 +281,21 @@ def _observe_channel(
     fetched_waveforms: dict[int, WaveformData],
     expectations: dict[int, dict[str, Any]],
     expectation_results: dict[int, dict[str, Any]],
+    execution: _ObservationExecution,
 ) -> dict[str, Any]:
     section: dict[str, Any] = {
         "channel": channel,
         "scope_status": _attempt(
-            lambda: asdict(service.status(channel=channel)),
+            lambda: service.observation_status(channel=channel),
             warnings=warnings,
             name=f"ch{channel}_scope_status",
+            execution=execution, io=True,
         ),
         "coupling": _attempt(
             lambda: _coupling_payload(service, channel, allow_50ohm=allow_50ohm),
             warnings=warnings,
             name=f"ch{channel}_coupling",
+            execution=execution, io=True,
         ),
     }
     if not fetch_waveform:
@@ -239,9 +306,11 @@ def _observe_channel(
             channel,
             allow_50ohm=allow_50ohm,
             fetched_waveforms=fetched_waveforms,
+            execution=execution,
         ),
         warnings=warnings,
         name=f"ch{channel}_waveform",
+        execution=execution, mutation=True,
     )
     if channel in expectations and section["waveform"]["status"] == "ok":
         result = _attempt(
@@ -268,16 +337,32 @@ def _observe_channel(
     return section
 
 
-def _attempt(call, *, warnings: list[str], name: str) -> dict[str, Any]:
+def _attempt(
+    call, *, warnings: list[str], name: str,
+    execution: _ObservationExecution | None = None, io: bool = False, mutation: bool = False,
+) -> dict[str, Any]:
+    if execution is not None:
+        execution.check_health()
+        reason = execution.stop_reason or (execution.mutation_block_reason if mutation else None)
+        if reason and (io or mutation):
+            warnings.append(f"{name}_skipped: {reason}")
+            return {"status": "skipped", "reason": reason}
     try:
-        return {"status": "ok", "data": call()}
+        data = call()
+        if execution is not None:
+            execution.check_health()
+        return {"status": "ok", "data": data}
     except WaveBenchError as exc:
+        if execution is not None and io:
+            execution.io_failed(exc)
         warnings.append(f"{name}_unavailable: {exc}")
         return {
             "status": "unavailable",
             "error": {"type": type(exc).__name__, "message": str(exc)},
         }
     except Exception as exc:
+        if execution is not None and io:
+            execution.io_failed(exc)
         warnings.append(f"{name}_unavailable: {type(exc).__name__}: {exc}")
         return {
             "status": "unavailable",
@@ -291,12 +376,7 @@ def _coupling_payload(
     *,
     allow_50ohm: bool,
 ) -> dict[str, Any]:
-    coupling = service.require_high_impedance(channel, allow_50ohm=allow_50ohm)
-    return {
-        "channel": channel,
-        "coupling": coupling,
-        "accepted_for_capture": True,
-    }
+    return service.observation_input_safety(channel, allow_50ohm=allow_50ohm)
 
 
 def _waveform_payload(
@@ -305,16 +385,27 @@ def _waveform_payload(
     *,
     allow_50ohm: bool,
     fetched_waveforms: dict[int, WaveformData],
+    execution: _ObservationExecution,
 ) -> dict[str, Any]:
-    service.require_high_impedance(channel, allow_50ohm=allow_50ohm)
-    waveform = service.fetch_waveform(channel=channel)
+    try:
+        service.observation_input_safety(channel, allow_50ohm=allow_50ohm)
+        execution.check_health()
+        if execution.stop_reason:
+            raise ConfigError(execution.stop_reason)
+        waveform = service.fetch_waveform(channel=channel)
+    except Exception as exc:
+        execution.io_failed(exc)
+        if isinstance(exc, ConfigError):
+            execution.mutation_block_reason = f"waveform mutation stopped after ConfigError: {exc}"
+        raise
+    summary = waveform.summary(
+        expected_frequency_hz=service.config.waveform.expected_frequency_hz,
+        frequency_tolerance_ratio=service.config.waveform.frequency_tolerance_ratio,
+    )
     fetched_waveforms[channel] = waveform
     return {
         "channel": channel,
-        "summary": waveform.summary(
-            expected_frequency_hz=service.config.waveform.expected_frequency_hz,
-            frequency_tolerance_ratio=service.config.waveform.frequency_tolerance_ratio,
-        ),
+        "summary": summary,
         "raw_samples_included": False,
     }
 
@@ -364,7 +455,7 @@ def _agent_hints(
         elif result["status"] == "unavailable":
             hints.append(f"CH{channel}_expectation_unavailable: acceptance could not be evaluated")
     if sections.get("scope_status", {}).get("status") == "unavailable":
-        hints.append("driver lacks scope.snapshot or the status query failed; use identity cautiously")
+        hints.append("pure-query V2 snapshot is unavailable; identity alone cannot assess scope settings")
     if sections.get("coupling", {}).get("status") == "unavailable":
         hints.append("do not run capture until input coupling safety is confirmed")
     if warnings:
diff --git a/src/wavebench/services/scope_service.py b/src/wavebench/services/scope_service.py
index a0d655e5..3c755041 100644
--- a/src/wavebench/services/scope_service.py
+++ b/src/wavebench/services/scope_service.py
@@ -6,7 +6,7 @@
 import time
 import traceback
 from collections.abc import Iterator, Mapping
-from contextlib import contextmanager
+from contextlib import contextmanager, nullcontext
 from dataclasses import asdict, dataclass
 from pathlib import Path
 from typing import Any, TypeVar, cast
@@ -14,9 +14,11 @@
 
 import numpy as np
 
-from wavebench.config import WaveBenchConfig
+from wavebench.config import WaveBenchConfig, normalize_waveform_points
 from wavebench.data.package import new_package_dir
-from wavebench.errors import ConfigError, DataError, SessionHealthError, WaveBenchError
+from wavebench.errors import (
+    ConfigError, DataError, SessionCloseError, SessionHealthError, WaveBenchError,
+)
 from wavebench.instruments.api import InstrumentDescriptor, ScopeCouplingPolicy
 from wavebench.instruments.capabilities import require_capabilities
 from wavebench.instruments.contracts import (
@@ -61,6 +63,7 @@
     WaveformData,
 )
 from wavebench.instruments.registry import resolve_instrument_descriptor
+from wavebench.instruments.scope_extension_capabilities import validate_scope_descriptor
 from wavebench.instruments.scope_extensions import (
     ErrorCheckSpec,
     ScopeContinuousAcquisitionRequest,
@@ -339,7 +342,7 @@ def _verify_scope_identity(self, scope: ScopeDriver, *, spec: OperationSpec) ->
                 coordinator.complete_verification(authorization)
                 return identity
 
-    def _open_scope(self) -> ScopeDriver:
+    def _open_scope(self, *, defer_driver_io: bool = False) -> ScopeDriver:
         self._prepare_session_open("scope")
         if self.lease is None:
             self.lease = ResourceLease(
@@ -360,6 +363,7 @@ def _open_scope(self) -> ScopeDriver:
             options=getattr(self.config.scope, "options", {}),
             access=getattr(self.config.scope, "access", "read_write"),
             lease=self.lease,
+            force_deferred_io=defer_driver_io,
         )
         self.descriptor = opened.descriptor
         self.transport = opened.transport
@@ -374,15 +378,168 @@ def open_session(self) -> ScopeDriver:
         return self._open_scope()
 
     @contextmanager
-    def _scope_session(self) -> Iterator[ScopeDriver]:
+    def session_context(self, *, observation: bool = False) -> Iterator[ScopeService]:
+        """Retain one epoch and lease; borrowed sessions remain owned by their caller.
+
+        Observation-owned sessions defer constructor I/O until factory validation.
+        """
+
         if self.session is not None:
-            yield self.session
+            if isinstance(self.transport, GuardedAuditedTransport) and self.transport.session_state is not self.session_state:
+                raise ConfigError("scope service and guarded transport require shared instrument session state")
+            lock = self.session_state.transaction_lock if self.session_state else nullcontext()
+            with lock:
+                yield self
             return
-        scope = self._open_scope()
+        owned_lease = self.lease is None or not self.lease.acquired
+        scope = self._open_scope(defer_driver_io=True) if observation else self._open_scope()
+        self.session = scope
+        body_error: BaseException | None = None
         try:
-            yield scope
+            lock = self.session_state.transaction_lock if self.session_state else nullcontext()
+            with lock:
+                yield self
+        except BaseException as exc:
+            body_error = exc
+            raise
         finally:
-            scope.close()
+            failures: list[tuple[str, BaseException]] = []
+            # A plugin close may fail before delegating to its core transport.
+            # The guard's close is idempotent and invalidates the epoch first.
+            for component, close in (
+                ("driver", scope.close),
+                ("transport", getattr(self.transport, "close", None)),
+                ("state", getattr(self.session_state, "close", None)),
+                ("lease", self.lease.release if owned_lease and self.lease else None),
+            ):
+                if callable(close):
+                    try:
+                        close()
+                    except BaseException as exc:
+                        failures.append((component, exc))
+            self.session = None
+            if failures:
+                error = SessionCloseError(failures)
+                if body_error is not None:
+                    body_error.add_note(str(error))
+                else:
+                    raise error from failures[0][1]
+
+    @contextmanager
+    def _scope_session(self) -> Iterator[ScopeDriver]:
+        with self.session_context():
+            assert self.session is not None
+            yield self.session
+
+    def _observation_state(self) -> InstrumentSessionState:
+        state = self.session_state
+        if (
+            not isinstance(self.transport, GuardedAuditedTransport)
+            or state is None
+            or self.transport.session_state is not state
+        ):
+            raise ConfigError("scope observation requires a core guarded session")
+        if self.transport.lease is None or not self.transport.lease.acquired:
+            raise ConfigError("scope observation requires an exclusive resource lease")
+        if state.health is not SessionHealth.HEALTHY:
+            raise SessionHealthError(
+                "scope observation requires a healthy session",
+                health=state.health.value,
+                io_kind="observation_preflight",
+                epoch_id=state.epoch_id,
+            )
+        return state
+
+    @contextmanager
+    def observation_queries(self, operation: str, *, max_queries: int = 128) -> Iterator[None]:
+        """Gate public transport I/O to text queries on the retained healthy epoch.
+
+        This is a transport authorization boundary, not a Python plugin sandbox.
+        Legacy snapshots have no pure-read field contract and are not used here.
+        """
+
+        state = self._observation_state()
+        timeout_ms = self.config.connection.timeout_ms
+        with SessionTransactionCoordinator(state).authorize_normal(
+            operation_id=operation,
+            allowed_io=("query",),
+            fields=("scope.observation",),
+            timeout_ms=timeout_ms,
+            max_steps=max_queries,
+            context_id="scope_observation",
+            correlation_id=uuid4().hex,
+            phase="query",
+            absolute_deadline=time.monotonic() + timeout_ms / 1000.0,
+        ):
+            yield
+
+    def observation_identity(self) -> str:
+        spec = self._require("scope.idn", "scope.idn")
+        with self._scope_session() as scope:
+            state = self._observation_state()
+            if "scope.identity" not in state.verified_fields:
+                return self._verify_scope_identity(scope, spec=spec)
+            with self.observation_queries("scope.observation.identity", max_queries=1):
+                return scope.idn()
+
+    def observation_status(self, channel: int) -> dict[str, Any]:
+        """Read a descriptor's pure-query V2 snapshot without consuming error queues."""
+
+        with self._scope_session():
+            self._observation_state()
+            return asdict(self.snapshot_v2(channel))
+
+    def observation_input_safety(self, channel: int, *, allow_50ohm: bool = False) -> dict[str, Any]:
+        descriptor = self.descriptor or resolve_instrument_descriptor(
+            self.config.scope.driver, expected_kind="scope",
+        )
+        with self._scope_session():
+            with self.observation_queries("scope.observation.input_safety"):
+                if "scope.channel_input_state_v2" in descriptor.capabilities:
+                    state = assert_scope_input_state_safe(
+                        self.channel_input_state_v2(channel), allow_50ohm=allow_50ohm,
+                    )
+                    return {
+                        "channel": channel, "coupling": state.coupling,
+                        "termination": state.termination, "accepted_for_capture": True,
+                    }
+                coupling = self.require_high_impedance(channel, allow_50ohm=allow_50ohm)
+                return {"channel": channel, "coupling": coupling, "accepted_for_capture": True}
+
+    def validate_observation_access(self) -> None:
+        """Reject disabled access and missing identity support before any factory runs."""
+
+        self._require("scope.idn", "scope.idn")
+
+    def validate_observation_fetch(self) -> None:
+        """Reject all offline-invalid fetch settings before report mutation starts."""
+
+        self._waveform_fetch_spec()
+        self.validate_observation_access()
+        descriptor = self.descriptor or resolve_instrument_descriptor(
+            self.config.scope.driver, expected_kind="scope",
+        )
+        validate_scope_descriptor(descriptor)
+        if not {"scope.channel_input_state_v2", "scope.channel_coupling"} & set(descriptor.capabilities):
+            raise ConfigError("scope waveform observation has no safe query-only channel verifier")
+        profile = self._waveform_binary_profile()
+        if profile is not None:
+            try:
+                profile.operation_for("fetch")
+            except ValueError as exc:
+                raise ConfigError(str(exc)) from exc
+
+    def preflight_observation_fetch(self, channels: tuple[int, ...], *, allow_50ohm: bool = False) -> None:
+        self.validate_observation_fetch()
+        # Standard waveform profiles do not declare an operation-specific channel
+        # range. Validate every requested channel through the guarded input query;
+        # unrelated display/focus profiles cannot prove fetch support.
+        with self._scope_session() as scope:
+            for channel in channels:
+                self.observation_input_safety(channel, allow_50ohm=allow_50ohm)
+            if self._waveform_binary_profile() is not None:
+                self._bounded_waveform_executor(scope)
+            self._session_preflight("scope.fetch_waveform", scope)
 
     def idn(self) -> str:
         spec = self._require("scope.idn", "scope.idn")
@@ -1192,11 +1349,12 @@ def autoscale(self) -> None:
                 check_errors=self.config.autoscale.check_errors,
             )
 
-    def fetch_waveform(self, channel: int) -> WaveformData:
+    def _waveform_fetch_spec(self) -> OperationSpec:
         if self.config.waveform.format.lower() != "real":
             raise ConfigError("MVP-1 only supports waveform.format = 'real'")
         if self.config.waveform.byte_order.lower() != "lsbf":
             raise ConfigError("MVP-1 only supports waveform.byte_order = 'lsbf'")
+        normalize_waveform_points(self.config.waveform.points)
         bounded_profile = self._waveform_binary_profile()
         required = ["scope.fetch_waveform"]
         if bounded_profile is not None:
@@ -1205,7 +1363,11 @@ def fetch_waveform(self, channel: int) -> WaveformData:
                 required.append("scope.error_drain_v1")
         elif self.config.scope.check_errors:
             required.append("scope.errors")
-        self._require("scope.fetch_waveform", *required)
+        return self._require("scope.fetch_waveform", *required)
+
+    def fetch_waveform(self, channel: int) -> WaveformData:
+        self._waveform_fetch_spec()
+        bounded_profile = self._waveform_binary_profile()
         with self._scope_session() as scope:
             if bounded_profile is not None:
                 result = self._bounded_waveform_executor(scope).fetch(
diff --git a/tests/test_agent_advise.py b/tests/test_agent_advise.py
index 5d5df8c8..9ec150e2 100644
--- a/tests/test_agent_advise.py
+++ b/tests/test_agent_advise.py
@@ -1,4 +1,6 @@
 import shlex
+from contextlib import nullcontext
+from dataclasses import asdict
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from unittest.mock import patch
@@ -34,6 +36,13 @@ def _write_config(root: Path) -> Path:
 class _FakeScopeService:
     def __init__(self, *, config, logger):
         self.config = config
+        self.session_state = None
+
+    def session_context(self, **kwargs):
+        return nullcontext(self)
+
+    def validate_observation_access(self):
+        pass
 
     def idn(self):
         return "RIGOL TECHNOLOGIES,DS1104Z Plus,123,1.0"
@@ -44,6 +53,17 @@ def status(self, channel):
     def require_high_impedance(self, channel, *, allow_50ohm=False):
         return "DC"
 
+    def observation_identity(self):
+        return self.idn()
+
+    def observation_status(self, channel):
+        return asdict(self.status(channel))
+
+    def observation_input_safety(self, channel, *, allow_50ohm=False):
+        return {"channel": channel,
+                "coupling": self.require_high_impedance(channel, allow_50ohm=allow_50ohm),
+                "accepted_for_capture": True}
+
 
 def _snapshot(channel: int):
     from wavebench.instruments.models import (
@@ -248,3 +268,77 @@ def test_focus_command_hide_others_uses_real_cli_flag(hide_others):
     assert request.channels == (2,)
     assert request.vertical_scales[0].scale_v_per_div == 0.25
     assert request.hide_others is hide_others
+
+
+def test_advice_without_evidence_does_not_recommend_keeping_settings():
+    observation = _observation(fetch_waveform=False)
+    observation["status"] = "partial"
+    for section in observation["channels"]:
+        section["scope_status"] = {"status": "unavailable", "reason": "snapshot unsupported"}
+    payload = scope_advise_from_observation(observation)
+    assert [item["id"] for item in payload["recommendations"]] == [
+        "advice_unavailable", "advice_unavailable",
+    ]
+    assert [item["channel"] for item in payload["recommendations"]] == [1, 2]
+    assert all("command" not in item for item in payload["recommendations"])
+    assert payload["status"] == "partial"
+
+
+def test_unavailable_channel_does_not_hide_other_channel_advice():
+    observation = _observation(fetch_waveform=False)
+    observation["status"] = "partial"
+    observation["channels"][1]["scope_status"] = {"status": "unavailable"}
+    payload = scope_advise_from_observation(observation)
+    assert [(item["id"], item["channel"]) for item in payload["recommendations"]] == [
+        ("focus_channel", 1), ("advice_unavailable", 2),
+    ]
+
+
+def test_unavailable_snapshot_data_is_not_reused_as_evidence():
+    observation = _observation(fetch_waveform=False)
+    observation["channels"] = [observation["channels"][0]]
+    observation["channels"][0]["scope_status"]["status"] = "unavailable"
+    payload = scope_advise_from_observation(observation)
+    assert payload["recommendations"][0]["id"] == "advice_unavailable"
+
+
+def test_identity_only_v2_snapshot_returns_unavailable_advice():
+    from wavebench.instruments.models import (
+        SCOPE_SNAPSHOT_V2_FIELD_ORDER, ScopeIdentitySnapshot, ScopeSnapshotV2,
+    )
+
+    snapshot = ScopeSnapshotV2(
+        identity=ScopeIdentitySnapshot("FAKE", "Scope", "OFFLINE", "1.0"),
+        unavailable_fields=tuple(
+            field for field in SCOPE_SNAPSHOT_V2_FIELD_ORDER if not field.startswith("identity.")
+        ),
+    )
+    observation = _observation(fetch_waveform=False)
+    observation["channels"] = [observation["channels"][0]]
+    observation["channels"][0]["scope_status"]["data"] = asdict(snapshot)
+    payload = scope_advise_from_observation(observation)
+    assert payload["recommendations"][0]["id"] == "advice_unavailable"
+
+
+def test_configured_frequency_can_support_advice_without_snapshot():
+    observation = _observation(fetch_waveform=False)
+    observation["channels"] = [observation["channels"][0]]
+    observation["channels"][0]["scope_status"] = {"status": "unavailable"}
+    payload = scope_advise_from_observation(observation, expected_frequencies_hz={1: 1000.0})
+    recommendation = payload["recommendations"][0]
+    assert recommendation["id"] == "focus_channel"
+    assert recommendation["parameters"]["time_range_s"] == pytest.approx(0.01)
+    assert recommendation["parameters"]["frequency_confidence"] == "configured"
+
+
+@pytest.mark.parametrize("invalid", [float("nan"), float("inf"), -float("inf")])
+def test_nonfinite_metrics_are_not_usable_advice_evidence(invalid):
+    observation = _observation(fetch_waveform=True, measured_frequency={1: 1000.0})
+    section = observation["channels"][0]
+    observation["channels"] = [section]
+    section["scope_status"]["data"]["channel"]["scale_v_per_div"] = invalid
+    section["waveform"]["data"]["summary"].update(
+        frequency_estimate_hz=invalid, voltage_vpp_v=invalid, quality_warnings=[],
+    )
+    payload = scope_advise_from_observation(observation)
+    assert payload["recommendations"][0]["id"] == "advice_unavailable"
diff --git a/tests/test_agent_observe.py b/tests/test_agent_observe.py
index 38c351c9..ea224eb7 100644
--- a/tests/test_agent_observe.py
+++ b/tests/test_agent_observe.py
@@ -1,3 +1,5 @@
+from contextlib import contextmanager
+from dataclasses import asdict
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from unittest.mock import patch
@@ -89,6 +91,7 @@ class _FakeScopeService:
 
     def __init__(self, *, config, logger):
         self.config = config
+        self.session_state = None
         self.fetched_channels: list[int] = []
         self.allow_50ohm_seen: list[bool] = []
         _FakeScopeService.instances.append(self)
@@ -99,6 +102,33 @@ def idn(self):
     def status(self, channel):
         return _snapshot(channel)
 
+    @contextmanager
+    def session_context(self, *, observation=False):
+        yield self
+
+    def observation_identity(self):
+        return self.idn()
+
+    def observation_status(self, channel):
+        return asdict(self.status(channel))
+
+    def observation_input_safety(self, channel, *, allow_50ohm=False):
+        return {
+            "channel": channel,
+            "coupling": self.require_high_impedance(channel, allow_50ohm=allow_50ohm),
+            "accepted_for_capture": True,
+        }
+
+    def validate_observation_fetch(self):
+        pass
+
+    def validate_observation_access(self):
+        pass
+
+    def preflight_observation_fetch(self, channels, *, allow_50ohm=False):
+        for channel in channels:
+            self.observation_input_safety(channel, allow_50ohm=allow_50ohm)
+
     def require_high_impedance(self, channel, *, allow_50ohm=False):
         self.allow_50ohm_seen.append(allow_50ohm)
         return "DC"
@@ -245,9 +275,13 @@ def evaluate(waveform, expectation):
         str(channel): "unavailable" if channel in failed_channels else "pass"
         for channel in (1, 2)
     }
+    # Any unsafe channel blocks the whole report before acquisition writes.
+    # An in-fetch ConfigError conservatively stops later mutations too.
+    if failure == "coupling" or (failure == "waveform" and 1 in failed_channels):
+        expected_statuses = {"1": "unavailable", "2": "unavailable"}
     assert payload["status"] == "partial"
     assert payload["expectations"] == {
-        "status": "unavailable" if len(failed_channels) == 2 else "partial",
+        "status": "unavailable" if set(expected_statuses.values()) == {"unavailable"} else "partial",
         "channels": expected_statuses,
     }
     for channel in failed_channels:
diff --git a/tests/test_scope_observation_contracts.py b/tests/test_scope_observation_contracts.py
new file mode 100644
index 00000000..28aa5125
--- /dev/null
+++ b/tests/test_scope_observation_contracts.py
@@ -0,0 +1,616 @@
+"""Offline observation contracts using real factory, guard, epochs and leases."""
+from __future__ import annotations
+
+from dataclasses import replace
+from pathlib import Path
+import subprocess
+import sys
+
+import numpy as np
+import pytest
+
+from wavebench.config import load_config
+from wavebench.errors import (
+    ConfigError, ConnectionError, DataError, ResourceBusyError,
+    SessionCloseError, SessionHealthError, TransportIOError,
+)
+from wavebench.instruments.api import InstrumentDescriptor
+from wavebench.instruments.builtin import BUILTIN_INSTRUMENTS
+from wavebench.instruments.models import (
+    SCOPE_SNAPSHOT_V2_FIELD_ORDER, ScopeAnalogChannelSnapshotV2,
+    ScopeChannelInputStateV2, ScopeIdentitySnapshot, ScopeSnapshotV2,
+    WaveformData, WaveformHeader,
+)
+from wavebench.instruments.registry import InstrumentRegistry
+from wavebench.instruments.scope_extensions import (
+    ScopeDescriptorExtensions, ScopeSnapshotProfileV2,
+    ScopeWaveformBinaryOperationProfile, ScopeWaveformBinaryProfile,
+)
+from wavebench.logging import CommandLogger
+from wavebench.services.agent_observe import scope_observe_payload, scope_waveform_report_payload
+from wavebench.services.resource_lease import ResourceLease
+from wavebench.services.scope_service import ScopeService
+from wavebench.transport.contracts import (
+    CommandTransmission, ReplayPolicy, ResponseProgress, Synchronization, TransportPhase,
+)
+from wavebench.transport.guarded import GuardedAuditedTransport
+from wavebench.transport.session import SessionHealth
+
+
+RESOURCE = "TCPIP::offline-scope-contract.invalid::INSTR"
+FIELDS = (
+    "identity.manufacturer", "identity.model", "identity.serial_number",
+    "identity.firmware", "identity.options", "channel.channel", "channel.coupling",
+)
+
+
+class AnalysisFailureWaveform(WaveformData):
+    def summary(self, *args, **kwargs):
+        raise DataError("deterministic waveform analysis failed")
+
+
+class ConcreteTransport:
+    resource = RESOURCE
+
+    def __init__(self, harness):
+        self.harness = harness
+        self.events = []
+
+    def query(self, command, *, replay=ReplayPolicy.NO_REPLAY):
+        assert replay is ReplayPolicy.NO_REPLAY
+        self.events.append(("query", command))
+        if self.harness.query_failure:
+            raise OSError("connection lost during safety query")
+        if command == "*IDN?":
+            return "RIGOL TECHNOLOGIES,DS1104Z Plus,OFFLINE,1.0"
+        if "COUPLING?" in command:
+            channel = int(command.rsplit("CH", 1)[1])
+            self.harness.input_reads.append(channel)
+            if len(self.harness.input_reads) == self.harness.input_failure_at:
+                raise OSError("connection lost during current input check")
+            if self.harness.compete:
+                self.harness.check_competitor()
+            if self.harness.change_input and len(self.harness.input_reads) >= 3:
+                return "DC"
+            return "DCL"
+        if command.endswith(":COUPling?"):
+            return "DC"
+        if command.startswith("STATUS?"):
+            return "DCL"
+        if command == ":WAVeform:PREamble?":
+            return "0,0,1000,1,0.00001,0,0,0.01,128,0"
+        if command == ":SYSTem:ERRor?":
+            return '0,"No error"'
+        raise AssertionError(f"unexpected query: {command}")
+
+    def write(self, command):
+        self.events.append(("write", command))
+        if command == "FETCH CH1" and self.harness.failure in {"uncertain", "poisoned", "oserror"}:
+            if self.harness.failure == "oserror":
+                raise OSError("connection lost after dispatch")
+            raise TransportIOError(
+                "waveform write result unknown", operation="write", phase=TransportPhase.SENDING,
+                replay_policy=ReplayPolicy.NO_REPLAY, command_transmission=CommandTransmission.UNKNOWN,
+                response_progress=ResponseProgress.NONE, attempts=1,
+                synchronization=(Synchronization.PROVEN if self.harness.failure == "uncertain"
+                                 else Synchronization.LOST),
+            )
+
+    def query_bin_block(self, command, *, replay=ReplayPolicy.NO_REPLAY):
+        self.events.append(("binary_query", command))
+        return np.full(1000, 128, dtype=np.uint8).tobytes()
+
+    def record_event(self, direction, text):
+        pass
+
+    def close(self):
+        self.events.append(("close", ""))
+        if self.harness.backend_close_failure:
+            raise OSError("backend close failed")
+
+
+class QueryScope:
+    def __init__(self, harness, transport):
+        self.harness = harness
+        self.transport = transport
+
+    def idn(self):
+        return self.transport.query("*IDN?")
+
+    def channel_coupling(self, channel):
+        if channel not in {1, 2}:
+            raise DataError("channel outside fake scope range")
+        if self.harness.mutating_input:
+            self.transport.write("FORBIDDEN INPUT WRITE")
+        return self.transport.query(f"COUPLING? CH{channel}")
+
+    def get_channel_input_state_v2(self, channel):
+        self.channel_coupling(channel)
+        return ScopeChannelInputStateV2(channel, "dc", "unknown", unavailable_fields=("impedance_ohm",))
+
+    def get_snapshot(self, channel):
+        # Legacy snapshots can consume state through text queries. Observation
+        # must use the descriptor's V2 contract instead of this method.
+        self.transport.query(":SYSTem:ERRor?")
+        raise AssertionError("legacy snapshot must not run")
+
+    def get_snapshot_v2(self, channel, *, fields):
+        assert fields == FIELDS
+        if self.harness.mutating_snapshot:
+            self.transport.write("FORBIDDEN SNAPSHOT WRITE")
+        self.idn()
+        coupling = self.transport.query(f"STATUS? CH{channel}")
+        return ScopeSnapshotV2(
+            identity=ScopeIdentitySnapshot("RIGOL", "DS1104Z Plus", "OFFLINE", "1.0", ()),
+            channel=ScopeAnalogChannelSnapshotV2(channel, coupling=coupling),
+            unavailable_fields=tuple(field for field in SCOPE_SNAPSHOT_V2_FIELD_ORDER if field not in FIELDS),
+        )
+
+    def fetch_waveform(self, *, channel, points, check_errors):
+        assert points.upper() == "DEF" and not check_errors
+        self.harness.fetch_health.append(self.transport.session_state.health)
+        if channel == 1:
+            if self.harness.failure == "session_health":
+                raise SessionHealthError(
+                    "session unavailable", health="uncertain", io_kind="write",
+                    epoch_id=self.transport.session_state.epoch_id,
+                )
+            if self.harness.failure == "connection":
+                raise ConnectionError("connection lost outside transport")
+            if self.harness.failure == "config":
+                raise ConfigError("fetch configuration invalid")
+            if self.harness.failure == "data":
+                raise DataError("malformed waveform after completed query")
+        self.transport.write(f"FETCH CH{channel}")
+        if channel == 1 and self.harness.failure == "returned_poisoned":
+            self.transport.session_state.degrade(SessionHealth.POISONED, reason="fake_lost_sync")
+        waveform_type = AnalysisFailureWaveform if channel == 1 and self.harness.failure == "analysis" else WaveformData
+        return waveform_type(
+            channel, WaveformHeader(0.0, 0.01, 1000),
+            np.sin(np.linspace(0.0, 20 * np.pi, 1000)),
+        )
+
+    def close(self):
+        if self.harness.driver_close_failure:
+            raise OSError("plugin close failed before delegation")
+        self.transport.close()
+
+
+class Harness:
+    def __init__(self, root: Path, monkeypatch):
+        self.root = root
+        self.monkeypatch = monkeypatch
+        self.transports = []
+        self.guards = []
+        self.constructor_calls = 0
+        self.constructor_io = None
+        self.input_reads = []
+        self.fetch_health = []
+        self.failure = None
+        self.query_failure = False
+        self.input_failure_at = None
+        self.mutating_input = False
+        self.mutating_snapshot = False
+        self.change_input = False
+        self.compete = False
+        self.driver_close_failure = False
+        self.backend_close_failure = False
+        self.config_path = root / "scope.toml"
+        monkeypatch.setenv("WAVEBENCH_LEASE_DIR", str(root / "leases"))
+        monkeypatch.setattr("wavebench.instruments.factory._open_transport", self.open_transport)
+        self.install()
+
+    def open_transport(self, **kwargs):
+        assert kwargs["resource"] == RESOURCE
+        transport = ConcreteTransport(self)
+        self.transports.append(transport)
+        return transport
+
+    def factory(self, context):
+        self.constructor_calls += 1
+        transport = context.open_transport()
+        assert isinstance(transport, GuardedAuditedTransport)
+        self.guards.append(transport)
+        if self.constructor_io == "write":
+            transport.write(":STOP")
+        elif self.constructor_io == "query":
+            transport.query("*IDN?")
+        elif self.constructor_io == "binary_query":
+            transport.query_bin_block(":WAVeform:DATA?")
+        return QueryScope(self, transport)
+
+    def install(self, *, access="read_write", builtin=False, capabilities=None):
+        if builtin:
+            descriptor = next(item for item in BUILTIN_INSTRUMENTS if item.driver_id == "rigol.ds1104")
+            original_factory = descriptor.factory
+
+            def factory(context):
+                driver = original_factory(context)
+                self.guards.append(driver.transport)
+                return driver
+
+            descriptor = replace(descriptor, factory=factory)
+        else:
+            descriptor = InstrumentDescriptor(
+                driver_id="test.observation-scope", kind="scope", display_name="Offline scope",
+                manufacturer="RIGOL", models=("DS1104Z Plus",), aliases=(),
+                capabilities=capabilities or (
+                    "scope.idn", "scope.snapshot", "scope.snapshot_v2",
+                    "scope.channel_coupling", "scope.fetch_waveform",
+                ),
+                idn_patterns=("RIGOL",), backends=("pyvisa",), option_specs=(),
+                permissions=("instrument.io",), factory=self.factory,
+                scope_coupling_policy="switchable-termination", wavebench_min_version="0.8.24",
+                scope_extensions=ScopeDescriptorExtensions(
+                    snapshot_profile_v2=ScopeSnapshotProfileV2(FIELDS, max_queries=2),
+                ) if capabilities is None or "scope.snapshot_v2" in capabilities else None,
+            )
+        self.descriptor = descriptor
+        registry = InstrumentRegistry(builtins=(descriptor,))
+        self.monkeypatch.setattr("wavebench.instruments.registry.build_instrument_registry", lambda **kwargs: registry)
+        self.config_path.write_text(
+            f'[connection]\nresource = "{RESOURCE}"\nbackend = "pyvisa"\n'
+            f'[scope]\ndriver = "{descriptor.driver_id}"\naccess = "{access}"\ncheck_errors = false\n'
+            '[waveform]\npoints = "def"\n', encoding="utf-8",
+        )
+
+    @property
+    def writes(self):
+        return [command for transport in self.transports for kind, command in transport.events if kind == "write"]
+
+    def report(self, channels=(1, 2)):
+        return scope_waveform_report_payload(config_path=self.config_path, channels=channels)
+
+    def service(self):
+        return ScopeService(load_config(self.config_path), CommandLogger())
+
+    def check_competitor(self):
+        # A separate process exercises the actual OS lock rather than an
+        # in-process mock. It tries to take the lease while a safety query runs.
+        result = subprocess.run(
+            [sys.executable, "-c", (
+                "from wavebench.services.resource_lease import ResourceLease\n"
+                "from wavebench.errors import ResourceBusyError\n"
+                f"lease = ResourceLease({RESOURCE!r})\n"
+                "try:\n    lease.acquire()\n"
+                "except ResourceBusyError:\n    print('busy')\n"
+                "else:\n    lease.release()\n    print('acquired')\n"
+            )], capture_output=True, text=True, timeout=20,
+        )
+        assert result.returncode == 0, result.stderr
+        assert result.stdout.strip() == "busy"
+
+
+@pytest.fixture
+def harness(tmp_path, monkeypatch):
+    return Harness(tmp_path, monkeypatch)
+
+
+def test_report_retains_one_epoch_and_lease_through_all_preflight_and_fetch(harness):
+    harness.compete = True
+    payload = harness.report()
+    assert payload["status"] == "ok"
+    assert len(harness.guards) == len(harness.transports) == 1
+    events = harness.transports[0].events
+    assert events.index(("query", "COUPLING? CH2")) < events.index(("write", "FETCH CH1"))
+    assert harness.input_reads == [1, 2, 1, 1, 2, 2]
+    assert harness.fetch_health == [SessionHealth.HEALTHY, SessionHealth.HEALTHY]
+    assert harness.guards[0].session_state.health is SessionHealth.CLOSED
+    assert harness.guards[0].lease.acquired is False
+    assert payload["waveform_source"]["same_acquisition"] is False
+    assert payload["relationships"][0]["correlation"]["status"] == "skipped"
+
+
+def test_builtin_ds1104_invalid_later_channel_has_zero_writes(harness):
+    harness.install(builtin=True)
+    payload = harness.report((1, 5))
+    assert payload["status"] == "partial"
+    assert [item["waveform"]["status"] for item in payload["channels"]] == ["skipped", "skipped"]
+    assert harness.writes == []
+    assert len(harness.guards) == 1
+
+
+def test_query_only_preflight_blocks_a_plugin_input_write(harness):
+    harness.mutating_input = True
+    payload = harness.report()
+    assert harness.guards[0].access == "read_write"
+    assert harness.writes == []
+    assert all(item["waveform"]["status"] == "skipped" for item in payload["channels"])
+    assert any("SessionHealthError" in warning for warning in payload["warnings"])
+
+
+@pytest.mark.parametrize("access", ["read_write", "read_only", "disabled"])
+def test_readonly_observation_tightens_access_and_uses_v2_query_phase(harness, access):
+    harness.install(access=access)
+    payload = scope_observe_payload(config_path=harness.config_path, channel=1)
+    assert harness.writes == []
+    assert payload["read_only"] and payload["query_only"]
+    if access == "disabled":
+        assert harness.guards == harness.transports == []
+        assert harness.constructor_calls == 0
+        assert payload["scope_status"]["status"] == "skipped"
+        return
+    assert harness.guards[0].access == "read_only"
+    assert payload["scope_status"]["status"] == "ok"
+    assert not any(command == ":SYSTem:ERRor?" for kind, command in harness.transports[0].events)
+
+
+def test_readonly_snapshot_plugin_write_is_blocked(harness):
+    harness.mutating_snapshot = True
+    payload = scope_observe_payload(config_path=harness.config_path)
+    assert payload["scope_status"]["status"] == "unavailable"
+    assert harness.writes == []
+    assert harness.guards[0].counters.blocked_write_requests == 1
+
+
+def test_legacy_snapshot_is_not_used_without_a_pure_query_field_contract(harness):
+    harness.install(capabilities=("scope.idn", "scope.snapshot", "scope.channel_coupling", "scope.fetch_waveform"))
+    payload = scope_observe_payload(config_path=harness.config_path)
+    assert payload["status"] == "partial"
+    assert payload["scope_status"]["status"] == "unavailable"
+    assert payload["identity"]["status"] == payload["coupling"]["status"] == "ok"
+    assert harness.writes == []
+
+
+@pytest.mark.parametrize("failure", [
+    "uncertain", "poisoned", "oserror", "session_health", "connection", "returned_poisoned", "config",
+])
+def test_report_stops_after_uncertain_io_or_fetch_config_error_without_reopen(harness, failure):
+    harness.failure = failure
+    payload = harness.report()
+    assert payload["status"] == "partial"
+    assert payload["channels"][1]["waveform"]["status"] == "skipped"
+    assert payload["channels"][1]["waveform"]["reason"]
+    if failure != "returned_poisoned":
+        expected_type = {
+            "uncertain": "TransportIOError", "poisoned": "TransportIOError", "oserror": "OSError",
+            "session_health": "SessionHealthError", "connection": "ConnectionError", "config": "ConfigError",
+        }[failure]
+        assert payload["channels"][0]["waveform"]["error"]["type"] == expected_type
+    assert "FETCH CH2" not in harness.writes
+    assert len(harness.guards) == 1
+
+
+def test_definitive_data_failure_can_continue_on_the_same_healthy_epoch(harness):
+    harness.failure = "data"
+    payload = harness.report()
+    assert payload["channels"][0]["waveform"]["status"] == "unavailable"
+    assert payload["channels"][1]["waveform"]["status"] == "ok"
+    assert harness.writes == ["FETCH CH2"]
+    assert len(harness.guards) == 1
+
+
+def test_deterministic_analysis_failure_preserves_later_channel_and_expectations(harness):
+    harness.failure = "analysis"
+    payload = scope_waveform_report_payload(
+        config_path=harness.config_path, channels=(1, 2),
+        expectations={1: {"vpp_v": 2.0}, 2: {"vpp_v": 2.0}},
+    )
+    assert payload["status"] == "partial"
+    assert payload["channels"][0]["waveform"]["error"]["type"] == "DataError"
+    assert payload["channels"][1]["waveform"]["status"] == "ok"
+    assert payload["expectations"]["channels"] == {"1": "unavailable", "2": "pass"}
+    assert harness.writes == ["FETCH CH1", "FETCH CH2"]
+    assert len(harness.guards) == 1
+
+
+def test_later_input_uncertainty_preserves_first_waveform_and_stops_mutation(harness):
+    harness.input_failure_at = 5
+    payload = harness.report()
+    assert payload["channels"][0]["waveform"]["status"] == "ok"
+    assert payload["channels"][1]["coupling"]["error"]["type"] == "OSError"
+    assert payload["channels"][1]["waveform"]["status"] == "skipped"
+    assert harness.writes == ["FETCH CH1"]
+    assert len(harness.guards) == 1
+
+
+def test_current_input_is_rechecked_before_fetch(harness):
+    harness.change_input = True
+    payload = harness.report((1,))
+    assert harness.input_reads == [1, 1, 1]
+    assert payload["channels"][0]["waveform"]["status"] == "unavailable"
+    assert "50 ohm" in payload["channels"][0]["waveform"]["error"]["message"]
+    assert harness.writes == []
+
+
+def test_unknown_input_termination_is_rejected_even_with_allow_50ohm(harness):
+    harness.install(capabilities=("scope.idn", "scope.channel_input_state_v2", "scope.fetch_waveform"))
+    payload = scope_waveform_report_payload(config_path=harness.config_path, allow_50ohm=True)
+    assert payload["channels"][0]["waveform"]["status"] == "skipped"
+    assert harness.writes == []
+
+
+@pytest.mark.parametrize("access", ["read_only", "disabled"])
+def test_mutating_report_does_not_enlarge_existing_access(harness, access):
+    harness.install(access=access)
+    payload = harness.report()
+    assert harness.guards == harness.transports == []
+    assert harness.constructor_calls == 0
+    assert harness.writes == []
+    assert all(item["waveform"]["status"] == "skipped" for item in payload["channels"])
+
+
+@pytest.mark.parametrize("missing", ["scope.idn", "scope.fetch_waveform", "scope.channel_coupling"])
+def test_missing_required_capability_hard_blocks_mutation(harness, missing):
+    harness.install(capabilities=tuple(item for item in (
+        "scope.idn", "scope.fetch_waveform", "scope.channel_coupling",
+    ) if item != missing))
+    payload = harness.report()
+    assert harness.writes == []
+    assert harness.guards == harness.transports == []
+    assert harness.constructor_calls == 0
+    assert all(item["waveform"]["status"] == "skipped" for item in payload["channels"])
+
+
+@pytest.mark.parametrize("setting", ['format = "ascii"', 'byte_order = "msbf"'])
+def test_offline_waveform_settings_block_mutation_before_fetch(harness, setting):
+    harness.constructor_io = "write"
+    with harness.config_path.open("a", encoding="utf-8") as file:
+        file.write(setting + "\n")
+    payload = harness.report()
+    assert harness.writes == []
+    assert harness.guards == harness.transports == []
+    assert harness.constructor_calls == 0
+    assert all(item["waveform"]["status"] == "skipped" for item in payload["channels"])
+
+
+def test_direct_offline_fetch_validation_never_runs_factory(harness):
+    harness.constructor_io = "write"
+    with harness.config_path.open("a", encoding="utf-8") as file:
+        file.write('format = "ascii"\n')
+    with pytest.raises(ConfigError, match="waveform.format"):
+        harness.service().validate_observation_fetch()
+    assert harness.constructor_calls == 0
+    assert harness.transports == harness.guards == []
+
+
+def test_invalid_bounded_profile_never_runs_factory(harness):
+    harness.install(capabilities=(
+        "scope.idn", "scope.fetch_waveform", "scope.capture_waveform", "scope.channel_coupling",
+    ))
+    service = harness.service()
+    harness.constructor_io = "write"
+    # The profile omits capture_single although the capability declares it.
+    # Registry/static profile validation must fail before invoking its factory.
+    descriptor = replace(harness.descriptor, scope_extensions=ScopeDescriptorExtensions(
+        waveform_binary_profile=ScopeWaveformBinaryProfile(operations=(
+            ScopeWaveformBinaryOperationProfile(
+                operation_kind="fetch", response_max_bytes=1024,
+                operation_max_bytes=4096, query_max_count=4, resynchronization_max_bytes=0,
+                restore_order=("scope.waveform_source",), snapshot_max_steps=1,
+                restore_max_steps=1, verify_max_steps=1,
+            ),
+        )),
+    ))
+    registry = InstrumentRegistry(builtins=(descriptor,))
+    harness.monkeypatch.setattr("wavebench.instruments.registry.build_instrument_registry", lambda **kwargs: registry)
+    service.descriptor = descriptor
+    with pytest.raises(ConfigError, match="match declared standard waveform capabilities"):
+        service.validate_observation_fetch()
+    with pytest.raises(ConfigError, match="match declared standard waveform capabilities"):
+        harness.report()
+    assert harness.constructor_calls == 0
+    assert harness.transports == harness.guards == []
+
+
+def test_invalid_points_fail_before_any_factory_io(harness):
+    text = harness.config_path.read_text(encoding="utf-8").replace('points = "def"', 'points = "invalid"')
+    harness.config_path.write_text(text, encoding="utf-8")
+    with pytest.raises(ConfigError, match="waveform points"):
+        harness.report()
+    assert not harness.transports
+
+
+@pytest.mark.parametrize("body_failure", [False, True])
+def test_borrowed_context_keeps_driver_state_and_lease_alive(harness, body_failure):
+    bootstrap = harness.service()
+    driver = bootstrap.open_session()
+    service = ScopeService(
+        bootstrap.config, bootstrap.logger, session=driver, descriptor=bootstrap.descriptor,
+        transport=bootstrap.transport, session_state=bootstrap.session_state, lease=bootstrap.lease,
+    )
+    try:
+        def use_borrowed():
+            with service.session_context(observation=True):
+                service.preflight_observation_fetch((1, 2))
+                with service.session_context():
+                    service.fetch_waveform(1)
+                if body_failure:
+                    raise DataError("borrowed body failed")
+
+        if body_failure:
+            with pytest.raises(DataError, match="borrowed body failed"):
+                use_borrowed()
+        else:
+            use_borrowed()
+        assert service.session is driver
+        assert service.session_state.health is SessionHealth.HEALTHY
+        assert service.lease.acquired
+        with pytest.raises(ResourceBusyError):
+            ResourceLease(RESOURCE).acquire()
+    finally:
+        driver.close()
+
+
+@pytest.mark.parametrize("body_failure", [False, True])
+def test_owned_context_cleans_up_plugin_and_backend_close_failures(harness, body_failure):
+    harness.driver_close_failure = harness.backend_close_failure = True
+    service = harness.service()
+    original = DataError("original body failure")
+    with pytest.raises(DataError if body_failure else SessionCloseError) as caught:
+        with service.session_context():
+            if body_failure:
+                raise original
+    if body_failure:
+        assert caught.value is original
+    assert service.session is None
+    assert service.session_state.health is SessionHealth.CLOSED
+    assert not service.lease.acquired
+    lease = ResourceLease(RESOURCE).acquire()
+    lease.release()
+
+
+def test_query_failure_stops_preflight_and_all_waveforms(harness):
+    harness.query_failure = True
+    payload = harness.report()
+    assert payload["identity"]["error"]["type"] == "OSError"
+    assert all(item["waveform"]["status"] == "skipped" for item in payload["channels"])
+    assert harness.writes == []
+    assert len(harness.guards) == 1
+
+
+def test_report_preserves_completed_sections_when_close_fails(harness):
+    harness.driver_close_failure = harness.backend_close_failure = True
+    payload = harness.report()
+    assert payload["status"] == "partial"
+    assert all(item["waveform"]["status"] == "ok" for item in payload["channels"])
+    assert payload["session"]["error"]["type"] == "SessionCloseError"
+    assert harness.guards[0].session_state.health is SessionHealth.CLOSED
+    assert not harness.guards[0].lease.acquired
+    assert len(harness.guards) == 1
+
+
+def test_readonly_missing_identity_never_executes_factory(harness):
+    harness.install(capabilities=("scope.channel_coupling",))
+    payload = scope_observe_payload(config_path=harness.config_path, channels=(1, 2))
+    assert harness.constructor_calls == 0
+    assert harness.transports == harness.guards == []
+    assert payload["status"] == "partial"
+    assert all(item["scope_status"]["status"] == "skipped" for item in payload["channels"])
+
+
+@pytest.mark.parametrize("constructor_io", ["write", "query", "binary_query"])
+@pytest.mark.parametrize("channels", [(1, 2), (1, 5)])
+def test_observation_legacy_factory_io_is_latched_before_preflight(harness, constructor_io, channels):
+    harness.install(capabilities=("scope.idn", "scope.channel_coupling", "scope.fetch_waveform"))
+    harness.constructor_io = constructor_io
+    payload = harness.report(channels)
+    assert harness.constructor_calls == 1
+    assert harness.transports[0].events == [("close", "")]
+    assert harness.writes == []
+    assert harness.guards[0].session_state.health is SessionHealth.CLOSED
+    assert not harness.guards[0].lease.acquired
+    assert all(item["waveform"]["status"] == "skipped" for item in payload["channels"])
+    assert payload["session"]["error"]["type"] == "ConfigError"
+    lease = ResourceLease(RESOURCE).acquire()
+    lease.release()
+
+
+def test_observation_passive_legacy_factory_is_released_after_validation(harness):
+    harness.install(capabilities=("scope.idn", "scope.channel_coupling", "scope.fetch_waveform"))
+    payload = harness.report()
+    assert harness.constructor_calls == 1
+    assert harness.guards[0].construction_latched is False
+    assert all(item["waveform"]["status"] == "ok" for item in payload["channels"])
+    assert harness.writes == ["FETCH CH1", "FETCH CH2"]
+
+
+def test_ordinary_legacy_open_keeps_existing_factory_io_behavior(harness):
+    harness.install(capabilities=("scope.idn", "scope.channel_coupling", "scope.fetch_waveform"))
+    harness.constructor_io = "write"
+    service = harness.service()
+    driver = service.open_session()
+    try:
+        assert harness.writes == [":STOP"]
+    finally:
+        driver.close()
diff --git a/tests/test_scope_observe_cli.py b/tests/test_scope_observe_cli.py
index 83fd6a57..2eed4198 100644
--- a/tests/test_scope_observe_cli.py
+++ b/tests/test_scope_observe_cli.py
@@ -1,6 +1,7 @@
 import io
 import json
-from contextlib import redirect_stderr, redirect_stdout
+from contextlib import nullcontext, redirect_stderr, redirect_stdout
+from dataclasses import asdict
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from unittest.mock import patch
@@ -65,9 +66,34 @@ class _FakeScopeService:
 
     def __init__(self, *, config, logger):
         self.config = config
+        self.session_state = None
         self.fetched_channels: list[int] = []
         _FakeScopeService.instances.append(self)
 
+    def session_context(self, **kwargs):
+        return nullcontext(self)
+
+    def validate_observation_access(self):
+        pass
+
+    def validate_observation_fetch(self):
+        pass
+
+    def preflight_observation_fetch(self, channels, *, allow_50ohm=False):
+        for channel in channels:
+            self.require_high_impedance(channel, allow_50ohm=allow_50ohm)
+
+    def observation_identity(self):
+        return self.idn()
+
+    def observation_status(self, channel):
+        return asdict(self.status(channel))
+
+    def observation_input_safety(self, channel, *, allow_50ohm=False):
+        return {"channel": channel,
+                "coupling": self.require_high_impedance(channel, allow_50ohm=allow_50ohm),
+                "accepted_for_capture": True}
+
     def idn(self):
         return "RIGOL TECHNOLOGIES,DS1104Z Plus,123,1.0"
 
diff --git a/tests/test_waveform_binary_factory.py b/tests/test_waveform_binary_factory.py
index d247de7f..eaf161eb 100644
--- a/tests/test_waveform_binary_factory.py
+++ b/tests/test_waveform_binary_factory.py
@@ -128,7 +128,7 @@ def fetch_waveform_bounded(self, channel, points="dmax", *, baseline):
         return object()
 
 
-def _open() -> object:
+def _open(*, force_deferred_io: bool = False) -> object:
     return open_instrument_driver(
         driver_reference="example.waveform",
         expected_kind="scope",
@@ -139,10 +139,12 @@ def _open() -> object:
         read_retry_attempts=1,
         read_retry_delay_ms=1,
         logger=CommandLogger(),
+        force_deferred_io=force_deferred_io,
     )
 
 
-def test_opt_in_factory_latch_blocks_all_instrument_io_until_validation(monkeypatch) -> None:
+@pytest.mark.parametrize("force_legacy", [False, True])
+def test_opt_in_factory_latch_blocks_all_instrument_io_until_validation(monkeypatch, force_legacy) -> None:
     inner = _InnerTransport()
     errors: list[TransportIOError] = []
 
@@ -167,7 +169,7 @@ def factory(context):
             errors.append(raised.value)
         return _BoundedDriver()
 
-    descriptor = _descriptor(factory=factory, profile=_profile())
+    descriptor = _descriptor(factory=factory, profile=None if force_legacy else _profile())
     monkeypatch.setattr(
         "wavebench.instruments.factory.resolve_instrument_descriptor",
         lambda reference, expected_kind: descriptor,
@@ -178,14 +180,14 @@ def factory(context):
         lambda **kwargs: None,
     )
 
-    opened = _open()
+    opened = _open(force_deferred_io=force_legacy)
 
     assert len(errors) == 7
     assert all(error.reason_code == "factory_construction_pending" for error in errors)
     assert all(error.attempts == 0 for error in errors)
     assert inner.queries == []
     assert inner.writes == []
-    assert opened.transport._has_verified_bounded_binary_backend()
+    assert opened.transport._has_verified_bounded_binary_backend() is (not force_legacy)
     assert opened.transport.query("*IDN?") == "ok"
     assert inner.queries == ["*IDN?"]
 
diff --git a/tests/test_waveform_relationships.py b/tests/test_waveform_relationships.py
index 21880cdd..34d50ef6 100644
--- a/tests/test_waveform_relationships.py
+++ b/tests/test_waveform_relationships.py
@@ -18,7 +18,7 @@ def test_waveform_pair_reports_frequency_voltage_and_phase_for_related_signals()
     left = _waveform(1, np.sin(2 * np.pi * 1000 * t), stop=float(t[-1]))
     right = _waveform(2, 0.5 * np.sin(2 * np.pi * 1000 * (t - 0.00025)) + 0.2, stop=float(t[-1]))
 
-    relationship = analyze_waveform_pair(left, right)
+    relationship = analyze_waveform_pair(left, right, same_acquisition=True)
 
     assert relationship["channels"] == [1, 2]
     assert relationship["common_time"]["overlap"] is True
@@ -60,7 +60,7 @@ def test_waveform_pair_reports_phase_lag_in_degrees():
             stop=float(t[-1]),
         )
 
-        relationship = analyze_waveform_pair(left, right)
+        relationship = analyze_waveform_pair(left, right, same_acquisition=True)
 
         assert relationship["phase_degrees_at_left_frequency"] == pytest.approx(
             expected_degrees, abs=0.5
@@ -74,6 +74,7 @@ def test_waveform_pair_reports_180_degrees_for_inverted_signal():
     relationship = analyze_waveform_pair(
         _waveform(1, left, stop=float(t[-1])),
         _waveform(2, -left, stop=float(t[-1])),
+        same_acquisition=True,
     )
 
     # 用相关峰绝对值选 lag 会把它报成 0°
@@ -87,7 +88,7 @@ def test_waveform_pair_phase_rejects_dc_leakage_in_noninteger_cycle_window():
         2, np.sin(2 * np.pi * 1000 * t - np.pi / 2) + 5.0, stop=float(t[-1]),
     )
 
-    relationship = analyze_waveform_pair(left, right)
+    relationship = analyze_waveform_pair(left, right, same_acquisition=True)
 
     assert relationship["frequency"]["left_hz"] == pytest.approx(1000.0, abs=0.1)
     assert relationship["frequency"]["right_hz"] == pytest.approx(1000.0, abs=0.1)
@@ -113,7 +114,7 @@ def test_waveform_pair_warns_when_frequency_confidence_is_low():
     left = _waveform(1, np.sin(2 * np.pi * 1000 * t), stop=float(t[-1]))
     right = _waveform(2, np.sin(2 * np.pi * 2000 * t), stop=float(t[-1]))
 
-    relationship = analyze_waveform_pair(left, right)
+    relationship = analyze_waveform_pair(left, right, same_acquisition=True)
 
     assert relationship["frequency"]["left_hz"] is None
     assert any("frequency_low_confidence" in warning for warning in relationship["warnings"])
@@ -124,7 +125,7 @@ def test_waveform_pair_reports_intersection_points():
     left = _waveform(1, t - 0.25, stop=float(t[-1]))
     right = _waveform(2, np.zeros_like(t), stop=float(t[-1]))
 
-    relationship = analyze_waveform_pair(left, right)
+    relationship = analyze_waveform_pair(left, right, same_acquisition=True)
 
     intersections = relationship["intersections"]
     assert intersections["mode"] == "finite"
@@ -141,7 +142,7 @@ def test_waveform_pair_can_truncate_many_intersections():
     left = _waveform(1, np.sin(2 * np.pi * 1000 * t), stop=float(t[-1]))
     right = _waveform(2, np.zeros_like(t), stop=float(t[-1]))
 
-    relationship = analyze_waveform_pair(left, right, max_intersections=3)
+    relationship = analyze_waveform_pair(left, right, same_acquisition=True, max_intersections=3)
 
     assert relationship["intersections"]["count"] > 3
     assert relationship["intersections"]["returned"] == 3
@@ -156,8 +157,32 @@ def test_waveform_pair_marks_coincident_waveforms_as_unbounded_intersections():
     relationship = analyze_waveform_pair(
         _waveform(1, values, stop=float(t[-1])),
         _waveform(2, values, stop=float(t[-1])),
+        same_acquisition=True,
     )
 
     assert relationship["intersections"]["mode"] == "coincident"
     assert relationship["intersections"]["count"] is None
     assert "waveforms_coincident_intersections_unbounded" in relationship["warnings"]
+
+
+def test_relationship_entry_points_default_to_unproven_timing():
+    t = np.linspace(0.0, 0.009, 1000)
+    waves = {
+        1: _waveform(1, np.sin(2 * np.pi * 1000 * t)),
+        2: _waveform(2, np.sin(2 * np.pi * 1000 * t - np.pi / 2)),
+    }
+    pair = analyze_waveform_pair(waves[1], waves[2])
+    multiple = analyze_waveform_relationships(waves)
+    for result in (pair, multiple[0]):
+        assert result["common_time"]["same_acquisition"] is False
+        assert result["phase_degrees_at_left_frequency"] is None
+        assert result["correlation"]["status"] == "skipped"
+        assert result["intersections"]["status"] == "skipped"
+        assert result["frequency"]["ratio_high_over_low"] == 1.0
+
+
+@pytest.mark.parametrize("assertion", ["false", 1, None])
+def test_relationship_rejects_truthy_non_boolean_sync_assertions(assertion):
+    wave = _waveform(1, np.zeros(10))
+    with pytest.raises(ValueError, match="same_acquisition"):
+        analyze_waveform_pair(wave, wave, same_acquisition=assertion)