> ## Documentation Index
> Fetch the complete documentation index at: https://adcp-docs-ja.pier1.co.jp/llms.txt
> Use this file to discover all available pages before exploring further.

# オーケストレーター設計

> AdCP オーケストレーター設計: マルチベンダーキャンペーンワークフロー向けのステートマシンパターン、永続的オペレーション追跡、非同期ファーストアーキテクチャ、再同期。

このガイドは、非同期オペレーションや保留状態、Human-in-the-Loop を扱う AdCP オーケストレーターのベストプラクティスをまとめています。

## 基本設計原則

### 1. 非同期を前提に

AdCP プロトコルは本質的に非同期です。処理は秒〜日単位でかかることがあります。

**DO:**

* すべてのオペレーションを async/await で設計
* オペレーション状態を永続化
* オーケストレーター再起動に耐える
* 適切なタイムアウトを実装

**DON'T:**

* 即時完了を前提にしません
* 同期ブロッキング呼び出しを使わない
* 状態をメモリだけに置かない
* バックオフなしに無限リトライしません

### 2. ステータス駆動ロジック

オペレーションは標準化されたステータス値を通じて進行する:

```python theme={null}
TASK_STATUSES = {
    "submitted",      # Long-running (hours to days) - provide webhook or poll
    "working",        # Processing (< 120 seconds) - poll frequently
    "input-required", # Need user input/approval - continue conversation
    "completed",      # Success - process results
    "failed",         # Error - handle appropriately
    "canceled",       # User canceled
    "auth-required"   # Need authentication
}
```

### 3. ステートマシン設計

AdCP タスクステータスに合わせた適切なステートマシンを実装します:

```python theme={null}
class OperationState(Enum):
    # Local orchestrator states
    REQUESTED = "requested"
    CALLING_ADCP = "calling_adcp"

    # AdCP task states (match server responses)
    SUBMITTED = "submitted"
    WORKING = "working"
    INPUT_REQUIRED = "input_required"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELED = "canceled"

# Valid state transitions
VALID_TRANSITIONS = {
    "requested": ["calling_adcp"],
    "calling_adcp": ["submitted", "working", "input_required", "completed", "failed"],
    "submitted": ["working", "completed", "failed", "canceled"],
    "working": ["completed", "failed", "input_required"],
    "input_required": ["submitted", "working", "completed", "failed"]
}
```

## オペレーションの追跡

### 永続ストレージ

すべてのオペレーションを詳細に追跡しながら保存します:

```python theme={null}
class OperationTracker:
    def __init__(self, db):
        self.db = db

    async def create_operation(self, operation_type, request_data, webhook_config=None):
        operation = {
            "id": str(uuid.uuid4()),
            "type": operation_type,
            "status": "requested",
            "request": request_data,
            "webhook_config": webhook_config,
            "created_at": datetime.now(),
            "updated_at": datetime.now(),
            "task_id": None,
            "context_id": None,
            "result": None,
            "error": None
        }
        await self.db.operations.insert_one(operation)
        return operation["id"]

    async def update_status(self, operation_id, status, **kwargs):
        update = {
            "status": status,
            "updated_at": datetime.now()
        }
        update.update(kwargs)

        await self.db.operations.update_one(
            {"id": operation_id},
            {"$set": update}
        )

    async def get_pending_operations(self):
        """Get all operations that need monitoring"""
        return await self.db.operations.find({
            "status": {"$in": ["submitted", "working", "input_required"]}
        }).to_list(length=None)
```

### 状態の再同期

起動時にローカル状態をサーバーと同期します:

```python theme={null}
async def reconcile_with_server(self, adcp_client):
    """Sync local state with server using AdCP task reconciliation"""
    server_tasks = await adcp_client.call('list_tasks', {
        'filters': {'statuses': ['submitted', 'working', 'input_required']}
    })

    server_task_ids = {task['task_id'] for task in server_tasks['tasks']}
    local_operations = await self.get_pending_operations()
    local_task_ids = {op['task_id'] for op in local_operations if op['task_id']}

    return {
        'orphaned_on_server': server_task_ids - local_task_ids,
        'missing_from_server': local_task_ids - server_task_ids,
        'total_pending_server': len(server_tasks['tasks']),
        'total_pending_local': len(local_operations)
    }
```

