resilient-data-gathering

Resilient Data Gathering Workflow

Purpose

When search_web, shell_agent, and execute_code_sandbox all fail repeatedly with 'unknown error' or similar tool execution failures, fall back to embedding known domain data directly into run_shell Python scripts and persisting intermediate data to JSON files for auditability and recovery.

When to Use This Pattern

Apply this pattern when:

1. Trigger condition: 2-3 consecutive failures across multiple tools (search_web, shell_agent, execute_code_sandbox) with 'unknown error' messages
2. You have domain knowledge: The required data is known or can be reasonably estimated from context
3. Auditability needed: Intermediate results must be preserved for verification or rollback

Step-by-Step Instructions

Step 1: Recognize Tool Failure Pattern

Monitor for repeated failures across multiple execution tools:

• search_web returns errors or empty results
• shell_agent fails to complete autonomous tasks
• execute_code_sandbox throws 'unknown error' repeatedly

Decision point: After 2-3 failures, switch to the fallback pattern rather than continuing to retry failing tools.

Step 2: Gather Known Domain Data

Collect all data you already know or can reasonably infer:

• Product specifications, prices, SKUs
• Competitor information from context
• Business rules and constraints
• Historical data from previous task phases

Document this data in a structured format before embedding.

Step 3: Embed Data in run_shell Python Script

Create a Python script that embeds the known data directly as literals or constants:

import json
import os
from datetime import datetime

# ============================================
# EMBEDDED DOMAIN DATA (known from context)
# ============================================
PRODUCT_DATA = {
    "sku_001": {
        "name": "Product A",
        "competitor_price": 29.99,
        "weight_oz": 12,
        "category": "beverage"
    },
    "sku_002": {
        "name": "Product B", 
        "competitor_price": 34.99,
        "weight_oz": 16,
        "category": "snack"
    }
}

BUSINESS_RULES = {
    "margin_target": 0.25,
    "price_floor": 19.99,
    "price_ceiling": 99.99
}

# ============================================
# ANALYSIS LOGIC
# ============================================
def analyze_products(products, rules):
    results = {}
    for sku, data in products.items():
        price_per_oz = data["competitor_price"] / data["weight_oz"]
        recommended_price = data["competitor_price"] * (1 + rules["margin_target"])
        recommended_price = max(rules["price_floor"], min(rules["price_ceiling"], recommended_price))
        
        results[sku] = {
            **data,
            "price_per_oz": round(price_per_oz, 2),
            "recommended_price": round(recommended_price, 2),
            "analysis_timestamp": datetime.now().isoformat()
        }
    return results

# Execute analysis
analysis_results = analyze_products(PRODUCT_DATA, BUSINESS_RULES)

# ============================================
# PERSIST INTERMEDIATE DATA (audit trail)
# ============================================
output_file = "intermediate_analysis.json"
with open(output_file, "w") as f:
    json.dump({
        "metadata": {
            "generated_at": datetime.now().isoformat(),
            "source": "embedded_domain_data",
            "fallback_reason": "tool_execution_failures"
        },
        "results": analysis_results
    }, f, indent=2)

# Signal artifact location for downstream tools
print(f"ARTIFACT_PATH:{os.path.abspath(output_file)}")

# Output results for immediate consumption
print("\n=== ANALYSIS RESULTS ===")
print(json.dumps(analysis_results, indent=2))