Python の subprocess で AI エージェントの CLI をタイムアウト付きで確実に止める実装手順 — timeout 後も孫プロセスが生き残る・PIPE 詰まりでデッドロック・communicate の timeout は子を殺さない、3つのハマりどころ【2026】

はじめに / 対象と前提

AI エージェントの CLI(例:claude -p "..." のようなヘッドレス実行)を Python から起動して、夜間や無人で回している人向けの記事。

エージェントはたまに 固まる。API 応答待ち、許可待ち、子プロセスの無限ループなど原因はいろいろだが、親側でタイムアウトを掛けて止めないと、次の実行が詰まったり、裏で API を叩き続けたりする。

自分も「timeout= を付けたから大丈夫」と思っていたら、例外は出たのにプロセスは生きていた という事故を踏んだ。この記事ではその挙動を手元で再現し、確実に止めるウォッチドッグ(監視して止める係)を実装する。

• 動作確認環境:macOS 15.7 / Python 3.14(3.9 以降なら同じ挙動)
• Linux でも同じ考え方で動く(POSIX 前提。Windows は対象外)

TL;DR

• subprocess.run(..., timeout=) が殺すのは 直接の子だけ。孫プロセスは生き残る
• start_new_session=True で起動して、os.killpg() で プロセスグループごと SIGTERM → 猶予 → SIGKILL
• 出力は PIPE で溜めずにファイルへ流す。Popen.communicate(timeout=) は タイムアウトしても子を殺さない

手順 / 動かし方

完成形のウォッチドッグ

import os, signal, subprocess

def run_with_watchdog(cmd, timeout, grace=10, log_path="agent.log"):
    with open(log_path, "wb") as log:
        proc = subprocess.Popen(
            cmd,
            stdout=log,                # PIPE にしない(後述のデッドロック対策)
            stderr=subprocess.STDOUT,
            stdin=subprocess.DEVNULL,  # 入力待ちで固まるのを防ぐ
            start_new_session=True,    # 新しいプロセスグループのリーダーにする
        )
        try:
            return proc.wait(timeout=timeout)
        except subprocess.TimeoutExpired:
            _kill_group(proc, grace)
            raise

def _kill_group(proc, grace):
    pgid = proc.pid  # start_new_session=True なので pgid == pid
    for sig, wait in ((signal.SIGTERM, grace), (signal.SIGKILL, 5)):
        try:
            os.killpg(pgid, sig)
        except ProcessLookupError:
            return               # グループが既に空
        try:
            proc.wait(timeout=wait)
        except subprocess.TimeoutExpired:
            continue             # リーダーが SIGTERM を無視 → SIGKILL へ
        try:
            os.killpg(pgid, 0)   # シグナル 0 = 生存確認だけ
        except ProcessLookupError:
            return               # 孫も含めて全滅を確認

ポイントは3つ。

1. start_new_session=True で子を 新しいセッション(=新しいプロセスグループ) で起動する。子が作る孫もこのグループに入る
2. タイムアウトしたら os.killpg() でグループ全体にまず SIGTERM。エージェント側に後始末の時間を grace 秒与える
3. それでも残っていたら SIGKILL。リーダーが先に死んでも孫が残っていれば killpg(pgid, 0) で検知して SIGKILL に進む

動作確認

SIGTERM を無視し、さらに孫プロセスを抱えたまま固まる「行儀の悪いエージェント」を bash で作って止めてみる。

import subprocess, time
t = time.time()
try:
    run_with_watchdog(
        ["bash", "-c", "trap '' TERM; sleep 40 & sleep 40; wait"],
        timeout=2, grace=2,
    )
except subprocess.TimeoutExpired:
    print("timeout", round(time.time() - t, 1))
time.sleep(0.3)
print(subprocess.run("pgrep -fl 'sleep 40'", shell=True,
                     capture_output=True, text=True).stdout or "(none)")

timeout 4.3
(none)

