feat: ForceStreamOutput 新增同幀未濾波量測欄位

App 需要濾波前的 raw 與 N 供錄製、分析與標定取樣；診斷幀保留 raw 特徵、N 為 NaN。已以帶標定、中值加低通的會話逐幀對照直通結果驗證。

docs/force-stream-session.zh.md

@@ -17,6 +17,9 @@
   为指定参考版本确认无载归零；可选窗口上限。
 - `process(measurement) -> ForceStreamOutput`：消费一个 T05 `ImageMeasurement`，返回逐帧结果
   （`frame` 携带 T03 帧头与 `reference_generation`，以及 `raw/newton/status/quality/zero`）。
+  `measurement_raw`/`measurement_newton` 是同帧、**未经输出滤波**的测量三元组（T09c）：接受帧取
+  `ForceOutput.measurement` 的未滤波 raw/N；诊断帧的 `measurement_raw` 等于该帧 `force_raw`、
+  `measurement_newton` 为 NaN。直通设置下它们与 `raw`/`newton` 逐位相同。
 
 ## 规则
 

src/orisys/force_stream_session.py

@@ -93,6 +93,9 @@ class ForceStreamOutput:
     warmup_required_frames: int = 0
     warmup_elapsed_frames: int = 0
     warming_up: bool = False
+    #: T09c: the same frame's unfiltered measurement, before the output chain.
+    measurement_raw: tuple[float, float, float] = (float("nan"),) * 3
+    measurement_newton: tuple[float, float, float] = (float("nan"),) * 3
 
     @property
     def reference_generation(self) -> int:
@@ -359,6 +362,8 @@ class ForceStreamSession:
         accepted: bool = True,
         duplicate: bool = False,
         late: bool = False,
+        measurement_raw: tuple[float, float, float] | None = None,
+        measurement_newton: tuple[float, float, float] | None = None,
     ) -> ForceStreamOutput:
         output = ForceStreamOutput(
             frame=frame,
@@ -381,6 +386,16 @@ class ForceStreamSession:
             warmup_required_frames=self._filter.warmup_required_frames,
             warmup_elapsed_frames=self._filter.warmup_elapsed_frames,
             warming_up=self._filter.warming_up,
+            measurement_raw=(
+                (float("nan"),) * 3
+                if measurement_raw is None
+                else tuple(float(value) for value in measurement_raw)
+            ),
+            measurement_newton=(
+                (float("nan"),) * 3
+                if measurement_newton is None
+                else tuple(float(value) for value in measurement_newton)
+            ),
         )
         self._last_output = output
         return output
@@ -409,6 +424,8 @@ class ForceStreamSession:
             accepted=False,
             duplicate=duplicate,
             late=late,
+            measurement_raw=raw,
+            measurement_newton=(float("nan"),) * 3,
         )
 
     @staticmethod
@@ -652,4 +669,8 @@ class ForceStreamSession:
             quality=quality,
             diagnostics=tuple(output.measurement.diagnostics),
             accepted=True,
+            measurement_raw=tuple(float(value) for value in output.measurement.sdk.raw),
+            measurement_newton=tuple(
+                float(value) for value in output.measurement.newton
+            ),
         )

tests/test_force_stream_session.py

@@ -669,3 +669,84 @@ def test_fn_only_model_uncalibrated_session_stays_explicit() -> None:
     assert all(np.isnan(value) for value in results[-1].newton)
     assert set(results[-1].newton_status) == {"uncalibrated"}
     assert results[-1].quality == "uncalibrated"
+
+
+def test_unfiltered_measurement_fields_survive_the_output_filter() -> None:
+    inputs = [100.0, 100.0, 100.0, 900.0, 100.0, 120.0, 140.0, 160.0]
+    filtered = _new_chain(settings=helpers.legacy_settings(median_window=5))
+    passthrough = _new_chain(settings=helpers.legacy_settings(median_window=1))
+    for frame_id, value in enumerate(inputs, start=1):
+        one = filtered.process(_measurement(frame_id, (value, 0.0, 0.0)))
+        two = passthrough.process(_measurement(frame_id, (value, 0.0, 0.0)))
+        assert one.measurement_raw == (value, 0.0, 0.0)
+        assert two.raw == (value, 0.0, 0.0)
+        assert two.measurement_raw == one.measurement_raw
+        assert one.measurement_newton == two.measurement_newton
+    assert filtered.process(_measurement(9, (900.0, 0.0, 0.0))).raw[0] != 900.0
+    assert passthrough.process(_measurement(9, (900.0, 0.0, 0.0))).raw[0] == 900.0
+
+
+def test_diagnostic_rows_keep_the_unfiltered_measurement() -> None:
+    session = _new_chain(settings=helpers.legacy_settings(median_window=5))
+    session.process(_measurement(20, _raw_for(20)))
+    duplicate = session.process(_measurement(20, _raw_for(20)))
+    assert duplicate.quality == "duplicate"
+    assert duplicate.measurement_raw == tuple(float(value) for value in _raw_for(20))
+    assert duplicate.raw == duplicate.measurement_raw
+    assert all(np.isnan(value) for value in duplicate.measurement_newton)
+    assert all(np.isnan(value) for value in duplicate.newton)
+
+
+def test_calibrated_measurement_newton_is_unfiltered() -> None:
+    filt = _new_chain(
+        helpers.box_cal(),
+        helpers.legacy_settings(
+            median_window=5, lowpass_hz=3.0, lowpass_order=2, sample_rate_hz=30.0
+        ),
+    )
+    passthrough = _new_chain(
+        helpers.box_cal(), helpers.legacy_settings(median_window=1)
+    )
+    for session in (filt, passthrough):
+        session.request_zero(
+            reference_id="r",
+            declared_unloaded=True,
+            expected_reference_generation=0,
+        )
+    rows = []
+    pass_rows = []
+    for frame_id in range(1, 80):
+        raw = list(_raw_for(1))
+        if frame_id == 60:
+            raw[0] += 20000.0
+        rows.append(filt.process(_measurement(frame_id, tuple(raw))))
+        pass_rows.append(passthrough.process(_measurement(frame_id, tuple(raw))))
+    for filtered, plain in zip(rows, pass_rows):
+        for measured, reference in zip(
+            filtered.measurement_newton, plain.newton
+        ):
+            assert (np.isnan(measured) and np.isnan(reference)) or measured == reference
+        assert filtered.measurement_raw == plain.raw
+    spike = rows[59]
+    assert spike.newton[0] != spike.measurement_newton[0]
+
+
+def test_late_frame_carries_unfiltered_measurement_fields() -> None:
+    session = _new_chain(
+        helpers.box_cal(),
+        helpers.legacy_settings(
+            median_window=5, lowpass_hz=3.0, lowpass_order=2, sample_rate_hz=30.0
+        ),
+    )
+    session.request_zero(
+        reference_id="r", declared_unloaded=True, expected_reference_generation=0
+    )
+    for frame_id in range(1, 32):
+        session.process(_measurement(frame_id, _raw_for(frame_id)))
+    raw = _raw_for(32)
+    late = session.process(
+        _measurement(32, raw, capture_time_ns=31 * FRAME_NS)
+    )
+    assert late.late is True
+    assert late.measurement_raw == tuple(float(value) for value in raw)
+    assert all(np.isnan(value) for value in late.measurement_newton)