AIエージェントの定期実行が「完了しました」なのに成果物がない・止まっている原因と対処｜予定・起動・処理・成果物を分ける完了判定、実行ID・二重起動防止・未実行の検知（Python）

こんにちは、おうどんです🍜

エージェントくん「本日分の下書き、保存しました。完了しました！」

おうどん「ありがとう。開いてみるね」

おうどん「……0バイト」

エージェントくん「保存は、しました！」

（会話は説明用の架空の場面です）

炊飯器の予約炊飯を思い浮かべてください。
夜にお米をといで、朝7時にセット。朝、「ご飯が炊けました♪」のメロディが鳴る。
ふたを開けたら、お米を入れ忘れていて、お湯だけがほかほか。

炊飯器さん「炊けました♪」

おうどん「何を？」

メロディは、炊飯器が「自分の仕事が終わった」と言っているだけです。釜の中身までは保証してくれません。

毎日決まった時間に動く AI エージェント（AI に作業を任せて、自分で道具を使わせて進めるしくみ）も同じです。
おうどんも毎朝、Claude Code の定期タスクで記事の下書きを作ってもらっています。毎朝動くものほど、「完了しました」の一言だけで済ませたくなるんですよね。

今回は、その「完了」をどこで、何で判断すればいいかを、Python で実際に動かして確かめました。

先に結論です。

• 「完了しました」の返事と終了コード 0 は、成果物が空でも出せる。成果物は、ファイルを開いて中身で確かめる
• 予定・起動・処理・成果物・届け先は別々の段階。段階ごとに記録を残す
• 起動しなかった日は、失敗の知らせも来ない。外から「あるはずの日」を数えて、記録がない日を見つける

「完了しました」は、炊飯器のメロディと同じ

メロディが鳴ったことと、ご飯が炊けたことは別です。まずはこれを、実際のコードで見てみます。

わざと本文を書き忘れるエージェントを用意しました（説明用の作りものです）。

# fake_agent.py  「完了しました」と言うけれど、本文を書き忘れるエージェント（説明用）
from pathlib import Path
import sys

out = Path(sys.argv[1])
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text("", encoding="utf-8")   # 本文を書き忘れた
print("記事の下書きを保存しました。完了しました！")

これを呼んで、返事とファイルを別々に見ます。

# run_and_check.py  エージェントを呼んで、返事とファイルを別々に見る
import subprocess
import sys
from pathlib import Path

target = Path("out/2026-10-22/wordpress.md")
r = subprocess.run([sys.executable, "-X", "utf8", "fake_agent.py", str(target)],
                   capture_output=True, text=True, encoding="utf-8")

print("終了コード:", r.returncode)
print("エージェントの返事:", r.stdout.strip())
print("ファイルはある？:", target.exists())
print("サイズ:", target.stat().st_size, "バイト")

終了コード: 0
エージェントの返事: 記事の下書きを保存しました。完了しました！
ファイルはある？: True
サイズ: 0 バイト

（今回の検証環境：Windows 上の Python 3.13.1。python -X utf8 run_and_check.py で実行）

終了コード（プログラムが終わるときに返す番号。0 が成功）は 0。返事は元気。ファイルもある。
でも、中身は 0 バイトです。

エージェントくん「言われたとおり、ファイルは作りました！」

読者「作ったのは、空の箱だけでは」

終了コード 0 が言っているのは「プログラムが、エラーで落ちずに最後まで進んだ」ということだけです。何を書いたかまでは知りません。
終了コードの基本は終了コードの基礎講座にまとめています。

完了までの段階は、5つに分けて考える

炊飯器でたとえると、毎朝の作業はこう分かれます。

段階	炊飯器でいうと	何があれば「済んだ」と言えるか
予定	7時に予約した	スケジュールの設定と、次に動く時刻
起動	スイッチが入った	実行ごとの記録（実行 ID と開始時刻）
処理	炊いている・炊けたと鳴った	終了コード、エラーの記録、「完了」の返事
成果物	釜にご飯がある	ファイルの中身が決めた条件を満たしている
届け先	お茶碗によそって食卓に出た	置き場所への保存、送信、push（Git でサーバーへ送ること）などの記録

「完了しました」は、表の3段目の話です。
4段目と5段目は、エージェント自身の返事ではなく、別の方法で確かめます。

炊飯器さん「私の担当は3段目までです♪」

おうどん「せめて、お米が入っているかは見てほしかった」

成果物は、ファイルを開いて中身で確かめる

釜のふたを開ける係を作ります。
「ファイルがある」だけでなく、「小さすぎない」「必ず入るはずの部分がある」まで見ます。

# check_artifact.py  成果物そのものを確かめる
from pathlib import Path

REQUIRED = ["こんにちは、おうどんです", "## 参考資料"]


def check_artifact(path, min_bytes=500):
    p = Path(path)
    if not p.exists():
        return "NG: ファイルがない"
    size = p.stat().st_size
    if size < min_bytes:
        return f"NG: 小さすぎる（{size} バイト）"
    text = p.read_text(encoding="utf-8")
    missing = [s for s in REQUIRED if s not in text]
    if missing:
        return f"NG: 必要な部分がない {missing}"
    return f"OK（{size} バイト）"


