修复工单PHP CLI同步问题 系统设置可以设置并发阈值

_php_code/Huojian.php

@@ -172,7 +172,7 @@ class Huojian extends AbstractScrmSpider
     {
         $unifiedData = $this->unifiedData;
 
-        $unifiedData->todayNewCount = (int) ($allListPagesData[0]['data']['counterWorker']['newTodayFriend'] ?? 0);
+        $unifiedData->todayNewCount = (int) ($allListPagesData[0]['data']['counterWorker']['newTodayFriend'] ?? 0) + (int) ($allListPagesData[0]['data']['counterWorker']['newRemovedTodayFriend'] ?? 0);
         
         $count = 0;
         foreach ($allListPagesData as $pageRaw) {

_php_code/index.php

@@ -85,7 +85,7 @@ try {
     // 长链：快速验证业务页拦截（推荐先测通）
     // $pageUrl = 'https://user.a2c.chat/visitors/counter/share?id=1b2cf9a91e2647c185b252e723c871de';
     // 短链：与客户后台一致，上线前必须回归
-    $pageUrl = 'https://yyk.ink/91SEWrL';
+    $pageUrl = 'https://yyk.ink/3xj1B2';
     $username = "";
     $password = "";
     $spider = (new A2c($pageUrl, $username, $password))->setVerbose(true);

application/admin/command/Install/split_config_node_channel.sql

@@ -0,0 +1,19 @@
+-- 双通道定时同步：Node 通道上限与候选池配置（已有环境增量执行）
+SET NAMES utf8mb4;
+
+INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`)
+SELECT 'split_sync_max_per_cron', 'split', 'Node通道单次同步上限', '每分钟 cron 在 Node 通道最多处理几条到期工单（仅影响需调用 Node 的类型）；纯 PHP 直连类型不受此限制', 'number', '', '5', '', '', '', NULL
+FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_max_per_cron' LIMIT 1);
+
+INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`)
+SELECT 'split_sync_node_candidate_multiplier', 'split', 'Node通道候选池倍数', 'Node 候选工单数 = 单次上限 × 本倍数，用于公平轮转', 'number', '', '10', '', '', '', NULL
+FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_node_candidate_multiplier' LIMIT 1);
+
+UPDATE `fa_config`
+SET `title` = 'Node通道单次同步上限',
+    `tip` = '每分钟 cron 在 Node 通道最多处理几条到期工单（仅影响需调用 Node 的类型）；纯 PHP 直连类型不受此限制'
+WHERE `name` = 'split_sync_max_per_cron';
+
+UPDATE `fa_config`
+SET `value` = '5'
+WHERE `name` = 'split_sync_max_per_cron' AND (`value` = '' OR `value` = '2');

application/common/library/scrm/spider/HuojianSpider.php

@@ -80,7 +80,7 @@ class HuojianSpider extends AbstractScrmSpider
     protected function parseToUnifiedData($detailData, array $allListPagesData): UnifiedScrmData
     {
         $unifiedData = $this->unifiedData;
-        $unifiedData->todayNewCount = (int) ($allListPagesData[0]['data']['counterWorker']['newTodayFriend'] ?? 0);
+        $unifiedData->todayNewCount = (int) ($allListPagesData[0]['data']['counterWorker']['newTodayFriend'] ?? 0) + (int) ($allListPagesData[0]['data']['counterWorker']['newRemovedTodayFriend'] ?? 0);
         $count = 0;
         foreach ($allListPagesData as $pageRaw) {
             $records = $pageRaw['data']['counterCsAccountVo'] ?? [];

application/common/service/SplitScrmSpiderFactory.php

@@ -4,6 +4,7 @@ declare(strict_types=1);
 
 namespace app\common\service;
 
+use app\common\library\scrm\AbstractScrmSpider;
 use app\common\library\scrm\ScrmSpiderInterface;
 use app\common\library\scrm\spider\A2cSpider;
 use app\common\library\scrm\spider\ChatknowSpider;
@@ -59,6 +60,49 @@ class SplitScrmSpiderFactory
         return array_keys(self::MAP);
     }
 
+    /**
+     * 同步是否依赖 Node Headless（纯 PHP cURL 等为 false）
+     */
+    public static function requiresNode(string $ticketType): bool
+    {
+        $class = self::resolveClass($ticketType);
+        if ($class === null) {
+            return true;
+        }
+
+        return is_subclass_of($class, AbstractScrmSpider::class);
+    }
+
+    /**
+     * @return list<string>
+     */
+    public static function listDirectSyncTypes(): array
+    {
+        $types = [];
+        foreach (self::listSupportedTypes() as $ticketType) {
+            if (!self::requiresNode($ticketType)) {
+                $types[] = $ticketType;
+            }
+        }
+
+        return $types;
+    }
+
+    /**
+     * @return list<string>
+     */
+    public static function listNodeSyncTypes(): array
+    {
+        $types = [];
+        foreach (self::listSupportedTypes() as $ticketType) {
+            if (self::requiresNode($ticketType)) {
+                $types[] = $ticketType;
+            }
+        }
+
+        return $types;
+    }
+
     /**
      * @return ScrmSpiderInterface|null
      */

application/common/service/SplitSyncConfigService.php

@@ -57,11 +57,23 @@ class SplitSyncConfigService
     {
         $value = self::getConfigValue('split_sync_max_per_cron');
         if ($value === '') {
-            return 2;
+            return 5;
         }
         return max(1, min(20, (int) $value));
     }
 
+    /**
+     * Node 通道候选池倍数（候选数 = maxPerRun × 本值）
+     */
+    public static function getNodeCandidatePoolMultiplier(): int
+    {
+        $value = self::getConfigValue('split_sync_node_candidate_multiplier');
+        if ($value === '') {
+            return 10;
+        }
+        return max(3, min(50, (int) $value));
+    }
+
     /**
      * 全局 cron 锁过期秒数，防止异常退出后永久占锁
      */
@@ -86,6 +98,53 @@ class SplitSyncConfigService
         return max(0, (int) $value);
     }
 
+    /**
+     * 后台投递 / Cron 使用的 PHP CLI 可执行文件路径
+     *
+     * Web(FPM) 环境下 PHP_BINARY 指向 php-fpm，不能用于 think 命令。
+     */
+    public static function getCliPhpBinary(): string
+    {
+        $configured = trim(self::getConfigValue('split_sync_cli_php'));
+        if ($configured !== '' && self::isUsableCliPhp($configured)) {
+            return $configured;
+        }
+
+        if (defined('PHP_BINDIR') && PHP_BINDIR !== '') {
+            $candidate = rtrim(PHP_BINDIR, '/\\') . DIRECTORY_SEPARATOR . 'php';
+            if (self::isUsableCliPhp($candidate)) {
+                return $candidate;
+            }
+        }
+
+        if (PHP_SAPI === 'cli' && defined('PHP_BINARY') && PHP_BINARY !== '' && self::isUsableCliPhp(PHP_BINARY)) {
+            return PHP_BINARY;
+        }
+
+        foreach (['/usr/bin/php', '/usr/local/bin/php'] as $candidate) {
+            if (self::isUsableCliPhp($candidate)) {
+                return $candidate;
+            }
+        }
+
+        return 'php';
+    }
+
+    /**
+     * 判断路径是否为可用的 PHP CLI（排除 php-fpm）
+     */
+    private static function isUsableCliPhp(string $path): bool
+    {
+        if (stripos($path, 'fpm') !== false) {
+            return false;
+        }
+        if ($path === 'php') {
+            return true;
+        }
+
+        return is_file($path) && is_executable($path);
+    }
+
     private static function getConfigValue(string $name): string
     {
         $site = Config::get('site.' . $name);

application/common/service/SplitTicketSyncDispatchService.php

@@ -51,6 +51,7 @@ class SplitTicketSyncDispatchService
             'queued'  => $queued,
             'skipped' => $skipped,
             'failed'  => $failed,
+            'phpCli'  => $php,
         ]);
 
         return [
@@ -130,13 +131,43 @@ class SplitTicketSyncDispatchService
 
     private function resolvePhpBinary(): string
     {
-        if (defined('PHP_BINARY') && PHP_BINARY !== '') {
+        if (method_exists(SplitSyncConfigService::class, 'getCliPhpBinary')) {
-            return PHP_BINARY;
+            return SplitSyncConfigService::getCliPhpBinary();
         }
 
+        return $this->resolvePhpBinaryFallback();
+    }
+
+    /**
+     * 当 SplitSyncConfigService 未部署 getCliPhpBinary 时的兜底解析
+     */
+    private function resolvePhpBinaryFallback(): string
+    {
+        if (defined('PHP_BINDIR') && PHP_BINDIR !== '') {
+            $candidate = rtrim(PHP_BINDIR, '/\\') . DIRECTORY_SEPARATOR . 'php';
+            if ($this->isUsableCliPhp($candidate)) {
+                return $candidate;
+            }
+        }
+        foreach (['/usr/bin/php', '/usr/local/bin/php'] as $candidate) {
+            if ($this->isUsableCliPhp($candidate)) {
+                return $candidate;
+            }
+        }
         return 'php';
     }
 
+    private function isUsableCliPhp(string $path): bool
+    {
+        if (stripos($path, 'fpm') !== false) {
+            return false;
+        }
+        if ($path === 'php') {
+            return true;
+        }
+        return is_file($path) && is_executable($path);
+    }
+
     private function resolveThinkScript(): string
     {
         $root = $this->resolveRootPath();

application/common/service/SplitTicketSyncService.php

@@ -80,131 +80,49 @@ class SplitTicketSyncService
     }
 
     /**
-     * 扫描到期工单并同步（限流 + Node 队列感知）
+     * 扫描到期工单并同步（直连 PHP 与 Node 双通道）
      */
     public function syncDueTickets(): int
     {
-        $maxPerRun = SplitSyncConfigService::getMaxTicketsPerCronRun();
         $autoSyncTypes = $this->resolveAutoSyncTicketTypes();
-        $failThreshold = SplitSyncConfigService::getFailPauseThreshold();
-
         if ($autoSyncTypes === []) {
             SplitTicketSyncLogger::logCron('scan skip', ['reason' => 'no auto-sync ticket types']);
             return 0;
         }
 
-        if (!SplitNodeHealthService::canAcceptWork()) {
+        $failThreshold = SplitSyncConfigService::getFailPauseThreshold();
-            SplitTicketSyncLogger::logCron('scan skip', [
+        $directTypes = array_values(array_intersect(
-                'reason' => 'node queue busy',
+            $autoSyncTypes,
-                'queue'  => SplitNodeHealthService::getQueueSnapshot(),
+            SplitScrmSpiderFactory::listDirectSyncTypes()
-            ]);
+        ));
-            return 0;
+        $nodeTypes = array_values(array_intersect(
-        }
+            $autoSyncTypes,
+            SplitScrmSpiderFactory::listNodeSyncTypes()
+        ));
 
-        $query = Ticket::where('status', 'normal')->whereIn('ticket_type', $autoSyncTypes);
-        if ($failThreshold > 0) {
-            $query->where('sync_fail_count', '<', $failThreshold);
-        }
-        $list = $query->order('sync_time', 'asc')->order('id', 'asc')
-            ->limit($maxPerRun * 5)
-            ->select();
-        $list = $this->sortTicketsBySessionKey($list);
-
-        $candidateCount = count($list);
         SplitTicketSyncLogger::logCron('scan start', [
-            'candidateCount' => $candidateCount,
-            'maxPerRun'      => $maxPerRun,
             'autoSyncTypes' => $autoSyncTypes,
+            'directTypes'   => $directTypes,
+            'nodeTypes'     => $nodeTypes,
             'failThreshold' => $failThreshold,
             'nodeQueue'     => SplitNodeHealthService::getQueueSnapshot(),
-            'groupedBy'      => 'sessionKey',
-        ]);
-        SplitTicketSyncLogger::log('cron', 'scan start', [
-            'candidateCount' => $candidateCount,
-            'maxPerRun'      => $maxPerRun,
-            'autoSyncTypes'  => $autoSyncTypes,
-            'nodeQueue'      => SplitNodeHealthService::getQueueSnapshot(),
         ]);
 
-        /** @var list<array<string, mixed>> $processed */
+        $directResult = $this->runDirectCronChannel($directTypes, $failThreshold);
-        $processed = [];
+        $nodeResult = $this->runNodeCronChannel($nodeTypes, $failThreshold);
-        /** @var list<array<string, mixed>> $skipped */
-        $skipped = [];
-        $count = 0;
-        $stoppedByNodeBusy = false;
 
-        foreach ($list as $index => $ticket) {
+        $processedCount = $directResult['processedCount'] + $nodeResult['processedCount'];
-            $ticketId = (int) $ticket['id'];
+        $processed = array_merge($directResult['processed'], $nodeResult['processed']);
-            $ticketUrl = (string) $ticket['ticket_url'];
+        $skipped = array_merge($directResult['skipped'], $nodeResult['skipped']);
-            $ticketType = (string) $ticket['ticket_type'];
+        $candidateIds = array_merge($directResult['candidateIds'], $nodeResult['candidateIds']);
-            $sessionKey = AntiBotConfigBuilder::resolveSessionKey($ticketType, $ticketUrl);
-
-            if ($count >= $maxPerRun) {
-                $reason = '本轮处理上限已满';
-                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $reason);
-                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $reason);
-                continue;
-            }
-
-            $skip = $this->shouldSkip($ticket);
-            if ($skip !== null) {
-                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skip);
-                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skip);
-                continue;
-            }
-
-            if (!SplitNodeHealthService::canAcceptWork()) {
-                $reason = 'Node 队列繁忙，停止本轮后续同步';
-                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $reason);
-                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $reason);
-                $stoppedByNodeBusy = true;
-                for ($i = $index + 1; $i < $candidateCount; $i++) {
-                    $remain = $list[$i];
-                    $remainReason = 'Node 队列繁忙，本轮未执行';
-                    $skipped[] = $this->buildCronTicketEntry(
-                        (int) $remain['id'],
-                        (string) $remain['ticket_type'],
-                        (string) $remain['ticket_url'],
-                        $remainReason
-                    );
-                }
-                SplitTicketSyncLogger::log('cron', 'node busy, stop batch', [
-                    'processed' => $count,
-                    'queue'     => SplitNodeHealthService::getQueueSnapshot(),
-                ]);
-                break;
-            }
-
-            $result = $this->syncOne($ticketId, false, true, 'auto');
-            if (!empty($result['skipped'])) {
-                $skipReason = (string) ($result['message'] ?? '已跳过');
-                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skipReason);
-                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skipReason);
-                continue;
-            }
-
-            $processed[] = [
-                'ticketId'   => $ticketId,
-                'ticketType' => $ticketType,
-                'ticketUrl'  => $ticketUrl,
-                'sessionKey' => $sessionKey,
-                'success'    => !empty($result['success']),
-                'message'    => (string) ($result['message'] ?? ''),
-            ];
-            $count++;
-        }
-
-        $candidateIds = array_map(static function ($row) {
-            return (int) (is_array($row) ? ($row['id'] ?? 0) : ($row['id'] ?? 0));
-        }, $list);
         $openNotSynced = $this->buildOpenNotSyncedAudit(
             $autoSyncTypes,
             $failThreshold,
-            $maxPerRun,
+            SplitSyncConfigService::getMaxTicketsPerCronRun(),
             $candidateIds,
             $processed,
             $skipped,
-            $stoppedByNodeBusy
+            $nodeResult['stoppedByNodeBusy']
         );
 
         $successCount = count(array_filter($processed, static function (array $row): bool {
@@ -213,21 +131,362 @@ class SplitTicketSyncService
         $failedCount = count($processed) - $successCount;
 
         SplitTicketSyncLogger::logCron('scan end', [
-            'processedCount' => $count,
+            'processedCount' => $processedCount,
             'successCount'   => $successCount,
             'failedCount'    => $failedCount,
             'skippedCount'   => count($skipped),
+            'directChannel'  => $directResult['summary'],
+            'nodeChannel'    => $nodeResult['summary'],
             'processed'      => $processed,
             'skipped'        => $skipped,
             'openNotSynced'  => $openNotSynced,
         ]);
         SplitTicketSyncLogger::log('cron', 'scan end', [
-            'processedCount' => $count,
+            'processedCount' => $processedCount,
             'successCount'   => $successCount,
             'failedCount'    => $failedCount,
         ]);
 
-        return $count;
+        return $processedCount;
+    }
+
+    /**
+     * 直连 PHP 通道：按周期同步，不占 Node 名额
+     *
+     * @param list<string> $directTypes
+     * @return array{
+     *   processedCount:int,
+     *   processed:list<array<string,mixed>>,
+     *   skipped:list<array<string,mixed>>,
+     *   candidateIds:list<int>,
+     *   summary:array<string,mixed>
+     * }
+     */
+    private function runDirectCronChannel(array $directTypes, int $failThreshold): array
+    {
+        if ($directTypes === []) {
+            return $this->emptyCronChannelResult('direct');
+        }
+
+        $list = $this->loadOpenTicketsForTypes($directTypes, $failThreshold, 0);
+        $candidateIds = $this->extractTicketIds($list);
+
+        SplitTicketSyncLogger::logCron('direct channel start', [
+            'types'           => $directTypes,
+            'candidateCount'  => count($list),
+        ]);
+
+        /** @var list<array<string, mixed>> $processed */
+        $processed = [];
+        /** @var list<array<string, mixed>> $skipped */
+        $skipped = [];
+        $count = 0;
+
+        foreach ($list as $ticket) {
+            $ticketId = (int) $ticket['id'];
+            $ticketType = (string) $ticket['ticket_type'];
+            $ticketUrl = (string) $ticket['ticket_url'];
+
+            $skip = $this->shouldSkip($ticket);
+            if ($skip !== null) {
+                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skip);
+                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skip, 'direct');
+                continue;
+            }
+
+            $result = $this->syncOne($ticketId, false, true, 'auto');
+            if (!empty($result['skipped'])) {
+                $skipReason = (string) ($result['message'] ?? '已跳过');
+                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skipReason);
+                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skipReason, 'direct');
+                continue;
+            }
+
+            $processed[] = [
+                'ticketId'   => $ticketId,
+                'ticketType' => $ticketType,
+                'ticketUrl'  => $ticketUrl,
+                'channel'    => 'direct',
+                'success'    => !empty($result['success']),
+                'message'    => (string) ($result['message'] ?? ''),
+            ];
+            $count++;
+        }
+
+        return [
+            'processedCount' => $count,
+            'processed'      => $processed,
+            'skipped'        => $skipped,
+            'candidateIds'   => $candidateIds,
+            'summary'        => [
+                'channel'        => 'direct',
+                'processedCount' => $count,
+                'skippedCount'   => count($skipped),
+            ],
+            'stoppedByNodeBusy' => false,
+        ];
+    }
+
+    /**
+     * Node 通道：限流 + 类型公平轮转
+     *
+     * @param list<string> $nodeTypes
+     * @return array{
+     *   processedCount:int,
+     *   processed:list<array<string,mixed>>,
+     *   skipped:list<array<string,mixed>>,
+     *   candidateIds:list<int>,
+     *   summary:array<string,mixed>,
+     *   stoppedByNodeBusy:bool
+     * }
+     */
+    private function runNodeCronChannel(array $nodeTypes, int $failThreshold): array
+    {
+        if ($nodeTypes === []) {
+            return $this->emptyCronChannelResult('node');
+        }
+
+        $maxPerRun = SplitSyncConfigService::getMaxTicketsPerCronRun();
+        $poolMultiplier = SplitSyncConfigService::getNodeCandidatePoolMultiplier();
+        $candidateLimit = $maxPerRun * $poolMultiplier;
+
+        $pool = $this->loadOpenTicketsForTypes($nodeTypes, $failThreshold, $candidateLimit);
+        $candidateIds = $this->extractTicketIds($pool);
+        $duePool = [];
+        foreach ($pool as $ticket) {
+            if ($this->shouldSkip($ticket) === null) {
+                $duePool[] = $ticket;
+            }
+        }
+        $picked = $this->pickFairNodeTickets($duePool, $maxPerRun);
+        $picked = $this->sortTicketsBySessionKey($picked);
+        $pickedIdMap = array_fill_keys($this->extractTicketIds($picked), true);
+
+        SplitTicketSyncLogger::logCron('node channel start', [
+            'types'          => $nodeTypes,
+            'maxPerRun'      => $maxPerRun,
+            'poolCount'      => count($pool),
+            'dueCount'       => count($duePool),
+            'pickedCount'    => count($picked),
+            'nodeQueue'      => SplitNodeHealthService::getQueueSnapshot(),
+        ]);
+
+        if (!SplitNodeHealthService::canAcceptWork()) {
+            SplitTicketSyncLogger::logCron('node channel skip', [
+                'reason' => 'node queue busy',
+                'queue'  => SplitNodeHealthService::getQueueSnapshot(),
+            ]);
+            $skipped = [];
+            foreach ($picked as $ticket) {
+                $skipped[] = $this->buildCronTicketEntry(
+                    (int) $ticket['id'],
+                    (string) $ticket['ticket_type'],
+                    (string) $ticket['ticket_url'],
+                    'Node 队列繁忙，本轮未执行'
+                );
+            }
+            foreach ($pool as $ticket) {
+                $ticketId = (int) $ticket['id'];
+                if (isset($pickedIdMap[$ticketId])) {
+                    continue;
+                }
+                $skip = $this->shouldSkip($ticket);
+                if ($skip !== null) {
+                    continue;
+                }
+                $skipped[] = $this->buildCronTicketEntry(
+                    $ticketId,
+                    (string) $ticket['ticket_type'],
+                    (string) $ticket['ticket_url'],
+                    sprintf('未进入本轮 Node 候选(前%d条到期工单公平轮转)', $candidateLimit)
+                );
+            }
+
+            return [
+                'processedCount'    => 0,
+                'processed'         => [],
+                'skipped'           => $skipped,
+                'candidateIds'      => $candidateIds,
+                'summary'           => [
+                    'channel'        => 'node',
+                    'processedCount' => 0,
+                    'skippedCount'   => count($skipped),
+                    'reason'         => 'node queue busy',
+                ],
+                'stoppedByNodeBusy' => true,
+            ];
+        }
+
+        /** @var list<array<string, mixed>> $processed */
+        $processed = [];
+        /** @var list<array<string, mixed>> $skipped */
+        $skipped = [];
+        $count = 0;
+        $stoppedByNodeBusy = false;
+
+        foreach ($pool as $ticket) {
+            $ticketId = (int) $ticket['id'];
+            $ticketType = (string) $ticket['ticket_type'];
+            $ticketUrl = (string) $ticket['ticket_url'];
+
+            if (!isset($pickedIdMap[$ticketId])) {
+                $skip = $this->shouldSkip($ticket);
+                if ($skip !== null) {
+                    continue;
+                }
+                $skipped[] = $this->buildCronTicketEntry(
+                    $ticketId,
+                    $ticketType,
+                    $ticketUrl,
+                    sprintf('未进入本轮 Node 候选(前%d条到期工单公平轮转)', $candidateLimit)
+                );
+                continue;
+            }
+
+            if (!SplitNodeHealthService::canAcceptWork()) {
+                $stoppedByNodeBusy = true;
+                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, 'Node 队列繁忙，停止本轮后续同步');
+                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, 'Node 队列繁忙，停止本轮后续同步', 'node');
+                continue;
+            }
+
+            $sessionKey = AntiBotConfigBuilder::resolveSessionKey($ticketType, $ticketUrl);
+            $result = $this->syncOne($ticketId, false, true, 'auto');
+            if (!empty($result['skipped'])) {
+                $skipReason = (string) ($result['message'] ?? '已跳过');
+                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skipReason);
+                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skipReason, 'node');
+                continue;
+            }
+
+            $processed[] = [
+                'ticketId'   => $ticketId,
+                'ticketType' => $ticketType,
+                'ticketUrl'  => $ticketUrl,
+                'sessionKey' => $sessionKey,
+                'channel'    => 'node',
+                'success'    => !empty($result['success']),
+                'message'    => (string) ($result['message'] ?? ''),
+            ];
+            $count++;
+        }
+
+        return [
+            'processedCount'    => $count,
+            'processed'         => $processed,
+            'skipped'           => $skipped,
+            'candidateIds'      => $candidateIds,
+            'summary'           => [
+                'channel'        => 'node',
+                'processedCount' => $count,
+                'skippedCount'   => count($skipped),
+                'maxPerRun'      => $maxPerRun,
+            ],
+            'stoppedByNodeBusy' => $stoppedByNodeBusy,
+        ];
+    }
+
+    /**
+     * @param list<string> $types
+     * @return list<Ticket>
+     */
+    private function loadOpenTicketsForTypes(array $types, int $failThreshold, int $limit): array
+    {
+        if ($types === []) {
+            return [];
+        }
+
+        $query = Ticket::where('status', 'normal')->whereIn('ticket_type', $types);
+        if ($failThreshold > 0) {
+            $query->where('sync_fail_count', '<', $failThreshold);
+        }
+        $query->order('sync_time', 'asc')->order('id', 'asc');
+        if ($limit > 0) {
+            $query->limit($limit);
+        }
+
+        $list = $query->select();
+        return $list instanceof \think\Collection ? $list->all() : (array) $list;
+    }
+
+    /**
+     * Node 通道：按工单类型公平轮转选取（每轮每类型优先取 1 条）
+     *
+     * @param list<Ticket> $dueTickets
+     * @return list<Ticket>
+     */
+    private function pickFairNodeTickets(array $dueTickets, int $maxPerRun): array
+    {
+        if ($dueTickets === [] || $maxPerRun <= 0) {
+            return [];
+        }
+
+        /** @var array<string, list<Ticket>> $byType */
+        $byType = [];
+        foreach ($dueTickets as $ticket) {
+            $type = (string) $ticket['ticket_type'];
+            $byType[$type][] = $ticket;
+        }
+
+        $typeKeys = array_keys($byType);
+        sort($typeKeys, SORT_STRING);
+
+        /** @var list<Ticket> $picked */
+        $picked = [];
+        while (count($picked) < $maxPerRun) {
+            $added = false;
+            foreach ($typeKeys as $type) {
+                if (count($picked) >= $maxPerRun) {
+                    break;
+                }
+                if ($byType[$type] === []) {
+                    continue;
+                }
+                $picked[] = array_shift($byType[$type]);
+                $added = true;
+            }
+            if (!$added) {
+                break;
+            }
+        }
+
+        return $picked;
+    }
+
+    /**
+     * @param list<Ticket|array<string,mixed>> $list
+     * @return list<int>
+     */
+    private function extractTicketIds(array $list): array
+    {
+        $ids = [];
+        foreach ($list as $row) {
+            $ids[] = (int) (is_array($row) ? ($row['id'] ?? 0) : ($row['id'] ?? 0));
+        }
+
+        return $ids;
+    }
+
+    /**
+     * @return array{
+     *   processedCount:int,
+     *   processed:list<array<string,mixed>>,
+     *   skipped:list<array<string,mixed>>,
+     *   candidateIds:list<int>,
+     *   summary:array<string,mixed>,
+     *   stoppedByNodeBusy:bool
+     * }
+     */
+    private function emptyCronChannelResult(string $channel): array
+    {
+        return [
+            'processedCount'    => 0,
+            'processed'         => [],
+            'skipped'           => [],
+            'candidateIds'      => [],
+            'summary'           => ['channel' => $channel, 'processedCount' => 0, 'skippedCount' => 0],
+            'stoppedByNodeBusy' => false,
+        ];
     }
 
     /**
@@ -466,13 +725,14 @@ class SplitTicketSyncService
         ];
     }
 
-    private function logCronCandidateSkipped(int $ticketId, string $ticketType, string $ticketUrl, string $reason): void
+    private function logCronCandidateSkipped(int $ticketId, string $ticketType, string $ticketUrl, string $reason, string $channel = 'legacy'): void
     {
         $ctx = [
             'ticketId'   => $ticketId,
             'ticketType' => $ticketType,
             'ticketUrl'  => $ticketUrl,
             'reason'     => $reason,
+            'channel'    => $channel,
         ];
         SplitTicketSyncLogger::logCron('candidate skipped', $ctx);
         SplitTicketSyncLogger::log('cron', 'candidate skipped', $ctx);
@@ -515,14 +775,15 @@ class SplitTicketSyncService
             ->select();
 
         $result = [];
-        foreach ($openList as $ticket) {
+        foreach ($openList as $ticketRow) {
-            $ticketId = (int) $ticket['id'];
+            $ticket = $ticketRow instanceof Ticket ? $ticketRow->toArray() : (array) $ticketRow;
-            if (isset($handledIds[$ticketId])) {
+            $ticketId = (int) ($ticket['id'] ?? 0);
+            if ($ticketId <= 0 || isset($handledIds[$ticketId])) {
                 continue;
             }
 
-            $ticketType = (string) $ticket['ticket_type'];
+            $ticketType = (string) ($ticket['ticket_type'] ?? '');
-            $ticketUrl = (string) $ticket['ticket_url'];
+            $ticketUrl = (string) ($ticket['ticket_url'] ?? '');
             $reason = $this->resolveOpenNotSyncedReason(
                 $ticket,
                 $autoSyncTypes,
@@ -568,7 +829,13 @@ class SplitTicketSyncService
             return '该类型未配置自动同步周期';
         }
         if (!isset($candidateIdMap[(int) ($ticket['id'] ?? 0)])) {
-            return sprintf('未进入本轮候选队列(仅扫描最久未同步的前%d条)', $maxPerRun * 5);
+            if (SplitScrmSpiderFactory::requiresNode($ticketType)) {
+                $poolSize = SplitSyncConfigService::getMaxTicketsPerCronRun()
+                    * SplitSyncConfigService::getNodeCandidatePoolMultiplier();
+                return sprintf('未进入本轮 Node 候选(前%d条到期工单公平轮转)', $poolSize);
+            }
+
+            return '直连通道：未到同步周期或本轮已处理';
         }
         if ($stoppedByNodeBusy) {
             return 'Node 队列繁忙，本轮未执行';

application/extra/site.php

@@ -58,5 +58,6 @@ return array (
   'split_sync_interval_whatshub' => '5',
   'split_sync_interval_sihai' => '0',
   'split_sync_fail_pause_threshold' => '6',
+  'split_sync_cli_php' => '',
   'split_sync_interval_chatknow' => '5',
 );

patches/application/admin/command/Install/split_config_node_channel.sql

@@ -0,0 +1,19 @@
+-- 双通道定时同步：Node 通道上限与候选池配置（已有环境增量执行）
+SET NAMES utf8mb4;
+
+INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`)
+SELECT 'split_sync_max_per_cron', 'split', 'Node通道单次同步上限', '每分钟 cron 在 Node 通道最多处理几条到期工单（仅影响需调用 Node 的类型）；纯 PHP 直连类型不受此限制', 'number', '', '5', '', '', '', NULL
+FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_max_per_cron' LIMIT 1);
+
+INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`)
+SELECT 'split_sync_node_candidate_multiplier', 'split', 'Node通道候选池倍数', 'Node 候选工单数 = 单次上限 × 本倍数，用于公平轮转', 'number', '', '10', '', '', '', NULL
+FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_node_candidate_multiplier' LIMIT 1);
+
+UPDATE `fa_config`
+SET `title` = 'Node通道单次同步上限',
+    `tip` = '每分钟 cron 在 Node 通道最多处理几条到期工单（仅影响需调用 Node 的类型）；纯 PHP 直连类型不受此限制'
+WHERE `name` = 'split_sync_max_per_cron';
+
+UPDATE `fa_config`
+SET `value` = '5'
+WHERE `name` = 'split_sync_max_per_cron' AND (`value` = '' OR `value` = '2');

patches/application/admin/command/Install/split_config_sync.sql

@@ -58,9 +58,13 @@ SELECT 'split_sync_interval_sihai', 'split', '四海云控同步周期(分钟)',
 FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_interval_sihai' LIMIT 1);
 
 INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`)
-SELECT 'split_sync_max_per_cron', 'split', '单次定时同步上限', '每分钟 cron 最多处理几条到期工单，防止打满 Node Browser 槽位', 'number', '', '2', '', '', '', NULL
+SELECT 'split_sync_max_per_cron', 'split', 'Node通道单次同步上限', '每分钟 cron 在 Node 通道最多处理几条到期工单（仅影响需调用 Node 的类型）；纯 PHP 直连类型不受此限制', 'number', '', '5', '', '', '', NULL
 FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_max_per_cron' LIMIT 1);
 
+INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`)
+SELECT 'split_sync_node_candidate_multiplier', 'split', 'Node通道候选池倍数', 'Node 候选工单数 = 单次上限 × 本倍数，用于公平轮转', 'number', '', '10', '', '', '', NULL
+FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_node_candidate_multiplier' LIMIT 1);
+
 INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`)
 SELECT 'split_sync_cron_lock_ttl', 'split', '定时同步全局锁TTL(秒)', '异常退出后自动释放全局锁，避免后续 cron 永久跳过', 'number', '', '1800', '', '', '', NULL
 FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_cron_lock_ttl' LIMIT 1);

patches/application/common/service/SplitScrmSpiderFactory.php

@@ -4,6 +4,7 @@ declare(strict_types=1);
 
 namespace app\common\service;
 
+use app\common\library\scrm\AbstractScrmSpider;
 use app\common\library\scrm\ScrmSpiderInterface;
 use app\common\library\scrm\spider\A2cSpider;
 use app\common\library\scrm\spider\ChatknowSpider;
@@ -59,6 +60,49 @@ class SplitScrmSpiderFactory
         return array_keys(self::MAP);
     }
 
+    /**
+     * 同步是否依赖 Node Headless（纯 PHP cURL 等为 false）
+     */
+    public static function requiresNode(string $ticketType): bool
+    {
+        $class = self::resolveClass($ticketType);
+        if ($class === null) {
+            return true;
+        }
+
+        return is_subclass_of($class, AbstractScrmSpider::class);
+    }
+
+    /**
+     * @return list<string>
+     */
+    public static function listDirectSyncTypes(): array
+    {
+        $types = [];
+        foreach (self::listSupportedTypes() as $ticketType) {
+            if (!self::requiresNode($ticketType)) {
+                $types[] = $ticketType;
+            }
+        }
+
+        return $types;
+    }
+
+    /**
+     * @return list<string>
+     */
+    public static function listNodeSyncTypes(): array
+    {
+        $types = [];
+        foreach (self::listSupportedTypes() as $ticketType) {
+            if (self::requiresNode($ticketType)) {
+                $types[] = $ticketType;
+            }
+        }
+
+        return $types;
+    }
+
     /**
      * @return ScrmSpiderInterface|null
      */

