# 機能設計書 24-チェックポイント永続化

## 概要

本ドキュメントは、Liam DBシステムにおけるチェックポイント永続化機能の設計を記述するものである。本機能は、LangGraphワークフローの状態をSupabaseデータベースに永続化し、ワークフローの中断・再開やセッション間での状態維持を可能にする。

### 本機能の処理概要

チェックポイント永続化機能は、LangGraphのBaseCheckpointSaverインターフェースを実装し、Supabaseをバックエンドとしてワークフロー状態を永続化する。これにより、長時間実行されるAIエージェントワークフローの信頼性と継続性を確保する。

**業務上の目的・背景**：AIエージェントによるデータベース設計ワークフローは複数のステップを経て完了するため、途中での中断や再開が必要になる場合がある。チェックポイント永続化により、セッションの状態を保持し、ブラウザを閉じた後でも作業を継続できるようにする。また、マルチテナント環境での組織単位のデータ分離を実現する。

**機能の利用シーン**：デザインセッションでAIエージェントと対話中に、各ノードの実行完了時に自動的にチェックポイントが保存される。ユーザーが後日セッションを再開する際、保存されたチェックポイントから状態を復元してワークフローを継続できる。

**主要な処理内容**：
1. ワークフロー状態（Checkpoint）のSupabaseへの保存
2. チャンネル値（channel_values）のBLOB形式での保存
3. 保留中の書き込み（pending_writes）の管理
4. 組織ID（organization_id）によるマルチテナント分離
5. Base64/Hex形式でのBYTEA列のエンコード・デコード

**関連システム・外部連携**：Supabase PostgreSQLデータベースと連携し、checkpoints、checkpoint_blobs、checkpoint_writesテーブルにデータを永続化する。

**権限による制御**：チェックポイントデータはorganization_idでフィルタリングされ、同一組織内のユーザーのみアクセス可能。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 14 | デザインセッション詳細画面 | 主画面 | エージェント状態のSupabaseベース永続化 |

## 機能種別

データ連携 / 永続化処理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| config | RunnableConfig | Yes | スレッドID、チェックポイント名前空間等を含む設定 | thread_id必須 |
| checkpoint | Checkpoint | Yes（putの場合） | 保存対象のチェックポイント | - |
| metadata | CheckpointMetadata | Yes（putの場合） | チェックポイントのメタデータ | - |
| newVersions | ChannelVersions | Yes（putの場合） | 更新されたチャンネルバージョン | - |
| organizationId | string | Yes | 組織の識別子（コンストラクタで設定） | UUID形式 |

### 入力データソース

- LangGraphワークフローからの呼び出し
- RunnableConfigに含まれるスレッド情報

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| CheckpointTuple | CheckpointTuple | チェックポイント、メタデータ、設定を含むタプル |
| RunnableConfig | RunnableConfig | 保存後の設定（新しいcheckpoint_id含む） |

### 出力先

- Supabaseデータベースの以下のテーブル：
  - checkpoints: チェックポイント本体
  - checkpoint_blobs: チャンネル値（BLOB形式）
  - checkpoint_writes: 保留中の書き込み

## 処理フロー

### 処理シーケンス

```
1. getTuple（チェックポイント取得）
   └─ thread_id, checkpoint_ns, checkpoint_idでクエリ
   └─ checkpoint_blobs, checkpoint_writesも取得
   └─ _loadCheckpoint, _loadMetadata, _loadWritesで復元
2. put（チェックポイント保存）
   └─ _dumpCheckpointでシリアライズ
   └─ _dumpBlobsでチャンネル値をBase64エンコード
   └─ RPCでアトミックに保存
3. putWrites（書き込み保存）
   └─ _dumpWritesでシリアライズ
   └─ upsertで保存
4. list（チェックポイント一覧）
   └─ フィルタ、ページング対応
   └─ AsyncGeneratorでイテレート
```

### フローチャート

