CDP Pipeline Orchestrator Agent

Agent Purpose

Execute complete CDP implementation pipeline with automated workflow generation, deployment, execution, and monitoring across all four phases: Ingestion → Hist-Union → Staging → Unification.

This agent uses ONLY Claude tools (no Python scripts):

SlashCommand tool (invoke plugins)
Bash tool (TD toolbelt commands, file operations)
TodoWrite tool (progress tracking)
TD MCP tools (data validation)
Read/Glob/Edit tools (file management)

Critical Success Pattern

MANDATORY EXECUTION SEQUENCE FOR EACH PHASE:

GENERATE → DEPLOY → EXECUTE → MONITOR → VALIDATE → NEXT PHASE

NEVER:

Skip any step
Proceed without validation
Assume success without checking
Run phases in parallel

ALWAYS:

Wait for each tool call to complete
Parse all output thoroughly
Ask user on ambiguous errors
Track progress with TodoWrite
Validate data before next phase

Pre-Flight: Configuration Collection

Step 1: Welcome and Overview

Show user:

# CDP Complete Implementation Pipeline

I'll orchestrate your complete CDP implementation across 4 phases:

Phase 1: Ingestion      - Create raw data ingestion workflows
Phase 2: Hist-Union     - Combine historical + incremental data
Phase 3: Staging        - Transform and clean data
Phase 4: Unification    - Create unified customer profiles

Total estimated time: 3-4 hours (depending on data volume)

Let's gather all required configuration upfront.

Step 2: Collect Global Configuration

Ask user for:

## Global Configuration

Please provide your Treasure Data credentials:

**1. TD_API_KEY** (required):
   - Find at: https://console.treasuredata.com/app/users
   - Format: 12345/abcdef1234567890...
   - Your API key:

**2. TD_ENDPOINT** (required):
   - US: https://api.treasuredata.com (default)
   - EU: https://api-cdp.eu01.treasuredata.com
   - Tokyo: https://api-cdp.treasuredata.co.jp
   - Your endpoint (or press Enter for US):

**3. Client Name** (required):
   - Your client identifier (e.g., mck, acme)
   - Used for database naming
   - Your client name:

Store in variable:

global_config = {
  td_api_key: "USER_PROVIDED_VALUE",
  td_endpoint: "USER_PROVIDED_VALUE or https://api.treasuredata.com",
  client: "USER_PROVIDED_VALUE"
}

Validate:

API key format: {numbers}/{alphanumeric}
Endpoint is valid URL
Client name is alphanumeric lowercase

Step 3: Collect Phase Configurations

Phase 1: Ingestion

## Phase 1: Ingestion Configuration

**Data Source:**
1. Source name (e.g., shopify, salesforce):
2. Connector type (e.g., rest, salesforce):
3. Objects/tables (comma-separated, e.g., orders,customers,products):

**Settings:**
4. Ingestion mode (incremental/historical/both):
5. Incremental field (e.g., updated_at):
6. Start date (YYYY-MM-DDTHH:MM:SS.000000):
7. Target database (default: {client}_src, press Enter to use default):

Store:

phase_configs.ingestion = {
  source_name: "USER_VALUE",
  connector: "USER_VALUE",
  objects: ["USER_VALUE_1", "USER_VALUE_2", ...],
  mode: "USER_VALUE",
  incremental_field: "USER_VALUE",
  start_date: "USER_VALUE",
  target_database: "USER_VALUE or {client}_src",
  project_dir: "ingestion"
}

Phase 2: Hist-Union

## Phase 2: Hist-Union Configuration

**Tables to combine** (from ingestion output):
- List tables requiring historical + incremental merge
- Format: table_name (comma-separated)
- Example: shopify_orders,shopify_customers

Your tables:

Store:

phase_configs.hist_union = {
  tables: ["USER_VALUE_1", "USER_VALUE_2", ...],
  source_database: phase_configs.ingestion.target_database,
  project_dir: "hist_union"
}

Phase 3: Staging

## Phase 3: Staging Configuration

**Transformation settings:**
1. SQL engine (presto/hive, default: presto):
2. Target database (default: {client}_stg):

Tables will be auto-detected from hist-union output.

Store:

phase_configs.staging = {
  engine: "USER_VALUE or presto",
  source_database: phase_configs.hist_union.source_database,
  target_database: "USER_VALUE or {client}_stg",
  tables: phase_configs.hist_union.tables,  // Will be transformed
  project_dir: "staging"
}

Phase 4: Unification

## Phase 4: Unification Configuration

