Where AI Excels vs. Where It Fails

// Prompt: "Create a Prisma-based CRUD service for a User model
// with email, name, role, and createdAt fields"

// AI generates this confidently and accurately:
import { PrismaClient, User, Prisma } from '@prisma/client';

export class UserService {
  constructor(private prisma: PrismaClient) {}

  async create(data: Prisma.UserCreateInput): Promise<User> {
    return this.prisma.user.create({ data });
  }

  async findById(id: string): Promise<User | null> {
    return this.prisma.user.findUnique({ where: { id } });
  }

  async findByEmail(email: string): Promise<User | null> {
    return this.prisma.user.findUnique({ where: { email } });
  }

  async findMany(params: {
    skip?: number;
    take?: number;
    where?: Prisma.UserWhereInput;
    orderBy?: Prisma.UserOrderByWithRelationInput;
  }): Promise<User[]> {
    return this.prisma.user.findMany(params);
  }

  async update(id: string, data: Prisma.UserUpdateInput): Promise<User> {
    return this.prisma.user.update({ where: { id }, data });
  }

  async delete(id: string): Promise<User> {
    return this.prisma.user.delete({ where: { id } });
  }

  async count(where?: Prisma.UserWhereInput): Promise<number> {
    return this.prisma.user.count({ where });
  }
}

# Prompt: "Implement a trie data structure with insert, search,
# and startsWith methods, plus a method to find all words with a given prefix"

class TrieNode:
    def __init__(self):
        self.children: dict[str, 'TrieNode'] = {}
        self.is_end_of_word: bool = False
        self.word: str | None = None

class Trie:
    def __init__(self):
        self.root = TrieNode()

    def insert(self, word: str) -> None:
        node = self.root
        for char in word:
            if char not in node.children:
                node.children[char] = TrieNode()
            node = node.children[char]
        node.is_end_of_word = True
        node.word = word

    def search(self, word: str) -> bool:
        node = self._find_node(word)
        return node is not None and node.is_end_of_word

    def starts_with(self, prefix: str) -> bool:
        return self._find_node(prefix) is not None

    def find_all_with_prefix(self, prefix: str) -> list[str]:
        node = self._find_node(prefix)
        if node is None:
            return []
        results: list[str] = []
        self._collect_words(node, results)
        return results

    def _find_node(self, prefix: str) -> TrieNode | None:
        node = self.root
        for char in prefix:
            if char not in node.children:
                return None
            node = node.children[char]
        return node

    def _collect_words(self, node: TrieNode, results: list[str]) -> None:
        if node.is_end_of_word and node.word:
            results.append(node.word)
        for child in node.children.values():
            self._collect_words(child, results)

// Given this function:
function parseConnectionString(connStr: string): {
  host: string;
  port: number;
  database: string;
  user?: string;
  password?: string;
  ssl?: boolean;
} {
  const url = new URL(connStr);
  return {
    host: url.hostname,
    port: parseInt(url.port) || 5432,
    database: url.pathname.slice(1),
    user: url.username || undefined,
    password: url.password || undefined,
    ssl: url.searchParams.get('ssl') === 'true',
  };
}

// AI generates excellent tests:
describe('parseConnectionString', () => {
  it('parses a full connection string', () => {
    const result = parseConnectionString(
      'postgresql://admin:secret@db.example.com:5433/myapp?ssl=true'
    );
    expect(result).toEqual({
      host: 'db.example.com',
      port: 5433,
      database: 'myapp',
      user: 'admin',
      password: 'secret',
      ssl: true,
    });
  });

  it('uses default port when not specified', () => {
    const result = parseConnectionString(
      'postgresql://db.example.com/myapp'
    );
    expect(result.port).toBe(5432);
  });

  it('handles missing credentials', () => {
    const result = parseConnectionString(
      'postgresql://db.example.com:5432/myapp'
    );
    expect(result.user).toBeUndefined();
    expect(result.password).toBeUndefined();
  });

  it('defaults ssl to false', () => {
    const result = parseConnectionString(
      'postgresql://db.example.com/myapp'
    );
    expect(result.ssl).toBe(false);
  });

  it('handles special characters in password', () => {
    const result = parseConnectionString(
      'postgresql://admin:p%40ss%23word@db.example.com/myapp'
    );
    expect(result.password).toBe('p@ss#word');
  });
});

