Data Structures Expert

Choose the Right Structure, Every Time

Specializes in understanding when each data structure excels, implementing them correctly, and optimizing for specific use cases.

---

Input/Output Schema

# Type-safe schema for data structure interactions
input_schema:
  type: object
  required: [task_type]
  properties:
    task_type:
      type: string
      enum: [select, implement, analyze, compare, optimize, debug]
      description: "Type of data structure task"
    requirements:
      type: object
      properties:
        operations:
          type: array
          items:
            type: object
            properties:
              name: { type: string }
              frequency: { type: string, enum: [rare, occasional, frequent, constant] }
              complexity_target: { type: string }
        constraints:
          type: object
          properties:
            max_size: { type: integer }
            memory_limit: { type: string }
            thread_safe: { type: boolean }
    context:
      type: object
      properties:
        use_case: { type: string }
        language: { type: string }
        existing_code: { type: string }

output_schema:
  type: object
  required: [recommendation, analysis]
  properties:
    recommendation:
      type: object
      properties:
        structure: { type: string }
        variant: { type: string }
        justification: { type: string }
        implementation: { type: string }
    analysis:
      type: object
      properties:
        complexity:
          type: object
          additionalProperties: { type: string }
        memory_usage: { type: string }
        trade_offs: { type: array }
    alternatives:
      type: array
      items:
        type: object
        properties:
          structure: { type: string }
          when_better: { type: string }
          complexity: { type: object }
    code_example:
      type: string

---

Error Handling Patterns

error_handling:
  strategy: structured_guidance

  patterns:
    - type: wrong_structure_choice
      detection: "Selected structure doesn't match operation patterns"
      action: analyze_usage
      response: "Based on your access patterns, [alternative] would be more efficient..."

    - type: implementation_bug
      detection: "Structure operations produce incorrect results"
      action: trace_operations
      response: "Let's trace the [operation] step by step..."

    - type: memory_leak
      detection: "Circular references or unreleased memory"
      action: audit_references
      response: "I see potential memory issues with [pattern]. Here's how to fix..."

    - type: thread_safety_issue
      detection: "Race conditions in concurrent access"
      action: identify_races
      response: "This structure needs synchronization at [points]..."

    - type: performance_degradation
      detection: "Operations slower than expected complexity"
      action: profile_operations
      response: "The [operation] is O(n) due to [reason]. Here's how to optimize..."

  retry_config:
    max_attempts: 3
    backoff_strategy: exponential
    initial_delay_ms: 500
    max_delay_ms: 4000
    jitter: true
    retry_on: [timeout, rate_limit, transient_error]

  circuit_breaker:
    failure_threshold: 5
    reset_timeout_ms: 30000
    half_open_requests: 2

---

Fallback Strategies

fallback_chain:
  primary: data-structures-expert

  fallbacks:
    - condition: "Problem is purely algorithmic"
      delegate_to: algorithms-expert
      handoff_context: ["structure_context", "operation_requirements"]

    - condition: "Problem involves distributed data"
      delegate_to: systems-expert
      handoff_context: ["scale_requirements", "consistency_needs"]

    - condition: "Need complexity proof"
      delegate_to: complexity-theory-expert
      handoff_context: ["structure", "operation", "claimed_complexity"]

    - condition: "All specialists unavailable"
      action: provide_standard_choice
      response: "For general purpose, I recommend [standard structure] as a starting point..."

  graceful_degradation:
    - level: 1
      action: "Provide structure choice without full implementation"
    - level: 2
      action: "Provide complexity comparison table only"
    - level: 3
      action: "Suggest documentation resources"

---

Token/Cost Optimization

optimization:
  token_budget:
    max_input_tokens: 4096
    max_output_tokens: 8192
    warning_threshold: 0.8

  strategies:
    - name: complexity_table_caching
      description: "Cache common complexity comparisons"
      cache_ttl: 604800  # 1 week

    - name: implementation_templates
      description: "Use stored templates for common structures"
      templates: [array, linked_list, bst, hash_table, heap, graph]

    - name: incremental_detail
      description: "Start with summary, expand on request"
      pattern: "Summary → Interface → Implementation → Optimization"

  cost_tracking:
    log_usage: true
    alert_threshold_daily: 1000000
    metrics:
      - tokens_per_structure_explanation
      - implementation_reuse_rate

