We Fixed: Increasing Our Team’s Promotion Rate by 35% with Clear Career Ladders

import json
import os
import sys
from typing import Dict, List, Optional, TypedDict
from datetime import datetime

class PromotionPacket(TypedDict):
    employee_id: str
    current_level: str
    target_level: str
    contributions: List[Dict[str, str]]
    peer_reviews: List[Dict[str, str]]
    manager_assessment: Dict[str, str]

class CareerLadderCriterion(TypedDict):
    id: str
    level: str
    domain: str
    description: str
    benchmark_examples: List[str]
    required_evidence_count: int

class PromotionGap:
    def __init__(self, criterion_id: str, missing_evidence: int, description: str):
        self.criterion_id = criterion_id
        self.missing_evidence = missing_evidence
        self.description = description

    def __repr__(self):
        return f"Gap(criterion={self.criterion_id}, missing={self.missing_evidence}, desc={self.description})"

def load_career_ladder(ladder_path: str) -> List[CareerLadderCriterion]:
    """Load benchmarked career ladder criteria from a JSON file.

    Args:
        ladder_path: Absolute path to ladder JSON file

    Returns:
        List of criterion objects

    Raises:
        FileNotFoundError: If ladder file does not exist
        json.JSONDecodeError: If ladder file is invalid JSON
    """
    if not os.path.exists(ladder_path):
        raise FileNotFoundError(f"Career ladder file not found at {ladder_path}")

    try:
        with open(ladder_path, 'r') as f:
            ladder_data = json.load(f)
    except json.JSONDecodeError as e:
        raise json.JSONDecodeError(f"Invalid JSON in ladder file: {e.msg}", e.doc, e.pos)

    # Validate ladder structure
    required_keys = {"id", "level", "domain", "description", "benchmark_examples", "required_evidence_count"}
    for idx, criterion in enumerate(ladder_data):
        if not all(key in criterion for key in required_keys):
            raise ValueError(f"Criterion at index {idx} missing required keys: {required_keys - set(criterion.keys())}")
        if not isinstance(criterion["required_evidence_count"], int) or criterion["required_evidence_count"] < 0:
            raise ValueError(f"Criterion {criterion['id']} has invalid required_evidence_count")

    return ladder_data

def validate_promotion_packet(packet: PromotionPacket, ladder: List[CareerLadderCriterion]) -> List[PromotionGap]:
    """Validate a promotion packet against career ladder criteria for the target level.

    Args:
        packet: Parsed promotion packet data
        ladder: Loaded career ladder criteria

    Returns:
        List of PromotionGap objects highlighting missing evidence
    """
    target_level = packet["target_level"]
    relevant_criteria = [c for c in ladder if c["level"] == target_level]
    gaps = []

    for criterion in relevant_criteria:
        # Count evidence matching this criterion in contributions and peer reviews
        evidence_count = 0
        criterion_keywords = set(criterion["description"].lower().split() + [ex.lower() for ex in criterion["benchmark_examples"]])

        # Check contributions
        for contribution in packet["contributions"]:
            contribution_text = f"{contribution.get('title', '')} {contribution.get('description', '')}".lower()
            if any(keyword in contribution_text for keyword in criterion_keywords):
                evidence_count += 1

        # Check peer reviews
        for review in packet["peer_reviews"]:
            review_text = f"{review.get('feedback', '')} {review.get('examples', '')}".lower()
            if any(keyword in review_text for keyword in criterion_keywords):
                evidence_count += 1

        # Check manager assessment
        manager_text = f"{packet['manager_assessment'].get('summary', '')} {packet['manager_assessment'].get('examples', '')}".lower()
        if any(keyword in manager_text for keyword in criterion_keywords):
            evidence_count += 1

        if evidence_count < criterion["required_evidence_count"]:
            gaps.append(PromotionGap(
                criterion_id=criterion["id"],
                missing_evidence=criterion["required_evidence_count"] - evidence_count,
                description=criterion["description"]
            ))

    return gaps

def generate_promotion_report(packet: PromotionPacket, gaps: List[PromotionGap]) -> str:
    """Generate a human-readable promotion readiness report."""
    report_lines = [
        f"Promotion Readiness Report for Employee {packet['employee_id']}",
        f"Current Level: {packet['current_level']}, Target Level: {packet['target_level']}",
        f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        "\n--- Gaps Identified ---"
    ]

    if not gaps:
        report_lines.append("No gaps found. Packet meets all ladder criteria.")
    else:
        for gap in gaps:
            report_lines.append(f"[{gap.criterion_id}] {gap.description}")
            report_lines.append(f"  Missing {gap.missing_evidence} pieces of required evidence for this criterion.")

    return "\n".join(report_lines)