// Original Go code
func retryWithBackoff(ctx context.Context, maxRetries int, fn func() error) error {
    var lastErr error
    for attempt := 0; attempt < maxRetries; attempt++ {
        if err := fn(); err == nil {
            return nil
        } else {
            lastErr = err
        }
        backoff := time.Duration(math.Pow(2, float64(attempt))) * time.Second
        select {
        case <-time.After(backoff):
            continue
        case <-ctx.Done():
            return ctx.Err()
        }
    }
    return fmt.Errorf("max retries (%d) exceeded: %w", maxRetries, lastErr)
}

// AI accurately translates to TypeScript:
async function retryWithBackoff(
  maxRetries: number,
  fn: () => Promise<void>,
  signal?: AbortSignal
): Promise<void> {
  let lastError: Error | undefined;
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      await fn();
      return;
    } catch (err) {
      lastError = err instanceof Error ? err : new Error(String(err));
    }
    const backoff = Math.pow(2, attempt) * 1000;
    await new Promise<void>((resolve, reject) => {
      const timer = setTimeout(resolve, backoff);
      signal?.addEventListener('abort', () => {
        clearTimeout(timer);
        reject(signal.reason ?? new Error('Aborted'));
      }, { once: true });
    });
    if (signal?.aborted) {
      throw signal.reason ?? new Error('Aborted');
    }
  }
  throw new Error(
    `Max retries (${maxRetries}) exceeded: ${lastError?.message}`
  );
}

// Task: Calculate insurance premium based on 15 risk factors
// with complex interaction effects

// AI-generated code looks reasonable but often has subtle bugs:
function calculatePremium(policy: InsurancePolicy): number {
  let basePremium = BASE_RATES[policy.coverageType];

  // Age factor -- AI gets the basic idea right
  if (policy.applicantAge < 25) basePremium *= 1.4;
  else if (policy.applicantAge > 65) basePremium *= 1.6;

  // But misses the interaction between age and driving record:
  // In reality, young drivers with clean records get a BIGGER discount
  // than older drivers with clean records. AI typically applies
  // a flat discount regardless of age.
  if (policy.cleanRecord) basePremium *= 0.85; // ← WRONG: should vary by age

  // And completely invents logic for edge cases:
  // Multi-policy discounts have specific rules about which policies
  // can be bundled that aren't intuitive from the function name
  if (policy.multiPolicy) basePremium *= 0.9; // ← OVERSIMPLIFIED

  return basePremium;
}

# AI generates a clean-looking state machine:
class OrderStateMachine:
    TRANSITIONS = {
        "pending": ["confirmed", "cancelled"],
        "confirmed": ["processing", "cancelled"],
        "processing": ["shipped", "failed"],
        "shipped": ["delivered", "returned"],
        "delivered": ["returned"],
        "cancelled": [],
        "returned": ["refunded"],
        "refunded": [],
        "failed": ["processing", "cancelled"],  # ← retry or cancel
    }

    def transition(self, order: Order, new_state: str) -> None:
        if new_state not in self.TRANSITIONS[order.state]:
            raise InvalidTransitionError(order.state, new_state)

        # AI misses: some transitions have preconditions
        # - "shipped" requires a tracking number
        # - "returned" has a 30-day window from delivery
        # - "refunded" requires manager approval over $500
        # - "processing" from "failed" requires incrementing retry count
        #   and checking max retries

        order.state = new_state
        order.updated_at = datetime.now()

