fix(studio-3d): 字段级来源与失效状态 —— 旧光流/算法失败不再冒充实时测量

- What: 1. **采集侧声明每个派生数组的归属**（`acquisition/models.py`）：`SdkPacket` 新增 `flow_valid/flow_frame_id/depth_valid/depth_frame_id/forces_valid`（`None` = 未声明）。两个真实生产者都填写：`finger_worker`/`sdk_worker` 逐帧记录"该数组实际算出来的 frame_id"，链失败时 `*_valid=False` 并把上一次成功帧号带出来。光流与形变链分开判定（力读取失败不等于光流失败），但算法链失败时本帧的一切派生量都不可声明。 2. **adapter 按声明收数据**（`scene3d/adapters.py`）：`_field_validity()` 三态——未声明 → 拒绝并报 `field_unverified`（没声明就不算数）；声明为假 → `field_stale`（把 `invalid_reason` 带进诊断）；声明帧号不是本帧 → `field_stale`（两个帧号都写进 details）。`_intersect_device_mask` 换成 `_device_mask`：设备 mask 形状不符**拒绝该帧**（`field_domain_mismatch`，两个形状都写进诊断），既不拉伸也不静默忽略。发布的帧带 `computed_frame_id`，身份一路带到显示侧。 3. **契约**（`scene3d/contracts.py`）：`SceneState` 加 `"stale"`；诊断码加 `field_stale`/`field_unverified` （warning）；`SurfaceFieldFrame.computed_frame_id` 与 `RenderFrame.computed_frame_id`（由 composer 原样透传）；`validate_surface_frame` 校验帧号非负且不晚于所属帧。新增 locale： `scene3d.state_stale`（中英各一条）。 4. **协调器不再"边失败边报 live"**（`scene3d/coordinator.py`）：新增 `_stale_since` 宽限状态机与 `note_field_failure()`——字段被拒、帧被判 invalid、compose 抛错、或合成结果的 `computed_frame_id` 落后于本帧，都立即把状态改为 `stale` 并保留最后一帧；宽限（`stale_grace_s`，默认 1.0 s）到期后回参考模型并转 `no_data`；收到好帧即回 `live` 并清掉宽限。顺带修掉"等待首帧时每个 tick 都把参考模型重新 present 一次"（`_rest_presented`，16 ms 一次的无谓上传）。 5. **诊断真正到达界面**（`scene3d/view.py`）：`submit_packet()` 在 adapter 拒绝时把 `adapter.last_diagnostic` 交给 `note_field_failure()`；`tick()` 把协调器诊断的变化经已有 `diagnosticRaised` 冒泡（协调器自身仍然零 Qt 依赖）。新增 `_snapshot_diagnostic()` 与 `_adapter` 引用（后续"按能力禁用控件"也要问它）。 - Why: 审查报告 P1/R2：`FingerFrameAdapter.snapshot()` 从不读 `valid`/`stale_forces`/`invalid_reason`，只看到有限的 `flow` 就当作新数据上色（复现：`valid=False, stale_forces=True` + 有限 flow → 仍有有效着色点、diagnostic=None）；compose 抛错只写 `last_diagnostic`，画面与 `state` 保持不变； `snapshot()` 返回 None 时诊断无人读取；mask 域不符时静默忽略会扩大可信范围。力读数失败不该机械地判死三维，但算法失败也不能因为旧数组是有限值就当成新数据——所以按**字段级来源**处理，并让失效有明确的、可见的状态而不是继续保持 live。 - Verification: · `test_scene3d_adapters.py` 23 通过（+6 新增）：stale flow 被拒且带原因、未声明 flow → `field_unverified`、帧号不符 → `field_stale`（details 两个帧号）、新帧记录 `computed_frame_id`、box 侧"形变链失败仍用光流"、 box 全部字段不可声明时拒绝；mask 域不符由"被忽略"改为"拒绝该帧"（原断言已改）。 · `test_scene3d_coordinator.py` 30 通过（+6 新增）：落后帧号的合成结果永不上屏且状态 stale、字段被拒 → 宽限内保留画面/宽限后回参考模型/好帧回 live、compose 抛错不再保持 live、invalid 帧也计入宽限。 · `test_scene3d_view.py`（+3 新增，offscreen 可跑）：adapter 拒绝 → 协调器 stale 且诊断经 `diagnosticRaised` 回到面板、`scene3d.state_stale` 不是裸键；未声明字段 → `field_unverified`；新鲜字段照常上屏回到 live。 · 全量：apps/studio 228 项 0 失败 14 跳过；shared 301 项 0 失败 1 跳过（`--junitxml` 计数）。 - Not in this change: `docs/architecture/studio-3d-contracts.pyi` 与新契约的同步放在阶段 6 的文档提交里一起做（本阶段契约改动较多，避免每提交改一次文档）；设备几何身份（R3）、资产加载请求号（R7）、二维后端统一、限速与真实统计均未动；真机上"链失败时 VRAW 是否也过期"这一条仍需实测（若证明链失败不影响 VRAW，`flow_valid` 放宽为只看读取结果是一行改动，已写进计划的风险项）。
2026-09-14 21:03:07 +08:00
parent d06e116c15
commit 7d1fca4c93
15 changed files with 721 additions and 38 deletions
@@ -411,6 +411,7 @@
    "state_live": "Live",
    "state_no_data": "No valid measurement in this frame — showing the surface model",
    "state_suspended": "Suspended (the view is hidden)",
+    "state_stale": "Measurement expired — the last frame is still shown but is not current",
    "state_unavailable": "3D is unavailable",
    "source_live": "Live acquisition",
    "source_live_tip": "The picture comes from the data being acquired right now.",
