1.生成剧本资产接口改成异步生成逻辑2.新增定时任务调度跑生成剧本资产任务3.新增work子进程并发跑调度任务

app/Console/Commands/TextScriptGenerateDispatchCommand.php

@@ -0,0 +1,283 @@
+<?php
+
+namespace App\Console\Commands;
+
+use Illuminate\Console\Command;
+use Illuminate\Support\Facades\DB;
+
+/**
+ * 剧本资产生成异步任务调度器
+ *
+ * 由 cron 每分钟触发一次。进入后以 5 秒为间隔轮询 pending 任务，
+ * 发现任务立即拉起独立 worker 子进程（text:generate:run）；
+ * 若连续 50 秒没有任何任务则退出，避免长时间占用调度进程。
+ */
+class TextScriptGenerateDispatchCommand extends Command
+{
+    /**
+     * The name and signature of the console command.
+     *
+     * @var string
+     */
+    protected $signature = 'text:generate:dispatch';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = '剧本资产生成异步任务调度：每5秒查询pending任务并拉起worker';
+
+    /**
+     * Execute the console command.
+     *
+     * @return int
+     */
+    public function handle()
+    {
+        dLog('command')->info('开始执行剧本资产生成任务调度...');
+
+        // 每次调度窗口执行一次历史 chunks 清理（任务完结后 chunks 仅供 SSE 推送，无长期保留价值）
+        $cleaned = $this->cleanupExpiredChunks();
+        if ($cleaned > 0) {
+            dLog('command')->info("已清理过期剧本资产生成chunks: {$cleaned} 条");
+        }
+
+        $concurrency = (int)env('SCRIPT_TEXT_GENERATION_CONCURRENCY', 3);
+        if ($concurrency < 1) {
+            $concurrency = 1;
+        }
+
+        $start = time();
+        $idleSince = null;
+        $spawnedTotal = 0;
+
+        while (true) {
+            $now = time();
+
+            // 孤儿任务恢复：processing 但超过5分钟且处理锁已空闲，说明 worker 异常退出，重置回 pending
+            $this->recoverOrphanTasks();
+
+            // 连续 50 秒没有任何任务则退出
+            if ($idleSince !== null && $now - $idleSince >= 50) {
+                dLog('command')->info('已连续50秒无待处理任务，退出本次调度');
+                break;
+            }
+
+            // 整体安全上限（防止异常情况下长时间占用）
+            if ($now - $start > 75) {
+                dLog('command')->info('达到调度窗口安全上限，退出本次调度');
+                break;
+            }
+
+            // 并发上限 = 配置上限 - 已在运行的 processing 任务数 - 本窗口已拉起数
+            $activeCount = (int)DB::table('mp_script_generate_tasks')
+                ->where('status', 'processing')
+                ->count();
+            $available = $concurrency - $activeCount - $spawnedTotal;
+            if ($available <= 0) {
+                dLog('command')->info("当前活跃任务已达并发上限，等待worker完成...");
+                sleep(5);
+                continue;
+            }
+
+            $pendingIds = DB::table('mp_script_generate_tasks')
+                ->where('status', 'pending')
+                ->orderBy('id')
+                ->pluck('id')
+                ->all();
+
+            if (empty($pendingIds)) {
+                if ($idleSince === null) {
+                    $idleSince = $now;
+                }
+                if ($now - $idleSince >= 50) {
+                    break;
+                }
+                sleep(5);
+                continue;
+            }
+
+            $idleSince = null;
+            foreach ($pendingIds as $taskId) {
+                if ($available <= 0) {
+                    break;
+                }
+
+                // 仅当该任务的处理锁空闲时才拉起，避免与手动/其它调度重复
+                $lockName = TextScriptGenerateRunCommand::LOCK_PREFIX . $taskId;
+                $isFree = DB::selectOne('SELECT IS_FREE_LOCK(?) AS is_free', [$lockName]);
+                if (!$isFree || (int)$isFree->is_free !== 1) {
+                    dLog('command')->info("task_id: {$taskId} 正在被其他进程处理，跳过");
+                    continue;
+                }
+
+                try {
+                    $this->spawnWorker((int)$taskId);
+                    $spawnedTotal++;
+                    $available--;
+                    dLog('command')->info("已拉起剧本资产生成worker, task_id: {$taskId}");
+                } catch (\Throwable $e) {
+                    dLog('command')->error("拉起剧本资产生成worker失败 task_id: {$taskId}: " . $e->getMessage());
+                    logDB('command', 'error', '剧本资产生成拉起失败', [
+                        'task_id' => (int)$taskId,
+                        'error'   => $e->getMessage(),
+                    ]);
+                }
+            }
+
+            sleep(5);
+        }
+
+        dLog('command')->info("剧本资产生成任务调度结束，本次拉起worker数: {$spawnedTotal}");
+        return 0;
+    }
+
+    /**
+     * 清理已完结任务的过期 chunks
+     *
+     * 保留期默认 30 天，可通过环境变量 SCRIPT_GENERATE_CHUNK_RETENTION_DAYS 调整。
+     * 只清理 status 为 success/failed 且 completed_at 早于保留期的任务；
+     * 按任务 id 升序游标分批删除，避免单次删除量过大造成长时间锁表。
+     *
+     * @return int 删除的 chunks 记录数
+     */
+    private function cleanupExpiredChunks(): int
+    {
+        $retentionDays = (int)env('SCRIPT_GENERATE_CHUNK_RETENTION_DAYS', 7);
+        if ($retentionDays < 1) {
+            $retentionDays = 7;
+        }
+
+        $cutoff = date('Y-m-d H:i:s', strtotime("-{$retentionDays} days"));
+        $batchRows = (int)env('SCRIPT_GENERATE_CHUNK_DELETE_BATCH', 5000);
+        if ($batchRows < 1000) {
+            $batchRows = 5000;
+        }
+
+        $deletedTotal = 0;
+
+        // 按行数小批循环删除，避免单次 DELETE 量过大造成长事务/长锁。
+        // MySQL 单表 DELETE 支持 LIMIT；目标表是 chunks，子查询引用 tasks，不同表无同表限制。
+        $deleteSql = 'DELETE FROM mp_script_generate_task_chunks '
+            . 'WHERE task_id IN ('
+            . 'SELECT id FROM mp_script_generate_tasks '
+            . 'WHERE status IN (?, ?) AND completed_at IS NOT NULL AND completed_at < ?'
+            . ') LIMIT ' . (int)$batchRows;
+
+        $maxRounds = 2000; // 防御性上限，正常数据远用不到
+        for ($round = 0; $round < $maxRounds; $round++) {
+            $deleted = DB::affectingStatement($deleteSql, ['success', 'failed', $cutoff]);
+            if ($deleted <= 0) {
+                break;
+            }
+            $deletedTotal += $deleted;
+            if ($deleted < $batchRows) {
+                break;
+            }
+        }
+
+        return $deletedTotal;
+    }
+
+    /**
+     * 拉起一个后台 worker 子进程处理指定任务
+     *
+     * @param int $taskId
+     * @return void
+     */
+    private function spawnWorker(int $taskId)
+    {
+        $command = [
+            $this->phpBinary(),
+            base_path('artisan'),
+            'text:generate:run',
+            (string)$taskId,
+        ];
+
+        if (DIRECTORY_SEPARATOR === '/') {
+            // Linux：用 nohup 后台启动并立即返回
+            $logFile = storage_path('logs/text-generate-worker-' . date('Ymd') . '.log');
+            $commandLine = 'nohup ' . implode(' ', array_map('escapeshellarg', $command))
+                . ' >> ' . escapeshellarg($logFile) . ' 2>&1 &';
+            exec($commandLine);
+            dLog('command')->info('拉起剧本资产生成worker命令: ' . $commandLine);
+        } else {
+            // Windows：使用 cmd start /B 分离启动。
+            // 注意：不能用 Symfony Process::start()，其对象析构时会 stop(0) 终止刚拉起的子进程。
+            $logFile = storage_path('logs/text-generate-worker-' . date('Ymd') . '.log');
+            $commandLine = 'start /B "" ' . implode(' ', array_map('escapeshellarg', $command))
+                . ' >> ' . escapeshellarg($logFile) . ' 2>&1';
+            pclose(popen($commandLine, 'r'));
+            dLog('command')->info('拉起剧本资产生成worker命令: ' . $commandLine);
+        }
+    }
+
+    /**
+     * 恢复异常退出遗留的孤儿任务
+     *
+     * worker 进程被强杀/崩溃时不会执行收尾，任务会一直停留在 processing。
+     * 若任务已 processing 超过 5 分钟且其处理锁处于空闲状态，说明原 worker 已不在，
+     * 将任务重置回 pending，等待重新调度。
+     *
+     * @return int 恢复的任务数
+     */
+    private function recoverOrphanTasks(): int
+    {
+        $recovered = 0;
+        $threshold = date('Y-m-d H:i:s', strtotime('-5 minutes'));
+
+        $candidates = DB::table('mp_script_generate_tasks')
+            ->where('status', 'processing')
+            ->where('started_at', '<', $threshold)
+            ->pluck('id')
+            ->all();
+
+        foreach ($candidates as $taskId) {
+            $lockName = TextScriptGenerateRunCommand::LOCK_PREFIX . $taskId;
+            $isFree = DB::selectOne('SELECT IS_FREE_LOCK(?) AS is_free', [$lockName]);
+            if (!$isFree || (int)$isFree->is_free !== 1) {
+                continue;
+            }
+
+            $updated = DB::table('mp_script_generate_tasks')
+                ->where('id', $taskId)
+                ->where('status', 'processing')
+                ->update([
+                    'status'        => 'pending',
+                    'error_message' => null,
+                    'completed_at'  => null,
+                    'updated_at'    => date('Y-m-d H:i:s'),
+                ]);
+            if ($updated) {
+                $recovered++;
+                dLog('command')->warning("孤儿任务恢复, task_id: {$taskId} 已重置为pending");
+                logDB('command', 'warning', '孤儿任务恢复', [
+                    'task_id' => (int)$taskId,
+                    'message' => 'worker异常退出，任务重置为pending等待重新调度',
+                ]);
+            }
+        }
+
+        return $recovered;
+    }
+
+    /**
+     * 获取当前 PHP 可执行文件的真实路径
+     *
+     * @return string
+     */
+    private function phpBinary()
+    {
+        $envBinary = getenv('PHP_BINARY');
+        if ($envBinary && is_executable($envBinary)) {
+            return $envBinary;
+        }
+
+        if (DIRECTORY_SEPARATOR === '/' && ($real = @readlink('/proc/self/exe')) && is_executable($real)) {
+            return $real;
+        }
+
+        return PHP_BINARY;
+    }
+}

