接続が切れた瞬間、その投稿は通ったのか — 無人パイプラインで MCP 書き込みを二重実行せずにやり直す

import json, os, time, uuid, tempfile
 
LEDGER = os.path.expanduser("~/.cache/mcp_ledger.jsonl")
 
def _append(rec: dict) -> None:
    # 追記は fsync まで含めて確実に永続化する（クラッシュ復旧の前提）
    os.makedirs(os.path.dirname(LEDGER), exist_ok=True)
    with open(LEDGER, "a", encoding="utf-8") as f:
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
        f.flush()
        os.fsync(f.fileno())
 
def begin_operation(action: str, args: dict) -> str:
    op_id = str(uuid.uuid4())
    # 呼び出しの「前」に pending を記録しておく
    _append({"op_id": op_id, "action": action, "args": args,
             "status": "pending", "ts": time.time()})
    return op_id
 
def mark(op_id: str, status: str) -> None:
    _append({"op_id": op_id, "status": status, "ts": time.time()})

TRANSIENT = (TimeoutError, ConnectionError, ConnectionResetError)
 
def call_with_idempotency(client, action: str, args: dict):
    op_id = begin_operation(action, args)
    payload = {**args, "idempotency_key": op_id}
    try:
        result = client.call_tool(action, payload)   # MCP ツール呼び出し
        mark(op_id, "committed")
        return result
    except TRANSIENT:
        # 通信が切れた — 実行されたかは不明。ここでは再送しない
        mark(op_id, "uncertain")
        raise
    except Exception:
        # サーバーが明確に拒否した — 安全に失敗扱いにできる
        mark(op_id, "failed")
        raise

def reconcile(client, op_id: str, action: str, args: dict) -> bool:
    """その操作の効果がサーバー側に存在するかを確かめる。
    存在すれば True（再送不要）、なければ False（再送してよい）。"""
    if action == "post_status":
        # 相関トークンを含む自分の投稿を検索する
        found = client.call_tool("search_own_posts", {"contains": op_id})
        return len(found.get("items", [])) > 0
    if action == "create_issue":
        found = client.call_tool("search_issues", {"external_id": op_id})
        return len(found.get("items", [])) > 0
    # 照合できないアクションは「確かめられない」を呼び出し側へ返す
    raise LookupError(f"no reconcile strategy for action={action}")
 
def recover_uncertain(client) -> None:
    """前回までに uncertain で終わった操作を点検し、安全に決着させる。"""
    for op_id, action, args in load_uncertain(LEDGER):
        # 相関トークンを書き込み内容へ必ず混ぜておくのが前提
        args = {**args, "correlation_token": op_id}
        try:
            if reconcile(client, op_id, action, args):
                mark(op_id, "committed")           # 実は通っていた
            else:
                client.call_tool(action, args)     # 通っていなかった → 再送
                mark(op_id, "committed")
        except LookupError:
            # 照合できないアクションは自動で再送しない（人手の確認へ回す）
            mark(op_id, "needs_review")