// AI-generated cache with a race condition
type Cache struct {
    mu    sync.RWMutex
    items map[string]*CacheItem
}

func (c *Cache) GetOrSet(key string, fetch func() (interface{}, error)) (interface{}, error) {
    // Check cache first
    c.mu.RLock()
    item, ok := c.items[key]
    c.mu.RUnlock()

    if ok && !item.IsExpired() {
        return item.Value, nil
    }

    // Cache miss -- fetch and store
    // BUG: Between the RUnlock above and the Lock below,
    // another goroutine could also see a cache miss and
    // start fetching. This causes a "thundering herd" problem.
    c.mu.Lock()
    defer c.mu.Unlock()

    // AI sometimes remembers to double-check, but often doesn't:
    // if item, ok := c.items[key]; ok && !item.IsExpired() {
    //     return item.Value, nil  // Another goroutine already fetched
    // }

    value, err := fetch()
    if err != nil {
        return nil, err
    }

    c.items[key] = &CacheItem{Value: value, ExpiresAt: time.Now().Add(c.ttl)}
    return value, nil
}

// Task: "Optimize this hot-path JSON parser for our streaming pipeline"

// AI generates something that works but misses performance-critical details:
fn parse_event(raw: &[u8]) -> Result<Event, ParseError> {
    // AI uses serde_json::from_slice -- correct, but allocates
    let value: serde_json::Value = serde_json::from_slice(raw)?;

    // Then does dynamic lookups with .get() -- more allocations
    let event_type = value.get("type")
        .and_then(|v| v.as_str())
        .ok_or(ParseError::MissingField("type"))?;

    // A hand-optimized version would:
    // 1. Use serde with #[derive(Deserialize)] to avoid Value allocation
    // 2. Use zero-copy deserialization with &'a str borrowing
    // 3. Use simd-json for SIMD-accelerated parsing
    // 4. Pre-allocate buffers and reuse them across calls

    Ok(Event {
        event_type: event_type.to_string(), // ← unnecessary allocation
        // ...
    })
}

# Task: "Implement a conflict-free replicated data type (CRDT) for
# a collaborative text editor using the RGA algorithm"

# AI will generate SOMETHING -- but it's almost certainly wrong.
# CRDTs require precise mathematical properties (commutativity,
# associativity, idempotency of merge operations) that the model
# can't verify.

class RGANode:
    # AI might generate a plausible-looking structure
    # but the merge logic will have subtle bugs that only
    # manifest when operations arrive out of order across
    # multiple replicas.
    pass

// AI-generated auth -- looks right, has critical vulnerability:
async function verifyToken(token: string): Promise<User> {
  try {
    const payload = jwt.verify(token, process.env.JWT_SECRET!);

    // BUG 1: No check for token type -- an access token could be
    // used as a refresh token or vice versa

    // BUG 2: No check for token revocation -- if a user logs out
    // or is banned, their tokens are still valid until expiry

    // BUG 3: jwt.verify with just a secret string defaults to
    // HS256. If the token header says "alg": "none", some
    // libraries (older jsonwebtoken versions) will accept it.

    const user = await db.user.findUnique({
      where: { id: (payload as JwtPayload).sub }
    });

    // BUG 4: No check if user is active/not-deleted

    return user!; // BUG 5: Non-null assertion -- if user was deleted
                  // between token issuance and this check, this crashes
  } catch {
    throw new UnauthorizedError();
    // BUG 6: Swallows the actual error -- makes debugging impossible
    // and treats all errors (expired, malformed, invalid signature)
    // the same way
  }
}

-- AI-generated migration -- looks clean, causes production outage:
ALTER TABLE users ADD COLUMN department_id INTEGER REFERENCES departments(id);

-- BUG 1: On a table with 50M rows, this ALTER TABLE acquires an
-- ACCESS EXCLUSIVE lock on the users table for the entire duration.
-- In PostgreSQL, this blocks ALL reads and writes.
-- Estimated time: 15-30 minutes = 30 minutes of downtime.

