Tb Quarantine Triage

name: tb-quarantine-triage description: > On-demand Tinybird quarantine investigator. Detects datasources with live quarantined rows, runs per-datasource root-cause analysis, presents a diagnosis plan for human review, then creates IN Jira tickets and git worktrees for approved datasources. No automation — always requires human sign-off before any Jira or Git action. allowed-tools: Bash, Read, Glob, Grep, Agent, AskUserQuestion, Skill, mcp__tinybird__list_datasources, mcp__tinybird__execute_query, mcp__plugin_context-mode_context-mode__ctx_execute, mcp__mcp-atlassian__searchJiraIssuesUsingJql, mcp__mcp-atlassian__createJiraIssue

Tinybird Quarantine Investigator

You are an on-demand quarantine analysis tool. You never auto-create tickets or branches without explicit human approval. You never read or log environment variables containing TOKEN, SECRET, or KEY.

The repo root is the current working directory (crowd.dev).

Phase 1 — Detect

Detection is two-pass. Do not include a FORMAT clause in any Tinybird query.

Important: Quarantine tables (e.g. activities_quarantine) are Tinybird system tables. They never appear in list_datasources output. You must build the candidate list from the regular datasource names and then probe each quarantine table directly via SQL.

Pass A — Candidate list from datasource listing

Call the list_datasources MCP tool. The result may be large and saved to a file path rather than returned inline. If a file path is returned, use ctx_execute (language: javascript) to read and parse the file — never load it raw into conversation. The response is a JSON array of objects with a name field (not a {datasources: [...]} wrapper).

Extract all datasource name values. Exclude any where the name:

• starts with raul_ or test_
• ends with _old or contains _backup
• ends with _MV or _MV_ds or _copy_ds, or matches *_MV_ds_\d+ (materialized views, numbered MV shards, and copy targets — they do not receive direct ingestion)
• ends with _sorted or _sorted_alt
• matches *_bucket_*_ds or *_deduplicated_cleaned_bucket_*_ds or *_enriched_deduplicated_bucket_*_ds or *_collection_bucket_*_ds or *_collection_deduplicated_cleaned_bucket_*_ds (sharded bucket datasources)

These are the candidate base DS names for Pass B.

Pass B — Live count per quarantine table

For each candidate from Pass A, query (in parallel where possible):

SELECT count() AS live_quarantined, max(insertion_date) AS last_seen
FROM {DS_NAME}_quarantine

Skip datasources where the query errors (many will — quarantine tables that have never received rows may still error). Keep only those with live_quarantined > 0.

If no datasources remain after filtering, report:

No live quarantined rows found. Nothing to investigate.

And stop.

Phase 2 — Diagnose (parallel subagents)

For each affected datasource, spawn one investigator subagent in parallel. Read the prompt from .claude/skills/tb-quarantine-triage/prompts/investigator.md and pass:

• DS_NAME — datasource name
• QUARANTINED_COUNT — count from Pass B
• LAST_SEEN — timestamp from Pass B
• REPO_ROOT — absolute path to the repo (resolve via pwd)

Each subagent returns a diagnosis bundle (JSON):

{
  "datasource": "<name>",
  "quarantined_count": <N>,
  "ds_total_rows": <N>,
  "last_seen": "<ISO8601>",
  "error_uniformity": "uniform | mixed",
  "dominant_error": {
    "column": "<col>",
    "error": "<normalized>",
    "occurrences": <N>,
    "pct_of_quarantined": <0-100>
  },
  "all_error_types": [
    {"column": "<col>", "error": "<normalized>", "occurrences": <N>, "first_seen": "<date>", "last_seen": "<date>"}
  ],
  "offending_columns": ["<col1>"],
  "schema_file": "<rel/path or 'not_found'>",
  "likely_producer_files": ["<rel/path>:<line>"],
  "postgres_source_table": "<table_name or 'not_found'>",
  "postgres_type_conflicts": [
    {"column": "<col>", "postgres_type": "<type>", "tinybird_type": "<type>", "safe_mapping": true}
  ],
  "downstream_impacts": [
    {"file": "<rel/path>", "usage": "<how the column is used>", "requires_change": true, "change_needed": "<description or null>"}
  ],
  "fix_type": "producer_cast | producer_guard | schema_add_column | schema_type_change | schema_type_change_with_downstream | ambiguous",
  "fix_description": "<what specifically needs to change, including file paths and the exact change>",
  "backfill_required": true | false,
  "backfill_risk": "low | medium | high | none",
  "fingerprint": "<sha256(ds|sorted_cols|dominant_error_sig)[:16]>"
}

Phase 3 — Present diagnosis plan

After all subagents complete, discard any bundles where dominant_error is null (zero-row race condition). Log each discarded datasource as "expired before diagnosis — no action needed." Only bundles with a non-null dominant_error proceed to the report and approval gate.

Present a structured report of the remaining bundles:

Summary table

| Datasource | Quarantined | Uniformity | Fix type | Backfill? |
|------------|-------------|------------|----------|-----------|
| activities |     47      |  uniform   | producer_cast | no  |
| members    |      3      |  mixed     | ambiguous | —        |

Per-datasource detail

For each datasource, print a section:

## <datasource_name>

**Quarantined rows:** <N> of <ds_total_rows> total (<pct>%)
**Last seen:** <last_seen>
**Error uniformity:** uniform | mixed

### Errors
| Column | Error | Occurrences | First seen | Last seen |
|--------|-------|-------------|------------|-----------|
| ...    | ...   | ...         | ...        | ...       |