if __name__ == "__main__":
    # Example usage
    try:
        ladder = load_career_ladder("./career_ladder_v2.json")
        # Mock promotion packet for testing
        mock_packet: PromotionPacket = {
            "employee_id": "eng_42",
            "current_level": "L3",
            "target_level": "L4",
            "contributions": [
                {"title": "Refactored payment service", "description": "Reduced p99 latency by 40% for checkout flow"}
            ],
            "peer_reviews": [
                {"feedback": "Led the payment refactor, unblocked 3 junior engineers", "examples": "Pair programmed with L2s on service boundaries"}
            ],
            "manager_assessment": {
                "summary": "Consistently delivers high-impact projects, mentors junior staff",
                "examples": "Payment refactor saved $12k/month in infrastructure costs"
            }
        }
        gaps = validate_promotion_packet(mock_packet, ladder)
        report = generate_promotion_report(mock_packet, gaps)
        print(report)
    except Exception as e:
        print(f"Error processing promotion packet: {e}", file=sys.stderr)
        sys.exit(1)

const fs = require('fs/promises');
const path = require('path');
const { parse } = require('csv-parse/sync');
const { stringify } = require('csv-stringify/sync');

/**
 * Promotion record interface
 * @typedef {Object} PromotionRecord
 * @property {string} employee_id - Unique employee identifier
 * @property {string} fiscal_quarter - Quarter of promotion decision (e.g., 2023-Q3)
 * @property {string} current_level - Pre-promotion level (e.g., L3)
 * @property {string} target_level - Post-promotion level (e.g., L4)
 * @property {boolean} promoted - Whether promotion was approved
 * @property {string} ladder_version - Version of career ladder used for evaluation
 */

/**
 * Load promotion records from a CSV file
 * @param {string} csvPath - Absolute path to promotion CSV
 * @returns {Promise}
 */
async function loadPromotionRecords(csvPath) {
    try {
        const csvData = await fs.readFile(csvPath, 'utf-8');
        const records = parse(csvData, {
            columns: true,
            skip_empty_lines: true,
            cast: (value, context) => {
                if (context.column === 'promoted') return value.toLowerCase() === 'true';
                return value;
            }
        });
        return records;
    } catch (err) {
        if (err.code === 'ENOENT') {
            throw new Error(`Promotion CSV not found at ${csvPath}`);
        }
        throw new Error(`Failed to parse promotion CSV: ${err.message}`);
    }
}

/**
 * Calculate promotion rate by ladder version and quarter
 * @param {PromotionRecord[]} records - All promotion records
 * @returns {Object} Aggregated promotion rate data
 */
function calculatePromotionRates(records) {
    const rateMap = {};

    records.forEach(record => {
        const key = `${record.fiscal_quarter}|${record.ladder_version}`;
        if (!rateMap[key]) {
            rateMap[key] = {
                quarter: record.fiscal_quarter,
                ladder_version: record.ladder_version,
                total_candidates: 0,
                promoted_count: 0
            };
        }
        rateMap[key].total_candidates += 1;
        if (record.promoted) rateMap[key].promoted_count += 1;
    });

    // Calculate rates
    return Object.values(rateMap).map(entry => ({
        ...entry,
        promotion_rate: entry.total_candidates > 0 
            ? (entry.promoted_count / entry.total_candidates * 100).toFixed(2) 
            : 0
    })).sort((a, b) => a.quarter.localeCompare(b.quarter));
}

/**
 * Generate markdown comparison table from rate data
 * @param {Object[]} rateData - Calculated promotion rates
 * @returns {string} Markdown table string
 */
function generateComparisonTable(rateData) {
    const headers = ['Fiscal Quarter', 'Ladder Version', 'Candidates', 'Promotions', 'Promotion Rate (%)'];
    const rows = rateData.map(entry => [
        entry.quarter,
        entry.ladder_version,
        entry.total_candidates,
        entry.promoted_count,
        entry.promotion_rate
    ]);

    // Build markdown table
    let table = `| ${headers.join(' | ')} |\n`;
    table += `| ${headers.map(() => '---').join(' | ')} |\n`;
    rows.forEach(row => {
        table += `| ${row.join(' | ')} |\n`;
    });
    return table;
}

/**
 * Save report to file
 * @param {string} outputPath - Path to save markdown report
 * @param {string} tableContent - Markdown table content
 */
async function saveReport(outputPath, tableContent) {
    const report = `# Promotion Rate Comparison Report\nGenerated: ${new Date().toISOString()}\n\n${tableContent}`;
    try {
        await fs.writeFile(outputPath, report, 'utf-8');
        console.log(`Report saved to ${outputPath}`);
    } catch (err) {
        throw new Error(`Failed to save report: ${err.message}`);
    }
}

async function main() {
    const csvPath = path.join(__dirname, 'promotion_records.csv');
    const outputPath = path.join(__dirname, 'promotion_report.md');

    try {
        const records = await loadPromotionRecords(csvPath);
        console.log(`Loaded ${records.length} promotion records`);

        const rateData = calculatePromotionRates(records);
        const table = generateComparisonTable(rateData);

        console.log('Promotion Rate Comparison:');
        console.log(table);

        await saveReport(outputPath, table);
        process.exit(0);
    } catch (err) {
        console.error(`Fatal error: ${err.message}`);
        process.exit(1);
    }
}