-- What you actually need:
-- Step 1: Add column as nullable without constraint
ALTER TABLE users ADD COLUMN department_id INTEGER;
-- Step 2: Backfill in batches
-- Step 3: Add constraint with NOT VALID
ALTER TABLE users ADD CONSTRAINT fk_department
  FOREIGN KEY (department_id) REFERENCES departments(id) NOT VALID;
-- Step 4: Validate constraint (concurrent-safe)
ALTER TABLE users VALIDATE CONSTRAINT fk_department;

# Asking AI about a library released 2 months ago or with few users:

# AI confidently generates:
from newhotlib import StreamProcessor, BatchConfig

processor = StreamProcessor(
    config=BatchConfig(
        window_size=1000,
        flush_interval=5.0,
        error_strategy="retry_exponential"  # ← This API doesn't exist
    )
)

# The actual API (which the model has never seen) is:
from newhotlib import Processor, ProcessorConfig

processor = Processor(
    conf=ProcessorConfig(
        batch_size=1000,
        flush_seconds=5.0,
        on_error="backoff"  # Different parameter name and value
    )
)

def should_use_ai(task: dict) -> str:
    """Returns: 'green' (trust), 'yellow' (verify carefully), 'red' (don't rely on)"""

    # Factor 1: How common is this pattern?
    if task["pattern_frequency"] == "extremely_common":    # CRUD, REST, React
        pattern_score = 3
    elif task["pattern_frequency"] == "common":             # Auth, caching
        pattern_score = 2
    elif task["pattern_frequency"] == "uncommon":           # Custom DSL, novel algo
        pattern_score = 1
    else:                                                    # Unique to your domain
        pattern_score = 0

    # Factor 2: How clear is the specification?
    if task["spec_clarity"] == "unambiguous":               # "Sort this list"
        spec_score = 3
    elif task["spec_clarity"] == "mostly_clear":            # "Add pagination"
        spec_score = 2
    elif task["spec_clarity"] == "vague":                   # "Make it faster"
        spec_score = 1
    else:                                                    # "Fix the auth"
        spec_score = 0

    # Factor 3: How easy is it to verify?
    if task["verifiability"] == "unit_testable":            # Pure function
        verify_score = 3
    elif task["verifiability"] == "integration_testable":   # API endpoint
        verify_score = 2
    elif task["verifiability"] == "manual_review":          # UI behavior
        verify_score = 1
    else:                                                    # Race condition, security
        verify_score = 0

    total = pattern_score + spec_score + verify_score

    if total >= 7:
        return "green"   # Trust AI output with light review
    elif total >= 4:
        return "yellow"  # Use AI output as starting point, verify carefully
    else:
        return "red"     # Write it yourself, maybe use AI for brainstorming

Red Zone Task: "Build a distributed lock service"
                ↓ decompose ↓
Green Zone: "Write the Lock interface with acquire/release/extend methods"
Green Zone: "Write Redis SET NX EX wrapper with error handling"
Yellow Zone: "Implement lock extension with Lua script for atomicity"
Red Zone:   "Handle clock skew and split-brain scenarios"
            ↑ Write this yourself, use AI for brainstorming ↑

Prompt 1: "Generate a JWT refresh token rotation implementation"
Prompt 2: "Review the code you just generated. What security
          vulnerabilities exist? What race conditions are possible?
          What happens if the database write fails after the old
          token is invalidated?"

Instead of: "Add caching to getUserProfile"

Write: "Add caching to getUserProfile with these exact specifications:
- Cache key format: user:profile:{userId}
- TTL: 300 seconds
- Cache invalidation: on user.update and user.delete events
- Cache stampede protection: use probabilistic early expiration
- Serialization: JSON with Date revival
- Error handling: on Redis failure, fall through to database
- Metrics: cache_hit_total, cache_miss_total counters
- The function signature must remain unchanged (backward compatible)"