godcrm/backend/services/agent-worker/execution.js

// agent-worker/execution.js — Ticket execution, conversation binding, agent config, context builders
import { dbGet, dbRun, isPostgres, safeJsonParse } from '../../database/connection.js';
import { apiLogger } from '../../utils/logger.js';
import { agentLoop as executeAgentToolLoop, saveStepMessage, resolveAllowedTools } from '../AgentLoopService.js';
import {
  resolveAgentProvider as sharedResolveAgentProvider,
  buildAgentSystemPrompt as sharedBuildAgentSystemPrompt,
  detectProvider,
} from '../chat/agent-execution-shared.js';
import { createAndDispatchJob } from '../AgentJobService.js';
import ChainHandoffService, { STATE, AGENT_USERS } from '../ChainHandoffService.js';
import { generateBaseId } from '../../utils/baseId.js';
import {
  TICKETS_TABLE_ID,
  AI_AGENTS_TABLE_ID,
  SPACE_ID,
  AGENT_USER_TO_ROW,
  normalizeAgentId,
} from './constants.js';

/**
 * Execute a single ticket: assign → execute → review/error.
 * @param {Object} ticket - Ticket row from DB
 * @param {Map} _activeJobs - Reference to the active jobs map (for duration tracking)
 */
async function executeTicket(ticket, _activeJobs) {
  const ticketData = safeJsonParse(ticket.data, {});
  // Normalise assigned_to: may be an integer, numeric string, or slug.
  const rawAssignedTo = ticketData.assigned_to;
  const agentUserId = normalizeAgentId(rawAssignedTo);

  if (agentUserId !== rawAssignedTo) {
    apiLogger.info({
      ticket_id: ticket.id,
      raw: rawAssignedTo,
      resolved: agentUserId,
    }, 'AgentWorker: Normalised string-slug assigned_to to integer user ID');
  }

  const agentName = ChainHandoffService.getAgentName(agentUserId);

  apiLogger.info({
    ticket_id: ticket.id,
    what: (ticketData.what || '').substring(0, 80),
    agentUserId,
    agentName,
  }, 'AgentWorker: Executing ticket');

  try {
    // 1. Transition to in_progress (acts as lock)
    await ChainHandoffService.updateTicketStatus({
      ticket_id: ticket.id,
      new_state: STATE.IN_PROGRESS,
      agent_id: agentUserId,
      notes: 'AgentWorker: picked up for execution',
    });

    // 2. Get or create a bound conversation
    const conversationId = await ensureTicketConversation(ticket.id, agentUserId, ticketData);

    // 3. Load agent config from AI Agents table
    const agentConfig = await loadAgentConfig(agentUserId);
    if (!agentConfig) {
      throw new Error(`No agent config found for user ${agentUserId} (${agentName})`);
    }

    // 4. Build execution context
    const systemPrompt = await buildTicketSystemPrompt(agentConfig, ticketData);
    const userMessage = buildTicketUserMessage(ticketData);

    // 5. Post the task as a user message in the ticket conversation
    await saveStepMessage(conversationId, {
      content: userMessage,
      role: 'user',
      senderType: 'system',
      contentType: 'text',
      metadata: JSON.stringify({
        source: 'agent_worker',
        ticket_id: ticket.id,
      }),
    });

    // 6. Resolve provider and detect type
    const resolved = await sharedResolveAgentProvider(agentConfig);
    const agentRowId = AGENT_USER_TO_ROW[agentUserId] || null;
    const { isClaudeCode } = detectProvider(resolved.provider, resolved.model);

    let responseText;

    if (isClaudeCode) {
      // ── Claude Code agents: async dispatch via AgentJobService ──
      apiLogger.info({
        ticket_id: ticket.id,
        agentName,
        provider: resolved.provider,
      }, 'AgentWorker: Claude-code agent detected, dispatching async job');

      const agentForJob = {
        id: agentUserId,
        name: agentName,
        managed_by_agent_row_id: agentRowId,
        _agentConfig: agentConfig,
      };

      await createAndDispatchJob({
        conversationId,
        agent: agentForJob,
        triggeredByUserId: agentUserId,
        messageContent: userMessage,
        options: { agent_mode: 'agent' },
        ticketId: ticket.id,
      });

      // Claude Code job runs async — do NOT transition ticket here.
      apiLogger.info({ ticket_id: ticket.id, agentName }, 'AgentWorker: Claude-code job dispatched, ticket stays in_progress until job completes');
      return; // Skip the synchronous REVIEW transition below
    } else {
      // ── Non-claude-code agents: synchronous tool loop ──
      responseText = await executeAgentToolLoop({
        conversationId,
        systemPrompt,
        history: [],
        userMessage,
        agentConfig,
        resolved,
        agentRowId,
        senderId: agentUserId,
        spaceId: SPACE_ID,
        userId: agentUserId,
        ticketData,
      });
    }

    // 7. Success → transition to review
    await ChainHandoffService.updateTicketStatus({
      ticket_id: ticket.id,
      new_state: STATE.REVIEW,
      agent_id: agentUserId,
      notes: `Completed. Response: ${(responseText || '').substring(0, 200)}`,
    });

    // 8. Log activity
    await ChainHandoffService.logActivity({
      action: 'agent_worker_completed',
      agent_id: agentUserId,
      ticket_id: ticket.id,
      chain_id: ticketData._chain?.chain_id,
      details: {
        response_length: responseText?.length || 0,
        duration_ms: Date.now() - (_activeJobs.get(ticket.id)?.startedAt || Date.now()),
      },
    });

    apiLogger.info({
      ticket_id: ticket.id,
      agentName,
      responseLength: responseText?.length || 0,
    }, 'AgentWorker: Ticket completed successfully');

  } catch (err) {
    apiLogger.error({ err, ticket_id: ticket.id, agentName }, 'AgentWorker: Ticket execution failed');

    // Save error to ticket conversation
    try {
      const convId = await findTicketConversation(ticket.id);
      if (convId) {
        await saveStepMessage(convId, {
          content: `Agent "${agentName}" encountered an error: ${err.message || 'Unknown error'}`,
          role: 'system',
          senderType: 'system',
          contentType: 'text',
          metadata: JSON.stringify({
            type: 'agent_error',
            error: err.message,
            ticket_id: ticket.id,
          }),
        });
      }
    } catch (msgErr) {
      apiLogger.error({ err: msgErr }, 'AgentWorker: Failed to save error message');
    }

    // Update ticket back to backlog so it can be retried
    try {
      await ChainHandoffService.updateTicketStatus({
        ticket_id: ticket.id,
        new_state: STATE.BACKLOG,
        agent_id: agentUserId,
        notes: `Error (will retry): ${err.message || 'Unknown error'}`,
      });
    } catch (statusErr) {
      apiLogger.error({ err: statusErr }, 'AgentWorker: Failed to update ticket status on error');
    }
  } finally {
    _activeJobs.delete(ticket.id);
  }
}