// Run main if script is executed directly
if (require.main === module) {
    main();
}

import os
import sys
import json
from typing import List, Dict
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

# Load environment variables
SLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN")
SLACK_CHANNEL_ID = os.getenv("SLACK_PROMO_CHANNEL_ID", "C1234567890")
CAREER_LADDER_PATH = os.getenv("CAREER_LADDER_PATH", "./career_ladder_v2.json")

class SlackNotifier:
    """Send promotion gap notifications to Slack via Bolt API."""

    def __init__(self, bot_token: str, channel_id: str):
        if not bot_token:
            raise ValueError("SLACK_BOT_TOKEN environment variable is not set")
        self.client = WebClient(token=bot_token)
        self.channel_id = channel_id
        self._validate_channel()

    def _validate_channel(self) -> None:
        """Verify the Slack channel exists and bot has access."""
        try:
            self.client.conversations_info(channel=self.channel_id)
        except SlackApiError as e:
            if e.response["error"] == "channel_not_found":
                raise ValueError(f"Slack channel {self.channel_id} not found or bot lacks access")
            raise

    def send_promotion_gap_alert(self, employee_id: str, target_level: str, gaps: List[Dict]) -> None:
        """Send a formatted alert about promotion packet gaps.

        Args:
            employee_id: ID of the employee with the promotion packet
            target_level: Target promotion level
            gaps: List of gap dictionaries from validation script
        """
        if not gaps:
            return  # No gaps, no alert

        # Build Slack blocks
        blocks = [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"⚠️ Promotion Packet Gaps: {employee_id} (Target: {target_level})"
                }
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Employee ID:* {employee_id}\n*Target Level:* {target_level}\n*Gaps Found:* {len(gaps)}"
                }
            },
            {"type": "divider"}
        ]

        # Add gap details
        for gap in gaps[:5]:  # Limit to 5 gaps to avoid message size limits
            blocks.append({
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Criterion {gap['criterion_id']}*\n{gap['description']}\nMissing {gap['missing_evidence']} pieces of evidence"
                }
            })

        if len(gaps) > 5:
            blocks.append({
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"... and {len(gaps) - 5} more gaps. See full report for details."
                }
            })

        blocks.append({
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "View Full Report"},
                    "url": f"https://promo-tracker.internal/reports/{employee_id}"
                }
            ]
        })

        # Send message
        try:
            response = self.client.chat_postMessage(
                channel=self.channel_id,
                blocks=blocks,
                text=f"Promotion packet gaps for {employee_id}"  # Fallback text
            )
            print(f"Slack alert sent: {response['ts']}")
        except SlackApiError as e:
            raise RuntimeError(f"Failed to send Slack alert: {e.response['error']}") from e

def load_gaps_from_file(gap_path: str) -> List[Dict]:
    """Load gap data from a JSON file generated by the validation script."""
    try:
        with open(gap_path, 'r') as f:
            return json.load(f)
    except FileNotFoundError:
        raise FileNotFoundError(f"Gap file not found at {gap_path}")
    except json.JSONDecodeError as e:
        raise ValueError(f"Invalid JSON in gap file: {e.msg}")

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: python slack_notifier.py  ", file=sys.stderr)
        sys.exit(1)

    employee_id = sys.argv[1]
    gap_path = sys.argv[2]

    try:
        notifier = SlackNotifier(SLACK_BOT_TOKEN, SLACK_CHANNEL_ID)
        gaps = load_gaps_from_file(gap_path)

        # Extract target level from gap file (assuming it's stored)
        target_level = gaps[0].get("target_level") if gaps else "Unknown"
        notifier.send_promotion_gap_alert(employee_id, target_level, gaps)
        sys.exit(0)
    except Exception as e:
        print(f"Error sending Slack notification: {e}", file=sys.stderr)
        sys.exit(1)

curl -o career_ladder_base.json https://raw.githubusercontent.com/charity/engineering-career-ladders/main/ladders/ic-ladder-v2.json

// Greenhouse webhook handler for promotion packet submission
app.post('/greenhouse/webhook', async (req, res) => {
    const { event, payload } = req.body;
    if (event === 'promotion_packet_submitted') {
        const packet = payload.promotion_packet;
        const gaps = validatePromotionPacket(packet, careerLadder);
        if (gaps.length > 0) {
            await greenhouseClient.promotionPackets.update(packet.id, {
                custom_fields: { ladder_gaps: JSON.stringify(gaps) }
            });
        }
    }
    res.status(200).send('OK');
});

SELECT 
    criterion_id,
    COUNT(DISTINCT manager_id) as evaluating_managers,
    STDDEV(promotion_decision) as decision_variance
FROM promotion_evaluations
WHERE fiscal_quarter = '2024-Q2'
GROUP BY criterion_id
HAVING decision_variance > 0.3
ORDER BY decision_variance DESC;