commit 18e579c9f0ef4793e24cf84d988990fd51cf945e
Author: Eiichi Shimazu <studio@adaptdesign.jp>
Date:   Sat Nov 14 23:52:49 2026 +0900

    補正エピソードの 1 コマごとに実測か埋めた値かの印 state_source を残す (依頼 196)
    
    build_correction_episodes.py: gap_fill_flags で state_source (int8. 0 実測 / 1 動いた抜けを埋めた値 /
    2 動かなかった抜けを埋めた値) を作り episode.npz に足す. meta.json に filled_steps / filled_motion_steps /
    filled_ratio / state_source_codes. README に印の説明. check_compat はお手本にない state_source を
    キーの違いとして数えない. state / actions / t / t_orig は直す前とバイト単位で同じ (test_state_source.py).
    diag_fill_flags.py: 既存の補正データセットを /tmp で作り直して印を数え, diag/fill_flags_20261114/ に CSV.
    データセットは書き換えていない. 補正エピソード 95 本 (14 データセット)すべて state / actions / t が一致.
    
    Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pi05/build_correction_episodes.py |  39 +++++++++-
 pi05/diag_fill_flags.py           | 132 +++++++++++++++++++++++++++++++++
 pi05/test_state_source.py         | 152 ++++++++++++++++++++++++++++++++++++++
 3 files changed, 321 insertions(+), 2 deletions(-)

diff --git a/pi05/build_correction_episodes.py b/pi05/build_correction_episodes.py
index e5bce79..ae813b3 100644
--- a/pi05/build_correction_episodes.py
+++ b/pi05/build_correction_episodes.py
@@ -16,6 +16,7 @@
   - 使うのは 結果が「成功」(y) の run だけ
   - 時間軸は 10Hz. state は実測角の線形補間 + グリッパーは指令の時刻で 0/1 (build_demo_episodes と同じ)
     実測角の抜け (送信中) は --gap-fill で埋める (linear = corr_v1 / speed20 = corr_v2. 依頼 46)
+    埋めたコマには episode.npz の state_source で印を残す (0 実測 / 1 動いた抜け / 2 動かなかった抜け. 依頼 196)
   - 学習に使うのは状態が model / correct のステップだけ. rewind / home / wait (確認の入力待ち.
     人がブロックを置き直すと手が写る) / regrip (補正の R で開いてから閉じ終わるまで. 2026-10-30 依頼 35)
     のステップと,