@@ -411,6 +411,7 @@
    "state_live": "实时",
    "state_no_data": "本帧没有有效测量 — 显示表面模型",
    "state_suspended": "已挂起（视图被隐藏）",
+    "state_stale": "测量已过期 — 上一帧仍显示，但已不是当前数据",
    "state_unavailable": "三维不可用",
    "source_live": "实时采集",
    "source_live_tip": "画面来自此刻正在采集的数据。",
@@ -12,6 +12,12 @@ The constraints here are about *where* the data may be read, not about optics:
  field is on a different grid than the scene's UV domain. A remap that is not
  configured is not guessed at: the adapter says which shapes it saw, so the
  mismatch is a fact in the diagnostics rather than a picture that looks plausible.
+* it publishes **nothing** it cannot claim. The packet states, per derived array,
+  whether that array is this frame's own measurement (``flow_valid``/``flow_frame_id``,
+  ``depth_valid``/``depth_frame_id``); an unstated array is unverified, a failed chain's
+  leftover array is stale, and both are refused with a diagnostic instead of being
+  coloured. A device validity mask that describes another grid refuses the frame too:
+  stretching it invents valid samples, and ignoring it widens the trusted area.

 Two things the v1 adapters deliberately do **not** do:

@@ -74,22 +80,74 @@ def _shape_of(array: Optional[np.ndarray]) -> str:
    return "none" if array is None else "x".join(str(int(v)) for v in array.shape[:2])


-def _intersect_device_mask(
-    packet: SdkPacket, mask: np.ndarray, shape_hw: tuple[int, int]
-) -> np.ndarray:
-    """AND in the device's own validity, when it describes this grid.
+def _field_validity(packet: SdkPacket, field: str, frame_id: int) -> Optional[Diagnostic]:
+    """Whether the packet may claim ``field`` for *this* frame.

