// schedule-trigger/pipeline-executors.js — Pipeline automation action executors // Ticket #75053: ticket_routing, agent_health_check, dora_metrics, failure_alerting import { dbAll, dbGet, dbRun } from '../../database/connection.js'; import { apiLogger } from '../../utils/logger.js'; import { sendToTopic } from '../TelegramService.js'; import { getPipelineConfig } from '../pipeline-config.js'; const LOG_PREFIX = '[PipelineAutomation]'; // ─── Ticket Auto-Routing ──────────────────────────────────────────────────── // Trigger: row_create on Tickets table // Logic: assigns unassigned tickets to agents based on ticket type // Type → agent mapping (Space 11 defaults) const TYPE_TO_AGENT_MAP = { // bug → Developer Ralph (strongest debugger) bug: 'DEV_RALPH', // task → Developer (general) task: 'DEVELOPER', // frontend task → Frontend frontend: 'FRONTEND', // docs → Document Agent docs: 'DOCUMENT_AGENT', documentation: 'DOCUMENT_AGENT', // architecture → Architect architecture: 'ARCHITECT', spike: 'ARCHITECT', // testing → Test Runner test: 'TEST_RUNNER', testing: 'TEST_RUNNER', // widget → Widget Developer widget: 'WIDGET_DEVELOPER', }; /** * Route a ticket to the appropriate agent based on type/title keywords. * Called as a row_create trigger on the Tickets table. * * @param {Object} config - action_config from automation record * @param {Object} rowData - the ticket row data * @param {number} rowId - the ticket row ID * @returns {Promise} */ async function executeTicketRouting(config, rowData, rowId) { try { const spaceId = config.space_id || 11; const pipelineConfig = getPipelineConfig(spaceId); // Skip if already assigned if (rowData.assigned_to) { return { success: true, skipped: true, reason: 'Ticket already has assigned_to' }; } // Determine agent from type option ID or title keywords let agentKey = null; // 1. Try type-based routing via config map const typeRouting = config.type_routing || {}; const ticketType = rowData.type || rowData.ticket_type; if (ticketType && typeRouting[String(ticketType)]) { agentKey = typeRouting[String(ticketType)]; } // 2. Fallback: keyword matching on title if (!agentKey && rowData.what) { const title = String(rowData.what).toLowerCase(); for (const [keyword, agent] of Object.entries(TYPE_TO_AGENT_MAP)) { if (title.includes(keyword)) { agentKey = agent; break; } } } // 3. Default agent if (!agentKey) { agentKey = config.default_agent || 'DEV_RALPH'; } // Resolve agent key to user ID const agentUserId = pipelineConfig.AGENT_USERS[agentKey]; if (!agentUserId) { return { success: false, error: `Unknown agent key: ${agentKey}` }; } // Update the ticket with assigned_to and state → assigned const data = typeof rowData === 'string' ? JSON.parse(rowData) : { ...rowData }; data.assigned_to = agentUserId; data.state = pipelineConfig.STATE.ASSIGNED; await dbRun( 'UPDATE table_rows SET data = ?, updated_at = ? WHERE id = ?', [JSON.stringify(data), new Date().toISOString(), rowId] ); apiLogger.info( { rowId, agentKey, agentUserId }, `${LOG_PREFIX} Ticket routed to ${agentKey} (userId: ${agentUserId})` ); return { success: true, routed_to: agentKey, agent_user_id: agentUserId }; } catch (err) { return { success: false, error: err.message }; } } // ─── Agent Health Check ───────────────────────────────────────────────────── // Trigger: schedule (every 30 min) // Logic: check each agent's last activity, flag dead agents /** * Check agent health by scanning recent activity. * Agents with no activity in the threshold period are flagged as dead. * * @param {Object} config - action_config * @param {Object} contextData - schedule context * @returns {Promise} */ async function executeAgentHealthCheck(config, contextData) { try { const thresholdMs = (config.threshold_hours || 2) * 60 * 60 * 1000; const now = Date.now(); // Use agent_jobs table for reliable health check (no LIKE on JSON text) const agents = await dbAll( `SELECT DISTINCT agent_user_id, agent_name FROM agent_jobs WHERE agent_user_id IS NOT NULL GROUP BY agent_user_id, agent_name`, [] ); const deadAgents = []; const healthyAgents = []; for (const agent of agents) { // Find most recent completed/processing job const recentJob = await dbGet( `SELECT id, agent_name, status, completed_at, started_at, created_at FROM agent_jobs WHERE agent_user_id = ? ORDER BY created_at DESC LIMIT 1`, [agent.agent_user_id] ); if (!recentJob) continue; const lastTime = new Date(recentJob.completed_at || recentJob.started_at || recentJob.created_at).getTime(); const timeSinceActivity = now - lastTime; if (timeSinceActivity > thresholdMs) { deadAgents.push({ id: agent.agent_user_id, name: agent.agent_name, last_activity: recentJob.completed_at || recentJob.started_at || recentJob.created_at, hours_since: Math.round(timeSinceActivity / (60 * 60 * 1000) * 10) / 10, }); } else { healthyAgents.push({ id: agent.agent_user_id, name: agent.agent_name }); } } // Send alert if dead agents found if (deadAgents.length > 0 && config.notify !== false) { const agentList = deadAgents .map(a => ` - ${a.name}: last seen ${a.hours_since}h ago`) .join('\n'); const message = `⚠️ *Agent Health Check*\n\n` + `🔴 *${deadAgents.length} dead agent(s) detected:*\n${agentList}\n\n` + `✅ ${healthyAgents.length} agent(s) healthy\n` + `⏰ Threshold: ${config.threshold_hours || 2}h`; await sendToTopic(config.topic || 'notifications', message); } apiLogger.info( { dead: deadAgents.length, healthy: healthyAgents.length }, `${LOG_PREFIX} Agent health check complete` ); return { success: true, dead_agents: deadAgents, healthy_agents: healthyAgents.length, total_checked: agents.length, }; } catch (err) { return { success: false, error: err.message }; } } // ─── DORA Metrics ─────────────────────────────────────────────────────────── // Trigger: schedule (daily at 23:00) // Logic: calculate deployment frequency, lead time, change failure rate, MTTR /** * Calculate DORA metrics from ticket and activity data. * * @param {Object} config - action_config * @param {Object} contextData - schedule context * @returns {Promise} */ async function executeDoraMetrics(config, contextData) { try { const spaceId = config.space_id || 11; const pipelineConfig = getPipelineConfig(spaceId); const ticketsTableId = pipelineConfig.TICKETS_TABLE_ID; const activityTableId = pipelineConfig.AGENT_ACTIVITY_TABLE_ID; const periodDays = config.period_days || 1; const since = new Date(Date.now() - periodDays * 24 * 60 * 60 * 1000).toISOString(); // 1. Deployment Frequency — tickets moved to DONE in period const doneTickets = await dbAll( `SELECT id, data, updated_at FROM table_rows WHERE table_id = ? AND updated_at >= ? ORDER BY updated_at DESC`, [ticketsTableId, since] ); const completedTickets = doneTickets.filter(t => { const d = typeof t.data === 'string' ? JSON.parse(t.data) : t.data; return d.state === pipelineConfig.STATE.DONE; }); const deploymentFrequency = completedTickets.length; // 2. Lead Time — avg time from created_at to state=done let totalLeadTimeMs = 0; let leadTimeCount = 0; for (const ticket of completedTickets) { const d = typeof ticket.data === 'string' ? JSON.parse(ticket.data) : ticket.data; const createdDate = d.created_date || d.created_at; const completedDate = d.completed_date || ticket.updated_at; if (createdDate && completedDate) { totalLeadTimeMs += new Date(completedDate).getTime() - new Date(createdDate).getTime(); leadTimeCount++; } } const avgLeadTimeHours = leadTimeCount > 0 ? Math.round(totalLeadTimeMs / leadTimeCount / (60 * 60 * 1000) * 10) / 10 : 0; // 3. Change Failure Rate — from agent_jobs table (reliable) const jobStats = await dbGet( `SELECT COUNT(*) as total, COUNT(*) FILTER (WHERE status = 'failed') as failed FROM agent_jobs WHERE created_at >= ?`, [since] ); let totalActivities = Number(jobStats?.total || 0); let failedActivities = Number(jobStats?.failed || 0); const changeFailureRate = totalActivities > 0 ? Math.round(failedActivities / totalActivities * 100 * 10) / 10 : 0; // 4. MTTR — avg duration of failed jobs (started_at → completed_at) const mttrResult = await dbGet( `SELECT AVG(EXTRACT(EPOCH FROM (completed_at - started_at)) / 60) as avg_minutes FROM agent_jobs WHERE status = 'failed' AND completed_at IS NOT NULL AND started_at IS NOT NULL AND created_at >= ?`, [since] ); const mttrMinutes = Math.round((Number(mttrResult?.avg_minutes) || 0) * 10) / 10; const metrics = { period_days: periodDays, deployment_frequency: deploymentFrequency, avg_lead_time_hours: avgLeadTimeHours, change_failure_rate_pct: changeFailureRate, mttr_minutes: mttrMinutes, total_activities: totalActivities, failed_activities: failedActivities, calculated_at: new Date().toISOString(), }; // Send report if (config.notify !== false) { const message = `📊 *DORA Metrics Report* (${periodDays}d)\n\n` + `🚀 Deployment Frequency: *${deploymentFrequency}* tickets done\n` + `⏱ Avg Lead Time: *${avgLeadTimeHours}h*\n` + `💥 Change Failure Rate: *${changeFailureRate}%*\n` + `🔧 MTTR: *${mttrMinutes} min*\n\n` + `📈 Total activities: ${totalActivities} | Failed: ${failedActivities}`; await sendToTopic(config.topic || 'notifications', message); } apiLogger.info({ metrics }, `${LOG_PREFIX} DORA metrics calculated`); return { success: true, metrics }; } catch (err) { return { success: false, error: err.message }; } } // ─── Failure Alerting ─────────────────────────────────────────────────────── // Trigger: schedule (every 15 min) // Logic: check recent failure rate, alert if above threshold /** * Check recent failure rate and send alert if above threshold. * * @param {Object} config - action_config * @param {Object} contextData - schedule context * @returns {Promise} */ async function executeFailureAlerting(config, contextData) { try { const windowMinutes = config.window_minutes || 30; const thresholdPct = config.threshold_pct || 30; const since = new Date(Date.now() - windowMinutes * 60 * 1000).toISOString(); // Use agent_jobs for reliable failure tracking const recentJobs = await dbAll( `SELECT id, agent_name, agent_user_id, status, error_message FROM agent_jobs WHERE created_at >= ?`, [since] ); let total = recentJobs.length; let failures = 0; const failureDetails = []; for (const job of recentJobs) { if (job.status === 'failed') { failures++; failureDetails.push({ job_id: job.id, agent_name: job.agent_name, error: job.error_message || 'unknown', }); } } const failureRate = total > 0 ? Math.round(failures / total * 100 * 10) / 10 : 0; const isAboveThreshold = failureRate >= thresholdPct && total >= (config.min_sample || 3); if (isAboveThreshold) { // Group failures by agent name const byAgent = {}; for (const f of failureDetails) { const key = f.agent_name || 'unknown'; if (!byAgent[key]) byAgent[key] = 0; byAgent[key]++; } const agentBreakdown = Object.entries(byAgent) .map(([name, count]) => ` ${name}: ${count} failures`) .join('\n'); const message = `🚨 *Failure Rate Alert*\n\n` + `Rate: *${failureRate}%* (threshold: ${thresholdPct}%)\n` + `Window: last ${windowMinutes} min\n` + `Failed: ${failures}/${total} jobs\n\n` + `*By agent:*\n${agentBreakdown}`; await sendToTopic(config.topic || 'notifications', message); apiLogger.warn( { failureRate, failures, total, threshold: thresholdPct }, `${LOG_PREFIX} Failure rate alert triggered` ); } return { success: true, failure_rate_pct: failureRate, failures, total, alert_triggered: isAboveThreshold, }; } catch (err) { return { success: false, error: err.message }; } } export { executeTicketRouting, executeAgentHealthCheck, executeDoraMetrics, executeFailureAlerting, };