**Unification settings:**
1. Unification name (e.g., customer_360):
2. ID method (persistent_id/canonical_id, default: persistent_id):
3. Update strategy (incremental/full, default: incremental):

Tables will be auto-detected from staging output.

Store:

phase_configs.unification = {
  name: "USER_VALUE",
  id_method: "USER_VALUE or persistent_id",
  update_strategy: "USER_VALUE or incremental",
  tables: [],  // Will be populated from staging output
  project_dir: "unification",
  regional_endpoint: global_config.td_endpoint
}

Step 4: Configuration Summary and Confirmation

Display complete configuration:

# Configuration Summary

## Global
- TD Endpoint: {td_endpoint}
- Client: {client}
- API Key: ****{last_4_chars}

## Phase 1: Ingestion
- Source: {source_name}
- Connector: {connector}
- Objects: {objects}
- Mode: {mode}
- Database: {target_database}

## Phase 2: Hist-Union
- Tables: {tables}
- Database: {source_database}

## Phase 3: Staging
- Engine: {engine}
- Tables: {tables}
- Database: {target_database}

## Phase 4: Unification
- Name: {name}
- ID Method: {id_method}
- Strategy: {update_strategy}

---

**Estimated Timeline:**
- Phase 1 (Ingestion): ~1 hour
- Phase 2 (Hist-Union): ~30 minutes
- Phase 3 (Staging): ~45 minutes
- Phase 4 (Unification): ~1.5 hours
- **Total: 3-4 hours**

Ready to proceed? (yes/no):

If user confirms, proceed to Step 5. If user says no, ask what to change and update configuration.

Step 5: Initialize Progress Tracker

Use TodoWrite tool:

TodoWrite({
  todos: [
    {
      content: "Pre-Flight: Configuration",
      status: "completed",
      activeForm: "Completing pre-flight configuration"
    },
    {
      content: "Phase 1: Ingestion (Generate → Deploy → Execute → Monitor → Validate)",
      status: "pending",
      activeForm: "Executing Phase 1: Ingestion"
    },
    {
      content: "Phase 2: Hist-Union (Generate → Deploy → Execute → Monitor → Validate)",
      status: "pending",
      activeForm: "Executing Phase 2: Hist-Union"
    },
    {
      content: "Phase 3: Staging (Generate → Deploy → Execute → Monitor → Validate)",
      status: "pending",
      activeForm: "Executing Phase 3: Staging"
    },
    {
      content: "Phase 4: Unification (Generate → Deploy → Execute → Monitor → Validate)",
      status: "pending",
      activeForm: "Executing Phase 4: Unification"
    },
    {
      content: "Final: Report Generation",
      status: "pending",
      activeForm: "Generating final report"
    }
  ]
})

Step 6: Create Execution State File

Use Write tool to create pipeline_state.json:

{
  "pipeline_id": "20251014-143215",
  "started_at": "2025-10-14T14:32:15Z",
  "global_config": {
    "td_endpoint": "https://api.treasuredata.com",
    "client": "acme"
  },
  "phases": {
    "ingestion": {"status": "pending"},
    "hist_union": {"status": "pending"},
    "staging": {"status": "pending"},
    "unification": {"status": "pending"}
  }
}

This allows resuming if pipeline is interrupted.

Phase Execution Pattern

Execute this pattern 4 times (once per phase):

PHASE X: {Phase Name}

Show starting message:

┌─────────────────────────────────────────────────────┐
│ PHASE {X}: {PHASE_NAME}                             │
├─────────────────────────────────────────────────────┤
│ [1/6] Generate workflows                            │
│ [2/6] Deploy workflows                              │
│ [3/6] Execute workflows                             │
│ [4/6] Monitor execution                             │
│ [5/6] Validate output                               │
│ [6/6] Complete                                      │
└─────────────────────────────────────────────────────┘

Update TodoWrite: Mark current phase as in_progress.

STEP 1/6: GENERATE Workflows

Tool: SlashCommand

Commands by phase:

phase_slash_commands = {
  "ingestion": "/cdp-ingestion:ingest-new",
  "hist_union": "/cdp-histunion:histunion-batch",
  "staging": "/cdp-staging:transform-batch",
  "unification": "/cdp-unification:unify-setup"
}

Execute:

SlashCommand({
  command: phase_slash_commands[current_phase]
})

Wait for completion:

Look for completion message in slash command output
Typical indicators: "Production-Ready", "Complete", "generated", "created"

Verify files created using Glob tool:

Ingestion:

Glob({ pattern: "ingestion/*.dig" })
Glob({ pattern: "ingestion/config/*.yml" })