## 非同期オペレーションハンドラー

### レスポンスの振り分け

ステータスに応じてレスポンスを処理します:

```python theme={null}
class AsyncOperationHandler:
    def __init__(self, adcp_client, tracker, notifier):
        self.adcp = adcp_client
        self.tracker = tracker
        self.notifier = notifier
        self.polling_tasks = {}

    async def handle_operation_response(self, operation_id, response):
        """Handle any AdCP response with proper status routing"""
        status = response.get("status")

        # レスポンスの詳細でオペレーションを更新
        await self.tracker.update_status(
            operation_id,
            status,
            task_id=response.get("task_id"),
            context_id=response.get("context_id"),
            result=response.get("result") if status == "completed" else None,
            error=response.get("error") if status == "failed" else None
        )

        # ステータスに応じて処理を振り分け
        if status == "completed":
            await self._handle_completed(operation_id, response)
        elif status == "failed":
            await self._handle_failed(operation_id, response)
        elif status == "submitted":
            await self._handle_submitted(operation_id, response)
        elif status == "working":
            await self._handle_working(operation_id, response)
        elif status == "input_required":
            await self._handle_input_required(operation_id, response)
```

### Submitted（長時間）オペレーション

長時間の処理を扱います:

```python theme={null}
async def _handle_submitted(self, operation_id, response):
    """Handle long-running operations"""
    task_id = response["task_id"]

    # Check if webhook is configured
    operation = await self.tracker.get_operation(operation_id)
    webhook_config = operation.get("webhook_config")

    if webhook_config:
        # Webhook will handle completion notification
        await self.notifier.notify_submitted_with_webhook(operation_id, task_id)
    else:
        # Start polling for completion
        polling_task = asyncio.create_task(
            self._poll_for_completion(operation_id, task_id, interval=60)
        )
        self.polling_tasks[task_id] = polling_task
```

### バックオフ付きポーリング

効率的なポーリングを実装します:

```python theme={null}
async def _poll_for_completion(self, operation_id, task_id, interval=60):
    """Poll task status until completion"""
    max_polls = 1440 if interval == 60 else 24  # 24 hours or 2 minutes
    poll_count = 0

    while poll_count < max_polls:
        try:
            await asyncio.sleep(interval)
            poll_count += 1

            task_response = await self.adcp.call('get_task_status', {
                'task_id': task_id,
                'include_result': True
            })

            await self.handle_operation_response(operation_id, task_response)

            if task_response["status"] in ["completed", "failed", "canceled"]:
                break

        except Exception as e:
            await self.tracker.update_status(
                operation_id,
                "failed",
                error=f"Polling error: {str(e)}"
            )
            break

    self.polling_tasks.pop(task_id, None)

    if poll_count >= max_polls:
        await self.tracker.update_status(
            operation_id,
            "failed",
            error="Task polling timeout"
        )
```

## Webhook サポート

### 信頼性の高い Webhook ハンドラー

信頼性パターンを使って Webhook を実装します:

```python theme={null}
class WebhookHandler:
    def __init__(self, tracker, notifier, secret_key):
        self.tracker = tracker
        self.notifier = notifier
        self.secret_key = secret_key
        self.processed_events = {}

    def verify_webhook_signature(self, payload: bytes, signature: str) -> bool:
        """Verify webhook authenticity"""
        expected_signature = hmac.new(
            self.secret_key.encode(),
            payload,
            hashlib.sha256
        ).hexdigest()
        return signature == f"sha256={expected_signature}"

    async def is_replay_attack(self, timestamp: str, event_id: str) -> bool:
        """Prevent replay attacks using timestamp and event ID"""
        event_time = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
        now = datetime.now()

        if now - event_time > timedelta(minutes=5):
            return True

        return event_id in self.processed_events
```

### Webhook とポーリングの併用（バックアップ）

Webhook のみに依存しないようにします:

```python theme={null}
class ReliableWebhookOrchestrator:
    def __init__(self):
        self.webhook_timeout = timedelta(minutes=10)
        self.backup_polling_delay = timedelta(minutes=2)

    async def _handle_submitted_with_webhook(self, operation_id, task_id):
        """Handle submitted task with webhook + backup polling"""

        async def backup_polling():
            await asyncio.sleep(self.backup_polling_delay.total_seconds())

            operation = await tracker.get_operation(operation_id)
            if operation["status"] not in ["completed", "failed", "canceled"]:
                logger.info(f"Starting backup polling for task {task_id}")
                await self._poll_for_completion(operation_id, task_id, interval=60)

        asyncio.create_task(backup_polling())
```

