memind/goal-run-service.mjs

import crypto from 'node:crypto';
import { buildGoalContextEnvelope as buildGoalContextEnvelopeText } from './goal-run-context.mjs';
import { shouldRequireCheckpointApproval } from './goal-run-policy.mjs';

const GOAL_TABLE = 'h5_goal_runs';
const CHECKPOINT_TABLE = 'h5_goal_checkpoints';

const TERMINAL_CHECKPOINT_STATUSES = new Set([
  'approved',
  'skipped',
  'failed',
]);

export function buildGoalRunSchemaSql() {
  return [
    `CREATE TABLE IF NOT EXISTS \`${GOAL_TABLE}\` (
      id CHAR(36) PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      title VARCHAR(500) NOT NULL,
      intent_summary TEXT NOT NULL,
      status ENUM(
        'draft',
        'active',
        'paused',
        'awaiting_user',
        'completed',
        'failed',
        'cancelled'
      ) NOT NULL DEFAULT 'active',
      priority TINYINT UNSIGNED NOT NULL DEFAULT 5,
      source_channel ENUM('h5', 'wechat', 'agent', 'api') NOT NULL DEFAULT 'h5',
      source_session_id VARCHAR(128) NULL,
      source_message_id VARCHAR(128) NULL,
      current_checkpoint_id CHAR(36) NULL,
      context_json JSON NULL,
      memory_snapshot_json JSON NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      completed_at BIGINT NULL,
      KEY idx_goal_user_status (user_id, status, updated_at),
      KEY idx_goal_user_active (user_id, status, created_at),
      CONSTRAINT fk_goal_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
    `CREATE TABLE IF NOT EXISTS \`${CHECKPOINT_TABLE}\` (
      id CHAR(36) PRIMARY KEY,
      goal_run_id CHAR(36) NOT NULL,
      sequence INT UNSIGNED NOT NULL,
      title VARCHAR(300) NOT NULL,
      description TEXT NULL,
      status ENUM(
        'pending',
        'running',
        'awaiting_approval',
        'approved',
        'skipped',
        'failed'
      ) NOT NULL DEFAULT 'pending',
      agent_run_id CHAR(36) NULL,
      output_summary TEXT NULL,
      output_artifact_ids JSON NULL,
      user_feedback TEXT NULL,
      approved_at BIGINT NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      started_at BIGINT NULL,
      completed_at BIGINT NULL,
      UNIQUE KEY uq_goal_checkpoint_seq (goal_run_id, sequence),
      KEY idx_checkpoint_goal_status (goal_run_id, status, sequence),
      CONSTRAINT fk_checkpoint_goal FOREIGN KEY (goal_run_id) REFERENCES \`${GOAL_TABLE}\`(id) ON DELETE CASCADE,
      CONSTRAINT fk_checkpoint_agent_run FOREIGN KEY (agent_run_id) REFERENCES h5_agent_runs(id) ON DELETE SET NULL
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
  ];
}

export async function ensureGoalRunSchema(pool, { columnExists, indexExists } = {}) {
  if (!pool?.query) throw new Error('Goal run schema requires a MySQL pool');
  for (const sql of buildGoalRunSchemaSql()) {
    await pool.query(sql);
  }
  const hasColumn = columnExists ?? (async (table, column) => {
    const [rows] = await pool.query(
      `SELECT 1 AS ok FROM information_schema.columns
       WHERE table_schema = DATABASE() AND table_name = ? AND column_name = ?
       LIMIT 1`,
      [table, column],
    );
    return rows.length > 0;
  });
  const hasIndex = indexExists ?? (async (table, index) => {
    const [rows] = await pool.query(
      `SELECT 1 AS ok FROM information_schema.statistics
       WHERE table_schema = DATABASE() AND table_name = ? AND index_name = ?
       LIMIT 1`,
      [table, index],
    );
    return rows.length > 0;
  });

  if (!(await hasColumn('h5_agent_runs', 'goal_run_id'))) {
    await pool.query(
      'ALTER TABLE h5_agent_runs ADD COLUMN goal_run_id CHAR(36) NULL AFTER agent_session_id',
    );
  }
  if (!(await hasColumn('h5_agent_runs', 'goal_checkpoint_id'))) {
    await pool.query(
      'ALTER TABLE h5_agent_runs ADD COLUMN goal_checkpoint_id CHAR(36) NULL AFTER goal_run_id',
    );
  }
  if (!(await hasIndex('h5_agent_runs', 'idx_agent_run_goal'))) {
    await pool.query(
      'ALTER TABLE h5_agent_runs ADD KEY idx_agent_run_goal (goal_run_id, updated_at)',
    );
  }
}

function parseJson(value, fallback = null) {
  if (value == null) return fallback;
  if (typeof value === 'object') return value;
  try {
    return JSON.parse(String(value));
  } catch {
    return fallback;
  }
}

function normalizeGoalRow(row) {
  return {
    id: String(row.id),
    userId: String(row.user_id),
    title: String(row.title),
    intentSummary: String(row.intent_summary),
    status: String(row.status),
    priority: Number(row.priority ?? 5),
    sourceChannel: String(row.source_channel ?? 'h5'),
    sourceSessionId: row.source_session_id == null ? null : String(row.source_session_id),
    sourceMessageId: row.source_message_id == null ? null : String(row.source_message_id),
    currentCheckpointId: row.current_checkpoint_id == null ? null : String(row.current_checkpoint_id),
    context: parseJson(row.context_json, null),
    memorySnapshot: parseJson(row.memory_snapshot_json, null),
    createdAt: Number(row.created_at ?? 0),
    updatedAt: Number(row.updated_at ?? 0),
    completedAt: row.completed_at == null ? null : Number(row.completed_at),
  };
}

function normalizeCheckpointRow(row) {
  return {
    id: String(row.id),
    goalRunId: String(row.goal_run_id),
    sequence: Number(row.sequence ?? 0),
    title: String(row.title),
    description: row.description == null ? null : String(row.description),
    status: String(row.status),
    agentRunId: row.agent_run_id == null ? null : String(row.agent_run_id),
    outputSummary: row.output_summary == null ? null : String(row.output_summary),
    outputArtifactIds: parseJson(row.output_artifact_ids, []),
    userFeedback: row.user_feedback == null ? null : String(row.user_feedback),
    approvedAt: row.approved_at == null ? null : Number(row.approved_at),
    createdAt: Number(row.created_at ?? 0),
    updatedAt: Number(row.updated_at ?? 0),
    startedAt: row.started_at == null ? null : Number(row.started_at),
    completedAt: row.completed_at == null ? null : Number(row.completed_at),
  };
}

export function createGoalRunService({ pool = null, now = () => Date.now(), env = process.env } = {}) {
  if (!pool?.query) return null;

  async function advanceGoalAfterCheckpointApproved(goal, approvedCheckpoint) {
    const timestamp = now();
    const nextPending = goal.checkpoints.find(
      (item) => item.sequence > approvedCheckpoint.sequence
        && ['pending', 'failed'].includes(item.status),
    );
    if (nextPending) {
      await pool.query(
        `UPDATE ${GOAL_TABLE}
         SET status = 'active', current_checkpoint_id = ?, updated_at = ?
         WHERE id = ?`,
        [nextPending.id, timestamp, goal.id],
      );
      return { nextCheckpointId: nextPending.id, goalCompleted: false };
    }
    await pool.query(
      `UPDATE ${GOAL_TABLE}
       SET status = 'completed',
           current_checkpoint_id = ?,
           completed_at = ?,
           updated_at = ?
       WHERE id = ?`,
      [approvedCheckpoint.id, timestamp, timestamp, goal.id],
    );
    return { nextCheckpointId: null, goalCompleted: true };
  }

  return {
    ensureSchema: () => ensureGoalRunSchema(pool),

    async createGoalRun({
      userId,
      title,
      intentSummary,
      sourceChannel = 'h5',
      sourceSessionId = null,
      sourceMessageId = null,
      checkpoints = [],
      context = null,
      memorySnapshot = null,
    } = {}) {
      if (!userId || !title || !intentSummary) {
        throw new Error('createGoalRun requires userId, title, and intentSummary');
      }
      const goalId = crypto.randomUUID();
      const timestamp = now();
      const initialCheckpoints = checkpoints.length
        ? checkpoints
        : [{ title: '启动', description: intentSummary }];
      await pool.query(
        `INSERT INTO ${GOAL_TABLE}
          (id, user_id, title, intent_summary, status, priority, source_channel,
           source_session_id, source_message_id, context_json, memory_snapshot_json,
           created_at, updated_at)
         VALUES (?, ?, ?, ?, 'active', 5, ?, ?, ?, ?, ?, ?, ?)`,
        [
          goalId,
          String(userId),
          String(title).slice(0, 500),
          String(intentSummary),
          String(sourceChannel),
          sourceSessionId,
          sourceMessageId,
          context == null ? null : JSON.stringify(context),
          memorySnapshot == null ? null : JSON.stringify(memorySnapshot),
          timestamp,
          timestamp,
        ],
      );

      let firstCheckpointId = null;
      for (let index = 0; index < initialCheckpoints.length; index += 1) {
        const checkpoint = initialCheckpoints[index];
        const checkpointId = crypto.randomUUID();
        if (index === 0) firstCheckpointId = checkpointId;
        await pool.query(
          `INSERT INTO ${CHECKPOINT_TABLE}
            (id, goal_run_id, sequence, title, description, status, created_at, updated_at)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
          [
            checkpointId,
            goalId,
            index + 1,
            String(checkpoint.title ?? `阶段 ${index + 1}`).slice(0, 300),
            checkpoint.description == null ? null : String(checkpoint.description),
            index === 0 ? 'pending' : 'pending',
            timestamp,
            timestamp,
          ],
        );
      }

      if (firstCheckpointId) {
        await pool.query(
          `UPDATE ${GOAL_TABLE} SET current_checkpoint_id = ?, updated_at = ? WHERE id = ?`,
          [firstCheckpointId, timestamp, goalId],
        );
      }

      return this.getGoalRun({ userId, goalRunId: goalId });
    },

    async getGoalRun({ userId, goalRunId } = {}) {
      if (!userId || !goalRunId) return null;
      const [rows] = await pool.query(
        `SELECT * FROM ${GOAL_TABLE} WHERE id = ? AND user_id = ? LIMIT 1`,
        [String(goalRunId), String(userId)],
      );
      const row = rows[0];
      if (!row) return null;
      const goal = normalizeGoalRow(row);
      const [checkpointRows] = await pool.query(
        `SELECT * FROM ${CHECKPOINT_TABLE} WHERE goal_run_id = ? ORDER BY sequence ASC`,
        [goal.id],
      );
      goal.checkpoints = checkpointRows.map(normalizeCheckpointRow);
      return goal;
    },

    async listGoalRuns({ userId, statuses = ['active', 'awaiting_user', 'paused'], limit = 20 } = {}) {
      if (!userId) return [];
      const normalizedStatuses = (statuses ?? []).map(String).filter(Boolean);
      if (!normalizedStatuses.length) return [];
      const placeholders = normalizedStatuses.map(() => '?').join(', ');
      const safeLimit = Math.max(1, Math.min(100, Number(limit) || 20));
      const [rows] = await pool.query(
        `SELECT * FROM ${GOAL_TABLE}
         WHERE user_id = ? AND status IN (${placeholders})
         ORDER BY updated_at DESC
         LIMIT ?`,
        [String(userId), ...normalizedStatuses, safeLimit],
      );
      return rows.map(normalizeGoalRow);
    },

    buildGoalContextEnvelope(goal) {
      return buildGoalContextEnvelopeText(goal);
    },

    async startNextCheckpoint({ userId, goalRunId, agentRunId = null } = {}) {
      if (!userId || !goalRunId) {
        throw new Error('startNextCheckpoint requires userId and goalRunId');
      }
      const goal = await this.getGoalRun({ userId, goalRunId });
      if (!goal) {
        const error = new Error('目标不存在');
        error.code = 'GOAL_RUN_NOT_FOUND';
        throw error;
      }
      if (!['active', 'awaiting_user'].includes(goal.status)) {
        const error = new Error(`目标状态 ${goal.status} 不可启动阶段`);
        error.code = 'GOAL_RUN_NOT_STARTABLE';
        throw error;
      }

      const running = goal.checkpoints.find((item) => item.status === 'running');
      if (running) {
        if (agentRunId && !running.agentRunId) {
          const timestamp = now();
          await pool.query(
            `UPDATE ${CHECKPOINT_TABLE}
             SET agent_run_id = ?, updated_at = ?
             WHERE id = ? AND goal_run_id = ?`,
            [String(agentRunId), timestamp, running.id, goal.id],
          );
        }
        return {
          goalRunId: goal.id,
          checkpointId: running.id,
          checkpoint: running,
        };
      }

      const nextCheckpoint = goal.checkpoints.find((item) => item.status === 'pending')
        ?? goal.checkpoints.find((item) => item.status === 'failed');
      if (!nextCheckpoint) {
        const error = new Error('没有可启动的阶段');
        error.code = 'GOAL_CHECKPOINT_UNAVAILABLE';
        throw error;
      }

      const timestamp = now();
      await pool.query(
        `UPDATE ${CHECKPOINT_TABLE}
         SET status = 'running',
             agent_run_id = ?,
             started_at = COALESCE(started_at, ?),
             updated_at = ?
         WHERE id = ? AND goal_run_id = ?`,
        [
          agentRunId ? String(agentRunId) : null,
          timestamp,
          timestamp,
          nextCheckpoint.id,
          goal.id,
        ],
      );
      await pool.query(
        `UPDATE ${GOAL_TABLE}
         SET status = 'active', current_checkpoint_id = ?, updated_at = ?
         WHERE id = ? AND user_id = ?`,
        [nextCheckpoint.id, timestamp, goal.id, String(userId)],
      );

      return {
        goalRunId: goal.id,
        checkpointId: nextCheckpoint.id,
        checkpoint: {
          ...nextCheckpoint,
          status: 'running',
          agentRunId: agentRunId ? String(agentRunId) : nextCheckpoint.agentRunId,
        },
      };
    },

    async attachAgentRunToCheckpoint({ checkpointId, agentRunId } = {}) {
      if (!checkpointId || !agentRunId) return false;
      const timestamp = now();
      const [result] = await pool.query(
        `UPDATE ${CHECKPOINT_TABLE}
         SET agent_run_id = ?, updated_at = ?
         WHERE id = ? AND (agent_run_id IS NULL OR agent_run_id = ?)`,
        [String(agentRunId), timestamp, String(checkpointId), String(agentRunId)],
      );
      return Number(result?.affectedRows ?? 0) > 0;
    },

    async onAgentRunCompleted({ agentRunId, status, outputSummary = null } = {}) {
      if (!agentRunId) return { handled: false, reason: 'missing_agent_run_id' };
      const [runRows] = await pool.query(
        `SELECT id, goal_run_id, goal_checkpoint_id, status
         FROM h5_agent_runs
         WHERE id = ?
         LIMIT 1`,
        [String(agentRunId)],
      );
      const run = runRows[0];
      if (!run?.goal_checkpoint_id) return { handled: false, reason: 'not_goal_run' };

      const [checkpointRows] = await pool.query(
        `SELECT c.*, g.user_id, g.status AS goal_status
         FROM ${CHECKPOINT_TABLE} c
         INNER JOIN ${GOAL_TABLE} g ON g.id = c.goal_run_id
         WHERE c.id = ?
         LIMIT 1`,
        [String(run.goal_checkpoint_id)],
      );
      const checkpoint = checkpointRows[0];
      if (!checkpoint) return { handled: false, reason: 'checkpoint_missing' };
      if (TERMINAL_CHECKPOINT_STATUSES.has(String(checkpoint.status))) {
        return { handled: true, reason: 'already_terminal', checkpointId: checkpoint.id };
      }

      const timestamp = now();
      const normalizedStatus = String(status ?? run.status ?? '').trim();
      if (normalizedStatus === 'succeeded') {
        const [pendingRows] = await pool.query(
          `SELECT id FROM ${CHECKPOINT_TABLE}
           WHERE goal_run_id = ? AND status IN ('pending', 'failed')
           ORDER BY sequence ASC`,
          [checkpoint.goal_run_id],
        );
        const [countRows] = await pool.query(
          `SELECT COUNT(*) AS total FROM ${CHECKPOINT_TABLE} WHERE goal_run_id = ?`,
          [checkpoint.goal_run_id],
        );
        const pendingCheckpointCount = pendingRows.length;
        const totalCheckpointCount = Number(countRows[0]?.total ?? 0);
        const requiresApproval = shouldRequireCheckpointApproval({
          env,
          pendingCheckpointCount,
          totalCheckpointCount,
        });

        if (requiresApproval) {
          await pool.query(
            `UPDATE ${CHECKPOINT_TABLE}
             SET status = 'awaiting_approval',
                 output_summary = COALESCE(?, output_summary),
                 completed_at = ?,
                 updated_at = ?
             WHERE id = ? AND status = 'running'`,
            [
              outputSummary == null ? null : String(outputSummary).slice(0, 4000),
              timestamp,
              timestamp,
              checkpoint.id,
            ],
          );
          await pool.query(
            `UPDATE ${GOAL_TABLE}
             SET status = 'awaiting_user', updated_at = ?
             WHERE id = ?`,
            [timestamp, checkpoint.goal_run_id],
          );
          return {
            handled: true,
            checkpointId: checkpoint.id,
            goalRunId: checkpoint.goal_run_id,
            awaitingApproval: true,
          };
        }

        await pool.query(
          `UPDATE ${CHECKPOINT_TABLE}
           SET status = 'approved',
               output_summary = COALESCE(?, output_summary),
               completed_at = ?,
               approved_at = COALESCE(approved_at, ?),
               updated_at = ?
           WHERE id = ? AND status = 'running'`,
          [
            outputSummary == null ? null : String(outputSummary).slice(0, 4000),
            timestamp,
            timestamp,
            timestamp,
            checkpoint.id,
          ],
        );
        if (pendingRows[0]?.id) {
          await pool.query(
            `UPDATE ${GOAL_TABLE}
             SET status = 'active',
                 current_checkpoint_id = ?,
                 updated_at = ?
             WHERE id = ?`,
            [pendingRows[0].id, timestamp, checkpoint.goal_run_id],
          );
          return {
            handled: true,
            checkpointId: checkpoint.id,
            goalRunId: checkpoint.goal_run_id,
            nextCheckpointId: pendingRows[0].id,
          };
        }
        await pool.query(
          `UPDATE ${GOAL_TABLE}
           SET status = 'completed',
               current_checkpoint_id = ?,
               completed_at = ?,
               updated_at = ?
           WHERE id = ?`,
          [checkpoint.id, timestamp, timestamp, checkpoint.goal_run_id],
        );
        return {
          handled: true,
          checkpointId: checkpoint.id,
          goalRunId: checkpoint.goal_run_id,
          goalCompleted: true,
        };
      }

      if (normalizedStatus === 'failed') {
        await pool.query(
          `UPDATE ${CHECKPOINT_TABLE}
           SET status = 'failed',
               output_summary = COALESCE(?, output_summary),
               completed_at = ?,
               updated_at = ?
           WHERE id = ? AND status = 'running'`,
          [
            outputSummary == null ? null : String(outputSummary).slice(0, 4000),
            timestamp,
            checkpoint.id,
          ],
        );
        await pool.query(
          `UPDATE ${GOAL_TABLE}
           SET status = 'failed', updated_at = ?
           WHERE id = ? AND status <> 'cancelled'`,
          [timestamp, checkpoint.goal_run_id],
        );
        return {
          handled: true,
          checkpointId: checkpoint.id,
          goalRunId: checkpoint.goal_run_id,
          failed: true,
        };
      }

      return { handled: false, reason: 'non_terminal_status', status: normalizedStatus };
    },

    async approveCheckpoint({
      userId,
      goalRunId,
      checkpointId,
      feedback = null,
    } = {}) {
      const goal = await this.getGoalRun({ userId, goalRunId });
      if (!goal) {
        const error = new Error('目标不存在');
        error.code = 'GOAL_RUN_NOT_FOUND';
        throw error;
      }
      const checkpoint = goal.checkpoints.find((item) => item.id === String(checkpointId));
      if (!checkpoint) {
        const error = new Error('阶段不存在');
        error.code = 'GOAL_CHECKPOINT_NOT_FOUND';
        throw error;
      }
      if (checkpoint.status !== 'awaiting_approval') {
        const error = new Error(`阶段状态 ${checkpoint.status} 不可确认`);
        error.code = 'GOAL_CHECKPOINT_NOT_APPROVABLE';
        throw error;
      }
      const timestamp = now();
      await pool.query(
        `UPDATE ${CHECKPOINT_TABLE}
         SET status = 'approved',
             user_feedback = ?,
             approved_at = ?,
             updated_at = ?
         WHERE id = ? AND goal_run_id = ?`,
        [
          feedback == null ? null : String(feedback).slice(0, 4000),
          timestamp,
          timestamp,
          checkpoint.id,
          goal.id,
        ],
      );
      const refreshed = await this.getGoalRun({ userId, goalRunId: goal.id });
      const approvedCheckpoint = refreshed.checkpoints.find((item) => item.id === checkpoint.id);
      await advanceGoalAfterCheckpointApproved(refreshed, approvedCheckpoint);
      return this.getGoalRun({ userId, goalRunId: goal.id });
    },

    async pauseGoal({ userId, goalRunId } = {}) {
      const timestamp = now();
      const [result] = await pool.query(
        `UPDATE ${GOAL_TABLE}
         SET status = 'paused', updated_at = ?
         WHERE id = ? AND user_id = ? AND status IN ('active', 'awaiting_user')`,
        [timestamp, String(goalRunId), String(userId)],
      );
      if (Number(result?.affectedRows ?? 0) === 0) {
        const error = new Error('目标不存在或不可暂停');
        error.code = 'GOAL_RUN_NOT_PAUSABLE';
        throw error;
      }
      return this.getGoalRun({ userId, goalRunId });
    },

    async resumeGoal({ userId, goalRunId } = {}) {
      const goal = await this.getGoalRun({ userId, goalRunId });
      if (!goal) {
        const error = new Error('目标不存在');
        error.code = 'GOAL_RUN_NOT_FOUND';
        throw error;
      }
      if (!['paused', 'awaiting_user', 'failed'].includes(goal.status)) {
        const error = new Error(`目标状态 ${goal.status} 不可续作`);
        error.code = 'GOAL_RUN_NOT_RESUMABLE';
        throw error;
      }
      const timestamp = now();
      await pool.query(
        `UPDATE ${GOAL_TABLE}
         SET status = 'active', updated_at = ?
         WHERE id = ? AND user_id = ?`,
        [timestamp, goal.id, String(userId)],
      );
      const started = await this.startNextCheckpoint({ userId, goalRunId: goal.id });
      return {
        goal: await this.getGoalRun({ userId, goalRunId: goal.id }),
        checkpointId: started.checkpointId,
      };
    },

    async cancelGoal({ userId, goalRunId } = {}) {
      const timestamp = now();
      const [result] = await pool.query(
        `UPDATE ${GOAL_TABLE}
         SET status = 'cancelled', updated_at = ?, completed_at = COALESCE(completed_at, ?)
         WHERE id = ? AND user_id = ? AND status NOT IN ('completed', 'cancelled')`,
        [timestamp, timestamp, String(goalRunId), String(userId)],
      );
      if (Number(result?.affectedRows ?? 0) === 0) {
        const error = new Error('目标不存在或已结束');
        error.code = 'GOAL_RUN_NOT_CANCELLABLE';
        throw error;
      }
      await pool.query(
        `UPDATE ${CHECKPOINT_TABLE}
         SET status = 'skipped', updated_at = ?
         WHERE goal_run_id = ? AND status IN ('pending', 'running', 'awaiting_approval')`,
        [timestamp, String(goalRunId)],
      );
      return this.getGoalRun({ userId, goalRunId });
    },
  };
}