f89e074dd1
Governed substrate for autonomous agents: scoped identity (passports), audited actions, MCP workspace. Infra IPs and secrets redacted for public release.
636 lines
28 KiB
JavaScript
636 lines
28 KiB
JavaScript
/**
|
|
* AgentJobService — Lifecycle Module
|
|
*
|
|
* cancelJob, recoverStuckJobs, setShuttingDown, isShuttingDown,
|
|
* and internal helpers for startup recovery.
|
|
*/
|
|
|
|
import { dbGet, dbRun, dbAll, isPostgres } from '../../database/connection.js';
|
|
import { apiLogger } from '../../utils/logger.js';
|
|
import { saveStepMessage } from '../AgentLoopService.js';
|
|
import { setConversationProcessing } from '../chat/agent-execution-shared.js';
|
|
import { resolveAgentUser } from '../agent-users.js';
|
|
import { getJob, getStalledJobs } from './query.js';
|
|
import { createAndDispatchJob } from './create.js';
|
|
import {
|
|
updateJobStatus,
|
|
failJob,
|
|
safeParse,
|
|
JOB_STATUS,
|
|
TICKETS_TABLE_ID,
|
|
TICKET_STATE,
|
|
MAX_RESTART_RETRIES,
|
|
MAX_JOB_RESTART_RETRIES,
|
|
MAX_RECOVERY_REDISPATCH,
|
|
} from './shared.js';
|
|
|
|
// ─── PID LIVENESS CHECK ─────────────────────────────────────────
|
|
/**
|
|
* Check if a process with given PID is still alive.
|
|
* Uses kill(pid, 0) which checks existence without sending a signal.
|
|
*/
|
|
function _isPidAlive(pid) {
|
|
try {
|
|
process.kill(pid, 0); // Signal 0 = check existence only
|
|
return true;
|
|
} catch {
|
|
return false; // ESRCH = no such process
|
|
}
|
|
}
|
|
|
|
// ─── SHUTDOWN STATE ──────────────────────────────────────────────
|
|
// FIX-A: Race condition guard. During graceful shutdown, server.js marks jobs
|
|
// as failed with a recovery marker. Without this flag, processJobLocally()'s
|
|
// catch block runs concurrently and overwrites the marker with the real error,
|
|
// causing recoverStuckJobs() Phase 0 to miss these jobs on restart.
|
|
let _isShuttingDown = false;
|
|
|
|
/**
|
|
* Signal that graceful shutdown has begun.
|
|
* Called from server.js BEFORE killing child processes / marking jobs.
|
|
* Once set, processJobLocally() catch blocks will NOT overwrite error_message.
|
|
*/
|
|
export function setShuttingDown() {
|
|
_isShuttingDown = true;
|
|
}
|
|
|
|
/**
|
|
* Check if shutdown is in progress (for external callers).
|
|
*/
|
|
export function isShuttingDown() {
|
|
return _isShuttingDown;
|
|
}
|
|
|
|
/**
|
|
* Cancel a job
|
|
*/
|
|
export async function cancelJob(jobId) {
|
|
const job = await getJob(jobId);
|
|
if (!job) return { success: false, error: 'Job not found' };
|
|
|
|
// Kill the worker process tree (including MCP children) if running
|
|
if (job.worker_pid) {
|
|
try {
|
|
// Kill entire process group (negative PID) to catch MCP sub-processes
|
|
process.kill(-job.worker_pid, 'SIGTERM');
|
|
} catch {
|
|
// Process group may not exist — try direct kill
|
|
try { process.kill(job.worker_pid, 'SIGTERM'); } catch { /* already dead */ }
|
|
}
|
|
// Schedule SIGKILL for stubborn processes
|
|
setTimeout(() => {
|
|
try { process.kill(-job.worker_pid, 'SIGKILL'); } catch { /* already dead */ }
|
|
}, 5000).unref();
|
|
}
|
|
|
|
await updateJobStatus(jobId, JOB_STATUS.CANCELLED);
|
|
await setConversationProcessing(job.conversation_id, false);
|
|
|
|
return { success: true };
|
|
}
|
|
|
|
// ─── SENTINEL CONSUMPTION (T-146479) ────────────────────────────
|
|
|
|
/**
|
|
* Mark a graceful-shutdown sentinel row as "consumed" so Phase 0 of
|
|
* recoverStuckJobs will NOT pick it up on subsequent restarts.
|
|
*
|
|
* Mutates BOTH error_message AND result_metadata to break the OR-joined
|
|
* WHERE clause in Phase 0 (see `gracefulJobs` query). Without this, every
|
|
* restart re-finds the same rows and re-dispatches / re-spams the chats.
|
|
*/
|
|
async function _consumeSentinelMarker(jobId, reason) {
|
|
// Replace result_metadata wholesale with a fresh marker — strips the
|
|
// shutdown_recovery=true flag and any stale fields. We don't try to
|
|
// preserve the original shutdown_at; it's redundant with completed_at.
|
|
const consumedMetadata = JSON.stringify({
|
|
shutdown_recovery: false,
|
|
sentinel_consumed_at: new Date().toISOString(),
|
|
sentinel_consume_reason: reason,
|
|
});
|
|
await dbRun(
|
|
isPostgres()
|
|
? `UPDATE agent_jobs SET error_message = $1, result_metadata = $2 WHERE id = $3`
|
|
: `UPDATE agent_jobs SET error_message = ?, result_metadata = ? WHERE id = ?`,
|
|
[`Auto-recovery consumed (${reason})`, consumedMetadata, jobId]
|
|
);
|
|
}
|
|
|
|
// ─── STARTUP RECOVERY ───────────────────────────────────────────
|
|
|
|
/**
|
|
* Reset a ticket back to BACKLOG so AgentWorkerService re-picks it up.
|
|
* Called when the associated agent_job was killed by a server restart.
|
|
*
|
|
* @param {number} conversationId - Conversation bound to the ticket
|
|
* @param {string} reason - Why the ticket is being reset
|
|
* @returns {Promise<{reset: boolean, ticketId?: number}>}
|
|
*/
|
|
async function resetBoundTicketToBacklog(conversationId, reason) {
|
|
try {
|
|
// Find the conversation's bound ticket
|
|
const conv = await dbGet(
|
|
isPostgres()
|
|
? `SELECT type, bound_table_id, bound_row_id FROM conversations WHERE id = $1`
|
|
: `SELECT type, bound_table_id, bound_row_id FROM conversations WHERE id = ?`,
|
|
[conversationId]
|
|
);
|
|
|
|
// Only reset ticket_chat conversations (created by AgentWorkerService)
|
|
if (conv?.type !== 'ticket_chat' || !conv.bound_row_id) {
|
|
return { reset: false };
|
|
}
|
|
|
|
const ticketRow = await dbGet(
|
|
isPostgres()
|
|
? `SELECT id, data FROM table_rows WHERE id = $1 AND table_id = $2`
|
|
: `SELECT id, data FROM table_rows WHERE id = ? AND table_id = ?`,
|
|
[conv.bound_row_id, TICKETS_TABLE_ID]
|
|
);
|
|
|
|
if (!ticketRow) return { reset: false };
|
|
|
|
const ticketData = safeParse(ticketRow.data, {});
|
|
const currentState = parseInt(ticketData.state, 10);
|
|
|
|
// Only reset if ticket is in REVIEW or IN_PROGRESS (not already done/backlog)
|
|
if (currentState !== TICKET_STATE.REVIEW && currentState !== TICKET_STATE.IN_PROGRESS) {
|
|
return { reset: false };
|
|
}
|
|
|
|
// Check restart count — don't retry forever
|
|
const restartCount = (ticketData._restart_count || 0) + 1;
|
|
if (restartCount > MAX_RESTART_RETRIES) {
|
|
apiLogger.warn({
|
|
ticket_id: ticketRow.id,
|
|
restartCount,
|
|
}, 'AgentJobService: Ticket exceeded max restart retries — leaving in current state');
|
|
return { reset: false };
|
|
}
|
|
|
|
// Reset ticket to BACKLOG with restart metadata
|
|
const updatedData = {
|
|
...ticketData,
|
|
state: TICKET_STATE.BACKLOG,
|
|
_restart_count: restartCount,
|
|
_last_restart: new Date().toISOString(),
|
|
_restart_reason: reason,
|
|
};
|
|
|
|
await dbRun(
|
|
isPostgres()
|
|
? `UPDATE table_rows SET data = $1, updated_at = NOW() WHERE id = $2`
|
|
: `UPDATE table_rows SET data = ?, updated_at = datetime('now') WHERE id = ?`,
|
|
[JSON.stringify(updatedData), ticketRow.id]
|
|
);
|
|
|
|
apiLogger.info({
|
|
ticket_id: ticketRow.id,
|
|
restart_count: restartCount,
|
|
previous_state: currentState,
|
|
}, 'AgentJobService: Reset ticket to backlog for auto-retry after restart');
|
|
|
|
return { reset: true, ticketId: ticketRow.id };
|
|
} catch (err) {
|
|
apiLogger.error({ err, conversationId }, 'AgentJobService: Failed to reset bound ticket');
|
|
return { reset: false };
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Try to automatically re-dispatch a failed/pending job after server restart.
|
|
* Uses the stored context (message_content, agent config) to create a new job.
|
|
*
|
|
* Tracks _restart_count in conversation metadata to prevent infinite restart loops.
|
|
*
|
|
* @param {Object} job - The failed job row (must include context, agent_row_id, etc.)
|
|
* @returns {Promise<boolean>} true if successfully re-dispatched
|
|
*/
|
|
async function _tryRedispatchJob(job) {
|
|
try {
|
|
const context = safeParse(job.context, {});
|
|
const messageContent = context.message_content;
|
|
|
|
// Can't re-dispatch without the original message
|
|
if (!messageContent) {
|
|
apiLogger.warn({ jobId: job.id }, 'AgentJobService: Cannot re-dispatch — no message_content in context');
|
|
return false;
|
|
}
|
|
|
|
// ADR-0057 WP-A: serialize recovery per source-job-id with an advisory lock.
|
|
// pg_try_advisory_xact_lock returns false when another transaction already
|
|
// holds the lock for the same hash — we abort cleanly. The lock auto-releases
|
|
// at commit/rollback. SQLite path skips the lock (no parallel servers there).
|
|
if (isPostgres()) {
|
|
const lockKey = `agent_recover:${job.id}`;
|
|
const lockRow = await dbGet(`SELECT pg_try_advisory_lock(hashtext($1)) AS locked`, [lockKey]);
|
|
if (!lockRow?.locked) {
|
|
apiLogger.info({ jobId: job.id }, 'AgentJobService: recovery advisory lock busy — peer is redispatching, skipping');
|
|
return false;
|
|
}
|
|
}
|
|
|
|
// Check restart count to prevent infinite loops (stored in settings JSONB)
|
|
const conv = await dbGet(
|
|
isPostgres()
|
|
? `SELECT id, settings FROM conversations WHERE id = $1`
|
|
: `SELECT id, settings FROM conversations WHERE id = ?`,
|
|
[job.conversation_id]
|
|
);
|
|
if (!conv) {
|
|
if (isPostgres()) await dbGet(`SELECT pg_advisory_unlock(hashtext($1))`, [`agent_recover:${job.id}`]);
|
|
return false;
|
|
}
|
|
|
|
const convSettings = safeParse(conv.settings, {});
|
|
|
|
// Reset restart count if last restart was more than 10 minutes ago
|
|
// This prevents accumulation during development/debugging with frequent restarts
|
|
const lastRestart = convSettings._last_restart ? new Date(convSettings._last_restart) : null;
|
|
const RESTART_WINDOW_MS = 10 * 60 * 1000; // 10 minutes
|
|
const baseCount = (lastRestart && (Date.now() - lastRestart.getTime()) > RESTART_WINDOW_MS)
|
|
? 0 // reset — enough time has passed since last restart
|
|
: (convSettings._job_restart_count || 0);
|
|
const restartCount = baseCount + 1;
|
|
|
|
if (restartCount > MAX_JOB_RESTART_RETRIES) {
|
|
apiLogger.warn({
|
|
conversationId: job.conversation_id, restartCount, lastRestart: convSettings._last_restart,
|
|
}, 'AgentJobService: Exceeded max restart retries for conversation — user must resend');
|
|
if (isPostgres()) await dbGet(`SELECT pg_advisory_unlock(hashtext($1))`, [`agent_recover:${job.id}`]);
|
|
return false;
|
|
}
|
|
|
|
// Update conversation restart metadata in settings
|
|
const updatedSettings = { ...convSettings, _job_restart_count: restartCount, _last_restart: new Date().toISOString() };
|
|
await dbRun(
|
|
isPostgres()
|
|
? `UPDATE conversations SET settings = $1 WHERE id = $2`
|
|
: `UPDATE conversations SET settings = ? WHERE id = ?`,
|
|
[JSON.stringify(updatedSettings), job.conversation_id]
|
|
);
|
|
|
|
// Resolve the agent user
|
|
const agentResolution = await resolveAgentUser(job.agent_row_id || job.agent_name);
|
|
if (!agentResolution) {
|
|
apiLogger.warn({ jobId: job.id, agentRowId: job.agent_row_id }, 'AgentJobService: Cannot resolve agent for re-dispatch');
|
|
if (isPostgres()) await dbGet(`SELECT pg_advisory_unlock(hashtext($1))`, [`agent_recover:${job.id}`]);
|
|
return false;
|
|
}
|
|
|
|
const agent = agentResolution.user;
|
|
|
|
// Save system message about auto-resume
|
|
await saveStepMessage(job.conversation_id, {
|
|
content: `🔄 Agent "${job.agent_name || 'AI Agent'}" was interrupted by a server restart. Automatically resuming... (attempt ${restartCount}/${MAX_JOB_RESTART_RETRIES})`,
|
|
contentType: 'system', role: 'system', senderType: 'system',
|
|
});
|
|
|
|
// Build continuation message so agent knows to continue, not restart from scratch
|
|
// The conversation history (loaded by processJobLocally) already contains previous work,
|
|
// but the agent needs an explicit instruction to continue rather than re-do everything.
|
|
const continuationMessage = [
|
|
`[SYSTEM CONTINUATION — Server restart recovery, attempt ${restartCount}/${MAX_JOB_RESTART_RETRIES}]`,
|
|
`You were previously working on a task but were interrupted by a server restart.`,
|
|
`Your previous responses and progress are visible in the conversation history above.`,
|
|
`IMPORTANT: Continue from where you left off. Do NOT repeat work you already completed.`,
|
|
`If you're unsure where you stopped, review your last messages in the conversation history.`,
|
|
``,
|
|
`Original request:`,
|
|
messageContent,
|
|
].join('\n');
|
|
|
|
// Re-dispatch with stored context
|
|
const options = {
|
|
agent_mode: context.agent_mode || 'agent',
|
|
thinking_enabled: context.thinking_enabled || false,
|
|
attachments: context.attachments || [],
|
|
attachmentBaseUrl: context.attachmentBaseUrl || '',
|
|
};
|
|
|
|
// ADR-0057 WP-A: lineage on the new row anchors it to the source-job's
|
|
// recovery chain. `recovered_from_job_id` is the ORIGINAL job (chain root,
|
|
// not the immediate predecessor) so the unique partial index serializes
|
|
// ALL retries against the same source. `restart_attempt` is monotonic
|
|
// (no need to track previous attempts for the index — restartCount IS the
|
|
// per-conversation counter and is unique within a chain).
|
|
const recoveredFromJobId = job.recovered_from_job_id || job.id;
|
|
const result = await createAndDispatchJob({
|
|
conversationId: job.conversation_id,
|
|
agent,
|
|
triggeredByUserId: job.trigger_user_id,
|
|
messageContent: continuationMessage,
|
|
options,
|
|
triggerMessageId: job.trigger_message_id,
|
|
recoveredFromJobId,
|
|
restartAttempt: restartCount,
|
|
});
|
|
|
|
if (isPostgres()) await dbGet(`SELECT pg_advisory_unlock(hashtext($1))`, [`agent_recover:${job.id}`]);
|
|
|
|
if (result?.skipped) {
|
|
apiLogger.info({ oldJobId: job.id, reason: result.reason }, 'AgentJobService: redispatch idempotent — peer already started');
|
|
return false;
|
|
}
|
|
|
|
apiLogger.info({
|
|
oldJobId: job.id, conversationId: job.conversation_id,
|
|
agentName: job.agent_name, attempt: restartCount,
|
|
recoveredFromJobId, restartAttempt: restartCount,
|
|
}, 'AgentJobService: Successfully re-dispatched job after restart');
|
|
|
|
return true;
|
|
} catch (err) {
|
|
apiLogger.error({ err, jobId: job.id }, 'AgentJobService: Failed to re-dispatch job');
|
|
try { if (isPostgres()) await dbGet(`SELECT pg_advisory_unlock(hashtext($1))`, [`agent_recover:${job.id}`]); } catch { /* lock may not be held */ }
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Recover stuck jobs on server startup.
|
|
*
|
|
* When pm2/server restarts, any in-flight jobs (status='processing') were killed
|
|
* mid-execution. This function:
|
|
* 1. Marks all 'processing' jobs as 'failed' (they can't be resumed)
|
|
* 2. Clears is_processing on their conversations (unblocks the chat UI)
|
|
* 3. Resets bound tickets to BACKLOG so AgentWorkerService re-dispatches them
|
|
* 4. Saves a system message about the restart
|
|
*
|
|
* Should be called once during server startup, BEFORE AgentWorkerService.start().
|
|
*/
|
|
export async function recoverStuckJobs() {
|
|
try {
|
|
let ticketsReset = 0;
|
|
let chatsRedispatched = 0;
|
|
let pendingRecovered = 0;
|
|
let syncConversationsFixed = 0;
|
|
let gracefulRecovered = 0;
|
|
let sentinelsConsumed = 0;
|
|
let totalRedispatched = 0; // Global counter — limits CLI spawns across all phases
|
|
|
|
// ── Phase 0: Recover gracefully-shutdown jobs ─────────────────────
|
|
// Jobs marked as failed by graceful shutdown have clean context and should be recovered first.
|
|
// Since we no longer kill CLI processes on shutdown, some may still be alive — check PID first.
|
|
// FIX-D: Check BOTH error_message text AND result_metadata JSON flag for reliability.
|
|
const gracefulJobs = await dbAll(
|
|
isPostgres()
|
|
? `SELECT id, job_id, conversation_id, agent_name, agent_row_id, agent_user_id,
|
|
context, trigger_message_id, trigger_user_id, started_at, worker_pid,
|
|
recovered_from_job_id
|
|
FROM agent_jobs
|
|
WHERE status = 'failed'
|
|
AND (error_message = 'Graceful shutdown — will auto-recover on restart'
|
|
OR result_metadata::text LIKE '%"shutdown_recovery":true%'
|
|
OR result_metadata::text LIKE '%"shutdown_recovery": true%')`
|
|
: `SELECT id, job_id, conversation_id, agent_name, agent_row_id, agent_user_id,
|
|
context, trigger_message_id, trigger_user_id, started_at, worker_pid,
|
|
recovered_from_job_id
|
|
FROM agent_jobs
|
|
WHERE status = 'failed'
|
|
AND (error_message = 'Graceful shutdown — will auto-recover on restart'
|
|
OR result_metadata LIKE '%"shutdown_recovery":true%'
|
|
OR result_metadata LIKE '%"shutdown_recovery": true%')`
|
|
);
|
|
|
|
let stillAliveCount = 0;
|
|
if (gracefulJobs && gracefulJobs.length > 0) {
|
|
apiLogger.info({ count: gracefulJobs.length }, 'AgentJobService: Found gracefully-shutdown jobs — recovering');
|
|
for (const job of gracefulJobs) {
|
|
try {
|
|
// Check if the CLI process is still alive (we no longer kill them on shutdown)
|
|
if (job.worker_pid && _isPidAlive(job.worker_pid)) {
|
|
// Process survived the restart — set it back to processing, don't re-dispatch.
|
|
// updateJobStatus flips status away from 'failed' so Phase 0 won't re-match this row.
|
|
await updateJobStatus(job.id, JOB_STATUS.PROCESSING);
|
|
await setConversationProcessing(job.conversation_id, true);
|
|
stillAliveCount++;
|
|
apiLogger.info({
|
|
jobId: job.id, pid: job.worker_pid, agentName: job.agent_name,
|
|
}, 'AgentJobService: CLI process still alive after restart — resuming job');
|
|
continue;
|
|
}
|
|
|
|
await setConversationProcessing(job.conversation_id, false);
|
|
|
|
// Try ticket reset first
|
|
const ticketResult = await resetBoundTicketToBacklog(
|
|
job.conversation_id,
|
|
'Graceful shutdown — auto-recovering'
|
|
);
|
|
|
|
let consumeReason;
|
|
if (ticketResult.reset) {
|
|
ticketsReset++;
|
|
consumeReason = `ticket-reset (ticket ${ticketResult.ticketId})`;
|
|
await saveStepMessage(job.conversation_id, {
|
|
content: `🔄 Agent "${job.agent_name || 'AI Agent'}" was interrupted by a server restart. Automatically resuming...`,
|
|
contentType: 'system', role: 'system', senderType: 'system',
|
|
});
|
|
} else if (totalRedispatched < MAX_RECOVERY_REDISPATCH) {
|
|
const redispatched = await _tryRedispatchJob(job);
|
|
if (redispatched) {
|
|
gracefulRecovered++;
|
|
totalRedispatched++;
|
|
consumeReason = 'redispatched';
|
|
} else {
|
|
consumeReason = 'redispatch-failed';
|
|
await saveStepMessage(job.conversation_id, {
|
|
content: `⚠️ Agent "${job.agent_name || 'AI Agent'}" was interrupted by a server restart. Please send your message again to continue.`,
|
|
contentType: 'system', role: 'system', senderType: 'system',
|
|
});
|
|
}
|
|
} else {
|
|
consumeReason = 'recovery-cap-reached';
|
|
}
|
|
|
|
// T-146479: Consume the sentinel marker so Phase 0 will NOT pick this row
|
|
// up on the next restart. Without this, every restart re-finds the same
|
|
// jobs and re-dispatches / re-spams them — accumulating zombie chats.
|
|
await _consumeSentinelMarker(job.id, consumeReason);
|
|
sentinelsConsumed++;
|
|
|
|
apiLogger.info({
|
|
jobId: job.id, conversationId: job.conversation_id,
|
|
agentName: job.agent_name, ticketReset: ticketResult.reset,
|
|
consumeReason,
|
|
}, 'AgentJobService: Recovered graceful-shutdown job');
|
|
} catch (jobErr) {
|
|
apiLogger.error({ err: jobErr, jobId: job.id }, 'AgentJobService: Failed to recover graceful-shutdown job');
|
|
}
|
|
}
|
|
}
|
|
|
|
// ── Phase 1: Recover stuck 'processing' jobs ──────────────────────
|
|
const stuckJobs = await dbAll(
|
|
isPostgres()
|
|
? `SELECT id, job_id, conversation_id, agent_name, agent_row_id, agent_user_id,
|
|
context, trigger_message_id, trigger_user_id, started_at, worker_pid,
|
|
recovered_from_job_id
|
|
FROM agent_jobs WHERE status = 'processing'`
|
|
: `SELECT id, job_id, conversation_id, agent_name, agent_row_id, agent_user_id,
|
|
context, trigger_message_id, trigger_user_id, started_at, worker_pid,
|
|
recovered_from_job_id
|
|
FROM agent_jobs WHERE status = 'processing'`
|
|
);
|
|
|
|
if (stuckJobs && stuckJobs.length > 0) {
|
|
apiLogger.warn({ count: stuckJobs.length }, 'AgentJobService: Found stuck processing jobs — recovering');
|
|
|
|
for (const job of stuckJobs) {
|
|
try {
|
|
// Check if CLI process is still alive — if so, leave it running
|
|
if (job.worker_pid && _isPidAlive(job.worker_pid)) {
|
|
stillAliveCount++;
|
|
apiLogger.info({
|
|
jobId: job.id, pid: job.worker_pid, agentName: job.agent_name,
|
|
}, 'AgentJobService: Processing job still has alive PID — leaving it');
|
|
continue;
|
|
}
|
|
|
|
// Mark old job as failed
|
|
await failJob(job.id, 'Server restarted while job was processing');
|
|
|
|
// Clear conversation processing state
|
|
await setConversationProcessing(job.conversation_id, false);
|
|
|
|
// Try ticket reset first (for ticket_chat conversations)
|
|
const ticketResult = await resetBoundTicketToBacklog(
|
|
job.conversation_id,
|
|
'Server restarted while agent was processing'
|
|
);
|
|
|
|
if (ticketResult.reset) {
|
|
ticketsReset++;
|
|
await saveStepMessage(job.conversation_id, {
|
|
content: `⚠️ Agent "${job.agent_name || 'AI Agent'}" was interrupted by a server restart. Task will be automatically re-dispatched.`,
|
|
contentType: 'system', role: 'system', senderType: 'system',
|
|
});
|
|
} else if (totalRedispatched < MAX_RECOVERY_REDISPATCH) {
|
|
// Regular conversation — try to auto re-dispatch (within limit)
|
|
const redispatched = await _tryRedispatchJob(job);
|
|
if (redispatched) {
|
|
chatsRedispatched++;
|
|
totalRedispatched++;
|
|
} else {
|
|
// Fallback: tell user to resend (only if re-dispatch impossible)
|
|
await saveStepMessage(job.conversation_id, {
|
|
content: `⚠️ Agent "${job.agent_name || 'AI Agent'}" was interrupted by a server restart. Please send your message again.`,
|
|
contentType: 'system', role: 'system', senderType: 'system',
|
|
});
|
|
}
|
|
}
|
|
|
|
apiLogger.info({
|
|
jobId: job.id, jobUuid: job.job_id,
|
|
conversationId: job.conversation_id,
|
|
agentName: job.agent_name,
|
|
ticketReset: ticketResult.reset,
|
|
ticketId: ticketResult.ticketId,
|
|
}, 'AgentJobService: Recovered stuck job');
|
|
} catch (jobErr) {
|
|
apiLogger.error({ err: jobErr, jobId: job.id }, 'AgentJobService: Failed to recover stuck job');
|
|
}
|
|
}
|
|
} else {
|
|
apiLogger.info('AgentJobService: No stuck processing jobs found on startup');
|
|
}
|
|
|
|
// ── Phase 2: Recover pending jobs (not yet timed out → re-dispatch) ──
|
|
const pendingJobs = await dbAll(
|
|
isPostgres()
|
|
? `SELECT id, job_id, conversation_id, agent_name, agent_row_id, agent_user_id,
|
|
context, trigger_message_id, trigger_user_id, recovered_from_job_id
|
|
FROM agent_jobs WHERE status = 'pending' AND timeout_at >= NOW()`
|
|
: `SELECT id, job_id, conversation_id, agent_name, agent_row_id, agent_user_id,
|
|
context, trigger_message_id, trigger_user_id, recovered_from_job_id
|
|
FROM agent_jobs WHERE status = 'pending' AND timeout_at >= datetime('now')`
|
|
);
|
|
|
|
if (pendingJobs && pendingJobs.length > 0) {
|
|
apiLogger.info({ count: pendingJobs.length }, 'AgentJobService: Found pending jobs to re-dispatch');
|
|
for (const job of pendingJobs) {
|
|
try {
|
|
// Cancel old pending job
|
|
await failJob(job.id, 'Server restarted — re-dispatching');
|
|
await setConversationProcessing(job.conversation_id, false);
|
|
if (totalRedispatched < MAX_RECOVERY_REDISPATCH) {
|
|
const redispatched = await _tryRedispatchJob(job);
|
|
if (redispatched) {
|
|
pendingRecovered++;
|
|
totalRedispatched++;
|
|
}
|
|
}
|
|
} catch (jobErr) {
|
|
apiLogger.error({ err: jobErr, jobId: job.id }, 'AgentJobService: Failed to recover pending job');
|
|
}
|
|
}
|
|
}
|
|
|
|
// Also clean up truly stalled pending jobs (past timeout)
|
|
const stalledPending = await dbAll(
|
|
isPostgres()
|
|
? `SELECT id, job_id, conversation_id, agent_name FROM agent_jobs WHERE status = 'pending' AND timeout_at < NOW()`
|
|
: `SELECT id, job_id, conversation_id, agent_name FROM agent_jobs WHERE status = 'pending' AND timeout_at < datetime('now')`
|
|
);
|
|
|
|
if (stalledPending && stalledPending.length > 0) {
|
|
for (const job of stalledPending) {
|
|
await failJob(job.id, 'Job timed out in pending state');
|
|
await setConversationProcessing(job.conversation_id, false);
|
|
apiLogger.info({ jobId: job.id }, 'AgentJobService: Cleaned up timed-out pending job');
|
|
}
|
|
}
|
|
|
|
// ── Phase 3: Fix orphaned sync-agent conversations ────────────────
|
|
// Conversations where is_processing=true but NO corresponding agent_job exists
|
|
// This happens when non-claude-code (sync) agents crash mid-execution
|
|
const orphanedConversations = await dbAll(
|
|
isPostgres()
|
|
? `SELECT c.id, c.processing_agent_name
|
|
FROM conversations c
|
|
WHERE c.is_processing = true
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM agent_jobs aj
|
|
WHERE aj.conversation_id = c.id
|
|
AND aj.status IN ('pending', 'processing')
|
|
)`
|
|
: `SELECT c.id, c.processing_agent_name
|
|
FROM conversations c
|
|
WHERE c.is_processing = 1
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM agent_jobs aj
|
|
WHERE aj.conversation_id = c.id
|
|
AND aj.status IN ('pending', 'processing')
|
|
)`
|
|
);
|
|
|
|
if (orphanedConversations && orphanedConversations.length > 0) {
|
|
apiLogger.warn({ count: orphanedConversations.length }, 'AgentJobService: Found orphaned sync-agent conversations — clearing is_processing');
|
|
for (const conv of orphanedConversations) {
|
|
try {
|
|
await setConversationProcessing(conv.id, false);
|
|
await saveStepMessage(conv.id, {
|
|
content: `⚠️ Agent "${conv.processing_agent_name || 'AI Agent'}" was interrupted by a server restart. Chat has been unlocked — you can continue.`,
|
|
contentType: 'system', role: 'system', senderType: 'system',
|
|
});
|
|
syncConversationsFixed++;
|
|
} catch (err) {
|
|
apiLogger.error({ err, conversationId: conv.id }, 'AgentJobService: Failed to clear orphaned conversation');
|
|
}
|
|
}
|
|
}
|
|
|
|
const total = (gracefulJobs?.length || 0) + (stuckJobs?.length || 0) + (pendingJobs?.length || 0) + (stalledPending?.length || 0) + syncConversationsFixed;
|
|
apiLogger.info({
|
|
total, stillAliveCount, gracefulRecovered, ticketsReset, chatsRedispatched, pendingRecovered, syncConversationsFixed,
|
|
sentinelsConsumed,
|
|
totalRedispatched, maxRecoveryRedispatch: MAX_RECOVERY_REDISPATCH,
|
|
}, 'AgentJobService: Startup recovery complete');
|
|
|
|
return { recovered: total, stillAliveCount, gracefulRecovered, ticketsReset, chatsRedispatched, pendingRecovered, syncConversationsFixed, sentinelsConsumed };
|
|
} catch (err) {
|
|
apiLogger.error({ err }, 'AgentJobService: Startup recovery failed');
|
|
return { recovered: 0, ticketsReset: 0, error: err.message };
|
|
}
|
|
}
|