@@ -176,6 +177,26 @@ def fill_motion_gaps(t, at, aq, mode="speed20", gap_sec=GAP_SEC, speed=SPEED20_D
     return q, gaps
 
 
+# ---- 1 コマごとの state の出どころ (state_source. int8. 2026-11-14 依頼 196) ----
+# 関節 6 列が実測角か, fill_motion_gaps が抜けを埋めた値かの印. 爪の列は指令なのでどれでも同じ
+SRC_MEASURED = 0     # 実測角 (0.25 秒以内の前後の実測角の線形補間. ふだんは 0.1 秒おき)
+SRC_FILLED_MOVE = 1  # 動いた抜け (前後の実測角の差 > GAP_MOVE_MIN_DEG) の中を埋めた値
+SRC_FILLED_STILL = 2  # 動かなかった抜け (差 <= GAP_MOVE_MIN_DEG. 爪の送信など) の中を埋めた値
+STATE_SOURCE_CODES = {SRC_MEASURED: "measured", SRC_FILLED_MOVE: "filled_motion_gap",
+                      SRC_FILLED_STILL: "filled_still_gap"}
+
+
+def gap_fill_flags(t, at, aq, gap_sec=GAP_SEC, min_move=GAP_MOVE_MIN_DEG):
+    """時刻 t のコマごとの state_source (int8). fill_motion_gaps と同じ「抜け」(実測角の間が gap_sec より長い
+    [ta, tb]) の中 (ta < t < tb) が埋めた値. 埋め方 (linear / speed20) によらず同じ印になる."""
+    src = np.zeros(len(t), dtype=np.int8)
+    for i in np.flatnonzero(np.diff(at) > gap_sec):
+        ta, tb = at[i], at[i + 1]
+        moved = float(np.max(np.abs(aq[i + 1] - aq[i]))) > min_move
+        src[(t > ta) & (t < tb)] = SRC_FILLED_MOVE if moved else SRC_FILLED_STILL
+    return src
+
+
 def logged_sends(log):
     """log.jsonl から, 送った目標とそのときの時刻の手がかりを集める.
 
@@ -443,9 +464,12 @@ def build_one(run_dir, out_dir, args):
         return {"run": stem, "used": False,
                 "reason": f"フレーム抽出不足 written={written} steps={len(tt)} (mp4 と frame_log の不一致)"}
 
+    state_source = gap_fill_flags(t, at, aq)[sel["idx"]]
     np.savez_compressed(out_dir / "episode.npz",
                         t=(np.arange(len(tt)) / FPS).astype(np.float64),
-                        t_orig=tt - tt[0], state=sel["state"], actions=sel["actions"])
+                        t_orig=tt - tt[0], state=sel["state"], actions=sel["actions"],
+                        state_source=state_source)
+    filled = int((state_source != SRC_MEASURED).sum())
     c = sel["counts"]
     meta = {
         "run_stem": stem, "task": bde.TASK, "fps": FPS, "steps": int(len(tt)),
@@ -475,6 +499,11 @@ def build_one(run_dir, out_dir, args):
         "gap_fill": {"mode": gap_fill, "gap_sec": GAP_SEC,
                      "speed_deg_per_sec": SPEED20_DEG_PER_SEC if gap_fill == "speed20" else None,
                      "gaps": len(gaps), **gap_match},
+        # 依頼 196: episode.npz の state_source (コマごとの印) の数
+        "filled_steps": filled,
+        "filled_motion_steps": int((state_source == SRC_FILLED_MOVE).sum()),
+        "filled_ratio": round(filled / len(tt), 4),
+        "state_source_codes": {str(k): v for k, v in STATE_SOURCE_CODES.items()},
         "max_step_deg": [round(float(v), 2) for v in
                          (np.abs(np.diff(sel["state"][:, :6], axis=0)).max(axis=0) if len(sel["state"]) > 1
                           else np.zeros(6))],
@@ -506,7 +535,8 @@ def check_compat(ours, ref):
     for ep in a:
         m = json.loads((ep / "meta.json").read_text())
         z = np.load(ep / "episode.npz")
-        if sorted(z.files) != sorted(rz.files):
+        # 依頼 196: state_source (コマごとの印) はお手本にないキー. 読む側は state / actions しか使わないので比べない
+        if sorted(set(z.files) - {"state_source"}) != sorted(set(rz.files) - {"state_source"}):
             probs.append(f"{ep.name}: npz のキー {sorted(z.files)} != お手本 {sorted(rz.files)}")
         for k in ("state", "actions"):
             if z[k].dtype != rz[k].dtype or z[k].shape[1:] != rz[k].shape[1:]:
@@ -554,6 +584,11 @@ def write_readme(ds, results, args):
         + ("前後の実測角をまっすぐ結ぶ (linear)" if args.gap_fill == "linear" else
            f"抜けの頭から速度 20 の速さ ({SPEED20_DEG_PER_SEC:.0f}°/秒. いちばん大きく動く関節) で前後の実測角を結ぶ"
            "直線を進み, 着いたら止まる (speed20)"),
+        "- 1 コマごとの state の出どころを episode.npz の `state_source` (int8. state と同じ長さ) に残す: "
+        "0 = 実測角 (0.25 秒以内の前後の実測角の線形補間) / 1 = 動いた抜け (前後の実測角の差が "
+        f"{GAP_MOVE_MIN_DEG}° を超える) の中を上の埋め方で埋めた値 / 2 = 動かなかった抜け (爪の送信など) の中を埋めた値。"
+        "印は関節 6 列のもの (爪の列はいつも指令)。actions の関節は次のコマの state なので, その出どころは次のコマの印。"
+        "meta.json の filled_steps (1 と 2 のコマの数) / filled_motion_steps (1 の数) / filled_ratio (全コマに対する 1 と 2 の割合)",
         "- 戻りきれなかった巻き戻しがある run は使わない (前後の姿勢が続かない)",
     ]
     if getattr(args, "max_protrude_px", None) is not None:
diff --git a/pi05/diag_fill_flags.py b/pi05/diag_fill_flags.py
new file mode 100644
index 0000000..4402ca4
--- /dev/null
+++ b/pi05/diag_fill_flags.py
@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+"""diag_fill_flags.py — 既存の補正データセットで, 実測角の抜けを埋めたコマの割合を測る (2026-11-14 依頼 196).
+
+datasets/ の名前に corr / deep_even10 が入るデータセットの, 補正エピソード (meta.json の source が pi05_policy)
+ごとに, 元の runs/pi05_policy/{run_stem}/ から build_correction_episodes.build_one を meta.json と同じ条件
+(gap_fill の mode・trim_mid_idle・trim_hold) で /tmp に作り直し, state_source (コマごとの印) を数える.
+作り直した state / actions / t がデータセットの episode.npz と一致しない本は数えない (理由を CSV に残す).
+
+データセット・元の記録は読むだけ (書かない). 書くのは /tmp と diag/fill_flags_20261114/ だけ.
+
+  python3 pi05/diag_fill_flags.py
+"""
+import argparse
+import csv
+import json
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+import numpy as np
+
+_HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(_HERE))
+
+import build_correction_episodes as bce  # noqa: E402
+import build_demo_episodes as bde  # noqa: E402
+
+OUT_NAME = "fill_flags_20261114"
+TMP_ROOT = Path("/tmp") / OUT_NAME
+# 学習に使ったことがあるデータセットを先に並べる (依頼 196)
+TRAINED_FIRST = ("mecharm_redblock_box_teach_v1_trim2_corr_v5", "mecharm_redblock_box_deep_even10_v1",
+                 "mecharm_redblock_box_teach_v1_trim2_corr_v5_deep")
+
+
+def target_datasets(ds_root):
+    names = sorted(p.name for p in ds_root.iterdir()
+                   if p.is_dir() and not p.name.startswith(".") and ("corr" in p.name or "deep_even10" in p.name))
+    return sorted(names, key=lambda n: (n not in TRAINED_FIRST, TRAINED_FIRST.index(n) if n in TRAINED_FIRST else 0, n))
+
+
+def rebuild(run_stem, meta, cache):
+    """meta.json と同じ条件で /tmp に作り直す. 戻り値: (npz の中身 dict / None, 理由)."""
+    gf = meta.get("gap_fill") or {}
+    key = (run_stem, gf.get("mode"), bool(meta.get("trim_mid_idle")), meta.get("trim_hold") is not None)
+    if key in cache:
+        return cache[key]
+    out = TMP_ROOT / "rebuild" / "_".join(str(k) for k in key) / run_stem
+    args = SimpleNamespace(gap_fill=gf.get("mode") or "linear", trim_hold=key[3],
+                           no_trim_mid_idle=not key[2], max_protrude_px=None, force=True)
+    res = bce.build_one(bce.RUNS_ROOT / run_stem, out, args)
+    if not res.get("used"):
+        cache[key] = (None, f"作り直せない: {res.get('reason')}")
+    else:
+        z = np.load(out / "episode.npz")
+        cache[key] = ({k: z[k] for k in z.files}, "ok")
+    return cache[key]
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--dataset", action="append", help="このデータセットだけ (複数可)")
+    args = ap.parse_args()
+    bde.ensure_ssd()
+    ds_root = bde.DATA_ROOT / "datasets"
+    out_dir = bde.DATA_ROOT / "diag" / OUT_NAME
+    if out_dir.exists():
+        sys.exit(f"[ABORT] 出力先がもうある: {out_dir}")
+    names = args.dataset or target_datasets(ds_root)
+    rows, cache = [], {}
+    for name in names:
+        for mp in sorted((ds_root / name / "raw_episodes").glob("[!.]*/meta.json")):
+            meta = json.loads(mp.read_text())
+            if not str(meta.get("source", "")).startswith("pi05_policy"):
+                continue
+            ep = mp.parent
+            stem = meta.get("run_stem", ep.name)
+            z = np.load(ep / "episode.npz")
+            row = {"dataset": name, "episode": ep.name, "run_stem": stem,
+                   "gap_fill": (meta.get("gap_fill") or {}).get("mode"),
+                   "trim_hold": meta.get("trim_hold") is not None, "steps": len(z["state"]),
+                   "filled_steps": "", "filled_motion_steps": "", "filled_ratio": "",
+                   "gaps": (meta.get("gap_fill") or {}).get("gaps"),
+                   "motion_gaps": (meta.get("gap_fill") or {}).get("motion_gaps"), "status": ""}
+            new, why = rebuild(stem, meta, cache)
+            if new is None:
+                row["status"] = why
+            elif not all(k in new and new[k].shape == z[k].shape and new[k].tobytes() == z[k].tobytes()
+                         for k in ("state", "actions", "t")):
+                row["status"] = "state が一致しない (数えない)"
+            else:
+                src = new["state_source"]
+                n_f = int((src != bce.SRC_MEASURED).sum())
+                row.update(filled_steps=n_f, filled_motion_steps=int((src == bce.SRC_FILLED_MOVE).sum()),
+                           filled_ratio=round(n_f / len(src), 4), status="ok")
+            rows.append(row)
+            print(f"{name}  {ep.name}  {row['status']}  {row['filled_steps']}/{row['steps']}")
+
+    if not rows:   # codex P2: 出力先を作る前に止める (空のまま作ると次に走らせられない)
+        sys.exit("[ABORT] 対象の補正エピソードがない")
+    out_dir.mkdir(parents=True)
+    with (out_dir / "episodes.csv").open("w", newline="") as f:
+        w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
+        w.writeheader(); w.writerows(rows)
+    summ = []
+    for name in names:
+        rs = [r for r in rows if r["dataset"] == name]
+        ok = [r for r in rs if r["status"] == "ok"]
+        steps = sum(r["steps"] for r in ok); filled = sum(r["filled_steps"] for r in ok)
+        top = max(ok, key=lambda r: r["filled_ratio"]) if ok else None
+        summ.append({"dataset": name, "corr_episodes": len(rs), "counted": len(ok),
+                     "not_counted": len(rs) - len(ok), "steps": steps, "filled_steps": filled,
+                     "filled_motion_steps": sum(r["filled_motion_steps"] for r in ok),
+                     "filled_ratio": round(filled / steps, 4) if steps else "",
+                     "max_episode": top["episode"] if top else "",
+                     "max_episode_ratio": top["filled_ratio"] if top else ""})
+    with (out_dir / "summary.csv").open("w", newline="") as f:
+        w = csv.DictWriter(f, fieldnames=list(summ[0].keys()))
+        w.writeheader(); w.writerows(summ)
+    (out_dir / "README.md").write_text(
+        "# fill_flags_20261114 (依頼 196)\n\n"
+        "既存の補正データセットで, 実測角の抜けを埋めたコマ (state_source が 1 または 2) の数と割合。\n"
+        "`python3 pi05/diag_fill_flags.py` で作った。データセットは読むだけ。作り直しは /tmp。\n\n"
+        "- state_source: 0 = 実測角 / 1 = 動いた抜けを埋めた値 / 2 = 動かなかった抜けを埋めた値\n"
+        "- filled_ratio = (1 と 2 のコマ) / 全コマ。filled_motion_steps は 1 のコマの数\n"
+        "- status が ok でない本 (作り直した state / actions / t がデータセットと一致しない・作り直せない) は数えない\n"
+        "- episodes.csv: 本ごと / summary.csv: データセットごと (max_episode = 割合がいちばん多い本)\n")
+    print(f"→ {out_dir}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/pi05/test_state_source.py b/pi05/test_state_source.py
new file mode 100644
index 0000000..cb6b6ae
--- /dev/null
+++ b/pi05/test_state_source.py
@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+"""test_state_source.py — 1 コマごとの state の出どころ (state_source. 依頼 196) の確かめ.
+
+  python3 pi05/test_state_source.py -v
+
+前半はファイルを読まない部分だけ. 後半 (TestSameAsBefore) は KIOXIA の runs/pi05_policy を読み,
+直す前のコード (git の HEAD) と今のコードで /tmp に作り, state / actions / t がバイト単位で同じことを見る
+(KIOXIA がなければ飛ばす. データセットにもアームにも触らない).
+"""
+import importlib.util
+import json
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+import numpy as np
+
+_HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(_HERE))
+
+import build_correction_episodes as bce  # noqa: E402
+
+
+def samples(still=False):
+    """0〜1 秒は 0.1 秒おき → 1.0〜2.6 秒は抜け (j1 が 6° 動く. still なら動かない) → 2.6〜3.6 秒は 0.1 秒おき."""
+    a = np.round(np.arange(0.0, 1.01, 0.1), 3)
+    b = np.round(np.arange(2.6, 3.61, 0.1), 3)
+    at = np.concatenate([a, b])
+    aq = np.zeros((len(at), 6))
+    if not still:
+        aq[len(a):, 1] = 6.0
+    return at, aq
+
+
+def runs_of_nonzero(src):
+    nz = np.r_[0, (np.asarray(src) != 0).astype(int)]
+    return int((np.diff(nz) == 1).sum())
+
+
+class TestFlags(unittest.TestCase):
+    def test_motion_gap_is_1(self):
+        at, aq = samples()
+        t = np.arange(0.0, 3.6, 0.1)
+        src = bce.gap_fill_flags(t, at, aq)
+        self.assertEqual(src.dtype, np.int8)
+        inside = (t > 1.0) & (t < 2.6)
+        self.assertTrue(np.all(src[inside] == bce.SRC_FILLED_MOVE))
+        self.assertTrue(np.all(src[~inside] == bce.SRC_MEASURED))
+
+    def test_still_gap_is_2(self):
+        at, aq = samples(still=True)
+        t = np.arange(0.0, 3.6, 0.1)
+        src = bce.gap_fill_flags(t, at, aq)
+        self.assertEqual(set(np.unique(src)), {bce.SRC_MEASURED, bce.SRC_FILLED_STILL})
+
+    def test_flags_mark_exactly_what_speed20_changes(self):
+        """speed20 が線形補間から変えたコマは, すべて印が 0 でないコマ."""
+        at, aq = samples()
+        t = np.arange(0.0, 3.6, 0.1)
+        q_lin, _ = bce.fill_motion_gaps(t, at, aq, mode="linear")
+        q_sp, _ = bce.fill_motion_gaps(t, at, aq, mode="speed20")
+        src = bce.gap_fill_flags(t, at, aq)
+        changed = np.any(np.abs(q_lin - q_sp) > 1e-9, axis=1)
+        self.assertTrue(changed.any())
+        self.assertTrue(np.all(src[changed] != bce.SRC_MEASURED))
+
+    def test_no_gap_no_flag(self):
+        at = np.round(np.arange(0.0, 3.01, 0.1), 3)
+        aq = np.cumsum(np.ones((len(at), 6)), axis=0)
+        self.assertEqual(int(bce.gap_fill_flags(np.arange(0.0, 3.0, 0.1), at, aq).sum()), 0)
+
+    def test_compat_ignores_state_source(self):
+        """お手本にない state_source のキーがあっても check_compat はキーの違いとして数えない."""
+        with tempfile.TemporaryDirectory() as d:
+            import cv2
+            for name, extra in (("ours", True), ("ref", False)):
+                ep = Path(d) / name / "e0"
+                (ep / "frames").mkdir(parents=True)
+                cv2.imwrite(str(ep / "frames" / "000000.jpg"), np.zeros((4, 6, 3), np.uint8))
+                kw = {"state_source": np.zeros(1, np.int8)} if extra else {}
+                np.savez_compressed(ep / "episode.npz", t=np.zeros(1), t_orig=np.zeros(1),
+                                    state=np.zeros((1, 7), np.float32), actions=np.zeros((1, 7), np.float32), **kw)
+                (ep / "meta.json").write_text(json.dumps({"fps": 10, "image_size": [6, 4], "state_dim": 7,
+                                                          "state_names": [], "task": "x"}))
+            self.assertEqual(bce.check_compat(Path(d) / "ours", Path(d) / "ref"), [])
+
+
+def _old_module():
+    """直す前の build_correction_episodes.py (git の HEAD) を別の名前で読み込む."""
+    src = subprocess.run(["git", "show", "HEAD:pi05/build_correction_episodes.py"], cwd=_HERE,
+                         capture_output=True, text=True, check=True).stdout
+    tmp = Path(tempfile.mkdtemp()) / "bce_old.py"
+    tmp.write_text(src)
+    spec = importlib.util.spec_from_file_location("bce_old", tmp)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+# 補正の多い本・巻き戻しのある本など. 依頼 196 の時点で KIOXIA にあるもの
+SAMPLE_RUNS = ("pi05_20261031_175527", "pi05_20261031_180435")
+
+
+@unittest.skipUnless(Path("/Volumes/KIOXIA").is_mount() and bce.RUNS_ROOT.is_dir(), "KIOXIA がない")
+class TestSameAsBefore(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.old = _old_module()
+
+    def _build(self, mod, run, d, gap_fill, trim_hold):
+        args = SimpleNamespace(gap_fill=gap_fill, trim_hold=trim_hold, no_trim_mid_idle=False,
+                               max_protrude_px=None, force=True)
+        out = Path(d) / mod.__name__ / f"{gap_fill}_{trim_hold}" / run
+        res = mod.build_one(bce.RUNS_ROOT / run, out, args)
+        self.assertTrue(res.get("used"), res)
+        z = np.load(out / "episode.npz")
+        return {k: z[k] for k in z.files}, json.loads((out / "meta.json").read_text())
+
+    def test_bytes_and_counts(self):
+        with tempfile.TemporaryDirectory() as d:
+            for run in SAMPLE_RUNS:
+                if not (bce.RUNS_ROOT / run).is_dir():
+                    self.skipTest(f"{run} がない")
+                for gap_fill, trim_hold in (("linear", False), ("speed20", False), ("speed20", True)):
+                    with self.subTest(run=run, gap_fill=gap_fill, trim_hold=trim_hold):
+                        a, _ = self._build(self.old, run, d, gap_fill, trim_hold)
+                        b, meta = self._build(bce, run, d, gap_fill, trim_hold)
+                        self.assertNotIn("state_source", a)
+                        for k in ("t", "t_orig", "state", "actions"):
+                            self.assertEqual(a[k].dtype, b[k].dtype, k)
+                            self.assertEqual(a[k].tobytes(), b[k].tobytes(), k)
+                        src = b["state_source"]
+                        self.assertEqual(src.dtype, np.int8)
+                        self.assertEqual(len(src), len(b["state"]))
+                        self.assertTrue(set(np.unique(src)) <= set(bce.STATE_SOURCE_CODES))
+                        gf = meta["gap_fill"]
+                        filled = int((src != 0).sum())
+                        self.assertEqual(meta["filled_steps"], filled)
+                        self.assertEqual(meta["filled_motion_steps"], int((src == 1).sum()))
+                        self.assertAlmostEqual(meta["filled_ratio"], filled / len(src), places=4)
+                        # 印のつながりの数は抜けの数を超えない (1 つの抜けが印を 2 つに分けることはない)
+                        self.assertLessEqual(runs_of_nonzero(src), gf["gaps"])
+                        self.assertLessEqual(runs_of_nonzero(src == 1), gf["motion_gaps"])
+                        if gf["gaps"] == 0:
+                            self.assertEqual(filled, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()
