Gemini CLIのhooksで使っているスクリプトが動かなくなったので原因調査・改修してみた

#!/usr/bin/env python3
"""
Gemini CLI の会話履歴を Obsidian に自動保存するフックスクリプト。

対応イベント:
  - SessionStart : Obsidian に Markdown ファイルを作成
  - AfterAgent   : transcript_path を読んで差分を追記
"""

import json
import sys
from datetime import datetime, timezone, timedelta
from pathlib import Path

# ============================================================
# 定数
# ============================================================
JST = timezone(timedelta(hours=9))
OBSIDIAN_VAULT = Path.home() / "Documents" / "obsidian_保管庫"
OUTPUT_BASE = OBSIDIAN_VAULT / "gemini-cli"
STATE_FILE = Path.home() / ".gemini" / "obsidian_state.json"
LOG_FILE = Path.home() / ".gemini" / "logs" / "obsidian_save.log"


# ============================================================
# ユーティリティ
# ============================================================
def log(msg: str) -> None:
    print(msg, file=sys.stderr)
    try:
        LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
        with open(LOG_FILE, "a", encoding="utf-8") as f:
            ts = datetime.now(JST).strftime("%Y-%m-%d %H:%M:%S")
            f.write(f"[{ts}] {msg}\n")
    except Exception:
        pass


def load_state() -> dict:
    if STATE_FILE.exists():
        try:
            return json.loads(STATE_FILE.read_text(encoding="utf-8"))
        except Exception:
            return {}
    return {}


def save_state(state: dict) -> None:
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    STATE_FILE.write_text(
        json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8"
    )


def read_stdin() -> dict:
    try:
        data = sys.stdin.read()
        if data.strip():
            return json.loads(data)
    except Exception:
        pass
    return {}


def now_jst() -> datetime:
    return datetime.now(JST)


def parse_timestamp(ts_str: str) -> datetime:
    try:
        return datetime.fromisoformat(ts_str.replace("Z", "+00:00")).astimezone(JST)
    except Exception:
        return now_jst()


# ============================================================
# テキスト抽出
# ============================================================
def extract_text(content) -> str:
    """contentからテキストを抽出する（str / list 両対応）"""
    if isinstance(content, str):
        return content.strip()
    if isinstance(content, list):
        parts = []
        for block in content:
            if isinstance(block, dict) and "text" in block:
                parts.append(block["text"].strip())
        return "\n".join(parts)
    return ""


# ============================================================
# JSONL パース
# ============================================================
def load_jsonl(transcript_path: str, start_line: int = 0) -> tuple[list[dict], int]:
    """start_line 以降を読み込み、メッセージリストと次回開始行を返す"""
    messages_map: dict[str, dict] = {}  # id -> message
    insert_order: list[str] = []
    next_line = start_line

    path = Path(transcript_path)
    if not path.exists():
        log(f"load_jsonl: ファイルが存在しません: {transcript_path}")
        return [], start_line

    try:
        with open(path, encoding="utf-8") as f:
            for i, line in enumerate(f):
                if i < start_line:
                    continue

                stripped = line.strip()
                if not stripped:
                    next_line = i + 1
                    continue

                try:
                    record = json.loads(stripped)
                except json.JSONDecodeError:
                    log(f"load_jsonl: 不正なJSON行をスキップ (line {i})")
                    continue  # 壊れた行は next_line を進めない

                next_line = i + 1  # パース成功後に更新

                record_type = record.get("type", "")
                record_id = record.get("id", "")

                if record_type == "metadata":
                    continue

                # 指定 ID 以降を削除（会話の巻き戻し）
                if "$rewindTo" in record:
                    rewind_id = record["$rewindTo"]
                    if rewind_id in insert_order:
                        idx = insert_order.index(rewind_id)
                        for removed_id in insert_order[idx + 1:]:
                            messages_map.pop(removed_id, None)
                        insert_order = insert_order[:idx + 1]
                    continue

                # 既存メッセージの上書き更新（ストリーミング途中）
                if "$set" in record and record_id:
                    if record_id in messages_map:
                        messages_map[record_id].update(record["$set"])
                    else:
                        log(f"load_jsonl: $set 対象が範囲外のためスキップ (id={record_id})")
                    continue

                if record_id and record_type in ("user", "gemini"):
                    if record_id not in messages_map:
                        insert_order.append(record_id)
                    messages_map[record_id] = record

    except Exception as e:
        log(f"load_jsonl: 読み込みエラー: {e}")
        return [], start_line

    messages = [messages_map[mid] for mid in insert_order if mid in messages_map]
    return messages, next_line


