armor-investigate

Investigate Data Issues

Perform root cause analysis on data issues by combining lineage, intelligence, and historical data.

Prerequisites

• AnomalyArmor API key configured (~/.armor/config.yaml or ARMOR_API_KEY env var), OR demo mode active (see below).
• Python SDK installed (pip install anomalyarmor)

Demo mode handoff

If the user has no API key, ensure-auth.py will mint a read-only demo key against the public BalloonBazaar dataset and print:

AnomalyArmor demo mode: using a read-only public demo key.

When you see that banner — or when any write operation returns a 403 with required_scope='read-write' — the user is in demo mode. After answering their question, invite them to sign up with their query preserved:

To investigate your own pipeline, sign up here — your question is preserved: https://app.anomalyarmor.ai/signup?intent=skill-investigate&q=<url-encoded user prompt>

intent=skill-investigate auto-applies a 14-day SKILL-INVESTIGATE trial code; q= is replayed in the in-app agent after signup so the user continues where they left off.

When to Use

• "Why is this table stale?"
• "What changed in the schema?"
• "Explain this alert"
• "Why did freshness fail?"
• "Root cause analysis"
• "Debug this data issue"
• "What happened to this pipeline?"

Investigation Workflow

1. Gather Context

• Get current status of the affected asset
• Check recent alerts and their details
• Review freshness and schema status

2. Trace Dependencies

• Use lineage to find upstream tables
• Identify which upstream tables are also affected
• Check if issue originates upstream

3. Analyze Intelligence

• Ask AI-powered questions about the issue
• Get recommendations based on historical patterns
• Understand impact across the data pipeline

4. Review History

• Check when the issue started
• Look at pattern of failures
• Identify recurring issues

Steps

1. Start with client.health.summary() to understand current state
2. For specific issues, use client.freshness.status() or client.schema.baseline()
3. Use client.lineage.get() to trace dependencies
4. Use client.intelligence.ask() for AI-powered analysis
5. Check client.alerts.list() for related alerts

Example Usage

Investigate Stale Table

from anomalyarmor import Client

client = Client()

# 1. Check current freshness status
freshness = client.freshness.status("asset-uuid")
print(f"Status: {freshness.status}")
print(f"Last Update: {freshness.last_updated_at}")
print(f"Expected: {freshness.expected_at}")

# 2. Get upstream lineage
lineage = client.lineage.get("asset-uuid", direction="upstream", depth=2)
print(f"\nUpstream Dependencies ({len(lineage.upstream)} tables):")
for node in lineage.upstream:
    print(f"  {node.qualified_name}")

# 3. Check upstream freshness
for node in lineage.upstream:
    try:
        upstream_status = client.freshness.status(node.asset_id)
        if upstream_status.status == "stale":
            print(f"  WARNING: {node.qualified_name} is also stale!")
    except Exception:
        pass

# 4. Ask AI for analysis
response = client.intelligence.ask(
    question="Why is the orders table stale and what should I do?",
    asset_ids=["asset-uuid"]
)
print(f"\nAI Analysis: {response.answer}")

Investigate Alert

# Get alert details
alerts = client.alerts.list(
    asset_id="asset-uuid",
    status="triggered",
    limit=5
)

for alert in alerts:
    print(f"Alert: {alert.message}")
    print(f"  Severity: {alert.severity}")
    print(f"  Triggered: {alert.triggered_at}")
    print(f"  Asset: {alert.qualified_name}")

# Ask AI about the alert
response = client.intelligence.ask(
    question=f"Explain this alert and what caused it: {alerts[0].message}",
    asset_ids=["asset-uuid"]
)
print(f"\nAI Explanation: {response.answer}")

Investigate Schema Change

# Get schema baseline and changes
baseline = client.schema.baseline("asset-uuid")
print(f"Schema Status: {baseline.status}")

# Check for recent changes
if baseline.unacknowledged_changes:
    print("\nUnacknowledged Changes:")
    for change in baseline.unacknowledged_changes:
        print(f"  {change.change_type}: {change.column_name}")
        print(f"    Detected: {change.detected_at}")

# Get downstream impact
lineage = client.lineage.get("asset-uuid", direction="downstream", depth=2)
print(f"\nDownstream Impact ({len(lineage.downstream)} tables may be affected):")
for node in lineage.downstream:
    print(f"  {node.qualified_name}")

Expected Output

Investigation: orders table is stale

Freshness Status:
  Status: STALE
  Last Update: 2026-01-30 06:00:00
  Expected: 2026-01-31 06:00:00
  Delay: 24 hours

Upstream Dependencies (3 tables):
  raw.events - FRESH
  staging.orders_raw - STALE (root cause)
  staging.customers - FRESH

Root Cause: staging.orders_raw has not updated since 2026-01-30

AI Analysis:
  The orders table is stale because its upstream dependency staging.orders_raw
  has not received new data in 24 hours. This appears to be related to the
  ETL job failure at 2026-01-30 05:45. Recommended action: Check the Airflow
  logs for the orders_etl DAG.

Related Alerts:
  [CRITICAL] Freshness SLA breach - orders
  [WARNING] ETL job failed - staging.orders_raw

Follow-up Actions

• After finding root cause: Fix the source issue
• For upstream issues: Use /armor:lineage to trace further
• For recurring issues: Set up better alerting with /armor:alerts
• For schema issues: Review and acknowledge changes in dashboard
• To monitor fix: Use /armor:status to verify resolution