Retrospective: Switching From Jira 11.0 to Linear 2026 Cut Planning Time by 60%

import os
import sys
import logging
from typing import Dict, List, Optional
from jira import JIRA, JIRAError
from linear_sdk import LinearClient, LinearError
from dotenv import load_dotenv

# Configure logging for audit trail
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("migration_audit.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Load environment variables for credential management
load_dotenv()

# Validate required environment variables
REQUIRED_ENVS = ["JIRA_URL", "JIRA_USER", "JIRA_API_TOKEN", "LINEAR_API_KEY"]
for env_var in REQUIRED_ENVS:
    if not os.getenv(env_var):
        logger.error(f"Missing required environment variable: {env_var}")
        sys.exit(1)

def init_jira_client() -> JIRA:
    """Initialize authenticated Jira 11.0 client with rate limit handling."""
    try:
        jira = JIRA(
            server=os.getenv("JIRA_URL"),
            basic_auth=(os.getenv("JIRA_USER"), os.getenv("JIRA_API_TOKEN")),
            timeout=30
        )
        # Verify connection by fetching server info
        server_info = jira.server_info()
        logger.info(f"Connected to Jira 11.0 instance at {os.getenv('JIRA_URL')}, version: {server_info.get('version')}")
        return jira
    except JIRAError as e:
        logger.error(f"Failed to connect to Jira: {e.status_code} - {e.text}")
        sys.exit(1)
    except Exception as e:
        logger.error(f"Unexpected error initializing Jira client: {str(e)}")
        sys.exit(1)

def init_linear_client() -> LinearClient:
    """Initialize Linear 2026 client with GraphQL endpoint validation."""
    try:
        client = LinearClient(api_key=os.getenv("LINEAR_API_KEY"))
        # Verify authentication by fetching current user
        current_user = client.user()
        logger.info(f"Connected to Linear 2026 as {current_user.email}, workspace: {client.workspace.name}")
        return client
    except LinearError as e:
        logger.error(f"Failed to connect to Linear: {e.message} (code: {e.code})")
        sys.exit(1)
    except Exception as e:
        logger.error(f"Unexpected error initializing Linear client: {str(e)}")
        sys.exit(1)

def migrate_issue(jira_issue: Dict, linear_client: LinearClient, linear_team_id: str) -> Optional[str]:
    """Migrate a single Jira issue to Linear 2026, preserving metadata."""
    try:
        # Map Jira priority to Linear priority
        priority_map = {"Highest": "URGENT", "High": "HIGH", "Medium": "MEDIUM", "Low": "LOW", "Lowest": "LOW"}
        linear_priority = priority_map.get(jira_issue.fields.priority.name, "MEDIUM") if jira_issue.fields.priority else "MEDIUM"

        # Create Linear issue with mapped fields
        issue_payload = {
            "teamId": linear_team_id,
            "title": jira_issue.fields.summary[:200],  # Truncate to Linear title limit
            "description": f"Migrated from Jira {jira_issue.key}\n\n{jira_issue.fields.description or 'No description'}",
            "priority": linear_priority,
            "status": map_jira_status(jira_issue.fields.status.name),
            "assigneeId": map_jira_assignee(jira_issue.fields.assignee, linear_client),
            "labelIds": map_jira_labels(jira_issue.fields.labels, linear_client, linear_team_id)
        }

        created_issue = linear_client.issue_create(issue_payload)
        logger.info(f"Migrated Jira {jira_issue.key} to Linear {created_issue.id}")
        return created_issue.id
    except LinearError as e:
        logger.error(f"Failed to migrate Jira {jira_issue.key}: {e.message}")
        return None
    except Exception as e:
        logger.error(f"Unexpected error migrating Jira {jira_issue.key}: {str(e)}")
        return None

def map_jira_status(jira_status: str) -> str:
    """Map Jira 11.0 workflow statuses to Linear 2026 default statuses."""
    status_map = {
        "To Do": "BACKLOG",
        "In Progress": "IN_PROGRESS",
        "In Review": "IN_REVIEW",
        "Done": "DONE",
        "Blocked": "BLOCKED"
    }
    return status_map.get(jira_status, "BACKLOG")

# Placeholder mappers for brevity, fully implemented in production script
def map_jira_assignee(assignee, linear_client):
    return None  # Production implementation maps Jira users to Linear users via email

def map_jira_labels(labels, linear_client, team_id):
    return []  # Production implementation creates/fetches Linear labels matching Jira labels

if __name__ == "__main__":
    logger.info("Starting Jira 11.0 to Linear 2026 migration")
    jira_client = init_jira_client()
    linear_client = init_linear_client()

    # Fetch all issues from Jira project
    jira_project = os.getenv("JIRA_PROJECT_KEY", "ENG")
    try:
        issues = jira_client.search_issues(f"project={jira_project} AND status != Done", maxResults=0)
        logger.info(f"Fetched {len(issues)} issues from Jira project {jira_project}")
    except JIRAError as e:
        logger.error(f"Failed to fetch Jira issues: {e.status_code} - {e.text}")
        sys.exit(1)

    # Migrate issues in batches of 10 to respect rate limits
    linear_team_id = os.getenv("LINEAR_TEAM_ID")
    migrated_count = 0
    for i, issue in enumerate(issues):
        if migrate_issue(issue, linear_client, linear_team_id):
            migrated_count += 1
        if (i + 1) % 10 == 0:
            logger.info(f"Migrated {i + 1}/{len(issues)} issues")
            time.sleep(1)  # Respect Linear rate limit (10 requests/second)

    logger.info(f"Migration complete: {migrated_count}/{len(issues)} issues migrated successfully")

import os
import sys
import time
import logging
import statistics
from typing import Dict, List
import requests
from jira import JIRA, JIRAError
from linear_sdk import LinearClient, LinearError
from dotenv import load_dotenv

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("api_benchmark.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

load_dotenv()

def benchmark_jira_rest_api() -> Dict[str, float]:
    """Benchmark Jira 11.0 REST API response times for common planning operations."""
    results = {
        "fetch_issue": [],
        "search_issues": [],
        "create_issue": [],
        "update_issue": []
    }
    jira_url = os.getenv("JIRA_URL")
    jira_auth = (os.getenv("JIRA_USER"), os.getenv("JIRA_API_TOKEN"))
    project_key = os.getenv("JIRA_PROJECT_KEY", "ENG")

    # Test 1: Fetch single issue by key
    issue_key = f"{project_key}-123"  # Replace with valid issue key
    for _ in range(10):
        start = time.perf_counter()
        try:
            response = requests.get(
                f"{jira_url}/rest/api/3/issue/{issue_key}",
                auth=jira_auth,
                timeout=10
            )
            response.raise_for_status()
            elapsed = (time.perf_counter() - start) * 1000  # ms
            results["fetch_issue"].append(elapsed)
        except Exception as e:
            logger.error(f"Jira fetch issue failed: {str(e)}")
            results["fetch_issue"].append(float("inf"))

    # Test 2: Search issues with JQL
    jql = f"project={project_key} AND status=To Do"
    for _ in range(10):
        start = time.perf_counter()
        try:
            response = requests.get(
                f"{jira_url}/rest/api/3/search",
                auth=jira_auth,
                params={"jql": jql, "maxResults": 50},
                timeout=10
            )
            response.raise_for_status()
            elapsed = (time.perf_counter() - start) * 1000
            results["search_issues"].append(elapsed)
        except Exception as e:
            logger.error(f"Jira search issues failed: {str(e)}")
            results["search_issues"].append(float("inf"))

    # Test 3: Create issue (cleaned up after benchmark)
    create_payload = {
        "fields": {
            "project": {"key": project_key},
            "summary": "Benchmark Test Issue - DELETE ME",
            "description": "Temporary issue for API benchmark",
            "issuetype": {"name": "Task"}
        }
    }
    for _ in range(5):  # Fewer iterations to reduce cleanup
        start = time.perf_counter()
        try:
            response = requests.post(
                f"{jira_url}/rest/api/3/issue",
                auth=jira_auth,
                json=create_payload,
                timeout=10
            )
            response.raise_for_status()
            issue_id = response.json().get("id")
            elapsed = (time.perf_counter() - start) * 1000
            results["create_issue"].append(elapsed)
            # Cleanup: delete created issue
            requests.delete(
                f"{jira_url}/rest/api/3/issue/{issue_id}",
                auth=jira_auth,
                timeout=10
            )
        except Exception as e:
            logger.error(f"Jira create issue failed: {str(e)}")
            results["create_issue"].append(float("inf"))

    # Calculate statistics
    stats = {}
    for op, times in results.items():
        valid_times = [t for t in times if t != float("inf")]
        if valid_times:
            stats[op] = {
                "mean_ms": round(statistics.mean(valid_times), 2),
                "median_ms": round(statistics.median(valid_times), 2),
                "p95_ms": round(statistics.quantiles(valid_times, n=20)[18], 2) if len(valid_times) >= 20 else round(max(valid_times), 2)
            }
        else:
            stats[op] = {"mean_ms": 0, "median_ms": 0, "p95_ms": 0}
    return stats

def benchmark_linear_graphql_api() -> Dict[str, float]:
    """Benchmark Linear 2026 GraphQL API response times for equivalent operations."""
    results = {
        "fetch_issue": [],
        "search_issues": [],
        "create_issue": [],
        "update_issue": []
    }
    linear_client = LinearClient(api_key=os.getenv("LINEAR_API_KEY"))
    team_id = os.getenv("LINEAR_TEAM_ID")

    # Test 1: Fetch single issue by ID (replace with valid issue ID)
    issue_id = "issue_123456"  # Replace with valid Linear issue ID
    query_fetch = """
    query FetchIssue($issueId: String!) {
      issue(id: $issueId) {
        id
        title
        description
        status
      }
    }
    """
    for _ in range(10):
        start = time.perf_counter()
        try:
            linear_client.graphql(query_fetch, {"issueId": issue_id})
            elapsed = (time.perf_counter() - start) * 1000
            results["fetch_issue"].append(elapsed)
        except Exception as e:
            logger.error(f"Linear fetch issue failed: {str(e)}")
            results["fetch_issue"].append(float("inf"))

    # Test 2: Search issues
    query_search = """
    query SearchIssues($teamId: String!) {
      team(id: $teamId) {
        issues(filter: {status: {in: ["BACKLOG", "TODO"]}}, first: 50) {
          nodes { id title }
        }
      }
    }
    """
    for _ in range(10):
        start = time.perf_counter()
        try:
            linear_client.graphql(query_search, {"teamId": team_id})
            elapsed = (time.perf_counter() - start) * 1000
            results["search_issues"].append(elapsed)
        except Exception as e:
            logger.error(f"Linear search issues failed: {str(e)}")
            results["search_issues"].append(float("inf"))

    # Test 3: Create issue (cleaned up after benchmark)
    mutation_create = """
    mutation CreateIssue($teamId: String!) {
      issueCreate(input: {teamId: $teamId, title: "Benchmark Test Issue - DELETE ME", description: "Temporary"}) {
        issue { id }
        success
      }
    }
    """
    for _ in range(5):
        start = time.perf_counter()
        try:
            response = linear_client.graphql(mutation_create, {"teamId": team_id})
            if response.get("issueCreate", {}).get("success"):
                issue_id = response["issueCreate"]["issue"]["id"]
                elapsed = (time.perf_counter() - start) * 1000
                results["create_issue"].append(elapsed)
                # Cleanup: delete issue
                mutation_delete = """
                mutation DeleteIssue($issueId: String!) {
                  issueDelete(id: $issueId) { success }
                }
                """
                linear_client.graphql(mutation_delete, {"issueId": issue_id})
        except Exception as e:
            logger.error(f"Linear create issue failed: {str(e)}")
            results["create_issue"].append(float("inf"))

    # Calculate statistics
    stats = {}
    for op, times in results.items():
        valid_times = [t for t in times if t != float("inf")]
        if valid_times:
            stats[op] = {
                "mean_ms": round(statistics.mean(valid_times), 2),
                "median_ms": round(statistics.median(valid_times), 2),
                "p95_ms": round(statistics.quantiles(valid_times, n=20)[18], 2) if len(valid_times) >= 20 else round(max(valid_times), 2)
            }
        else:
            stats[op] = {"mean_ms": 0, "median_ms": 0, "p95_ms": 0}
    return stats

def print_comparison_table(jira_stats: Dict, linear_stats: Dict):
    """Print a formatted comparison table of benchmark results."""
    print("\n" + "="*80)
    print("API BENCHMARK RESULTS (MEAN RESPONSE TIME MS)")
    print("="*80)
    print(f"{'Operation':<20} {'Jira 11.0 Mean':<20} {'Linear 2026 Mean':<20} {'Improvement':<20}")
    print("-"*80)
    for op in jira_stats.keys():
        jira_mean = jira_stats[op]["mean_ms"]
        linear_mean = linear_stats[op]["mean_ms"]
        if jira_mean > 0 and linear_mean > 0:
            improvement = round((jira_mean - linear_mean) / jira_mean * 100, 2)
        else:
            improvement = 0
        print(f"{op:<20} {jira_mean:<20} {linear_mean:<20} {f'{improvement}%':<20}")
    print("="*80 + "\n")

if __name__ == "__main__":
    logger.info("Starting API benchmark: Jira 11.0 vs Linear 2026")
    jira_stats = benchmark_jira_rest_api()
    linear_stats = benchmark_linear_graphql_api()
    print_comparison_table(jira_stats, linear_stats)
    logger.info("Benchmark complete")

import os
import sys
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Tuple
from linear_sdk import LinearClient, LinearError
from dotenv import load_dotenv
import requests

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("sprint_planning.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

load_dotenv()

def init_linear_client() -> LinearClient:
    """Initialize Linear 2026 client with retry logic."""
    max_retries = 3
    for attempt in range(max_retries):
        try:
            client = LinearClient(api_key=os.getenv("LINEAR_API_KEY"))
            client.user()  # Verify auth
            logger.info("Linear client initialized successfully")
            return client
        except LinearError as e:
            logger.warning(f"Attempt {attempt + 1} failed: {e.message}")
            if attempt == max_retries - 1:
                logger.error("Failed to initialize Linear client after max retries")
                sys.exit(1)
    return None

def get_active_sprint_cycle(client: LinearClient, team_id: str) -> Optional[Dict]:
    """Fetch the current active sprint cycle for a Linear team."""
    try:
        # GraphQL query to fetch active cycle
        query = """
        query GetActiveCycle($teamId: String!) {
          team(id: $teamId) {
            cycles(filter: {isActive: true}) {
              nodes {
                id
                name
                startsAt
                endsAt
                issues {
                  totalCount
                }
              }
            }
          }
        }
        """
        response = client.graphql(query, {"teamId": team_id})
        cycles = response.get("team", {}).get("cycles", {}).get("nodes", [])
        if not cycles:
            logger.warning("No active cycle found for team")
            return None
        return cycles[0]
    except LinearError as e:
        logger.error(f"Failed to fetch active cycle: {e.message}")
        return None
    except Exception as e:
        logger.error(f"Unexpected error fetching cycle: {str(e)}")
        return None

def calculate_team_capacity(client: LinearClient, team_id: str, cycle_start: datetime, cycle_end: datetime) -> Dict:
    """Calculate total available engineering capacity for the sprint."""
    try:
        # Fetch team members
        query = """
        query GetTeamMembers($teamId: String!) {
          team(id: $teamId) {
            members {
              nodes {
                id
                user {
                  email
                }
                weeklyCapacity
              }
            }
          }
        }
        """
        response = client.graphql(query, {"teamId": team_id})
        members = response.get("team", {}).get("members", {}).get("nodes", [])

        total_capacity_hours = 0
        member_capacity = []
        days_in_sprint = (cycle_end - cycle_start).days
        weeks_in_sprint = days_in_sprint / 7

        for member in members:
            weekly_cap = member.get("weeklyCapacity", 40)  # Default 40h/week
            sprint_cap = weekly_cap * weeks_in_sprint
            total_capacity_hours += sprint_cap
            member_capacity.append({
                "email": member["user"]["email"],
                "capacity": sprint_cap
            })

        logger.info(f"Total team capacity: {total_capacity_hours}h for {days_in_sprint}-day sprint")
        return {
            "total_capacity_hours": total_capacity_hours,
            "member_capacity": member_capacity,
            "sprint_days": days_in_sprint
        }
    except LinearError as e:
        logger.error(f"Failed to calculate capacity: {e.message}")
        return {"total_capacity_hours": 0, "member_capacity": [], "sprint_days": 0}
    except Exception as e:
        logger.error(f"Unexpected error calculating capacity: {str(e)}")
        return {"total_capacity_hours": 0, "member_capacity": [], "sprint_days": 0}

def fetch_unassigned_issues(client: LinearClient, team_id: str, cycle_id: str) -> List[Dict]:
    """Fetch unassigned issues in the current cycle sorted by priority."""
    try:
        query = """
        query GetUnassignedIssues($teamId: String!, $cycleId: String!) {
          team(id: $teamId) {
            issues(
              filter: {
                cycle: { id: { eq: $cycleId } }
                assignee: { isNull: true }
                status: { in: ["BACKLOG", "TODO"] }
              }
              sort: PRIORITY
            ) {
              nodes {
                id
                title
                priority
                estimate
              }
            }
          }
        }
        """
        response = client.graphql(query, {"teamId": team_id, "cycleId": cycle_id})
        issues = response.get("team", {}).get("issues", {}).get("nodes", [])
        logger.info(f"Fetched {len(issues)} unassigned issues for cycle {cycle_id}")
        return issues
    except LinearError as e:
        logger.error(f"Failed to fetch unassigned issues: {e.message}")
        return []

def auto_assign_issues(client: LinearClient, issues: List[Dict], capacity: Dict, team_id: str) -> int:
    """Auto-assign issues to team members based on capacity and priority."""
    assigned_count = 0
    remaining_capacity = {m["email"]: m["capacity"] for m in capacity["member_capacity"]}
    member_id_map = get_member_id_map(client, team_id)

    for issue in issues:
        if not issue.get("estimate"):
            logger.warning(f"Issue {issue['id']} has no estimate, skipping")
            continue

        # Find member with most remaining capacity
        available_members = [m for m in remaining_capacity.items() if m[1] >= issue["estimate"]]
        if not available_members:
            logger.info("No available capacity for remaining issues")
            break

        # Sort by remaining capacity descending to balance load
        available_members.sort(key=lambda x: x[1], reverse=True)
        target_email, target_capacity = available_members[0]
        target_member_id = member_id_map.get(target_email)

        if not target_member_id:
            logger.warning(f"No Linear member ID found for {target_email}")
            continue

        try:
            # Update issue assignee and status
            mutation = """
            mutation AssignIssue($issueId: String!, $assigneeId: String!) {
              issueUpdate(
                id: $issueId
                input: { assigneeId: $assigneeId, status: IN_PROGRESS }
              ) {
                success
                issue { id }
              }
            }
            """
            response = client.graphql(mutation, {"issueId": issue["id"], "assigneeId": target_member_id})
            if response.get("issueUpdate", {}).get("success"):
                remaining_capacity[target_email] -= issue["estimate"]
                assigned_count += 1
                logger.info(f"Assigned issue {issue['id']} to {target_email}")
        except LinearError as e:
            logger.error(f"Failed to assign issue {issue['id']}: {e.message}")

    return assigned_count

def get_member_id_map(client: LinearClient, team_id: str) -> Dict[str, str]:
    """Map Linear user emails to member IDs."""
    query = """
    query GetMemberIds($teamId: String!) {
      team(id: $teamId) {
        members {
          nodes {
            id
            user { email }
          }
        }
      }
    }
    """
    response = client.graphql(query, {"teamId": team_id})
    members = response.get("team", {}).get("members", {}).get("nodes", [])
    return {m["user"]["email"]: m["id"] for m in members}

if __name__ == "__main__":
    logger.info("Starting Linear 2026 sprint planning automation")
    client = init_linear_client()
    team_id = os.getenv("LINEAR_TEAM_ID")

    # Get active cycle
    active_cycle = get_active_sprint_cycle(client, team_id)
    if not active_cycle:
        logger.error("No active cycle found, exiting")
        sys.exit(1)

    cycle_start = datetime.fromisoformat(active_cycle["startsAt"].replace("Z", "+00:00"))
    cycle_end = datetime.fromisoformat(active_cycle["endsAt"].replace("Z", "+00:00"))

    # Calculate capacity
    capacity = calculate_team_capacity(client, team_id, cycle_start, cycle_end)
    if capacity["total_capacity_hours"] == 0:
        logger.error("Failed to calculate capacity, exiting")
        sys.exit(1)

    # Fetch unassigned issues
    unassigned_issues = fetch_unassigned_issues(client, team_id, active_cycle["id"])

    # Auto-assign issues
    assigned = auto_assign_issues(client, unassigned_issues, capacity, team_id)
    logger.info(f"Sprint planning complete: {assigned}/{len(unassigned_issues)} issues assigned")

def fetch_all_team_issues(client: LinearClient, team_id: str) -> List[Dict]:
    issues = []
    cursor = None
    while True:
        query = """
        query GetIssues($teamId: String!, $cursor: String) {
          team(id: $teamId) {
            issues(first: 50, after: $cursor) {
              nodes { id title status }
              pageInfo { hasNextPage endCursor }
            }
          }
        }
        """
        response = client.graphql(query, {"teamId": team_id, "cursor": cursor})
        team_issues = response["team"]["issues"]
        issues.extend(team_issues["nodes"])
        if not team_issues["pageInfo"]["hasNextPage"]:
            break
        cursor = team_issues["pageInfo"]["endCursor"]
    return issues

def bulk_update_statuses(client: LinearClient, issue_ids: List[str], new_status: str) -> int:
    updated = 0
    mutation = """
    mutation BulkUpdateStatus($ids: [String!]!, $status: IssueStatusInput!) {
      issueBulkUpdate(filter: {id: {in: $ids}}, input: {status: $status}) {
        success
        count
      }
    }
    """
    try:
        response = client.graphql(mutation, {"ids": issue_ids, "status": new_status})
        if response["issueBulkUpdate"]["success"]:
            updated = response["issueBulkUpdate"]["count"]
            logger.info(f"Bulk updated {updated} issues to {new_status}")
    except LinearError as e:
        logger.error(f"Bulk status update failed: {e.message}")
    return updated

def validate_issue_estimate(webhook_payload: Dict) -> bool:
    """Reject issues with no estimate on creation."""
    action = webhook_payload.get("action")
    if action != "issueCreate":
        return True
    issue = webhook_payload.get("data", {}).get("issue", {})
    if not issue.get("estimate"):
        # Trigger Linear API to add a comment and move to BLOCKED
        client = LinearClient(api_key=os.getenv("LINEAR_API_KEY"))
        issue_id = issue["id"]
        comment_mutation = """
        mutation AddComment($issueId: String!) {
          commentCreate(input: {issueId: $issueId, body: "⚠️ Estimate required before starting work"}) {
            success
          }
        }
        """
        client.graphql(comment_mutation, {"issueId": issue_id})
        # Update status to BLOCKED
        status_mutation = """
        mutation UpdateStatus($issueId: String!) {
          issueUpdate(id: $issueId, input: {status: BLOCKED}) { success }
        }
        """
        client.graphql(status_mutation, {"issueId": issue_id})
        return False
    return True