debugging-signals-pipeline

シグナルパイプラインのデバッグ

パイプラインの流れ

emit_signals_from_fixture
  → signal-emitter (Temporal workflow)
    → buffer-signals (batches signals, 5s flush timer)
      → safety_filter_activity
      → flush_signals_to_s3_activity
      → signal_with_start_grouping_v2_activity
        → team-signal-grouping-v2 (30s batch collect window)
          → read_signals_from_s3_activity
          → get_embedding_activity + generate_search_queries_activity
          → run_signal_semantic_search_activity
          → match_signal_to_report_activity
          → assign_and_emit_signal_activity
          → wait_for_signal_in_clickhouse_activity
          → (if new report) signal-report-summary
            → fetch_signals_for_report_activity
            → report_safety_judge_activity
            → select_repository_activity (spawns Docker sandbox)

テストシグナルの送信

# Emit a single signal from the Zendesk fixture at offset 26
DEBUG=1 python manage.py emit_signals_from_fixture --type zendesk --team-id 1 --offset 26 --limit 1

# Clean up all signal data before re-emitting (avoids stale matches)
DEBUG=1 python manage.py cleanup_signals --team-id 1 --yes

# Check pipeline status
python manage.py signal_pipeline_status --team-id 1 --wait --expected-signals 1 --poll-interval 10

古い埋め込みが残り、レポートに誤って一致するのを防ぐため、再送信する前に必ずクリーンアップしてください。

Temporal ワークフローの監視

Temporal UI は http://localhost:8081 で実行されています。スクリプトでの調査には REST API が便利です。

最近のワークフローを一覧表示する

curl -s 'http://localhost:8081/api/v1/namespaces/default/workflows?query=ORDER+BY+StartTime+DESC&maximumPageSize=15' \
  | python3 -c "
import sys, json
for wf in json.load(sys.stdin).get('executions', []):
    info = wf['execution']
    status = wf['status'].replace('WORKFLOW_EXECUTION_STATUS_', '')
    print(f'{wf[\"startTime\"][:19]}  {status:20s} {wf[\"type\"][\"name\"]:35s} {info[\"workflowId\"][:90]}')
"

ワークフロー履歴を調べる

WF_ID="buffer-signals-1"  # or team-signal-grouping-v2-1, signals-report:1:<uuid>
curl -s "http://localhost:8081/api/v1/namespaces/default/workflows/$WF_ID/history?maximumPageSize=200" \
  | python3 -c "
import sys, json
for event in json.load(sys.stdin).get('history', {}).get('events', []):
    etype = event['eventType'].replace('EVENT_TYPE_', '')
    etime = event['eventTime'][:19]
    details = ''
    for key, attrs in event.items():
        if key.endswith('Attributes') and isinstance(attrs, dict):
            if 'activityType' in attrs: details = attrs['activityType'].get('name', '')
            elif 'signalName' in attrs: details = f'signal: {attrs[\"signalName\"]}'
            elif 'startToFireTimeout' in attrs: details = f'timer: {attrs[\"startToFireTimeout\"]}'
            elif 'failure' in attrs: details = f'FAILED: {attrs[\"failure\"].get(\"message\", \"\")[:200]}'
    if details: print(f'  {etime}  {etype:50s} {details}')
"

前回の実行を調べる（continued-as-new）

ワークフローが continued-as-new になった場合は、execution.runId クエリパラメーターを使用します。

curl -s "http://localhost:8081/api/v1/namespaces/default/workflows/$WF_ID/history?execution.runId=<run-id>&maximumPageSize=200"

サンドボックスエージェントのログを読む

エージェントのログは、オブジェクトストレージ（ローカル環境では MinIO）に JSONL ファイルとして保存されています。 ログ URL は TaskRun モデルにあります。

# In Django shell (python manage.py shell)
from products.tasks.backend.models import TaskRun
from posthog.storage import object_storage

# Find the most recent task run
run = TaskRun.objects.order_by("-created_at").first()
print(f"status: {run.status}, error: {run.error_message}")
print(f"log_url: {run.log_url}")

# Read the log
content = object_storage.read(run.log_url, missing_ok=True)

# Print last 3000 chars (most useful — shows what happened before failure)
print(content[-3000:])

ログは、次のようなエントリーを含む JSONL です。

{
  "type": "notification",
  "timestamp": "...",
  "notification": { "jsonrpc": "2.0", "method": "_posthog/console", "params": { "level": "debug", "message": "..." } }
}

ログ末尾で確認する主な項目：

• agentsh のネットワークイベント — DENY エントリーは、ブロックされたネットワーク呼び出しを示します
• _posthog/progress イベント — サンドボックスがどのセットアップ手順まで進んだかを示します
• _posthog/console のデバッグメッセージ — サンドボックスのプロビジョニング、クローン作成、エージェントの起動状況を示します

Docker サンドボックスコンテナの調査