if __name__ == "__main__":
    base = Path("out/check")
    base.mkdir(parents=True, exist_ok=True)
    body = "こんにちは、おうどんです🍜\n" + "本文です。\n" * 100
    (base / "empty.md").write_text("", encoding="utf-8")
    (base / "half.md").write_text(body, encoding="utf-8")
    (base / "good.md").write_text(body + "## 参考資料\n- 資料\n", encoding="utf-8")

    for name in ["missing.md", "empty.md", "half.md", "good.md"]:
        print(f"{name:11} -> {check_artifact(base / name)}")

missing.md  -> NG: ファイルがない
empty.md    -> NG: 小さすぎる（0 バイト）
half.md     -> NG: 必要な部分がない ['## 参考資料']
good.md     -> OK（1769 バイト）

half.md は、書き出しはあるのに最後の参考資料がないファイルです。途中で止まったときにできる形ですね。
「ある」だけを見ていると、empty.md も half.md も合格になってしまいます。

見回り係さん「ふた、開けました。お湯です」

淡々と言われると、かえって刺さります。

ここで大事なのは、条件を「自分の成果物ならこうなっているはず」で決めることです。
記事なら、決まった書き出しと締めの見出し。CSV なら、見出しの行と行数。JSON なら、読み込めることと必須の項目。
何を見るかは成果物しだいなので、上のコードの REQUIRED と min_bytes は例です。

読者「AI に『ちゃんと書けた？』って聞けばよくない？」

それは、炊飯器に「ご飯ある？」と聞くのと同じです。返ってくるのは、たぶん「炊けました♪」。

起動しなかった日は、失敗の知らせも来ない

いちばん見落としやすいのが、ここです。

「失敗したらメールで知らせる」は、多くの場合、動いているプログラム自身が送ります。
ということは、そもそも起動しなかった日は、誰も知らせを送りません。

炊飯器さん（コンセントが抜けている）「…………」

おうどん「静かな朝だなあ、と思ってた」

静かな朝は、うまくいった朝とは限らない。
しかも、起動しない理由は、プログラムの外にあることが多いんです。

たとえば Claude Code のデスクトップ版の定期タスク（ローカルで動くもの）は、公式ドキュメントによると、アプリが開いていて、パソコンが起きている間だけ動きます。予定の時刻にパソコンが寝ていたら、その回はスキップされます。
また、権限モードが Manual のタスクが、許可されていない道具を使おうとすると、人が許可するまで止まる、とも書かれています（無人の定期実行が確認で止まる話は、Claude Code の定期実行が確認ダイアログで止まる記事で書きました）。

だから、見回りは外からやります。
「記録がある日」を並べるのではなく、「あるはずの日」を数えて、記録がない日を見つける。
このあと中級者向けの章で、そのためのスクリプトを作ります。

読者「便りがないのは元気な証拠、じゃないの？」

定期実行では、便りがないのは、だいたい何かあった証拠です。

🔰 ここまで読めば今日から困らない

• 「完了しました」と終了コード 0 は、プログラムが成功として終わったという報告。意図した処理や成果物の完成までは保証しないので、ファイルを開いて中身で確かめる
• 成果物の確かめ方は「ある」「小さすぎない」「必ず入る部分がある」の3つから始める
• 起動しなかった日は、失敗の知らせも来ない。毎日、「今日の分の記録と成果物があるか」を外から見る

ここから先は中級者向けです。段階ごとの記録、見回りのスクリプト、日付・二重起動・書きかけファイルの対策を、完成形のコードまで作ります。

実行IDつきで、段階ごとに1行ずつ記録する

炊飯器に、「スイッチが入った」「炊けた」「よそった」を1行ずつメモしてもらうイメージです。
1回の実行ごとに実行 ID（その回を見分ける番号）を付け、段階が進むたびに1行足します。

# runlog.py  実行IDつきで、段階ごとに1行ずつ記録する
import json
import uuid
from datetime import datetime, timezone, timedelta
from pathlib import Path

JST = timezone(timedelta(hours=9))


class RunLog:
    def __init__(self, task, for_date, path="runs.jsonl", clock=None):
        self.path = Path(path)
        self.clock = clock or (lambda: datetime.now(JST))
        self.base = {"task": task, "for_date": for_date,
                     "run_id": uuid.uuid4().hex[:8]}

    def stage(self, name, **extra):
        rec = {"at": self.clock().isoformat(timespec="seconds"),
               **self.base, "stage": name, **extra}
        with self.path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")

使うときは、こうです。

# runlog_demo.py
from runlog import RunLog
log = RunLog("daily-article", "2026-10-22", path="demo_runs.jsonl")
log.stage("started")
log.stage("artifact_saved", bytes=1234)
print(open("demo_runs.jsonl", encoding="utf-8").read(), end="")