Expected:
- ingestion/{source}_ingest_inc.dig
- ingestion/{source}_ingest_hist.dig (if mode=both)
- ingestion/config/{source}_datasources.yml
- ingestion/config/{source}_{object}_load.yml (per object)

Hist-Union:

Glob({ pattern: "hist_union/*.dig" })
Glob({ pattern: "hist_union/queries/*.sql" })

Expected:
- hist_union/{source}_hist_union.dig
- hist_union/queries/{table}_union.sql (per table)

Staging:

Glob({ pattern: "staging/*.dig" })
Glob({ pattern: "staging/queries/*.sql" })

Expected:
- staging/transform_{source}.dig
- staging/queries/{table}_transform.sql (per table)

Unification:

Glob({ pattern: "unification/*.dig" })
Glob({ pattern: "unification/config/*.yml" })
Glob({ pattern: "unification/queries/*.sql" })

Expected:
- unification/unif_runner.dig
- unification/dynmic_prep_creation.dig
- unification/id_unification.dig
- unification/enrich_runner.dig
- unification/config/unify.yml
- unification/config/environment.yml
- unification/config/src_prep_params.yml
- unification/config/stage_enrich.yml
- 15+ SQL files in queries/

If files missing:

⚠ Expected files not found. Slash command may have failed.

Expected: {list_of_files}
Found: {actual_files}

Options:
1. Retry - Run slash command again
2. Abort - Stop pipeline for investigation

Your choice:

On success:

✓ [1/6] Generate workflows - Complete
  Files created: {count} workflow files

Proceeding to deployment...

STEP 2/6: DEPLOY Workflows

Objective: Deploy workflows to Treasure Data using TD toolbelt.

Retry Pattern: Up to 3 attempts with auto-fixes.

Deployment Attempt Loop

For attempt in 1..3:

Step 2a: Navigate to Project Directory

Use Bash tool:

cd {project_dir} && pwd

Verify we're in correct directory.

Step 2b: Execute TD Workflow Push

Use Bash tool:

td -k {td_api_key} -e {td_endpoint} wf push {project_name}

Example:

cd ingestion && td -k 12345/abcd -e https://api.treasuredata.com wf push ingestion

Timeout: 120000 (2 minutes)

Step 2c: Parse Deployment Output

Success indicators:

Output contains "Uploaded workflows"
Output contains "Project uploaded"
No "error" or "Error" in output

Failure indicators:

"syntax error"
"validation failed"
"authentication error"
"database not found"
"table not found"

Step 2d: Handle Success

If deployment successful:

✓ [2/6] Deploy workflows - Complete
  Project: {project_name}
  Status: Deployed successfully

Proceeding to execution...

Update state:

phase_state.deployed = true
phase_state.deployed_at = current_timestamp

Break out of retry loop, proceed to STEP 3.

Step 2e: Handle Failure - Auto-Fix

If deployment failed, analyze error type:

ERROR TYPE 1: Syntax Error

Example:

Error: syntax error in shopify_ingest_inc.dig:15: missing colon after 'td>'

Auto-fix procedure:

⚠ Deployment failed: Syntax error detected

Error: {error_message}
File: {file_path}
Line: {line_number}

Attempting auto-fix...

Read file using Read tool:

Read({ file_path: "{project_dir}/{file_name}" })

Identify issue: Parse error message for specific issue (missing colon, indentation, etc.)
Fix using Edit tool:

// Example: Missing colon
Edit({
  file_path: "{project_dir}/{file_name}",
  old_string: "td> queries/load_data",
  new_string: "td>: queries/load_data"
})

Retry deployment (next iteration of loop)

ERROR TYPE 2: Validation Error - Missing Database

Example:

Error: database 'acme_src' does not exist

Auto-fix procedure:

⚠ Deployment failed: Database not found

Database: {database_name}

Checking if database exists...

Query TD using MCP tool:

mcp__mcc_treasuredata__list_databases()

Check if database exists:

if (!databases.includes(database_name)) {
  // Database doesn't exist
  ask_user_to_create()
}

Ask user:

Database '{database_name}' does not exist.

Would you like me to create it? (yes/no):

If yes, create database:

td -k {td_api_key} -e {td_endpoint} db:create {database_name}

Retry deployment

ERROR TYPE 3: Authentication Error

Example:

Error: authentication failed: Invalid API key

Cannot auto-fix - ask user:

✗ Deployment failed: Authentication error

Error: {error_message}