# ============================================================
# ファイル生成
# ============================================================
def build_frontmatter(session_id: str, cwd: str, start_dt: datetime) -> str:
    return (
        "---\n"
        f"session_id: {session_id}\n"
        f"project: {cwd}\n"
        f"time_start: {start_dt.isoformat()}\n"
        "tags:\n"
        "  - gemini-cli\n"
        "---\n\n"
    )


def build_header(session_id: str, cwd: str, start_dt: datetime) -> str:
    return (
        "# Gemini CLI セッション\n\n"
        "| Key | Value |\n"
        "|---|---|\n"
        f"| Session | `{session_id[:8]}` |\n"
        f"| Project | `{cwd}` |\n"
        f"| Started | {start_dt.strftime('%Y-%m-%d %H:%M:%S')} |\n\n"
        "---\n\n"
        "## 会話ログ\n\n"
    )


def create_md_file(session_id: str, cwd: str, start_dt: datetime) -> Path:
    date_dir = start_dt.strftime("%Y%m%d")
    time_prefix = start_dt.strftime("%H%M%S")
    short_id = session_id[:8]

    output_dir = OUTPUT_BASE / date_dir
    output_dir.mkdir(parents=True, exist_ok=True)

    path = output_dir / f"{time_prefix}_{short_id}.md"
    path.write_text(
        build_frontmatter(session_id, cwd, start_dt) + build_header(session_id, cwd, start_dt),
        encoding="utf-8",
    )
    return path


# ============================================================
# メッセージ追記
# ============================================================
def append_messages(path: Path, messages: list) -> None:
    blocks = []

    for msg in messages:
        msg_type = msg.get("type", "")
        timestamp = msg.get("timestamp", "")
        dt = parse_timestamp(timestamp)
        time_str = dt.strftime("%H:%M:%S")
        text = extract_text(msg.get("content", ""))

        if not text:
            continue

        if msg_type == "user":
            blocks.append(f"### {time_str} 🙋 User\n\n{text}\n\n---\n\n")
        elif msg_type == "gemini":
            blocks.append(f"### {time_str} 🤖 Gemini\n\n{text}\n\n---\n\n")

    if blocks:
        with open(path, "a", encoding="utf-8") as f:
            f.write("".join(blocks))


# ============================================================
# イベントハンドラ
# ============================================================
def handle_session_start(hook_input: dict) -> None:
    session_id = hook_input.get("session_id", "")
    cwd = hook_input.get("cwd", "")
    if not session_id:
        log("SessionStart: session_id が取得できませんでした")
        return

    start_dt = now_jst()
    path = create_md_file(session_id, cwd, start_dt)

    state = load_state()
    state[session_id] = {
        "output_path": str(path),
        "start_time": start_dt.isoformat(),
        "cwd": cwd,
        "last_line": 0,
    }
    save_state(state)
    log(f"SessionStart: {session_id[:8]} -> {path}")


def handle_after_agent(hook_input: dict) -> None:
    session_id = hook_input.get("session_id", "")
    transcript_path = hook_input.get("transcript_path", "")
    cwd = hook_input.get("cwd", "")

    if not session_id or not transcript_path:
        log("AfterAgent: session_id または transcript_path が取得できませんでした")
        return

    state = load_state()

    # resume 等で SessionStart が走っていない場合はここで初期化
    if session_id not in state:
        start_dt = now_jst()
        path = create_md_file(session_id, cwd, start_dt)
        state[session_id] = {
            "output_path": str(path),
            "start_time": start_dt.isoformat(),
            "cwd": cwd,
            "last_line": 0,
        }
        save_state(state)
        log(f"AfterAgent: セッション初期化 {session_id[:8]} -> {path}")

    path = Path(state[session_id]["output_path"])
    last_line = state[session_id].get("last_line", 0)

    new_messages, next_line = load_jsonl(transcript_path, last_line)

    if new_messages:
        append_messages(path, new_messages)
        log(f"AfterAgent: {len(new_messages)}件追記完了 {session_id[:8]}")
    else:
        log(f"AfterAgent: 新規メッセージなし {session_id[:8]}")

    if next_line != last_line:
        state[session_id]["last_line"] = next_line
        save_state(state)


# ============================================================
# メイン
# ============================================================
def main() -> None:
    hook_input = read_stdin()
    event = hook_input.get("hook_event_name", "")

    if event == "SessionStart":
        handle_session_start(hook_input)
    elif event == "AfterAgent":
        handle_after_agent(hook_input)
    else:
        log(f"未対応イベント: {event}")

    print("{}")


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        log(f"予期しないエラー: {e}")
        print("{}")
        sys.exit(0)

---
session_id: 9c9768bf-6e86-4159-9f30-95c5d0282887
project: /Users/xxxx/Documents/xxxx
time_start: 2026-04-25T14:11:24.917364+09:00
tags:
  - gemini-cli