// ----- TICKET-CONVERSATION BINDING -----

/**
 * Get or create a conversation bound to a ticket.
 */
async function ensureTicketConversation(ticketId, agentUserId, ticketData) {
  // Check for existing bound conversation
  const existing = await findTicketConversation(ticketId);
  if (existing) return existing;

  // Create new conversation bound to the ticket
  const title = `Ticket #${ticketId}: ${(ticketData.what || 'Task').substring(0, 100)}`;
  const baseId = generateBaseId('conv');

  const result = await dbRun(
    isPostgres()
      ? `INSERT INTO conversations (space_id, type, title, bound_table_id, bound_row_id, created_by, base_id, created_at, updated_at)
         VALUES ($1, 'ticket_chat', $2, $3, $4, $5, $6, NOW(), NOW())
         RETURNING id`
      : `INSERT INTO conversations (space_id, type, title, bound_table_id, bound_row_id, created_by, base_id, created_at, updated_at)
         VALUES (?, 'ticket_chat', ?, ?, ?, ?, ?, datetime('now'), datetime('now'))`,
    [SPACE_ID, title, TICKETS_TABLE_ID, ticketId, agentUserId, baseId]
  );

  const conversationId = result?.lastInsertRowid;

  if (!conversationId) {
    throw new Error(`Failed to create conversation for ticket #${ticketId}`);
  }

  // Add agent as participant
  try {
    await dbRun(
      isPostgres()
        ? `INSERT INTO conversation_participants (conversation_id, user_id, role, joined_at)
           VALUES ($1, $2, 'agent', NOW())
           ON CONFLICT (conversation_id, user_id) DO NOTHING`
        : `INSERT OR IGNORE INTO conversation_participants (conversation_id, user_id, role, joined_at)
           VALUES (?, ?, 'agent', datetime('now'))`,
      [conversationId, agentUserId]
    );
  } catch (partErr) {
    apiLogger.warn({ err: partErr }, 'AgentWorker: Failed to add agent as participant (non-fatal)');
  }

  apiLogger.info({
    conversationId,
    ticketId,
    agentUserId,
    title,
  }, 'AgentWorker: Created ticket conversation');

  return conversationId;
}