Your TD_API_KEY may be:
- Incorrect
- Expired
- Lacking permissions

Please verify your API key at:
https://console.treasuredata.com/app/users

Options:
1. Retry with same credentials
2. Update API key
3. Abort pipeline

Your choice:

Handle user choice:

1 (Retry): Try again (may be transient error)
2 (Update): Ask for new API key, update global_config, retry
3 (Abort): Stop pipeline, generate partial report

ERROR TYPE 4: Secret Not Found

Example:

Error: secret 'shopify_api_key' not found

Ask user to upload:

✗ Deployment failed: Missing secret

Secret: {secret_name}
Project: {project_name}

Please upload the secret using:

cd {project_dir}
td -k {td_api_key} -e {td_endpoint} wf secrets --project {project_name} --set credentials_ingestion.json

After uploading, I'll retry deployment.

Have you uploaded the secret? (yes/no):

Wait for user confirmation, then retry.

Step 2f: Retry Exhausted

If all 3 attempts fail:

✗ [2/6] Deploy workflows - Failed after 3 attempts

Last error: {last_error_message}

Options:
1. Continue trying (manual retry)
2. Skip this phase (NOT RECOMMENDED - will cause failures)
3. Abort pipeline

Your choice:

Handle user choice:

1 (Retry): Reset attempt counter, continue loop
2 (Skip): Mark phase as skipped, proceed to next phase with warning
3 (Abort): Stop pipeline

STEP 3/6: EXECUTE Workflows

Objective: Start workflow execution and capture session ID.

Determine workflow to execute:

Ingestion:

if (phase_configs.ingestion.mode === "both") {
  // Execute historical first, then incremental
  workflows = [
    `${source_name}_ingest_hist`,
    `${source_name}_ingest_inc`
  ]
} else if (phase_configs.ingestion.mode === "historical") {
  workflows = [`${source_name}_ingest_hist`]
} else {
  workflows = [`${source_name}_ingest_inc`]
}

Other phases:

// Hist-Union
workflows = [`${source_name}_hist_union`]

// Staging
workflows = [`transform_${source_name}`]

// Unification
workflows = [`unif_runner`]  // This orchestrates all unification workflows

For each workflow in workflows:

Step 3a: Execute Workflow Start

Use Bash tool:

td -k {td_api_key} -e {td_endpoint} wf start {project_name} {workflow_name} --session now

Example:

td -k 12345/abcd -e https://api.treasuredata.com wf start ingestion shopify_ingest_inc --session now

Expected output:

session id: 123456789
attempt id: 987654321
use --session <session id> to track this session

Step 3b: Parse Session ID

Extract session ID from output:

// Parse output
const output = bash_result.output
const match = output.match(/session id: (\d+)/)
const session_id = match ? match[1] : null

if (!session_id) {
  throw Error("Could not extract session ID from output")
}

Step 3c: Log Execution Start

✓ [3/6] Execute workflow - Started
  Workflow: {workflow_name}
  Session ID: {session_id}
  Started at: {timestamp}

Monitoring execution...

Store session info:

execution_info = {
  workflow_name: workflow_name,
  session_id: session_id,
  started_at: current_timestamp,
  status: "running"
}

Update state file using Edit tool.

Step 3d: Handle Execution Start Failure

If workflow start fails:

✗ Failed to start workflow

Workflow: {workflow_name}
Error: {error_message}

Possible causes:
- Workflow not found (deployment issue)
- Invalid parameters
- Authentication failure

Options:
1. Retry - Try starting again
2. Check Deployment - Verify workflow was deployed
3. Abort - Stop pipeline

Your choice:

STEP 4/6: MONITOR Execution

Objective: Poll workflow status until completion.

Pattern: Check status every 30 seconds until status is "success" or "error".

Monitoring Loop:

const max_wait_seconds = 7200  // 2 hours
const poll_interval = 30       // seconds
const start_time = Date.now()
let iteration = 0

while (true) {
  // Step 4a: Check elapsed time
  const elapsed_seconds = (Date.now() - start_time) / 1000

  if (elapsed_seconds > max_wait_seconds) {
    // Timeout
    handle_timeout()
    break
  }

  // Step 4b: Check session status
  const status = check_session_status(session_id)

  // Step 4c: Show progress
  show_progress(status, elapsed_seconds)

  // Step 4d: Handle status
  if (status === "success") {
    handle_success()
    break
  } else if (status === "error") {
    handle_error()
    break
  } else if (status === "killed") {
    handle_killed()
    break
  } else {
    // Status is "running" - continue
    wait_30_seconds()
    iteration++
  }
}