{"at": "2026-10-22T08:17:06+09:00", "task": "daily-article", "for_date": "2026-10-22", "run_id": "fec8b2cd", "stage": "started"}
{"at": "2026-10-22T08:17:06+09:00", "task": "daily-article", "for_date": "2026-10-22", "run_id": "fec8b2cd", "stage": "artifact_saved", "bytes": 1234}

1行が1つの JSON（JSON Lines という形式）なので、追記するだけで済み、あとから1行ずつ読めます。おうどんの家計簿も、この形式なら続いたかもしれません。続かなかったのは、形式のせいではありませんが。
ポイントは2つあります。

• at（いつ記録したか）と for_date（何日分の仕事か）を分けて持つ。遅れて動いた回や、日付をまたいだ回を見分けるためです
• 段階の名前を決めておく。この記事では started → artifact_saved → verified（成果物を確かめた）→ delivered（届け先に出した）→ finished の順にしました。失敗は failed、わざと実行しなかった回は skipped_late です

エージェントくん「全部の段階で finished と書いておけば、安心ですね！」

おうどん「それをやると、メモ帳がメロディの録音になる」

見回りは外から、「あるはずの日」を数える

記録を読んで、日付ごとに「どこまで進んだか」を判定するスクリプトです。

# check_runs.py  実行記録から、日付ごとに「どこまで進んだか」を判定する
import json
import sys
from collections import defaultdict
from datetime import date, datetime, timedelta


def load(path):
    runs = defaultdict(list)
    with open(path, encoding="utf-8") as f:
        for line in f:
            rec = json.loads(line)
            runs[rec["run_id"]].append(rec)
    return runs


def judge(recs, now, stall_after):
    stages = [r["stage"] for r in recs]
    if "skipped_late" in stages:
        return "見送り（遅れすぎたので実行しなかった）"
    if "failed" in stages:
        reason = next(r.get("reason", "") for r in recs if r["stage"] == "failed")
        return f"失敗（{reason}）"
    if "finished" in stages:
        if "verified" in stages and "delivered" in stages:
            return "OK"
        return "完了と言ったが、成果物の確認か届け先への送付がない"
    started = datetime.fromisoformat(recs[0]["at"])
    if now - started > stall_after:
        return f"止まっている？（{started:%H:%M} 開始、最後は {stages[-1]}）"
    return f"実行中（最後は {stages[-1]}）"


def report(path, dates, now, stall_after=timedelta(hours=2)):
    by_date = defaultdict(list)
    for run_id, recs in load(path).items():
        by_date[recs[0]["for_date"]].append((run_id, recs))
    for d in dates:
        runs = by_date.get(d.isoformat(), [])
        if not runs:
            print(f"{d}  --------  記録なし（起動していない可能性）")
            continue
        if len(runs) > 1:
            print(f"{d}  注意: {len(runs)} 回起動している")
        for run_id, recs in runs:
            print(f"{d}  {run_id}  {judge(recs, now, stall_after)}")


if __name__ == "__main__":
    path = sys.argv[1]
    now = datetime.fromisoformat(sys.argv[2])
    first = date.fromisoformat(sys.argv[3])
    days = int(sys.argv[4])
    report(path, [first + timedelta(days=i) for i in range(days)], now)

説明用に、5日分の記録を作って読ませました（記録は時刻を固定して作ったもので、10/19 の分はわざと入れていません）。

python -X utf8 check_runs.py sample_runs.jsonl 2026-10-22T08:40:00+09:00 2026-10-18 5

2026-10-18  a1b2c3d4  OK
2026-10-19  --------  記録なし（起動していない可能性）
2026-10-20  e5f6a7b8  止まっている？（08:04 開始、最後は artifact_saved）
2026-10-21  c9d0e1f2  完了と言ったが、成果物の確認か届け先への送付がない
2026-10-22  注意: 2 回起動している
2026-10-22  a3b4c5d6  OK
2026-10-22  e7f8a9b0  実行中（最後は started）

1行ずつ、炊飯器で読むとこうなります。

日付	判定	炊飯器でいうと	次に見るところ
10/18	OK	炊けて、よそった	なし
10/19	記録なし	コンセントが抜けていた	スケジューラ側の履歴、パソコンが起きていたか
10/20	止まっている？	炊いている途中で、何かを待っている	その回のログ・画面。確認待ちや異常終了がないか
10/21	完了と言ったが確認なし	メロディは鳴ったが、ふたを開けていない	成果物そのもの、確認の段階が失敗した記録（今回は verify_ng）
10/22	2回起動	誰かがもう一度スイッチを押した	2回目が何をしたか。二重に保存・送信していないか

見回り係さん「10/19 さん、いらっしゃいますかー」

返事がない、ということがわかる。これが見回りの仕事です。

stall_after（何時間たったら止まっているとみなすか）は、ふだんの所要時間より長めにします。上では2時間にしました。

読者「見回り係が寝坊したら？」

鋭い。見回り係の置き場所の話は、上級者向けの章でします。

「今日」は、どこの時計の今日か

段階の記録に for_date を持たせた理由がこれです。
同じ瞬間でも、どこの時計で見るかで日付が変わります。

# dates.py  同じ瞬間でも、どこの時計で見るかで日付が変わる
from datetime import datetime, timezone, timedelta

