How We Cut 30% of Engineer Turnover by Switching from Jira 2026 to Linear 2.0 and Slack 5.0

import os
import time
import logging
from typing import Dict, List, Optional
from jira import JIRA, JIRAError
from linear import LinearClient, LinearError
import requests

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

# Load environment variables for API credentials
JIRA_URL = os.getenv("JIRA_2026_URL", "https://jira.acme.com")
JIRA_USER = os.getenv("JIRA_USER")
JIRA_API_TOKEN = os.getenv("JIRA_API_TOKEN")
LINEAR_API_KEY = os.getenv("LINEAR_API_KEY")
LINEAR_TEAM_ID = os.getenv("LINEAR_TEAM_ID")

# Validate required environment variables
required_vars = [JIRA_USER, JIRA_API_TOKEN, LINEAR_API_KEY, LINEAR_TEAM_ID]
if any(var is None for var in required_vars):
    raise ValueError("Missing required environment variables. Check JIRA_USER, JIRA_API_TOKEN, LINEAR_API_KEY, LINEAR_TEAM_ID")

# Initialize Jira 2026 client with rate limiting (100 req/min max for Jira 2026 Cloud)
jira_client = JIRA(
    server=JIRA_URL,
    basic_auth=(JIRA_USER, JIRA_API_TOKEN),
    max_retries=3,
    retry_backoff_factor=2
)

# Initialize Linear 2.0 client with default rate limit (1000 req/min)
linear_client = LinearClient(api_key=LINEAR_API_KEY)

def fetch_jira_issues(project_key: str, batch_size: int = 50) -> List[Dict]:
    """Fetch all open issues from Jira 2026 project with pagination."""
    issues = []
    start_at = 0
    while True:
        try:
            # Jira 2026 uses JQL for filtering, fetch only open issues to reduce migration scope
            jql = f"project = {project_key} AND status != Done ORDER BY created ASC"
            batch = jira_client.search_issues(
                jql_str=jql,
                startAt=start_at,
                maxResults=batch_size,
                expand="changelog,comments"
            )
            if not batch:
                break
            for issue in batch:
                issues.append({
                    "jira_id": issue.key,
                    "title": issue.fields.summary,
                    "description": issue.fields.description or "",
                    "status": issue.fields.status.name,
                    "assignee": issue.fields.assignee.emailAddress if issue.fields.assignee else None,
                    "labels": [label.name for label in issue.fields.labels],
                    "comments": [c.body for c in issue.fields.comment.comments]
                })
            start_at += batch_size
            # Respect Jira 2026 rate limits: sleep 0.6s between batches (100 req/min)
            time.sleep(0.6)
            logger.info(f"Fetched {len(issues)} issues so far...")
        except JIRAError as e:
            logger.error(f"Jira API error: {e.status_code} - {e.text}")
            if e.status_code == 429:
                retry_after = int(e.headers.get("Retry-After", 60))
                logger.warning(f"Rate limited. Retrying after {retry_after}s")
                time.sleep(retry_after)
            else:
                raise
        except Exception as e:
            logger.error(f"Unexpected error fetching Jira issues: {e}")
            raise
    return issues