## オーケストレーターの例

オーケストレーターの実装例:

```python theme={null}
class AdCPOrchestrator:
    def __init__(self):
        self.adcp = AdCPClient()
        self.tracker = OperationTracker(db)
        self.handler = AsyncOperationHandler(self.adcp, self.tracker, UserNotifier())
        self.webhook_base_url = "https://orchestrator.com/webhooks"

    async def create_campaign(self, user_id, request, enable_webhook=True):
        """Create a campaign with governance validation and full async handling.

        Plans must already be synced via sync_plans before calling this method.
        Plan creation happens during the planning phase, not at campaign creation time.
        """

        # 1. Run intent check (plan must already exist)
        if request.get("governance_context"):
            gov_check = await self.adcp.call("check_governance", {
                "plan_id": request["governance_context"]["plan_id"],
                "caller": request["governance_context"]["caller"],
                "tool": "create_media_buy",
                "payload": request
            })
            if gov_check["status"] == "denied":
                raise GovernanceDeniedError(gov_check["explanation"])
            if gov_check["status"] == "conditions":
                raise GovernanceConditionsError(gov_check["conditions"])
            # If check_governance needs human review internally, it returns
            # async task status (submitted/working) and resolves to
            # approved or denied — standard task lifecycle.

        # 2. Create the media buy
        await self._create_media_buy(user_id, request, enable_webhook)

    async def _create_media_buy(self, user_id, request, enable_webhook=True):
        """Create a media buy with full async handling."""

        # 1. Prepare webhook configuration
        webhook_config = None
        if enable_webhook:
            webhook_config = {
                "webhook_url": f"{self.webhook_base_url}/adcp/{user_id}",
                "webhook_auth": {
                    "type": "bearer",
                    "credentials": await self.get_webhook_token(user_id)
                }
            }

        # 2. Create operation record
        operation_id = await self.tracker.create_operation(
            "create_media_buy",
            request,
            webhook_config=webhook_config
        )

        try:
            # 3. Call AdCP
            response = await self.adcp.call("create_media_buy", request, webhook_config)

            # 4. Handle response
            await self.handler.handle_operation_response(operation_id, response)

            # 5. Return appropriate response to user
            return self._format_user_response(operation_id, response)

        except Exception as e:
            await self.tracker.update_status(operation_id, "failed", error=str(e))
            raise

    async def reconcile_state_on_startup(self):
        """Recover from orchestrator restart"""
        reconciliation = await self.tracker.reconcile_with_server(self.adcp)
        logger.info(f"State reconciliation: {reconciliation}")

        for task_id in reconciliation["orphaned_on_server"]:
            # Resume monitoring orphaned tasks
            operation_id = await self.tracker.create_operation(
                "unknown",
                {},
                status="submitted"
            )
            await self.tracker.update_status(operation_id, "submitted", task_id=task_id)
            asyncio.create_task(
                self.handler._poll_for_completion(operation_id, task_id)
            )
```

## キャンペーンライフサイクルにおけるガバナンス

プラン作成（[`sync_plans`](/docs/governance/campaign/tasks/sync_plans)）は計画フェーズ中 — キャンペーンが存在する前に発生します。ガバナンスチェックはキャンペーン実行中に発生します。これらは別々の関心事です。

**計画フェーズ**（メディアプランごとに 1 回）:

```
sync_plans — オーケストレーターがプランをガバナンスエージェントにプッシュする
```

**キャンペーン実行**（メディアバイごと）:

```
check_governance(tool + payload) → create_media_buy → check_governance(media_buy_id + planned_delivery) → delivery → report_plan_outcome
```

