debugging-signals-pipeline

调试 signals 流水线

流水线流程

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}')
"

检查之前的运行（以新运行继续）

工作流以新运行继续后，请使用 execution.runId 查询参数：

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

查看沙盒 Agent 日志

Agent 日志以 JSONL 文件的形式存储在对象存储中（本地使用 MinIO）。 日志 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 调试消息 — 显示沙盒配置、克隆和 Agent 启动情况

检查 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 Agent 空闲时间超过了 TASKS_INACTIVITY_TIMEOUT_SECONDS。未设置时， 超时时间默认为 2 小时 — 在本地设置 TASKS_INACTIVITY_TIMEOUT_SECONDS=30 可触发快速失败。

诊断： 从对象存储读取 Agent 日志（见上文）。检查日志末尾的内容：

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。 这是因为发射器将信号发送给了之前的 buffer 运行，而该运行随后以新运行继续； 新运行启动时没有待处理信号。请重新发出信号。

清理期间 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 fixtures 发出测试信号
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
• 沙箱 Dockerfiles：products/tasks/backend/sandbox/images/
• Agent 日志轮询：products/tasks/backend/logic/services/custom_prompt_internals.py
• 清理命令：products/signals/backend/management/commands/cleanup_signals.py
• 管理命令文档：products/signals/backend/management/CLAUDE.md