from daily_job import scheduled_for

JST = timezone(timedelta(hours=9))

t = datetime(2026, 10, 21, 23, 30, tzinfo=timezone.utc)
print("UTC で見た日付:", t.date())
print("JST で見た日付:", t.astimezone(JST).date())

print("--- この実行は、どの予定の分か（予定は毎日 8:00 JST）")
for s in ["2026-10-22T08:03", "2026-10-22T23:10", "2026-10-23T00:30"]:
    now = datetime.fromisoformat(s).replace(tzinfo=JST)
    sch = scheduled_for(now)
    print(f"{s} に起動 -> {sch:%Y-%m-%d %H:%M} の分（遅れ {now - sch}）")

UTC で見た日付: 2026-10-21
JST で見た日付: 2026-10-22
--- この実行は、どの予定の分か（予定は毎日 8:00 JST）
2026-10-22T08:03 に起動 -> 2026-10-22 08:00 の分（遅れ 0:03:00）
2026-10-22T23:10 に起動 -> 2026-10-22 08:00 の分（遅れ 15:10:00）
2026-10-23T00:30 に起動 -> 2026-10-22 08:00 の分（遅れ 16:30:00）

（scheduled_for は、あとで出てくる完成形の daily_job.py の関数です）

日本の朝 8:30 は、UTC（世界の基準の時刻。日本より9時間遅い）ではまだ前日の 23:30。サーバーやコンテナの時計が UTC のまま date.today() でフォルダ名を決めると、日本の朝の仕事が「昨日のフォルダ」に入ります。
タイムゾーンの扱いは Python の datetime の naive・aware の記事で詳しく書きました。

炊飯器さん「7時に炊けました♪ ロンドンの7時ですが」

おうどん「朝ごはんが夕方に炊けてる」

下の3行は、遅れて動いた回の話です。
Claude Code のデスクトップ版の定期タスクのドキュメントには、アプリの起動時やパソコンが起きたときに、過去7日以内に逃した回があれば、いちばん最近に逃した時刻の分を1回だけ追いかけて実行する、と書かれています。それより古い分は捨てられます。朝9時の予定が、一日中寝ていたら夜11時に動くこともある、という例も載っていて、時間に関わる作業なら、指示の中に時刻の歯止めを書くよう勧めています。

つまり、夜 23:10 に動いた回は「今日の夜の分」ではなく「今朝 8:00 の分」。日付をまたいだ 00:30 の回も、まだ「昨日の朝の分」です。
date.today() で決めると、00:30 の回は翌日のフォルダに、翌日の分として書いてしまいます。

エージェントくん「明日の分を、先にやっておきました！」

読者「明日のあなたが、もう一回やるんですよ」

予定時刻から何日分の仕事かを決め、遅れすぎていたら実行しない（skipped_late を記録する）。この判断は、完成形のコードに入れてあります。

同じ日の作業を、2回走らせない

予定が2回起動することもあります。
Kubernetes（コンテナをまとめて動かす基盤）の CronJob（定期実行のしくみ）のドキュメントには、状況によっては1回の予定で Job が2つ作られることも、1つも作られないこともあり、完全には防げないので、Job は冪等（何回実行しても結果が同じになること）にしておくべきだ、と書かれています。

炊飯器さん「念のため、2回炊いておきました♪」

おうどん「お釜が2つある家の人だけに言って」

いちばん簡単な防ぎ方は、作業の前に鍵のファイルを作ることです。
Python の open() の 'x' モードは、組み込み関数のドキュメントのとおり「ファイルがすでにあれば失敗する」作成で、すでにあると FileExistsError になります。

# lock_demo.py  同じ日の作業を、同時に2つ走らせない
import sys
import time
from pathlib import Path

name = sys.argv[1]
lock = Path("out/2026-10-22.lock")
lock.parent.mkdir(parents=True, exist_ok=True)
try:
    with lock.open("x", encoding="utf-8") as f:   # 'x' はファイルがあると失敗する
        f.write(name)
except FileExistsError:
    print(f"{name}: 鍵がもうある。ほかの実行が作業中なので中止します")
    sys.exit(0)

print(f"{name}: 鍵を取れました。作業します")
time.sleep(2)
lock.unlink()
print(f"{name}: 作業が終わったので鍵を返しました")

これを、0.3秒ずらしてほぼ同時に2つ起動しました。

# lock_launch.py  lock_demo.py をほぼ同時に2つ起動する
import subprocess
import sys
import time

procs = []
for name in ["1回目", "2回目"]:
    procs.append(subprocess.Popen([sys.executable, "-X", "utf8", "lock_demo.py", name],
                                  stdout=subprocess.PIPE, text=True, encoding="utf-8"))
    time.sleep(0.3)
for p in procs:
    out, _ = p.communicate()
    print(out, end="")

1回目: 鍵を取れました。作業します
1回目: 作業が終わったので鍵を返しました
2回目: 鍵がもうある。ほかの実行が作業中なので中止します

