Contrarian View: Agile Is Dead – Waterfall Is Better for Large Teams – 2026 Stripe Data

import datetime
import json
from typing import List, Dict, Optional, Literal
from dataclasses import dataclass, field

PhaseStatus = Literal['pending', 'in_progress', 'blocked', 'completed', 'failed']

@dataclass
class WaterfallPhase:
    '''Represents a single phase in a Waterfall delivery pipeline'''
    id: str
    name: str
    status: PhaseStatus
    start_date: datetime.date
    end_date: datetime.date
    dependencies: List[str] = field(default_factory=list)
    deliverables: List[str] = field(default_factory=list)
    owner: str = ''
    notes: str = ''

    def validate(self) -> List[str]:
        '''Validate phase constraints, return list of error messages'''
        errors = []
        # Check date sanity
        if self.end_date < self.start_date:
            errors.append(f'Phase {self.id}: End date {self.end_date} is before start date {self.start_date}')
        # Check deliverables exist
        if self.status == 'completed' and not self.deliverables:
            errors.append(f'Phase {self.id}: Completed phase has no deliverables')
        # Check dependencies are valid (non-empty IDs)
        for dep in self.dependencies:
            if not dep.strip():
                errors.append(f'Phase {self.id}: Empty dependency ID found')
        return errors

class WaterfallProject:
    '''Manages a full Waterfall project with phased delivery'''
    def __init__(self, project_id: str, name: str, team_size: int):
        self.project_id = project_id
        self.name = name
        self.team_size = team_size
        self.phases: List[WaterfallPhase] = []
        self.created_at = datetime.date.today()

    def add_phase(self, phase: WaterfallPhase) -> None:
        '''Add a phase to the project, validate no duplicate IDs'''
        existing_ids = [p.id for p in self.phases]
        if phase.id in existing_ids:
            raise ValueError(f'Phase ID {phase.id} already exists in project {self.project_id}')
        self.phases.append(phase)

    def validate_project(self) -> Dict[str, List[str]]:
        '''Validate all phases and inter-phase dependencies'''
        all_errors = {'phase_errors': [], 'dependency_errors': []}
        # Validate individual phases
        for phase in self.phases:
            phase_errors = phase.validate()
            all_errors['phase_errors'].extend([f'Phase {phase.id}: {err}' for err in phase_errors])
        # Validate dependencies exist
        phase_ids = [p.id for p in self.phases]
        for phase in self.phases:
            for dep in phase.dependencies:
                if dep not in phase_ids:
                    all_errors['dependency_errors'].append(
                        f'Phase {phase.id} depends on non-existent phase {dep}'
                    )
        # Validate phase ordering (no phase starts before dependency ends)
        phase_map = {p.id: p for p in self.phases}
        for phase in self.phases:
            for dep_id in phase.dependencies:
                dep_phase = phase_map.get(dep_id)
                if dep_phase and phase.start_date < dep_phase.end_date:
                    all_errors['dependency_errors'].append(
                        f'Phase {phase.id} starts on {phase.start_date} before dependency {dep_id} ends on {dep_phase.end_date}'
                    )
        return all_errors

    def get_completion_rate(self) -> float:
        '''Calculate percentage of phases completed'''
        if not self.phases:
            return 0.0
        completed = sum(1 for p in self.phases if p.status == 'completed')
        return round((completed / len(self.phases)) * 100, 2)

# Example usage with error handling
if __name__ == '__main__':
    try:
        # Initialize a 50-person Stripe payments project
        project = WaterfallProject(
            project_id='stripe-pay-v2',
            name='Stripe Payments V2 Core Overhaul',
            team_size=52
        )
        # Add requirements phase
        req_phase = WaterfallPhase(
            id='req-001',
            name='Requirements Gathering',
            status='completed',
            start_date=datetime.date(2025, 1, 1),
            end_date=datetime.date(2025, 2, 28),
            deliverables=['PRD v1.0', 'Compliance Checklist', 'API Spec Draft'],
            owner='product-eng-lead'
        )
        project.add_phase(req_phase)
        # Add design phase (depends on req-001)
        design_phase = WaterfallPhase(
            id='design-001',
            name='System Design & Architecture',
            status='completed',
            start_date=datetime.date(2025, 3, 1),
            end_date=datetime.date(2025, 4, 30),
            dependencies=['req-001'],
            deliverables=['Architecture Diagram', 'DB Schema', 'Service Contracts'],
            owner='staff-engineer'
        )
        project.add_phase(design_phase)
        # Validate project
        errors = project.validate_project()
        if errors['phase_errors'] or errors['dependency_errors']:
            print(f'Validation errors: {json.dumps(errors, indent=2)}')
        else:
            print(f'Project {project.name} is valid. Completion rate: {project.get_completion_rate()}%')
    except ValueError as e:
        print(f'Failed to add phase: {e}')
    except Exception as e:
        print(f'Unexpected error: {e}')

