完走したジョブを拾えない停滞を検知・再開する方法――31時間の実測

完走した修正ジョブが12時54分に残り、製品リポジトリへ取り込まれないまま約21時間が過ぎました。壊れていたのは実装レーンではなく、「終わった」と気づく主体でした。

長時間ジョブを任せるホームラボ運用者は、通知を受けてから手で再開依頼を運ぶ仕組みを減らせます。ここでは、成果物の進みと進捗ファイルの鮮度から停滞を判定し、検知後に非同期の再開便を投入する形までを、固定ログの再生で確認します。

先に結論を置くと、プロセスの生存確認だけでは「完走したのに拾われない」停滞を捕まえられません。完走後の取り込み待ちと、進捗表示が古くなった状態は、別の閾値で見る必要がありました。

完走した仕事が約21時間残り、通知だけの監視では止まりました

2026年8月5日、実装レーンの修正ジョブは12時54分に完走していました。しかし成果物は製品リポジトリに取り込まれず、約21時間残りました。初版の監視は翌8月6日に設置しましたが、停滞を通知するだけで停止していました。再開は所長からセッションへの伝言に依存しており、監視が見つけた問題を監視自身では閉じられなかったのです。

私も当初は、通知が届くことを運用上の完了と扱っていました。実際には、通知の後に誰かが内容を読み、再開依頼を作り、渡す工程が残ります。そこに約21時間の空白を作りました。

長時間ジョブが無言で止まる問題については、先行する進捗ゼロを異常として鳴らす監視で扱いました。今回は、その一段後ろにある「終わったのに拾われない」状態が対象です。

プロセスではなく、成果物が進んだかで停滞を分けました

通常、死活監視ではプロセスを見るのが合理的です。ファイルの存在だけを生存証明にしない理由は、死活監視は「プロセス」を見よで整理しています。ただし、プロセスが消えたこと自体が正常な完走である実装レーンでは、逆向きの観測面が要ります。

そこで、ワーカーログ、リポジトリのHEAD、進捗ファイルの更新時刻を組み合わせました。進捗の値と鮮度を別々に見ない考え方は、値ではなく「値と鮮度の組」を監視するとも接続します。

90分は、実測した取り込み所要20〜40分の倍として置きました。8時間は、既存の48時間警告より先に鮮度低下を拾うための値です。両者は同じ「遅い」ではありません。前者は完走済み成果の未回収、後者は進行中表示の更新停止です。

判定の流れは次のとおりです。

図1で重要なのは、停止ファイルとstate: idleを異常にしない点です。意図的に止めたレーンへ再開便を送らないため、停止と停滞を同じ信号にしません。

通知で止めず、検知から再開便の投入までを同じ処理へつなげました

2026年8月8日、通知だけ来て手で再開依頼を運ぶ状態を改め、停滞検知から非同期の再開便投入までを同じ処理につなげました。効果の範囲は、観測、状態保存、再開便の投入、task branchの作成までです。本流への統合、公開、金銭的な処理はこの道具の効果に含めていません。

設置時は20分ごとのcronでした。2026年8月18日の監査後は、毎時の定常点検に加え、観測材料が更新された場合だけ10分刻みの追加点検を行う形へ縮退しています。常駐プロセスはありません。

固定入力を再生すると、31時間の進捗停止を同じJSONで判定できます

ここでは外部stateを書き換えず、2026年8月8日10時40分01秒（JST）の記録を固定入力として再生します。必要なのはPython 3の標準ライブラリだけで、ネットワーク接続や外部パッケージは使いません。実行前に、次の内容をe1-stall-check.pyとして保存してください。

#!/usr/bin/env python3
"""Replay the output-based stall decision with an offline fixed sample."""

from __future__ import annotations

import argparse
import json
from datetime import datetime, timedelta, timezone


JST = timezone(timedelta(hours=9))
UNPICKED_JOB_MINUTES = 90
STATUS_QUIET_HOURS = 8


def parse_time(value: str) -> datetime:
    parsed = datetime.fromisoformat(value)
    return parsed if parsed.tzinfo else parsed.replace(tzinfo=JST)