```mermaid
flowchart TD
    A[ワークフローノード実行] --> B{操作種別}
    B -->|取得| C[getTuple]
    B -->|保存| D[put]
    B -->|書き込み追加| E[putWrites]
    C --> F[Supabase SELECT]
    D --> G[_dumpCheckpoint]
    G --> H[_dumpBlobs]
    H --> I[RPC put_checkpoint]
    E --> J[_dumpWrites]
    J --> K[Supabase UPSERT]
    F --> L[_loadCheckpoint]
    L --> M[CheckpointTuple返却]
    I --> N[RunnableConfig返却]
    K --> O[完了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-24-01 | 組織分離 | 全クエリにorganization_idフィルタを適用 | 常時 |
| BR-24-02 | アトミック保存 | checkpointとblobsはRPCでアトミックに保存 | put操作時 |
| BR-24-03 | バージョン管理 | 整数インクリメントでバージョン管理 | getNextVersion |
| BR-24-04 | BYTEA処理 | SupabaseではBase64で送信、Hexで受信 | BLOB操作時 |

### 計算ロジック

- バージョン算出: `getNextVersion(current) => (current ?? 0) + 1`
- 最新BLOB選択: `_selectLatestBlobs`でchannelVersionsに基づき最新版を選択

## データベース操作仕様

### 操作別データベース影響一覧

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| getTuple | checkpoints, checkpoint_blobs, checkpoint_writes | SELECT | チェックポイント取得 |
| put | checkpoints, checkpoint_blobs | INSERT/UPDATE | チェックポイント保存（RPC経由） |
| putWrites | checkpoint_writes | UPSERT | 書き込み保存 |
| list | checkpoints, checkpoint_blobs, checkpoint_writes | SELECT | 一覧取得 |
| deleteThread | checkpoints, checkpoint_blobs, checkpoint_writes | DELETE | スレッド削除 |

### テーブル別操作詳細

#### checkpoints

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| INSERT | thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, checkpoint, metadata, organization_id | ワークフロー状態 | RPCで実行 |

#### checkpoint_blobs

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| INSERT | thread_id, checkpoint_ns, channel, version, type, blob, organization_id | チャンネル値（Base64） | RPCで実行 |

#### checkpoint_writes

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| UPSERT | thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, blob, organization_id | 保留書き込み | onConflict指定 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| MISSING_CONFIG | 設定エラー | configurable未設定 | 例外スロー |
| SAVE_FAILED | 保存エラー | RPC/UPSERT失敗 | リトライ後に例外スロー |
| DELETE_FAILED | 削除エラー | DELETE失敗 | 例外スロー |

### リトライ仕様

- `retry`関数（@liam-hq/db）を使用してDB操作をリトライ
- 一時的なエラー（ネットワーク等）に対応

## トランザクション仕様

- put操作: `put_checkpoint` RPCで checkpoints と checkpoint_blobs をアトミックに保存
- putWrites操作: 単一テーブルへのupsertのため自動トランザクション

## パフォーマンス要件

- チェックポイント保存: ノード実行完了後、非同期で実行
- チェックポイント取得: ワークフロー開始時に1回

## セキュリティ考慮事項

- organization_idによるデータ分離
- SupabaseのRow Level Security（RLS）と連携
- 機密データ（スキーマ情報等）のBLOB暗号化は現状未実装

## 備考

SupabaseCheckpointSaverはLangGraph 0.4.x互換。BYTEAカラムのエンコーディングはSupabase REST API特有の処理が必要。

---

## コードリーディングガイド

本機能を理解するために参照すべきファイルと、推奨する読み解き順序を以下に示す。

### 推奨読解順序

#### Step 1: データ構造を理解する

チェックポイントの構造とシリアライズ形式を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | types.ts | `frontend/internal-packages/agent/src/checkpoint/types.ts` | SerializedCheckpoint型定義 |
| 1-2 | schemas.ts | `frontend/internal-packages/agent/src/checkpoint/schemas.ts` | パーススキーマ定義 |

**読解のコツ**: LangGraphのCheckpoint型とカスタムのSerializedCheckpoint型の違いに注目。

#### Step 2: メインクラスを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SupabaseCheckpointSaver.ts | `frontend/internal-packages/agent/src/checkpoint/SupabaseCheckpointSaver.ts` | クラス全体構造 |

**主要処理フロー**:
- **69-78行目**: コンストラクタ（client, organizationId設定）
- **80-173行目**: getTuple（チェックポイント取得）
- **290-356行目**: put（チェックポイント保存）
- **358-402行目**: putWrites（書き込み保存）
- **404-406行目**: getNextVersion（バージョン算出）

#### Step 3: シリアライズ・デシリアライズを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | byteaUtils.ts | `frontend/internal-packages/agent/src/checkpoint/byteaUtils.ts` | Base64/Hex変換ユーティリティ |

**主要処理フロー**:
- hexToUint8Array: Supabaseから受信したHex文字列をバイナリに変換
- uint8ArrayToBase64: バイナリをBase64に変換してSupabaseに送信

#### Step 4: ワークフローとの統合を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | createGraph.ts | `frontend/internal-packages/agent/src/createGraph.ts` | checkpointerの利用箇所 |

**主要処理フロー**:
- **107行目**: graph.compile({ checkpointer })でチェックポイント機能を有効化

### プログラム呼び出し階層図

```
createGraph (createGraph.ts)
    │
    └─ graph.compile({ checkpointer })
           │
           └─ SupabaseCheckpointSaver
                  │
                  ├─ getTuple
                  │      ├─ client.from('checkpoints').select()
                  │      ├─ client.from('checkpoint_blobs').select()
                  │      ├─ client.from('checkpoint_writes').select()
                  │      ├─ _loadCheckpoint
                  │      │      └─ _loadBlobs → hexToUint8Array
                  │      ├─ _loadMetadata
                  │      └─ _loadWrites → hexToUint8Array
                  │
                  ├─ put
                  │      ├─ _dumpCheckpoint
                  │      ├─ _dumpBlobs → uint8ArrayToBase64
                  │      └─ client.rpc('put_checkpoint')
                  │
                  ├─ putWrites
                  │      └─ client.from('checkpoint_writes').upsert()
                  │
                  └─ list
                         └─ AsyncGenerator<CheckpointTuple>
