#!/usr/bin/env python3 """依頼箱の見張り係（inbox_runner） 依頼箱/新規/ に置かれた依頼（.md）を1件ずつ Claude Code（claude -p）に渡し、 Claude Code の報告を 依頼箱/結果/ に書き戻す。 使い方: python3 inbox_runner.py 見張りを続ける（ターミナルの窓を開けたままにする） python3 inbox_runner.py --once 今ある依頼を処理したら終わる python3 inbox_runner.py --check 設定・claude の場所と版を確かめるだけ 止め方: ターミナルで Ctrl+C。 一時停止は 依頼箱/ に「止める」という名前のファイルを置く（消すと再開）。 """ import argparse import datetime as dt import fcntl import json import os import re import shutil import signal import subprocess import sys import tempfile import time from pathlib import Path HERE = Path(__file__).resolve().parent CONFIG = HERE / "config.json" SETTINGS = HERE / "settings.json" FOLDERS = ["新規", "確認待ち", "実行中", "済み", "失敗", "結果"] PREAMBLE = """これは「依頼箱」から自動で渡された依頼です。人はこの画面を見ていません。質問はできません。 守ること: - 依頼の範囲だけを行う。範囲の外は変えない。 - エイイチが決めること（方針・数値の決め方・公開・実機を動かすこと）に当たったら、そこで手を止め、最後の報告の「決めてほしいこと」に書く。推測で決めて進めない。 - プロジェクトの決まり（MASTER・CLAUDE.md）に従う。git は個別にファイルを指定して add する。git clean は使わない。 - 依頼文で報告ファイルの置き場所が指定されていれば、そこにも書く。 - 最後の返答は日本語の報告にする。英語の文を混ぜない。形は次のとおり: やったこと／結果（数字があれば数字）／変えたファイル／決めてほしいこと（なければ「なし」） 依頼ファイル: {name} ----- ここから依頼 ----- """ def now(): return dt.datetime.now() def stamp(): return now().strftime("%Y%m%d_%H%M") def load_config(): with open(CONFIG, encoding="utf-8") as f: cfg = json.load(f) cfg["inbox"] = Path(os.path.expanduser(cfg["inbox"])) cfg["allowed_dirs"] = [Path(os.path.expanduser(p)).resolve() for p in cfg["allowed_dirs"]] return cfg def log(cfg, msg): line = f"{now().isoformat(timespec='seconds')} {msg}" print(line, flush=True) try: with open(cfg["inbox"] / "記録.log", "a", encoding="utf-8") as f: f.write(line + "\n") except OSError as e: print(f"記録.log に書けません: {e}", flush=True) def notify(title, msg): """Mac の通知を出す（失敗しても止めない）""" if sys.platform != "darwin": return msg = msg.replace("\n", " ") script = (f"display notification {json.dumps(msg, ensure_ascii=False)} " f"with title {json.dumps(title, ensure_ascii=False)}") try: subprocess.run(["osascript", "-e", script], capture_output=True, timeout=10) except Exception: pass HEADER_RE = re.compile(r"\A---[ \t]*\n(.*?)\n---[ \t]*\n", re.S) def parse_request(text): """頭の --- で囲んだ部分を読む。例: 作業フォルダ: /Users/... 上限時間: 120""" text = text.lstrip("").replace("\r\n", "\n") meta, body = {}, text m = HEADER_RE.match(text) if m: for line in m.group(1).splitlines(): parts = re.split(r"[:：]", line, maxsplit=1) if len(parts) == 2: meta[parts[0].strip()] = parts[1].strip() body = text[m.end():] return meta, body def is_allowed(cfg, workdir): for base in cfg["allowed_dirs"]: if workdir == base or base in workdir.parents: return True return False def pending(cfg): d = cfg["inbox"] / "新規" items = [] for p in sorted(d.iterdir()): n = p.name if n.startswith(".") and n.endswith(".icloud"): # iCloud に退避されたファイルは取り寄せてから次の回で拾う real = d / n[1:-len(".icloud")] subprocess.run(["brctl", "download", str(real)], capture_output=True) continue if n.startswith(".") or not n.endswith(".md"): continue if time.time() - p.stat().st_mtime < cfg.get("settle_seconds", 15): continue # 書き込み中かもしれない items.append(p) return items def move(src, folder): dst = src.parent.parent / folder / src.name if dst.exists(): dst = dst.with_name(f"{dst.stem}_{stamp()}{dst.suffix}") shutil.move(str(src), str(dst)) return dst def child_env(cfg): """Claude Code に渡す環境。APIキーを外してプランの利用枠で動かす（config.json の drop_env / set_env）""" env = dict(os.environ) for k in cfg.get("drop_env", []): env.pop(k, None) env.update({k: str(v) for k, v in cfg.get("set_env", {}).items()}) return env def run_claude(cfg, prompt, workdir, limit_min): cmd = [cfg.get("claude", "claude"), "-p", prompt, "--output-format", "json", "--permission-mode", cfg["permission_mode"], "--permission-prompts", "none", "--settings", str(SETTINGS)] + list(cfg.get("extra_args", [])) # 出力はファイルで受ける（Claude Code が残した子プロセスが管をふさいで待ち続けるのを防ぐ） with tempfile.TemporaryFile("w+", encoding="utf-8") as fo, \ tempfile.TemporaryFile("w+", encoding="utf-8") as fe: p = subprocess.Popen(cmd, cwd=str(workdir), stdout=fo, stderr=fe, start_new_session=True, env=child_env(cfg)) timed_out = False try: p.wait(timeout=limit_min * 60) except subprocess.TimeoutExpired: timed_out = True # 先に SIGINT（その回の作業を終わらせる）→ だめなら SIGTERM → 最後に SIGKILL for sig, wait in ((signal.SIGINT, 120), (signal.SIGTERM, 60), (signal.SIGKILL, 30)): try: os.killpg(p.pid, sig) except ProcessLookupError: pass try: p.wait(timeout=wait) break except subprocess.TimeoutExpired: continue fo.seek(0) fe.seek(0) return p.returncode, fo.read(), fe.read(), timed_out def write_report(cfg, req, meta, workdir, started, rc, out, err, timed_out, note=""): res_dir = cfg["inbox"] / "結果" base = f"{req.stem}_{started.strftime('%Y%m%d_%H%M')}" data = None try: data = json.loads(out) if out.strip() else None except json.JSONDecodeError: data = None if out.strip(): (res_dir / f"{base}.json").write_text(out, encoding="utf-8") if err.strip(): (res_dir / f"{base}_stderr.txt").write_text(err, encoding="utf-8") is_error = bool(data.get("is_error")) if isinstance(data, dict) else True if timed_out: state = "時間切れ（途中で止めました）" elif note: state = "失敗" elif rc == 0 and not is_error: state = "完了" else: state = "失敗" ended = now() mins = (ended - started).total_seconds() / 60 lines = [f"# 結果：{req.stem}", "", f"- 状態：{state}", f"- 始めた時刻：{started.strftime('%Y-%m-%d %H:%M')}", f"- 終わった時刻：{ended.strftime('%Y-%m-%d %H:%M')}（{mins:.0f}分）", f"- 作業フォルダ：{workdir}"] if isinstance(data, dict) and data.get("session_id"): lines.append(f"- この続きを Claude Code で話すとき：作業フォルダで `claude --resume {data['session_id']}`") denials = data.get("permission_denials") if isinstance(data, dict) else None if denials: lines.append(f"- 止められた操作：{len(denials)}件（いちばん下に一覧）") if "ANTHROPIC_API_KEY" in err or "auth source is set" in err: lines.append("- ★ APIキーで動いた可能性があります（プランの利用枠ではなく従量課金）。_stderr.txt を見てください") lines += ["", "## Claude Code の報告", ""] if note: lines.append(note) elif isinstance(data, dict) and data.get("result"): lines.append(str(data["result"])) else: lines.append("（報告がありません。終了コード " + str(rc) + "。_stderr.txt を見てください。ログインが切れているときは、ターミナルで claude を開いて /login）") if denials: lines += ["", "## 止められた操作", ""] for d in denials: lines.append("- " + json.dumps(d, ensure_ascii=False)[:500]) path = res_dir / f"{base}.md" path.write_text("\n".join(lines) + "\n", encoding="utf-8") return state, path def handle(cfg, req): started = now() req = move(req, "実行中") log(cfg, f"開始 {req.name}") meta, body = parse_request(req.read_text(encoding="utf-8")) wd_text = meta.get("作業フォルダ", "") workdir = Path(os.path.expanduser(wd_text)).resolve() if wd_text else None note = "" if not workdir: note = "依頼の頭に「作業フォルダ:」がありません。実行していません。" elif not workdir.is_dir(): note = f"作業フォルダが見つかりません：{workdir}。実行していません。" elif not is_allowed(cfg, workdir): note = (f"作業フォルダ {workdir} は自動で回してよい場所（config.json の allowed_dirs）に入っていません。" "実行していません。手で頼むか、allowed_dirs に足すかを決めてください。") try: limit = int(re.sub(r"[^0-9]", "", meta.get("上限時間", "")) or cfg["default_limit_minutes"]) except ValueError: limit = cfg["default_limit_minutes"] if note: rc, out, err, timed_out = -1, "", "", False else: prompt = PREAMBLE.format(name=req.name) + body try: rc, out, err, timed_out = run_claude(cfg, prompt, workdir, limit) except FileNotFoundError: rc, out, err, timed_out = -1, "", "", False note = "claude が見つかりません。config.json の claude に絶対パスを書いてください。" state, report = write_report(cfg, req, meta, workdir, started, rc, out, err, timed_out, note) move(req, "済み" if state == "完了" else "失敗") log(cfg, f"{state} {req.name} → 結果/{report.name}") notify("依頼箱", f"{state}：{req.stem}") def check(cfg): ok = True print("依頼箱:", cfg["inbox"], "あり" if cfg["inbox"].is_dir() else "★ ない") ok &= cfg["inbox"].is_dir() for d in cfg["allowed_dirs"]: print("自動で回してよい場所:", d, "あり" if d.is_dir() else "★ ない") exe = shutil.which(cfg.get("claude", "claude")) print("claude:", exe or "★ 見つからない") if exe: v = subprocess.run([exe, "--version"], capture_output=True, text=True) print("claude の版:", (v.stdout or v.stderr).strip()) else: ok = False print("承認の扱い:", cfg["permission_mode"], "／ 拒否の決まり:", SETTINGS) print("外す環境変数:", ", ".join(cfg.get("drop_env", [])) or "なし", "／ うち今この窓で設定されているもの:", ", ".join(k for k in cfg.get("drop_env", []) if k in os.environ) or "なし") return ok def main(): ap = argparse.ArgumentParser() ap.add_argument("--once", action="store_true") ap.add_argument("--check", action="store_true") a = ap.parse_args() cfg = load_config() if a.check: sys.exit(0 if check(cfg) else 1) for f in FOLDERS: (cfg["inbox"] / f).mkdir(parents=True, exist_ok=True) lock = open(HERE / ".runner.lock", "w") try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: print("見張り係はもう動いています（2つ目は起動しません）") sys.exit(1) if sys.platform == "darwin" and not a.once: # 見張っている間は Mac をスリープさせない（この係が終わると解除） subprocess.Popen(["caffeinate", "-i", "-w", str(os.getpid())]) log(cfg, "見張り係を起動" + ("（1回だけ）" if a.once else "")) # 前回、作業の途中で止まった依頼（Mac の再起動など）は「失敗」へ回す。 # 途中まで変わったファイルがあるかもしれないので、自動ではやり直さない。 for stale in sorted((cfg["inbox"] / "実行中").glob("*.md")): write_report(cfg, stale, {}, "不明", now(), -1, "", "", False, note="見張り係が止まった（Mac の再起動など）ため、この依頼は途中で終わっています。" "途中まで変わったファイルがあるかもしれません。やり直すかどうかは、確かめてから決めます。") move(stale, "失敗") log(cfg, f"中断されていた依頼を失敗へ {stale.name}") notify("依頼箱", f"中断されていた依頼があります：{stale.stem}") paused_logged = False try: while True: if (cfg["inbox"] / "止める").exists(): if not paused_logged: log(cfg, "「止める」があるので一時停止中") paused_logged = True else: paused_logged = False for req in pending(cfg): if (cfg["inbox"] / "止める").exists(): break try: handle(cfg, req) except Exception as e: # 1件の失敗で見張りを止めない log(cfg, f"★ {req.name} の処理中に失敗: {e!r}") if a.once: break time.sleep(cfg.get("poll_seconds", 30)) except KeyboardInterrupt: log(cfg, "見張り係を止めました") if __name__ == "__main__": main()