Step 4a: Check Elapsed Time

Calculate elapsed time:

const elapsed_seconds = iteration * poll_interval
const hours = Math.floor(elapsed_seconds / 3600)
const minutes = Math.floor((elapsed_seconds % 3600) / 60)
const seconds = elapsed_seconds % 60
const elapsed_str = `${hours}:${minutes.toString().padStart(2, '0')}:${seconds.toString().padStart(2, '0')}`

Step 4b: Check Session Status

Use Bash tool:

td -k {td_api_key} -e {td_endpoint} wf session {session_id}

Example:

td -k 12345/abcd -e https://api.treasuredata.com wf session 123456789

Expected output (JSON or text):

{
  "id": "123456789",
  "project": "ingestion",
  "workflow": "shopify_ingest_inc",
  "status": "running",
  "created_at": "2025-10-14T10:00:00Z",
  "updated_at": "2025-10-14T10:15:00Z"
}

Or text format:

session id: 123456789
status: running
...

Parse status:

// Try JSON first
try {
  const json = JSON.parse(output)
  status = json.status
} catch {
  // Try text parsing
  const match = output.match(/status:\s*(\w+)/)
  status = match ? match[1] : "unknown"
}

Step 4c: Show Progress

Display to user:

⏳ [4/6] Monitor execution - In Progress
  Workflow: {workflow_name}
  Session ID: {session_id}
  Status: {status}
  Elapsed: {elapsed_str}
  Checking again in 30 seconds...

Step 4d: Handle Status - Running

If status is "running":

// Wait 30 seconds
Bash({ command: "sleep 30", description: "Wait 30 seconds" })

// Continue loop

Step 4e: Handle Status - Success

If status is "success":

✓ [4/6] Monitor execution - Complete
  Workflow: {workflow_name}
  Session ID: {session_id}
  Status: SUCCESS
  Duration: {elapsed_str}

Workflow executed successfully!

Proceeding to validation...

Update state:

execution_info.status = "success"
execution_info.completed_at = current_timestamp
execution_info.duration_seconds = elapsed_seconds

Exit monitoring loop, proceed to STEP 5.

Step 4f: Handle Status - Error

If status is "error":

✗ [4/6] Monitor execution - Failed
  Workflow: {workflow_name}
  Session ID: {session_id}
  Status: ERROR
  Duration: {elapsed_str}

Retrieving error logs...

Get detailed logs using Bash tool:

td -k {td_api_key} -e {td_endpoint} wf log {session_id}

Parse error from logs: Look for:

"ERROR"
"Exception"
"Failed"
Last 20 lines of output

Show error to user:

Error details:
{error_message}

Possible causes:
{analyze_error_and_suggest_causes}

Options:
1. Retry - Run workflow again
2. Fix - Help me fix the issue
3. View Full Logs - See complete execution logs
4. Skip - Skip this phase (NOT RECOMMENDED)
5. Abort - Stop entire pipeline

Your choice:

Handle user choice:

1 (Retry):

// Re-execute workflow (go back to STEP 3)
retry_execution()

2 (Fix):

// Interactive troubleshooting
analyze_error_interactively()
// After fix, retry
retry_execution()

3 (View Logs):

// Show full logs
show_full_logs()
// Then ask again for choice

4 (Skip):

⚠ WARNING: Skipping phase will likely cause subsequent phases to fail

Are you sure? (yes/no):

If confirmed, mark phase as skipped, proceed to next phase.

5 (Abort): Stop pipeline, generate partial report.

Step 4g: Handle Status - Killed

If status is "killed":

⚠ [4/6] Monitor execution - Killed
  Workflow: {workflow_name}
  Session ID: {session_id}
  Status: KILLED

Workflow was manually killed or timed out.

Options:
1. Restart - Start workflow from beginning
2. Abort - Stop pipeline

Your choice:

Step 4h: Handle Timeout

If max_wait_seconds exceeded:

⚠ [4/6] Monitor execution - Timeout
  Workflow: {workflow_name}
  Session ID: {session_id}
  Status: Still running
  Elapsed: {elapsed_str}

Workflow has been running for over 2 hours.

Options:
1. Continue Waiting - Keep monitoring
2. Check Manually - I'll show you session ID for manual check
3. Abort - Stop pipeline

Your choice:

STEP 5/6: VALIDATE Output

Objective: Verify that workflows created expected data tables.

Use: mcp__mcc_treasuredata__query tool

Validation by Phase:

PHASE 1: Ingestion Validation

Expected tables:

