AIエージェントを無人で長時間回すときの状態ファイルを壊さない実装手順(Python) — 書き込み途中クラッシュで JSONDecodeError・os.replace の Cross-device link・PIDファイルのロック残留、3つのハマりどころ【2026】

はじめに / 対象と前提

Claude Code などの AI エージェントを cron やループで 無人のまま何時間も回している と、いつか必ず「状態ファイルが壊れて次の起動が全部コケる」日が来る。自分は完全自律実装システム(24 時間エージェントを回し続けている自作の仕組み)でこれを一通り踏んだので、壊れない書き方を実行結果つきでまとめる。

• 想定読者:AI エージェントや定期ジョブのタスク状態を JSON ファイルで管理している人
• 前提:Python の基本、Unix 系 OS(fcntl を使うので Windows は対象外)
• 動作確認環境:macOS 15.7 / Python 3.14.3(標準ライブラリのみ。3.10 以降なら同じ挙動のはず)

TL;DR

• 状態ファイルは 同じディレクトリに一時ファイル → fsync → os.replace で書く。/tmp に作ると別ディスクで Cross-device link になる
• 二重起動防止は PID ファイルではなく fcntl.flock。プロセスが kill -9 されても OS がロックを外してくれる
• 「作業中」のまま落ちたタスクは 期限付きの担当表明(lease) で回収する。期限切れなら別のワーカーが引き取る

手順 / 動かし方

1. まず壊れ方を再現する

エージェントがタスク一覧を書き込んでいる最中に、タイムアウトや OOM で強制終了されたケースを、途中までしか書かれていない JSON で再現する。

import json
open("state.json", "w").write(json.dumps({"tasks": [{"id": "t1", "status": "done"}]})[:20])
json.load(open("state.json"))

JSONDecodeError: Unterminated string starting at: line 1 column 19 (char 18)

open(path, "w") は 開いた瞬間にファイルを空にする。書き終わる前に落ちれば、前回の正しい状態ごと失われる。次回起動時に読み込みで落ち、以降のループが全部止まる。

2. アトミック書き込み

import json, os, tempfile
from pathlib import Path

def atomic_write_json(path, data):
    path = Path(path)
    # 一時ファイルは「書き込み先と同じディレクトリ」に作る(理由はハマりどころ2)
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
            f.flush()
            os.fsync(f.fileno())      # 中身をディスクまで押し込む
        os.replace(tmp, path)         # 置き換えは一瞬で起きる(中途半端な状態がない)
        dfd = os.open(path.parent, os.O_RDONLY)
        try:
            os.fsync(dfd)             # ディレクトリの更新(ファイル名の差し替え)も確定させる
        finally:
            os.close(dfd)
    except BaseException:
        try:
            os.unlink(tmp)
        except FileNotFoundError:
            pass
        raise

os.replace は同じファイルシステム内なら 「古いファイル」か「新しいファイル」のどちらかしか見えない。途中で落ちても残るのは一時ファイルのゴミだけで、本体は前回の正しい状態のまま。

3. 二重起動を fcntl.flock で防ぐ

cron で 5 分おきに起動していると、前回の実行が長引いたときに 2 つ目が走り出し、同じタスクを二重に処理する。

import fcntl, os, sys

def acquire_single_instance(lock_path):
    f = open(lock_path, "a+")
    try:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        f.close()
        return None
    f.seek(0); f.truncate(); f.write(str(os.getpid())); f.flush()
    return f   # 戻り値を変数に持ち続ける(GC で閉じられるとロックも外れる)

lock = acquire_single_instance("agent.lock")
if lock is None:
    print("already running, exit")
    sys.exit(0)

1 つ目を起動したまま 2 つ目を起動 → 1 つ目を kill -9 → 3 つ目を起動、の順で確認した。

acquired 23945
already running, exit
acquired 23949

強制終了された後でも、3 つ目はちゃんとロックを取れている。

4. 「作業中のまま落ちたタスク」を lease で回収する

ファイルが壊れなくても、status: "in_progress" に書き換えた直後に落ちると、そのタスクは永遠に誰も拾わない。そこで担当表明に期限(lease)を付ける。

import json, time
from pathlib import Path

LEASE_SEC = 2   # 動作確認用。実運用では 1 タスクの最大所要時間より長くする
STATE = Path("tasks.json")

def load():
    return json.loads(STATE.read_text(encoding="utf-8"))