patches/application/common/service/SplitSyncConfigService.php

@@ -57,11 +57,23 @@ class SplitSyncConfigService
     {
         $value = self::getConfigValue('split_sync_max_per_cron');
         if ($value === '') {
-            return 2;
+            return 5;
         }
         return max(1, min(20, (int) $value));
     }
 
+    /**
+     * Node 通道候选池倍数（候选数 = maxPerRun × 本值）
+     */
+    public static function getNodeCandidatePoolMultiplier(): int
+    {
+        $value = self::getConfigValue('split_sync_node_candidate_multiplier');
+        if ($value === '') {
+            return 10;
+        }
+        return max(3, min(50, (int) $value));
+    }
+
     /**
      * 全局 cron 锁过期秒数，防止异常退出后永久占锁
      */
@@ -86,6 +98,53 @@ class SplitSyncConfigService
         return max(0, (int) $value);
     }
 
+    /**
+     * 后台投递 / Cron 使用的 PHP CLI 可执行文件路径
+     *
+     * Web(FPM) 环境下 PHP_BINARY 指向 php-fpm，不能用于 think 命令。
+     */
+    public static function getCliPhpBinary(): string
+    {
+        $configured = trim(self::getConfigValue('split_sync_cli_php'));
+        if ($configured !== '' && self::isUsableCliPhp($configured)) {
+            return $configured;
+        }
+
+        if (defined('PHP_BINDIR') && PHP_BINDIR !== '') {
+            $candidate = rtrim(PHP_BINDIR, '/\\') . DIRECTORY_SEPARATOR . 'php';
+            if (self::isUsableCliPhp($candidate)) {
+                return $candidate;
+            }
+        }
+
+        if (PHP_SAPI === 'cli' && defined('PHP_BINARY') && PHP_BINARY !== '' && self::isUsableCliPhp(PHP_BINARY)) {
+            return PHP_BINARY;
+        }
+
+        foreach (['/usr/bin/php', '/usr/local/bin/php'] as $candidate) {
+            if (self::isUsableCliPhp($candidate)) {
+                return $candidate;
+            }
+        }
+
+        return 'php';
+    }
+
+    /**
+     * 判断路径是否为可用的 PHP CLI（排除 php-fpm）
+     */
+    private static function isUsableCliPhp(string $path): bool
+    {
+        if (stripos($path, 'fpm') !== false) {
+            return false;
+        }
+        if ($path === 'php') {
+            return true;
+        }
+
+        return is_file($path) && is_executable($path);
+    }
+
     private static function getConfigValue(string $name): string
     {
         $site = Config::get('site.' . $name);

patches/application/common/service/SplitTicketSyncDispatchService.php

@@ -51,6 +51,7 @@ class SplitTicketSyncDispatchService
             'queued'  => $queued,
             'skipped' => $skipped,
             'failed'  => $failed,
+            'phpCli'  => $php,
         ]);
 
         return [
@@ -130,13 +131,43 @@ class SplitTicketSyncDispatchService
 
     private function resolvePhpBinary(): string
     {
-        if (defined('PHP_BINARY') && PHP_BINARY !== '') {
+        if (method_exists(SplitSyncConfigService::class, 'getCliPhpBinary')) {
-            return PHP_BINARY;
+            return SplitSyncConfigService::getCliPhpBinary();
         }
 
+        return $this->resolvePhpBinaryFallback();
+    }
+
+    /**
+     * 当 SplitSyncConfigService 未部署 getCliPhpBinary 时的兜底解析
+     */
+    private function resolvePhpBinaryFallback(): string
+    {
+        if (defined('PHP_BINDIR') && PHP_BINDIR !== '') {
+            $candidate = rtrim(PHP_BINDIR, '/\\') . DIRECTORY_SEPARATOR . 'php';
+            if ($this->isUsableCliPhp($candidate)) {
+                return $candidate;
+            }
+        }
+        foreach (['/usr/bin/php', '/usr/local/bin/php'] as $candidate) {
+            if ($this->isUsableCliPhp($candidate)) {
+                return $candidate;
+            }
+        }
         return 'php';
     }
 
+    private function isUsableCliPhp(string $path): bool
+    {
+        if (stripos($path, 'fpm') !== false) {
+            return false;
+        }
+        if ($path === 'php') {
+            return true;
+        }
+        return is_file($path) && is_executable($path);
+    }
+
     private function resolveThinkScript(): string
     {
         $root = $this->resolveRootPath();

patches/application/common/service/SplitTicketSyncService.php

@@ -80,131 +80,49 @@ class SplitTicketSyncService
     }
 
     /**
-     * 扫描到期工单并同步（限流 + Node 队列感知）
+     * 扫描到期工单并同步（直连 PHP 与 Node 双通道）
      */
     public function syncDueTickets(): int
     {
-        $maxPerRun = SplitSyncConfigService::getMaxTicketsPerCronRun();
         $autoSyncTypes = $this->resolveAutoSyncTicketTypes();
-        $failThreshold = SplitSyncConfigService::getFailPauseThreshold();
-
         if ($autoSyncTypes === []) {
             SplitTicketSyncLogger::logCron('scan skip', ['reason' => 'no auto-sync ticket types']);
             return 0;
         }
 
-        if (!SplitNodeHealthService::canAcceptWork()) {
+        $failThreshold = SplitSyncConfigService::getFailPauseThreshold();
-            SplitTicketSyncLogger::logCron('scan skip', [
+        $directTypes = array_values(array_intersect(
-                'reason' => 'node queue busy',
+            $autoSyncTypes,
-                'queue'  => SplitNodeHealthService::getQueueSnapshot(),
+            SplitScrmSpiderFactory::listDirectSyncTypes()
-            ]);
+        ));
-            return 0;
+        $nodeTypes = array_values(array_intersect(
-        }
+            $autoSyncTypes,
+            SplitScrmSpiderFactory::listNodeSyncTypes()
+        ));
 
-        $query = Ticket::where('status', 'normal')->whereIn('ticket_type', $autoSyncTypes);
-        if ($failThreshold > 0) {
-            $query->where('sync_fail_count', '<', $failThreshold);
-        }
-        $list = $query->order('sync_time', 'asc')->order('id', 'asc')
-            ->limit($maxPerRun * 5)
-            ->select();
-        $list = $this->sortTicketsBySessionKey($list);
-
-        $candidateCount = count($list);
         SplitTicketSyncLogger::logCron('scan start', [
-            'candidateCount' => $candidateCount,
-            'maxPerRun'      => $maxPerRun,
             'autoSyncTypes' => $autoSyncTypes,
+            'directTypes'   => $directTypes,
+            'nodeTypes'     => $nodeTypes,
             'failThreshold' => $failThreshold,
             'nodeQueue'     => SplitNodeHealthService::getQueueSnapshot(),
-            'groupedBy'      => 'sessionKey',
-        ]);
-        SplitTicketSyncLogger::log('cron', 'scan start', [
-            'candidateCount' => $candidateCount,
-            'maxPerRun'      => $maxPerRun,
-            'autoSyncTypes'  => $autoSyncTypes,
-            'nodeQueue'      => SplitNodeHealthService::getQueueSnapshot(),
         ]);
 
-        /** @var list<array<string, mixed>> $processed */
+        $directResult = $this->runDirectCronChannel($directTypes, $failThreshold);
-        $processed = [];
+        $nodeResult = $this->runNodeCronChannel($nodeTypes, $failThreshold);
-        /** @var list<array<string, mixed>> $skipped */
-        $skipped = [];
-        $count = 0;
-        $stoppedByNodeBusy = false;
 
-        foreach ($list as $index => $ticket) {
+        $processedCount = $directResult['processedCount'] + $nodeResult['processedCount'];
-            $ticketId = (int) $ticket['id'];
+        $processed = array_merge($directResult['processed'], $nodeResult['processed']);
-            $ticketUrl = (string) $ticket['ticket_url'];
+        $skipped = array_merge($directResult['skipped'], $nodeResult['skipped']);
-            $ticketType = (string) $ticket['ticket_type'];
+        $candidateIds = array_merge($directResult['candidateIds'], $nodeResult['candidateIds']);
-            $sessionKey = AntiBotConfigBuilder::resolveSessionKey($ticketType, $ticketUrl);
-
-            if ($count >= $maxPerRun) {
-                $reason = '本轮处理上限已满';
-                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $reason);
-                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $reason);
-                continue;
-            }
-
-            $skip = $this->shouldSkip($ticket);
-            if ($skip !== null) {
-                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skip);
-                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skip);
-                continue;
-            }
-
-            if (!SplitNodeHealthService::canAcceptWork()) {
-                $reason = 'Node 队列繁忙，停止本轮后续同步';
-                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $reason);
-                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $reason);
-                $stoppedByNodeBusy = true;
-                for ($i = $index + 1; $i < $candidateCount; $i++) {
-                    $remain = $list[$i];
-                    $remainReason = 'Node 队列繁忙，本轮未执行';
-                    $skipped[] = $this->buildCronTicketEntry(
-                        (int) $remain['id'],
-                        (string) $remain['ticket_type'],
-                        (string) $remain['ticket_url'],
-                        $remainReason
-                    );
-                }
-                SplitTicketSyncLogger::log('cron', 'node busy, stop batch', [
-                    'processed' => $count,
-                    'queue'     => SplitNodeHealthService::getQueueSnapshot(),
-                ]);
-                break;
-            }
-
-            $result = $this->syncOne($ticketId, false, true, 'auto');
-            if (!empty($result['skipped'])) {
-                $skipReason = (string) ($result['message'] ?? '已跳过');
-                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skipReason);
-                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skipReason);
-                continue;
-            }
-
-            $processed[] = [
-                'ticketId'   => $ticketId,
-                'ticketType' => $ticketType,
-                'ticketUrl'  => $ticketUrl,
-                'sessionKey' => $sessionKey,
-                'success'    => !empty($result['success']),
-                'message'    => (string) ($result['message'] ?? ''),
-            ];
-            $count++;
-        }
-
-        $candidateIds = array_map(static function ($row) {
-            return (int) (is_array($row) ? ($row['id'] ?? 0) : ($row['id'] ?? 0));
-        }, $list);
         $openNotSynced = $this->buildOpenNotSyncedAudit(
             $autoSyncTypes,
             $failThreshold,
-            $maxPerRun,
+            SplitSyncConfigService::getMaxTicketsPerCronRun(),
             $candidateIds,
             $processed,
             $skipped,
-            $stoppedByNodeBusy
+            $nodeResult['stoppedByNodeBusy']
         );
 
         $successCount = count(array_filter($processed, static function (array $row): bool {
@@ -213,21 +131,362 @@ class SplitTicketSyncService
         $failedCount = count($processed) - $successCount;
 
         SplitTicketSyncLogger::logCron('scan end', [
-            'processedCount' => $count,
+            'processedCount' => $processedCount,
             'successCount'   => $successCount,
             'failedCount'    => $failedCount,
             'skippedCount'   => count($skipped),
+            'directChannel'  => $directResult['summary'],
+            'nodeChannel'    => $nodeResult['summary'],
             'processed'      => $processed,
             'skipped'        => $skipped,
             'openNotSynced'  => $openNotSynced,
         ]);
         SplitTicketSyncLogger::log('cron', 'scan end', [
-            'processedCount' => $count,
+            'processedCount' => $processedCount,
             'successCount'   => $successCount,
             'failedCount'    => $failedCount,
         ]);
 
-        return $count;
+        return $processedCount;
+    }
+
+    /**
+     * 直连 PHP 通道：按周期同步，不占 Node 名额
+     *
+     * @param list<string> $directTypes
+     * @return array{
+     *   processedCount:int,
+     *   processed:list<array<string,mixed>>,
+     *   skipped:list<array<string,mixed>>,
+     *   candidateIds:list<int>,
+     *   summary:array<string,mixed>
+     * }
+     */
+    private function runDirectCronChannel(array $directTypes, int $failThreshold): array
+    {
+        if ($directTypes === []) {
+            return $this->emptyCronChannelResult('direct');
+        }
+
+        $list = $this->loadOpenTicketsForTypes($directTypes, $failThreshold, 0);
+        $candidateIds = $this->extractTicketIds($list);
+
+        SplitTicketSyncLogger::logCron('direct channel start', [
+            'types'           => $directTypes,
+            'candidateCount'  => count($list),
+        ]);
+
+        /** @var list<array<string, mixed>> $processed */
+        $processed = [];
+        /** @var list<array<string, mixed>> $skipped */
+        $skipped = [];
+        $count = 0;
+
+        foreach ($list as $ticket) {
+            $ticketId = (int) $ticket['id'];
+            $ticketType = (string) $ticket['ticket_type'];
+            $ticketUrl = (string) $ticket['ticket_url'];
+
+            $skip = $this->shouldSkip($ticket);
+            if ($skip !== null) {
+                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skip);
+                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skip, 'direct');
+                continue;
+            }
+
+            $result = $this->syncOne($ticketId, false, true, 'auto');
+            if (!empty($result['skipped'])) {
+                $skipReason = (string) ($result['message'] ?? '已跳过');
+                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skipReason);
+                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skipReason, 'direct');
+                continue;
+            }
+
+            $processed[] = [
+                'ticketId'   => $ticketId,
+                'ticketType' => $ticketType,
+                'ticketUrl'  => $ticketUrl,
+                'channel'    => 'direct',
+                'success'    => !empty($result['success']),
+                'message'    => (string) ($result['message'] ?? ''),
+            ];
+            $count++;
+        }
+
+        return [
+            'processedCount' => $count,
+            'processed'      => $processed,
+            'skipped'        => $skipped,
+            'candidateIds'   => $candidateIds,
+            'summary'        => [
+                'channel'        => 'direct',
+                'processedCount' => $count,
+                'skippedCount'   => count($skipped),
+            ],
+            'stoppedByNodeBusy' => false,
+        ];
+    }
+
+    /**
+     * Node 通道：限流 + 类型公平轮转
+     *
+     * @param list<string> $nodeTypes
+     * @return array{
+     *   processedCount:int,
+     *   processed:list<array<string,mixed>>,
+     *   skipped:list<array<string,mixed>>,
+     *   candidateIds:list<int>,
+     *   summary:array<string,mixed>,
+     *   stoppedByNodeBusy:bool
+     * }
+     */
+    private function runNodeCronChannel(array $nodeTypes, int $failThreshold): array
+    {
+        if ($nodeTypes === []) {
+            return $this->emptyCronChannelResult('node');
+        }
+
+        $maxPerRun = SplitSyncConfigService::getMaxTicketsPerCronRun();
+        $poolMultiplier = SplitSyncConfigService::getNodeCandidatePoolMultiplier();
+        $candidateLimit = $maxPerRun * $poolMultiplier;
+
+        $pool = $this->loadOpenTicketsForTypes($nodeTypes, $failThreshold, $candidateLimit);
+        $candidateIds = $this->extractTicketIds($pool);
+        $duePool = [];
+        foreach ($pool as $ticket) {
+            if ($this->shouldSkip($ticket) === null) {
+                $duePool[] = $ticket;
+            }
+        }
+        $picked = $this->pickFairNodeTickets($duePool, $maxPerRun);
+        $picked = $this->sortTicketsBySessionKey($picked);
+        $pickedIdMap = array_fill_keys($this->extractTicketIds($picked), true);
+
+        SplitTicketSyncLogger::logCron('node channel start', [
+            'types'          => $nodeTypes,
+            'maxPerRun'      => $maxPerRun,
+            'poolCount'      => count($pool),
+            'dueCount'       => count($duePool),
+            'pickedCount'    => count($picked),
+            'nodeQueue'      => SplitNodeHealthService::getQueueSnapshot(),
+        ]);
+
+        if (!SplitNodeHealthService::canAcceptWork()) {
+            SplitTicketSyncLogger::logCron('node channel skip', [
+                'reason' => 'node queue busy',
+                'queue'  => SplitNodeHealthService::getQueueSnapshot(),
+            ]);
+            $skipped = [];
+            foreach ($picked as $ticket) {
+                $skipped[] = $this->buildCronTicketEntry(
+                    (int) $ticket['id'],
+                    (string) $ticket['ticket_type'],
+                    (string) $ticket['ticket_url'],
+                    'Node 队列繁忙，本轮未执行'
+                );
+            }
+            foreach ($pool as $ticket) {
+                $ticketId = (int) $ticket['id'];
+                if (isset($pickedIdMap[$ticketId])) {
+                    continue;
+                }
+                $skip = $this->shouldSkip($ticket);
+                if ($skip !== null) {
+                    continue;
+                }
+                $skipped[] = $this->buildCronTicketEntry(
+                    $ticketId,
+                    (string) $ticket['ticket_type'],
+                    (string) $ticket['ticket_url'],
+                    sprintf('未进入本轮 Node 候选(前%d条到期工单公平轮转)', $candidateLimit)
+                );
+            }
+
+            return [
+                'processedCount'    => 0,
+                'processed'         => [],
+                'skipped'           => $skipped,
+                'candidateIds'      => $candidateIds,
+                'summary'           => [
+                    'channel'        => 'node',
+                    'processedCount' => 0,
+                    'skippedCount'   => count($skipped),
+                    'reason'         => 'node queue busy',
+                ],
+                'stoppedByNodeBusy' => true,
+            ];
+        }
+
+        /** @var list<array<string, mixed>> $processed */
+        $processed = [];
+        /** @var list<array<string, mixed>> $skipped */
+        $skipped = [];
+        $count = 0;
+        $stoppedByNodeBusy = false;
+
+        foreach ($pool as $ticket) {
+            $ticketId = (int) $ticket['id'];
+            $ticketType = (string) $ticket['ticket_type'];
+            $ticketUrl = (string) $ticket['ticket_url'];
+
+            if (!isset($pickedIdMap[$ticketId])) {
+                $skip = $this->shouldSkip($ticket);
+                if ($skip !== null) {
+                    continue;
+                }
+                $skipped[] = $this->buildCronTicketEntry(
+                    $ticketId,
+                    $ticketType,
+                    $ticketUrl,
+                    sprintf('未进入本轮 Node 候选(前%d条到期工单公平轮转)', $candidateLimit)
+                );
+                continue;
+            }
+
+            if (!SplitNodeHealthService::canAcceptWork()) {
+                $stoppedByNodeBusy = true;
+                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, 'Node 队列繁忙，停止本轮后续同步');
+                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, 'Node 队列繁忙，停止本轮后续同步', 'node');
+                continue;
+            }
+
+            $sessionKey = AntiBotConfigBuilder::resolveSessionKey($ticketType, $ticketUrl);
+            $result = $this->syncOne($ticketId, false, true, 'auto');
+            if (!empty($result['skipped'])) {
+                $skipReason = (string) ($result['message'] ?? '已跳过');
+                $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skipReason);
+                $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skipReason, 'node');
+                continue;
+            }
+
+            $processed[] = [
+                'ticketId'   => $ticketId,
+                'ticketType' => $ticketType,
+                'ticketUrl'  => $ticketUrl,
+                'sessionKey' => $sessionKey,
+                'channel'    => 'node',
+                'success'    => !empty($result['success']),
+                'message'    => (string) ($result['message'] ?? ''),
+            ];
+            $count++;
+        }
+
+        return [
+            'processedCount'    => $count,
+            'processed'         => $processed,
+            'skipped'           => $skipped,
+            'candidateIds'      => $candidateIds,
+            'summary'           => [
+                'channel'        => 'node',
+                'processedCount' => $count,
+                'skippedCount'   => count($skipped),
+                'maxPerRun'      => $maxPerRun,
+            ],
+            'stoppedByNodeBusy' => $stoppedByNodeBusy,
+        ];
+    }
+
+    /**
+     * @param list<string> $types
+     * @return list<Ticket>
+     */
+    private function loadOpenTicketsForTypes(array $types, int $failThreshold, int $limit): array
+    {
+        if ($types === []) {
+            return [];
+        }
+
+        $query = Ticket::where('status', 'normal')->whereIn('ticket_type', $types);
+        if ($failThreshold > 0) {
+            $query->where('sync_fail_count', '<', $failThreshold);
+        }
+        $query->order('sync_time', 'asc')->order('id', 'asc');
+        if ($limit > 0) {
+            $query->limit($limit);
+        }
+
+        $list = $query->select();
+        return $list instanceof \think\Collection ? $list->all() : (array) $list;
+    }
+
+    /**
+     * Node 通道：按工单类型公平轮转选取（每轮每类型优先取 1 条）
+     *
+     * @param list<Ticket> $dueTickets
+     * @return list<Ticket>
+     */
+    private function pickFairNodeTickets(array $dueTickets, int $maxPerRun): array
+    {
+        if ($dueTickets === [] || $maxPerRun <= 0) {
+            return [];
+        }
+
+        /** @var array<string, list<Ticket>> $byType */
+        $byType = [];
+        foreach ($dueTickets as $ticket) {
+            $type = (string) $ticket['ticket_type'];
+            $byType[$type][] = $ticket;
+        }
+
+        $typeKeys = array_keys($byType);
+        sort($typeKeys, SORT_STRING);
+
+        /** @var list<Ticket> $picked */
+        $picked = [];
+        while (count($picked) < $maxPerRun) {
+            $added = false;
+            foreach ($typeKeys as $type) {
+                if (count($picked) >= $maxPerRun) {
+                    break;
+                }
+                if ($byType[$type] === []) {
+                    continue;
+                }
+                $picked[] = array_shift($byType[$type]);
+                $added = true;
+            }
+            if (!$added) {
+                break;
+            }
+        }
+
+        return $picked;
+    }
+
+    /**
+     * @param list<Ticket|array<string,mixed>> $list
+     * @return list<int>
+     */
+    private function extractTicketIds(array $list): array
+    {
+        $ids = [];
+        foreach ($list as $row) {
+            $ids[] = (int) (is_array($row) ? ($row['id'] ?? 0) : ($row['id'] ?? 0));
+        }
+
+        return $ids;
+    }
+
+    /**
+     * @return array{
+     *   processedCount:int,
+     *   processed:list<array<string,mixed>>,
+     *   skipped:list<array<string,mixed>>,
+     *   candidateIds:list<int>,
+     *   summary:array<string,mixed>,
+     *   stoppedByNodeBusy:bool
+     * }
+     */
+    private function emptyCronChannelResult(string $channel): array
+    {
+        return [
+            'processedCount'    => 0,
+            'processed'         => [],
+            'skipped'           => [],
+            'candidateIds'      => [],
+            'summary'           => ['channel' => $channel, 'processedCount' => 0, 'skippedCount' => 0],
+            'stoppedByNodeBusy' => false,
+        ];
     }
 
     /**
@@ -466,13 +725,14 @@ class SplitTicketSyncService
         ];
     }
 
-    private function logCronCandidateSkipped(int $ticketId, string $ticketType, string $ticketUrl, string $reason): void
+    private function logCronCandidateSkipped(int $ticketId, string $ticketType, string $ticketUrl, string $reason, string $channel = 'legacy'): void
     {
         $ctx = [
             'ticketId'   => $ticketId,
             'ticketType' => $ticketType,
             'ticketUrl'  => $ticketUrl,
             'reason'     => $reason,
+            'channel'    => $channel,
         ];
         SplitTicketSyncLogger::logCron('candidate skipped', $ctx);
         SplitTicketSyncLogger::log('cron', 'candidate skipped', $ctx);
@@ -515,14 +775,15 @@ class SplitTicketSyncService
             ->select();
 
         $result = [];
-        foreach ($openList as $ticket) {
+        foreach ($openList as $ticketRow) {
-            $ticketId = (int) $ticket['id'];
+            $ticket = $ticketRow instanceof Ticket ? $ticketRow->toArray() : (array) $ticketRow;
-            if (isset($handledIds[$ticketId])) {
+            $ticketId = (int) ($ticket['id'] ?? 0);
+            if ($ticketId <= 0 || isset($handledIds[$ticketId])) {
                 continue;
             }
 
-            $ticketType = (string) $ticket['ticket_type'];
+            $ticketType = (string) ($ticket['ticket_type'] ?? '');
-            $ticketUrl = (string) $ticket['ticket_url'];
+            $ticketUrl = (string) ($ticket['ticket_url'] ?? '');
             $reason = $this->resolveOpenNotSyncedReason(
                 $ticket,
                 $autoSyncTypes,
@@ -568,7 +829,13 @@ class SplitTicketSyncService
             return '该类型未配置自动同步周期';
         }
         if (!isset($candidateIdMap[(int) ($ticket['id'] ?? 0)])) {
-            return sprintf('未进入本轮候选队列(仅扫描最久未同步的前%d条)', $maxPerRun * 5);
+            if (SplitScrmSpiderFactory::requiresNode($ticketType)) {
+                $poolSize = SplitSyncConfigService::getMaxTicketsPerCronRun()
+                    * SplitSyncConfigService::getNodeCandidatePoolMultiplier();
+                return sprintf('未进入本轮 Node 候选(前%d条到期工单公平轮转)', $poolSize);
+            }
+
+            return '直连通道：未到同步周期或本轮已处理';
         }
         if ($stoppedByNodeBusy) {
             return 'Node 队列繁忙，本轮未执行';

patches/puppeteer-api/server.js

@@ -60,6 +60,7 @@ const {
     isHuojianShortEntryUrl,
     applyHuojianBusinessHostHint,
 } = require('./lib/huojian-url');
+const { runUiPagination } = require('./lib/ui-pagination-runner');
 
 const app = express();
 app.use(express.json({ limit: '50mb' }));