| フェーズ      | 呼び出し元     | タスク                                                                                                            | 失敗時に起こること                                                                                        |
| --------- | --------- | -------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------ |
| インテントチェック | オーケストレーター | [`check_governance`](/docs/governance/campaign/tasks/check_governance)（`tool` + `payload`）                     | キャンペーンがバイヤーのプランに違反 — いかなる支出も発生する前に拒否または条件付き。ガバナンスエージェントが人間のレビューを必要とする場合、タスクは非同期になり承認または拒否に解決される。 |
| 実行チェック    | セラー       | [`check_governance`](/docs/governance/campaign/tasks/check_governance)（`media_buy_id` + `planned_delivery`）    | セラーのデリバリープランがバイヤーの期待と一致しない — 購入がブロックされる                                                          |
| デリバリーチェック | セラー       | [`check_governance`](/docs/governance/campaign/tasks/check_governance)（`phase: delivery` + `delivery_metrics`） | ドリフトを検出 — ペーシング、地域、またはチャネル分布がプランから逸脱                                                             |
| プラン成果     | オーケストレーター | [`report_plan_outcome`](/docs/governance/campaign/tasks/report_plan_outcome)                                   | フィードバックループなし — ガバナンスエージェントが将来の推奨を改善できない                                                          |

コード例付きの完全なシーケンスは[メディアバイのガバナンスワークフロー](/docs/media-buy/index#governance)を、セラーの実行チェック義務は[セラー統合ガイド](/docs/building/operating/seller-integration#execution-checks)を参照してください。

## ベストプラクティス

### 1. 永続ストレージ

オペレーションの状態には常に永続ストレージを使用します:

* データベース（PostgreSQL、MongoDB）
* メッセージキュー（Redis、RabbitMQ）
* 分散キャッシュ（Redis Cluster）

### 2. 冪等性

すべてのオペレーションを冪等にする:

```python theme={null}
async def create_media_buy_idempotent(self, request):
    existing = await self.db.operations.find_one({
        "type": "create_media_buy",
        "request.po_number": request["po_number"],
        "status": {"$in": ["created", "active"]}
    })

    if existing:
        return existing["result"]

    return await self.create_media_buy(request)
```

### 3. タイムアウト処理

合理的なタイムアウトを実装します:

```python theme={null}
OPERATION_TIMEOUTS = {
    "create_media_buy": timedelta(hours=24),
    "update_media_buy": timedelta(hours=12),
    "creative_approval": timedelta(hours=48)
}
```

### 4. エラーリカバリー

サーキットブレーカー付きのリトライロジックを実装します:

```python theme={null}
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(min=1, max=60),
    retry=retry_if_exception_type(TransientError)
)
async def call_adcp_api(self, tool, params):
    try:
        return await self.adcp.call(tool, params)
    except RateLimitError:
        raise TransientError("Rate limited")
    except NetworkError:
        raise TransientError("Network error")
```

### 5. モニタリングとアラート

主要メトリクスを追跡する:

* タイプ別の保留オペレーション数
* 平均承認時間
* 拒否率
* タスクタイムアウト率
* API エラー率

## ユーザーへの通知

保留中のオペレーションについてユーザーに知らせ続ける:

```python theme={null}
class UserNotifier:
    async def notify_pending_approval(self, user_id, operation):
        message = {
            "type": "pending_approval",
            "operation_id": operation["id"],
            "message": "Your media buy requires publisher approval",
            "estimated_time": "2-4 hours"
        }
        await self.send_notification(user_id, message)

    async def notify_approval(self, user_id, operation):
        message = {
            "type": "operation_approved",
            "operation_id": operation["id"],
            "message": "Your media buy has been approved",
            "media_buy_id": operation["result"]["media_buy_id"]
        }
        await self.send_notification(user_id, message)
```

## まとめ

堅牢な AdCP オーケストレーターを構築するには次のことが必要だ:

1. 全体を通じた非同期設計
2. 永続化を伴う適切な状態管理
3. 保留状態の適切な処理
4. 長時間オペレーションでのユーザーへの通知
5. モニタリングと観測性

保留状態はエラーではなく、広告ワークフローの正常な一部です。

## 次のステップ

* **Task Lifecycle**: ステータス処理については [Task Lifecycle](/docs/building/by-layer/L3/task-lifecycle) を参照
* **Webhooks**: プッシュ通知については [Webhooks](/docs/building/by-layer/L3/webhooks) を参照
* **Security**: マルチテナントセキュリティについては [Security](/docs/building/by-layer/L1/security) を参照