debugging-signals-pipeline

What this skill does

# Debugging the signals pipeline

## Pipeline flow

```text
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)
```

## Emitting test signals

```bash
# 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
```

Always clean up before re-emitting to avoid stale embeddings causing phantom report matches.

## Monitoring Temporal workflows

The Temporal UI runs at `http://localhost:8081`. The REST API is useful for scripted inspection.

### List recent workflows

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

### Inspect workflow history

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

### Inspect a previous run (continued-as-new)

When a workflow has continued-as-new, use the `execution.runId` query param:

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

## Reading sandbox agent logs

Agent logs are stored in object storage (MinIO locally) as JSONL files.
The log URL is on the `TaskRun` model.

```python
# 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:])
```

The log is JSONL with entries like:

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

Key things to look for in the log tail:

- **agentsh network events** — `DENY` entries show blocked network calls
- **`_posthog/progress`** events — show which setup step the sandbox reached
- **`_posthog/console`** debug messages — show sandbox provisioning, cloning, agent startup

## Inspecting Docker sandbox containers

```bash
# 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
```

The container is named `task-sandbox-<task-id>-<random>` and uses the `posthog-sandbox-base` image.
Containers are ephemeral — they're removed after the task run completes, so inspect while running.

## Common failures

### `SignalReport matching query does not exist`

The `assign_and_emit_signal_activity` tried to assign a signal to a report that doesn't exist.
Usually caused by stale embeddings in ClickHouse after a `cleanup_signals` that failed to delete them.

**Root cause:** `CLICKHOUSE_DATABASE` not set in `.env`. The cleanup command uses `sync_execute`
which connects to the `CLICKHOUSE_DATABASE` (defaults to `default`), but the embedding tables
live in the `posthog` database.

**Fix:** Add `CLICKHOUSE_DATABASE=posthog` to `.env` and restart workers.

**Manual cleanup of stale embeddings:**

```bash
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"
```

**Verify embeddings are clean:**

```bash
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'"
```

### `Run timed out due to inactivity` on `select_repository_activity`

The sandbox Claude agent went idle for longer than `TASKS_INACTIVITY_TIMEOUT_SECONDS`. When unset
this falls back to a 2 hour timeout — set `TASKS_INACTIVITY_TIMEOUT_SECONDS=30` locally to force fast failures.

**Diagnosing:** Read the agent log from object storage (see above). Check the tail for:

1. **agentsh network denials** — `DENY host.docker.internal` means the MCP server URL is blocked
   by the sandbox network policy. The `SIGNALS_REPO_DISCOVERY` environment's domain allowlist
   doesn't include `host.docker.internal`.
2. **No log content at all** — sandbox failed to start, check Docker container logs.
3. **Claude API errors** — check if `ANTHROPIC_API_KEY` is valid.

### `buffer-signals` sits idle, never receives signals

The `signal-emitter` completed but `buffer-signals` never got the `submit_signal`.
This happens when the emitter sent the signal to a previous buffer run that then continued-as-new,
and the new run started fresh without the pending signal. Re-emit the signal.

### ClickHouse embedding tables "not found" during cleanup

The tables exist in the `posthog` database but `sync_execute` queries the `default` database.

```bash
# 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
```

## Useful management commands

| Command                                            | Purpose                                        |
| -------------------------------------------------- | ---------------------------------------------- |
| `emit_signals_from_fixture`                        | Emit test signals from JSON fixtures           |
| `DEBUG=1 cleanup_signals --team-id N --yes`        | Delete all signal data and terminate workflows |
| `signal_pipeline_status --team-id N --wait`