def evaluate(
    *,
    checked_at: str,
    status_updated_at: str,
    job_finished_at: str | None,
    repo_commit_at: str | None,
    stopped_on_purpose: bool = False,
    lane_idle: bool = False,
) -> dict[str, object]:
    """Return the same public verdict shape as the production lane."""
    now = parse_time(checked_at)
    if stopped_on_purpose:
        return {"product": "sashiko-repair", "stalled": False,
                "reason": "stopped-on-purpose"}
    if lane_idle:
        return {"product": "sashiko-repair", "stalled": False,
                "reason": "lane-idle"}

    if job_finished_at is not None:
        finished = parse_time(job_finished_at)
        idle_minutes = (now - finished).total_seconds() / 60
        moved_after = (
            repo_commit_at is not None and parse_time(repo_commit_at) > finished
        )
        if idle_minutes >= UNPICKED_JOB_MINUTES and not moved_after:
            return {
                "product": "sashiko-repair",
                "stalled": True,
                "kind": "job-unpicked",
                "detail": (
                    f"作業は{int(idle_minutes // 60)}時間"
                    f"{int(idle_minutes % 60)}分前に終わっていますが、"
                    "その成果がまだ取り込まれていません"
                ),
                "finished_at": finished.isoformat(),
            }

    updated = parse_time(status_updated_at)
    quiet_hours = (now - updated).total_seconds() / 3600
    if quiet_hours >= STATUS_QUIET_HOURS:
        return {
            "product": "sashiko-repair",
            "stalled": True,
            "kind": "status-quiet",
            "detail": f"進捗の更新が{int(quiet_hours)}時間止まっています",
            "updated_at": updated.isoformat(),
        }
    return {"product": "sashiko-repair", "stalled": False,
            "reason": "moving"}


SAMPLE = {
    "checked_at": "2026-08-08T10:40:01.288710+09:00",
    "status_updated_at": "2026-08-07T03:10:00+09:00",
    "job_finished_at": None,
    "repo_commit_at": None,
}


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--sample", action="store_true",
                        help="run the fixed offline production-log sample")
    args = parser.parse_args()
    if not args.sample:
        parser.error("--sample is required")
    record = {
        "schema": "impl-lane-watch/v1",
        "checked_at": SAMPLE["checked_at"],
        "results": [evaluate(**SAMPLE)],
    }
    print(json.dumps(record, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

固定入力なので、実行時刻ではなく記録されたchecked_atとupdated_atを比較します。

$ PYTHONDONTWRITEBYTECODE=1 python3 e1-stall-check.py --sample

{"schema": "impl-lane-watch/v1", "checked_at": "2026-08-08T10:40:01.288710+09:00", "results": [{"product": "sashiko-repair", "stalled": true, "kind": "status-quiet", "detail": "進捗の更新が31時間止まっています", "updated_at": "2026-08-07T03:10:00+09:00"}]}

終了コードは0でした。このstdoutは、当時の本番JSONLから採取した同時刻の選択レコードと、改行を除いて一致しました。図2は、出力が何を示すかを読みやすく整理したものです。

図2の31時間は「少し遅い」ではなく、進捗の鮮度が8時間閾値を大きく越えた状態です。この記録を起点に再開便が投入され、非同期タスクは完了状態まで追跡できました。

本番スクリプトを再生用に使うと、stateを変えてしまいます

最小再現では製品版を直接実行しません。本番スクリプトは外部stateの更新や再開便の投入を行い得るためです。上のstandaloneは判定と公開出力形だけを固定入力で再生します。

また、設置時の「20分ごと」を現在の実行条件として読まないでください。現在は毎時の定常点検と、観測材料が変わった場合の追加点検です。監視頻度と判定閾値を混同すると、90分と8時間の意味も崩れます。

現在は一つの実装レーンを見ており、次は拾われない完走を増やして見つけます

現物はACTIVEのまま、実装中のアプリ1本を対象に動いています。対象を並べる構造は複数製品を扱える形ですが、複数レーンでの効果はまだ測っていません。

長時間ジョブを増やす前に、まず自分の運用で「完走後に成果物がどこへ渡るか」と「誰が拾うか」を一行で書き出してください。後者が人名だけなら、通知の先に再開処理をつなぐ余地があります。