def map_jira_status_to_linear(jira_status: str) -> str:
    """Map Jira 2026 legacy statuses to Linear 2.0 default workflow 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")

def migrate_issue_to_linear(jira_issue: Dict) -> Optional[str]:
    """Migrate a single Jira issue to Linear 2.0, return Linear issue ID if successful."""
    try:
        # Map Jira assignee to Linear user (assumes email addresses match)
        assignee_id = None
        if jira_issue["assignee"]:
            users = linear_client.users(filter=f"email: {jira_issue['assignee']}")
            if users:
                assignee_id = users[0].id

        # Create Linear issue with mapped fields
        linear_issue = linear_client.issue_create(
            team_id=LINEAR_TEAM_ID,
            title=jira_issue["title"],
            description=f"Migrated from Jira {jira_issue['jira_id']}\n\n{jira_issue['description']}",
            status=map_jira_status_to_linear(jira_issue["status"]),
            assignee_id=assignee_id,
            labels=jira_issue["labels"]
        )
        logger.info(f"Migrated {jira_issue['jira_id']} to Linear {linear_issue.identifier}")

        # Migrate comments (Linear 2.0 supports batch comment creation)
        for comment in jira_issue["comments"]:
            linear_client.comment_create(
                issue_id=linear_issue.id,
                body=f"Migrated from Jira:\n{comment}"
            )

        return linear_issue.id
    except LinearError as e:
        logger.error(f"Linear API error migrating {jira_issue['jira_id']}: {e}")
        return None
    except Exception as e:
        logger.error(f"Unexpected error migrating {jira_issue['jira_id']}: {e}")
        return None

def main():
    project_key = os.getenv("JIRA_PROJECT_KEY", "ENG")
    logger.info(f"Starting migration for Jira project {project_key}")

    # Fetch all open Jira issues
    jira_issues = fetch_jira_issues(project_key)
    logger.info(f"Total Jira issues to migrate: {len(jira_issues)}")

    # Migrate issues with rate limiting (Linear 2.0 allows 1000 req/min, so 0.06s sleep)
    migrated_count = 0
    for issue in jira_issues:
        result = migrate_issue_to_linear(issue)
        if result:
            migrated_count += 1
        time.sleep(0.06)

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

if __name__ == "__main__":
    main()

import { App, LogLevel } from "@slack/bolt";
import { WebClient } from "@slack/web-api";
import { LinearClient } from "@linear/sdk";
import { Request, Response } from "express";
import dotenv from "dotenv";

dotenv.config();

// Initialize Slack 5.0 Bolt app with socket mode (no public endpoint required)
const slackApp = new App({
  token: process.env.SLACK_BOT_TOKEN,
  signingSecret: process.env.SLACK_SIGNING_SECRET,
  logLevel: LogLevel.INFO,
  socketMode: true,
  appToken: process.env.SLACK_APP_TOKEN
});

// Initialize Linear 2.0 client
const linearClient = new LinearClient({
  apiKey: process.env.LINEAR_API_KEY
});

// Slack WebClient for posting messages
const slackClient = new WebClient(process.env.SLACK_BOT_TOKEN);

// Map Linear team IDs to Slack channel IDs for notifications
const TEAM_CHANNEL_MAP: Record = {
  "team_123": "C1234567890", // Backend team
  "team_456": "C0987654321", // Frontend team
  "team_789": "C1122334455"  // DevOps team
};

// Linear 2.0 webhook event types we handle
type LinearWebhookEvent = {
  type: string;
  action: string;
  data: {
    id: string;
    identifier: string;
    title: string;
    status: string;
    team: { id: string };
    assignee?: { email: string };
  };
};

// Validate Linear webhook signature (Linear 2.0 uses HMAC-SHA256)
function validateLinearSignature(payload: string, signature: string, secret: string): boolean {
  const crypto = require("crypto");
  const hmac = crypto.createHmac("sha256", secret);
  hmac.update(payload);
  const expected = `sha256=${hmac.digest("hex")}`;
  return crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(signature));
}

// Endpoint to receive Linear 2.0 webhooks
slackApp.receiver.app.post(
  "/linear-webhook",
  async (req: Request, res: Response) => {
    try {
      // Validate webhook signature
      const signature = req.headers["linear-signature"] as string;
      const webhookSecret = process.env.LINEAR_WEBHOOK_SECRET;
      if (!validateLinearSignature(JSON.stringify(req.body), signature, webhookSecret)) {
        return res.status(401).send("Invalid signature");
      }

      const event: LinearWebhookEvent = req.body;
      const { type, action, data } = event;

      // Only handle issue creation and status updates
      if (type !== "issue" || !["create", "update"].includes(action)) {
        return res.status(200).send("Event ignored");
      }

      // Get Slack channel for the Linear team
      const channelId = TEAM_CHANNEL_MAP[data.team.id];
      if (!channelId) {
        console.warn(`No Slack channel mapped for Linear team ${data.team.id}`);
        return res.status(200).send("No channel mapped");
      }

      // Get assignee Slack ID if available
      let assigneeSlackId: string | undefined;
      if (data.assignee) {
        const slackUsers = await slackClient.users.lookupByEmail({
          email: data.assignee.email
        });
        assigneeSlackId = slackUsers.user?.id;
      }

      // Build Slack Block Kit message
      const blocks = [
        {
          type: "header",
          text: {
            type: "plain_text",
            text: `Linear Issue ${action === "create" ? "Created" : "Updated"}: ${data.identifier}`,
            emoji: true
          }
        },
        {
          type: "section",
          text: {
            type: "mrkdwn",
            text: `*${data.title}*\nStatus: ${data.status}\n${assigneeSlackId ? `Assignee: <@${assigneeSlackId}>` : "Unassigned"}`
          }
        },
        {
          type: "actions",
          elements: [
            {
              type: "button",
              text: {
                type: "plain_text",
                text: "View in Linear",
                emoji: true
              },
              url: `https://linear.app/acme/issue/${data.identifier}`
            }
          ]
        }
      ];

      // Post message to Slack channel
      await slackClient.chat.postMessage({
        channel: channelId,
        text: `Linear Issue ${data.identifier} ${action === "create" ? "created" : "updated"}`,
        blocks: blocks
      });

      res.status(200).send("Webhook processed successfully");
    } catch (error) {
      console.error("Error processing Linear webhook:", error);
      res.status(500).send("Internal server error");
    }
  }
);

// Handle Slack 5.0 slash command to create Linear issues directly from Slack
slackApp.command("/linear-create", async ({ command, ack, respond }) => {
  await ack();

  try {
    const [title, ...descriptionParts] = command.text.split("|");
    const description = descriptionParts.join("|") || "";

    // Create Linear issue from Slack command
    const issue = await linearClient.issueCreate({
      teamId: "team_123", // Default to backend team, can be extended with args
      title: title.trim(),
      description: `Created from Slack by <@${command.user_id}>\n\n${description}`,
      status: "Backlog"
    });

    await respond({
      text: `Created Linear issue ${issue.identifier}: ${issue.title}`,
      blocks: [
        {
          type: "section",
          text: {
            type: "mrkdwn",
            text: `✅ Created *${issue.identifier}*: ${issue.title}\n`
          }
        }
      ]
    });
  } catch (error) {
    console.error("Error creating Linear issue from Slack:", error);
    await respond({
      text: "Failed to create Linear issue. Please check permissions and try again.",
      response_type: "ephemeral"
    });
  }
});

// Start the Slack 5.0 app
(async () => {
  const port = process.env.PORT || 3000;
  await slackApp.start(port);
  console.log(`Slack 5.0 Linear integration app listening on port ${port}`);
})();

import os
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
from datetime import datetime, timedelta
from linear import LinearClient, LinearError
import requests
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Load environment variables
LINEAR_API_KEY = os.getenv("LINEAR_API_KEY")
HR_DATA_PATH = os.getenv("HR_DATA_PATH", "hr_data.csv")
OUTPUT_DIR = os.getenv("OUTPUT_DIR", "./analytics_output")

# Initialize Linear 2.0 client
linear_client = LinearClient(api_key=LINEAR_API_KEY)

def fetch_linear_team_metrics(team_id: str, start_date: datetime, end_date: datetime) -> pd.DataFrame:
    """Fetch issue cycle time and assignment metrics from Linear 2.0 API."""
    metrics = []
    try:
        # Fetch all issues for the team in date range
        issues = linear_client.issues(
            filter=f"team: {team_id} AND createdAt: {{>={start_date.isoformat()} <={end_date.isoformat()}}}",
            expand="assignee, cycle"
        )
        for issue in issues:
            # Calculate cycle time (time from start to completion)
            cycle_time = None
            if issue.startedAt and issue.completedAt:
                cycle_time = (issue.completedAt - issue.startedAt).total_seconds() / 3600  # hours

            # Check if issue was reassigned (context switch indicator)
            reassign_count = len(issue.assigneeHistory) if issue.assigneeHistory else 0

            metrics.append({
                "issue_id": issue.identifier,
                "created_at": issue.createdAt,
                "completed_at": issue.completedAt,
                "cycle_time_hours": cycle_time,
                "reassign_count": reassign_count,
                "assignee_email": issue.assignee.email if issue.assignee else None,
                "status": issue.status.name
            })
        return pd.DataFrame(metrics)
    except LinearError as e:
        logger.error(f"Linear API error fetching team metrics: {e}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error fetching Linear metrics: {e}")
        raise

def load_hr_turnover_data() -> pd.DataFrame:
    """Load HR turnover data from CSV (expects columns: email, start_date, end_date, reason)."""
    try:
        df = pd.read_csv(HR_DATA_PATH, parse_dates=["start_date", "end_date"])
        # Filter to engineering roles only
        df = df[df["reason"].str.contains("engineering|engineer", case=False, na=False)]
        return df
    except FileNotFoundError:
        logger.warning(f"HR data file not found at {HR_DATA_PATH}. Using mock data.")
        # Generate mock turnover data for demonstration
        dates = pd.date_range(start="2025-01-01", end="2026-03-31", freq="M")
        mock_data = []
        for date in dates:
            for i in range(3):  # 3 turnovers per month
                mock_data.append({
                    "email": f"engineer{i}@acme.com",
                    "start_date": date - timedelta(days=365),
                    "end_date": date,
                    "reason": "Resigned"
                })
        return pd.DataFrame(mock_data)

def calculate_turnover_correlation(linear_df: pd.DataFrame, hr_df: pd.DataFrame) -> pd.DataFrame:
    """Correlate Linear productivity metrics with turnover events."""
    # Merge Linear assignee data with HR data
    merged = pd.merge(
        linear_df,
        hr_df,
        left_on="assignee_email",
        right_on="email",
        how="inner"
    )

    # Calculate per-engineer metrics
    engineer_metrics = merged.groupby("assignee_email").agg(
        avg_cycle_time=("cycle_time_hours", "mean"),
        avg_reassign_count=("reassign_count", "mean"),
        turnover_date=("end_date", "max")
    ).reset_index()

    # Calculate correlation between reassignment count and turnover
    if len(engineer_metrics) > 1:
        corr = engineer_metrics["avg_reassign_count"].corr(
            pd.to_datetime(engineer_metrics["turnover_date"]).astype(int) / 1e9
        )
        logger.info(f"Correlation between reassignments and turnover: {corr:.2f}")

    return engineer_metrics

def generate_turnover_report(linear_df: pd.DataFrame, hr_df: pd.DataFrame, engineer_metrics: pd.DataFrame):
    """Generate visualization and report of turnover metrics."""
    os.makedirs(OUTPUT_DIR, exist_ok=True)

    # Plot 1: Monthly turnover rate vs avg cycle time
    monthly_turnover = hr_df.groupby(pd.Grouper(key="end_date", freq="M")).size()
    monthly_cycle_time = linear_df.groupby(pd.Grouper(key="created_at", freq="M"))["cycle_time_hours"].mean()

    fig, ax1 = plt.subplots(figsize=(12, 6))
    ax1.set_xlabel("Month")
    ax1.set_ylabel("Turnover Count", color="tab:red")
    ax1.plot(monthly_turnover.index, monthly_turnover.values, color="tab:red", label="Monthly Turnover")
    ax1.tick_params(axis="y", labelcolor="tab:red")

    ax2 = ax1.twinx()
    ax2.set_ylabel("Avg Cycle Time (hours)", color="tab:blue")
    ax2.plot(monthly_cycle_time.index, monthly_cycle_time.values, color="tab:blue", label="Avg Cycle Time")
    ax2.tick_params(axis="y", labelcolor="tab:blue")

    plt.title("Monthly Engineering Turnover vs Linear 2.0 Avg Cycle Time")
    fig.tight_layout()
    plt.savefig(f"{OUTPUT_DIR}/turnover_vs_cycle_time.png")
    logger.info(f"Saved turnover vs cycle time plot to {OUTPUT_DIR}")

    # Plot 2: Reassignment count vs turnover risk
    if not engineer_metrics.empty:
        plt.figure(figsize=(10, 6))
        plt.scatter(
            engineer_metrics["avg_reassign_count"],
            pd.to_datetime(engineer_metrics["turnover_date"]).astype(int) / 1e9,
            c="tab:orange",
            alpha=0.6
        )
        plt.xlabel("Average Reassignment Count per Issue")
        plt.ylabel("Turnover Date (Unix Timestamp)")
        plt.title("Linear Issue Reassignments vs Engineer Turnover Date")
        plt.savefig(f"{OUTPUT_DIR}/reassign_vs_turnover.png")
        logger.info(f"Saved reassignment vs turnover plot to {OUTPUT_DIR}")

    # Save summary report
    summary = {
        "total_turnover_6m": len(hr_df[hr_df["end_date"] > datetime.now() - timedelta(days=180)]),
        "avg_cycle_time_hours": linear_df["cycle_time_hours"].mean(),
        "avg_reassign_count": linear_df["reassign_count"].mean(),
        "correlation_reassign_turnover": engineer_metrics["avg_reassign_count"].corr(
            pd.to_datetime(engineer_metrics["turnover_date"]).astype(int) / 1e9
        ) if not engineer_metrics.empty else None
    }

    with open(f"{OUTPUT_DIR}/summary_report.json", "w") as f:
        import json
        json.dump(summary, f, indent=2)
    logger.info(f"Saved summary report to {OUTPUT_DIR}/summary_report.json")

def main():
    # Define analysis period: last 6 months
    end_date = datetime.now()
    start_date = end_date - timedelta(days=180)

    # Fetch Linear metrics for all engineering teams
    team_ids = ["team_123", "team_456", "team_789"]
    linear_dfs = []
    for team_id in team_ids:
        logger.info(f"Fetching Linear metrics for team {team_id}")
        df = fetch_linear_team_metrics(team_id, start_date, end_date)
        linear_dfs.append(df)
    linear_df = pd.concat(linear_dfs, ignore_index=True)

    # Load HR turnover data
    hr_df = load_hr_turnover_data()

    # Calculate correlation
    engineer_metrics = calculate_turnover_correlation(linear_df, hr_df)

    # Generate report
    generate_turnover_report(linear_df, hr_df, engineer_metrics)

    logger.info("Analytics report generation complete.")

if __name__ == "__main__":
    main()

# Audit Jira 2026 workflow usage
from jira import JIRA
jira = JIRA(server="https://jira.acme.com", basic_auth=("user", "token"))
workflows = jira.workflows()
for wf in workflows:
    status_count = {}
    issues = jira.search_issues(f"workflow = {wf.name}", maxResults=500)
    for issue in issues:
        status = issue.fields.status.name
        status_count[status] = status_count.get(status, 0) + 1
    print(f"Workflow {wf.name}: {status_count}")

{
  "blocks": [
    {
      "type": "header",
      "text": {
        "type": "plain_text",
        "text": "ENG-123: Fix p99 latency spike",
        "emoji": true
      }
    },
    {
      "type": "section",
      "text": {
        "type": "mrkdwn",
        "text": "*Status:* In Progress\n*Assignee:* @backend-lead\n*Priority:* High"
      }
    },
    {
      "type": "actions",
      "elements": [
        {
          "type": "button",
          "text": { "type": "plain_text", "text": "Claim Issue" },
          "action_id": "claim_linear_issue",
          "value": "ENG-123"
        },
        {
          "type": "button",
          "text": { "type": "plain_text", "text": "View in Linear" },
          "url": "https://linear.app/acme/issue/ENG-123"
        }
      ]
    }
  ]
}

# Prometheus counter for engineer context switches
from prometheus_client import Counter, start_http_server

context_switch_counter = Counter(
    "engineer_context_switches_total",
    "Total number of issue reassignments per engineer",
    ["engineer_email", "team_id"]
)

def track_linear_reassignment(issue_id: str, old_assignee: str, new_assignee: str):
    if old_assignee and new_assignee and old_assignee != new_assignee:
        context_switch_counter.labels(
            engineer_email=new_assignee,
            team_id="team_123"
        ).inc()
        print(f"Tracked context switch for {new_assignee} on {issue_id}")