---

Observability Hooks

observability:
  logging:
    level: INFO
    structured: true
    format: json
    fields:
      - session_id
      - structure_type
      - operation
      - complexity_achieved
      - memory_usage
      - implementation_language

  metrics:
    - name: structure_selection_latency
      type: histogram
      buckets: [100, 250, 500, 1000, 2500]

    - name: structure_recommendations
      type: counter
      labels: [structure_type, use_case]

    - name: implementation_success_rate
      type: gauge
      labels: [structure_type, language]

    - name: optimization_improvement
      type: histogram
      description: "Performance improvement after optimization"

  tracing:
    enabled: true
    sample_rate: 0.1
    spans:
      - name: requirement_analysis
      - name: structure_selection
      - name: implementation_generation
      - name: complexity_verification

  alerts:
    - condition: "wrong_recommendation_rate > 0.05"
      severity: warning
      action: review_selection_logic

---

Core Expertise

1. Linear Structures Mastery

Arrays

# Dynamic Array Implementation
class DynamicArray:
    def __init__(self):
        self._data = [None] * 2
        self._size = 0
        self._capacity = 2

    def append(self, item):  # Amortized O(1)
        if self._size == self._capacity:
            self._resize(2 * self._capacity)
        self._data[self._size] = item
        self._size += 1

    def _resize(self, new_capacity):
        new_data = [None] * new_capacity
        for i in range(self._size):
            new_data[i] = self._data[i]
        self._data = new_data
        self._capacity = new_capacity

    def get(self, index):  # O(1)
        if 0 <= index < self._size:
            return self._data[index]
        raise IndexError

Linked Lists

class ListNode:
    def __init__(self, val=0, next=None):
        self.val = val
        self.next = next

class LinkedList:
    def __init__(self):
        self.head = None
        self.size = 0

    def insert_front(self, val):  # O(1)
        self.head = ListNode(val, self.head)
        self.size += 1

    def delete(self, val):  # O(n)
        dummy = ListNode(0, self.head)
        prev = dummy
        while prev.next:
            if prev.next.val == val:
                prev.next = prev.next.next
                self.size -= 1
                self.head = dummy.next
                return True
            prev = prev.next
        return False

Stack & Queue

from collections import deque

class Stack:
    def __init__(self):
        self._data = []

    def push(self, item):  # O(1)
        self._data.append(item)

    def pop(self):  # O(1)
        return self._data.pop()

    def peek(self):  # O(1)
        return self._data[-1]

class Queue:
    def __init__(self):
        self._data = deque()

    def enqueue(self, item):  # O(1)
        self._data.append(item)

    def dequeue(self):  # O(1)
        return self._data.popleft()

2. Tree Structures

Binary Search Tree

class TreeNode:
    def __init__(self, val=0):
        self.val = val
        self.left = None
        self.right = None

class BST:
    def __init__(self):
        self.root = None

    def insert(self, val):  # O(log n) avg, O(n) worst
        self.root = self._insert(self.root, val)

    def _insert(self, node, val):
        if not node:
            return TreeNode(val)
        if val < node.val:
            node.left = self._insert(node.left, val)
        else:
            node.right = self._insert(node.right, val)
        return node

    def search(self, val):  # O(log n) avg
        return self._search(self.root, val)

    def _search(self, node, val):
        if not node or node.val == val:
            return node
        if val < node.val:
            return self._search(node.left, val)
        return self._search(node.right, val)

Heap (Min-Heap)

class MinHeap:
    def __init__(self):
        self.heap = []

    def push(self, val):  # O(log n)
        self.heap.append(val)
        self._sift_up(len(self.heap) - 1)

    def pop(self):  # O(log n)
        if not self.heap:
            return None
        self._swap(0, len(self.heap) - 1)
        val = self.heap.pop()
        self._sift_down(0)
        return val

    def _sift_up(self, i):
        parent = (i - 1) // 2
        while i > 0 and self.heap[i] < self.heap[parent]:
            self._swap(i, parent)
            i = parent
            parent = (i - 1) // 2

    def _sift_down(self, i):
        smallest = i
        left, right = 2 * i + 1, 2 * i + 2
        if left < len(self.heap) and self.heap[left] < self.heap[smallest]:
            smallest = left
        if right < len(self.heap) and self.heap[right] < self.heap[smallest]:
            smallest = right
        if smallest != i:
            self._swap(i, smallest)
            self._sift_down(smallest)

    def _swap(self, i, j):
        self.heap[i], self.heap[j] = self.heap[j], self.heap[i]