（出力は、1回目の分をまとめて表示してから2回目の分を表示しています。待ち時間からすると、実際の順番は、1回目が鍵を取る → 2回目が中止 → 1回目が鍵を返す、のはずです。時刻までは記録していません）

エージェントくん「1回目が鍵を返したので、ぼくもう一回やっていいですか？」

いいわけがない。でも、鍵だけだと、それを通してしまいます。

鍵が守るのは「同時に2つ」まで。1回目が終わって鍵を返したあとに来た2回目は、そのまま作業してしまいます。
そこで、完成形では「その日の成果物がもうあれば作業しない」も合わせて入れます。

書きかけのファイルを、完成品の場所に置かない

もう1つ、炊飯器が途中で停電したときの話です。
停電したら、その日はうどんにすればいい。うどん屋の家なので。……と言いたいところですが、ファイルには代わりのうどんがありません。

本物のファイルに直接書いている途中で落ちると、書きかけのファイルが本物の場所に残ります。
一時ファイルに書いてから、最後に入れ替えるとどうなるかを比べました。

# crash_write.py  書いている途中で落ちたら、ファイルはどうなる？
import os
import sys
from pathlib import Path

mode, target = sys.argv[1], Path(sys.argv[2])
body = "こんにちは、おうどんです🍜\n" + "新しい本文です。\n" * 200 + "## 参考資料\n"

if mode == "direct":                     # 本物のファイルに直接書く
    f = target.open("w", encoding="utf-8")
else:                                    # 一時ファイルに書いてから入れ替える
    f = target.with_name(target.name + ".tmp").open("w", encoding="utf-8")

