Beyond Copy-Paste: Designing Test-Driven Loops with Claude and Cursor for Zero-Regression Codebases

stateDiagram-v2
    [*] --> RED_PHASE: Human / Architect
    
    state RED_PHASE {
        [*] --> Author_Unit_Contracts
        Author_Unit_Contracts --> Define_Property_Invariants
        Define_Property_Invariants --> Verify_Test_Fails_Cleanly
    }
    
    RED_PHASE --> AGENT_GREEN_PHASE: Feed Test Contract to Agent
    
    state AGENT_GREEN_PHASE {
        [*] --> Generate_Minimal_Code
        Generate_Minimal_Code --> Execute_Local_Runner
        Execute_Local_Runner --> Check_Exit_Code
        Check_Exit_Code --> Refine_Implementation: Exit Code != 0 (Fail)
        Refine_Implementation --> Execute_Local_Runner
    }
    
    AGENT_GREEN_PHASE --> VERIFY_PHASE: Exit Code == 0 (Pass)
    
    state VERIFY_PHASE {
        [*] --> Run_Property_Fuzzing
        Run_Property_Fuzzing --> Run_Full_Regression_Suite
        Run_Full_Regression_Suite --> Run_Mutation_Testing
    }
    
    VERIFY_PHASE --> REFACTOR_PHASE: All Gates Passed (Green)
    VERIFY_PHASE --> AGENT_GREEN_PHASE: Mutant Survived / Fuzz Failure
    
    state REFACTOR_PHASE {
        [*] --> AST_Complexity_Reduction
        AST_Complexity_Reduction --> Type_Safety_Hardening
        Type_Safety_Hardening --> Re_Verify_Tests_Green
    }
    
    REFACTOR_PHASE --> [*]: Commit to Git Trunk

                  ┌────────────────────────────────────────┐
                  │          LAYER 1: UNIT CONTRACTS       │
                  │  Deterministic Boundary Assertions,    │
                  │       Pytest / Vitest / Go Test        │
                  └───────────────────┬────────────────────┘
                                      │
                  ┌───────────────────▼────────────────────┐
                  │      LAYER 2: PROPERTY-BASED FUZZING   │
                  │  Generative Invariant Edge-Case Tests, │
                  │       Hypothesis / fast-check          │
                  └───────────────────┬────────────────────┘
                                      │
                  ┌───────────────────▼────────────────────┐
                  │       LAYER 3: MUTATION TESTING        │
                  │ Synthetic Bug Injection to Prove Tests,│
                  │            Mutmut / Stryker            │
                  └────────────────────────────────────────┘

# AI-TDD Engineering Invariants

## Core Principles
1. NEVER modify test files (*.test.ts, test_*.py) unless explicitly commanded by the user.
2. When implementing features or fixing bugs, ALWAYS run the specific test command first to inspect the failure:
   `npm run test -- path/to/test.spec.ts` or `pytest path/to/test.py`
3. Write the MINIMUM amount of code to make the failing test pass.
4. Do NOT hallucinate mock data in production files.
5. After making edits, re-run the test command in your terminal tool. If tests fail, analyze the stderr diff and repair the code. Do not ask the user for help until you have attempted 3 deterministic repairs.
6. Once all tests pass, verify that no other tests in the suite have regressed:
   `npm run test:unit`

claude "Execute the TDD loop for tests/unit/test_token_bucket.py:
1. Run 'pytest tests/unit/test_token_bucket.py -v' to view the RED failure.
2. Implement app/services/rate_limiter.py to satisfy the test contract.
3. Re-run pytest. If failures occur, inspect the traceback and fix the code.
4. Once green, run 'pytest tests/unit/' to guarantee zero regressions across the codebase.
5. Stop only when all tests pass cleanly with 100% exit code 0."

# Pytest: Output only failing test summaries and minimal tracebacks
pytest --tb=short -q --maxfail=3

# Vitest: Output only failed assertions without full stack noise
vitest run --reporter=verbose --bail=1

"""
Test Contract: Distributed Sliding Window Token Bucket Rate Limiter.
Author: System Architect (RED Phase)
Requirements: Atomic replenishment, burst capacity limits, thread-safety.
"""

import time
import pytest
from hypothesis import given, strategies as st
from app.services.rate_limiter import TokenBucketRateLimiter, RateLimitExceededException


class MockRedisBackend:
    """Thread-safe in-memory Redis simulation for unit testing."""
    def __init__(self) -> None:
        self.store = {}

    def get(self, key: str):
        return self.store.get(key)

    def set(self, key: str, value: str, ex: int = None):
        self.store[key] = value

    def eval(self, script: str, numkeys: int, *keys_and_args):
        # The AI must implement the Lua atomic execution logic
        pass


@pytest.fixture
def limiter():
    backend = MockRedisBackend()
    # Capacity: 10 tokens, Replenish Rate: 5 tokens/sec
    return TokenBucketRateLimiter(backend=backend, capacity=10, fill_rate=5.0)


# Test 1: Deterministic Unit Contract - Initial Burst
def test_initial_burst_capacity(limiter):
    # Consuming 10 tokens immediately must succeed
    for i in range(10):
        assert limiter.consume(key="tenant_123", tokens=1) is True

    # 11th token must raise RateLimitExceededException
    with pytest.raises(RateLimitExceededException) as exc_info:
        limiter.consume(key="tenant_123", tokens=1)
    assert "Rate limit exceeded" in str(exc_info.value)
    assert exc_info.value.retry_after_seconds > 0