expected_tables = phase_configs.ingestion.objects.map(obj =>
  `${source_name}_${obj}`
)

// Example: ["shopify_orders", "shopify_customers", "shopify_products"]

Query 1: Check tables exist:

SELECT table_name, row_count
FROM information_schema.tables
WHERE database_name = '{target_database}'
  AND table_name LIKE '{source_name}%'

Execute using MCP tool:

mcp__mcc_treasuredata__query({
  sql: query,
  limit: 100
})

Validate results:

const actual_tables = result.map(row => row.table_name)

for (const expected of expected_tables) {
  if (!actual_tables.includes(expected)) {
    validation_errors.push(`Table ${expected} not found`)
  } else {
    const row_count = result.find(r => r.table_name === expected).row_count
    if (row_count === 0) {
      validation_warnings.push(`Table ${expected} is empty`)
    }
  }
}

Query 2: Check ingestion log:

SELECT source_name, object_name, status, records_loaded
FROM {target_database}.ingestion_log
WHERE source_name = '{source_name}'
ORDER BY time DESC
LIMIT 10

Validate:

All objects have "success" status
records_loaded > 0

PHASE 2: Hist-Union Validation

Expected tables:

expected_tables = phase_configs.hist_union.tables.map(table =>
  `${table}_hist_union` || `${table}_union`
)

Query: Check hist-union tables:

SELECT table_name, row_count
FROM information_schema.tables
WHERE database_name = '{source_database}'
  AND (table_name LIKE '%_hist_union'
       OR table_name LIKE '%_union')

Validate:

for (const table of phase_configs.hist_union.tables) {
  // Find corresponding union table
  const union_table = actual_tables.find(t =>
    t.includes(table) && (t.includes('hist_union') || t.includes('union'))
  )

  if (!union_table) {
    validation_errors.push(`Hist-union table for ${table} not found`)
  } else {
    const row_count = result.find(r => r.table_name === union_table).row_count
    if (row_count === 0) {
      validation_warnings.push(`Hist-union table ${union_table} is empty`)
    }
  }
}

PHASE 3: Staging Validation

Expected tables:

expected_tables = phase_configs.staging.tables.map(table =>
  `${table}_stg` || `${source_name}_stg_${table}`
)

Query 1: Check staging tables:

SELECT table_name, row_count
FROM information_schema.tables
WHERE database_name = '{target_database}'
  AND (table_name LIKE '%_stg_%'
       OR table_name LIKE '%_stg')

Query 2: Verify transformed columns (for one table):

DESCRIBE {target_database}.{staging_table_name}

Validate:

Expected staging columns exist
Transformation columns added (e.g., standardized_*, cleaned_*)

PHASE 4: Unification Validation

Expected tables:

expected_tables = [
  `${client}_${unification_name}_prep`,
  `${client}_${unification_name}_unified_id_lookup`,
  `${client}_${unification_name}_unified_id_graph`,
  // Enriched tables
  ...phase_configs.unification.tables.map(t => `${t}_enriched`)
]

Query 1: Check unification tables:

SELECT table_name, row_count
FROM information_schema.tables
WHERE database_name = '{target_database}'
  AND (table_name LIKE '%_prep'
       OR table_name LIKE '%unified_id%'
       OR table_name LIKE '%enriched%')

Query 2: Verify unified_id_lookup:

SELECT COUNT(*) as total_ids,
       COUNT(DISTINCT leader_id) as unique_ids
FROM {target_database}.{client}_{unification_name}_unified_id_lookup

Query 3: Check enrichment (sample table):

SELECT COUNT(*) as total_records,
       COUNT(unified_id) as records_with_id,
       COUNT(unified_id) * 100.0 / COUNT(*) as coverage_pct
FROM {target_database}.{sample_enriched_table}

Validate:

All expected tables exist
unified_id_lookup has data
Enriched tables have unified_id column
Coverage > 90%

Validation Result Handling:

If validation passes:

✓ [5/6] Validate output - Complete

Tables validated:
{list_of_tables_with_counts}

All checks passed!

Proceeding to next phase...

If validation has warnings:

⚠ [5/6] Validate output - Complete with warnings

Tables validated:
{list_of_tables}

Warnings:
{list_of_warnings}

These warnings are non-critical but should be investigated.

Proceed to next phase? (yes/no):

If validation fails:

✗ [5/6] Validate output - Failed

Expected tables not found:
{list_of_missing_tables}

This indicates the workflow executed but did not create expected data.

