Troubleshooting Guide

name: debug-workflow
description: Workflow with detailed logging

flow:
  - member: log-input
    type: Function
    input:
      level: "debug"
      message: "Input received"
      data: ${JSON.stringify(input)}

  - member: process-data
    type: Function

  - member: log-output
    type: Function
    input:
      level: "debug"
      message: "Processing complete"
      data: ${JSON.stringify(process-data.output)}

output:
  result: ${process-data.output}

// Worker with execution tracing
import { Conductor } from '@ensemble-edge/conductor';

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const traceId = crypto.randomUUID();
    console.log(`[${traceId}] Starting execution`);

    try {
      const conductor = new Conductor({
        env,
        onMemberStart: (member) => {
          console.log(`[${traceId}] Starting: ${member.name}`);
        },
        onMemberComplete: (member, result) => {
          console.log(`[${traceId}] Completed: ${member.name} (${result.duration}ms)`);
        }
      });

      const result = await conductor.executeEnsemble(
        'my-workflow',
        await request.json()
      );

      console.log(`[${traceId}] Execution complete`);
      return Response.json(result);
    } catch (error) {
      console.error(`[${traceId}] Error:`, error);
      return Response.json({ error: error.message }, { status: 500 });
    }
  }
};

flow:
  - member: log-initial-state
    type: Function
    state:
      use: all
    input:
      state: ${JSON.stringify(state)}

  - member: update-state
    type: Function
    state:
      set: [userData, processedAt]

  - member: log-updated-state
    type: Function
    state:
      use: all
    input:
      state: ${JSON.stringify(state)}

# Check ensemble exists
ls ensembles/my-workflow.yaml

# Verify wrangler.toml configuration
cat wrangler.toml | grep ensembles

# Deploy with ensembles
npx wrangler deploy

project/
├── ensembles/
│   └── my-workflow.yaml
├── src/
│   └── index.ts
└── wrangler.toml

# Add timeout and retry
- member: fetch-data
  type: Fetch
  config:
    url: "${env.API_URL}"
    timeout: 30000  # Increase timeout
    retries: 3      # Add retries
  continue_on_error: true  # Continue on failure

# Handle error
- member: handle-error
  condition: ${!fetch-data.success}
  type: Function
  input:
    error: ${fetch-data.error}

# ❌ Invalid - missing quotes
expression: $filter(items, status = active)

# ✅ Valid - quoted string
expression: $filter(items, status = "active")

# Test expressions separately
- member: test-transform
  type: Transform
  input:
    data: { "test": true }
    expression: "test"  # Simple test first

flow:
  # Set state first
  - member: load-user
    type: Data
    state:
      set: [userData]  # Set state

  # Then use it
  - member: process-user
    condition: ${state.userData != null}  # Check exists
    state:
      use: [userData]  # Use state

# Add exponential backoff
- member: ai-task
  type: Think
  retry:
    maxAttempts: 5
    backoff: exponential
    initialDelay: 1000

# Use AI Gateway caching
config:
  provider: openai
  routing: cloudflare-gateway  # Automatic caching

# Batch requests
- member: batch-classify
  type: Think
  input:
    prompt: |
      Classify each item:
      ${items.map((item, i) => `${i+1}. ${item}`).join('\n')}

-- Add index for common queries
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_orders_user_id ON orders(user_id);

-- Optimize query
-- ❌ Slow - no index
SELECT * FROM orders WHERE created_at > '2024-01-01';

-- ✅ Fast - indexed
SELECT * FROM orders WHERE user_id = ? ORDER BY created_at DESC LIMIT 10;

# Add caching
- member: fetch-user
  type: Data
  cache:
    ttl: 300  # Cache 5 minutes
  config:
    storage: d1
    operation: query
    query: "SELECT * FROM users WHERE id = ?"

# Split large data
- member: store-metadata
  type: Data
  config:
    storage: kv
    operation: put
  input:
    key: "data:${input.id}:metadata"
    value: ${metadata}  # Small metadata in KV

- member: store-content
  type: Data
  config:
    storage: r2  # Use R2 for large files
    operation: put
  input:
    key: "data:${input.id}:content"
    value: ${largeContent}

- member: request-review
  type: HITL
  config:
    timeout: 86400000  # Increase to 24 hours
    onTimeout: continue  # Or continue on timeout

# Handle timeout
- member: auto-approve-on-timeout
  condition: ${request-review.timedOut}
  type: Function
  input:
    reason: "Auto-approved due to timeout"