---

# Gemini CLI セッション

| Key | Value |
|---|---|
| Session | `9c9768bf` |
| Project | `/Users/xxxx/Documents/xxxx` |
| Started | 2026-04-25 14:11:24 |

---

## 会話ログ

### 14:12:53 🙋 User

Google CloudのVertex AIについて、サービス名が変わったのでしょうか？Google検索した上で答えてください

---

### 14:12:55 🤖 Gemini

Google CloudのVertex AIのサービス名が変わったかどうか、Google検索で調べてお答えします。

---

### 14:13:31 🤖 Gemini

はい、Google CloudのVertex AIはサービス名が変更されました。

2026年4月に開催されたGoogle Cloud Next '26にて、**Vertex AI**は**「Gemini Enterprise Agent Platform」**へとリブランディングされることが発表されました。これは、AIモデルの提供だけでなく、「AIエージェント」の構築・運用基盤へとシフトしたことを象徴する変更です。

これに伴い、**Vertex AI Search**は**Agent Search**に、Webサービス版の**Gemini Enterprise**は**Gemini Enterprise app**にそれぞれ名称が変更されています。

Vertex AIは、2021年5月にAI PlatformとAutoMLを統合して誕生して以来、これまでにも機能の統合やブランド統一のためにいくつかの名称変更を行ってきています。

---

### 14:19:35 🙋 User

Google Cloud Next '26 で発表されたのですね。セキュリティ製品で注目すべきアナウンスはありましたか？

---

### 14:19:45 🤖 Gemini

Google Cloud Next '26でのセキュリティ製品に関する注目すべき発表についてですね。Google検索で調べてお伝えします。

---

### 14:20:17 🤖 Gemini

Google Cloud Next '26では、「Agentic Defense（エージェントによる防御）」をテーマに、AIを活用したセキュリティの発表が多数ありました。主な注目点は以下の通りです。

1.  **Google Security Operations の新しいAIエージェント**:
    *   **Threat Hunting エージェント**: 未知の攻撃パターンやステルスな攻撃者の行動をプロアクティブに探索します。
    *   **Detection Engineering エージェント**: 防御のギャップを特定し、新しい検知ルールを自動生成することで、手動でのルール作成作業を自動化します。
    *   **Third-Party Context エージェント**: サードパーティのコンテキストデータをワークフローに統合し、より精度の高い分析を可能にします。

2.  **「エージェント・エンタープライズ」の保護**:
    *   **Gemini Enterprise Agent Platform**の強化: AIエージェントの構築・管理・統制のためのプラットフォームで、Agent Identity、Agent Gateway、Model Armorといった機能が発表されました。
    *   **AI-BOM (AI Bill of Materials)**: AIが生成したコードの安全性を確保し、「シャドーAI」のリスクを軽減するための構成管理ツールです。

3.  **Wiz との戦略的提携の深化**:
    *   **Wiz AI-APP (AI Application Protection Platform)**: コードからランタイムまで、AIアプリケーションの全レイヤーを保護します。マルチクラウド環境への対応も拡大しています。

4.  **インフラおよびネットワークセキュリティの進化**:
    *   **Cloud NGFW (次世代ファイアウォール)**: Palo Alto Networksの技術を活用した「高度なマルウェアサンドボックス」が年内にプレビュー公開されます。
    *   **Cloud Armor**: Thales Impervaの技術を活用した新しい管理ルールが導入され、Layer 7攻撃やゼロデイ脆弱性への対策が強化されます。
    *   **Confidential Computing**: NVIDIA Blackwell GPUを搭載したG4 VMでの機密コンピューティングがサポートされます。

5.  **セキュリティ運用とインテリジェンスの統合**:
    *   **Security Command Center (SCC)**: Cloud RunやGKE上の管理されていないエージェントワークロードの可視化機能が追加され、標準ティアでデータセキュリティポスチャ管理（DSPM）などが利用可能になります。
    *   **Google Threat Intelligence**: Mandiant、VirusTotal、Geminiを統合したインテリジェンスが提供されます。
    *   **Dark Web Intelligence**: Geminiモデルを使用して、ダークウェブ上の情報から組織固有の脅威プロファイルを構築します。

6.  **Google Cloud Fraud Defense**: reCAPTCHAの技術をベースにした新しい詐欺対策プラットフォームで、AIエージェントが介在するウェブ環境におけるユーザー体験とカスタマージャーニー全体を保護します。

これらの発表は、攻撃側がAIで攻撃を加速する現代において、防御側もAIを最大限に活用し、「マシンスピード」で対応していくというGoogle Cloudのセキュリティ戦略を示しています。

---