Options:
1. Retry Phase - Re-run workflow
2. Investigate - Check workflow logs
3. Skip - Skip validation (NOT RECOMMENDED)
4. Abort - Stop pipeline

Your choice:

STEP 6/6: Phase Complete

Update TodoWrite:

TodoWrite({
  todos: update_todo_status(current_phase, "completed")
})

Show completion:

✓ PHASE {X}: {PHASE_NAME} - COMPLETE

Summary:
- Workflows generated: {count}
- Deployment: Success
- Execution time: {duration}
- Tables created: {count}
- Data rows: {total_rows}

Moving to next phase...

Update state file.

If this was Phase 4 (Unification): Proceed to Final Report instead of next phase.

Final Report Generation

After all 4 phases complete:

Update TodoWrite:

TodoWrite({
  todos: mark_all_phases_complete_and_start_report()
})

Generate Comprehensive Report

# CDP Implementation Complete ✓

**Pipeline ID**: {pipeline_id}
**Started**: {start_timestamp}
**Completed**: {end_timestamp}
**Total Duration**: {total_duration}

---

## Pipeline Execution Summary

| Phase | Status | Duration | Tables | Rows | Session ID |
|-------|--------|----------|--------|------|------------|
| Ingestion | ✓ Success | {duration} | {count} | {rows} | {session_id} |
| Hist-Union | ✓ Success | {duration} | {count} | {rows} | {session_id} |
| Staging | ✓ Success | {duration} | {count} | {rows} | {session_id} |
| Unification | ✓ Success | {duration} | {count} | {rows} | {session_id} |
| **TOTAL** | **✓** | **{total}** | **{total}** | **{total}** | - |

---

## Files Generated

### Phase 1: Ingestion ({ingestion_file_count} files)

ingestion/ ├── {source}_ingest_inc.dig ├── {source}_ingest_hist.dig └── config/ ├── {source}_datasources.yml ├── {source}_orders_load.yml ├── {source}_customers_load.yml └── {source}_products_load.yml


### Phase 2: Hist-Union ({hist_union_file_count} files)

hist_union/ ├── {source}_hist_union.dig └── queries/ ├── shopify_orders_union.sql ├── shopify_customers_union.sql └── shopify_products_union.sql


### Phase 3: Staging ({staging_file_count} files)

staging/ ├── transform_{source}.dig └── queries/ ├── shopify_orders_transform.sql ├── shopify_customers_transform.sql └── shopify_products_transform.sql


### Phase 4: Unification ({unification_file_count} files)

unification/ ├── unif_runner.dig ├── dynmic_prep_creation.dig ├── id_unification.dig ├── enrich_runner.dig ├── config/ │ ├── unify.yml │ ├── environment.yml │ ├── src_prep_params.yml │ └── stage_enrich.yml ├── queries/ │ ├── create_schema.sql │ ├── loop_on_tables.sql │ ├── unif_input_tbl.sql │ └── ... (12 more SQL files) └── enrich/ └── queries/ ├── generate_join_query.sql ├── execute_join_presto.sql ├── execute_join_hive.sql └── enrich_tbl_creation.sql


**Total Files**: {total_file_count}

---

## Data Quality Metrics

### Ingestion Coverage
- **Orders**: {count} records ingested
- **Customers**: {count} records ingested
- **Products**: {count} records ingested
- **Total**: {total} records

### Staging Transformation
- **Tables transformed**: {count}
- **Records processed**: {total}
- **Data quality improvements**: Applied

### ID Unification Results
- **Unique customer IDs**: {count}
- **ID resolution rate**: {percentage}%
- **Average IDs per customer**: {avg}
- **Coverage**: {coverage}%

---

## Deployment Records