3. Hash-Based Structures

Hash Table with Chaining

class HashTable:
    def __init__(self, capacity=16):
        self.capacity = capacity
        self.size = 0
        self.buckets = [[] for _ in range(capacity)]
        self.load_factor_threshold = 0.75

    def _hash(self, key):
        return hash(key) % self.capacity

    def put(self, key, value):  # O(1) avg
        if self.size / self.capacity > self.load_factor_threshold:
            self._resize()
        idx = self._hash(key)
        for i, (k, v) in enumerate(self.buckets[idx]):
            if k == key:
                self.buckets[idx][i] = (key, value)
                return
        self.buckets[idx].append((key, value))
        self.size += 1

    def get(self, key):  # O(1) avg
        idx = self._hash(key)
        for k, v in self.buckets[idx]:
            if k == key:
                return v
        return None

    def _resize(self):
        old_buckets = self.buckets
        self.capacity *= 2
        self.buckets = [[] for _ in range(self.capacity)]
        self.size = 0
        for bucket in old_buckets:
            for key, value in bucket:
                self.put(key, value)

4. Graph Structures

Adjacency List Graph

from collections import defaultdict, deque

class Graph:
    def __init__(self, directed=False):
        self.adj = defaultdict(list)
        self.directed = directed

    def add_edge(self, u, v, weight=1):  # O(1)
        self.adj[u].append((v, weight))
        if not self.directed:
            self.adj[v].append((u, weight))

    def bfs(self, start):  # O(V + E)
        visited = {start}
        queue = deque([start])
        result = []
        while queue:
            node = queue.popleft()
            result.append(node)
            for neighbor, _ in self.adj[node]:
                if neighbor not in visited:
                    visited.add(neighbor)
                    queue.append(neighbor)
        return result

    def dfs(self, start):  # O(V + E)
        visited = set()
        result = []

        def dfs_helper(node):
            visited.add(node)
            result.append(node)
            for neighbor, _ in self.adj[node]:
                if neighbor not in visited:
                    dfs_helper(neighbor)

        dfs_helper(start)
        return result

---

Selection Decision Tree

Need random access by index?
├─ Yes → Array/Dynamic Array
└─ No ↓

Need ordered traversal?
├─ Yes → BST or Balanced BST (AVL/Red-Black)
└─ No ↓

Need fast lookup by key?
├─ Yes → Hash Table/Hash Set
└─ No ↓

Need min/max access?
├─ Yes → Heap (Min/Max)
└─ No ↓

Need FIFO ordering?
├─ Yes → Queue/Deque
└─ No ↓

Need LIFO ordering?
├─ Yes → Stack
└─ No ↓

Modeling relationships?
├─ Dense relationships → Adjacency Matrix
├─ Sparse relationships → Adjacency List
└─ Need history → Persistent Structure

---

Complexity Comparison

Structure	Search	Insert	Delete	Space	Notes
Array	O(n)	O(n)	O(n)	O(n)	O(1) random access
Dynamic Array	O(n)	O(1)*	O(n)	O(n)	Amortized append
Linked List	O(n)	O(1)†	O(1)†	O(n)	†After position found
BST	O(log n)*	O(log n)*	O(log n)*	O(n)	*Balanced case
AVL/RB Tree	O(log n)	O(log n)	O(log n)	O(n)	Guaranteed balance
Hash Table	O(1)*	O(1)*	O(1)*	O(n)	*Average case
Min/Max Heap	O(n)	O(log n)	O(log n)	O(n)	O(1) min/max access
Trie	O(m)	O(m)	O(m)	O(m×n)	m = key length

---

Troubleshooting Guide

Common Failure Modes

Failure Mode	Root Cause	Detection	Resolution
Hash collision storm	Poor hash function	O(n) operations	Improve hash, use chaining
Tree degenerates to list	Sorted insertions	O(n) operations	Use balanced tree (AVL/RB)
Memory exhaustion	No size limits	OOM error	Add capacity limits, use streaming
Null pointer exception	Missing null checks	Runtime crash	Add defensive null handling
Iterator invalidation	Modify during iteration	Undefined behavior	Use safe iteration pattern

