/** * ADR-093/113: Processing state management and plan handling. * * Extracted from agent-execution-shared.js */ import { dbGet, dbRun, isPostgres } from '../../../database/connection.js'; import { apiLogger } from '../../../utils/logger.js'; // ─── setConversationProcessing() ────────────────────────────── /** * ADR-093 D2 / ADR-0057 WP-C: Set or clear the is_processing flag on a conversation. * * Starting (isProcessing=true): * - is_processing flips to true unconditionally. * - processing_agent_id / _name / _started_at are set only if currently NULL * (COALESCE). The first agent in a multi-agent session keeps the scalar * slot; siblings are reflected via `active_agents` in the messages API. * * Clearing (isProcessing=false): * - If any OTHER agent_jobs row for this conversation is still in * 'pending'/'processing', we leave the scalars alone — at least one agent * is still working. The legacy single-agent UI continues to show the * original owner until everyone is done; the new UI reads `active_agents`. * - Otherwise we NULL the scalars as before. * * @param {number} conversationId - Conversation ID * @param {boolean} isProcessing - true to start, false to clear * @param {string|null} agentName - Agent display name (only used when isProcessing=true) * @param {number|null} agentRowId - Agent row ID (only used when isProcessing=true) */ export async function setConversationProcessing(conversationId, isProcessing, agentName = null, agentRowId = null) { try { if (isProcessing) { await dbRun( isPostgres() ? `UPDATE conversations SET is_processing = true, processing_started_at = COALESCE(processing_started_at, NOW()), processing_agent_id = COALESCE(processing_agent_id, $2), processing_agent_name = COALESCE(processing_agent_name, $3), updated_at = NOW() WHERE id = $1` : `UPDATE conversations SET is_processing = 1, processing_started_at = COALESCE(processing_started_at, datetime('now')), processing_agent_id = COALESCE(processing_agent_id, ?), processing_agent_name = COALESCE(processing_agent_name, ?), updated_at = datetime('now') WHERE id = ?`, isPostgres() ? [conversationId, agentRowId, agentName] : [agentRowId, agentName, conversationId] ); return; } // ADR-0057 WP-C: scope the clear — only NULL scalars if no sibling job // is still pending/processing for this conversation. const siblingCountRow = await dbGet( isPostgres() ? `SELECT COUNT(*)::int AS n FROM agent_jobs WHERE conversation_id = $1 AND status IN ('pending', 'processing')` : `SELECT COUNT(*) AS n FROM agent_jobs WHERE conversation_id = ? AND status IN ('pending', 'processing')`, [conversationId] ); const siblingsStillProcessing = Number(siblingCountRow?.n || 0); if (siblingsStillProcessing > 0) { // Another agent is still working — bump updated_at but leave processing fields. await dbRun( isPostgres() ? `UPDATE conversations SET updated_at = NOW() WHERE id = $1` : `UPDATE conversations SET updated_at = datetime('now') WHERE id = ?`, [conversationId] ); return; } await dbRun( isPostgres() ? `UPDATE conversations SET is_processing = false, processing_started_at = NULL, processing_agent_id = NULL, processing_agent_name = NULL, updated_at = NOW() WHERE id = $1` : `UPDATE conversations SET is_processing = 0, processing_started_at = NULL, processing_agent_id = NULL, processing_agent_name = NULL, updated_at = datetime('now') WHERE id = ?`, [conversationId] ); } catch (err) { apiLogger.error({ err, conversationId, isProcessing }, 'ADR-093/0057: setConversationProcessing failed'); } } // ─── handleManagePlan() ─────────────────────────────────────── /** * ADR-113: Shared manage_plan handler. * * Validates the tasks array, sanitises each task, then (when a conversationId * is provided) upserts a plan message in the conversation. If a plan message * (content_type = 'plan') already exists for the conversation, it is updated * in place. Otherwise a new message row is inserted. * * When called without a conversationId the function performs validation only * and returns the progress summary without touching the database. * * Both AgentToolsService.executeTool() and AgentLoopService inline handlers * delegate to this single implementation to avoid duplication. * * @param {Object|null|undefined} args - Tool arguments ({ tasks: Array }) * @param {number|null} conversationId - Current conversation ID (null = validation-only) * @param {string} agentName - Name of the calling agent * @param {Object} [options] - Optional extra context * @param {number|null} [options.agentId] - Agent user/row ID * @returns {Promise} Human-readable progress string, or error object */ export async function handleManagePlan(args, conversationId, agentName, options = {}) { const { tasks } = args || {}; // ── Validate input ── if (!Array.isArray(tasks) || tasks.length === 0) { return { error: 'tasks must be a non-empty array' }; } if (tasks.length > 20) { return { error: 'Maximum 20 tasks allowed in a plan' }; } const validStatuses = ['pending', 'in_progress', 'completed', 'blocked']; const validatedTasks = []; for (const task of tasks) { if (typeof task.id !== 'number') { return { error: `Task id must be a number. Got: ${JSON.stringify(task.id)}` }; } if (typeof task.title !== 'string' || !task.title.trim()) { return { error: `Task title must be a non-empty string. Task id=${task.id}` }; } if (!validStatuses.includes(task.status)) { return { error: `Invalid status "${task.status}" for task ${task.id}. Must be one of: ${validStatuses.join(', ')}` }; } validatedTasks.push({ id: task.id, title: task.title.trim(), status: task.status, ...(task.note ? { note: String(task.note) } : {}) }); } // ── Compute progress summary ── const completed = validatedTasks.filter(t => t.status === 'completed').length; const inProgress = validatedTasks.filter(t => t.status === 'in_progress').length; const pending = validatedTasks.filter(t => t.status === 'pending').length; const blocked = validatedTasks.filter(t => t.status === 'blocked').length; const total = validatedTasks.length; // ── Persist to DB when conversationId is available ── if (conversationId) { const { agentId = null } = options; const contentJson = JSON.stringify({ tasks: validatedTasks }); const metadataJson = JSON.stringify({ agent: agentName, tool: 'manage_plan', version: 1, updated_at: new Date().toISOString() }); const pg = isPostgres(); // Check for existing plan message in this conversation const existingPlan = await dbGet( pg ? `SELECT id FROM messages WHERE conversation_id = $1 AND content_type = 'plan' ORDER BY id DESC LIMIT 1` : `SELECT id FROM messages WHERE conversation_id = ? AND content_type = 'plan' ORDER BY id DESC LIMIT 1`, [conversationId] ); if (existingPlan) { // UPDATE existing plan in-place — frontend polls for plan updates via content_type check await dbRun( pg ? `UPDATE messages SET content = $1, metadata = $2, updated_at = NOW() WHERE id = $3` : `UPDATE messages SET content = ?, metadata = ?, updated_at = datetime('now') WHERE id = ?`, [contentJson, metadataJson, existingPlan.id] ); apiLogger.info({ planId: existingPlan.id, progress: `${completed}/${total}` }, 'handleManagePlan: plan updated in-place'); } else { // INSERT new plan message await dbRun( pg ? `INSERT INTO messages (conversation_id, sender_id, sender_type, role, content, content_type, agent_id, metadata, created_at, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW(), NOW())` : `INSERT INTO messages (conversation_id, sender_id, sender_type, role, content, content_type, agent_id, metadata, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now'))`, [conversationId, agentId || null, 'agent', 'tool_result', contentJson, 'plan', agentId || null, metadataJson] ); apiLogger.info({ conversationId, progress: `${completed}/${total}` }, 'handleManagePlan: plan created'); } } // ── Build human-readable confirmation ── const parts = []; if (completed > 0) parts.push(`${completed}/${total} completed`); if (inProgress > 0) parts.push(`${inProgress} in progress`); if (pending > 0) parts.push(`${pending} pending`); if (blocked > 0) parts.push(`${blocked} blocked`); const summary = parts.length > 0 ? parts.join(', ') : '0 tasks'; return `Plan updated: ${summary}`; }