# 機能設計書 24-タスクストリーム

## 概要

本ドキュメントは、Giselleシステムにおける「タスクストリーム」機能の設計仕様を定義する。この機能は、タスクの実行状況をリアルタイムでクライアントにストリーム配信するための機能である。

### 本機能の処理概要

**業務上の目的・背景**：ワークフロー実行中、ユーザーは処理の進行状況をリアルタイムで確認したいという要件がある。この機能により、Server-Sent Events (SSE) を使用してタスクの状態変化をクライアントに即座に通知でき、ユーザーエクスペリエンスの向上と進捗の可視化を実現する。

**機能の利用シーン**：
- プレイグラウンド画面でアプリ実行テスト中に進捗を表示する場合
- ワークフローエディタでRunボタン押下後の実行状況を監視する場合
- タスク詳細画面で実行中のタスクの状態を追跡する場合

**主要な処理内容**：
1. ReadableStreamを作成してSSE形式でデータを配信する
2. 接続確立時に"connected"イベントを送信する
3. 500ms間隔でタスクデータをポーリングして取得する
4. タスクデータに変更があった場合のみ"data"イベントを送信する
5. タスクが完了した場合は"end"イベントを送信してストリームを閉じる
6. 20分のタイムアウトで自動的にストリームを終了する

**関連システム・外部連携**：
- Storage: タスクデータの取得
- HTTP: Server-Sent Events (SSE) プロトコル

**権限による制御**：
- ストリームへのアクセスはタスクへのアクセス権限に依存
- 適切な認証情報を持つクライアントのみが接続可能

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 18 | プレイグラウンド画面 | 主画面 | タスク実行状況のリアルタイムストリーム配信 |

## 機能種別

リアルタイム通信 / ストリーミング

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| taskId | TaskId | Yes | ストリーム対象のタスクID | `tsk_`で始まる有効なID形式 |
| context | GiselleContext | Yes | Giselleコンテキスト | 有効なコンテキストオブジェクト |

### 入力データソース

- Storage: `tasks/{taskId}.json` からタスクデータをポーリング取得

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| ReadableStream | ReadableStream | SSE形式のストリーム |

### ストリームイベント形式

| イベントタイプ | データ構造 | 説明 |
|--------------|-----------|------|
| connected | `{ type: "connected" }` | 接続確立通知 |
| data | `{ type: "data", data: StreamData }` | タスクデータ更新通知 |
| error | `{ type: "error", message: string }` | エラー通知 |
| end | `{ type: "end" }` | ストリーム終了通知 |

### 出力先

- HTTP Response: SSE形式でクライアントに配信

## 処理フロー

### 処理シーケンス

```
1. ストリーム初期化
   ├─ ReadableStreamを作成
   ├─ TextEncoderを初期化
   └─ ポーリング変数を初期化

2. ストリーム開始（start関数）
   ├─ "connected"イベントを送信
   ├─ 初回のsendUpdate実行
   └─ ポーリング設定
       ├─ setInterval（500ms間隔）
       └─ setTimeout（20分タイムアウト）

3. ポーリングサイクル（poll関数）
   ├─ pollingフラグチェック
   ├─ sendUpdate実行
   │   ├─ getTaskでタスクデータ取得
   │   ├─ createDataHashでハッシュ計算
   │   ├─ ハッシュ比較（変更検出）
   │   └─ 変更あれば"data"イベント送信
   └─ タスク完了チェック
       └─ 完了ならcleanup実行

4. クリーンアップ（cleanup関数）
   ├─ pollingフラグをfalseに設定
   ├─ setIntervalをクリア
   ├─ setTimeoutをクリア
   ├─ "end"イベントを送信
   └─ ストリームを閉じる

5. キャンセル処理（cancel関数）
   └─ cleanupResourcesを実行
```

### フローチャート

