Implementation Guide: Monitor compensation benchmarks and flag roles drifting below market

az group create --name rg-comp-agent --location eastus2
az vm create --resource-group rg-comp-agent --name vm-comp-agent --image Ubuntu2204 --size Standard_B2ms --admin-username mspadmin --generate-ssh-keys --os-disk-size-gb 64 --public-ip-sku Standard

# Open port 443 outbound (default allowed) and 5678 for n8n UI (restrict to MSP IP)
az vm open-port --resource-group rg-comp-agent --name vm-comp-agent --port 5678 --priority 1010

# SSH into the VM
ssh mspadmin@<public-ip>

# Update system and install Docker
sudo apt update && sudo apt upgrade -y
curl -fsSL https://get.docker.com -o get-docker.sh && sudo sh get-docker.sh
sudo usermod -aG docker mspadmin
sudo apt install -y docker-compose-plugin

# Verify Docker
docker --version
docker compose version

# Create project directory
mkdir -p /opt/comp-agent && cd /opt/comp-agent

# Create Docker Compose file
cat > docker-compose.yml << 'EOF'
version: '3.8'

services:
  postgres:
    image: postgres:15-alpine
    container_name: comp-agent-db
    restart: unless-stopped
    environment:
      POSTGRES_USER: compagent
      POSTGRES_PASSWORD: ${DB_PASSWORD}
      POSTGRES_DB: comp_benchmarks
    volumes:
      - pgdata:/var/lib/postgresql/data
      - ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql
    ports:
      - '127.0.0.1:5432:5432'
    healthcheck:
      test: ['CMD-SHELL', 'pg_isready -U compagent']
      interval: 10s
      timeout: 5s
      retries: 5

  n8n:
    image: n8nio/n8n:latest
    container_name: comp-agent-n8n
    restart: unless-stopped
    environment:
      DB_TYPE: postgresdb
      DB_POSTGRESDB_HOST: postgres
      DB_POSTGRESDB_PORT: 5432
      DB_POSTGRESDB_DATABASE: comp_benchmarks
      DB_POSTGRESDB_USER: compagent
      DB_POSTGRESDB_PASSWORD: ${DB_PASSWORD}
      N8N_BASIC_AUTH_ACTIVE: 'true'
      N8N_BASIC_AUTH_USER: ${N8N_USER}
      N8N_BASIC_AUTH_PASSWORD: ${N8N_PASSWORD}
      N8N_HOST: ${N8N_HOST}
      N8N_PROTOCOL: https
      WEBHOOK_URL: https://${N8N_HOST}/
      GENERIC_TIMEZONE: America/New_York
      N8N_ENCRYPTION_KEY: ${ENCRYPTION_KEY}
    volumes:
      - n8ndata:/home/node/.n8n
    ports:
      - '5678:5678'
    depends_on:
      postgres:
        condition: service_healthy

volumes:
  pgdata:
  n8ndata:
EOF

# Create environment file
cat > .env << 'EOF'
DB_PASSWORD=<generate-strong-password-here>
N8N_USER=mspadmin
N8N_PASSWORD=<generate-strong-password-here>
N8N_HOST=comp-agent.yourmsp.com
ENCRYPTION_KEY=<generate-32-char-random-string>
EOF

# Secure the env file
chmod 600 .env

cat > /opt/comp-agent/init-db.sql << 'EOSQL'
-- Compensation Agent Database Schema