-    A device mask of the wrong shape is ignored rather than resized: a mask from
-    another grid says nothing about these pixels, and stretching it would mark
-    samples valid that the device never claimed.
+    Three answers, and the middle one is the point: ``None`` on the packet means the
+    producer made **no statement**, and an unstated field is unverified rather than
+    fresh — inventing "it is probably current" is how a stale array becomes a
+    measurement nobody measured. A failed chain leaves the previous frame's arrays in
+    place, so ``False`` (or a producer frame id that is not this frame's) is stale data
+    wearing a new frame's identity.
+    """
+    stated = getattr(packet, f"{field}_valid", None)
+    if stated is None:
+        return diagnostic(
+            "field_unverified",
+            f"this frame does not state whether its {field} belongs to it",
+        )
+    if not bool(stated):
+        reason = str(getattr(packet, "invalid_reason", "") or "")
+        return diagnostic(
+            "field_stale",
+            f"this frame's {field} is not this frame's measurement"
+            + (f": {reason}" if reason else ""),
+        )
+    produced = getattr(packet, f"{field}_frame_id", None)
+    if produced is not None and int(produced) != int(frame_id):
+        return diagnostic(
+            "field_stale",
+            f"this frame's {field} was computed for another frame",
+            computed=str(int(produced)),
+            frame=str(int(frame_id)),
+        )
+    return None
+
+
+def _computed_frame_id(packet: SdkPacket, field: str, frame_id: int) -> int:
+    """The frame the published array belongs to, for the display side to re-check."""
+    produced = getattr(packet, f"{field}_frame_id", None)
+    return int(frame_id if produced is None else int(produced))
+
+
+def _device_mask(
+    packet: SdkPacket, shape_hw: tuple[int, int]
+) -> tuple[Optional[np.ndarray], Optional[Diagnostic]]:
+    """The device's own validity, or a diagnostic when it describes another grid.
+
+    Both silent choices are wrong: stretching a mask from another grid marks samples
+    valid that the device never claimed, and ignoring it *widens* the trusted area to
+    the whole UV domain. So a mismatch refuses the frame and names both shapes.
    """
    explicit = getattr(packet, "valid_mask", None)
    if explicit is None:
+        return None, None
+    array = np.asarray(explicit, dtype=bool)
+    if tuple(array.shape) != tuple(shape_hw):
+        return None, diagnostic(
+            "field_domain_mismatch",
+            "the device's validity mask describes another grid",
+            mask=_shape_of(array),
+            domain=f"{int(shape_hw[0])}x{int(shape_hw[1])}",
+        )
+    return array, None
+
+
+def _intersect(mask: np.ndarray, device: Optional[np.ndarray]) -> np.ndarray:
+    """Inside the UV island AND the device's own valid area, when it stated one."""
+    if device is None:
        return freeze(mask, dtype=np.bool_)
-    explicit_array = np.asarray(explicit, dtype=bool)
-    if tuple(explicit_array.shape) == tuple(shape_hw):
-        return freeze(np.asarray(mask, dtype=bool) & explicit_array, dtype=np.bool_)
-    return freeze(mask, dtype=np.bool_)
+    return freeze(np.asarray(mask, dtype=bool) & device, dtype=np.bool_)


 class FingerFrameAdapter:
@@ -141,6 +199,12 @@ class FingerFrameAdapter:
    def snapshot(
        self, packet: SdkPacket, runtime: Any, identity: FrameIdentity
    ) -> Optional[SurfaceFieldFrame]:
+        # Provenance first, and before the cheap array work: a packet that cannot say
+        # its flow is this frame's must not be coloured, whatever the array holds.
+        problem = _field_validity(packet, "flow", identity.frame_id)
+        if problem is not None:
+            self.last_diagnostic = problem
+            return None
        magnitude = flow_magnitude(packet)
        if magnitude is None:
            self.last_diagnostic = diagnostic(
@@ -174,7 +238,11 @@ class FingerFrameAdapter:
            return None
        # Inside the UV island is the only place a flow value means anything: the
        # tracker is looking at masked black outside it.
-        mask = _intersect_device_mask(packet, mask, self.shape_hw)
+        device, mask_problem = _device_mask(packet, self.shape_hw)
+        if mask_problem is not None:
+            self.last_diagnostic = mask_problem
+            return None
+        mask = _intersect(mask, device)
        self.last_diagnostic = None
        return SurfaceFieldFrame(
            identity=identity,
@@ -183,6 +251,7 @@ class FingerFrameAdapter:
            scalar=scalar,
            scalar_descriptor=FLOW_MAGNITUDE,
            valid=mask,
+            computed_frame_id=_computed_frame_id(packet, "flow", identity.frame_id),
        )


@@ -218,19 +287,29 @@ class BoxFrameAdapter:
            )
            return None
        candidates = (
-            ("depth_response", getattr(packet, "depth", None)),
-            ("flow_magnitude", flow_magnitude(packet)),
+            ("depth", getattr(packet, "depth", None)),
+            ("flow", flow_magnitude(packet)),
        )
+        device, mask_problem = _device_mask(packet, self.shape_hw)
+        if mask_problem is not None:
+            self.last_diagnostic = mask_problem
+            return None
+        refused: Optional[Diagnostic] = None
        for name, field in candidates:
            if field is None:
                continue
+            # The depth channel is the packet's *deformation* output: a failed chain
+            # leaves the previous frame's map in place, so provenance decides whether
+            # this frame may colour it at all.
+            problem = _field_validity(packet, name, identity.frame_id)
+            if problem is not None:
+                refused = refused or problem
+                continue
            array = np.asarray(field, dtype=np.float32)
            if array.ndim != 2 or tuple(array.shape) != self.shape_hw:
                continue
            # The device's own mask wins: it knows where its surface is.
-            mask = _intersect_device_mask(
-                packet, np.isfinite(array), self.shape_hw
-            )
+            mask = _intersect(np.isfinite(array), device)
            self.last_diagnostic = None
            return SurfaceFieldFrame(
                identity=identity,
@@ -238,10 +317,14 @@ class BoxFrameAdapter:
                field_domain_id=self.field_domain_id,
                scalar=freeze(array, dtype=np.float32),
                scalar_descriptor=(
-                    DEPTH_RESPONSE if name == "depth_response" else FLOW_MAGNITUDE
+                    DEPTH_RESPONSE if name == "depth" else FLOW_MAGNITUDE
                ),
                valid=freeze(mask, dtype=np.bool_),
+                computed_frame_id=_computed_frame_id(packet, name, identity.frame_id),
            )
+        if refused is not None:
+            self.last_diagnostic = refused
+            return None
        self.last_diagnostic = diagnostic(
            "field_unavailable",
            "no usable scalar field of the scene's shape in this frame",
@@ -123,6 +123,14 @@ class ThreeDCoordinator:
        self._last_composed_identity: Optional[FrameIdentity] = None
        #: What is on screen and when it got there, for the pane's frame counter.
        self._last_presented: Optional[tuple[FrameIdentity, float]] = None
+        #: When the current measurement stopped being current, or ``None`` while the
+        #: view is live. Set by a field that could not be built, a compose that
+        #: failed, or a frame whose array belongs to an earlier acquisition frame.
+        self._stale_since: Optional[float] = None
+        #: Whether the reference model is already on screen for the current claim.
+        #: Without this the pane re-uploaded the same rest frame on every 16 ms tick
+        #: while it waited for the first measurement.
+        self._rest_presented = False
        #: The field frame currently on screen, so a gain change can redraw exactly
        #: it rather than a newer instant.
        self._last_field_frame: Optional[SurfaceFieldFrame] = None
@@ -193,6 +201,8 @@ class ThreeDCoordinator:
        self._last_composed_identity = None
        self._last_field_frame = None
        self._settings_dirty = False
+        self._stale_since = None
+        self._rest_presented = False
        item = self._inflight
        if item is not None:
            item.future.cancel()
@@ -212,7 +222,7 @@ class ThreeDCoordinator:
        self._generation += 1
        self._widget.set_scene(assets)
        if self._visible:
-            self._widget.present(self._composer.rest_frame(assets, self._identity(0)))
+            self._present_rest()
            self._set_state("ready")
        else:
            self._set_state("loading" if self._visible else "suspended")
@@ -254,7 +264,7 @@ class ThreeDCoordinator:
        if self._assets is not None:
            # The reference model, not the last measurement: the new session has not
            # produced anything yet, and an old picture would be a claim about it.
-            self._widget.present(self._composer.rest_frame(self._assets, self._identity(0)))
+            self._present_rest()
            self._set_state("ready" if self._visible else "suspended")

    # ---------------------------------------------------------------- producers
@@ -307,6 +317,9 @@ class ThreeDCoordinator:
        if problem is not None:
            self._invalid_frames += 1
            self._last_diagnostic = problem
+            # A frame that cannot be used is also "no measurement is arriving": the
+            # view must stop claiming the last picture is current, and say why.
+            self._mark_stale()
            return "invalid"
        with self._lock:
            if session_id is None:
@@ -315,6 +328,21 @@ class ThreeDCoordinator:
                self._last_request_id = max(self._last_request_id, identity.request_id)
        return self._incoming.offer(frame)

+    def note_field_failure(self, item: Optional[Diagnostic] = None) -> None:
+        """A producer could not build a field frame from this packet.
+
+        Called from the acquisition thread by the sink that owns the adapter: when the
+        adapter refuses (an unstated field, a stale one, a mask from another grid) there
+        is no frame to submit, and without this the view would keep saying ``live``
+        about a picture nothing is producing any more. The picture itself is not
+        replaced here — the GUI thread decides that, once the grace has passed.
+        """
+        if self._closed or not self._visible:
+            return
+        if item is not None:
+            self._last_diagnostic = item
+        self._mark_stale()
+
    @staticmethod
    def _reject_reason(
        frame: SurfaceFieldFrame, assets: SceneAssets
@@ -353,6 +381,7 @@ class ThreeDCoordinator:
        if self._closed or not self._visible or self._assets is None:
            return
        self._collect_inflight()
+        self._expire_stale()
        if self._inflight is None:
            if self._settings_dirty and self._last_field_frame is not None:
                # A gain change re-draws the field that is already on screen; taking
@@ -370,16 +399,33 @@ class ThreeDCoordinator:
            if not self._identity_is_current(composed.identity):
                self._stale_frames += 1
                return
+            if self._array_is_stale(composed):
+                # The colour belongs to an earlier acquisition frame. Presenting it as
+                # the current instant is exactly the "old array read as new data"
+                # failure, so the picture on screen stays and the state says stale.
+                self._stale_frames += 1
+                self._last_diagnostic = diagnostic(
+                    "field_stale",
+                    "the composed field belongs to an earlier acquisition frame",
+                    computed=str(composed.computed_frame_id),
+                    frame=str(composed.identity.frame_id),
+                )
+                self._mark_stale()
+                return
            self._widget.set_superseded_frames(self.superseded_frames)
            self._widget.present(composed)
            self._last_composed_identity = composed.identity
            self._last_presented = (composed.identity, self._clock())
+            # A frame arrived with its own fields: whatever went stale is current
+            # again, and the grace starts over next time it fails.
+            self._stale_since = None
            self._set_state("live" if composed.valid.any() else "no_data")
        elif self._state in ("ready", "loading"):
            # Nothing has arrived yet: show the reference model rather than a blank
-            # window, and say that no measurement is being displayed.
-            if self._last_composed_identity is None:
-                self._widget.present(self._composer.rest_frame(self._assets, self._identity(0)))
+            # window, and say that no measurement is being displayed. Only once per
+            # claim — re-uploading the same model every tick is pure cost.
+            if self._last_composed_identity is None and not self._rest_presented:
+                self._present_rest()
                self._set_state("no_data")

    def set_settings(self, settings: DisplaySettings) -> None:
@@ -408,7 +454,7 @@ class ThreeDCoordinator:
        if self._closed or self._assets is None or not self._visible:
            return
        self._invalidate()
-        self._widget.present(self._composer.rest_frame(self._assets, self._identity(0)))
+        self._present_rest()
        self._set_state("no_data")

    def close(self) -> None:
@@ -420,6 +466,49 @@ class ThreeDCoordinator:
        self._set_state("disabled")

    # ------------------------------------------------------------------ internal
+    def _present_rest(self) -> None:
+        """Put the asset's own geometry on screen, once, as the no-measurement state."""
+        assets = self._assets
+        if assets is None:
+            return
+        self._widget.present(self._composer.rest_frame(assets, self._identity(0)))
+        self._rest_presented = True
+
+    def _mark_stale(self) -> None:
+        """The measurement is no longer current: say so, and start the grace clock.
+
+        Only the *claim* changes here. What is on screen is replaced by the reference
+        model once ``_expire_stale`` decides the gap is no longer a hiccup — a single
+        refused frame must not make the operator's model vanish.
+        """
+        if self._stale_since is None:
+            self._stale_since = self._clock()
+        if self._assets is not None:
+            self._set_state("stale")
+
+    def _expire_stale(self) -> None:
+        """GUI side of :meth:`_mark_stale`: after the grace, drop the old picture."""
+        since = self._stale_since
+        if since is None or self._assets is None:
+            return
+        if (self._clock() - since) < self._stale_grace_s:
+            return
+        # Held for the grace and no good frame since: the picture is now a claim about
+        # a measurement nobody is making, so the reference model goes back on screen.
+        self._present_rest()
+        self._last_composed_identity = None
+        self._last_field_frame = None
+        self._stale_since = None
+        self._incoming.clear()
+        self._set_state("no_data")
+
+    @staticmethod
+    def _array_is_stale(composed: RenderFrame) -> bool:
+        computed = composed.computed_frame_id
+        if computed is None:
+            return False
+        return int(computed) != int(composed.identity.frame_id)
+
    def _identity(self, request_id: int) -> FrameIdentity:
        return FrameIdentity(
            source_id=self._source_id,
@@ -478,6 +567,10 @@ class ThreeDCoordinator:
            self._last_diagnostic = getattr(exc, "diagnostic", None) or diagnostic(
                "field_unavailable", f"compose failed: {type(exc).__name__}: {exc}"
            )
+            # A compose that failed means the picture on screen is no longer being
+            # produced; it must not keep the status "live" on the strength of the
+            # frame that happened to be there before.
+            self._mark_stale()
            return
        if composed is None:
            return
@@ -100,11 +100,17 @@ class Scene3DView(QWidget):
        self._assets: Optional[SceneAssets] = None
        self._expected: Optional[ExpectedGeometry] = None
        self._snapshot: Optional[Callable[[Any, FrameIdentity], Any]] = None
+        #: The adapter behind ``_snapshot``: it owns the reason a packet was refused,
+        #: and (for a source that cannot provide a displacement) what the pane may
+        #: offer. Kept rather than closed over so those answers can be asked.
+        self._adapter: Any = None
        self._source_id = ""
        self._enabled = False
        self._last_diagnostic: Optional[Diagnostic] = None
        self._last_descriptor: Optional[FieldDescriptor] = None
        self._last_stats: Optional[RenderStats] = None
+        #: The coordinator's last diagnostic as *this* view has already reported it.
+        self._reported_diagnostic: Optional[Diagnostic] = None

        self._layout = QVBoxLayout(self)
        self._layout.setContentsMargins(0, 0, 0, 0)
@@ -209,6 +215,7 @@ class Scene3DView(QWidget):
        here keeps the acquisition path free of both variants.
        """
        asset_id = str(getattr(assets, "asset_id", "") or "")
+        self._adapter = None
        common = {
            "geometry_id": assets.geometry_id,
            "field_domain_id": assets.field_domain_id,
@@ -224,11 +231,13 @@ class Scene3DView(QWidget):
                )
                return None
            adapter = BoxFrameAdapter(**common)
+            self._adapter = adapter
            return lambda packet, identity: adapter.snapshot(packet, identity, expected)
        # The finger's flow arrives already unrolled on the scene's UV domain
        # (measured: apps/studio/tests/test_scene3d_adapters.py), so no remap is
        # passed here. Remapping it again would index the flow at source pixels.
        adapter = FingerFrameAdapter(**common)
+        self._adapter = adapter
        return lambda packet, identity: adapter.snapshot(packet, None, identity)

    # --------------------------------------------------------------- producer
@@ -245,9 +254,19 @@ class Scene3DView(QWidget):
            frame = snapshot(packet, identity)
            if frame is not None:
                coordinator.submit(frame)
+                return
+            # The adapter refused this packet, so there is no frame to submit — but
+            # the *reason* has to reach the display, or the pane keeps claiming the
+            # last picture is live while nothing is producing it any more.
+            coordinator.note_field_failure(self._snapshot_diagnostic())
        except Exception:  # noqa: BLE001 - a display sink must never break capture
            logger.exception("the 3D sink failed for one packet")

+    def _snapshot_diagnostic(self) -> Optional[Diagnostic]:
+        """Why the adapter refused the last packet (the adapter owns the answer)."""
+        item = getattr(self._adapter, "last_diagnostic", None)
+        return item if isinstance(item, Diagnostic) else None
+
    def identity_for(self, packet: Any) -> Optional[FrameIdentity]:
        """The identity a field frame built from this packet would carry."""
        if self._coordinator is None or self._assets is None:
@@ -486,11 +505,22 @@ class Scene3DView(QWidget):
        if self._coordinator is None or not self._enabled:
            return
        self._coordinator.tick()
+        # The coordinator owns the truths a pane cannot see from here — a refused
+        # field, an expired measurement, a failed compose. It has no Qt signals by
+        # design, so the one place that already runs per tick publishes them.
+        self._publish_coordinator_diagnostic(self._coordinator.last_diagnostic)
        state = self._coordinator.state
        if self._panel is not None:
            self._panel.set_state(state)
        self.stateChanged.emit(state)

+    def _publish_coordinator_diagnostic(self, item: Optional[Diagnostic]) -> None:
+        """Raise a coordinator diagnostic once, when it changes."""
+        if item is None or item == self._reported_diagnostic:
+            return
+        self._reported_diagnostic = item
+        self._on_diagnostic(item)
+
    def _on_capabilities(self, capabilities: Any) -> None:
        software = bool(getattr(capabilities, "software_renderer", False))
        if software and self._panel is not None:
@@ -526,6 +556,7 @@ class Scene3DView(QWidget):
        """Drop the scene, keep the pane: a lock violation is not a crash."""
        self._assets = None
        self._snapshot = None
+        self._adapter = None
        if self._coordinator is not None:
            self._coordinator.close()
        if self._widget is not None:
@@ -32,12 +32,24 @@ from orisys_studio.scene3d.adapters import (

@dataclass
 class FakePacket:
-    """The parts of ``SdkPacket`` the adapters look at."""
+    """The parts of ``SdkPacket`` the adapters look at.
+
+    The provenance flags default to *stated fresh*: that is what both production
+    workers now write, and it keeps the older tests about grids and units about grids
+    and units. A test that wants a stale or unstated field says so — that is the
+    behaviour under test.
+    """

    flow: Optional[np.ndarray] = None
    depth: Optional[np.ndarray] = None
    valid_mask: Optional[np.ndarray] = None
    frame_id: int = 1
+    invalid_reason: str = ""
+    flow_valid: Optional[bool] = True
+    flow_frame_id: Optional[int] = None
+    depth_valid: Optional[bool] = True
+    depth_frame_id: Optional[int] = None
+    forces_valid: Optional[bool] = True


 def _identity(generation: int = 1) -> FrameIdentity:
@@ -242,7 +254,8 @@ def test_the_finger_adapter_does_not_colour_outside_the_uv_island(scene):
    )


-def test_a_device_mask_from_another_grid_is_ignored_not_stretched(scene):
+def test_a_device_mask_from_another_grid_refuses_the_frame(scene):
+    """Neither stretching it nor ignoring it is honest — so the frame is refused."""
    adapter = FingerFrameAdapter(
        geometry_id=scene.geometry_id,
        field_domain_id=scene.field_domain_id,
@@ -253,8 +266,117 @@ def test_a_device_mask_from_another_grid_is_ignored_not_stretched(scene):
    frame = adapter.snapshot(
        FakePacket(flow=flow, valid_mask=wrong), runtime=None, identity=_identity()
    )
+    assert frame is None, "a mask from another grid must not widen the trusted area"
+    item = adapter.last_diagnostic
+    assert item is not None and item.code == "field_domain_mismatch"
+    assert item.details["mask"] == "7x7"
+    assert item.details["domain"] == f"{scene.shape_hw[0]}x{scene.shape_hw[1]}"
+
+
+# ------------------------------------------------- field provenance (stage 6, R2)
+def _finger(scene) -> FingerFrameAdapter:
+    return FingerFrameAdapter(
+        geometry_id=scene.geometry_id,
+        field_domain_id=scene.field_domain_id,
+        shape_hw=scene.shape_hw,
+    )
+
+
+def test_a_stale_flow_is_refused_instead_of_coloured(scene):
+    """R2: the array is still finite, but it is not this frame's measurement.
+
+    This is the audit's reproduction: a packet with ``valid=False``/``stale_forces``
+    and a filled ``flow`` used to be coloured as valid data with no diagnostic.
+    """
+    adapter = _finger(scene)
+    flow = np.full((*scene.shape_hw, 2), 4.0, dtype=np.float32)
+    frame = adapter.snapshot(
+        FakePacket(
+            flow=flow,
+            flow_valid=False,
+            invalid_reason="compute: ValueError: boom",
+        ),
+        runtime=None,
+        identity=_identity(),
+    )
+    assert frame is None, "a failed chain's leftover flow was coloured as current"
+    item = adapter.last_diagnostic
+    assert item is not None and item.code == "field_stale"
+    assert "boom" in item.message
+
+
+def test_an_unstated_flow_is_unverified_not_fresh(scene):
+    """No statement is not the same as "it is mine"."""
+    adapter = _finger(scene)
+    flow = np.full((*scene.shape_hw, 2), 1.0, dtype=np.float32)
+    frame = adapter.snapshot(
+        FakePacket(flow=flow, flow_valid=None), runtime=None, identity=_identity()
+    )
+    assert frame is None
+    item = adapter.last_diagnostic
+    assert item is not None and item.code == "field_unverified"
+
+
+def test_flow_computed_for_another_frame_is_refused(scene):
+    adapter = _finger(scene)
+    flow = np.full((*scene.shape_hw, 2), 2.0, dtype=np.float32)
+    frame = adapter.snapshot(
+        FakePacket(flow=flow, flow_frame_id=3),
+        runtime=None,
+        identity=_identity(),  # frame_id == 1
+    )
+    assert frame is None
+    item = adapter.last_diagnostic
+    assert item is not None and item.code == "field_stale"
+    assert item.details["computed"] == "3" and item.details["frame"] == "1"
+
+
+def test_a_fresh_flow_records_the_frame_it_belongs_to(scene):
+    """The identity travels with the frame, so the display side can re-check it."""
+    adapter = _finger(scene)
+    flow = np.full((*scene.shape_hw, 2), 3.0, dtype=np.float32)
+    frame = adapter.snapshot(
+        FakePacket(flow=flow, flow_frame_id=1), runtime=None, identity=_identity()
+    )
    assert frame is not None
-    assert np.asarray(frame.valid).all()
+    assert frame.computed_frame_id == 1
+    assert validate_surface_frame(frame) == ()
+
+
+def test_the_box_adapter_skips_a_stale_depth_and_uses_the_flow(scene):
+    """A failed deformation chain does not have to cost the flow as well."""
+    adapter = BoxFrameAdapter(
+        geometry_id=scene.geometry_id,
+        field_domain_id=scene.field_domain_id,
+        shape_hw=scene.shape_hw,
+    )
+    depth = np.full(scene.shape_hw, 0.5, dtype=np.float32)
+    flow = np.zeros((*scene.shape_hw, 2), dtype=np.float32)
+    flow[..., 0] = 1.0
+    frame = adapter.snapshot(
+        FakePacket(depth=depth, depth_valid=False, flow=flow),
+        _identity(),
+        ExpectedGeometry("box", "rev", scene.geometry_id, scene.field_domain_id, scene.shape_hw),
+    )
+    assert frame is not None and frame.scalar_descriptor == FLOW_MAGNITUDE
+    assert np.allclose(np.asarray(frame.scalar), 1.0)
+
+
+def test_the_box_adapter_refuses_when_no_field_is_its_own(scene):
+    adapter = BoxFrameAdapter(
+        geometry_id=scene.geometry_id,
+        field_domain_id=scene.field_domain_id,
+        shape_hw=scene.shape_hw,
+    )
+    depth = np.full(scene.shape_hw, 0.5, dtype=np.float32)
+    frame = adapter.snapshot(
+        FakePacket(depth=depth, depth_valid=False),
+        _identity(),
+        ExpectedGeometry("box", "rev", scene.geometry_id, scene.field_domain_id, scene.shape_hw),
+    )
+    assert frame is None
+    item = adapter.last_diagnostic
+    assert item is not None and item.code == "field_stale"


 def _run_map():
@@ -80,7 +80,7 @@ class CountingComposer(SceneComposer):


 def _frame(scene, *, generation=1, source="live-A", session=1, frame_id=1, value=1.0,
-           domain=None, valid=True):
+           domain=None, valid=True, computed=None):
    scalar = np.full(scene.shape_hw, value, dtype=np.float32)
    mask = np.full(scene.shape_hw, valid, dtype=bool)
    return SurfaceFieldFrame(
@@ -90,6 +90,7 @@ def _frame(scene, *, generation=1, source="live-A", session=1, frame_id=1, value
        scalar=freeze(scalar, dtype=np.float32),
        scalar_descriptor=FieldDescriptor("flow_magnitude", "px", "approximate", False),
        valid=freeze(mask, dtype=np.bool_),
+        computed_frame_id=computed,
    )


@@ -640,3 +641,151 @@ def test_a_result_from_an_older_request_is_dropped_on_arrival(scene):
        "a result for a superseded request was presented"
    )
    coordinator.close()
+
+
+# --------------------------------------------- a field that is not this frame's (R2)
+class Clock:
+    """A monotonic clock the test moves by hand."""
+
+    def __init__(self, now: float = 0.0) -> None:
+        self.now = float(now)
+
+    def __call__(self) -> float:
+        return self.now
+
+    def advance(self, seconds: float) -> None:
+        self.now += float(seconds)
+
+
+class FailingComposer(SceneComposer):
+    """A composer that can be switched to raising, like a broken chain."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.fail = False
+
+    def compose(self, assets, frame, settings):
+        if self.fail:
+            raise RuntimeError("synthetic compose error")
+        return super().compose(assets, frame, settings)
+
+
+def _stale_diagnostic():
+    from orisys_app_shared.scene3d.contracts import diagnostic
+
+    return diagnostic("field_stale", "this frame's flow is not its own measurement")
+
+
+def test_a_stale_array_is_never_presented_as_the_current_instant(scene):
+    """R2: a finite array from an earlier frame is not this frame's measurement."""
+    widget = FakeWidget()
+    coordinator = ThreeDCoordinator(widget, SceneComposer(), ImmediateExecutor())
+    coordinator.set_visible(True)
+    coordinator.activate(scene, "live-A")
+    presented = len(widget.frames)
+
+    assert coordinator.submit(_frame(scene, frame_id=5, computed=2, value=6.0)) == "accepted"
+    coordinator.tick()
+    coordinator.tick()
+
+    assert not any(frame.valid.any() for frame in widget.frames[presented:]), (
+        "a stale array was painted as the current instant"
+    )
+    assert all(frame.identity.frame_id != 5 for frame in widget.frames[presented:])
+    assert coordinator.state == "stale"
+    assert coordinator.stale_frames >= 1
+    assert coordinator.last_diagnostic is not None
+    assert coordinator.last_diagnostic.code == "field_stale"
+    coordinator.close()
+
+
+def test_a_refused_field_expires_the_measurement_after_the_grace(scene):
+    clock = Clock()
+    widget = FakeWidget()
+    coordinator = ThreeDCoordinator(
+        widget, SceneComposer(), ImmediateExecutor(), clock=clock, stale_grace_s=1.0
+    )
+    coordinator.set_visible(True)
+    coordinator.activate(scene, "live-A")
+    coordinator.submit(_frame(scene, value=8.0))
+    coordinator.tick()
+    coordinator.tick()
+    assert coordinator.state == "live"
+    assert np.allclose(np.asarray(widget.frames[-1].scalar), 8.0)
+
+    # The producer cannot build a field any more: the picture stays for the grace, but
+    # the pane must stop calling it live immediately.
+    coordinator.note_field_failure(_stale_diagnostic())
+    assert coordinator.state == "stale"
+    coordinator.tick()
+    assert np.allclose(np.asarray(widget.frames[-1].scalar), 8.0), (
+        "a single refused frame blanked the model"
+    )
+    assert coordinator.state == "stale"
+
+    clock.advance(1.5)
+    coordinator.tick()
+    assert not widget.frames[-1].valid.any(), "the expired measurement is still on screen"
+    assert widget.frames[-1].scalar_descriptor.quantity == "unavailable"
+    assert coordinator.state == "no_data"
+    coordinator.close()
+
+
+def test_a_good_frame_after_a_failure_goes_back_to_live(scene):
+    clock = Clock()
+    widget = FakeWidget()
+    coordinator = ThreeDCoordinator(
+        widget, SceneComposer(), ImmediateExecutor(), clock=clock, stale_grace_s=5.0
+    )
+    coordinator.set_visible(True)
+    coordinator.activate(scene, "live-A")
+    coordinator.note_field_failure(_stale_diagnostic())
+    assert coordinator.state == "stale"
+
+    coordinator.submit(_frame(scene, frame_id=7, value=2.0))
+    coordinator.tick()
+    coordinator.tick()
+    assert coordinator.state == "live"
+    assert np.allclose(np.asarray(widget.frames[-1].scalar), 2.0)
+    coordinator.close()
+
+
+def test_a_failed_compose_does_not_leave_the_state_live(scene):
+    """R2: the error used to be recorded while the picture and "live" stayed."""
+    composer = FailingComposer()
+    widget = FakeWidget()
+    coordinator = ThreeDCoordinator(widget, composer, ImmediateExecutor())
+    coordinator.set_visible(True)
+    coordinator.activate(scene, "live-A")
+    coordinator.submit(_frame(scene, frame_id=1, value=1.0))
+    coordinator.tick()
+    coordinator.tick()
+    assert coordinator.state == "live"
+
+    composer.fail = True
+    coordinator.submit(_frame(scene, frame_id=2, value=2.0))
+    coordinator.tick()
+    coordinator.tick()
+    assert coordinator.state == "stale", "a failed compose kept reporting live"
+    assert coordinator.last_diagnostic is not None
+    assert coordinator.last_diagnostic.code == "field_unavailable"
+    assert "synthetic compose error" in coordinator.last_diagnostic.message
+    coordinator.close()
+
+
+def test_an_invalid_frame_starts_the_stale_grace_too(scene):
+    """A domain mismatch is "no measurement is arriving", not "everything is fine"."""
+    clock = Clock()
+    widget = FakeWidget()
+    coordinator = ThreeDCoordinator(
+        widget, SceneComposer(), ImmediateExecutor(), clock=clock, stale_grace_s=1.0
+    )
+    coordinator.set_visible(True)
+    coordinator.activate(scene, "live-A")
+    assert coordinator.submit(_frame(scene, domain="other-domain")) == "invalid"
+    assert coordinator.state == "stale"
+    assert coordinator.invalid_frames == 1
+    clock.advance(2.0)
+    coordinator.tick()
+    assert coordinator.state == "no_data"
+    coordinator.close()
@@ -381,7 +381,12 @@ def test_the_probe_reports_this_machine_truthfully():

 # ------------------------------------------------- stage 4: the assembled view
 class _Packet:
-    """The parts of an ``SdkPacket`` the finger adapter and the identity need."""
+    """The parts of an ``SdkPacket`` the finger adapter and the identity need.
+
+    The provenance flags are part of the contract now: a producer has to say that the
+    array it hands over belongs to this frame, and the adapter refuses one that does
+    not say (see ``adapters._field_validity``).
+    """

    def __init__(self, frame_id: int, shape_hw: tuple[int, int], value: float = 0.5) -> None:
        self.session_id = 1
@@ -389,6 +394,15 @@ class _Packet:
        self.mono_t = float(frame_id) / 30.0
        self.flow = np.full((*shape_hw, 2), float(value), dtype=np.float32)
        self.raw_frame = np.zeros((4, 4, 3), dtype=np.uint8)
+        self.valid = True
+        self.invalid_reason = ""
+        self.flow_valid = True
+        self.flow_frame_id = int(frame_id)
+        self.depth = None
+        self.depth_valid = None
+        self.depth_frame_id = None
+        self.forces_valid = True
+        self.valid_mask = None


@pytest.fixture(scope="module")
@@ -22,6 +22,7 @@ from __future__ import annotations

 import itertools
 import shutil
+import time
 from dataclasses import replace
 from pathlib import Path

@@ -610,6 +611,89 @@ def test_a_finger_scene_uses_the_finger_adapter(tmp_path, write_poc, qt_app):
        view.close()


+# ------------------------------------------- stage 6: a field that is not this frame's
+def _stale_scene_view(qt_app):
+    """A real view with the synthetic scene, or a skip when GL is not buildable."""
+    from orisys_studio.scene3d.view import Scene3DView
+
+    view = Scene3DView()
+    if not view.available:
+        view.close()
+        pytest.skip("no GL binding here")
+    if not view.use_scene(synthetic.poc_scene(), source_id="poc-live"):
+        view.close()
+        pytest.skip("the synthetic scene was refused")
+    view.set_enabled(True)
+    return view
+
+
+def test_a_packet_that_cannot_claim_its_field_marks_the_measurement_stale(qt_app):
+    """R2 through the view: the refusal used to be dropped on the floor.
+
+    ``snapshot()`` returning ``None`` left the pane claiming ``live`` about a picture
+    nothing was producing. Now the reason travels to the coordinator *and* back out
+    to the pane.
+    """
+    view = _stale_scene_view(qt_app)
+    try:
+        packet = ui_check._fake_packet(1)
+        packet.flow_valid = False
+        packet.invalid_reason = "compute: RuntimeError: boom"
+        view.submit_packet(packet)
+
+        coordinator = view.coordinator
+        assert coordinator is not None
+        assert coordinator.state == "stale", "a refused field kept reporting live"
+        assert coordinator.last_diagnostic is not None
+        assert coordinator.last_diagnostic.code == "field_stale"
+
+        view.tick()
+        assert view.last_diagnostic is not None
+        assert view.last_diagnostic.code == "field_stale", (
+            "the pane never learned why the measurement stopped"
+        )
+        assert view.panel is not None
+        assert view.panel.state_label.text() not in ("", "scene3d.state_stale"), (
+            "the stale state has no operator-facing sentence"
+        )
+    finally:
+        view.close()
+
+
+def test_an_unstated_field_is_refused_rather_than_assumed(qt_app):
+    view = _stale_scene_view(qt_app)
+    try:
+        packet = ui_check._fake_packet(1)
+        packet.flow_valid = None
+        view.submit_packet(packet)
+        coordinator = view.coordinator
+        assert coordinator is not None
+        assert coordinator.state == "stale"
+        assert coordinator.last_diagnostic is not None
+        assert coordinator.last_diagnostic.code == "field_unverified"
+    finally:
+        view.close()
+
+
+def test_a_fresh_field_reaches_the_screen_and_clears_the_stale_state(qt_app):
+    view = _stale_scene_view(qt_app)
+    try:
+        shape = tuple(view.assets.shape_hw)
+        packet = ui_check._fake_packet(1)
+        packet.flow = np.full((*shape, 2), 0.5, dtype=np.float32)
+        view.submit_packet(packet)
+        assert view._snapshot_diagnostic() is None, "a stated, fresh field was refused"
+        coordinator = view.coordinator
+        for _ in range(60):
+            view.tick()
+            if coordinator is not None and coordinator.state == "live":
+                break
+            time.sleep(0.01)
+        assert coordinator is not None and coordinator.state == "live", coordinator.state
+    finally:
+        view.close()
+
+
 # ------------------------------------------------------- stage 5: the record block
 def test_a_recording_says_what_the_3d_pane_was(
    make_window, next_record_root, write_poc, qt_app, monkeypatch
@@ -763,6 +847,9 @@ def test_every_run_gets_a_new_generation(make_window, qt_app):
            packet = ui_check._fake_packet(1)
            packet.flow = np.full((*shape, 2), 0.5, dtype=np.float32)
            view.submit_packet(packet)
+            assert view._snapshot_diagnostic() is None, (
+                "the adapter refused the second session's packet"
+            )
            assert view.coordinator.stale_frames == 0, (
                "the second session's frames were refused as stale"
            )
@@ -213,6 +213,15 @@ def _fake_packet(
        invalid_reason=reason,
        raw_available=True,
        depth=depth,
+        # Per-field provenance: a producer has to say whether the arrays it hands over
+        # are this frame's own (the 3D sink refuses an unstated field rather than
+        # colouring something it cannot claim). A failed frame is stale here, exactly
+        # as the workers mark it.
+        flow_valid=bool(valid),
+        flow_frame_id=frame_id if valid else None,
+        depth_valid=bool(valid) if depth is not None else None,
+        depth_frame_id=frame_id if (valid and depth is not None) else None,
+        forces_valid=bool(valid),
    )