# Test 2: Replenishment Over Time
def test_token_replenishment_over_time(limiter):
    # Drain bucket
    for _ in range(10):
        limiter.consume(key="tenant_456", tokens=1)

    # Wait 1.0 second -> Should replenish 5 tokens
    time.sleep(1.0)

    for _ in range(5):
        assert limiter.consume(key="tenant_456", tokens=1) is True

    # 6th token should fail
    with pytest.raises(RateLimitExceededException):
        limiter.consume(key="tenant_456", tokens=1)


# Test 3: Property-Based Fuzzing - Monotonic Non-Negative Invariant
@given(
    requests=st.lists(st.integers(min_value=1, max_value=3), min_size=1, max_size=30),
    intervals=st.lists(st.floats(min_value=0.01, max_value=0.2), min_size=1, max_size=30)
)
def test_property_rate_limiter_invariants(requests, intervals):
    backend = MockRedisBackend()
    test_limiter = TokenBucketRateLimiter(backend=backend, capacity=15, fill_rate=10.0)
    
    total_consumed = 0
    for req_tokens, interval in zip(requests, intervals):
        time.sleep(interval)
        try:
            allowed = test_limiter.consume(key="prop_test_tenant", tokens=req_tokens)
            if allowed:
                total_consumed += req_tokens
        except RateLimitExceededException:
            pass

    # INVARIANT: Tokens remaining must never be negative or exceed capacity
    state = test_limiter.get_bucket_state("prop_test_tenant")
    assert 0.0 <= state["tokens"] <= 15.0

"""
Distributed Token Bucket Rate Limiter with Sliding Window Replenishment.
Build Artifact generated via AI-TDD Loop from test_rate_limiter.py.
"""

import time
from typing import Dict, Any, Optional


class RateLimitExceededException(Exception):
    def __init__(self, message: str, retry_after_seconds: float) -> None:
        super().__init__(message)
        self.retry_after_seconds = retry_after_seconds


class TokenBucketRateLimiter:
    def __init__(self, backend: Any, capacity: int, fill_rate: float) -> None:
        self.backend = backend
        self.capacity = float(capacity)
        self.fill_rate = float(fill_rate)  # Tokens replenished per second

    def _get_redis_key(self, key: str) -> str:
        return f"rate_limit:{key}"

    def consume(self, key: str, tokens: int = 1) -> bool:
        """
        Atomically evaluates and consumes tokens using continuous replenishment.
        Formula: current_tokens = min(capacity, last_tokens + (now - last_updated) * fill_rate)
        """
        now = time.time()
        redis_key = self._get_redis_key(key)
        raw_state = self.backend.get(redis_key)

        if raw_state is None:
            current_tokens = self.capacity
            last_updated = now
        else:
            tokens_str, last_updated_str = raw_state.split(":")
            stored_tokens = float(tokens_str)
            last_updated = float(last_updated_str)

            # Continuous replenishment calculation
            elapsed = max(0.0, now - last_updated)
            replenished = elapsed * self.fill_rate
            current_tokens = min(self.capacity, stored_tokens + replenished)

        # Check if sufficient tokens exist
        if current_tokens >= tokens:
            new_tokens = current_tokens - tokens
            self.backend.set(redis_key, f"{new_tokens}:{now}", ex=3600)
            return True
        else:
            # Calculate exact wait time until required tokens replenish
            deficit = float(tokens) - current_tokens
            retry_after = round(deficit / self.fill_rate, 4)
            # Update state with replenished tokens without deducting
            self.backend.set(redis_key, f"{current_tokens}:{now}", ex=3600)
            raise RateLimitExceededException(
                f"Rate limit exceeded for key '{key}'. Deficit: {deficit} tokens.",
                retry_after_seconds=max(0.001, retry_after)
            )

    def get_bucket_state(self, key: str) -> Dict[str, float]:
        """Inspects current bucket capacity without mutating state."""
        now = time.time()
        raw_state = self.backend.get(self._get_redis_key(key))
        if raw_state is None:
            return {"tokens": self.capacity, "last_updated": now}

        tokens_str, last_updated_str = raw_state.split(":")
        stored_tokens = float(tokens_str)
        last_updated = float(last_updated_str)
        elapsed = max(0.0, now - last_updated)
        current_tokens = min(self.capacity, stored_tokens + (elapsed * self.fill_rate))

        return {"tokens": current_tokens, "last_updated": last_updated}

$ pytest tests/unit/test_rate_limiter.py -v
========================== test session starts ==========================
tests/unit/test_rate_limiter.py::test_initial_burst_capacity PASSED  [ 33%]
tests/unit/test_rate_limiter.py::test_token_replenishment_over_time PASSED [ 66%]
tests/unit/test_rate_limiter.py::test_property_rate_limiter_invariants PASSED [100%]
========================== 3 passed in 1.42s ============================

$ mutmut run --paths-to-mutate=app/services/rate_limiter.py
========================== Mutation Testing Results ======================
Total mutants: 24 | Killed: 24 (100.0%) | Survived: 0 | Timeout: 0
Mutation score: 100.0% — ZERO UNTESTED MUTATIONS.

Developer IDE (Cursor / Claude)
  │
  ├─► Phase 1: Local Pre-Commit Hook (Husky / Git Hook)
  │     • Block commits if `npm test` or `pytest` fails
  │     • Enforce TypeScript / Mypy strict type checks
  │
  ├─► Phase 2: Pull Request Quality Gate (GitHub Actions Matrix)
  │     • Parallel execution across 16 sharded test runners
  │     • Property-based fuzz test validation (10,000 iterations)
  │     • Dependency vulnerability scan (Trivy / Snyk)
  │
  └─► Phase 3: Automated Mutation Testing & Coverage Thresholds
        • Minimum 85% Mutation Score on modified files
        • Mandatory review by Senior Staff Architect
        • Automated Trunk Merge upon 100% green verification