app/Console/Commands/TextScriptGenerateRunCommand.php

@@ -0,0 +1,96 @@
+<?php
+
+namespace App\Console\Commands;
+
+use App\Services\DeepSeek\DeepSeekService;
+use Illuminate\Console\Command;
+use Illuminate\Support\Facades\DB;
+
+/**
+ * 剧本资产生成异步任务 worker
+ *
+ * 由 text:generate:dispatch 拉起；通过 MySQL 命名锁防止同一任务被重复执行。
+ */
+class TextScriptGenerateRunCommand extends Command
+{
+    /**
+     * 任务处理锁前缀（配合 MySQL GET_LOCK 使用）
+     */
+    public const LOCK_PREFIX = 'script_text_generate_';
+
+    /**
+     * The name and signature of the console command.
+     *
+     * @var string
+     */
+    protected $signature = 'text:generate:run {task_id : 剧本资产生成任务ID}';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = '剧本资产生成worker：执行指定任务的流式模型生成并落库';
+
+    /**
+     * @var DeepSeekService
+     */
+    protected $deepSeekService;
+
+    /**
+     * Create a new command instance.
+     *
+     * @return void
+     */
+    public function __construct(DeepSeekService $deepSeekService)
+    {
+        parent::__construct();
+        $this->deepSeekService = $deepSeekService;
+    }
+
+    /**
+     * Execute the console command.
+     *
+     * @return int
+     */
+    public function handle()
+    {
+        set_time_limit(0);
+
+        $taskId = (int)$this->argument('task_id');
+        $lockName = self::LOCK_PREFIX . $taskId;
+
+        // 非阻塞抢占该任务的处理锁；抢不到说明已有其他进程正在处理
+        $acquired = DB::selectOne('SELECT GET_LOCK(?, 0) AS acquired', [$lockName]);
+        if (!$acquired || (int)$acquired->acquired !== 1) {
+            dLog('command')->info("task_id: {$taskId} 正在被其他进程处理，本次直接退出");
+            return 0;
+        }
+
+        dLog('command')->info("task_id: {$taskId} 抢占处理锁成功，开始处理...");
+
+        try {
+            $result = $this->deepSeekService->runScriptGenerateTaskAsync($taskId);
+            dLog('command')->info('task_id: ' . $taskId . ' 处理结果: ' . json_encode($result, JSON_UNESCAPED_UNICODE));
+            if (empty($result['ok'])) {
+                logDB('command', 'error', '剧本资产生成worker异常', [
+                    'task_id' => $taskId,
+                    'status'  => $result['status'] ?? null,
+                    'error'   => $result['error'] ?? '未知错误',
+                ]);
+            }
+        } catch (\Throwable $e) {
+            dLog('command')->error("task_id: {$taskId} 处理异常: " . $e->getMessage());
+            logDB('command', 'error', '剧本资产生成worker异常', [
+                'task_id' => $taskId,
+                'error'   => $e->getMessage(),
+            ]);
+        } finally {
+            // 显式释放处理锁；即使进程异常退出，连接断开后 MySQL 也会自动释放
+            DB::statement('SELECT RELEASE_LOCK(?)', [$lockName]);
+            dLog('command')->info("task_id: {$taskId} 处理结束，已释放处理锁");
+        }
+
+        return 0;
+    }
+}