```

### データフロー図

```
[入力]                    [処理]                         [出力]

Checkpoint ────────────▶ _dumpCheckpoint ───────────▶ JSON
       │                        │
channel_values ────────▶ _dumpBlobs ────────────────▶ Base64 BLOB
       │                        │
       │                        ▼
       └───────────────▶ client.rpc('put_checkpoint') ──▶ Supabase DB
                                │
                                │
[復元]                          │
                                │
Supabase DB ───────────▶ SELECT ─────────────────────▶ Hex BLOB
       │                        │
       │                        ▼
       │                 hexToUint8Array
       │                        │
       │                        ▼
       │                 serde.loadsTyped
       │                        │
       └───────────────────────▶ CheckpointTuple
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SupabaseCheckpointSaver.ts | `frontend/internal-packages/agent/src/checkpoint/SupabaseCheckpointSaver.ts` | ソース | メインクラス |
| types.ts | `frontend/internal-packages/agent/src/checkpoint/types.ts` | ソース | 型定義 |
| schemas.ts | `frontend/internal-packages/agent/src/checkpoint/schemas.ts` | ソース | パーススキーマ |
| byteaUtils.ts | `frontend/internal-packages/agent/src/checkpoint/byteaUtils.ts` | ソース | エンコード/デコード |
| createGraph.ts | `frontend/internal-packages/agent/src/createGraph.ts` | ソース | チェックポイント利用 |
| database.types.ts | `frontend/internal-packages/db/supabase/database.types.ts` | 型定義 | DBスキーマ型 |
| retry.ts | `frontend/internal-packages/db/src/utils/retry.ts` | ソース | リトライユーティリティ |