CREATE TABLE IF NOT EXISTS roles (
    id SERIAL PRIMARY KEY,
    external_hris_id VARCHAR(255) UNIQUE,
    job_title VARCHAR(500) NOT NULL,
    normalized_title VARCHAR(500),
    department VARCHAR(255),
    location_city VARCHAR(255),
    location_state VARCHAR(100),
    location_country VARCHAR(100) DEFAULT 'US',
    employment_type VARCHAR(50) DEFAULT 'full-time',
    experience_level VARCHAR(50),
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS compensation_snapshots (
    id SERIAL PRIMARY KEY,
    role_id INTEGER REFERENCES roles(id),
    snapshot_date DATE NOT NULL,
    current_base_salary NUMERIC(12,2),
    current_total_comp NUMERIC(12,2),
    salary_currency VARCHAR(10) DEFAULT 'USD',
    source VARCHAR(100) DEFAULT 'hris',
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS market_benchmarks (
    id SERIAL PRIMARY KEY,
    role_id INTEGER REFERENCES roles(id),
    benchmark_date DATE NOT NULL,
    market_p25 NUMERIC(12,2),
    market_p50 NUMERIC(12,2),
    market_p75 NUMERIC(12,2),
    market_p90 NUMERIC(12,2),
    data_source VARCHAR(100),
    geography VARCHAR(255),
    industry VARCHAR(255),
    sample_size INTEGER,
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS drift_analyses (
    id SERIAL PRIMARY KEY,
    role_id INTEGER REFERENCES roles(id),
    analysis_date DATE NOT NULL,
    current_salary NUMERIC(12,2),
    market_p50 NUMERIC(12,2),
    drift_percentage NUMERIC(6,2),
    drift_severity VARCHAR(20),
    target_percentile VARCHAR(10) DEFAULT 'p50',
    recommendation TEXT,
    llm_reasoning TEXT,
    alert_sent BOOLEAN DEFAULT FALSE,
    alert_sent_at TIMESTAMP,
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS agent_runs (
    id SERIAL PRIMARY KEY,
    run_date TIMESTAMP DEFAULT NOW(),
    run_type VARCHAR(50),
    roles_analyzed INTEGER,
    alerts_generated INTEGER,
    errors TEXT,
    duration_seconds INTEGER,
    status VARCHAR(50),
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS role_mappings (
    id SERIAL PRIMARY KEY,
    internal_title VARCHAR(500),
    benchmark_title VARCHAR(500),
    benchmark_code VARCHAR(100),
    confidence_score NUMERIC(4,2),
    mapping_method VARCHAR(50),
    verified_by_hr BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_drift_date ON drift_analyses(analysis_date);
CREATE INDEX idx_drift_severity ON drift_analyses(drift_severity);
CREATE INDEX idx_comp_snapshot_date ON compensation_snapshots(snapshot_date);
CREATE INDEX idx_benchmark_date ON market_benchmarks(benchmark_date);
EOSQL

cd /opt/comp-agent && docker compose up -d

docker compose ps

docker exec -it comp-agent-db psql -U compagent -d comp_benchmarks -c '\dt'

curl -X GET 'https://api.unified.to/hris/employees?connection_id=YOUR_CONNECTION_ID&fields=compensations,title,department,location' \
  -H 'Authorization: Bearer YOUR_UNIFIED_API_KEY' \
  -H 'Content-Type: application/json' | python3 -m json.tool

{
  "results": [
    {
      "id": "emp_123",
      "name": "Jane Doe",
      "title": "Software Engineer II",
      "department": "Engineering",
      "location": { "city": "Austin", "state": "TX", "country": "US" },
      "compensations": [
        { "type": "salary", "amount": 125000, "currency": "USD", "frequency": "annual" }
      ]
    }
  ]
}

curl -X GET 'https://api.salary.com/v1/companalyst/market-data?job_title=Software+Engineer+II&location=Austin,TX&industry=Technology' \
  -H 'Authorization: Bearer YOUR_COMPANALYST_API_KEY' \
  -H 'Accept: application/json' | python3 -m json.tool

curl https://api.openai.com/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -H 'Authorization: Bearer YOUR_OPENAI_API_KEY' \
  -d '{
    "model": "gpt-5.4-mini",
    "messages": [{
      "role": "system",
      "content": "You are a compensation analyst. Given a job title, normalize it to a standard market title and suggest a SOC code."
    }, {
      "role": "user",
      "content": "Normalize this job title: Sr. Full Stack Dev (Backend Heavy) in Austin, TX"
    }],
    "temperature": 0.2
  }'

curl -X POST 'YOUR_TEAMS_WEBHOOK_URL' \
  -H 'Content-Type: application/json' \
  -d '{
    "@type": "MessageCard",
    "summary": "Test Alert",
    "themeColor": "FF0000",
    "title": "⚠️ Compensation Drift Alert — TEST",
    "sections": [{
      "facts": [
        { "name": "Role", "value": "Software Engineer II" },
        { "name": "Current Salary", "value": "$105,000" },
        { "name": "Market P50", "value": "$125,000" },
        { "name": "Drift", "value": "-16.0%" }
      ],
      "text": "This role is 16% below the 50th percentile market rate for Austin, TX."
    }]
  }'

# Install Nginx and Certbot
sudo apt install -y nginx certbot python3-certbot-nginx

# Create Nginx config
sudo cat > /etc/nginx/sites-available/comp-agent << 'EOF'
server {
    listen 80;
    server_name comp-agent.yourmsp.com;

    location / {
        proxy_pass http://127.0.0.1:5678;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection 'upgrade';
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_read_timeout 300s;
        chunked_transfer_encoding off;
        proxy_buffering off;
    }
}
EOF

# Enable the site
sudo ln -s /etc/nginx/sites-available/comp-agent /etc/nginx/sites-enabled/
sudo rm -f /etc/nginx/sites-enabled/default
sudo nginx -t && sudo systemctl reload nginx

# Get SSL certificate
sudo certbot --nginx -d comp-agent.yourmsp.com --non-interactive --agree-tos -m admin@yourmsp.com

# Verify auto-renewal
sudo certbot renew --dry-run

# Restrict n8n to only accept connections from localhost (via Nginx)
# Edit docker-compose.yml: change n8n ports from '5678:5678' to '127.0.0.1:5678:5678'
cd /opt/comp-agent
sed -i "s/'5678:5678'/'127.0.0.1:5678:5678'/" docker-compose.yml
docker compose up -d

# Create backup script
sudo cat > /opt/comp-agent/backup.sh << 'EOF'
#!/bin/bash
set -e
BACKUP_DIR=/opt/comp-agent/backups
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
RETENTION_DAYS=90

mkdir -p $BACKUP_DIR

# Dump PostgreSQL
docker exec comp-agent-db pg_dump -U compagent -Fc comp_benchmarks > $BACKUP_DIR/comp_benchmarks_$TIMESTAMP.dump

# Compress
gzip $BACKUP_DIR/comp_benchmarks_$TIMESTAMP.dump

# Upload to Azure Blob Storage (or AWS S3)
# az storage blob upload --account-name compagentbackups --container-name backups \
#   --file $BACKUP_DIR/comp_benchmarks_$TIMESTAMP.dump.gz \
#   --name $TIMESTAMP/comp_benchmarks.dump.gz

# Clean up local backups older than retention
find $BACKUP_DIR -name '*.dump.gz' -mtime +$RETENTION_DAYS -delete

echo "Backup completed: comp_benchmarks_$TIMESTAMP.dump.gz"
EOF

chmod +x /opt/comp-agent/backup.sh

# Schedule daily backup at 2 AM
(crontab -l 2>/dev/null; echo '0 2 * * * /opt/comp-agent/backup.sh >> /opt/comp-agent/backups/backup.log 2>&1') | crontab -

# Test backup
/opt/comp-agent/backup.sh

docker exec -it comp-agent-db psql -U compagent -d comp_benchmarks -c "
CREATE USER retool_readonly WITH PASSWORD 'strong-password-here';
GRANT CONNECT ON DATABASE comp_benchmarks TO retool_readonly;
GRANT USAGE ON SCHEMA public TO retool_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO retool_readonly;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO retool_readonly;
"

pip install streamlit psycopg2-binary plotly pandas

// Place in Node 3 of the 'Comp Agent — HRIS Data Pull' workflow

// n8n Code Node — JavaScript
// Place this in Node 3 of the 'Comp Agent — HRIS Data Pull' workflow

const employees = $input.all();
const today = new Date().toISOString().split('T')[0];
const results = [];

for (const item of employees) {
  const emp = item.json;
  
  // Skip employees without compensation data
  if (!emp.compensations || emp.compensations.length === 0) continue;
  
  // Get the primary compensation (most recent or highest)
  const primaryComp = emp.compensations
    .filter(c => c.type === 'salary' || c.type === 'base')
    .sort((a, b) => (b.amount || 0) - (a.amount || 0))[0];
  
  if (!primaryComp || !primaryComp.amount) continue;
  
  // Normalize to annual salary
  let annualSalary = primaryComp.amount;
  const freq = (primaryComp.frequency || 'annual').toLowerCase();
  if (freq === 'hourly') annualSalary = primaryComp.amount * 2080;
  else if (freq === 'monthly') annualSalary = primaryComp.amount * 12;
  else if (freq === 'biweekly') annualSalary = primaryComp.amount * 26;
  else if (freq === 'weekly') annualSalary = primaryComp.amount * 52;
  
  // Extract location
  const location = emp.location || {};
  const city = location.city || 'Unknown';
  const state = location.state || location.region || 'Unknown';
  const country = location.country || 'US';
  
  // Determine experience level from title heuristics
  const title = (emp.title || '').toLowerCase();
  let experienceLevel = 'mid';
  if (title.includes('junior') || title.includes('jr') || title.includes('entry') || title.includes('associate')) experienceLevel = 'junior';
  else if (title.includes('senior') || title.includes('sr') || title.includes('lead') || title.includes('principal') || title.includes('staff')) experienceLevel = 'senior';
  else if (title.includes('director') || title.includes('vp') || title.includes('chief') || title.includes('head of')) experienceLevel = 'executive';
  else if (title.includes('manager') || title.includes('mgr')) experienceLevel = 'manager';
  
  results.push({
    json: {
      external_hris_id: emp.id,
      job_title: emp.title || 'Unknown Title',
      department: emp.department || 'Unknown',
      location_city: city,
      location_state: state,
      location_country: country,
      employment_type: emp.employment_type || 'full-time',
      experience_level: experienceLevel,
      snapshot_date: today,
      current_base_salary: Math.round(annualSalary * 100) / 100,
      salary_currency: primaryComp.currency || 'USD'
    }
  });
}

return results;

// Example 1
// Input: Internal Title: Sr. Full Stack Dev (Backend Heavy), Department: Engineering, Location: Austin, TX, Experience Level: senior
{
  "normalized_title": "Senior Software Engineer",
  "soc_code": "15-1252",
  "confidence_score": 0.92,
  "reasoning": "Full Stack Dev with backend emphasis maps to Senior Software Engineer. The 'Sr.' prefix confirms senior level. SOC 15-1252 covers Software Developers and Programmers.",
  "needs_human_review": false,
  "alternative_titles": ["Senior Backend Developer", "Senior Full Stack Developer"]
}

// Example 2
// Input: Internal Title: People Ops Ninja, Department: Human Resources, Location: San Francisco, CA, Experience Level: mid
{
  "normalized_title": "Human Resources Generalist",
  "soc_code": "13-1071",
  "confidence_score": 0.72,
  "reasoning": "'People Ops' maps to HR operations. 'Ninja' is a non-standard modifier that doesn't indicate seniority. Most likely an HR Generalist role. Low confidence due to ambiguous title.",
  "needs_human_review": true,
  "alternative_titles": ["HR Coordinator", "HR Business Partner", "People Operations Specialist"]
}

TRIGGER: Cron schedule — 0 8 * * 1 (Monday 8 AM)

SELECT 
  r.id AS role_id,
  r.job_title,
  r.normalized_title,
  r.department,
  r.location_city,
  r.location_state,
  r.experience_level,
  cs.current_base_salary,
  cs.snapshot_date,
  mb.market_p25,
  mb.market_p50,
  mb.market_p75,
  mb.market_p90,
  mb.data_source AS benchmark_source,
  mb.sample_size,
  -- Previous drift for trend
  prev.drift_percentage AS previous_drift
FROM roles r
INNER JOIN compensation_snapshots cs ON r.id = cs.role_id
  AND cs.snapshot_date = (SELECT MAX(snapshot_date) FROM compensation_snapshots WHERE role_id = r.id)
INNER JOIN market_benchmarks mb ON r.id = mb.role_id
  AND mb.benchmark_date = (SELECT MAX(benchmark_date) FROM market_benchmarks WHERE role_id = r.id)
INNER JOIN role_mappings rm ON r.job_title = rm.internal_title
  AND (rm.verified_by_hr = true OR rm.confidence_score >= 0.85)
LEFT JOIN LATERAL (
  SELECT drift_percentage 
  FROM drift_analyses 
  WHERE role_id = r.id 
  ORDER BY analysis_date DESC 
  LIMIT 1
) prev ON true
WHERE cs.current_base_salary IS NOT NULL 
  AND mb.market_p50 IS NOT NULL
  AND mb.market_p50 > 0;

// CODE: Calculate Drift Metrics (configure node with 2 outputs)

const roles = $input.all();
const today = new Date().toISOString().split('T')[0];
const flaggedRoles = [];
const allRoles = [];

for (const item of roles) {
  const r = item.json;
  const salary = parseFloat(r.current_base_salary);
  const p50 = parseFloat(r.market_p50);
  
  if (!salary || !p50 || p50 === 0) continue;
  
  const driftPct = ((salary - p50) / p50) * 100;
  const roundedDrift = Math.round(driftPct * 100) / 100;
  
  let severity = 'OK';
  if (roundedDrift <= -15) severity = 'CRITICAL';
  else if (roundedDrift <= -10) severity = 'WARNING';
  else if (roundedDrift <= -5) severity = 'WATCH';
  
  // Detect worsening trend
  const prevDrift = r.previous_drift ? parseFloat(r.previous_drift) : null;
  const trendDirection = prevDrift !== null ? 
    (roundedDrift < prevDrift ? 'worsening' : roundedDrift > prevDrift ? 'improving' : 'stable') : 'new';
  
  const roleData = {
    ...r,
    drift_percentage: roundedDrift,
    drift_severity: severity,
    trend_direction: trendDirection,
    analysis_date: today,
    salary_vs_p25: Math.round(((salary - parseFloat(r.market_p25)) / parseFloat(r.market_p25)) * 10000) / 100,
    salary_vs_p75: Math.round(((salary - parseFloat(r.market_p75)) / parseFloat(r.market_p75)) * 10000) / 100
  };
  
  allRoles.push({ json: roleData });
  
  if (severity === 'CRITICAL' || severity === 'WARNING') {
    flaggedRoles.push({ json: roleData });
  }
}

// Output 0: All roles (for logging)
// Output 1: Flagged roles only (for LLM analysis + alerts)
return [allRoles, flaggedRoles];

const items = $input.all();
const results = [];

for (const item of items) {
  let llmResponse;
  try {
    const content = item.json.message?.content || item.json.text || JSON.stringify(item.json);
    llmResponse = JSON.parse(content);
  } catch (e) {
    llmResponse = { 
      risk_level: 'UNKNOWN', 
      analysis: item.json.message?.content || 'LLM response parsing failed',
      recommendation: 'Manual review required',
      urgency: 'Next Quarter'
    };
  }
  results.push({ json: { ...item.json, llm_analysis: llmResponse } });
}
return results;

EMAIL SUBJECT: ⚠️ Weekly Compensation Drift Report — {{analysis_date}} | {{total_flagged}} Roles Below Market

<!DOCTYPE html>
<html>
<head><style>
  body { font-family: 'Segoe UI', Arial, sans-serif; color: #333; max-width: 800px; margin: 0 auto; }
  .header { background: #1a365d; color: white; padding: 20px; border-radius: 8px 8px 0 0; }
  .header h1 { margin: 0; font-size: 20px; }
  .header p { margin: 5px 0 0; opacity: 0.8; font-size: 14px; }
  .summary-cards { display: flex; gap: 12px; padding: 16px; background: #f7fafc; }
  .card { flex: 1; background: white; border-radius: 8px; padding: 16px; text-align: center; box-shadow: 0 1px 3px rgba(0,0,0,0.1); }
  .card .number { font-size: 32px; font-weight: bold; }
  .card .label { font-size: 12px; color: #718096; text-transform: uppercase; }
  .critical .number { color: #e53e3e; }
  .warning .number { color: #dd6b20; }
  .watch .number { color: #d69e2e; }
  table { width: 100%; border-collapse: collapse; margin: 16px 0; }
  th { background: #edf2f7; padding: 10px 12px; text-align: left; font-size: 12px; text-transform: uppercase; color: #4a5568; }
  td { padding: 10px 12px; border-bottom: 1px solid #e2e8f0; font-size: 14px; }
  .severity-critical { background: #fff5f5; color: #c53030; font-weight: bold; }
  .severity-warning { background: #fffaf0; color: #c05621; font-weight: bold; }
  .trend-worsening { color: #e53e3e; }
  .trend-improving { color: #38a169; }
  .trend-stable { color: #718096; }
  .recommendation { background: #ebf8ff; border-left: 4px solid #3182ce; padding: 12px; margin: 8px 0; border-radius: 0 4px 4px 0; font-size: 13px; }
  .footer { padding: 16px; font-size: 12px; color: #a0aec0; border-top: 1px solid #e2e8f0; }
</style></head>
<body>
  <div class="header">
    <h1>📊 Compensation Drift Report</h1>
    <p>Week of {{analysis_date}} | Generated by Compensation Monitor Agent</p>
  </div>
  
  <div class="summary-cards">
    <div class="card critical"><div class="number">{{critical_count}}</div><div class="label">Critical (≤-15%)</div></div>
    <div class="card warning"><div class="number">{{warning_count}}</div><div class="label">Warning (-10 to -15%)</div></div>
    <div class="card watch"><div class="number">{{watch_count}}</div><div class="label">Watch (-5 to -10%)</div></div>
  </div>
  
  <table>
    <thead><tr>
      <th>Role</th><th>Department</th><th>Location</th><th>Current Salary</th><th>Market P50</th><th>Drift</th><th>Trend</th><th>Urgency</th>
    </tr></thead>
    <tbody>
      {{#each flagged_roles}}
      <tr class="severity-{{lowercase drift_severity}}">
        <td><strong>{{job_title}}</strong><br><small>{{normalized_title}}</small></td>
        <td>{{department}}</td>
        <td>{{location_city}}, {{location_state}}</td>
        <td>${{formatNumber current_base_salary}}</td>
        <td>${{formatNumber market_p50}}</td>
        <td><strong>{{drift_percentage}}%</strong></td>
        <td class="trend-{{trend_direction}}">{{trend_icon}} {{trend_direction}}</td>
        <td>{{llm_analysis.urgency}}</td>
      </tr>
      <tr><td colspan="8"><div class="recommendation">💡 {{llm_analysis.recommendation}}</div></td></tr>
      {{/each}}
    </tbody>
  </table>
  
  <div class="footer">
    <p>This report was generated automatically by the Compensation Monitoring Agent. Data sources: {{benchmark_source}} (market data), {{hris_source}} (employee data). For questions, contact your MSP account manager.</p>
    <p>⚙️ Alert thresholds: Critical ≤ -15% | Warning -10% to -15% | Watch -5% to -10% | Target: {{target_percentile}} percentile</p>
  </div>
</body>
</html>

# send via HTTP POST to Teams webhook URL

{
  "@type": "MessageCard",
  "@context": "http://schema.org/extensions",
  "themeColor": "E53E3E",
  "summary": "🚨 Critical Compensation Drift: {{job_title}}",
  "sections": [
    {
      "activityTitle": "🚨 CRITICAL Compensation Drift Detected",
      "activitySubtitle": "{{job_title}} — {{department}}",
      "activityImage": "",
      "facts": [
        { "name": "Role", "value": "{{job_title}} ({{normalized_title}})" },
        { "name": "Location", "value": "{{location_city}}, {{location_state}}" },
        { "name": "Current Salary", "value": "${{current_base_salary}}" },
        { "name": "Market P50", "value": "${{market_p50}}" },
        { "name": "Market P75", "value": "${{market_p75}}" },
        { "name": "Drift", "value": "**{{drift_percentage}}%** below market median" },
        { "name": "Trend", "value": "{{trend_direction}} (prev: {{previous_drift}}%)" },
        { "name": "Risk Level", "value": "{{llm_analysis.risk_level}}" },
        { "name": "Retention Risk", "value": "{{llm_analysis.retention_risk}}" }
      ],
      "markdown": true
    },
    {
      "title": "💡 AI Recommendation",
      "text": "{{llm_analysis.recommendation}}\n\n**Suggested Range:** ${{llm_analysis.suggested_salary_range.min}} — ${{llm_analysis.suggested_salary_range.max}}\n\n**Urgency:** {{llm_analysis.urgency}}"
    }
  ],
  "potentialAction": [
    {
      "@type": "OpenUri",
      "name": "View Full Report",
      "targets": [{ "os": "default", "uri": "https://comp-dashboard.yourmsp.com/drift/{{role_id}}" }]
    }
  ]
}

-- Used by: Comp Agent — Drift Analyzer (Node 2)

-- Core Drift Comparison Query
-- Used by: Comp Agent — Drift Analyzer (Node 2)
-- Performance: Executes in <100ms for up to 1,000 roles with proper indexing

WITH latest_comp AS (
    SELECT DISTINCT ON (role_id)
        role_id,
        current_base_salary,
        snapshot_date
    FROM compensation_snapshots
    ORDER BY role_id, snapshot_date DESC
),
latest_benchmark AS (
    SELECT DISTINCT ON (role_id)
        role_id,
        market_p25,
        market_p50,
        market_p75,
        market_p90,
        data_source,
        geography,
        sample_size,
        benchmark_date
    FROM market_benchmarks
    ORDER BY role_id, benchmark_date DESC
),
previous_drift AS (
    SELECT DISTINCT ON (role_id)
        role_id,
        drift_percentage,
        drift_severity,
        analysis_date AS previous_analysis_date
    FROM drift_analyses
    ORDER BY role_id, analysis_date DESC
)
SELECT
    r.id AS role_id,
    r.job_title,
    COALESCE(rm.benchmark_title, r.normalized_title, r.job_title) AS normalized_title,
    r.department,
    r.location_city,
    r.location_state,
    r.experience_level,
    lc.current_base_salary,
    lc.snapshot_date,
    lb.market_p25,
    lb.market_p50,
    lb.market_p75,
    lb.market_p90,
    lb.data_source AS benchmark_source,
    lb.sample_size,
    lb.benchmark_date,
    pd.drift_percentage AS previous_drift,
    pd.drift_severity AS previous_severity,
    pd.previous_analysis_date,
    -- Staleness check: flag if benchmark data is older than 90 days
    CASE WHEN lb.benchmark_date < CURRENT_DATE - INTERVAL '90 days'
         THEN true ELSE false END AS stale_benchmark
FROM roles r
INNER JOIN latest_comp lc ON r.id = lc.role_id
INNER JOIN latest_benchmark lb ON r.id = lb.role_id
INNER JOIN role_mappings rm ON r.job_title = rm.internal_title
    AND (rm.verified_by_hr = true OR rm.confidence_score >= 0.85)
LEFT JOIN previous_drift pd ON r.id = pd.role_id
WHERE lc.current_base_salary > 0
    AND lb.market_p50 > 0
ORDER BY r.department, r.job_title;

CREATE INDEX IF NOT EXISTS idx_comp_role_date ON compensation_snapshots(role_id, snapshot_date DESC);
CREATE INDEX IF NOT EXISTS idx_bench_role_date ON market_benchmarks(role_id, benchmark_date DESC);
CREATE INDEX IF NOT EXISTS idx_drift_role_date ON drift_analyses(role_id, analysis_date DESC);
CREATE INDEX IF NOT EXISTS idx_mapping_title ON role_mappings(internal_title) WHERE verified_by_hr = true OR confidence_score >= 0.85;

# Bulk job title normalization for initial setup

#!/usr/bin/env python3
"""initial_role_mapping.py — Bulk job title normalization for initial setup.

Usage: python3 initial_role_mapping.py --db-url postgresql://compagent:PASSWORD@localhost:5432/comp_benchmarks --openai-key sk-...

This script:
1. Fetches all unmapped roles from PostgreSQL
2. Sends each title through GPT-5.4 mini for normalization
3. Stores mappings in role_mappings table
4. Exports a CSV for HR review
"""

import argparse
import csv
import json
import time
import sys
from datetime import datetime

import psycopg2
import psycopg2.extras
import requests

SYSTEM_PROMPT = """You are an expert compensation analyst specializing in job title normalization.
Map the given internal job title to a standardized market benchmark title.
Consider the department and location context.

Respond ONLY with valid JSON:
{
  "normalized_title": "Standard Market Title",
  "soc_code": "XX-XXXX",
  "confidence_score": 0.95,
  "reasoning": "Brief explanation",
  "needs_human_review": false,
  "alternative_titles": ["Alt 1", "Alt 2"]
}"""

def normalize_title(title: str, department: str, location: str, api_key: str) -> dict:
    """Call GPT-5.4 mini to normalize a single job title."""
    user_msg = f"""Normalize this internal job title to a standard market benchmark title:

Internal Title: {title}
Department: {department}
Location: {location}"""
    
    resp = requests.post(
        "https://api.openai.com/v1/chat/completions",
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        },
        json={
            "model": "gpt-5.4-mini",
            "messages": [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": user_msg}
            ],
            "temperature": 0.1,
            "max_tokens": 300,
            "response_format": {"type": "json_object"}
        },
        timeout=30
    )
    resp.raise_for_status()
    content = resp.json()["choices"][0]["message"]["content"]
    return json.loads(content)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--db-url", required=True)
    parser.add_argument("--openai-key", required=True)
    parser.add_argument("--output-csv", default=f"role_mappings_review_{datetime.now().strftime('%Y%m%d')}.csv")
    parser.add_argument("--batch-size", type=int, default=5)
    parser.add_argument("--delay", type=float, default=1.0, help="Seconds between API calls")
    args = parser.parse_args()

    conn = psycopg2.connect(args.db_url)
    cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)

    # Fetch unmapped roles
    cur.execute("""
        SELECT r.id, r.job_title, r.department, r.location_city, r.location_state
        FROM roles r
        LEFT JOIN role_mappings rm ON r.job_title = rm.internal_title
        WHERE rm.id IS NULL
        ORDER BY r.department, r.job_title
    """)
    roles = cur.fetchall()
    print(f"Found {len(roles)} unmapped roles")

    results = []
    for i, role in enumerate(roles):
        location = f"{role['location_city']}, {role['location_state']}"
        print(f"[{i+1}/{len(roles)}] Normalizing: {role['job_title']} ({role['department']})")
        
        try:
            mapping = normalize_title(
                role['job_title'], role['department'], location, args.openai_key
            )
            
            # Insert into database
            cur.execute("""
                INSERT INTO role_mappings 
                    (internal_title, benchmark_title, benchmark_code, confidence_score, 
                     mapping_method, verified_by_hr)
                VALUES (%s, %s, %s, %s, %s, %s)
                ON CONFLICT (internal_title) DO UPDATE SET
                    benchmark_title = EXCLUDED.benchmark_title,
                    benchmark_code = EXCLUDED.benchmark_code,
                    confidence_score = EXCLUDED.confidence_score,
                    updated_at = NOW()
            """, (
                role['job_title'],
                mapping['normalized_title'],
                mapping.get('soc_code', ''),
                mapping.get('confidence_score', 0.5),
                'gpt-5.4-mini-initial',
                False  # Always false until HR reviews
            ))
            
            results.append({
                'internal_title': role['job_title'],
                'department': role['department'],
                'location': location,
                'normalized_title': mapping['normalized_title'],
                'soc_code': mapping.get('soc_code', ''),
                'confidence': mapping.get('confidence_score', 0),
                'needs_review': mapping.get('needs_human_review', True),
                'reasoning': mapping.get('reasoning', ''),
                'alternatives': ', '.join(mapping.get('alternative_titles', []))
            })
            print(f"  -> {mapping['normalized_title']} (confidence: {mapping.get('confidence_score', '?')})")
            
        except Exception as e:
            print(f"  ERROR: {e}")
            results.append({
                'internal_title': role['job_title'],
                'department': role['department'],
                'location': location,
                'normalized_title': 'ERROR — MANUAL MAPPING REQUIRED',
                'soc_code': '',
                'confidence': 0,
                'needs_review': True,
                'reasoning': str(e),
                'alternatives': ''
            })
        
        time.sleep(args.delay)  # Rate limiting
    
    conn.commit()
    
    # Export CSV for HR review
    with open(args.output_csv, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=[
            'internal_title', 'department', 'location', 'normalized_title',
            'soc_code', 'confidence', 'needs_review', 'reasoning', 'alternatives'
        ])
        writer.writeheader()
        writer.writerows(results)
    
    print(f"\nCompleted! {len(results)} mappings processed.")
    print(f"Review CSV exported to: {args.output_csv}")
    print(f"High confidence (>=0.85): {sum(1 for r in results if r['confidence'] >= 0.85)}")
    print(f"Needs HR review: {sum(1 for r in results if r['needs_review'])}")
    
    cur.close()
    conn.close()


if __name__ == '__main__':
    main()

pip install psycopg2-binary requests

python3 initial_role_mapping.py \
  --db-url 'postgresql://compagent:PASSWORD@localhost:5432/comp_benchmarks' \
  --openai-key 'sk-...' \
  --output-csv role_mappings_review.csv

psql -U compagent -d comp_benchmarks -c \
  "UPDATE role_mappings SET verified_by_hr = true WHERE confidence_score >= 0.85;"