import datetime
import json
from typing import List, Dict, Optional, Literal
from dataclasses import dataclass, field

SprintStatus = Literal['planning', 'active', 'closed', 'failed']

@dataclass
class UserStory:
    '''Represents a single user story in Agile sprint'''
    id: str
    name: str
    story_points: int
    status: Literal['todo', 'in_progress', 'done', 'blocked'] = 'todo'
    assignee: str = ''
    tags: List[str] = field(default_factory=list)

    def validate(self) -> List[str]:
        errors = []
        if self.story_points <= 0:
            errors.append(f'Story {self.id}: Story points must be positive, got {self.story_points}')
        if not self.name.strip():
            errors.append(f'Story {self.id}: Name cannot be empty')
        return errors

@dataclass
class Sprint:
    '''Represents an Agile sprint'''
    id: str
    name: str
    status: SprintStatus
    start_date: datetime.date
    end_date: datetime.date
    team_size: int
    stories: List[UserStory] = field(default_factory=list)
    velocity: Optional[int] = None  # Historical velocity if known

    def validate(self) -> List[str]:
        errors = []
        if self.end_date < self.start_date:
            errors.append(f'Sprint {self.id}: End date before start date')
        if self.team_size <= 0:
            errors.append(f'Sprint {self.id}: Team size must be positive')
        if self.status == 'active' and not self.stories:
            errors.append(f'Sprint {self.id}: Active sprint has no stories')
        # Validate all stories
        for story in self.stories:
            story_errors = story.validate()
            errors.extend([f'Story {story.id}: {err}' for err in story_errors])
        return errors

    def calculate_burndown(self) -> Dict[str, float]:
        '''Calculate daily burndown data: remaining points per day'''
        if self.status not in ['active', 'closed']:
            return {}
        total_points = sum(story.story_points for story in self.stories)
        if total_points == 0:
            return {}
        # Generate list of days in sprint
        days = []
        current_day = self.start_date
        while current_day <= self.end_date:
            days.append(current_day)
            current_day += datetime.timedelta(days=1)
        # For simplicity, assume linear burndown (real world would use actual done points)
        burndown = {}
        num_days = len(days)
        if num_days == 0:
            return burndown
        points_per_day = total_points / num_days
        remaining = total_points
        for day in days:
            burndown[str(day)] = round(remaining, 2)
            remaining -= points_per_day
            if remaining < 0:
                remaining = 0
        return burndown

    def get_sprint_velocity(self) -> int:
        '''Calculate velocity as completed story points / sprint'''
        if self.status != 'closed':
            return 0
        completed_points = sum(story.story_points for story in self.stories if story.status == 'done')
        return completed_points

class AgileProject:
    '''Manages Agile project with sprints'''
    def __init__(self, project_id: str, name: str):
        self.project_id = project_id
        self.name = name
        self.sprints: List[Sprint] = []

    def add_sprint(self, sprint: Sprint) -> None:
        existing_ids = [s.id for s in self.sprints]
        if sprint.id in existing_ids:
            raise ValueError(f'Sprint ID {sprint.id} already exists')
        self.sprints.append(sprint)

    def calculate_average_velocity(self) -> float:
        closed_sprints = [s for s in self.sprints if s.status == 'closed']
        if not closed_sprints:
            return 0.0
        total_velocity = sum(s.get_sprint_velocity() for s in closed_sprints)
        return round(total_velocity / len(closed_sprints), 2)

# Example usage with Stripe Agile team data
if __name__ == '__main__':
    try:
        # Stripe 2026 Agile team: 52-person payments team (same size as Waterfall example)
        project = AgileProject(
            project_id='stripe-pay-agile-v2',
            name='Stripe Payments V2 Agile Iteration'
        )
        # Add sprint 1
        sprint1 = Sprint(
            id='sprint-001',
            name='Payments V2 Sprint 1: Core API',
            status='closed',
            start_date=datetime.date(2025, 1, 1),
            end_date=datetime.date(2025, 1, 14),
            team_size=52,
            velocity=42  # Historical velocity from Stripe data
        )
        # Add stories to sprint 1
        story1 = UserStory(
            id='story-001',
            name='Implement /v2/payments endpoint',
            story_points=8,
            status='done',
            assignee='backend-lead'
        )
        story2 = UserStory(
            id='story-002',
            name='Add idempotency key support',
            story_points=5,
            status='done',
            assignee='senior-backend'
        )
        sprint1.stories.extend([story1, story2])
        project.add_sprint(sprint1)
        # Validate sprint
        errors = sprint1.validate()
        if errors:
            print(f'Sprint errors: {json.dumps(errors, indent=2)}')
        else:
            burndown = sprint1.calculate_burndown()
            print(f'Sprint {sprint1.name} burndown: {json.dumps(burndown, indent=2)}')
            print(f'Average project velocity: {project.calculate_average_velocity()}')
    except ValueError as e:
        print(f'Failed to add sprint: {e}')
    except Exception as e:
        print(f'Unexpected error: {e}')