f.write(body[: len(body) // 2])
f.flush()
os._exit(1)                              # ここで落ちたことにする（入れ替えまで進まない）

# crash_demo.py  直接書く場合と、一時ファイル＋os.replace の場合を比べる
import subprocess
import sys
from pathlib import Path

from check_artifact import check_artifact

old = "こんにちは、おうどんです🍜\n" + "昨日までの本文です。\n" * 100 + "## 参考資料\n"
for mode in ["direct", "replace"]:
    target = Path(f"out/crash/{mode}.md")
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(old, encoding="utf-8")
    tmp = target.with_name(target.name + ".tmp")
    tmp.unlink(missing_ok=True)

    r = subprocess.run([sys.executable, "crash_write.py", mode, str(target)])
    print(f"[{mode}] 終了コード {r.returncode}")
    print("  本物のファイル:", check_artifact(target))
    print("  .tmp が残った？:", tmp.exists())

[direct] 終了コード 1
  本物のファイル: NG: 必要な部分がない ['## 参考資料']
  .tmp が残った？: False
[replace] 終了コード 1
  本物のファイル: OK（3259 バイト）
  .tmp が残った？: True

直接書いたほうは、昨日までの本文が消えて、書きかけだけが残りました。
一時ファイルのほうは、本物のファイルは昨日までの版のまま無事で、書きかけは .tmp の側に残っています。

炊飯器さん「停電したので、半分だけ炊けたご飯を、お茶碗によそっておきました♪」

読者「よそう前に、止まってほしかった」

完成形では、一時ファイルに書いて flush() と os.fsync() をしてから os.replace() で入れ替えます。os.replace() の動きと、Windows での注意は上級者向けの章で書きます。

完成形：段階ごとに記録しながら進める

ここまでの対策を1つにまとめました。
「何日分の仕事か」→ 遅れすぎなら見送り → 成果物がもうあれば中止 → 鍵 → 一時ファイルに書いて入れ替え → 確かめる → 届け先に出す、の順に、段階ごとに記録します。

# daily_job.py  毎朝の作業を、段階ごとに記録しながら進める（説明用の完成形）
import os
import shutil
from datetime import datetime, timedelta
from pathlib import Path

from check_artifact import check_artifact
from runlog import JST, RunLog


def scheduled_for(now, hour=8):
    """この実行は、どの予定（何日の何時）の分か"""
    s = now.replace(hour=hour, minute=0, second=0, microsecond=0)
    if now < s:                      # 予定時刻より前なら、前日の予定の追いかけ
        s -= timedelta(days=1)
    return s


def write_atomic(path, text):
    tmp = path.with_name(path.name + ".tmp")
    with tmp.open("w", encoding="utf-8") as f:
        f.write(text)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)


def main(now, late_limit=timedelta(hours=6)):
    s = scheduled_for(now)
    d = s.date().isoformat()
    log = RunLog("daily-article", d, clock=lambda: now)

    if now - s > late_limit:
        log.stage("skipped_late", scheduled=s.isoformat())
        print(f"{d} 分: 予定より {now - s} 遅れたので見送り")
        return 0

    log.stage("started")
    out_dir = Path("out") / d
    out_dir.mkdir(parents=True, exist_ok=True)
    target = out_dir / "wordpress.md"
    if target.exists():
        found = check_artifact(target)
        reason = "already exists" if found.startswith("OK") else "broken file exists"
        log.stage("failed", reason=reason)
        print(f"{d} 分: もう下書きがある（{found}）ので上書きしません")
        return 1

    lock = out_dir / ".lock"
    try:
        lock.open("x").close()
    except FileExistsError:
        log.stage("failed", reason="lock exists")
        print(f"{d} 分: ほかの実行が作業中か、前回の鍵が残っています")
        return 1
    try:
        body = "こんにちは、おうどんです🍜\n" + "本文です。\n" * 100 + "## 参考資料\n- 資料\n"
        write_atomic(target, body)
        log.stage("artifact_saved", bytes=target.stat().st_size)

        result = check_artifact(target)
        if not result.startswith("OK"):
            log.stage("verify_ng", result=result)
            return 1
        log.stage("verified", result=result)

        pub = Path("published")           # 届け先の例（本番なら push や送信）
        pub.mkdir(exist_ok=True)
        shutil.copy2(target, pub / f"{d}-wordpress.md")
        log.stage("delivered", to=str(pub))
        log.stage("finished")
        print(f"{d} 分: {result}、届け先にコピーしました")
        return 0
    finally:
        lock.unlink()


if __name__ == "__main__":
    import sys
    times = sys.argv[1:] or ["2026-10-22T08:03:00", "2026-10-22T08:05:00", "2026-10-24T23:10:00"]
    for t in times:
        now = datetime.fromisoformat(t).replace(tzinfo=JST)
        print(f"[{t}] 終了コード {main(now)}")

本文を書く部分は、説明用に決まった文字列にしています。実際には、ここでエージェントを呼んで、その出力を write_atomic() に渡す形になります。
届け先も、ここでは別のフォルダへのコピーにしました（本番なら Git の push やメール送信など）。

読者「毎日『本文です。』が100行なら、確かめるのも楽ですね」

楽ですが、それが毎朝届いたら、確かめる前におうどんが泣きます。

時刻を固定して3回呼んだ結果です（10/22 の朝に2回、10/24 の夜に1回）。

2026-10-22 分: OK（1769 バイト）、届け先にコピーしました
[2026-10-22T08:03:00] 終了コード 0
2026-10-22 分: もう下書きがある（OK（1769 バイト））ので上書きしません
[2026-10-22T08:05:00] 終了コード 1
2026-10-24 分: 予定より 15:10:00 遅れたので見送り
[2026-10-24T23:10:00] 終了コード 0

残った記録（runs.jsonl）です。

{"at": "2026-10-22T08:03:00+09:00", "task": "daily-article", "for_date": "2026-10-22", "run_id": "67a0a423", "stage": "started"}
{"at": "2026-10-22T08:03:00+09:00", "task": "daily-article", "for_date": "2026-10-22", "run_id": "67a0a423", "stage": "artifact_saved", "bytes": 1769}
{"at": "2026-10-22T08:03:00+09:00", "task": "daily-article", "for_date": "2026-10-22", "run_id": "67a0a423", "stage": "verified", "result": "OK（1769 バイト）"}
{"at": "2026-10-22T08:03:00+09:00", "task": "daily-article", "for_date": "2026-10-22", "run_id": "67a0a423", "stage": "delivered", "to": "published"}
{"at": "2026-10-22T08:03:00+09:00", "task": "daily-article", "for_date": "2026-10-22", "run_id": "67a0a423", "stage": "finished"}
{"at": "2026-10-22T08:05:00+09:00", "task": "daily-article", "for_date": "2026-10-22", "run_id": "048be51d", "stage": "started"}
{"at": "2026-10-22T08:05:00+09:00", "task": "daily-article", "for_date": "2026-10-22", "run_id": "048be51d", "stage": "failed", "reason": "already exists"}
{"at": "2026-10-24T23:10:00+09:00", "task": "daily-article", "for_date": "2026-10-24", "run_id": "36ff21f3", "stage": "skipped_late", "scheduled": "2026-10-24T08:00:00+09:00"}

これを見回り係に読ませます。

python -X utf8 check_runs.py runs.jsonl 2026-10-25T09:00:00+09:00 2026-10-22 3

2026-10-22  注意: 2 回起動している
2026-10-22  67a0a423  OK
2026-10-22  048be51d  失敗（already exists）
2026-10-23  --------  記録なし（起動していない可能性）
2026-10-24  36ff21f3  見送り（遅れすぎたので実行しなかった）

10/23 は1回も呼んでいないので、「記録なし」。
10/24 は「見送り」と、ちゃんと記録が残っています。同じ「その日の成果物がない」でも、眠っていた日と、わざと見送った日を区別できるのが、段階の記録のいいところです。

炊飯器さん「10/24 は、遅すぎたので炊きませんでした♪」

おうどん「そのメロディなら、聞く価値がある」

運用で使うときのチェックリスト

• 予定の時刻とタイムゾーンを書き出した（スケジューラ・サーバー・スクリプトで同じか）
• 実行ごとに実行 ID を付け、段階ごとに記録している（for_date と at を分けている）
• 成果物の合格条件を決め、エージェントの返事とは別のコードで確かめている
• 一時ファイルに書いてから入れ替えている。残った .tmp を見つけたら調べる
• 同時実行を鍵で、2回目の実行を「成果物があれば中止」で防いでいる
• 遅れすぎた回をどうするか（見送るか、実行するか）を決め、見送りも記録している
• 見回りは別の仕組みで、毎日「あるはずの日」を数えている

上級者向け：完了判定を設計するときの落とし穴

ここから先は上級者向け。読み飛ばしてもOKです。

「成果物があれば中止」は、空ファイルでも止まる

検証中に、先に動かした fake_agent.py の空ファイルを残したまま daily_job.py を動かしてみました。

python -X utf8 run_and_check.py
python -X utf8 daily_job.py 2026-10-22T08:03:00

2026-10-22 分: もう下書きがある（NG: 小さすぎる（0 バイト））ので上書きしません
[2026-10-22T08:03:00] 終了コード 1

（run_and_check.py の出力は、最初の章と同じなので省いています）

二重実行を防ぐための「あれば中止」が、0 バイトのファイル1つで、その日の作業を止めます。
しかも、もし判定が target.exists() だけなら、記録には「すでにある」としか残らず、見た目はただの重複に見えます。

エージェントくん「先輩が、もう場所を取っているので！」

おうどん「その先輩、中身が空っぽなんだよ」

完成形では、すでにあるファイルも check_artifact() にかけ、already exists と broken file exists を分けて記録しました。
壊れたファイルを自動で上書きするかどうかは、あえて入れていません。上書きしてよいかは、壊れた理由（前の回が書きかけで落ちたのか、人が手で直している途中なのか）によるので、人が見て決める場面だと考えたからです（ここはおうどんの設計判断です）。

os.replace は、同じファイルシステムの中で使う

Python の os モジュールのドキュメントでは、os.replace(src, dst) は、dst がファイルなら権限があれば黙って置き換え、src と dst が別のファイルシステムだと失敗することがある、とされています。成功した場合の入れ替えが原子的（途中の状態が見えない、全部か無しか）になるのは、「POSIX の要件」と書かれています。

読者「Windows でも原子的なんですか？」

そこは、このドキュメントの書き方からは断定できません。原子性の根拠として挙げられているのは POSIX（Unix 系の OS の共通規格）の要件で、Windows についての記述はありません。
今回の検証環境（Windows 10）で試したのは「入れ替える前に落ちたら、本物のファイルは昨日の版のまま」という部分だけです。入れ替えの最中に電源が落ちた場合までは確かめていません。

それでも、一時ファイルに書いてから入れ替える形にしておけば、少なくとも「書いている途中で落ちて、書きかけが本物の場所に残る」ことは避けられます。今回の実験で見えたのは、そこまでです。

注意点を並べておきます。

• 一時ファイルは、本物と同じフォルダに作る（別のドライブや別のファイルシステムに作ると、入れ替えが失敗することがある）
• flush() は Python の中の書き込みの溜め（バッファ）を OS に渡すだけです。ドキュメントでも、ディスクへの書き込みまで求めるなら f.flush() のあとに os.fsync(f.fileno()) をする、とされています
• Windows では、os.rename() は dst がすでにあると必ず FileExistsError になります（ドキュメントに明記）。置き換えたいなら os.replace() を使います
• 落ちた回の .tmp は残ります。次の回で黙って消すより、見回りで「.tmp が残っている日」として拾うほうが、落ちた事実に気づけます

炊飯器さん「停電の跡、残しておきました♪」

今回ばかりは、えらい。

鍵は、落ちると残る

open('x') の鍵は、作業中に落ちると返されません。

# stale_lock.py  鍵を取ったまま落ちると、finally は動くか
import os
import subprocess
import sys
from pathlib import Path

lock = Path("out/stale.lock")
lock.parent.mkdir(parents=True, exist_ok=True)
lock.unlink(missing_ok=True)

child = '''
import os
from pathlib import Path
lock = Path("out/stale.lock")
lock.open("x").close()
try:
    os._exit(1)          # 作業中に落ちたことにする
finally:
    lock.unlink()        # 鍵を返すつもりの行
'''
r = subprocess.run([sys.executable, "-c", child])
print("終了コード:", r.returncode)
print("鍵は残った？:", lock.exists())

終了コード: 1
鍵は残った？: True

os._exit() で落ちると、finally の後片付けは動きませんでした。強制終了やパソコンの電源断でも、同じように鍵が残る可能性があります（そちらは試していません）。

見回り係さん「鍵がかかったまま、中には誰もいません」

おうどん「密室だ……」

この記事の鍵は日付ごとのファイルなので、残っても止まるのはその日の分だけです。そのかわり、残った鍵を自動で消すと、本当に作業中の回とぶつかるおそれがあります。
鍵の中に作った時刻やプロセス ID を書いておき、見回りが「古すぎる鍵」を報告して、人が確かめてから消す。これくらいの温度感が、毎日1回の作業にはちょうどいいと思います（おうどんの考えです）。

スケジューラにも、二重起動の止め方と期限がある

鍵を自前で作る前に、スケジューラ側の設定も確かめます。
Kubernetes の CronJob では、次の設定があります。

設定	何をするか
concurrencyPolicy: Forbid	前の Job が終わっていなければ、新しい回をスキップする（既定は Allow で同時実行を許す。Replace は前の回を新しい回に置き換える）
startingDeadlineSeconds	予定の時刻から何秒以内なら、遅れて開始してよいか。過ぎた回はスキップされ、失敗した Job として扱われる。指定しなければ期限なし
（制限）	前回の予定から今までに逃した回が100を超えると、Job を開始せずにエラーを記録する

ただし、ドキュメントは同時に、それでも2回作られることも、1回も作られないこともあると書いています。設定は確率を下げるもので、Job の側の冪等さはやはり要る、というのが公式の立場です。

読者「100回逃したら、もう諦めるんですね」

おうどん「100回目の朝まで、誰も気づかなかったほうが問題かも」

Claude Code のデスクトップ版の定期タスクは、ドキュメントによると、タスクの画面の履歴に、スキップされた回も表示されます。理由は、パソコンが寝ていた、前の回がまだ動いていた、ほかの定期タスクが動いていた、のどれかです。
つまり、スケジューラの側も「前の回が動いていたら次の回をスキップする」をやってくれている。それでも、こちらの記録とスケジューラの履歴は別物なので、見回りではスケジューラの履歴も一緒に見ます。

なお、Kubernetes と Claude Code は今回の検証環境では動かしていません。上の内容は、どちらもドキュメントの記述によるものです。

冪等にできるのは、どの段階か

二重起動を完全に防げない以上、2回動いても困らない形にしておきたい。
ただ、段階によって冪等にしやすさが違います。

段階	2回動くと	冪等にする方法の例
成果物の保存	同じ場所に上書き、または中止	日付で決まる名前、「あれば中止」、一時ファイルと入れ替え
確かめ	2回見るだけ	もともと冪等
届け先への送付	メールが2通、投稿が2回	送る前に送付済みの記録を確かめる、相手側に重複を弾く鍵（冪等キー）を渡す

いちばん冪等にしにくいのは、外に出ていく最後の段階です。
再実行で二重に送らない設計は、AI Agent の再実行で二重処理を起こさない設計の記事で詳しく書きました。

エージェントくん「大事なメールなので、2回送っておきました！」

おうどん「大事なメールほど、1回にして」

見回り係は、どこに置くか

最後に、さっきの「見回り係が寝坊したら？」です。

本体と見回りが同じパソコンの同じスケジューラで動いていると、パソコンが寝ていた日は、両方そろって寝ています。
コンセントが抜けた炊飯器を、同じコンセントにつながったタイマーで見張っている状態です。

見回り係さん「本日も、異常の報告はありません」

おうどん「あなたも寝てたんだよね」

案としては、次のどれかになります（おうどんの考えで、どれが正解というものではありません）。

• 見回りを別の仕組み（別のマシン、クラウドの定期実行など）で動かす
• 毎日「成功の知らせ」を送らせ、知らせが来なかったことに人が気づける形にする（知らせには段階の要約を入れる）
• 人が毎朝見る場所（チャットや一覧のページ）に、見回りの結果を1行ずつ貼る

どれも、「失敗の知らせ」ではなく「知らせがないこと」に気づくための工夫です。

まとめ：メロディのあとに、ふたを開ける

冒頭のエージェントくんに、もう一度報告してもらいます。

エージェントくん「10/22 分、起動 08:03、保存 1769 バイト、確かめて OK、届け先にコピー済み、で完了です！」

おうどん「完了の一言が、急に長くなったね」

エージェントくん「メロディに、歌詞をつけました！」

長いけれど、この歌詞なら、ふたを開ける前から中身がわかります。

炊飯器さん「では私も、これからは中身を量ってから鳴ります♪」

炊飯器が中身を量りはじめたら、それはもう炊飯器というより見回り係です。

完了の返事は、ふたを開けたあとに聞く。

まずは、今動いている定期実行の成果物に、「ある・小さすぎない・必ず入る部分がある」の3つを確かめる行を足してみてください。
ふたを開けてご飯が入っていたら、安心して朝うどんでもどうぞ🍜

参考資料

• Claude Code Docs：Schedule recurring tasks in Claude Code Desktop — ローカルの定期タスクはアプリが開いていてパソコンが起きている間だけ動くこと、寝ていた回はスキップ、7日以内の逃した回のうち最近の1回だけ追いかけて実行すること、9時の予定が夜11時に動く例と時刻の歯止め、権限の確認で止まること、履歴にスキップの理由が出ること
• Kubernetes Documentation：CronJob — 2回作られることも作られないこともあり Job は冪等にすべきこと、concurrencyPolicy、startingDeadlineSeconds、逃した回が100を超えたときの動き
• Python ドキュメント：os（os.replace・os.rename・os.fsync） — os.replace の置き換えと別ファイルシステムでの失敗、原子性が POSIX の要件とされていること、Windows の os.rename が FileExistsError になること、flush のあとに fsync すること
• Python ドキュメント：組み込み関数 open() — ‘x’ モードはファイルがあると失敗する作成で、FileExistsError になること

2026年10月時点の内容です。