mechArm 270 Pi 受け渡し動作の成立を探る Log#4 | E-4からE-6まで。台帳・撮影・53件のテストで制御コードを閉じる
受け渡し動作の制御コードには、実装ブロックが D-1 から G まで並んでいる。前回（Log#3）までに E-3、つまり PRESENT → WAIT → RELEASE → 退避 → HOME のシーケンス本体までが完成していた。残りは E-4（試行の記録層）、E-5（撮影と VLM 判定）、E-6（検証テスト一式）。この3ブロックを閉じれば、実機の前に書くコードはなくなる。
その作業の開始点は、少し変な形をしていた。引き継ぎ書の最優先事項が「E-4 は未実装の疑い。実装した AI に差し戻せ」だったからだ。
「E-4 未実装」の引き継ぎ判定が、grep 一発で覆った
前セッションの引き継ぎ書にはこう書いてあった。E-4（台帳追記 + emergency journal + recovery）を発注し、Claude Code は「完了」と報告した。だが検証すると handover_journal.py というファイルが存在せず、atomic write（書きかけのファイルが残らない書き込み方式）のコードも本体に見当たらない——報告と実物が食い違う3度目のパターンだ、と。
指示どおり、最初に実物を確認した。
$ cd ~/r2r2r-scripts
$ grep -rn "emergency\|journal\|atomic\|fsync\|os.replace\|/tmp/handover\|recover" dynamic_tracking_limits/place/place_modules/run_pick_handover.py
dynamic_tracking_limits/place/place_modules/run_pick_handover.py:11:orchestration は E-2、handover sequence 本体は E-3、台帳追記 / emergency
dynamic_tracking_limits/place/place_modules/run_pick_handover.py:12:journal は E-4、VLM / 撮影は E-5、モックテストは E-6 で足す。
dynamic_tracking_limits/place/place_modules/run_pick_handover.py:108: persist_or_journal,
dynamic_tracking_limits/place/place_modules/run_pick_handover.py:109: recover_journals,
dynamic_tracking_limits/place/place_modules/run_pick_handover.py:436: physical safety has priority and a later write failure goes to the journal.
dynamic_tracking_limits/place/place_modules/run_pick_handover.py:1226: """Finalize annotations, run JSON and one ledger append (or journal)."""
dynamic_tracking_limits/place/place_modules/run_pick_handover.py:1270: result = persist_or_journal(payload)
dynamic_tracking_limits/place/place_modules/run_pick_handover.py:1468: "--recover",
dynamic_tracking_limits/place/place_modules/run_pick_handover.py:1484: recovered, skipped, failed = recover_journals(args.recover or None)108行目と109行目。persist_or_journal と recover_journals を、本体が別モジュールから import している。E-4 の機能は呼ばれている。呼ばれている以上、定義がどこかに実在するはずだ。定義元をファイル名の当て推量でなく grep で確定した。
$ f=$(grep -rl "def persist_or_journal" dynamic_tracking_limits/place/place_modules/)
$ echo "DEFINER: $f"
DEFINER: dynamic_tracking_limits/place/place_modules/handover_recording.py
$ grep -n "def persist_or_journal\|def recover_journals\|/tmp/handover\|os.replace\|fsync" "$f"
39:EMERGENCY_JOURNAL_DIR = Path("/tmp/handover_emergency_journal")
269: f.flush()
270: os.fsync(f.fileno())
271: os.replace(tmp_name, path)
274: os.fsync(dir_fd)
324:def persist_or_journal(payload: dict) -> str:
341:def recover_journals(run_stem: Optional[str] = None) -> tuple:E-4 は最初から実装されていた。ファイル名が handover_journal.py ではなく handover_recording.py だっただけだ。前セッションの検証は、存在するはずのファイル名を先に決めて探し、「無い」と結論していた。報告と実装の食い違いではなく、検証側の当て推量が原因だった。
差し戻しは不要になった。代わりに、この実装が仕様（data_manifest_rules.md 9節）を満たしているかのレビューへ進んだ。
handover_recording.py の中身 — 44列台帳・/tmp journal・recovery の三点セット
受け渡しの試行記録は3層でできている。正本は KIOXIA SSD 上の台帳 handover_ledger_v1.csv（44列・1試行1行・追記のみ）。SSD が試行の途中で外れた場合は、記録より物理安全を優先してシーケンスを完了させ、そのあと試行データを /tmp/handover_emergency_journal/ へ退避する。/tmp は正本の代わりではなく、復旧用の痕跡。SSD が戻ったら --recover で journal を読み、台帳に1回だけ追記する（既存行は絶対に書き換えない）。
書き込みはすべて atomic write で行う。一時ファイルに書いて flush と fsync（OS のバッファをディスクへ確実に落とす命令）を通してから、os.replace で本来の名前に差し替える。この順序なら、途中で電源が落ちても書きかけの中途半端なファイルが正規の名前で残ることはない。
codex レビュー2ラウンドを通過した確定版を全量で載せる。
#!/usr/bin/env python3
"""Handover E-4 recording, emergency journal, and recovery.
The KIOXIA copy is canonical. Files under ``EMERGENCY_JOURNAL_DIR`` are only
recovery traces; after the SSD is restored, call ``recover_journals()`` (or
``run_pick_handover.py --recover``). Recovery checks ``run_stem`` while holding
the ledger lock and appends a missing row once; it never updates an existing row.
Operator recovery procedure::
# 1. Restore/mount KIOXIA and verify the canonical ledger is present.
# 2. Recover all finalized journals (``*.tmp`` is deliberately ignored):
python3 run_pick_handover.py --recover
# Or recover exactly one trace:
python3 run_pick_handover.py --recover RUN_STEM
The canonical ``handover_parameters.json`` is created only when absent. Journal
files are retained as traces; recovery does not move, overwrite, or delete them.
"""
from __future__ import annotations
import csv
import fcntl
import json
import os
import re
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Optional
from handover_ledger import (
HANDOVER_LEDGER_COLS,
KIOXIA_MOUNT,
LEDGER_PATH,
ensure_ssd_mounted,
)
RUNS_ROOT = KIOXIA_MOUNT / "r2r2r-data" / "runs" / "run_handover"
EMERGENCY_JOURNAL_DIR = Path("/tmp/handover_emergency_journal")
PARAMETERS_JSON_NAME = "handover_parameters.json"
OUTCOMES = ("taken", "drop", "stuck", "unknown", "not_applicable")
RELEASE_OBSERVED = ("open", "not_open", "unknown")
RELEASE_SOURCES = ("operator", "top_camera", "side_camera")
OUTCOME_SOURCES_E4 = ("operator_visual", "unknown")
# E-4 Mod 1 (❌ A4/P1): run_stem 検証正規表現。make_run_stem() (L108-110) が生成する
# "run_handover_{YYYYMMDD}_{HHMMSS}_{uuuuuu}_{NNN}" (strftime("%Y%m%d_%H%M%S_%f")
# + "_{:03d}".format(trial_no)) と一致。CLI 値と journal payload の run_stem の
# 両方に適用し、journal ファイル名との整合も検査する (recover_journals 内で使用)。
RUN_STEM_RE = re.compile(r"^run_handover_\d{8}_\d{6}_\d{6}_\d{3}$")
def validate_run_stem(value) -> str:
"""run_stem が make_run_stem() 形式に合うことを保証。不正なら ValueError。
recovery 経路で信頼できない外部入力 (CLI --recover 値・journal payload) を
ledger append 前段で早期に弾くために使う (E-4 Mod 1・❌ A4/P1)。"""
if not isinstance(value, str) or not RUN_STEM_RE.match(value):
raise ValueError(
"invalid run_stem: {!r} (expected pattern {})".format(
value, RUN_STEM_RE.pattern))
return value
class AppendAborted(RuntimeError):
"""Ledger append was refused without modifying the ledger."""
class RecordLost(RuntimeError):
"""Emergency journal write also failed. Trial record is unrecoverable.
E-4 Mod 3 (⚠️ C2/D5) 反映: 台帳追記・run JSON 書込・journal 書込のいずれも
失敗した場合、記録は失われる。呼び出し側は [JOURNAL-FAIL] を stderr に
出した上で、常駐ループを制御された形で終了 (非0 exit code) すべき。"""
class _InputUnavailable(Exception):
pass
def _prompt_choice(prompt: str, choices: tuple, input_fn=None) -> str:
input_fn = input if input_fn is None else input_fn
while True:
try:
value = input_fn("{} [{}]: ".format(prompt, "/".join(choices))).strip().lower()
except (EOFError, KeyboardInterrupt) as e:
print("\n[ANNOTATION] input unavailable/interrupted -> remaining fields unknown")
raise _InputUnavailable() from e
if value in choices:
return value
print("[ANNOTATION] invalid value {!r}".format(value))
def collect_operator_annotations(log_record: dict, input_fn=None) -> dict:
"""Collect mutually exclusive E-4 fields without ever dropping recording."""
if log_record.get("trial_status") == "pick_failed":
annotations = {
"outcome": "not_applicable",
"outcome_source": "unknown",
"release_observed": "unknown",
"release_observed_source": "operator",
}
else:
try:
outcome = _prompt_choice("outcome", OUTCOMES[:-1], input_fn)
release_observed = _prompt_choice(
"release_observed", RELEASE_OBSERVED, input_fn)
release_source = (
"operator" if release_observed == "unknown" else
_prompt_choice("release_observed_source", RELEASE_SOURCES, input_fn)
)
outcome_source = (
"unknown" if outcome == "unknown" else
_prompt_choice("outcome_source", OUTCOME_SOURCES_E4, input_fn)
)
annotations = {
"outcome": outcome,
"outcome_source": outcome_source,
"release_observed": release_observed,
"release_observed_source": release_source,
}
except _InputUnavailable:
annotations = {
"outcome": "unknown",
"outcome_source": "unknown",
"release_observed": "unknown",
"release_observed_source": "operator",
}
log_record["operator_annotations"] = annotations
return annotations
def make_run_stem(trial_no: int, now: Optional[datetime] = None) -> str:
now = now or datetime.now()
return "run_handover_{}_{:03d}".format(now.strftime("%Y%m%d_%H%M%S_%f"), trial_no)
def _wall_iso(value) -> str:
if not isinstance(value, (int, float)):
return ""
return datetime.fromtimestamp(float(value)).astimezone().isoformat()
def _bool_or_blank(value):
return "" if value is None else bool(value)
def _mm_or_blank(values, index):
if not isinstance(values, (list, tuple)) or len(values) <= index:
return ""
value = values[index]
return value * 1000.0 if isinstance(value, (int, float)) else ""
def _find_sequence(handover: dict, label: str, occurrence: int = 0) -> dict:
found = [x for x in handover.get("sequence", []) if x.get("label") == label]
return found[occurrence] if len(found) > occurrence else {}
def _pick_lookup(log_record: dict) -> dict:
for item in reversed(log_record.get("lookups", [])):
if item.get("phase") == "pick" or item.get("label") in ("handover_pick", "stage4_pick"):
return item
return {}
def _abort_parts(log_record: dict) -> tuple:
abort = log_record.get("abort") or {}
return log_record.get("abort_stage", abort.get("stage", "")), abort.get("reason", "")
def build_ledger_row(log_record: dict, pose_data: dict, plan: dict) -> dict:
"""Map one completed/aborted trial to all HANDOVER_LEDGER_COLS."""
hv = log_record.get("handover") or {}
wait = hv.get("wait") or {}
release = hv.get("release") or {}
capture = hv.get("capture") or {}
annotations = log_record.get("operator_annotations") or {}
corr = (log_record.get("pick_correction") or {}).get("corr_meta") or {}
block_xz = (log_record.get("pick_correction") or {}).get("block_xz")
pick_lookup = _pick_lookup(log_record)
abort_stage, abort_reason = _abort_parts(log_record)
present = _find_sequence(hv, "PRESENT")
retract = _find_sequence(hv, "RETRACT_SMALL")
pmid_retreat = _find_sequence(hv, "PRESENT_MID", 1)
home = _find_sequence(hv, "HOME")
judgments = log_record.get("judgments") or {}
pick_judgment = judgments.get("pick") or {}
vlm = hv.get("vlm") or {}
row = {name: "" for name in HANDOVER_LEDGER_COLS}
row.update({
"timestamp": log_record.get("ended_at") or datetime.now().astimezone().isoformat(),
"run_stem": log_record["run_stem"],
"trial_no": log_record.get("trial_index", ""),
"trial_status": log_record.get("trial_status", "unexpected_error"),
"manual_recovery_required": bool(log_record.get("manual_recovery_required", False)),
"abort_stage": abort_stage,
"abort_reason": abort_reason,
"pick_verdict": pick_judgment.get("verdict", ""),
"pick_reason": pick_judgment.get("reason", ""),
"pick_confidence": pick_judgment.get("confidence", ""),
"block_pick_x": _mm_or_blank(block_xz, 0),
"block_pick_z": _mm_or_blank(block_xz, 1),
# DAY277 removed PICK visual servo; these columns intentionally remain blank.
"align_j0_final": "",
"align_converged": "",
"pick_pose_id": pick_lookup.get("pose_id", ""),
"corr_cell_key": corr.get("cell_key", ""),
"corr_applied_dx": corr.get("dx_mm", ""),
"corr_applied_dz": corr.get("dz_mm", ""),
"handover_pose_id": ((pose_data.get("poses") or {}).get("PRESENT") or {}).get("pose_id", ""),
"object_profile_id": (log_record.get("object_profile") or {}).get("id", ""),
"wait_plan_id": plan.get("wait_plan_id", ""),
"wait_plan_index": plan.get("wait_plan_index", ""),
"shuffle_seed": plan.get("shuffle_seed", ""),
"wait_sec_requested": wait.get("wait_sec_requested", plan.get("wait_sec_requested", "")),
"wait_sec_actual": wait.get("wait_sec_actual", ""),
"wait_started_at": _wall_iso(wait.get("wait_started_wall")),
"release_sent_at": _wall_iso(release.get("release_sent_at")),
"release_responded_at": _wall_iso(release.get("release_responded_at")),
"release_command_ok": _bool_or_blank(release.get("release_command_ok")),
"release_observed": annotations.get("release_observed", "unknown"),
"release_observed_source": annotations.get("release_observed_source", "operator"),
"present_reached_bool": bool(present.get("arrived", False)),
"retreat_completed_bool": bool(retract.get("arrived", False) and pmid_retreat.get("arrived", False)),
"home_reached_bool": bool(home.get("arrived", False)),
"hand_clear_confirmed": bool(hv.get("hand_clear_confirmed", False)),
"outcome": annotations.get("outcome", "unknown"),
"outcome_source": annotations.get("outcome_source", "unknown"),
"vlm_taken_verdict": vlm.get("verdict", ""),
"vlm_status": vlm.get("status", ""),
"vlm_reason": vlm.get("reason", ""),
"top_capture_status": (capture.get("top") or {}).get("status", ""),
"side_capture_status": (capture.get("side") or {}).get("status", ""),
"top_capture_path": (capture.get("top") or {}).get("path", ""),
"side_capture_path": (capture.get("side") or {}).get("path", ""),
})
if list(row) != HANDOVER_LEDGER_COLS:
raise AssertionError("ledger mapping is not in HANDOVER_LEDGER_COLS order")
return row
def build_run_payload(log_record: dict, pose_data: dict, plan: dict, row: dict) -> dict:
hv = log_record.get("handover") or {}
return {
"schema_version": "handover_run_v1",
"run_stem": log_record["run_stem"],
"ledger_row": row,
"pose_ids": {
label: ((pose_data.get("poses") or {}).get(label) or {}).get("pose_id", "")
for label in ("HOME", "PRESENT_MID", "PRESENT", "RETRACT_SMALL")
},
"wait_plan": dict(plan),
"object_profile": log_record.get("object_profile") or {
"id": "", "weight_g": None, "dimensions_mm": None,
"status": "not_measured_e4",
},
"timing": {
"started_at": log_record.get("started_at"),
"ended_at": log_record.get("ended_at"),
"wait": hv.get("wait"),
"release": hv.get("release"),
},
"actual_angles": [
{"label": x.get("label"), "target_angles": x.get("target_angles"),
"max_axis_diff_deg": x.get("max_axis_diff_deg"),
"last_actual": x.get("last_actual"),
"ts_send_wall": x.get("ts_send_wall"),
"ts_arrival_wall": x.get("ts_arrival_wall"),
"arrived": x.get("arrived")}
for x in hv.get("sequence", [])
],
"interlocks": {
"interlock1": hv.get("interlock1"),
"interlock2": hv.get("interlock2"),
"hand_clear_confirmed": hv.get("hand_clear_confirmed", False),
},
"sigint": {
"release_interrupted_after_send": hv.get("release_interrupted_after_send", False),
"install_error": (hv.get("release") or {}).get("sigint_install_error"),
},
"log_record": log_record,
}
def _atomic_json(path: Path, payload: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_name = tempfile.mkstemp(prefix=".{}.".format(path.name), suffix=".tmp", dir=str(path.parent))
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(payload, f, ensure_ascii=False, indent=2, sort_keys=True)
f.write("\n")
f.flush()
os.fsync(f.fileno())
os.replace(tmp_name, path)
dir_fd = os.open(str(path.parent), os.O_RDONLY)
try:
os.fsync(dir_fd)
finally:
os.close(dir_fd)
except BaseException:
try:
os.unlink(tmp_name)
except FileNotFoundError:
pass
raise
def write_emergency_journal(payload: dict, reason: str) -> Path:
"""journal (recovery trace) を書く。KIOXIA 未マウント時のみ呼ばれる
(persist_or_journal からの fallback 経路)。
E-4 Mod 4 (C1) + R2-2: trial_status は物理結果のまま変更しない
(completed 等)。mount lost の事実は journal.storage_status フィールドで
別途表現する。値は「いつ外れたか」でなく「いつ検出したか」を表す語彙
(mount guard を置いていない区間の断定を避ける):
- completed → "mount_lost_detected_after_physical_sequence"
- その他 (interrupted / motion_failed 等) → "mount_lost_detected_mid_sequence:{trial_status}"
台帳44列は不変。C1 の状態語原則: 物理結果と storage 事情を混ぜない。
E-4 Mod 3 (⚠️ C2/D5): _atomic_json の失敗を握らず落とす。
[JOURNAL-FAIL] を stderr に明示し、RecordLost を raise する。呼び出し側
(常駐ループ) が制御された形で非0 終了する義務を負う (record が失われた
ことが画面に必ず残る)。
"""
journal = dict(payload)
trial_status = (payload.get("ledger_row") or {}).get("trial_status", "unknown")
if trial_status == "completed":
storage_status = "mount_lost_detected_after_physical_sequence"
else:
storage_status = "mount_lost_detected_mid_sequence:{}".format(trial_status)
journal["journal"] = {
"kind": "recovery_trace_not_canonical",
"reason": reason,
"storage_status": storage_status,
"created_at": datetime.now().astimezone().isoformat(),
}
path = EMERGENCY_JOURNAL_DIR / "{}.json".format(payload["run_stem"])
try:
_atomic_json(path, journal)
except Exception as e:
msg = "[JOURNAL-FAIL] run_stem={} 記録は失われた: {}: {}".format(
payload.get("run_stem", "?"), type(e).__name__, e)
print(msg, file=os.sys.stderr)
raise RecordLost(msg) from e
return path
def _header_and_stems_locked(f) -> set:
f.seek(0)
reader = csv.DictReader(f)
if reader.fieldnames != HANDOVER_LEDGER_COLS:
raise AppendAborted("[ABORT-APPEND] ledger header mismatch: {}".format(LEDGER_PATH))
return {r.get("run_stem", "") for r in reader}
def append_ledger_once(row: dict) -> bool:
"""Append under an exclusive lock. Return False if run_stem already exists."""
if not ensure_ssd_mounted():
raise AppendAborted("[ABORT-APPEND] KIOXIA unmounted immediately before append")
try:
with LEDGER_PATH.open("r+", newline="", encoding="utf-8") as f:
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
stems = _header_and_stems_locked(f)
if row["run_stem"] in stems:
return False
f.seek(0, os.SEEK_END)
csv.DictWriter(f, fieldnames=HANDOVER_LEDGER_COLS).writerow(row)
f.flush()
os.fsync(f.fileno())
return True
except FileNotFoundError as e:
raise AppendAborted("[ABORT-APPEND] ledger missing: {}".format(LEDGER_PATH)) from e
def persist_or_journal(payload: dict) -> str:
"""Write canonical run JSON then append. Any mount/write refusal makes a trace."""
try:
if not ensure_ssd_mounted():
raise AppendAborted("KIOXIA unmounted before run JSON")
run_dir = RUNS_ROOT / payload["run_stem"]
run_dir.mkdir(parents=True, exist_ok=False)
_atomic_json(run_dir / PARAMETERS_JSON_NAME, payload)
appended = append_ledger_once(payload["ledger_row"])
return "appended" if appended else "already_recorded"
except (AppendAborted, OSError) as e:
print(str(e), file=os.sys.stderr)
path = write_emergency_journal(payload, "{}: {}".format(type(e).__name__, e))
print("[EMERGENCY-JOURNAL] {}".format(path), file=os.sys.stderr)
return "journaled"
def recover_journals(run_stem: Optional[str] = None) -> tuple:
"""Recover finalized ``*.json`` traces after KIOXIA is mounted.
Existing ledger rows are left untouched. Canonical run JSON is copied only
when absent; an existing run directory is never overwritten.
E-4 Mod 1 (❌ A4/P1): CLI 経由の run_stem と各 journal payload["run_stem"]
の両方を validate_run_stem で検証。加えて journal ファイル名 stem との
一致も確認 (`path.stem == payload["run_stem"]`)。不正はその journal のみ
failed カウントし継続 (他の journal 回収は阻害しない)。CLI 値が不正なら
ValueError で即中断 (recovery 開始前に呼び出し元へ返す)。
E-4 Mod 2 (❌ D6/P1): 各 journal の run_dir.mkdir + _atomic_json 直前に
ensure_ssd_mounted() を再確認。失敗時は AppendAborted で loop 中断
(以降の journal は EMERGENCY_JOURNAL_DIR にそのまま残る = 再実行で回収可能・
無停電で mount が瞬断した場合も安全)。
E-4 R2-3: canonical run JSON への storage 反映は **新規作成時のみ**。
canonical は「journal ブロック全体を剥がす」のではなく、`storage_status` を
トップレベルに残して書く (kind/reason/created_at は剥がす)。既に canonical
が存在する場合は上書きしない (params_path.exists() ガード)。
**storage 障害の正は journal 側**。canonical への反映は新規作成時のみで、
全ケース反映は保証しない (既存 canonical は運用上そもそも journal と併存
しないため実務上稀・保守的な保証を優先)。
"""
if run_stem is not None:
validate_run_stem(run_stem) # Mod 1: CLI 値検証 (不正なら ValueError)
if not ensure_ssd_mounted():
raise AppendAborted("[RECOVER] KIOXIA is not mounted/writable")
paths = ([EMERGENCY_JOURNAL_DIR / "{}.json".format(run_stem)] if run_stem else
sorted(EMERGENCY_JOURNAL_DIR.glob("*.json")))
recovered = skipped = failed = 0
for path in paths:
if not path.is_file():
failed += 1
print("[RECOVER] journal missing: {}".format(path), file=os.sys.stderr)
continue
try:
payload = json.loads(path.read_text(encoding="utf-8"))
stem = payload["run_stem"]
# Mod 1: payload の run_stem 検証
validate_run_stem(stem)
# Mod 1: filename ↔ payload run_stem の整合 (二重チェック)
if path.stem != stem:
raise ValueError(
"[RECOVER] filename/run_stem mismatch: {} vs {}".format(
path.name, stem))
run_dir = RUNS_ROOT / stem
params_path = run_dir / PARAMETERS_JSON_NAME
if not params_path.exists():
# Mod 2: mkdir + _atomic_json の直前に mount を再確認。瞬断で
# 部分 write されると canonical が破損するため、書込前で止める。
if not ensure_ssd_mounted():
raise AppendAborted(
"[RECOVER] KIOXIA unmounted mid-recovery before writing {}"
.format(params_path))
run_dir.mkdir(parents=True, exist_ok=True)
# R2-3: canonical は「journal ブロックを丸ごと剥がす」ではなく
# storage_status だけを top-level に残して書く。kind/reason/
# created_at (recovery 実施時の trace メタデータ) は canonical
# に載せない (journal 側が正)。
canonical = dict(payload)
journal_block = canonical.pop("journal", None)
if isinstance(journal_block, dict):
storage_status = journal_block.get("storage_status")
if storage_status:
canonical["storage_status"] = storage_status
_atomic_json(params_path, canonical)
if append_ledger_once(payload["ledger_row"]):
recovered += 1
else:
skipped += 1
except AppendAborted as e:
# mount 瞬断 (Mod 2 経路) → loop 中断・残りの journal は次回に持ち越し
print(str(e), file=os.sys.stderr)
raise
except Exception as e:
failed += 1
print("[RECOVER] failed {}: {}: {}".format(path, type(e).__name__, e), file=os.sys.stderr)
return recovered, skipped, failedcodex が recovery に見つけた2つの穴 — run_stem のパス結合と mount 再確認の欠落
初回レビューで ❌ が2件出た。どちらも recovery（SSD 復旧後の回収）経路の穴で、通常試行の物理安全には触れない。
1件目は run_stem のパストラバーサル。--recover の引数と journal ファイル内の run_stem を、検証なしでそのままパスに結合していた。--recover ../../foo のような値や、改変された journal の "run_stem": "../../foo" で、基準ディレクトリの外を参照できてしまう。通常実行の make_run_stem() は日時と数字しか生成しないので安全だが、recovery は外から来る値を扱う。修正は正規表現での検証。run_handover_{8桁日付}_{6桁時刻}_{6桁マイクロ秒}_{3桁試行番号} に一致しない値は reject し、journal ファイル名と payload 内の run_stem の一致も確認する。
2件目は recovery 途中の mount 再確認漏れ。recovery は開始時に1回だけマウントを確認して複数の journal を処理していた。全件回収の途中で SSD が外れると、以降の run JSON 書き込みが古いマウントポイント側へ向かう可能性がある。修正は、各 journal の書き込み直前に ensure_ssd_mounted() を再実行し、失敗したらループを中断すること。残った journal は /tmp にそのまま残るので、再実行すれば回収できる。
あわせて、退避完了後に mount lost を検出した試行の記録方針も確定した。台帳の trial_status は物理結果のまま（completed 等）にし、ストレージの事情は journal 側の storage_status フィールドに分離する。値は「いつ外れたか」でなく「いつ検出したか」を表す語彙にした。mount guard を置いていない区間では、外れた瞬間は分からない。分かるのは検出した瞬間だけだ。断定できないことを断定する語彙にしない。
修正版が再レビューを通過し、E-4 がコミットに入った。
[main 2c935bf] DAY293 E-4: 台帳append + emergency journal + recovery。動作パラメータ不変・物理指令前mount guard追加（codex 2ラウンド通過）
2 files changed, 704 insertions(+), 22 deletions(-)
create mode 100644 dynamic_tracking_limits/place/place_modules/handover_recording.py撮影は RELEASE 直後、VLM は HOME 到達後 — 人の手が可動域にある間 API を呼ばない
E-5 は2つの部品でできている。RELEASE 直後（人の手がまだ近くにある時点）に上下カメラで1枚ずつ撮る観測撮影と、アームが HOME に帰りきってから、その画像を VLM（画像を判定できる AI モデル）へ投げて「ブロックは取られたか」を参考値として判定する事後判定。
順序が設計の核心になっている。撮影は RELEASE の直後でないと、受け渡しの瞬間の状態が写らない。だが VLM の API 呼び出しは、人の手が可動域にある間は絶対にやらない（恒久ルール）。API は何秒かかるか分からず、その間アームの隣で人が待つことになるからだ。だから撮るのは先、判定は HOME 到達後。コードの構造上、VLM 呼び出しには HOME 到達判定を通過しないと到達できない。
もう1つの原則は、撮影は観測のみということ。撮影コードからアームへの指令（send_angles・グリッパー・daemon 接続）は一切出さない。撮影が失敗してもシーケンスは止まらず、fail:{理由} を台帳に記録して続行する。
この E-5 も初回レビューで ❌ が3件出た。一番効いたのは撮影の例外処理で、cv2.VideoCapture()（カメラを開く処理）の生成と isOpened() が例外捕捉の外にあった。ここで例外が出ると、インターロック2・退避・HOME 復帰に進まずシーケンス全体が中断する。「撮影失敗は退避を巻き込まない」の約束が破れていた。修正は撮影経路全体を1つの try/except/finally に包むこと。あとの2件は、クレジット切れ判定が文字列検索で広すぎた件（HTTP ステータスコード 400 と本文の "credit balance" の2条件に修正）と、VLM が返した verdict を無検証で台帳に入れていた件（taken / not_taken / unknown の3語彙に強制変換）。
確定版の3関数を全量で載せる。
def _capture_handover_frame(camera_role: str, label: str,
run_stem: str, dry_run: bool) -> dict:
"""RELEASE 直後の観測撮影 1 枚。E-5 spec 2 (撮影は観測のみ・アーム指令ゼロ)。
- camera_role: "top" (Continuity/iPhone 14 Pro Max) or "side" (Iriun/iPhone 11)
- 保存先: RUNS_ROOT/{run_stem}/{label}.jpg (MASTER 保存先テーブル準拠)
- dry_run: 撮影 skip・"skipped:dry_run" 返し (E-4 と同じ原則)
- mount 未確認時は fail:mount_lost (ローカル fallback 禁止・E-5 spec 5)
- cv2 / anthropic 等の import は関数内 (dry_run / --recover では触らない)
E-5 R2-1 反映: dry_run 早期 return と run_stem 検査以降、**全処理を 1 つの
try/except/finally に包む** (from import・VideoCapture 生成・isOpened・
read・imwrite・release まで)。全例外を "fail:capture_{TypeName}" に
落として続行 (シーケンス本体を止めない・E-5 spec 「撮影の例外が退避・HOME
復帰を巻き込まない」)。cap=None で開始し、finally で `cap is not None` の
ときだけ release (VideoCapture 生成自体が例外の場合の safe cleanup)。
戻り値: {"status": "ok" | "fail:{reason}" | "skipped:dry_run", "path": str}
E-4 台帳の 41-44 列に直接マップされる (build_ledger_row L234-237 参照)。
"""
if dry_run:
return {"status": "skipped:dry_run", "path": ""}
if not run_stem:
return {"status": "fail:missing_run_stem", "path": ""}
cap = None
try:
# ---- mount 確認 (ローカル fallback 禁止) ----
if not ensure_handover_ssd_mounted():
return {"status": "fail:mount_lost", "path": ""}
# ---- camera 解決 (番号決め打ち禁止・camera_resolver 使用) ----
from camera_resolver import resolve_cameras # 関数内 import
cams = resolve_cameras(
require_top=(camera_role == "top"),
require_side=(camera_role == "side"),
verbose=False,
)
cam_idx = cams.get(camera_role) if cams else None
if cam_idx is None:
return {"status": "fail:no_{}_camera".format(camera_role), "path": ""}
# ---- 保存先ディレクトリ + cv2 VideoCapture ----
import cv2 # 関数内 import (dry_run / --recover では触らない)
from handover_recording import RUNS_ROOT # 関数内 import (循環回避)
run_dir = RUNS_ROOT / run_stem
run_dir.mkdir(parents=True, exist_ok=True)
out_path = run_dir / "{}.jpg".format(label)
cap = cv2.VideoCapture(int(cam_idx))
if not cap.isOpened():
return {"status": "fail:cap_open", "path": ""}
# ---- warm-up + drop + 1 frame ----
for _ in range(_CAPTURE_WARMUP_FRAMES):
cap.read()
for _ in range(_CAPTURE_DROP_FRAMES):
cap.read()
ret, frame = cap.read()
if not ret or frame is None:
return {"status": "fail:read_fail", "path": ""}
if not cv2.imwrite(str(out_path), frame):
return {"status": "fail:imwrite", "path": ""}
return {"status": "ok", "path": str(out_path)}
except Exception as e:
# E-5 R2-1: 撮影経路の全例外を fail:capture_{TypeName} で吸収。
# シーケンス本体 (interlock2/RETRACT/HOME) を巻き込まない。
return {"status": "fail:capture_{}".format(type(e).__name__), "path": ""}
finally:
# E-5 R2-1: VideoCapture 生成自体が例外の場合を含めた safe cleanup。
# cap=None のとき (生成前 or 生成失敗) は release を呼ばない。
if cap is not None:
try:
cap.release()
except Exception:
passdef _capture_after_release(run_stem: str, dry_run: bool) -> dict:
"""RELEASE 直後 (interlock2 前) の観測撮影 (top + side 各 1 枚・E-5 spec 1)。
E-3 の _capture_after_release_stub を置き換える。撮影は観測のみで、アーム
指令 (send_angles / gripper / daemon 新規接続) を一切出さない (E-5 spec 2)。
撮影失敗はシーケンスを止めない (E-5 spec 「fail:{reason} を記録して続行」)。
返り値の shape は E-4 台帳の 41-44 列 (top/side_capture_status/path) に
直接マップされる (handover_recording.build_ledger_row L234-237 参照)。
"""
top = _capture_handover_frame("top", "handover_release_top", run_stem, dry_run)
side = _capture_handover_frame("side", "handover_release_side", run_stem, dry_run)
if not dry_run:
print("[HANDOVER] capture top.status={} side.status={}".format(
top.get("status"), side.get("status")))
return {"top": top, "side": side}_HANDOVER_VLM_VERDICT_VOCAB = ("taken", "not_taken", "unknown")
# R3: 実機実体 place_modules/vlm_judge.py (2eb53e3 で追跡入り) L41/44/45 と一致
_HANDOVER_VLM_MODEL = "claude-sonnet-4-6" # place_modules/vlm_judge.py VLM_MODEL
_HANDOVER_VLM_TIMEOUT_SEC = 30.0 # 同 VLM_TIMEOUT_SEC
_HANDOVER_VLM_MAX_TOKENS = 512 # 同 VLM_MAX_TOKENS
def _judge_handover_taken(top_image_path: str, dry_run: bool) -> dict:
"""RELEASE 直後の top 画像を Anthropic に投げて「取られたか」を参考値判定
(E-5 spec 3-4)。**呼び出し timing は _run_handover_sequence 側の責務**で、
本関数は HOME 到達を仮定して呼ばれる (assertion は呼び出し側)。
E-5 R3 反映: 実機実体は `place_modules/vlm_judge.py` (urllib 直呼び・
2eb53e3 で追跡入り・実機で数百試行の判定実績あり)。本関数は model /
timeout / max_tokens をそれに一致させた **SDK 独立実装** (urllib 書換は
R2 で確定した credit 分類・例外捕捉・timeout を壊さないため見送り)。
認証は vlm_judge.py と同じ ANTHROPIC_API_KEY 環境変数を使う (key
fallback ファイル読取は本関数では実装せず・SDK が env を読む標準経路)。
既存 vlm_judge.py の関数・prompt は 1 文字も改変しない。
定数対応 (place_modules/vlm_judge.py L41/44/45 と bit-perfect 一致):
_HANDOVER_VLM_MODEL = "claude-sonnet-4-6" (VLM_MODEL)
_HANDOVER_VLM_TIMEOUT_SEC = 30.0 (VLM_TIMEOUT_SEC)
_HANDOVER_VLM_MAX_TOKENS = 512 (VLM_MAX_TOKENS)
E-5 R2-2 反映: Path.exists 検査を含む全非 dry-run 経路を try/except で
包み、どの例外でも `{"status": "error", ...}` を返す (試行を
unexpected_error に化けさせない)。credit 切れ判定は SDK 例外の
`status_code == 400` + 本文 "credit balance" の 2 条件 (文字列 "400" の
素朴 in 検索を廃止・誤分類回避)。
E-5 R2-3 反映: parsed["verdict"] を正式語彙 (taken/not_taken/unknown) に
coerce。非文字列や範囲外は "unknown"。reason も str に制限 (非文字列は
repr の短縮)。台帳 col 38 に不正語彙が入る経路をゼロに。
dry_run: {"verdict": "", "status": "skipped:dry_run", "reason": "dry_run"}
正常: {"verdict": "taken"|"not_taken"|"unknown", "status": "ok", "reason": str}
credit 切れ: {"verdict": "unknown", "status": "http_400_no_credit", ...}
その他エラー: {"verdict": "unknown", "status": "error", "reason": "{type}: {msg}"}
"""
if dry_run:
return {"verdict": "", "status": "skipped:dry_run", "reason": "dry_run"}
# R2-2: 全非 dry-run 経路を 1 つの try で包む (Path.exists 含む)
try:
# ---- 画像 path 検査 ----
if not top_image_path:
return {"verdict": "unknown", "status": "error", "reason": "no_image_path"}
p = Path(top_image_path)
if not p.exists():
return {"verdict": "unknown", "status": "error",
"reason": "top_image not available"}
# ---- base64 encode ----
import base64 as _base64
with open(p, "rb") as f:
image_b64 = _base64.b64encode(f.read()).decode("ascii")
# ---- Anthropic SDK 呼出し (R2-4: timeout 30s 明示) ----
from anthropic import Anthropic
client = Anthropic()
message = client.messages.create(
model=_HANDOVER_VLM_MODEL,
max_tokens=_HANDOVER_VLM_MAX_TOKENS,
timeout=_HANDOVER_VLM_TIMEOUT_SEC,
messages=[{
"role": "user",
"content": [
{"type": "image", "source": {
"type": "base64", "media_type": "image/jpeg",
"data": image_b64,
}},
{"type": "text", "text": _HANDOVER_VLM_PROMPT},
],
}],
)
response_text = message.content[0].text
# ---- JSON 抽出 ----
import re as _re
import json as _json
m = _re.search(r"\{.*?\}", response_text, _re.DOTALL)
if not m:
return {"verdict": "unknown", "status": "error",
"reason": "no_json_in_response"}
parsed = _json.loads(m.group())
# ---- R2-3: verdict / reason の語彙 coerce ----
verdict_raw = parsed.get("verdict")
if not isinstance(verdict_raw, str):
verdict = "unknown"
else:
verdict_norm = verdict_raw.strip().lower()
if verdict_norm in _HANDOVER_VLM_VERDICT_VOCAB:
verdict = verdict_norm
else:
verdict = "unknown"
reason_raw = parsed.get("reason", "")
if isinstance(reason_raw, str):
reason = reason_raw
else:
# 非文字列 → repr の短縮 (台帳の text 列に置ける安全形式)
reason = repr(reason_raw)[:200]
return {"verdict": verdict, "status": "ok", "reason": reason}
except Exception as e:
# R2-2: SDK 例外の status_code + 本文 "credit balance" で credit 切れ検出
# (文字列 "400" の in 検索を廃止・誤分類回避)。それ以外は "error"。
status = "error"
status_code = getattr(e, "status_code", None)
if status_code == 400:
# 本文の error.message を優先確認、fallback で str(e)
body = getattr(e, "body", None) or {}
body_msg = ""
if isinstance(body, dict):
err_dict = body.get("error") or {}
if isinstance(err_dict, dict):
body_msg = str(err_dict.get("message", "")).lower()
fallback_msg = str(e).lower()
if "credit balance" in body_msg or "credit balance" in fallback_msg:
status = "http_400_no_credit"
# reason は 400 文字で切る (台帳・JSON の肥大化防止)
return {"verdict": "unknown", "status": status,
"reason": "{}: {}".format(type(e).__name__, e)[:400]}vlm_judge.py が2本あった — 実機で動いていたのは追跡外の7月版
E-5 のレビューの過程で、想定していなかった事実が出た。VLM 判定モジュール vlm_judge.py が、リポジトリに2本ある。
$ md5 vlm_judge.py dynamic_tracking_limits/place/place_modules/vlm_judge.py
MD5 (vlm_judge.py) = a7bd8007ac642ac2527682bcaca099d7
MD5 (dynamic_tracking_limits/place/place_modules/vlm_judge.py) = fceb0aaa99864f5893ea8daa8405f4ad
$ grep -n "import vlm_judge\|from vlm_judge" dynamic_tracking_limits/place/place_modules/run_pick_place.py
dynamic_tracking_limits/place/place_modules/run_pick_place.py:158:from vlm_judge import judge as vlm_judge_call # noqa: E402
$ python3 -c "import sys; sys.path.insert(0, 'dynamic_tracking_limits/place/place_modules'); import vlm_judge; print(vlm_judge.__file__)"
/Users/shimazueiichi/r2r2r-scripts/dynamic_tracking_limits/place/place_modules/vlm_judge.pymd5 が違う。中身が別物だ。そして Python の import が解決するのは place_modules 側——つまり実機で数百試行の VLM 判定を担ってきた実体は、git に追跡されていない方のファイルだった。中身を見比べると、リポジトリ直下の追跡版は4月の古い実装（5,141バイト・SDK 方式）、place_modules 側は7月の新版（13,458バイト・urllib 方式・横カメラ判定の kind 拡張入り）。
$ ls -l vlm_judge.py dynamic_tracking_limits/place/place_modules/vlm_judge.py
-rw-r--r-- 1 shimazueiichi staff 13458 7月 11 16:44 dynamic_tracking_limits/place/place_modules/vlm_judge.py
-rwxr-xr-x@ 1 shimazueiichi staff 5141 4月 11 12:58 vlm_judge.py
$ grep -n "VLM_MODEL\|VLM_TIMEOUT_SEC\|VLM_MAX_TOKENS" dynamic_tracking_limits/place/place_modules/vlm_judge.py | head -5
41:VLM_MODEL = "claude-sonnet-4-6" # ビジョン対応・現行 Sonnet
44:VLM_TIMEOUT_SEC = 30.0
45:VLM_MAX_TOKENS = 512
115: "model": VLM_MODEL,
226: "model": VLM_MODEL,この2本問題は、レビューの現場で実害を出していた。codex が「vlm_judge.py の実物」として引いた値と、Claude Code が「実物」として引いた値が食い違い、どちらが正しいかで1往復無駄にした。答えは「両方とも実在するファイルを正確に読んでいた。読んでいたファイルが違った」。誰も間違ったコードを書いていないのに、全員が「正本」の指す先を取り違えていた。原因は同名2本の放置そのものだ。
解消は3手。実機実績を持つ place_modules 版を git 追跡に入れ、直下の旧版を archive へコピー退避し、直下から除去した。除去前に、直下版を import している他のスクリプトが無いことも grep で確認した（0件）。E-5 の VLM 定数（model・timeout・max_tokens）も、実機実績のある新版の値に揃えた。
グリッパーが開かなかったとき、アームは止まるか帰るか
この回で唯一、物理の挙動を変える判断があった。
RELEASE 失敗——アームが物を差し出し、グリッパーに「開け」と命令したのに開かなかった場合。それまでの実装は「その場で止まる」だった。ブロックを保持したまま静止し、記録を残し、手動復帰を待つ。グリッパーの状態が不明なまま自動で動くのは危険、という設計レビュー時の判断（Q1）に基づいていた。
この方針を今回撤回した。アームの動きを止めてしまうのは NG。人の手のすぐ横で、アームが保持姿勢のまま静止し続ける——それは「次に何が起きるか読めない」状態が続くということでもある。ブロックは保持したままでいい。開け直しの再試行はしない（handover 中の再送禁止は恒久ルール）。その代わり、アームは自分で HOME まで帰る。
ただし保持したまま人の近くで動く以上、丸ごと自動化はしない。設計はこうなった。
RELEASE 失敗
→ その場で静止（新規指令なし・グリッパー再送なし・ブロック保持のまま）
→ 画面表示「RELEASE 失敗・ブロック保持中。手を可動域から退避させて Enter」
→ 人の退避確認入力（インターロック2と同一の入力機構）
→ RETRACT_SMALL → PRESENT_MID → HOME（正常系と完全同一の退避経路・同一 deadline）
→ 記帳正常系との差は「開いたか開いていないか」だけで、退避の経路も確認手順も同一。人から見て、アームは毎回同じ振る舞いをする。この「同一」を口約束にしないため、退避3ウェイポイントを _run_retreat_3wp という1つの関数に切り出し、正常系と失敗系の両方が同じ関数を呼ぶ構造にした。経路・速度・deadline の一致が、コードの構造で保証される。
記録の語彙も1つ広げた。台帳の outcome=stuck は元々「開いたがブロックが爪に引っかかったまま」の意味だったが、「爪にブロックが残ったまま（理由問わず）」に解釈を広げ、従来の引っかかりと今回の RELEASE 失敗は release_observed 列（open / not_open）で区別する。ブロックの物理状態を語彙にするという視点で一貫させ、語彙の追加はゼロで済ませた。
確定版の3関数を全量で載せる。_run_handover_sequence はシーケンス全体の orchestration で、RELEASE 失敗分岐（6-b・6-c）と E-5 の撮影・VLM 呼び出し位置もこの中にある。
def _perform_release(gate, dry_run: bool) -> dict:
"""RELEASE を実行し SIGINT deferral 込みで結果を返す (E-3a C3 + DAY293
RELEASE 失敗退避のための test seam)。dry_run では skip・release_command_ok
=True 返し。本関数は _run_handover_sequence から呼ばれる。tests で mock 可能
な形で release 失敗経路 (release_command_ok=False) を検証するための seam
でもある。
返り値 dict:
release_command_ok / release_error / release_sent_at /
release_responded_at / deferred_was_interrupted / sigint_install_error
"""
if dry_run:
now = time.time()
return {
"release_command_ok": True,
"release_error": None,
"release_sent_at": now,
"release_responded_at": now,
"deferred_was_interrupted": False,
"sigint_install_error": None,
}
# E-3a C3: RELEASE 送信〜応答〜記録確定を SIGINT から保護
release_command_ok = False
release_error: Optional[str] = None
deferred = _DeferredSigint()
with deferred:
release_sent_at = time.time()
try:
gate.release_open() # daemon 送信 + Pi 側 wait 2s
release_command_ok = True
except ForbiddenGripperClose as e:
release_error = "ForbiddenGripperClose: {}".format(e)
release_command_ok = False
print("[HANDOVER] RELEASE で close が要求された (構造破損): {}"
.format(e))
except Exception as e:
release_error = "{}: {}".format(type(e).__name__, e)
release_command_ok = False
print("[HANDOVER] RELEASE 失敗 (再送・再接続なし・ブロック保持中): {}"
.format(e))
release_responded_at = time.time()
return {
"release_command_ok": release_command_ok,
"release_error": release_error,
"release_sent_at": release_sent_at,
"release_responded_at": release_responded_at,
"deferred_was_interrupted": deferred.was_interrupted,
"sigint_install_error": deferred._install_error,
}def _run_retreat_3wp(pose_data: dict, log_record: dict, dry_run: bool,
context_prefix: str = "") -> bool:
"""RETRACT_SMALL → PRESENT_MID → HOME の 3 waypoint 退避。
**正常系 (sequence 9-11) と RELEASE 失敗時退避 (DAY293) で共有し、経路・
speed・deadline を構造的に完全同一にする** (安全要件3・send_once_and_wait_arrival
が pose JSON の deadline 係数から算出する仕様に依存)。
context_prefix: log の abort stage 文字列 suffix (例: "_release_failed"・
normal path は空文字)。normal / release_failed の分離を可能にする。
返り値: True (3wp 全到達) / False (どこかで未到達)
False 時、log_record["handover"]["phase"] に "aborted_at_retract" /
"aborted_at_pmid_retreat" / "aborted_at_home" を set、
manual_recovery_required=True を set、_abort() で abort を記録する。
trial_status は本関数では触らない (呼び出し側の規約: normal path は
motion_failed 上書き、release_failed path は release_failed 保持)。
"""
rs_angles = pose_data["poses"]["RETRACT_SMALL"]["angles"]
r9 = _do_handover_move(rs_angles, "RETRACT_SMALL", pose_data,
log_record, dry_run)
if not r9["arrived"]:
log_record["handover"]["phase"] = "aborted_at_retract"
log_record["manual_recovery_required"] = True
_abort(log_record, "handover_seq9" + context_prefix,
"RETRACT_SMALL 未到達 (完了表示なし・手動復帰へ): {}"
.format(r9.get("reason")))
return False
pmid_angles = pose_data["poses"]["PRESENT_MID"]["angles"]
r10 = _do_handover_move(pmid_angles, "PRESENT_MID", pose_data,
log_record, dry_run)
if not r10["arrived"]:
log_record["handover"]["phase"] = "aborted_at_pmid_retreat"
log_record["manual_recovery_required"] = True
_abort(log_record, "handover_seq10" + context_prefix,
"PRESENT_MID (退避) 未到達 (完了表示なし・手動復帰へ): {}"
.format(r10.get("reason")))
return False
home_angles = pose_data["poses"]["HOME"]["angles"]
r11 = _do_handover_move(home_angles, "HOME", pose_data,
log_record, dry_run)
if not r11["arrived"]:
log_record["handover"]["phase"] = "aborted_at_home"
log_record["manual_recovery_required"] = True
_abort(log_record, "handover_seq11" + context_prefix,
"HOME 未到達 (完了表示なし・手動復帰へ): {}"
.format(r11.get("reason")))
return False
return Truedef _run_handover_sequence(
trial: Optional[dict],
pose_data: dict,
log_record: dict,
dry_run: bool,
) -> bool:
"""r3.4 2節 sequence 2-11 を実行。呼び出し前提: E-2 の PICK 完了 + HOME
復帰が成功していること。返り値: True (sequence 11 まで到達) / False (中断・
未到達・失敗のいずれか)。log_record["handover"] に全結果を残す (E-4 台帳
が拾える形)。close 送信ゼロは _HandoverPhaseGate が構造保証。
r3.4 2節 の中断条件対応:
- PRESENT 未到達 (3): インターロック1 表示を出さない (r3.4 2節3・
「手を出してください」を出さない = 人に手を出させない安全要件)
- 9-11 未到達: 完了表示を出さず manual_recovery_required を記録
"""
trial_wait_sec = int(trial["wait_sec"]) if trial else 5
log_record["handover"] = {
"phase": "start",
"trial_wait_sec_requested": trial_wait_sec,
"sequence": [],
}
gate = _HandoverPhaseGate()
gate.enter_handover()
try:
# ---- 2. PRESENT_MID へ移動 + 到達判定 ----
if not _guard_mount_before_sequence(log_record, "handover_seq2", dry_run):
log_record["handover"]["phase"] = "mount_lost_before_present_mid"
return False
pmid_angles = pose_data["poses"]["PRESENT_MID"]["angles"]
r2 = _do_handover_move(pmid_angles, "PRESENT_MID", pose_data,
log_record, dry_run)
if not r2["arrived"]:
log_record["handover"]["phase"] = "aborted"
log_record["trial_status"] = "motion_failed"
_abort(log_record, "handover_seq2",
"PRESENT_MID 未到達: {}".format(r2.get("reason")))
return False
# ---- 3. PRESENT へ移動 + 到達判定 ----
if not _guard_mount_before_sequence(log_record, "handover_seq3", dry_run):
log_record["handover"]["phase"] = "mount_lost_before_present"
return False
pres_angles = pose_data["poses"]["PRESENT"]["angles"]
r3 = _do_handover_move(pres_angles, "PRESENT", pose_data,
log_record, dry_run)
if not r3["arrived"]:
# r3.4 2節3: PRESENT 未到達なら「手を出してください」を表示せず
# 中断。**インターロック1 の入力プロンプトを出さない** (下記の 4
# 番へ進まないことで実現)。
log_record["handover"]["phase"] = "aborted_before_interlock1"
log_record["trial_status"] = "motion_failed"
log_record["handover"]["interlock1_skipped_reason"] = \
"PRESENT unreached (r3.4 2節3・手を出させない)"
_abort(log_record, "handover_seq3",
"PRESENT 未到達 (interlock1 表示なし・手動復帰へ): {}"
.format(r3.get("reason")))
print("[HANDOVER] PRESENT 未到達。インターロック1 は表示しません "
"(r3.4 2節3・手動復帰へ)。")
return False
# ---- 4. インターロック1 (WAIT 開始前・Enter まで送信なし・Q2: q で中断可) ----
# PRESENT 移動中に mount lost した場合も、人へ手を出すよう促す前に止める。
# この確認を最後の mount guard とし、Enter 通過後は退避完了まで確認で
# 止めない (記録より物理安全を優先し、最後に journal へ退避)。
if not _guard_mount_before_sequence(log_record, "interlock1", dry_run):
log_record["handover"]["phase"] = "mount_lost_before_interlock1"
return False
if dry_run:
interlock1 = InterlockConfirmation(
label="interlock1_wait_start", role="operator",
ts_monotonic=time.monotonic(), ts_wall=time.time())
print("[HANDOVER dry_run] interlock1 skip (input なし)")
else:
# Q2 反映: D-2 の wait_interlock ではなく E-3 wrapper 経由 (q 対応)。
# D-2 は不変。q → None で受けて Ctrl-C と同じ終着へ。
interlock1 = _wait_interlock_or_abort(
"interlock1_wait_start",
"静止・到達を確認。準備ができたら Enter (受け取り開始)",
role="operator")
if interlock1 is None:
log_record["handover"]["phase"] = "aborted_at_interlock1"
_record_interlock_abort(log_record, "interlock1_wait_start")
return False
log_record["handover"]["interlock1"] = {
"label": interlock1.label, "role": interlock1.role,
"ts_monotonic": interlock1.ts_monotonic,
"ts_wall": interlock1.ts_wall,
}
# ---- 5. WAIT: trial["wait_sec"] 秒・monotonic 実測・送信ゼロ ----
# E-3a B2 反映: 一発 time.sleep(wait_sec) から分割 sleep + monotonic
# 経過判定へ。Ctrl-C は従来どおり即応 (WAIT 中の中断は安全: RELEASE 前)、
# 中断時も wait_sec_actual が monotonic 実測で正確に残る。
# 送信ゼロは不変 (time.sleep と時刻取得のみ)。
_WAIT_SLICE_SEC = 0.1
wait_started_mono = time.monotonic()
wait_started_wall = time.time()
wait_interrupted = False
try:
if not dry_run:
while True:
elapsed = time.monotonic() - wait_started_mono
remaining = trial_wait_sec - elapsed
if remaining <= 0:
break
time.sleep(min(_WAIT_SLICE_SEC, remaining))
except KeyboardInterrupt:
wait_interrupted = True
finally:
wait_sec_actual = time.monotonic() - wait_started_mono
log_record["handover"]["wait"] = {
"wait_sec_requested": trial_wait_sec,
"wait_sec_actual": wait_sec_actual,
"wait_started_wall": wait_started_wall,
"dry_run": dry_run,
"interrupted": wait_interrupted,
}
if wait_interrupted:
# WAIT 中の Ctrl-C は RELEASE 前で安全 → 外側 except に伝播させ
# _record_handover_interrupt が interrupted 記録
raise KeyboardInterrupt()
# ---- 6. RELEASE: GripperOpenOnly.open() (再送・再接続なし) ----
# 防衛線: この直前で「open 以外の action は禁止」を宣言的に確認。
gate.assert_no_close_attempted("open")
# DAY293: RELEASE 実行を _perform_release helper に集約 (test seam
# でもある・SIGINT deferral 込み)。dry_run では skip される。
release_result = _perform_release(gate, dry_run)
log_record["handover"]["release"] = {
"release_command_ok": release_result["release_command_ok"],
"release_error": release_result["release_error"],
"release_sent_at": release_result["release_sent_at"],
"release_responded_at": release_result["release_responded_at"],
"dry_run": dry_run,
"sigint_deferred_during_send": (not dry_run),
"sigint_install_error": release_result["sigint_install_error"],
}
# E-3a C3: 遅延中に SIGINT が来ていた場合、記録が確定した今こそ
# KeyboardInterrupt を再送出する。外側 except handler
# (_record_handover_interrupt) が interrupted 記録を追加する
# (release_command_ok は既に上で確定済み・両方 log_record に残る)。
if release_result["deferred_was_interrupted"]:
log_record["handover"]["release_interrupted_after_send"] = True
raise KeyboardInterrupt()
if not release_result["release_command_ok"]:
# DAY293 仕様変更: RELEASE 失敗時はブロック保持のまま人退避確認 →
# 正常系と同一の 3wp 退避 (RETRACT_SMALL → PRESENT_MID → HOME)。
# Q1 の「その場停止・手動復帰」ポリシーは撤回。アームを止めっぱなし
# にせず、ブロック保持のまま HOME まで帰る (人退避確認を挟んで
# 安全境界を作る)。再送禁止は handover phase の恒久ルールと整合。
log_record["handover"]["phase"] = "release_failed_retreat_pending"
log_record["trial_status"] = "release_failed"
log_record["handover"]["release_failed_retreat_started"] = True
_abort(log_record, "handover_seq6_release_failed",
"release 失敗 (ブロック保持のまま退避へ・DAY293 仕様変更)")
# ---- 6-b. 人退避確認 (仕様書指定文言・safety req 2: 送角ゼロで
# 入力待ち・Ctrl-C はこの待ち中も有効・interrupted で記帳・
# 自動動作なし) ----
print("[HANDOVER] RELEASE 失敗・ブロック保持中。"
"アームは保持したまま退避します。")
print("[HANDOVER] 手を可動域から退避させて Enter "
"(safety-rated ではなく手順的保護)")
if dry_run:
interlock_clr = InterlockConfirmation(
label="interlock_release_failed_hand_clear",
role="assistant",
ts_monotonic=time.monotonic(), ts_wall=time.time())
print("[HANDOVER dry_run] release_failed hand_clear 確認 skip")
else:
interlock_clr = _wait_interlock_or_abort(
"interlock_release_failed_hand_clear",
"RELEASE 失敗・ブロック保持中。手を可動域から退避させて "
"Enter",
role="assistant")
if interlock_clr is None:
log_record["handover"]["phase"] = \
"release_failed_interrupted_at_hand_clear"
_record_interlock_abort(log_record,
"interlock_release_failed_hand_clear")
return False
log_record["handover"]["interlock_release_failed_hand_clear"] = {
"label": interlock_clr.label, "role": interlock_clr.role,
"ts_monotonic": interlock_clr.ts_monotonic,
"ts_wall": interlock_clr.ts_wall,
}
log_record["handover"]["hand_clear_confirmed"] = True
# ---- 6-c. 正常系と同一 3wp 退避 (安全要件3: 経路・deadline
# 完全同一・_run_retreat_3wp helper で構造保証) ----
if not _run_retreat_3wp(pose_data, log_record, dry_run,
context_prefix="_release_failed"):
# 退避中失敗 → helper が manual_recovery_required=True set 済み
# trial_status="release_failed" は保持 (DAY293 spec: 変更なし)
return False
log_record["handover"]["phase"] = "release_failed_retreated_to_home"
log_record["manual_recovery_required"] = False # HOME 到達
# trial 全体としては失敗 (RELEASE failed)・sequence 7/8/12 は skip
return False
# ---- 7. 撮影 (RELEASE 直後・interlock2 前・観測のみ・E-5) ----
# 撮影経路からアーム指令 (send_angles / gripper / daemon 接続新規)
# を一切出さない (E-5 spec 2)。撮影失敗はシーケンスを止めない。
run_stem = log_record.get("run_stem", "")
cap_status = _capture_after_release(run_stem, dry_run)
cap_status["captured_at_wall"] = time.time() if not dry_run else None
log_record["handover"]["capture"] = cap_status
# ---- 8. インターロック2 (退避開始前・hand_clear_confirmed・Q2: q で中断可) ----
if dry_run:
interlock2 = InterlockConfirmation(
label="interlock2_hand_clear", role="assistant",
ts_monotonic=time.monotonic(), ts_wall=time.time())
print("[HANDOVER dry_run] interlock2 skip (input なし)")
else:
# Q2 反映: interlock2 も同 wrapper 経由 (q 対応)。D-2 不変。
interlock2 = _wait_interlock_or_abort(
"interlock2_hand_clear",
"手を離し、アームの届く範囲の外へ出たことを確認したら Enter (退避開始)",
role="assistant")
if interlock2 is None:
log_record["handover"]["phase"] = "aborted_at_interlock2"
_record_interlock_abort(log_record, "interlock2_hand_clear")
return False
log_record["handover"]["interlock2"] = {
"label": interlock2.label, "role": interlock2.role,
"ts_monotonic": interlock2.ts_monotonic,
"ts_wall": interlock2.ts_wall,
}
log_record["handover"]["hand_clear_confirmed"] = True
# ---- 9-11. 退避 3wp (RETRACT_SMALL → PRESENT_MID → HOME) ----
# DAY293: _run_retreat_3wp helper 経由で RELEASE 失敗時退避と
# 経路・deadline を構造的に完全同一にする (安全要件3・pose JSON の
# deadline 係数から send_once_and_wait_arrival が算出する仕様)。
# helper は phase / manual_recovery_required / _abort を set するが
# trial_status は触らないため、退避失敗時は呼び出し側で motion_failed
# を明示 set (release_failed retreat の呼出し規約と対称)。
if not _run_retreat_3wp(pose_data, log_record, dry_run):
log_record["trial_status"] = "motion_failed"
return False
# ---- 12. VLM 事後判定 (E-5 spec 3・HOME 到達後のみ・参考値) ----
# r3.4 恒久ルール: 人の手が可動域にある間は API を呼ばない (VLM 実行は
# 退避完了 = HOME 到達後に限る)。本ブロックは _run_retreat_3wp が
# True (RETRACT_SMALL / PRESENT_MID / HOME すべて arrived) を返した
# 場合のみ到達する。したがって「home_reached_bool 通過前に VLM が
# 発火する」経路が構造上存在しない (E-5 検証 (e) の構造保証)。
# 加えて、DAY293 の RELEASE 失敗退避経路は sequence 6 内で return
# False で抜けるため、この VLM 呼出しに到達しない (release_failed
# では VLM を呼ばない・block は保持のままなので verdict は not_taken
# 定義的に自明)。
# 撮影失敗時 (top_path が空) は _judge_handover_taken が
# "image not available" で error を返し、シーケンスは止まらない
# (VLM は参考値・E-5 spec 4)。
top_path = ((cap_status.get("top") or {}).get("path")) or ""
vlm_result = _judge_handover_taken(top_path, dry_run)
vlm_result["called_at_wall"] = time.time() if not dry_run else None
log_record["handover"]["vlm"] = vlm_result
if not dry_run:
print("[HANDOVER] VLM: verdict={!r} status={!r} reason={!r}".format(
vlm_result.get("verdict"), vlm_result.get("status"),
vlm_result.get("reason")))
log_record["handover"]["phase"] = "complete"
log_record["trial_status"] = "completed"
return True
finally:
gate.exit_handover()テストが緑でも検証していなかった — dry_run の空履歴と mount mock 漏れ
E-6 は検証テスト一式。Ctrl-C の注入点、RELEASE 失敗、mount lost の3時点、グリッパー close 0回の構造保証、台帳の追記ガード、recovery の3経路、E-5 の失敗系。安全設計を「一度確認した」で終わらせず、回帰テストとして固定する。
ここでも codex レビューが2回、テストの穴を見つけた。どちらも「テストが緑なのに、何も検証していない」型だった。
1つ目。グリッパー送信履歴の動的検査が dry_run で走っていた。dry_run では RELEASE 自体が呼ばれないので、gripper の履歴は常に空。「close が0件」の assert は、何もしていないから0件、で常に通る。Log#3 の parity mock（既存本線と比べるはずが比べていなかった）と完全に同じ形だ。修正は、dry_run でなく完全モック環境で実際の release 経路を通し、「open（state=0）がちょうど1回記録される」ことを正の値で assert すること。空振りなら履歴が0件になり、テストが落ちる。
2つ目。修正後の新規7テストが、codex の実行環境で全部 mount_lost に落ちた。テスト共通の helper が ensure_handover_ssd_mounted() をモックしておらず、KIOXIA が繋がっていない環境では mount guard で止まって、検証したい分岐（RELEASE 失敗・退避確認・waypoint 失敗）に到達していなかった。こちらの手元では KIOXIA が繋がっていたから通り、codex の環境では落ちた——テストの結果がマウント状態という環境に依存していたこと自体が欠陥で、mock 1行の追加でこの依存を断った。
静的検査の側も1段強くした。AST（コードを文字列でなく構文木として解析する方式）で handover 中に実行される3関数（_run_handover_sequence / _perform_release / _run_retreat_3wp）を走査し、gripper close 系の直呼びが0件であることを検査する。from visual_servo_pick import daemon_set_gripper as _sg のような別名 import で呼んでも、別名を追跡して検出する。検査部を全量で載せる。
# §4: gripper close 0 回 (handover phase で close 系送信履歴が構造上ゼロ)
# ==========================================================================
# handover phase で実行される関数のリスト。将来 helper が増えたら追加する
# (S1 静的検査の走査対象・R2-1 反映)。
_HANDOVER_PHASE_FUNCS = (
"_run_handover_sequence", # 全体 orchestration
"_perform_release", # RELEASE (gate.release_open 呼出し)
"_run_retreat_3wp", # 退避 3wp
)
class GripperCloseZeroTest(unittest.TestCase):
"""D-2 docstring が予告していた検査そのもの: handover phase を通しても
close 系呼出しがゼロであることを送信履歴で担保。"""
def test_static_ast_no_close_in_handover_phase_funcs(self):
"""AST 検査: **handover phase で実行される全関数**
(_run_handover_sequence / _perform_release / _run_retreat_3wp) 本体に
daemon_set_gripper / daemon_set_gripper_value / state=1 の直呼びが
0 件。GripperOpenOnly 経由の open のみ許可。
R2-1 反映: 走査対象を単一関数から `_HANDOVER_PHASE_FUNCS` の 3 関数に
拡大。将来 handover phase の helper が増えたらリストに追加する。
別名 import (asname) 対策 (既存): モジュールトップの ImportFrom を
走査し、`from visual_servo_pick import daemon_set_gripper as _sg` の
ような alias 経由の呼出しも検出できるよう、asname を forbidden 集合に
追加する (bare Name 呼出しで `_sg(1)` と書いても検出)。
Attribute 呼出し (`vsp.daemon_set_gripper(1)`) は callee.attr で
検出されるので既存経路のまま。"""
src = Path(sys.modules["run_pick_handover"].__file__).read_text()
tree = ast.parse(src)
# 対象関数を全部集める (R2-1: 3 関数)
target_funcs = {}
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef) and node.name in _HANDOVER_PHASE_FUNCS:
target_funcs[node.name] = node
for fname in _HANDOVER_PHASE_FUNCS:
self.assertIn(fname, target_funcs,
"{} が見つからない (handover phase 関数リスト要更新)".format(fname))
# 基本の禁止名 (関数本名)
core_forbidden = {"daemon_set_gripper", "daemon_set_gripper_value"}
forbidden = set(core_forbidden)
# モジュールトップの ImportFrom を走査し、asname を forbidden に追加
alias_map = {}
for node in tree.body:
if isinstance(node, ast.ImportFrom):
for alias in node.names:
if alias.name in core_forbidden and alias.asname:
forbidden.add(alias.asname)
alias_map[alias.asname] = alias.name
# 各対象関数を個別に走査
all_found = []
for fname, func_node in target_funcs.items():
for node in ast.walk(func_node):
if isinstance(node, ast.Call):
callee = node.func
name = None
if isinstance(callee, ast.Name):
name = callee.id
elif isinstance(callee, ast.Attribute):
name = callee.attr
if name in forbidden:
detail = name
if name in alias_map:
detail = "{} (alias for {})".format(name, alias_map[name])
all_found.append("{} L{}: {}".format(fname, node.lineno, detail))
self.assertEqual(all_found, [],
"handover phase 関数内に close 系直呼び: {}".format(all_found))
def test_dynamic_gripper_history_release_success_open_once_no_close(self):
"""動的検査 (R2-1 反映・S1 dynamic 実質化): _run_handover_sequence を
dry_run=False で isolate 実行し、実際に gate.release_open が呼ばれ
daemon_set_gripper(state=0, ...) が **ちょうど 1 回** 記録されることを
検証。close 系 (state=1 / set_gripper_value) は **ゼロ**。
旧テストの穴 (S1 codex 指摘): dry_run で _perform_release が
release_open を呼ばないため gripper 履歴が常に空。「テストが緑でも
何も検証していない」parity mock 教訓 (E-2b) と同型の穴を塞ぐ。"""
success, log, gripper_hist, move_hist = _call_handover_isolated(
release_ok=True, # 実 gate.release_open を呼ぶ
)
self.assertTrue(success, "release 成功で handover 完走のはず: {}".format(
log.get("handover", {}).get("phase")))
# open (state=0) が実際に呼ばれたことの証明
open_calls = [c for c in gripper_hist
if c["kind"] == "set_gripper" and c.get("state") == 0]
close_calls = [c for c in gripper_hist
if c["kind"] == "set_gripper_value"
or (c["kind"] == "set_gripper" and c.get("state") == 1)]
self.assertEqual(len(open_calls), 1,
"handover phase の open は 1 回のみ (RELEASE): {}".format(gripper_hist))
self.assertEqual(close_calls, [],
"handover phase で close 系呼出しあり: {}".format(close_calls))
# 実物ログ (診断用): 履歴が非空であることを証明する
self.assertGreater(len(gripper_hist), 0,
"gripper_history が空 (S1 旧テストと同じ空振り穴)")
def test_dynamic_gripper_history_release_failure_no_close_during_retreat(self):
"""release 失敗経路でも close 系呼出しゼロ (hand_clear 〜 retreat 完走で
gripper 履歴が増えない・R2-1 補強)。"""
success, log, gripper_hist, move_hist = _call_handover_isolated(
release_ok=False, # _perform_release mock で failure 強制
)
self.assertFalse(success)
# release_ok=False では gate.release_open は呼ばれない → gripper 履歴は空
# (retreat 経路で send_once_and_wait_arrival は呼ばれるが gripper は
# 触らない)
self.assertEqual(gripper_hist, [],
"release 失敗時に gripper 呼出しなし: {}".format(gripper_hist))
# retreat 3wp は実行されていること
retreat_labels = [m["label"] for m in move_hist
if m["label"] in ("RETRACT_SMALL", "PRESENT_MID", "HOME")]
self.assertEqual(retreat_labels[-3:], ["RETRACT_SMALL", "PRESENT_MID", "HOME"])
def test_gripper_open_only_class_has_no_close_methods(self):
"""D-2 の GripperOpenOnly クラスが open メソッドのみ持ち close 系
メソッド (release_close / set_gripper_value 等) を一切公開しない。"""
from handover_release import GripperOpenOnly
g = GripperOpenOnly()
# open メソッド存在
self.assertTrue(hasattr(g, "open"))
# close 系メソッド不在 (構造保証)
forbidden = ["close", "release_close", "set_gripper_value",
"set_gripper", "toggle"]
for name in forbidden:
self.assertFalse(hasattr(g, name),
"GripperOpenOnly に禁止メソッド {} が存在".format(name))
# ==========================================================================
# §1: Ctrl-C 注入点テスト (handover phase の代表点)
# ==========================================================================最終の実行結果。
$ python3 -m unittest discover -s tests -p 'test_handover_*.py'
Ran 47 tests in 8.309s
OK
$ python3 -m unittest tests.test_pick_parity
Ran 6 tests in 1.173s
OKhandover 系 47件（safety 16 / recording 9 / capture・VLM 22）+ 既存 parity 6件 = 53件、失敗ゼロ。
この回で積んだ commit 列と、実機の前に残っているもの
4bb7405 E-6 + RELEASE失敗時退避（codex 2ラウンド通過）
dc72e1e E-5 R3 + 直下旧vlm_judge除去
2eb53e3 vlm_judge実機実体を追跡へ
4b26b38 E-5 本体（codex 2ラウンド通過）
1c22e0b E-4 tests（mount guard モック）
2c935bf E-4（codex 2ラウンド通過）D-1 から始まった受け渡し動作の制御コードは、これで全ブロックが閉じた。書くコードはもうない。
実機の前に残っているのは、コードでない作業だ。提示姿勢はまだ仮確定のままで、二次絞り込み（爪が人の方を向いているかの判定・張り出し・クリアランス）と swept volume（動作中に腕が通過する空間）の確認が済んでいない。非常停止の方法も確定していない。そして人なしの実機 dry-run——通常経路と、今回作った RELEASE 失敗退避の経路の両方を、人が手を出す前に実機で通す。
開かなかったときにアームがどう帰るかを決めた回だった。次は、その帰り方を実機で確かめる。
