コネクタに書き込みツールが増えた夜、無人ジョブは黙って何かを送れるようになる

# surface_diff.py
import json, hashlib
from pathlib import Path
from classify_tools import classify, Risk
 
SNAP = Path.home() / ".connector_surface" / "m365.json"
 
def current_surface(list_tools) -> dict:
    """list_tools(): コネクタのツール定義配列を返す関数を注入する"""
    surface = {}
    for t in list_tools():
        name = t["name"]
        surface[name] = {
            "risk": int(classify(name)),
            # 説明の変化も検知したいのでハッシュだけ持つ
            "desc_hash": hashlib.sha256(
                t.get("description", "").encode("utf-8")
            ).hexdigest()[:12],
        }
    return surface
 
def load_snapshot() -> dict:
    if SNAP.exists():
        return json.loads(SNAP.read_text())
    return {}
 
def diff(prev: dict, now: dict) -> dict:
    added = {k: now[k] for k in now.keys() - prev.keys()}
    removed = {k: prev[k] for k in prev.keys() - now.keys()}
    changed = {
        k: {"from": prev[k], "to": now[k]}
        for k in prev.keys() & now.keys()
        if prev[k] != now[k]  # 危険度や説明ハッシュが変わった
    }
    return {"added": added, "removed": removed, "changed": changed}
 
def save_snapshot(now: dict) -> None:
    SNAP.parent.mkdir(parents=True, exist_ok=True)
    SNAP.write_text(json.dumps(now, ensure_ascii=False, indent=2))

# preflight_gate.py
import json, sys
from pathlib import Path
from surface_diff import current_surface, load_snapshot, diff, save_snapshot
from classify_tools import Risk
 
APPROVED = Path.home() / ".connector_surface" / "approved_tools.json"
 
def approved() -> set:
    if APPROVED.exists():
        return set(json.loads(APPROVED.read_text()))
    return set()
 
def preflight(list_tools) -> None:
    now = current_surface(list_tools)
    d = diff(load_snapshot(), now)
    ok = approved()
 
    # 新規 or 危険度が上がった、書き込み以上のツールを洗い出す
    suspects = []
    for name, meta in d["added"].items():
        if meta["risk"] >= Risk.WRITE:
            suspects.append((name, meta["risk"]))
    for name, chg in d["changed"].items():
        if chg["to"]["risk"] >= Risk.WRITE and chg["to"]["risk"] > chg["from"]["risk"]:
            suspects.append((name, chg["to"]["risk"]))
 
    unapproved = [(n, r) for (n, r) in suspects if n not in ok]
    if unapproved:
        lines = [f"  - {n} (risk={Risk(r).name})" for n, r in unapproved]
        sys.stderr.write(
            "未承認の書き込みツールを検知しました。実行を中止します:\n"
            + "\n".join(lines)
            + f"\n承認するには {APPROVED} に名前を追記してください。\n"
        )
        sys.exit(3)  # ジョブ本体に入る前に落とす
 
    # 承認済みのものだけならスナップショットを更新して通す
    save_snapshot(now)

# dry_run_ledger.py
import json, time
from pathlib import Path
from classify_tools import classify, Risk
 
LEDGER = Path.home() / ".connector_surface" / "write_intents.jsonl"
DRY_RUN = True  # 慣らし期間は True。信頼できたら False へ
 
def guarded_call(tool_name: str, args: dict, real_call):
    risk = classify(tool_name)
    if risk >= Risk.WRITE and DRY_RUN:
        record = {
            "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
            "tool": tool_name,
            "risk": Risk(risk).name,
            "args": args,          # 送るつもりだった中身をそのまま保存
            "executed": False,
        }
        with LEDGER.open("a", encoding="utf-8") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")
        return {"dry_run": True, "logged": tool_name}
    # READ か、ドライランを解除した書き込みだけが本当に実行される
    return real_call(tool_name, args)