タイムアウト 2 秒 + SIGTERM の猶予 2 秒のあと SIGKILL で全滅し、sleep は1つも残っていない。正常終了するコマンドなら run_with_watchdog(["bash", "-c", "echo hi"], timeout=5) が 0 を返す。

ハマりどころ

1. subprocess.run(timeout=) のあとも孫プロセスが生き残る

subprocess.run("sleep 30 & sleep 30; wait", shell=True, timeout=1)

subprocess.TimeoutExpired: Command 'sleep 30 & sleep 30; wait' timed out after 1 seconds
$ pgrep -fl 'sleep 30'
26387 sleep 30
26388 sleep 30

原因:run() はタイムアウト時に process.kill() するが、対象は直接の子(ここでは /bin/sh)だけ。孫の sleep は親を失って孤児になり、そのまま走り続ける。shell=True でなくても、エージェント CLI 自体が MCP サーバーやツール実行用の子プロセスを起動するので同じことが起きる。

回避策:start_new_session=True + os.killpg() でグループごと止める(上の完成形)。TimeoutExpired が出た = 止まった、ではない。

2. stdout=PIPE + wait() でデッドロックする

p = subprocess.Popen(["python3", "-c", "import sys; sys.stdout.write('x'*200000)"],
                     stdout=subprocess.PIPE)
p.wait(timeout=3)   # → TimeoutExpired、poll() は None のまま

原因:パイプには OS のバッファ(数十 KB 程度)しかない。親が読まずに wait() すると、子はバッファが空くのを待って書き込みでブロックし、親は子の終了を待つ。お互い待ちで止まる。エージェントのログは長くなりがちなので、普段は動いて 長い出力のときだけ固まる のが厄介。公式ドキュメントにも wait() + PIPE の警告がある。

回避策:

• 出力を後で読めればいいなら、ファイルに直接流す(完成形の stdout=log)
• メモリに欲しいなら communicate() を使う(内部で読みながら待つので詰まらない)。ただし次のハマりどころに注意

3. communicate(timeout=) はタイムアウトしても子を殺さない

p = subprocess.Popen(["sleep", "30"], stdout=subprocess.PIPE)
try:
    p.communicate(timeout=1)
except subprocess.TimeoutExpired:
    print(p.poll())   # → None(まだ生きている)

原因:Popen.communicate() の timeout は「待つのをやめる」だけ。公式ドキュメントにも「タイムアウトしても子プロセスは kill されないので、自分で kill して communicate し直すこと」と書かれている。run() が kill してくれるのは run() の中だけの話。

回避策:except の中で必ず kill し、もう一度 communicate() してパイプを回収する。

try:
    out, _ = p.communicate(timeout=60)
except subprocess.TimeoutExpired:
    os.killpg(p.pid, signal.SIGKILL)  # start_new_session=True で起動している前提
    out, _ = p.communicate()          # 残りの出力回収とゾンビ回収

背景・補足

• なぜ SIGTERM から送るのか:いきなり SIGKILL すると、エージェントが書きかけの状態ファイルやロックファイルを残す。まず SIGTERM で後始末の機会を与え、駄目なら SIGKILL、の二段構えが無難
• start_new_session の副作用:子は端末から切り離されるので、手元で Ctrl+C しても子には届かない。対話的に使うときは親側で KeyboardInterrupt を捕まえて _kill_group() を呼ぶ
• タイムアウト値の決め方:自分は「普段の所要時間の 3〜5 倍」にしている。短すぎると正常な長考まで殺してしまう

まとめ

• timeout= で例外が出ても、孫プロセスは止まっていない ことがある
• start_new_session=True + os.killpg() でプロセスグループごと SIGTERM → SIGKILL
• 出力は PIPE + wait() にしない。ファイルに流すか communicate() を使う
• communicate(timeout=) の後は 自分で kill して回収 する
• 止めたあとは pgrep で本当に消えたか一度確認しておくと安心