app/Console/Kernel.php

@@ -53,6 +53,9 @@ class Kernel extends ConsoleKernel
         // 批量生成分集任务（每分钟执行一次，并发调度不同anime_id的子进程处理，默认并发5）
         $schedule->command('batch:generate-episodes')->everyMinute()->withoutOverlapping();
 
+        // 剧本资产生成异步任务调度（命令内部每5秒轮询 pending 任务并拉起 worker）
+        $schedule->command('text:generate:dispatch')->everyMinute()->withoutOverlapping();
+
         // 同步任务中心状态和结果（关联图片/视频任务）
         $schedule->command('taskCenter:sync')->everyMinute()->withoutOverlapping(5);
 

app/Http/Controllers/DeepSeek/DeepSeekController.php

@@ -547,6 +547,110 @@ class DeepSeekController extends BaseController
 
         $data = $request->all();
 
+        // stream=1 + script_id：剧本资产生成走异步任务模式（worker 执行，SSE 每 10s 轮询推送，最长 30 分钟）
+        if ((int)getProp($data, 'stream', 0) === 1 && (int)getProp($data, 'script_id', 0) > 0) {
+            $taskId = $this->deepseekService->createScriptGenerateAsyncTask($data);
+            $uid = (int)\App\Facade\Site::getUid();
+
+            return response()->stream(function () use ($taskId, $uid) {
+                if (ob_get_level()) {
+                    ob_end_clean();
+                }
+                ini_set('output_buffering', 'off');
+                ini_set('zlib.output_compression', 'off');
+                if (function_exists('apache_setenv')) {
+                    apache_setenv('no-gzip', '1');
+                }
+
+                $push = function (array $chunk) {
+                    echo "data: " . json_encode($chunk, JSON_UNESCAPED_UNICODE) . "\n\n";
+                    if (ob_get_level() > 0) {
+                        ob_flush();
+                    }
+                    flush();
+                };
+
+                // SSE 轮询：每 10s 查询一次任务状态与新增 chunks，最长等待 30 分钟
+                $deadline = time() + 1800;
+                $lastSeq = -1;
+                $terminalPushed = false;
+
+                try {
+                    while (time() < $deadline) {
+                        if (connection_aborted()) {
+                            // 浏览器断开：停止推送，后台 worker 继续执行并落库
+                            return;
+                        }
+
+                        $poll = $this->deepseekService->pollScriptGenerateTask($taskId, $uid, $lastSeq);
+                        if (!$poll['found']) {
+                            $push([
+                                'type' => 'error',
+                                'message' => '生成任务不存在或无权访问',
+                            ]);
+                            return;
+                        }
+
+                        foreach ($poll['chunks'] as $chunk) {
+                            $push($chunk);
+                            $lastSeq++;
+                            if (isset($chunk['type']) && in_array($chunk['type'], ['done', 'error'])) {
+                                $terminalPushed = true;
+                            }
+                        }
+
+                        $status = (string)getProp($poll, 'status', '');
+
+                        // 任务已结束：补齐可能尚未写入的 done/error chunk 后关闭连接
+                        if (in_array($status, ['success', 'failed'])) {
+                            $waitRound = 0;
+                            while (!$terminalPushed && $waitRound < 3) {
+                                sleep(3);
+                                $waitRound++;
+                                $pollRetry = $this->deepseekService->pollScriptGenerateTask($taskId, $uid, $lastSeq);
+                                foreach ($pollRetry['chunks'] as $chunk) {
+                                    $push($chunk);
+                                    $lastSeq++;
+                                    if (isset($chunk['type']) && in_array($chunk['type'], ['done', 'error'])) {
+                                        $terminalPushed = true;
+                                    }
+                                }
+                            }
+
+                            if (!$terminalPushed && $status === 'failed') {
+                                $push([
+                                    'type' => 'error',
+                                    'message' => (string)getProp($poll, 'error_message', '') !== ''
+                                        ? (string)getProp($poll, 'error_message', '')
+                                        : '内容生成失败',
+                                ]);
+                            }
+                            return;
+                        }
+
+                        // 未完成：每 10s 轮询一次
+                        sleep(10);
+                    }
+
+                    // 30 分钟未完成：向前端提示超时，任务继续在后台执行
+                    $push([
+                        'type' => 'error',
+                        'message' => '生成超时，任务仍在后台继续执行，请稍后查看任务列表',
+                    ]);
+                } catch (\Throwable $e) {
+                    $push([
+                        'type' => 'error',
+                        'message' => $e->getMessage(),
+                    ]);
+                }
+            }, 200, [
+                'Content-Type' => 'text/event-stream',
+                'Cache-Control' => 'no-cache',
+                'Connection' => 'keep-alive',
+                'X-Accel-Buffering' => 'no',
+            ]);
+        }
+
         // stream=1：SSE 流式输出（格式与 generateText 一致：yield type=content/reasoning/done）
         if ((int)getProp($data, 'stream', 0) === 1) {
             return response()->stream(function () use ($data) {

app/Models/MpScriptGenerateTask.php

@@ -22,6 +22,7 @@ class MpScriptGenerateTask extends Model
         'prompt',
         'result',
         'json_result',
+        'request_params',
         'error_message',
         'started_at',
         'completed_at',

app/Services/DeepSeek/DeepSeekService.php

@@ -762,7 +762,7 @@ class DeepSeekService
      * @param array $data 请求参数
      * @return \Generator
      */
-    public function newGenerateTextStream($data) {
+    public function newGenerateTextStream($data, $existingTask = null) {
         $model = getProp($data, 'model', 'deepseek-v4-pro');
         $messages = getProp($data, 'messages', []);
         $systemPrompt = getProp($data, 'system_prompt', '');
@@ -775,14 +775,15 @@ class DeepSeekService
         $sequence = getProp($data, 'sequence', 0);
 
         $generateTaskId = 0;
-        $uid = Site::getUid();
-        if ($script_id > 0) {
-            $existingTask = DB::table('mp_script_generate_tasks')
+        // 异步 worker 场景直接使用任务记录中的用户，避免命令行下无登录上下文
+        $uid = $existingTask ? (int)$existingTask->uid : (int)Site::getUid();
+        if ($script_id > 0 && !$existingTask) {
+            $runningTask = DB::table('mp_script_generate_tasks')
                 ->where('uid', $uid)
                 ->where('script_id', $script_id)
                 ->orderByDesc('id')
                 ->first();
-            if ($existingTask && in_array(getProp($existingTask, 'status'), ['pending', 'processing'])) {
+            if ($runningTask && in_array(getProp($runningTask, 'status'), ['pending', 'processing'])) {
                 Utils::throwError('20003:该剧本资产正在生成中，请稍后重试');
             }
         }
@@ -958,17 +959,22 @@ class DeepSeekService
 
         // 如果提供了script_id，创建剧本资产生成任务
         if ($script_id > 0) {
-            $uid = Site::getUid();
-            $generateTaskId = DB::table('mp_script_generate_tasks')->insertGetId([
-                'uid'        => $uid,
-                'script_id'  => $script_id,
-                'sequence'   => $sequence,
-                'status'     => 'processing',
-                'prompt'     => getProp($data, 'prompt', ''),
-                'started_at' => date('Y-m-d H:i:s'),
-                'created_at' => date('Y-m-d H:i:s'),
-                'updated_at' => date('Y-m-d H:i:s')
-            ]);
+            if ($existingTask) {
+                // 异步 worker：接管已创建的 pending 任务，不再新建
+                $generateTaskId = (int)$existingTask->id;
+            } else {
+                // 同步调用：保持原行为，创建 processing 任务
+                $generateTaskId = DB::table('mp_script_generate_tasks')->insertGetId([
+                    'uid'        => $uid,
+                    'script_id'  => $script_id,
+                    'sequence'   => $sequence,
+                    'status'     => 'processing',
+                    'prompt'     => getProp($data, 'prompt', ''),
+                    'started_at' => date('Y-m-d H:i:s'),
+                    'created_at' => date('Y-m-d H:i:s'),
+                    'updated_at' => date('Y-m-d H:i:s')
+                ]);
+            }
         }
 
         // 积分余额预检（仅存在有效剧本时按 chat 类型计费，不足直接报错）
@@ -1033,47 +1039,50 @@ class DeepSeekService
                     if (!empty($ctx['script_id'])) {
                         try {
                             $now = now();
-                            DB::table('mp_script_records')->insert([
-                                [
-                                    'uid' => $ctx['uid'],
-                                    'script_id' => $ctx['script_id'],
-                                    'sequence' => $ctx['sequence'],
-                                    'role' => 'user',
-                                    'content' => $ctx['original_prompt'],
-                                    'created_at' => $now,
-                                    'updated_at' => $now,
-                                ],
-                                [
-                                    'uid' => $ctx['uid'],
-                                    'script_id' => $ctx['script_id'],
-                                    'sequence' => $ctx['sequence'],
-                                    'role' => 'assistant',
-                                    'content' => $fullContent,
-                                    'created_at' => $now,
-                                    'updated_at' => $now,
-                                ],
-                            ]);
-
-                            $chunk['records'] = DB::table('mp_script_records')
-                                ->where('uid', $ctx['uid'])
-                                ->where('script_id', $ctx['script_id'])
-                                ->where('sequence', $ctx['sequence'])
-                                ->orderBy('id', 'desc')
-                                ->limit(2)
-                                ->get()
-                                ->map(function ($record) {
-                                    return [
-                                        'rid' => $record->id,
-                                        'script_id' => $record->script_id,
-                                        'sequence' => $record->sequence,
-                                        'role' => $record->role,
-                                        'content' => $record->content,
-                                        'created_at' => $record->created_at,
-                                    ];
-                                })
-                                ->reverse()
-                                ->values()
-                                ->toArray();
+                            // 两条对话记录（user/assistant）在同一事务内写入，保证原子性
+                            $chunk['records'] = DB::transaction(function () use ($ctx, $fullContent, $now) {
+                                DB::table('mp_script_records')->insert([
+                                    [
+                                        'uid' => $ctx['uid'],
+                                        'script_id' => $ctx['script_id'],
+                                        'sequence' => $ctx['sequence'],
+                                        'role' => 'user',
+                                        'content' => $ctx['original_prompt'],
+                                        'created_at' => $now,
+                                        'updated_at' => $now,
+                                    ],
+                                    [
+                                        'uid' => $ctx['uid'],
+                                        'script_id' => $ctx['script_id'],
+                                        'sequence' => $ctx['sequence'],
+                                        'role' => 'assistant',
+                                        'content' => $fullContent,
+                                        'created_at' => $now,
+                                        'updated_at' => $now,
+                                    ],
+                                ]);
+
+                                return DB::table('mp_script_records')
+                                    ->where('uid', $ctx['uid'])
+                                    ->where('script_id', $ctx['script_id'])
+                                    ->where('sequence', $ctx['sequence'])
+                                    ->orderBy('id', 'desc')
+                                    ->limit(2)
+                                    ->get()
+                                    ->map(function ($record) {
+                                        return [
+                                            'rid' => $record->id,
+                                            'script_id' => $record->script_id,
+                                            'sequence' => $record->sequence,
+                                            'role' => $record->role,
+                                            'content' => $record->content,
+                                            'created_at' => $record->created_at,
+                                        ];
+                                    })
+                                    ->reverse()
+                                    ->values()
+                                    ->toArray();
+                            });
                         } catch (\Exception $e) {
                             dLog('deepseek')->error('保存剧本对话记录失败: ' . $e->getMessage());
                             $chunk['records'] = [];
@@ -1084,15 +1093,28 @@ class DeepSeekService
             }
 
             if ($finished) {
-                if (!empty($ctx['generate_task_id'])) {
-                    DB::table('mp_script_generate_tasks')->where('id', $ctx['generate_task_id'])->update([
-                        'status'       => 'success',
-                        'result'       => $fullContent,
-                        'completed_at' => date('Y-m-d H:i:s'),
-                        'updated_at'   => date('Y-m-d H:i:s'),
-                    ]);
+                // 任务状态与扣费在同一事务内：扣费失败则任务状态一并回滚，由外层 catch 置为 failed
+                DB::transaction(function () use ($ctx, $fullContent, $usage) {
+                    if (!empty($ctx['generate_task_id'])) {
+                        DB::table('mp_script_generate_tasks')->where('id', $ctx['generate_task_id'])->update([
+                            'status'       => 'success',
+                            'result'       => $fullContent,
+                            'completed_at' => date('Y-m-d H:i:s'),
+                            'updated_at'   => date('Y-m-d H:i:s'),
+                        ]);
+                    }
+                    if ($ctx['has_valid_script'] && $fullContent !== '') {
+                        $this->chargeChatSuccess($ctx['uid'], $this->pointsService->getTokensFromUsage($usage), [
+                            'model' => $ctx['model'],
+                            'script_id' => $ctx['script_id'],
+                            'sequence' => $ctx['sequence'],
+                            'source' => 'newGenerateText',
+                        ]);
+                    }
+                });
 
-                    // JSON 输出结果的安全校验/修复与观察（不改变 result 与 SSE 推送内容）
+                // JSON 输出结果的安全校验/修复与观察（事务外执行，不改变 result 与 SSE 推送内容）
+                if (!empty($ctx['generate_task_id'])) {
                     $this->guardScriptGenerateJsonResult(
                         (int)$ctx['generate_task_id'],
                         (int)($ctx['script_id'] ?? 0),
@@ -1101,14 +1123,6 @@ class DeepSeekService
                         (string)($ctx['response_format'] ?? '')
                     );
                 }
-                if ($ctx['has_valid_script'] && $fullContent !== '') {
-                    $this->chargeChatSuccess($ctx['uid'], $this->pointsService->getTokensFromUsage($usage), [
-                        'model' => $ctx['model'],
-                        'script_id' => $ctx['script_id'],
-                        'sequence' => $ctx['sequence'],
-                        'source' => 'newGenerateText',
-                    ]);
-                }
             }
         } catch (\Throwable $e) {
             if (!empty($ctx['generate_task_id'])) {
@@ -1230,6 +1244,175 @@ class DeepSeekService
     }
 
     /**
+     * 创建剧本资产生成异步任务（stream=1 + script_id 走异步任务模式）
+     *
+     * @param array $data 原请求参数
+     * @return int 任务ID
+     */
+    public function createScriptGenerateAsyncTask(array $data): int
+    {
+        $script_id = (int)getProp($data, 'script_id', 0);
+        if ($script_id <= 0) {
+            Utils::throwError('20003:缺少剧本ID');
+        }
+
+        $uid = (int)Site::getUid();
+
+        // 与同步逻辑一致：同一剧本存在生成中任务时直接拒绝
+        $runningTask = DB::table('mp_script_generate_tasks')
+            ->where('uid', $uid)
+            ->where('script_id', $script_id)
+            ->orderByDesc('id')
+            ->first();
+        if ($runningTask && in_array(getProp($runningTask, 'status'), ['pending', 'processing'])) {
+            Utils::throwError('20003:该剧本资产正在生成中，请稍后重试');
+        }
+
+        // 前置校验剧本存在且内容非空，避免创建无效任务
+        $script = DB::table('mp_scripts')
+            ->where('id', $script_id)
+            ->where('is_deleted', 0)
+            ->first();
+        if (!$script) {
+            Utils::throwError('20003:剧本不存在或已删除');
+        }
+        if (empty($script->content)) {
+            Utils::throwError('20003:剧本内容为空');
+        }
+
+        // newGenerateText 请求不含上传文件对象，images/image_urls 以字符串形式传入，可直接序列化
+        $paramsJson = json_encode($data, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
+        if ($paramsJson === false) {
+            Utils::throwError('20003:请求参数序列化失败');
+        }
+
+        $now = date('Y-m-d H:i:s');
+        return (int)DB::table('mp_script_generate_tasks')->insertGetId([
+            'uid'            => $uid,
+            'script_id'      => $script_id,
+            'sequence'       => (int)getProp($data, 'sequence', 0),
+            'status'         => 'pending',
+            'prompt'         => (string)getProp($data, 'prompt', ''),
+            'request_params' => $paramsJson,
+            'started_at'     => $now,
+            'created_at'     => $now,
+            'updated_at'     => $now,
+        ]);
+    }
+
+    /**
+     * 供 SSE 轮询：查询任务状态与增量 chunks
+     *
+     * @param int $taskId
+     * @param int $uid 当前登录用户，防越权
+     * @param int $afterSeq 已推送的最大 seq
+     * @return array{found:bool, task:?object, chunks:array, status:?string, error_message:?string}
+     */
+    public function pollScriptGenerateTask(int $taskId, int $uid, int $afterSeq = -1): array
+    {
+        $task = DB::table('mp_script_generate_tasks')
+            ->where('id', $taskId)
+            ->where('uid', $uid)
+            ->first();
+        if (!$task) {
+            return ['found' => false, 'task' => null, 'chunks' => [], 'status' => null, 'error_message' => null];
+        }
+
+        $rows = DB::table('mp_script_generate_task_chunks')
+            ->where('task_id', $taskId)
+            ->where('seq', '>', $afterSeq)
+            ->orderBy('seq')
+            ->get();
+
+        $chunks = [];
+        foreach ($rows as $row) {
+            $payload = json_decode((string)$row->payload, true);
+            if (is_array($payload)) {
+                $chunks[] = $payload;
+            }
+        }
+
+        return [
+            'found'         => true,
+            'task'          => $task,
+            'chunks'        => $chunks,
+            'status'        => (string)$task->status,
+            'error_message' => (string)($task->error_message ?? ''),
+        ];
+    }
+
+    /**
+     * 异步 worker 执行入口：抢占 pending 任务并流式请求大模型
+     *
+     * 由 text:generate:run 命令调用；worker 内完成 chunks 落库，模型结束后
+     * 由流式包装（wrapNewGenerateTextStream）完成 records/扣费/任务状态/json 修复收尾。
+     *
+     * @param int $taskId
+     * @return array{ok:bool, task_id:int, status:?string, seq:int, error:?string}
+     */
+    public function runScriptGenerateTaskAsync(int $taskId): array
+    {
+        // 原子抢占：只允许 pending -> processing，避免重复执行
+        $claimed = DB::table('mp_script_generate_tasks')
+            ->where('id', $taskId)
+            ->where('status', 'pending')
+            ->update([
+                'status'     => 'processing',
+                'started_at' => date('Y-m-d H:i:s'),
+                'updated_at' => date('Y-m-d H:i:s'),
+            ]);
+        if (!$claimed) {
+            return ['ok' => false, 'task_id' => $taskId, 'status' => null, 'seq' => 0, 'error' => '任务已被处理或状态异常'];
+        }
+
+        $task = DB::table('mp_script_generate_tasks')->where('id', $taskId)->first();
+        if (!$task) {
+            return ['ok' => false, 'task_id' => $taskId, 'status' => null, 'seq' => 0, 'error' => '任务不存在'];
+        }
+
+        $params = json_decode((string)getProp($task, 'request_params', ''), true);
+        if (!is_array($params)) {
+            DB::table('mp_script_generate_tasks')->where('id', $taskId)->update([
+                'status'        => 'failed',
+                'error_message' => '任务请求参数缺失',
+                'completed_at'  => date('Y-m-d H:i:s'),
+                'updated_at'    => date('Y-m-d H:i:s'),
+            ]);
+            return ['ok' => false, 'task_id' => $taskId, 'status' => 'failed', 'seq' => 0, 'error' => '任务请求参数缺失'];
+        }
+
+        // 兼容旧参数未带 stream 的情况
+        $params['stream'] = 1;
+
+        $seq = 0;
+        try {
+            $generator = $this->newGenerateTextStream($params, $task);
+            foreach ($generator as $chunk) {
+                DB::table('mp_script_generate_task_chunks')->insert([
+                    'task_id'    => $taskId,
+                    'seq'        => $seq,
+                    'chunk_type' => (string)getProp($chunk, 'type', 'content'),
+                    'payload'    => json_encode($chunk, JSON_UNESCAPED_UNICODE | JSON_INVALID_UTF8_SUBSTITUTE),
+                    'created_at' => date('Y-m-d H:i:s'),
+                ]);
+                $seq++;
+            }
+
+            $finalStatus = (string)DB::table('mp_script_generate_tasks')->where('id', $taskId)->value('status');
+            return ['ok' => $finalStatus === 'success', 'task_id' => $taskId, 'status' => $finalStatus, 'seq' => $seq, 'error' => null];
+        } catch (\Throwable $e) {
+            // 兜底：正常流程中 wrap 会自行把任务置为 failed，这里防止未知异常导致任务卡死
+            DB::table('mp_script_generate_tasks')->where('id', $taskId)->update([
+                'status'        => 'failed',
+                'error_message' => $e->getMessage(),
+                'completed_at'  => date('Y-m-d H:i:s'),
+                'updated_at'    => date('Y-m-d H:i:s'),
+            ]);
+            return ['ok' => false, 'task_id' => $taskId, 'status' => 'failed', 'seq' => $seq, 'error' => $e->getMessage()];
+        }
+    }
+
+    /**
      * 通用文生文方法（非流式版本）- 支持多模型、图片输入、JSON输出和模板提示词
      * 
      * @param array $data 请求参数