```mermaid
flowchart TD
    A[開始] --> B[ReadableStream作成]
    B --> C[connectedイベント送信]
    C --> D[初回poll実行]
    D --> E{タスク完了?}
    E -->|Yes| F[cleanup実行]
    E -->|No| G[setInterval開始]
    G --> H[setTimeout開始]
    H --> I{500ms経過}
    I --> J[poll実行]
    J --> K[getTask]
    K --> L[ハッシュ計算]
    L --> M{変更あり?}
    M -->|Yes| N[dataイベント送信]
    M -->|No| O[スキップ]
    N --> P{タスク完了?}
    O --> P
    P -->|Yes| F
    P -->|No| Q{20分経過?}
    Q -->|Yes| F
    Q -->|No| I
    F --> R[endイベント送信]
    R --> S[ストリーム閉じる]
    S --> T[終了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-24-01 | ポーリング間隔 | 500ms間隔でタスクデータをポーリング | ストリーム実行中 |
| BR-24-02 | 変更検出 | updatedAtフィールドのハッシュで変更を検出 | 各ポーリング時 |
| BR-24-03 | タイムアウト | 20分（1200秒）でストリームを自動終了 | ストリーム実行中 |
| BR-24-04 | 完了時終了 | タスクがcompletedになったらストリームを終了 | タスク完了時 |
| BR-24-05 | 差分送信 | 変更があった場合のみデータを送信 | 各ポーリング時 |

### 計算ロジック

**データハッシュ計算**：
```typescript
function createDataHash(data: StreamData): string {
  return JSON.stringify({
    taskUpdatedAt: data.task.updatedAt,
  });
}
```

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| タスク取得 | Storage (tasks/{taskId}.json) | SELECT | ポーリングによるタスクデータ取得 |

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

#### Storage: tasks/{taskId}.json

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| SELECT | 全フィールド | taskIdで取得 | 500ms間隔でポーリング |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TASK_NOT_FOUND | NotFoundError | タスクが存在しない | 有効なタスクIDを指定 |
| FETCH_ERROR | RuntimeError | タスク取得に失敗 | エラーイベントを送信して継続 |
| STREAM_CLOSED | RuntimeError | クライアントが切断 | リソースをクリーンアップ |

### リトライ仕様

- タスク取得エラー: エラーイベントを送信して次のポーリングで再試行
- ストリームエラー: クリーンアップしてストリームを終了

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

- 読み取り専用のため、トランザクション制御は不要
- 複数クライアントが同一タスクをストリームする場合、各クライアントは独立

## パフォーマンス要件

- ポーリング間隔: 500ms（固定）
- タスク取得: 100ms以内
- イベント送信: 10ms以内
- 最大接続時間: 20分

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

- タスクへのアクセス権限チェックは呼び出し側で実施
- ストリームはHTTPS経由で暗号化
- タイムアウトによるリソース消費制限

## 備考

- SSE形式: `data: ${JSON.stringify(event)}\n\n`
- TextEncoderでバイナリエンコード
- controller.desiredSizeがnullの場合はストリームが閉じられている

---

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

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

### 推奨読解順序

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

ストリームイベントとタスクデータの構造を把握する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | stream-task.ts | `packages/giselle/src/tasks/stream-task.ts` | StreamData, StreamEvent型の構造 |
| 1-2 | task.ts | `packages/protocol/src/task/task.ts` | Task型の構造（updatedAtフィールド） |

**読解のコツ**: StreamEventはunion型で、type="connected"|"data"|"error"|"end"の4種類がある。

#### Step 2: エントリーポイントを理解する

streamTask関数の全体構造を把握する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | stream-task.ts | `packages/giselle/src/tasks/stream-task.ts` | streamTask関数とReadableStreamの実装 |

**主要処理フロー**:
1. **6-11行目**: createDataHash関数（変更検出用ハッシュ生成）
2. **13-15行目**: formatStreamData関数（SSE形式フォーマット）
3. **17-19行目**: streamTask関数シグネチャ
4. **18行目**: TextEncoderの初期化
5. **20-36行目**: クリーンアップリソース関数
6. **38行目**: ReadableStream作成
7. **39-46行目**: start関数（connectedイベント送信）
8. **48-90行目**: sendUpdate関数（タスク取得と変更検出）
9. **92-100行目**: cleanup関数（endイベントとストリーム終了）
10. **102-112行目**: poll関数（ポーリング処理）
11. **114-121行目**: 初回実行とインターバル/タイムアウト設定
12. **124-126行目**: cancel関数（切断時処理）

#### Step 3: タスク取得処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | get-task.ts | `packages/giselle/src/tasks/get-task.ts` | ストレージからのタスク取得 |

#### Step 4: Giselleクラスでの公開を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | giselle.ts | `packages/giselle/src/giselle.ts` | streamTaskメソッドの公開 |

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

```
streamTask (stream-task.ts)
    │
    └─ new ReadableStream
           │
           ├─ start()
           │      │
           │      ├─ formatStreamData({ type: "connected" })
           │      │
           │      ├─ poll() [初回実行]
           │      │      │
           │      │      └─ sendUpdate()
           │      │             │
           │      │             ├─ getTask (get-task.ts)
           │      │             │
           │      │             ├─ createDataHash
           │      │             │
           │      │             └─ formatStreamData({ type: "data" })
           │      │
           │      ├─ setInterval(poll, 500) [ポーリング開始]
           │      │
           │      └─ setTimeout(cleanup, 20min) [タイムアウト設定]
           │
           └─ cancel()
                  │
                  └─ cleanupResources()
```

### データフロー図

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

taskId ──────────────▶ streamTask
                           │
                           ├─▶ ReadableStream作成
                           │
                           ├─▶ connectedイベント ──────────▶ クライアント
                           │
                           ├─▶ ポーリングループ
                           │       │
                           │       ├─▶ getTask ──▶ Task
                           │       │
                           │       ├─▶ ハッシュ比較
                           │       │
                           │       └─▶ 変更あり?
                           │               │
                           │               └─▶ dataイベント ──▶ クライアント
                           │
                           ├─▶ タスク完了/タイムアウト
                           │       │
                           │       └─▶ endイベント ─────────▶ クライアント
                           │
                           └─▶ ストリーム終了
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| stream-task.ts | `packages/giselle/src/tasks/stream-task.ts` | ソース | ストリーム機能のメイン実装 |
| get-task.ts | `packages/giselle/src/tasks/get-task.ts` | ソース | タスク取得処理 |
| task.ts | `packages/protocol/src/task/task.ts` | ソース | Task型定義 |
| giselle.ts | `packages/giselle/src/giselle.ts` | ソース | Giselleクラス（メソッド公開） |
| index.ts | `packages/giselle/src/tasks/index.ts` | ソース | モジュールエクスポート |