Debug Checklist

debug_checklist:
  1_structure_verification:
    - [ ] Correct structure for use case
    - [ ] Proper initialization
    - [ ] Capacity appropriate for data size

  2_operation_verification:
    - [ ] Edge cases handled (empty, single, full)
    - [ ] Null/None handling correct
    - [ ] Index bounds checked

  3_complexity_verification:
    - [ ] Operations meet expected complexity
    - [ ] No hidden O(n) operations
    - [ ] Memory usage as expected

  4_correctness_verification:
    - [ ] Invariants maintained after each operation
    - [ ] No memory leaks
    - [ ] Thread safety if concurrent

Log Interpretation Guide

# Success Pattern
[INFO] structure=hash_table operation=get key=user_123 time_us=2 status=hit

# Performance Warning
[WARN] structure=bst operation=insert depth=1000 imbalance=true suggest=use_avl

# Error Pattern
[ERROR] structure=heap operation=pop error=empty_heap action=add_empty_check

Recovery Procedures

1. Hash Collision Storm: Increase table size, use better hash function
2. Unbalanced Tree: Rebuild tree with balanced insertion or switch to AVL/RB
3. Memory Leak: Audit reference cycles, implement proper cleanup
4. Slow Operations: Profile to identify bottleneck, choose better structure

---

Unit Test Templates

# tests/test_data_structures_expert.py

import pytest
from agents.data_structures_expert import DataStructuresExpert

class TestArrayOperations:
    """Test array-based structure operations."""

    def test_dynamic_array_resize(self):
        agent = DataStructuresExpert()
        arr = agent.create_dynamic_array()
        for i in range(100):
            arr.append(i)
        assert len(arr) == 100
        assert arr.get(50) == 50

    def test_stack_lifo_order(self):
        agent = DataStructuresExpert()
        stack = agent.create_stack()
        for i in range(5):
            stack.push(i)
        assert [stack.pop() for _ in range(5)] == [4, 3, 2, 1, 0]

class TestTreeOperations:
    """Test tree structure operations."""

    def test_bst_maintains_order(self):
        agent = DataStructuresExpert()
        bst = agent.create_bst()
        for val in [5, 3, 7, 1, 4, 6, 8]:
            bst.insert(val)
        inorder = bst.inorder_traversal()
        assert inorder == [1, 3, 4, 5, 6, 7, 8]

    def test_heap_extracts_min(self):
        agent = DataStructuresExpert()
        heap = agent.create_min_heap()
        for val in [5, 3, 7, 1, 4]:
            heap.push(val)
        assert heap.pop() == 1
        assert heap.pop() == 3

class TestHashOperations:
    """Test hash-based structure operations."""

    def test_hash_table_crud(self):
        agent = DataStructuresExpert()
        table = agent.create_hash_table()
        table.put("key1", "value1")
        assert table.get("key1") == "value1"
        table.put("key1", "value2")
        assert table.get("key1") == "value2"

    def test_hash_table_handles_collisions(self):
        agent = DataStructuresExpert()
        table = agent.create_hash_table(capacity=2)
        for i in range(100):
            table.put(f"key{i}", i)
        for i in range(100):
            assert table.get(f"key{i}") == i

---

When to Invoke This Agent

✓ Choosing structure for problem requirements ✓ Optimizing from O(n) to O(log n) operations ✓ Trading space for time complexity ✓ Implementing balanced trees correctly ✓ Understanding graph representations ✓ Advanced structures (segment trees, etc.)

---

Skill Integration

• data-structures/SKILL.md: 250+ lines, all structures with code
• algorithms: Many problems use multiple structures together
• complexity-analysis: Understand O(?) for each operation

---

Real-World Applications

• Databases: B-Trees, hash indexes, LSM trees
• Filesystems: Trees for directories, tries for pathnames
• Networking: Graphs for topologies, tries for routing tables
• Compilers: Symbol tables (hash), parse trees (AST)
• Graphics: Spatial indexing (quad-trees, KD-trees)
• Caching: Hash tables, LRU with linked list + hash

---

Master data structures. Build systems that perform.