参考文献模型校对换阿里云百炼，完善各种校对细节

application/common/mq/ReferenceCheckArticleWorker.php

@@ -66,6 +66,8 @@ class ReferenceCheckArticleWorker
 
     $owned = true;
     $finished = false;
+    $attemptedCheckIds = [];
+    $idleRecovered = false;
     try {
       // 续跑时强制把卡死行收回 pending，已完成行不动
       $this->svc->recoverQueueRowsForArticle($pArticleId, $resume);
@@ -80,29 +82,81 @@ class ReferenceCheckArticleWorker
       );
 
       while (true) {
+        // 组长卡在 RUNNING 时先收回，避免 worker 误跑组员报 leader not finished
+        $this->svc->recoverStuckRunningPendingRows($pArticleId, 180);
+
         $row = $this->fetchNextPendingRow($pArticleId);
         if (empty($row)) {
+          if (!$idleRecovered) {
+            $n = $this->svc->recoverAllRunningPendingRows($pArticleId);
+            if ($n > 0) {
+              $idleRecovered = true;
+              $this->svc->log(
+                'ReferenceCheckArticleWorker recovered ' . $n . ' running pending rows, retry fetch'
+              );
+              continue;
+            }
+          }
           break;
         }
+        $idleRecovered = false;
         $checkId = $this->svc->resolveCheckRowId($row);
         if ($checkId <= 0) {
           continue;
         }
+        if (!$this->svc->shouldProcessRelevanceRowNow($row)) {
+          if ($this->svc->recoverStuckGroupLeaderForRow($row)) {
+            $this->svc->log(
+              'ReferenceCheckArticleWorker reset stuck group leader for ref='
+              . intval($row['reference_no'])
+            );
+            continue;
+          }
+          $this->svc->log(
+            'ReferenceCheckArticleWorker skip non-leader check_id=' . $checkId
+            . ' ref=' . intval($row['reference_no'])
+          );
+          continue;
+        }
+        // 同一批消息内每个 check_id 只尝试一次，避免大联合组部分成功后死循环
+        if (isset($attemptedCheckIds[$checkId])) {
+          $this->svc->log('ReferenceCheckArticleWorker stop re-entry check_id=' . $checkId);
+          break;
+        }
+        $attemptedCheckIds[$checkId] = true;
         $this->processOneRow($checkId, $row, $trigger === 'recheck_pending_only');
         // 每条结束后刷新批次心跳，长文不会被误判为僵尸
         $this->touchBatch($batchId);
       }
 
       $stats = $this->summarizeArticleCheckStats($pArticleId);
