外部のデータ提供元から、大量のファイルを一括で受け取る処理を書いていました。
途中で、あるファイルが「壊れている」を意味する戻り値を返しました。処理はそこで止まります。
私はAIエージェントに「壊れたファイルは飛ばして続行する」を実装させ、その分岐を検査するテストまで書いて、「対処済み」と記録しました。
**間違っていました。**その対処は、取り直せば直るデータを、永久に失う設計でした。
しかも提供元の仕様書は、**2日前から手元にありました。**開いていなかっただけです。
この記事は、その失敗と、二度と同じことをしないために足した検査の話です。
実装した対処
こういう形でした。壊れたファイルを飛ばして、次へ進みます。
# 実装したもの（誤り）
while True:
rc, path = api.read_next()
if rc == BROKEN: # 壊れている
if SKIP_DAMAGED: # 環境変数で有効化
log.warning("破損ファイルを飛ばす")
api.skip() # ← このファイルを諦めて次へ
continue
raise DataError(rc)
if rc == EOF:
break
process(path)
**動きます。**止まらなくなりました。テストも通りました。
「破損ファイルがあっても処理が完走する」という検査を書いて、緑になったのを確認しています。
「エラーが出なくなった」ので、直ったと判断しました。
テストは通っていた ― 何を検査していたのか
「対処済み」と記録した根拠は、検査を書いて緑になったことでした。その検査はこうでした。
def test_broken_file_is_skipped():
api = FakeApi(broken_at=3) # 3件目を壊す
result = run_collection(api, skip_damaged=True)
assert result.completed is True # 完走した
通ります。実装どおりに動くからです。
問題は、この検査が「止まらないこと」を確かめていて、「失わないこと」を確かめていない点です。
守りたかったのはデータでした。検査したのは処理の完走でした。別の量を検査していた。
書くべきだったのはこちらです。
def test_no_permanent_loss():
api = FakeApi(broken_at=3)
run_collection(api, skip_damaged=True)
# 破損があっても、最終的に全件そろっていること
assert count_records() == EXPECTED_TOTAL # ← ここで落ちる
この検査を先に書いていれば、実装した瞬間に赤になっていました。
検査は、守りたいものを名指しで書く。「エラーが出ない」は、守りたいものではありません。
仕様書に書いてあったこと
あとから該当箇所を読みました。まったく別の対処が指定されていました。
指示の趣旨（原文は提供元のドキュメントなので引用しません）
読み出しAPIが返したファイル名のファイルを削除し、
開き直しからの処理をやり直すこと
削除すれば、次に開き直したときに再ダウンロードされます。
壊れていたのは手元のキャッシュであって、提供元のデータではなかったからです。
飛ばす（実装したもの） そのファイルの中身は**永久に手に入らない**
削除して取り直す（正解） 再ダウンロードされ、**欠測は出ない**
同じ「エラーで止まらない」でも、結果が正反対でした。
私が実装したのは、回復可能な失敗を、回復不能な失敗に変換する処理です。
しかも静かに動くので、**失われたことに誰も気づきません。**ログには
「破損ファイルを飛ばす」と警告が出るだけで、処理は正常終了します。
なぜ読まなかったのか
理由ははっきりしています。実測で振る舞いを確認したからです。
やったこと 壊れたファイルで再現させ、戻り値を確認し、
飛ばせば処理が続くことを確かめた
やらなかったこと 仕様書の該当ページを開くこと
動作を観察したことで、「分かった」と思ってしまった。
ここが今回の核心です。
「実測でこう振る舞った」は、仕様書が別の対処を定めていないことの証明にならない。
観察できるのは「何が起きるか」までです。「何をすべきか」は観察からは出てきません。
飛ばしても動く、は事実です。飛ばすべきだ、はどこにも書かれていませんでした。
どうやって気づいたか ― 偶然です
正直に書きます。検査で見つけたのではありません。
別の作業で同じ仕様書を開いたとき、たまたま該当ページが目に入りました。読んだら、
自分が実装したものと違う対処が書いてありました。それだけです。
見つけた方法 偶然、別件で仕様書を開いた
見つからなかった方法 テスト / レビュー / 実測での動作確認
偶然に頼っている、という事実のほうが、この失敗の本体だと思っています。
だから、偶然を待たない仕組みを足しました。
二重の失敗 ― 対処に必要な情報を、自分で捨てていた
もっと悪いことがありました。
私はコードに、こう注記していました。
# 破損したファイルの名前は例外に載らない。
# → 削除しようにも、どのファイルか分からない
**これも間違いでした。**読み出しAPIは、失敗時にもファイル名を返していました。
なぜ「載らない」と思ったか。自分で書いた例外ラッパが、捨てていたからです。
# 誤り ― 戻り値だけ見て、結果オブジェクトを作る前に投げる
def check_rc(rc, path):
if rc < 0:
raise DataError(f"読み出しに失敗 rc={rc}") # ← path を捨てている
return ReadResult(rc=rc, path=path)
例外にはエラーコードしか載りません。ファイル名は引数として渡ってきているのに、
例外を組み立てる時点で使っていませんでした。
# 修正 ― 失敗しても、対処に必要な情報を必ず載せる
def check_rc(rc, path):
if rc < 0:
raise DataError(f"読み出しに失敗 rc={rc}", path=path) # ← 載せる
return ReadResult(rc=rc, path=path)
仕様書どおりの対処（削除して取り直す）には、ファイル名が要ります。
その唯一の情報を、自分の例外ラッパが握り潰していました。
だから「飛ばすしかない」と思い込み、飛ばす実装が唯一の選択肢に見えていた。
仕様書を読まなかったことと、情報を捨てていたことが、噛み合って一つの誤りになっていました。
AIに実装させるときに、特に起きやすい
この失敗は、AIに書かせていたから起きた、という側面があります。
AIは動くコードを速く出す
→ 動いたので確認が済んだ気になる
→ 「動く」と「仕様どおり」の差が、確認されないまま残る
**AIは仕様書を読んでいません。**私が渡していないからです。
渡していない以上、AIが出せるのは「観察された振る舞いに合う実装」までで、
それは仕様が定める対処とは別物です。
そして厄介なことに、AIはそれらしい理由を付けてくれます。
「破損ファイルはスキップして処理を継続するのが一般的です」と言われれば、
一般論としては正しいので、疑う理由が消えます。
一般論として正しいことと、この提供元の仕様として正しいことは、別です。
足した検査 ― 引用が無ければ通さない
心がけでは同じことが起きるので、機械が見る形にしました。
エラーコードへの対処を実装するときは、仕様書の該当箇所を引用する。引用が無いものは通さない。
# 破損を示す戻り値への対処
#
# 仕様: 提供元SDKドキュメント「破損時の対処」の節
# 要旨: 返されたファイル名のファイルを削除し、開き直しからやり直す
#
# → skip ではなく delete + reopen。skip は永久欠測になる
def handle_broken(path):
os.remove(path)
return REOPEN
検査はこれだけです。エラーコード分岐の近くに、出典（文書名とページ）を含むコメントがあるか。
import re, sys, io, os
CITE = re.compile(r'(仕様|spec|ドキュメント|マニュアル).{0,40}(p\.?\s*\d+|§|セクション|\d+\.\d+)')
BRANCH = re.compile(r'(rc\s*[=<>!]=?\s*-?\d+|errno|ERROR_CODE|status_code\s*==)')
def check(path):
lines = io.open(path, encoding='utf-8', errors='ignore').read().splitlines()
bad = []
for i, line in enumerate(lines):
if not BRANCH.search(line):
continue
window = '\n'.join(lines[max(0, i - 12):i + 1]) # 直前12行に出典があるか
if not CITE.search(window):
bad.append((i + 1, line.strip()[:70]))
return bad
if __name__ == '__main__':
ng = 0
for root, _dirs, files in os.walk(sys.argv[1] if len(sys.argv) > 1 else '.'):
if any(s in root for s in ('.git', 'venv', '__pycache__')):
continue
for f in files:
if not f.endswith('.py'):
continue
p = os.path.join(root, f)
for ln, src in check(p):
print(f'{p}:{ln} 出典コメントが無いエラー分岐: {src}')
ng += 1
print(f'\n出典なしの分岐 {ng} 件')
sys.exit(1 if ng else 0)
**これは「正しさ」を検査していません。**引用があるかどうかしか見ていません。
それでも効きます。引用を書こうとした瞬間に、仕様書を開くことになるからです。
検査の目的は、間違いを見つけることではなく、読まずに書く経路を塞ぐことです。
同じ型が他にどれだけあるか、数えた
検査を書いたので、自分のコード全体に当ててみました。
エラーコード分岐 合計 172 件
出典コメントが無い分岐 148 件（86%） / 47 ファイル
**86%です。**1件の事故として処理していたものが、実際には自分の書き方の既定値でした。
もちろん、172件すべてが同じ危険を持つわけではありません。
**多くは無害です。**この検査は正しさを見ていないので、当然そうなります。
それでも、「1件やらかした」と「86%がその書き方だった」は、対処が変わります。
前者なら直して終わりですが、後者なら書き方そのものを変える必要があります。
事故を1件として数えるか、率として数えるかで、打つ手が変わる。
これは今回いちばん役に立った気づきでした。
この失敗の一般形
1 観察は「何が起きるか」しか教えない。「何をすべきか」は仕様にしか無い
2 エラー処理は、**失敗の種類を変えてしまう**ことがある
回復可能な失敗を、回復不能な失敗に変換していないか
3 例外を組み立てるとき、**対処に必要な情報を落としていないか**
落とすと、選べたはずの対処が見えなくなる
4 「エラーが出なくなった」は、直った証拠にならない
5 AIは仕様書を読んでいない。渡していないなら、読んでいない
**2 がいちばん怖いと思っています。**エラー処理は「止まらないようにする」ために書きますが、
止まらないようにした結果、静かに失うようになったなら、止まっていたほうがましでした。
止まるのは事故ですが、静かに失うのは、事故だと気づけません。
まとめ
- **仕様書は2日前から手元にあった。**読まなかっただけ
- 実測で振る舞いを確認したことが、読まない理由になっていた
- 実装した対処は、取り直せば直るデータを永久に失う設計だった
- 自分の例外ラッパが、対処に必要な唯一の情報を捨てていた
- 足した検査は「正しさ」ではなく 「出典が書いてあるか」 だけを見る
- 読まずに書く経路を塞ぐことが、検査の目的
この連載
AIエージェントを実際に回して、かかった費用と壊れた箇所を測って書いています。推測は書きません。実測値だけです。
- Claude Code のトークン使用量を実測したら1ターン23万 ― 節約に効いたのは「セッションを切る」だった
https://qiita.com/manabu49-ai/items/2598a30d5140e4445ab6 - 57分間、死んだジョブを誰も見ていなかった ― AIエージェントの見張りを機械に渡す
https://qiita.com/manabu49-ai/items/5c7506c94b5be58c040e - AIに100通り試させたら、効果ゼロなのに5個が「有意」だった
https://qiita.com/manabu49-ai/items/397a018948e1ab7ef29e - 本記事 ― 仕様書は2日前から手元にあった ― 読まずに実装した対処が、データを永久に失う設計だった
- 次 ― AI初心者が3日でエージェントを24時間回すまで（未公開）
番外 ― Claude のモデル提供終了、猶予は告知から約60日 ― 過去9回を数えて、手元を検査するスクリプトを書いた
https://qiita.com/manabu49-ai/items/bc2a7be2a6ee4ceb6185
**無料記事は結論まで全部書きます。**出し惜しみはしません。