/**
 * Find existing conversation bound to a ticket.
 */
async function findTicketConversation(ticketId) {
  const row = await dbGet(
    isPostgres()
      ? `SELECT id FROM conversations WHERE bound_table_id = $1 AND bound_row_id = $2`
      : `SELECT id FROM conversations WHERE bound_table_id = ? AND bound_row_id = ?`,
    [TICKETS_TABLE_ID, ticketId]
  );
  return row?.id || null;
}

// ----- AGENT CONFIG -----

/**
 * Load agent configuration from AI Agents table by user ID.
 */
async function loadAgentConfig(agentUserId) {
  // Map user ID to agent row ID
  const agentRowId = AGENT_USER_TO_ROW[agentUserId];

  if (agentRowId) {
    const row = await dbGet(
      isPostgres()
        ? `SELECT data FROM table_rows WHERE id = $1 AND table_id = $2`
        : `SELECT data FROM table_rows WHERE id = ? AND table_id = ?`,
      [agentRowId, AI_AGENTS_TABLE_ID]
    );
    if (row) {
      const config = safeJsonParse(row.data, {});
      config.row_id = agentRowId;
      return config;
    }
  }

  // Fallback: search by agent name
  const agentName = ChainHandoffService.getAgentName(agentUserId);
  const row = await dbGet(
    isPostgres()
      ? `SELECT tr.id, tr.data FROM table_rows tr
         WHERE tr.table_id = $1 AND tr.data->>'name' ILIKE $2
         LIMIT 1`
      : `SELECT tr.id, tr.data FROM table_rows tr
         WHERE tr.table_id = ? AND json_extract(tr.data, '$.name') LIKE ?
         LIMIT 1`,
    [AI_AGENTS_TABLE_ID, `%${agentName}%`]
  );

  if (row) {
    const config = safeJsonParse(row.data, {});
    config.row_id = row.id;
    return config;
  }

  return null;
}

// ----- CONTEXT BUILDERS -----

/**
 * Build the system prompt for ticket execution.
 * Combines the agent's own system prompt with ticket-specific context.
 */
async function buildTicketSystemPrompt(agentConfig, ticketData) {
  const basePrompt = await sharedBuildAgentSystemPrompt(agentConfig, {
    spaceId: SPACE_ID,
    agentMode: 'agent',
  }, 'account');

  const ticketContext = [
    '',
    '## Current Task (from Ticket)',
    '',
    `**Ticket**: #${ticketData._chain?.ticket_id || 'unknown'}`,
    `**Task**: ${ticketData.what || 'No description'}`,
    ticketData.acceptance_criteria ? `**Acceptance Criteria**: ${ticketData.acceptance_criteria}` : '',
    ticketData._chain ? `**Chain**: ${ticketData._chain.chain_id} (step ${ticketData._chain.step})` : '',
    '',
    'Complete this task. Use the available tools. Report your results.',
  ].filter(Boolean).join('\n');

  return basePrompt + ticketContext;
}

/**
 * Build the user message from ticket data.
 */
function buildTicketUserMessage(ticketData) {
  const parts = [
    `## Task: ${ticketData.what || 'No description'}`,
  ];

  if (ticketData.why) {
    parts.push('', `### Context`, ticketData.why);
  }

  if (ticketData.acceptance_criteria) {
    parts.push('', `### Acceptance Criteria`, ticketData.acceptance_criteria);
  }

  if (ticketData._chain) {
    parts.push('', `### Chain Info`, `Chain: ${ticketData._chain.chain_id} | Step: ${ticketData._chain.step}`);
    if (ticketData._chain.dispatched_by) {
      const dispatcherName = ChainHandoffService.getAgentName(ticketData._chain.dispatched_by);
      parts.push(`Dispatched by: ${dispatcherName}`);
    }
  }

  return parts.join('\n');
}

export {
  executeTicket,
  ensureTicketConversation,
  findTicketConversation,
  loadAgentConfig,
  buildTicketSystemPrompt,
  buildTicketUserMessage,
};