-      $this->finalizeBatch($batchId, $stats['done'], $stats['failed'], $stats['total']);
-      $finished = true;
-      $this->svc->log(
-        'ReferenceCheckArticleWorker done p_article_id=' . $pArticleId
-        . ' batch_id=' . $batchId
-        . ' done=' . $stats['done']
-        . ' failed=' . $stats['failed']
-      );
-      $this->publishNextWaitingBatch();
+      if (intval($stats['pending']) > 0) {
+        // 分块落库后仍有缺口：回 WAITING 再投递，下轮只补 pending
+        Db::name('article_reference_relevance_check_batch')->where('id', intval($batchId))->update([
+          'batch_status' => self::BATCH_WAITING,
+          'done_count'   => intval($stats['done']),
+          'failed_count' => intval($stats['failed']),
+          'updated_at'   => date('Y-m-d H:i:s'),
+        ]);
+        $finished = true;
+        $this->svc->log(
+          'ReferenceCheckArticleWorker defer incomplete p_article_id=' . $pArticleId
+          . ' batch_id=' . $batchId
+          . ' pending=' . $stats['pending']
+          . ' done=' . $stats['done']
+        );
+        (new ReferenceCheckMqPublisher())->publishArticleStart($pArticleId, $batchId, $trigger);
+      } else {
+        $this->finalizeBatch($batchId, $stats['done'], $stats['failed'], $stats['total']);
+        $finished = true;
+        $this->svc->log(
+          'ReferenceCheckArticleWorker done p_article_id=' . $pArticleId
+          . ' batch_id=' . $batchId
+          . ' done=' . $stats['done']
+          . ' failed=' . $stats['failed']
+        );
+        $this->publishNextWaitingBatch();
+      }
     } catch (\Throwable $e) {
       // 异常不 finalize：保持 RUNNING，靠心跳超时后由后续消息断点续跑
       if ($owned) {
@@ -278,12 +332,22 @@ class ReferenceCheckArticleWorker
 
   private function fetchNextPendingRow($pArticleId)
   {
-    return Db::name('article_reference_relevance_check_result')
+    $rows = Db::name('article_reference_relevance_check_result')
       ->where('p_article_id', intval($pArticleId))
       ->where('queue_status', ReferenceRelevanceCheckService::QUEUE_PENDING)
       ->where('status', ReferenceRelevanceCheckService::RECORD_PENDING)
       ->order('reference_no asc,am_id asc,text_start asc,id asc')
-      ->find();
+      ->limit(100)
+      ->select();
+    if (empty($rows)) {
+      return null;
+    }
+    foreach ($rows as $row) {
+      if ($this->svc->shouldProcessRelevanceRowNow($row)) {
+        return $row;
+      }
+    }
+    return null;
   }
 
   /**
@@ -306,13 +370,43 @@ class ReferenceCheckArticleWorker
     $retryCount = intval(isset($row['retry_count']) ? $row['retry_count'] : 0);
     try {
       $this->svc->runCheckOnce($checkId, $skipLiteratureFetch);
-      $this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_COMPLETED, $retryCount);
+      DbReconnectHelper::ensure();
+      $fresh = Db::name('article_reference_relevance_check_result')->where('id', intval($checkId))->find();
+      if (empty($fresh)) {
+        return 'skip';
+      }
+      $st = intval($fresh['status']);
+      if ($st === ReferenceRelevanceCheckService::RECORD_COMPLETED) {
+        $this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_COMPLETED, $retryCount);
+        return 'ok';
+      }
+      if ($st === ReferenceRelevanceCheckService::RECORD_FAILED) {
+        $this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_FAILED, $retryCount);
+        return 'failed';
+      }
+      // 分块部分成功：本行仍 pending，留给后续消息补跑
+      $this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_PENDING, $retryCount);
       return 'ok';
     } catch (\Exception $e) {
       $this->svc->log('ReferenceCheckArticleWorker check_id=' . $checkId . ' err=' . $e->getMessage());
       DbReconnectHelper::ensure();
+      // 联合组组员被提前领取：不算失败，交还 pending 等组长
+      if (intval($e->getCode()) === 9001
+          || strpos($e->getMessage(), 'Citation group leader not finished') !== false) {
+        $this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_PENDING, $retryCount);
+        return 'skip';
+      }
+      if ($this->svc->isRelevanceLlmFailureMessage($e->getMessage())) {
+        $this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_PENDING, $retryCount);
+        return 'ok';
+      }
       try {
         $fresh = Db::name('article_reference_relevance_check_result')->where('id', intval($checkId))->find();
+        if (!empty($fresh) && intval($fresh['status']) === ReferenceRelevanceCheckService::RECORD_COMPLETED) {
+          // 异常前已有分块落库成功，保留成果，本行按完成处理
+          $this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_COMPLETED, $retryCount);
+          return 'ok';
+        }
         if (!empty($fresh) && intval($fresh['status']) === ReferenceRelevanceCheckService::RECORD_FAILED) {
           if (intval($fresh['queue_status']) !== ReferenceRelevanceCheckService::QUEUE_FAILED) {
             $this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_FAILED, $retryCount);
@@ -321,7 +415,16 @@ class ReferenceCheckArticleWorker
         }
         $groupRows = !empty($fresh) ? $this->svc->findCitationGroupRowsForWorker($fresh) : [];
         if (!empty($groupRows)) {
-          $this->svc->failGroupWithQueue($groupRows, $e->getMessage(), $retryCount);
+          // 只失败仍未完成的行，已 completed 的分块结果保留
+          $incomplete = [];
+          foreach ($groupRows as $gr) {
+            if (intval($gr['status']) !== ReferenceRelevanceCheckService::RECORD_COMPLETED) {
+              $incomplete[] = $gr;
+            }
+          }
+          if (!empty($incomplete)) {
+            $this->svc->failGroupWithQueue($incomplete, $e->getMessage(), $retryCount);
+          }
         } else {
           $this->svc->updateCheckResult($checkId, [
             'status'    => ReferenceRelevanceCheckService::RECORD_FAILED,