# List running sandbox containers
docker ps --filter "name=task-sandbox" --format "table {{.Names}}\t{{.Status}}\t{{.Image}}"

# See processes inside a running sandbox
docker exec <container-name> ps aux

# Read the agent-server log inside the container (while it's still running)
docker exec <container-name> cat /tmp/agent-server.log

コンテナ名は task-sandbox-<task-id>-<random> で、posthog-sandbox-base イメージを使用します。 コンテナは一時的なもので、タスクの実行が完了すると削除されるため、実行中に調査してください。

よくある障害

SignalReport matching query does not exist

assign_and_emit_signal_activity が、存在しないレポートにシグナルを割り当てようとしました。 通常、cleanup_signals が埋め込みの削除に失敗した後、ClickHouse に古い埋め込みが残っていることが原因です。

根本原因： .env で CLICKHOUSE_DATABASE が設定されていません。クリーンアップコマンドは sync_execute を使用して CLICKHOUSE_DATABASE（デフォルトは default）に接続しますが、埋め込みテーブルは posthog データベースにあります。

修正方法： .env に CLICKHOUSE_DATABASE=posthog を追加し、ワーカーを再起動します。

古い埋め込みを手動でクリーンアップする：

curl -s 'http://localhost:8123/' --data-binary \
  "ALTER TABLE posthog.sharded_posthog_document_embeddings_text_embedding_3_small_1536 DELETE WHERE product = 'signals' AND team_id = 1 SETTINGS mutations_sync = 1"

埋め込みがクリーンアップされたことを確認する：

curl -s 'http://localhost:8123/' --data-binary \
  "SELECT count() FROM posthog.sharded_posthog_document_embeddings_text_embedding_3_small_1536 WHERE team_id = 1 AND product = 'signals'"

select_repository_activity で Run timed out due to inactivity が発生する

サンドボックスの Claude エージェントが TASKS_INACTIVITY_TIMEOUT_SECONDS を超えてアイドル状態になりました。未設定の場合は 2 時間のタイムアウトが適用されます。すぐに失敗させるには、ローカルで TASKS_INACTIVITY_TIMEOUT_SECONDS=30 を設定します。

診断方法： オブジェクトストレージからエージェントのログを読みます（上記参照）。ログ末尾を確認します。

1. agentsh によるネットワーク拒否 — DENY host.docker.internal は、MCP サーバーの URL が サンドボックスのネットワークポリシーによってブロックされていることを意味します。SIGNALS_REPO_DISCOVERY 環境のドメイン許可リストに host.docker.internal が含まれていません。
2. ログの内容がまったくない — サンドボックスの起動に失敗しています。Docker コンテナのログを確認します。
3. Claude API エラー — ANTHROPIC_API_KEY が有効か確認します。

buffer-signals がアイドル状態のままで、シグナルを受信しない

signal-emitter は完了しましたが、buffer-signals は submit_signal を受信しませんでした。 これは、エミッターがシグナルを送信した先の以前のバッファ実行がその後 continued-as-new になり、新しい実行が保留中のシグナルを引き継がずに開始した場合に発生します。シグナルを再送信してください。

クリーンアップ中に ClickHouse の埋め込みテーブルが「見つからない」と表示される

テーブルは posthog データベースに存在しますが、sync_execute は default データベースに対してクエリを実行しています。

# Verify tables exist
curl -s 'http://localhost:8123/' --data-binary "SHOW TABLES FROM posthog LIKE '%embed%'"

# Check current CLICKHOUSE_DATABASE setting
grep CLICKHOUSE_DATABASE .env

便利な管理コマンド

コマンド	用途
emit_signals_from_fixture	JSON フィクスチャからテストシグナルを送出します
DEBUG=1 cleanup_signals --team-id N --yes	すべてのシグナルデータを削除し、ワークフローを終了します
signal_pipeline_status --team-id N --wait	パイプラインの処理が完了するまで待機します
list_signal_reports --team-id N --signals --json	グループ化の結果を確認します
ingest_signals_json <file> --team-id N	処理済みのシグナルを JSON から取り込みます
ingest_report_json <file> --team-id N	調査済みのレポートを登録します（サンドボックスをスキップ）

主なファイルの場所

• パイプラインのワークフロー定義：products/signals/backend/temporal/
• バッファーワークフロー：products/signals/backend/temporal/buffer.py
• グループ化ワークフロー：products/signals/backend/temporal/grouping_v2.py
• レポート要約ワークフロー：products/signals/backend/temporal/summary.py
• Docker サンドボックスの実装：products/tasks/backend/logic/services/docker_sandbox.py
• サンドボックスの Dockerfile：products/tasks/backend/sandbox/images/
• エージェントログのポーリング：products/tasks/backend/logic/services/custom_prompt_internals.py
• クリーンアップコマンド：products/signals/backend/management/commands/cleanup_signals.py
• 管理コマンドのドキュメント：products/signals/backend/management/CLAUDE.md