--- name: kunanonj/cursor-plugin-posthog-debugging-signals-pipeline source: https://app.decimal.ai/s/kunanonj-cursor-plugin-posthog-debugging-signals-pipeline@1/SKILL.md source_sha256: 1f8e82b3aea1 --- # 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: 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=&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 ps aux # Read the agent-server log inside the container (while it's still running) docker exec cat /tmp/agent-server.log ``` The container is named `task-sandbox--` 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` | Wait for pipeline to finish processing | | `list_signal_reports --team-id N --signals --json` | Inspect grouping results | | `ingest_signals_json --team-id N` | Ingest pre-processed signals from JSON | | `ingest_report_json --team-id N` | Seed a pre-researched report (skip sandbox) | ## Key file locations - Pipeline workflow definitions: `products/signals/backend/temporal/` - Buffer workflow: `products/signals/backend/temporal/buffer.py` - Grouping workflow: `products/signals/backend/temporal/grouping_v2.py` - Report summary workflow: `products/signals/backend/temporal/summary.py` - Docker sandbox implementation: `products/tasks/backend/services/docker_sandbox.py` - Sandbox Dockerfiles: `products/tasks/backend/sandbox/images/` - Agent log polling: `products/tasks/backend/services/custom_prompt_internals.py` - Cleanup command: `products/signals/backend/management/commands/cleanup_signals.py` - Management command docs: `products/signals/backend/management/CLAUDE.md`