package main

import (
    \"encoding/json\"
    \"fmt\"
    \"log\"
    \"errors\"
    \"time\"
)

// DeliveryModel represents either Agile or Waterfall
type DeliveryModel string

const (
    Agile     DeliveryModel = \"agile\"
    Waterfall DeliveryModel = \"waterfall\"
)

// TeamConfig holds configuration for a delivery team
type TeamConfig struct {
    TeamSize      int           `json:\"team_size\"`
    Model         DeliveryModel `json:\"model\"`
    HourlyRate    float64       `json:\"hourly_rate_usd\"`
    ProjectMonths int           `json:\"project_months\"`
    ReworkRate    float64       `json:\"rework_rate\"` // Percentage of hours spent on rework (0.0 to 1.0)
    ContextSwitchPenalty float64 `json:\"context_switch_penalty_usd\"` // Monthly penalty for context switching
}

// Validate validates team configuration
func (t *TeamConfig) Validate() error {
    if t.TeamSize <= 0 {
        return errors.New(\"team size must be positive\")
    }
    if t.Model != Agile && t.Model != Waterfall {
        return errors.New(\"model must be either agile or waterfall\")
    }
    if t.HourlyRate <= 0 {
        return errors.New(\"hourly rate must be positive\")
    }
    if t.ProjectMonths <= 0 {
        return errors.New(\"project months must be positive\")
    }
    if t.ReworkRate < 0 || t.ReworkRate > 1 {
        return errors.New(\"rework rate must be between 0 and 1\")
    }
    if t.ContextSwitchPenalty < 0 {
        return errors.New(\"context switch penalty cannot be negative\")
    }
    return nil
}

// CostBreakdown holds total cost calculations
type CostBreakdown struct {
    Model                  DeliveryModel `json:\"model\"`
    TeamSize               int           `json:\"team_size\"`
    TotalLaborCost         float64       `json:\"total_labor_cost_usd\"`
    ReworkCost             float64       `json:\"rework_cost_usd\"`
    ContextSwitchCost      float64       `json:\"context_switch_cost_usd\"`
    TotalCost              float64       `json:\"total_cost_usd\"`
    CostPerEngineer        float64       `json:\"cost_per_engineer_usd\"`
}

// CalculateCosts computes total delivery costs for a team
func CalculateCosts(config TeamConfig) (CostBreakdown, error) {
    if err := config.Validate(); err != nil {
        return CostBreakdown{}, fmt.Errorf(\"invalid config: %w\", err)
    }
    // Total working hours: 40 hours/week * 4 weeks/month * project months * team size
    totalHours := 40 * 4 * config.ProjectMonths * config.TeamSize
    laborCost := float64(totalHours) * config.HourlyRate
    reworkCost := laborCost * config.ReworkRate
    contextSwitchCost := config.ContextSwitchPenalty * float64(config.ProjectMonths)
    totalCost := laborCost + reworkCost + contextSwitchCost

    return CostBreakdown{
        Model:             config.Model,
        TeamSize:          config.TeamSize,
        TotalLaborCost:    roundToTwoDecimals(laborCost),
        ReworkCost:        roundToTwoDecimals(reworkCost),
        ContextSwitchCost: roundToTwoDecimals(contextSwitchCost),
        TotalCost:         roundToTwoDecimals(totalCost),
        CostPerEngineer:   roundToTwoDecimals(totalCost / float64(config.TeamSize)),
    }, nil
}

// roundToTwoDecimals rounds a float to two decimal places
func roundToTwoDecimals(val float64) float64 {
    return float64(int(val*100+0.5)) / 100
}

// Stripe 2026 benchmark data constants
const (
    StripeAvgReworkAgile = 0.32 // 32% rework for Agile teams (Stripe 2026 data)
    StripeAvgReworkWaterfall = 0.19 // 19% rework for Waterfall teams
    StripeContextSwitchAgile = 12500.0 // $12.5k/month per team for Agile context switching
    StripeContextSwitchWaterfall = 4200.0 // $4.2k/month per team for Waterfall
    StripeHourlyRate = 85.0 // Average Stripe engineer hourly rate USD
)

func main() {
    // Compare 52-person team (same as earlier examples) over 6 months
    agileConfig := TeamConfig{
        TeamSize:      52,
        Model:         Agile,
        HourlyRate:    StripeHourlyRate,
        ProjectMonths: 6,
        ReworkRate:    StripeAvgReworkAgile,
        ContextSwitchPenalty: StripeContextSwitchAgile,
    }
    waterfallConfig := TeamConfig{
        TeamSize:      52,
        Model:         Waterfall,
        HourlyRate:    StripeHourlyRate,
        ProjectMonths: 6,
        ReworkRate:    StripeAvgReworkWaterfall,
        ContextSwitchPenalty: StripeContextSwitchWaterfall,
    }

    agileCosts, err := CalculateCosts(agileConfig)
    if err != nil {
        log.Fatalf(\"Failed to calculate Agile costs: %v\", err)
    }
    waterfallCosts, err := CalculateCosts(waterfallConfig)
    if err != nil {
        log.Fatalf(\"Failed to calculate Waterfall costs: %v\", err)
    }

    // Output comparison as JSON
    comparison := map[string]CostBreakdown{
        \"agile\":     agileCosts,
        \"waterfall\": waterfallCosts,
    }
    output, err := json.MarshalIndent(comparison, \"\", \"  \")
    if err != nil {
        log.Fatalf(\"Failed to marshal output: %v\", err)
    }
    fmt.Printf(\"Stripe 2026 52-Person Team 6-Month Cost Comparison:\\n%s\\n\", output)

    // Calculate savings
    savings := agileCosts.TotalCost - waterfallCosts.TotalCost
    savingsPerEngineer := savings / float64(agileConfig.TeamSize)
    fmt.Printf(\"\\nWaterfall saves: $%.2f total, $%.2f per engineer\\n\", savings, savingsPerEngineer)
}

# Snippet: Create a Waterfall compliance phase with DeliveryTracker API
import requests

DELIVERY_TRACKER_URL = 'https://delivery-tracker.stripe.com/api/v2'
API_KEY = 'your-stripe-api-key'

def create_compliance_phase(project_id: str, phase_name: str, start_date: str, end_date: str):
    payload = {
        'project_id': project_id,
        'name': phase_name,
        'type': 'compliance',
        'start_date': start_date,
        'end_date': end_date,
        'required_approvals': ['compliance-lead', 'staff-engineer', 'product-lead'],
        'deliverables': ['SOC2 Type II Checklist', 'GDPR Data Map', 'Audit Log Report']
    }
    headers = {'Authorization': f'Bearer {API_KEY}'}
    response = requests.post(f'{DELIVERY_TRACKER_URL}/phases', json=payload, headers=headers)
    response.raise_for_status()
    return response.json()

# Example: Create Q3 2026 compliance phase for Stripe Payments V2
phase = create_compliance_phase(
    project_id='stripe-pay-v2',
    phase_name='Q3 2026 SOC2 Compliance Review',
    start_date='2026-07-01',
    end_date='2026-08-15'
)
print(f'Created compliance phase: {phase[\"id\"]}')

# Snippet: Audit Jira sprints for fixed scope (Waterfall trap)
from jira import Jira

def audit_sprint_scope(jira_url: str, api_key: str, project_key: str, num_sprints: int = 3):
    jira = Jira(url=jira_url, token=api_key)
    project = jira.project(project_key)
    sprints = project.sprints()[:num_sprints]  # Get last 3 sprints
    for sprint in sprints:
        issues = sprint.issues()
        scope_changes = sum(1 for issue in issues if issue.changelog.has_scope_change())
        fixed_scope_pct = round((1 - (scope_changes / len(issues))) * 100, 2) if issues else 0
        print(f'Sprint {sprint.name}: {fixed_scope_pct}% fixed scope')
        if fixed_scope_pct >= 90:
            print('⚠️  Waterfall trap: 90%+ fixed scope, adopt Waterfall explicitly')

# Example: Audit Stripe Payments Agile project
audit_sprint_scope(
    jira_url='https://stripe.atlassian.net',
    api_key='your-jira-api-key',
    project_key='PAYV2'
)

# Snippet: Prometheus metric for Waterfall phase completion rate
from prometheus_client import Gauge, start_http_server
import time

# Define metric
phase_completion_rate = Gauge(
    'waterfall_phase_completion_rate_percent',
    'Percentage of completed phases in Waterfall project',
    ['project_id', 'team_size']
)

def update_metrics(project_id: str, team_size: int, completion_rate: float):
    phase_completion_rate.labels(project_id=project_id, team_size=str(team_size)).set(completion_rate)

if __name__ == '__main__':
    # Start Prometheus metrics server on port 8000
    start_http_server(8000)
    # Example: Update metric for Stripe Payments V2 project
    update_metrics(project_id='stripe-pay-v2', team_size=52, completion_rate=96.0)
    print('Metrics server running on :8000')
    # Keep process alive
    while True:
        time.sleep(1)