# Add timing
output:
  metrics:
    member1Duration: ${member1.duration}
    member2Duration: ${member2.duration}
    totalDuration: ${execution.duration}

# ❌ Sequential - 300ms
flow:
  - member: task1  # 100ms
  - member: task2  # 100ms
  - member: task3  # 100ms

# ✅ Parallel - 100ms
flow:
  parallel:
    - member: task1
    - member: task2
    - member: task3

- member: expensive-operation
  cache:
    ttl: 3600

# ❌ Slow - gpt-4o (~2s)
config:
  model: gpt-4o

# ✅ Fast - gpt-4o-mini (~200ms)
config:
  model: gpt-4o-mini

# Track token usage
- member: log-usage
  type: Data
  config:
    storage: d1
    operation: query
    query: |
      INSERT INTO usage_log (member, tokens, cost)
      VALUES (?, ?, ?)
  input:
    params:
      - "expensive-ai-call"
      - ${expensive-ai-call.output.usage.total_tokens}
      - ${expensive-ai-call.output.usage.total_tokens * 0.00001}

config:
  provider: openai
  routing: cloudflare-gateway  # Persistent cache

# ❌ Verbose prompt (100 tokens)
prompt: |
  I would like you to carefully analyze the following text
  and determine whether the sentiment is positive or negative.
  Please be thorough in your analysis.

  Text: ${text}

# ✅ Concise prompt (10 tokens)
prompt: "Classify sentiment (positive/negative): ${text}"

# Simple tasks
config:
  model: gpt-4o-mini  # $0.15/$0.60 per 1M tokens

# Complex tasks only
config:
  model: gpt-4o  # $5.00/$15.00 per 1M tokens

# Set secret
echo "sk-..." | npx wrangler secret put OPENAI_API_KEY

# Verify in wrangler.toml
cat wrangler.toml

# Use in workflow
config:
  provider: openai
  apiKey: ${env.OPENAI_API_KEY}

// Verify signature
import crypto from 'crypto';

async function verifySignature(
  request: Request,
  secret: string
): Promise<boolean> {
  const signature = request.headers.get('X-Signature');
  const body = await request.clone().text();

  const hmac = crypto.createHmac('sha256', secret);
  hmac.update(body);
  const expected = hmac.digest('hex');

  return signature === expected;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    // Handle preflight
    if (request.method === 'OPTIONS') {
      return new Response(null, {
        headers: {
          'Access-Control-Allow-Origin': '*',
          'Access-Control-Allow-Methods': 'GET, POST, OPTIONS',
          'Access-Control-Allow-Headers': 'Content-Type',
          'Access-Control-Max-Age': '86400'
        }
      });
    }

    // Handle request
    const response = await handleRequest(request, env);

    // Add CORS headers
    response.headers.set('Access-Control-Allow-Origin', '*');
    return response;
  }
};

// Ensure mocks are properly configured
const conductor = await TestConductor.create({
  mocks: {
    ai: {
      responses: {
        'member-name': { result: 'mocked response' }
      }
    },
    db: {
      users: [
        { id: 1, name: 'Test User' }
      ]
    }
  }
});

// ❌ Not waiting for completion
const result = conductor.executeEnsemble('workflow', input);
expect(result.output).toBeDefined();

// ✅ Await completion
const result = await conductor.executeEnsemble('workflow', input);
expect(result.output).toBeDefined();

// ✅ Wait for HITL
const execution = await conductor.executeEnsemble('workflow', input);
expect(execution.status).toBe('waiting_for_input');

await conductor.respondToHITL(execution.id, { approved: true });
const completed = await conductor.waitForCompletion(execution.id);
expect(completed.output.approved).toBe(true);

# 1. Verify authentication
npx wrangler whoami

# 2. Check wrangler.toml syntax
npx wrangler deploy --dry-run

# 3. Verify bindings
cat wrangler.toml | grep -A 5 "binding"

# 4. Check compatibility date
# wrangler.toml
compatibility_date = "2024-01-01"  # Not too old

# 5. Deploy with verbose logging
npx wrangler deploy --verbose

# Local development - .dev.vars
OPENAI_API_KEY=sk-...

# Production - secrets
echo "sk-..." | npx wrangler secret put OPENAI_API_KEY --env production

# Verify secrets exist
npx wrangler secret list --env production

const conductor = new Conductor({
  env,
  debug: true  // Verbose logging
});

console.log(JSON.stringify({
  version: '1.0.0',
  ensemble: 'my-workflow',
  input: input,
  error: error.message,
  stack: error.stack,
  timestamp: Date.now()
}));