def claim_next(worker_id):
    state, now = load(), time.time()
    for t in state["tasks"]:
        expired = t["status"] == "in_progress" and t.get("lease_until", 0) < now
        if t["status"] == "pending" or expired:
            if expired:
                t["attempts"] = t.get("attempts", 0) + 1
            t.update(status="in_progress", owner=worker_id, lease_until=now + LEASE_SEC)
            atomic_write_json(STATE, state)
            return t
    return None

def complete(task_id, worker_id):
    state = load()
    for t in state["tasks"]:
        if t["id"] == task_id:
            if t.get("owner") != worker_id:
                raise RuntimeError(f"lease lost: {task_id} now owned by {t.get('owner')}")
            t.update(status="done", lease_until=None)
    atomic_write_json(STATE, state)

ワーカー A が t1 を取って落ち、期限切れ後に B が引き取り、遅れて戻ってきた A が完了報告しようとするシナリオを実行した。

A claims t1
B claims t1 attempts 1
A: lease lost: t1 now owned by worker-B
[('t1', 'done'), ('t2', 'pending')]

A の古い完了報告は弾かれ、B の結果だけが残る。attempts を数えておけば「3 回落ちたら failed にして人間に回す」といった打ち切りも書ける。

ハマりどころ

ハマりどころ1:open(path, "w") で直接上書きすると JSONDecodeError

• エラー:json.decoder.JSONDecodeError: Unterminated string ... や Expecting value: line 1 column 1 (char 0)(空ファイル)
• 原因:"w" は開いた時点で中身を消す。書き終える前に強制終了されると壊れたファイルだけが残る
• 回避策:手順 2 のアトミック書き込みに統一する。読み込み側で try/except して握りつぶすのは NG(壊れた状態から空の一覧で再出発し、全タスクを最初からやり直す事故になる)

ハマりどころ2:一時ファイルを /tmp に作ると Cross-device link

tempfile.mkstemp() を引数なしで呼ぶと一時ファイルは /tmp 側にできる。状態ファイルが外付けディスクや別ボリュームにあると、os.replace が失敗する。

OSError: [Errno 18] Cross-device link

(Linux では Invalid cross-device link)

• 原因:os.replace は名前の付け替えなので、同じファイルシステム内でしか使えない。ディスクをまたぐとコピーが必要になり、アトミック性もなくなる
• 回避策:mkstemp(dir=path.parent) で必ず書き込み先と同じディレクトリに作る。開発機では同じディスクで通ってしまい、本番の外付けディスクで初めて落ちるのが厄介

ハマりどころ3:PID ファイル方式だとロックが残って二度と起動しない

よくある「起動時に agent.pid を作り、終了時に消す」方式は、kill -9 やマシンのスリープ中断で終了処理が走らないとファイルが残る。

FileExistsError: [Errno 17] File exists: 'agent.pid'

• 原因:ファイルの存在をロック代わりにしているので、後片付けされない限り永久にロック中扱い。「PID のプロセスが生きているか確認する」対策も、PID が別プロセスに再利用されると誤判定する
• 回避策:fcntl.flock を使う。ロックは プロセスが持つファイルの開き口 に紐づくので、プロセスが死ねば OS が自動で外す。ロックファイル自体は消さずに置いておいてよい
• 注意:flock は NFS などのネットワークファイルシステムでは期待通りに効かないことがある。ロックファイルはローカルディスクに置く

背景・補足

手順 4 の claim_next は「読む → 書き換える → 保存」の 3 段階なので、複数ワーカーが同時に呼ぶなら手順 3 のロックで囲む 必要がある。アトミック書き込みは「ファイルが壊れない」ことは保証するが、「2 人が同時に同じタスクを取らない」ことまでは保証しない。役割が違う、と覚えておくと設計を間違えにくい。

タスク数が数百を超えたり、ワーカーが複数マシンに分かれたりしたら、素直に SQLite(BEGIN IMMEDIATE でロック)に移行した方が楽になる。

まとめ

• 状態ファイルの書き込みは「同じディレクトリに一時ファイル → fsync → os.replace」
• /tmp に一時ファイルを作ると、別ディスク上で Cross-device link になる
• 二重起動防止は PID ファイルではなく fcntl.flock。kill -9 後も自動で外れる
• 作業中のまま落ちたタスクは lease(期限付き担当表明)で回収し、attempts で打ち切る
• アトミック書き込みとロックは別物。読み書きの一連の流れはロックで囲む