### Root cause
<fix_description — full explanation including file paths and exact changes needed>

### Schema file
<schema_file>

### Producer files
<likely_producer_files — one per line>

### Backfill
<backfill_required> — risk: <backfill_risk>

### Recovery

Recovery path depends on fix type and whether the quarantined data itself is valid:

**Schema type change (`schema_type_change` / `schema_type_change_with_downstream`) — quarantined data is valid:**
1. Pause the Sequin sink for this datasource
2. Update `{DS_NAME}.datasource` with the type fix (and any downstream pipes if `schema_type_change_with_downstream`)
3. Delete and recreate the datasource: `tb push datasources/{DS_NAME}.datasource --force`
4. Backfill from Sequin — re-sends all rows including previously-quarantined ones (now accepted by the fixed schema)
5. Restart the sink

**Add column (`schema_add_column`) — quarantined data is valid:**
1. Add the missing column to `{DS_NAME}.datasource`
2. Push the updated schema: `tb push datasources/{DS_NAME}.datasource` (no delete/recreate needed — additive)
3. Backfill from Sequin — re-sends previously-quarantined rows so the new column is populated

**Producer fix (`producer_cast` / `producer_guard`) — quarantined data is bad:**
1. Fix the producer code (cast or null-guard the offending value)
2. Re-run the sync / worker that feeds this datasource
3. Corrected rows land in `{DS_NAME}` directly; the quarantined rows (which had the bad value) are discarded
4. No explicit backfill needed — the records already exist in the main datasource with their pre-fix values; the re-sync updates them

> ⚠ Quarantine retention is **1 month**. Rows older than 30 days are deleted automatically. Check `first_seen` — if > 3 weeks ago, flag as urgent.

If error_uniformity is "mixed", add a warning:

⚠ Multiple distinct error types. The fix may not resolve all quarantined rows. Review each error type before proceeding.

Phase 4 — Ticket approval gate

Ask the user which datasources to file tickets for:

Use AskUserQuestion with one multi-select question listing each affected datasource as an option. Include "None — diagnose only" as an option.

Do not proceed to Phase 5 until the user responds.

Phase 5 — Create Jira tickets

For each datasource approved by the user:

1. Skip any bundle where dominant_error is null (zero-row race condition — rows expired before diagnosis). Report it as resolved.
2. Search for an existing open IN ticket with label tb-quarantine-fp-<fingerprint> using mcp__mcp-atlassian__searchJiraIssuesUsingJql with JQL: project = IN AND labels = "tb-quarantine-fp-<fingerprint>" AND statusCategory != Done. If found, report the existing key and skip ticket creation.
3. Otherwise, create a ticket with createJiraIssue:

• project: IN
• issuetype: Bug
• summary: Tinybird quarantine: <datasource_name> — <dominant_error.column>: <dominant_error.error[:60]>
• description: structured Jira description (see format below)
• labels: ["tinybird", "quarantine", "tb-quarantine-fp-<fingerprint>"]
• priority: High

Jira description format:

h2. Problem

Rows are being quarantined in the *<datasource_name>* datasource.

*Quarantined rows:* <N> of <ds_total_rows> total
*Last seen:* <last_seen>
*Error uniformity:* <uniform | mixed>

h2. Error breakdown

|| Column || Error || Occurrences || First seen || Last seen ||
| <col> | <error> | <N> | <date> | <date> |

h2. Root cause

<fix_description>

h2. Affected files

*Schema:* <schema_file>
*Producers:*
<likely_producer_files — one per line>
*Downstream pipes/datasources requiring changes:*
<downstream_impacts where requires_change=true — one per line with change_needed description, or "None">

h2. Backfill

Backfill required: <yes | no> — Risk: <backfill_risk>

Capture the returned Jira key (e.g. IN-1234). After all tickets are created, print a summary:

## Tickets created

| Datasource | Ticket | Link |
|------------|--------|------|
| <name>     | <KEY>  | https://crowddev.atlassian.net/browse/<KEY> |

Phase 6 — Worktree approval gate

Ask the user which tickets to start working on now:

Use AskUserQuestion with one multi-select question listing each created ticket as an option (<KEY> — <datasource_name>). Include "None — tickets only" as an option.

Do not proceed to Phase 7 until the user responds.

Phase 7 — Create git worktrees

For each ticket approved by the user, create a worktree.

Use the superpowers:using-git-worktrees skill:

• branch: fix/<JIRA_KEY>-tb-quarantine-<datasource_name>
• If the skill is not available, fall back to:

  # Exclude .worktrees locally without modifying tracked .gitignore
  grep -qxF '.worktrees' .git/info/exclude 2>/dev/null || echo '.worktrees' >> .git/info/exclude
  git worktree add .worktrees/<JIRA_KEY> -b fix/<JIRA_KEY>-tb-quarantine-<datasource_name> origin/main
  

After all worktrees are created, print a summary per datasource:

## <datasource_name> → <JIRA_KEY>

Ticket: <JIRA_KEY> (https://crowddev.atlassian.net/browse/<JIRA_KEY>)
Worktree: .worktrees/<JIRA_KEY>
Branch: fix/<JIRA_KEY>-tb-quarantine-<datasource_name>

Next steps:
<fix_description — verbatim from the diagnosis bundle>

Error handling

• If a subagent fails for one datasource, report the failure inline and continue with the rest.
• If Jira creation fails, report the error and exclude that datasource from the Phase 6 gate.
• If worktree creation fails (branch already exists, dirty state, etc.), report the error — do not force-push or reset.