openmaic/OpenMAIC/lib/makelore-runtime/large-course-control.ts

import {
  cancelCourseGenerationJob,
  isCourseGenerationRunning,
  regenerateCourseFramework,
  runCourseFrameworkGeneration,
  runCourseGenerationJob,
  runCourseModuleGeneration,
} from '@/lib/course-framework/runner';
import {
  isValidCourseId,
  readCourseRecordReconciled,
  updateCourseRecord,
} from '@/lib/course-framework/store';
import {
  CourseMutationInProgressError,
  isCoursePublishing,
} from '@/lib/course-framework/publish-state';
import {
  assertLearningCourseMutable,
  FrozenLearningCourseMutationError,
} from '@/lib/makelore-course/immutability';

export const LARGE_COURSE_CONTROL_ACTIONS = [
  'start',
  'cancel',
  'resume',
  'framework-regenerate',
  'module-regenerate',
] as const;

export type LargeCourseControlAction = (typeof LARGE_COURSE_CONTROL_ACTIONS)[number];

export type LargeCourseControlResult = {
  courseId: string;
  action: LargeCourseControlAction;
  status: 'framework_generating' | 'generating' | 'cancelling';
  message: string;
  moduleIndex?: number;
  run?: Promise<void>;
};

export class LargeCourseControlError extends Error {
  constructor(
    public readonly code: string,
    public readonly status: number,
    message: string,
    public readonly details?: Record<string, unknown>,
  ) {
    super(message);
    this.name = 'LargeCourseControlError';
  }
}

function conflict(message: string, code = 'invalid_course_state'): never {
  throw new LargeCourseControlError(code, 409, message);
}

/**
 * The only process-local entry point for large-course runner mutations.
 *
 * Engine runtime routes, including the legacy Works cancel/resume paths, call
 * this function so the controller maps and their cancellation signals always
 * belong to the same Node process.
 */
export async function controlLargeCourse(options: {
  courseId: string;
  ownerPrincipalId: string;
  action: LargeCourseControlAction;
  baseUrl: string;
  moduleIndex?: number;
}): Promise<LargeCourseControlResult> {
  const { courseId, ownerPrincipalId, action, baseUrl, moduleIndex } = options;
  if (!isValidCourseId(courseId) || !ownerPrincipalId.trim()) {
    throw new LargeCourseControlError('invalid_request', 400, 'Invalid course or owner');
  }
  const record = await readCourseRecordReconciled(courseId);
  if (!record || record.ownerPrincipalId !== ownerPrincipalId) {
    throw new LargeCourseControlError('job_not_found', 404, 'Course generation job not found');
  }

  try {
    // Cancellation remains available to stop an in-flight finalize race. All
    // actions that can create new source state are forbidden after freezing.
    if (action !== 'cancel') await assertLearningCourseMutable(courseId, record);
    switch (action) {
      case 'cancel': {
        if (!isCourseGenerationRunning(courseId) || !cancelCourseGenerationJob(courseId)) {
          conflict('Course generation is not running', 'job_not_running');
        }
        await updateCourseRecord(courseId, { status: 'cancelled' });
        return {
          courseId,
          action,
          status: 'cancelling',
          message: 'Cancellation requested',
        };
      }

      case 'start': {
        if (!record.framework) conflict('课程框架尚未生成，请等待框架生成完成');
        if (record.status !== 'framework_ready') {
          conflict(`当前状态（${record.status}）不允许启动模块生成`);
        }
        if (
          isCourseGenerationRunning(courseId)
          || isCoursePublishing(courseId)
          || record.modules.some((entry) => entry.status === 'generating')
        ) {
          conflict('该课程已有生成任务正在运行', 'job_already_running');
        }
        return {
          courseId,
          action,
          status: 'generating',
          message: '模块生成已开始',
          run: runCourseModuleGeneration(courseId, baseUrl),
        };
      }

      case 'resume': {
        if (record.status === 'framework_ready') {
          throw new LargeCourseControlError(
            'review_required',
            409,
            'Course framework requires operations review',
            { reviewUrl: `/learning-ops/courses/${record.id}` },
          );
        }
        if (record.status === 'completed') conflict('Course is already completed', 'job_not_resumable');
        if (isCourseGenerationRunning(courseId) || isCoursePublishing(courseId)) {
          conflict('该课程已有生成任务正在运行', 'job_not_resumable');
        }
        const hasFramework = Boolean(record.framework);
        return {
          courseId,
          action,
          status: hasFramework ? 'generating' : 'framework_generating',
          message: hasFramework ? '模块生成已继续' : '框架生成已继续',
          run: hasFramework
            ? runCourseModuleGeneration(courseId, baseUrl)
            : runCourseFrameworkGeneration(courseId, baseUrl),
        };
      }

      case 'framework-regenerate': {
        if (
          isCourseGenerationRunning(courseId)
          || isCoursePublishing(courseId)
          || record.modules.some((entry) => entry.status === 'generating')
        ) {
          conflict('模块正在生成中，无法重新生成框架', 'job_already_running');
        }
        return {
          courseId,
          action,
          status: 'framework_generating',
          message: '框架重新生成已开始',
          run: regenerateCourseFramework(courseId, baseUrl),
        };
      }

      case 'module-regenerate': {
        if (!Number.isInteger(moduleIndex) || moduleIndex! < 1) {
          throw new LargeCourseControlError('invalid_request', 400, 'Invalid module index');
        }
        if (!record.framework) {
          conflict('Course framework not generated yet; cannot regenerate a module');
        }
        if (!record.modules.some((entry) => entry.index === moduleIndex)) {
          throw new LargeCourseControlError(
            'module_not_found',
            404,
            `Module ${moduleIndex} not found`,
          );
        }
        if (
          isCourseGenerationRunning(courseId)
          || isCoursePublishing(courseId)
          || record.modules.some((entry) => entry.status === 'generating')
        ) {
          conflict('该课程已有生成任务正在运行', 'job_already_running');
        }
        return {
          courseId,
          action,
          moduleIndex,
          status: 'generating',
          message: `Module ${moduleIndex} and downstream regeneration started`,
          run: runCourseGenerationJob(courseId, baseUrl, { onlyModuleIndex: moduleIndex }),
        };
      }
    }
  } catch (error) {
    if (error instanceof LargeCourseControlError) throw error;
    if (error instanceof FrozenLearningCourseMutationError) {
      throw new LargeCourseControlError('course_frozen', 409, error.message);
    }
    if (error instanceof CourseMutationInProgressError) {
      throw new LargeCourseControlError('job_already_running', 409, error.message);
    }
    if (error instanceof Error && error.message.includes('already has an active generation run')) {
      throw new LargeCourseControlError('job_already_running', 409, error.message);
    }
    throw error;
  }
}

export function largeCourseControlErrorResponse(error: unknown): Response {
  if (error instanceof LargeCourseControlError) {
    return Response.json(
      { error: error.code, message: error.message, ...error.details },
      { status: error.status },
    );
  }
  return Response.json(
    {
      error: 'internal_error',
      message: error instanceof Error ? error.message : 'Large-course control failed',
    },
    { status: 500 },
  );
}