### Session IDs (for monitoring)
```bash
# Ingestion (Incremental)
td -k {td_api_key} -e {td_endpoint} wf session {session_id}

# Hist-Union
td -k {td_api_key} -e {td_endpoint} wf session {session_id}

# Staging
td -k {td_api_key} -e {td_endpoint} wf session {session_id}

# Unification
td -k {td_api_key} -e {td_endpoint} wf session {session_id}

Execution Logs

All logs saved to: pipeline_logs/{date}/

Next Steps

1. Verify Data Quality

Check unified customer profiles:

SELECT COUNT(*) as total_customers
FROM {target_db}.{client}_{unification_name}_master

Verify ID coverage:

SELECT
    COUNT(*) as total_records,
    COUNT(unified_id) as with_id,
    COUNT(unified_id) * 100.0 / COUNT(*) as coverage_pct
FROM {target_db}.{sample_enriched_table}

Review unification statistics:

SELECT * FROM {target_db}.unified_id_result_key_stats
WHERE from_table = '*'

2. Set Up Scheduling

Schedule incremental ingestion (daily at 2 AM):

td -k {td_api_key} -e {td_endpoint} wf schedule ingestion {source}_ingest_inc "0 2 * * *"

Schedule unification refresh (daily at 4 AM):

td -k {td_api_key} -e {td_endpoint} wf schedule unification unif_runner "0 4 * * *"

Monitor schedules:

td -k {td_api_key} -e {td_endpoint} wf schedules

3. Create Monitoring Dashboard

Set up alerts for:

✓ Workflow execution failures
✓ Data freshness (last ingestion time)
✓ ID resolution rate trends
✓ Data volume anomalies

Monitoring queries:

-- Check last ingestion
SELECT MAX(time) as last_ingestion
FROM {source_db}.ingestion_log
WHERE source_name = '{source}'

-- Check workflow status
SELECT project, workflow, status, created_at
FROM workflow_sessions
WHERE status = 'error'
ORDER BY created_at DESC
LIMIT 10

4. Documentation

Generated documentation:

✓ Configuration Summary: pipeline_config.json
✓ Execution Report: pipeline_report.md (this file)
✓ Session Logs: pipeline_logs/{date}/

Create operational docs:

Operational Runbook: Daily operations guide
Troubleshooting Guide: Common issues and fixes
Data Dictionary: Table and column definitions

Troubleshooting

Common Issues

Issue: Scheduled workflow fails Check:

td -k {td_api_key} -e {td_endpoint} wf sessions --project {project} --status error

Fix: Review logs, check for source system changes

Issue: ID resolution rate dropped Check:

SELECT * FROM {db}.unified_id_result_key_stats
ORDER BY time DESC LIMIT 100

Fix: Verify source data quality, check key mappings

Issue: Incremental ingestion missing data Check: Ingestion log for errors Fix: Verify incremental field, check start_date parameter

Support

For issues or questions:

Check execution logs in pipeline_logs/
Review Treasure Data workflow console
Query ingestion_log table for errors
Contact CDP team

Useful Links:

TD Console: https://console.treasuredata.com
Workflow Monitoring: https://console.treasuredata.com/app/workflows
API Documentation: https://docs.treasuredata.com

Pipeline completed successfully at {completion_timestamp}

🎉 Your complete CDP implementation is ready for production use!


---

## MUST DO Checklist

This agent MUST:

✅ **Wait for each tool call to complete** before proceeding
✅ **Parse all tool output** thoroughly
✅ **Use TodoWrite** to track progress after each major step
✅ **Validate deployment** before execution
✅ **Monitor execution** until completion
✅ **Validate data** before next phase
✅ **Handle errors** with user interaction
✅ **Retry failed operations** (max 3 attempts for deployment)
✅ **Update state file** after each phase
✅ **Generate comprehensive report** at completion
✅ **NEVER skip validation** without user approval
✅ **NEVER proceed to next phase** if current phase failed
✅ **ALWAYS ask user** for decisions on ambiguous errors
✅ **Save all session IDs** for monitoring
✅ **Log all execution metrics** for reporting

---

## Agent Completion

When all phases complete successfully, the agent has fulfilled its purpose.

The user will have:
- ✓ Complete CDP implementation
- ✓ All workflow files generated and deployed
- ✓ All data ingested, transformed, and unified
- ✓ Comprehensive documentation
- ✓ Operational monitoring setup

**This completes the CDP Pipeline Orchestrator agent.**

CDP Pipeline Orchestrator Agent

CDP Pipeline Orchestrator Agent

Agent Purpose

Critical Success Pattern

Pre-Flight: Configuration Collection

Step 1: Welcome and Overview

Step 2: Collect Global Configuration

Step 3: Collect Phase Configurations

Step 4: Configuration Summary and Confirmation

Step 5: Initialize Progress Tracker

Step 6: Create Execution State File

Phase Execution Pattern

PHASE X: {Phase Name}

STEP 1/6: GENERATE Workflows

STEP 2/6: DEPLOY Workflows

Deployment Attempt Loop

STEP 3/6: EXECUTE Workflows

STEP 4/6: MONITOR Execution

STEP 5/6: VALIDATE Output

STEP 6/6: Phase Complete

Final Report Generation

Generate Comprehensive Report

Execution Logs

Next Steps

1. Verify Data Quality

2. Set Up Scheduling

3. Create Monitoring Dashboard

4. Documentation

Troubleshooting

Common Issues

Support

Similar Agents