feat(代理): 代理配置按登录用户隔离——桌面端 current_uid 登录同步 + brand 页 proxy_users[uid] 读写，各用户各自代理池各自计费复用（未登录回退全局 proxy_url 兼容旧配置）

backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java

@@ -44,8 +44,14 @@ public class BrandCheckClient {
         this.externalCallMetrics = externalCallMetrics;
     }
 
-    private final ExecutorService checkExecutor = Executors.newFixedThreadPool(
+    // 有界队列 + CallerRuns：大批量（几千品牌）检查时队列洪峰不无界堆内存，
-            BRAND_CHECK_CONCURRENCY, namedThreadFactory("brand-check"));
+    // 队列打满时提交线程自己执行一个检查任务，上游天然限速。
+    private final ExecutorService checkExecutor = new java.util.concurrent.ThreadPoolExecutor(
+            BRAND_CHECK_CONCURRENCY, BRAND_CHECK_CONCURRENCY,
+            60L, java.util.concurrent.TimeUnit.SECONDS,
+            new java.util.concurrent.LinkedBlockingQueue<>(1000),
+            namedThreadFactory("brand-check"),
+            new java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy());
 
     @PreDestroy
     void shutdownCheckExecutor() {

backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java

@@ -58,6 +58,7 @@ import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
 import com.nanri.aiimage.modules.task.service.TaskFileJobService;
 import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
 import lombok.RequiredArgsConstructor;
+import com.nanri.aiimage.common.service.DistributedJobLockService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.poi.ss.usermodel.Cell;
 import org.apache.poi.ss.usermodel.DataFormatter;
@@ -147,6 +148,7 @@ public class CollectDataService {
     private final OssStorageService ossStorageService;
     private final ObjectMapper objectMapper;
     private final TransactionTemplate transactionTemplate;
+    private final DistributedJobLockService distributedJobLockService;
 
     /** ASIN 去重 + 无效品牌批量集合查询器：两段式查询合并为一次往返，语义与旧实现等价。 */
     private final CollectDataBatchQuery collectDataBatchQuery;
@@ -483,16 +485,25 @@ public class CollectDataService {
 
     @Scheduled(cron = "${aiimage.collect-data.stale-check-cron:*/30 * * * * *}")
     public void finalizeStaleTasks() {
-        long timeoutMinutes = Math.max(1L, staleTimeoutMinutes);
+        // 双节点互斥：Redis 不可用时 fail-open（仅记日志），兜底靠下方 per-task DB 锁，不会双写终态
-        LocalDateTime threshold = LocalDateTime.now().minusMinutes(timeoutMinutes);
+        DistributedJobLockService.LockHandle jobLock =
-        List<FileTaskEntity> tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
+                distributedJobLockService.tryLock("collect-data:stale-check", java.time.Duration.ofMinutes(2));
-                .eq(FileTaskEntity::getModuleType, MODULE_TYPE)
+        if (jobLock == null) {
-                .eq(FileTaskEntity::getStatus, STATUS_RUNNING)
+            log.info("[collect-data] stale-check skipped, another instance holds the distributed lock");
-                .lt(FileTaskEntity::getUpdatedAt, threshold)
+            return;
-                .orderByAsc(FileTaskEntity::getUpdatedAt)
+        }
-                .last("limit 200"));
+        try (jobLock) {
-        for (FileTaskEntity task : tasks) {
+            long timeoutMinutes = Math.max(1L, staleTimeoutMinutes);
-            finalizeStaleTask(task.getId(), threshold, timeoutMinutes);
+            LocalDateTime threshold = LocalDateTime.now().minusMinutes(timeoutMinutes);
+            List<FileTaskEntity> tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
+                    .eq(FileTaskEntity::getModuleType, MODULE_TYPE)
+                    .eq(FileTaskEntity::getStatus, STATUS_RUNNING)
+                    .lt(FileTaskEntity::getUpdatedAt, threshold)
+                    .orderByAsc(FileTaskEntity::getUpdatedAt)
+                    .last("limit 200"));
+            for (FileTaskEntity task : tasks) {
+                finalizeStaleTask(task.getId(), threshold, timeoutMinutes);
+            }
         }
     }
 
@@ -1201,36 +1212,6 @@ public class CollectDataService {
         return safe.isBlank() ? "collect-data" : safe;
     }
 
-    private void deleteTransientTaskPayloads(List<TaskChunkEntity> chunks, List<TaskResultItemEntity> items) {
-        if (chunks != null) {
-            Set<String> deleted = new HashSet<>();
-            for (TaskChunkEntity chunk : chunks) {
-                // 多 chunk 共享同一对象（deterministic key 残留场景）按值去重只删一次。
-                if (chunk.getPayloadJson() != null && deleted.add(chunk.getPayloadJson())) {
-                    transientPayloadStorageService.deletePayloadIfPresent(chunk.getPayloadJson());
-                }
-            }
-        }
-        deleteResultItemPayloads(items);
-    }
-
-    /** 删除结果明细 payload：chunk 级引用按对象去重后各删一次，旧格式逐行删。 */
-    private void deleteResultItemPayloads(List<TaskResultItemEntity> items) {
-        if (items == null) {
-            return;
-        }
-        Set<String> deletedPointers = new HashSet<>();
-        for (TaskResultItemEntity item : items) {
-            CollectDataResultDetailCodec.ChunkRef ref = resultDetailCodec.parseRef(item.getPayloadJson());
-            if (ref != null) {
-                if (deletedPointers.add(ref.pointer())) {
-                    transientPayloadStorageService.deletePayloadIfPresent(ref.pointer());
-                }
-            } else {
-                transientPayloadStorageService.deletePayloadIfPresent(item.getPayloadJson());
-            }
-        }
-    }
 
     private static class CollectDataStats {
         private int totalRows;
@@ -1252,9 +1233,38 @@ public class CollectDataService {
         private List<CollectDataSummaryRowDto> summaries = new ArrayList<>();
     }
 
-    @Transactional
+    // 不用 @Transactional（同类自调用代理失效）：用 TransactionTemplate 显式包 DB 段，
+    // 事务提交后再做远程删除（慢 IO），避免大任务删除时长事务占用连接池。
     public void deleteTask(Long taskId, Long userId) {
+        List<String> chunkPayloadPointers = new ArrayList<>();
+        List<String> resultPayloadJsons = new ArrayList<>();
+        transactionTemplate.executeWithoutResult(status ->
+                deleteTaskRows(taskId, userId, chunkPayloadPointers, resultPayloadJsons));
+        // 事务已提交，这里执行远程删除
+        Set<String> deleted = new HashSet<>();
+        for (String pointer : chunkPayloadPointers) {
+            if (deleted.add(pointer)) {
+                transientPayloadStorageService.deletePayloadIfPresent(pointer);
+            }
+        }
+        deletePayloadJsonsAndPointers(resultPayloadJsons, deleted);
+    }
+
+    private void deleteTaskRows(Long taskId, Long userId, List<String> chunkPayloadPointers, List<String> resultPayloadJsons) {
         FileTaskEntity task = requireTask(taskId, userId);
+        taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
+                .select(TaskChunkEntity::getPayloadJson)
+                .eq(TaskChunkEntity::getTaskId, task.getId())
+                .eq(TaskChunkEntity::getModuleType, MODULE_TYPE)).stream()
+                .map(TaskChunkEntity::getPayloadJson)
+                .filter(Objects::nonNull)
+                .forEach(chunkPayloadPointers::add);
+        taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
+                .select(TaskResultItemEntity::getPayloadJson)
+                .eq(TaskResultItemEntity::getTaskId, task.getId())
+                .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)).stream()
+                .map(TaskResultItemEntity::getPayloadJson)
+                .forEach(resultPayloadJsons::add);
         collectDataItemMapper.delete(new LambdaQueryWrapper<CollectDataItemEntity>()
                 .eq(CollectDataItemEntity::getTaskId, task.getId()));
         fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
@@ -1271,37 +1281,47 @@ public class CollectDataService {
         taskResultItemMapper.delete(new LambdaQueryWrapper<TaskResultItemEntity>()
                 .eq(TaskResultItemEntity::getTaskId, task.getId())
                 .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE));
-        deleteTransientTaskPayloads(
-                taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
-                        .select(TaskChunkEntity::getPayloadJson)
-                        .eq(TaskChunkEntity::getTaskId, task.getId())
-                        .eq(TaskChunkEntity::getModuleType, MODULE_TYPE)),
-                taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
-                        .select(TaskResultItemEntity::getPayloadJson)
-                        .eq(TaskResultItemEntity::getTaskId, task.getId())
-                        .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)));
         taskFileJobService.deleteTaskJobs(task.getId(), MODULE_TYPE);
         fileTaskMapper.deleteById(task.getId());
     }
 
+    private void deletePayloadJsonsAndPointers(List<String> payloadJsons, Set<String> deletedPointers) {
+        for (String payloadJson : payloadJsons) {
+            if (payloadJson == null) {
+                continue;
+            }
+            CollectDataResultDetailCodec.ChunkRef ref = resultDetailCodec.parseRef(payloadJson);
+            if (ref != null) {
+                if (deletedPointers.add(ref.pointer())) {
+                    transientPayloadStorageService.deletePayloadIfPresent(ref.pointer());
+                }
+            } else {
+                transientPayloadStorageService.deletePayloadIfPresent(payloadJson);
+            }
+        }
+    }
+
     public void deleteHistory(Long resultId, Long userId) {
         FileResultEntity row = fileResultMapper.selectById(resultId);
         if (row == null || !MODULE_TYPE.equals(row.getModuleType()) || !Objects.equals(userId, row.getUserId())) {
             throw new BusinessException("记录不存在");
         }
-        List<TaskResultItemEntity> resultItems = taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
+        List<String> resultPayloadJsons = transactionTemplate.execute(status -> {
-                .select(TaskResultItemEntity::getPayloadJson)
+            List<TaskResultItemEntity> resultItems = taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
-                .eq(TaskResultItemEntity::getTaskId, row.getTaskId())
+                    .select(TaskResultItemEntity::getPayloadJson)
-                .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)
+                    .eq(TaskResultItemEntity::getTaskId, row.getTaskId())
-                .eq(TaskResultItemEntity::getResultId, row.getId()));
+                    .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)
-        taskResultItemMapper.delete(new LambdaQueryWrapper<TaskResultItemEntity>()
+                    .eq(TaskResultItemEntity::getResultId, row.getId()));
-                .eq(TaskResultItemEntity::getTaskId, row.getTaskId())
+            taskResultItemMapper.delete(new LambdaQueryWrapper<TaskResultItemEntity>()
-                .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)
+                    .eq(TaskResultItemEntity::getTaskId, row.getTaskId())
-                .eq(TaskResultItemEntity::getResultId, row.getId()));
+                    .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)
-        // 与 deleteTask 一致：先删 DB 行再物理删对象，保证行删除与对象删除一致。
+                    .eq(TaskResultItemEntity::getResultId, row.getId()));
-        deleteResultItemPayloads(resultItems);
+            taskFileJobService.deleteResultJobs(row.getTaskId(), MODULE_TYPE, row.getId());
-        taskFileJobService.deleteResultJobs(row.getTaskId(), MODULE_TYPE, row.getId());
+            fileResultMapper.deleteById(resultId);
-        fileResultMapper.deleteById(resultId);
+            return resultItems.stream().map(TaskResultItemEntity::getPayloadJson).toList();
+        });
+        // 事务已提交，再物理删对象，保证行删除与对象删除一致。
+        deletePayloadJsonsAndPointers(resultPayloadJsons, new HashSet<>());
     }
 
     public CollectDataCountryPreferenceVo getCountryPreference(Long userId) {

backend-java/src/main/java/com/nanri/aiimage/modules/file/service/oss/OssStorageService.java

@@ -41,6 +41,8 @@ public class OssStorageService {
     private static final String DIGITAL_HUMAN_PREFIX = "digital-human/versions/";
     private static final String SOFTWARE_VERSION_PREFIX = "nanri-image/versions/";
     private static final String LEGACY_MINIO_ENDPOINT = "http://47.110.241.161:9000";
+    /** readObjectBytes 兜底读取上限（20MB）：防止大对象一次性读入内存导致 OOM。 */
+    private static final long DEFAULT_READ_MAX_BYTES = 20L * 1024 * 1024;
 
     private final OssProperties ossProperties;
     private final MinioClient minioClient;
@@ -248,13 +250,32 @@ public class OssStorageService {
     }
 
     public byte[] readObjectBytes(String bucket, String objectKey) {
+        // 防御性上限：防止被误用到大对象时一次性读入内存导致 OOM（当前仅模板下载场景，模板为 xlsx KB 级）
+        return readObjectBytes(bucket, objectKey, DEFAULT_READ_MAX_BYTES);
+    }
+
+    public byte[] readObjectBytes(String bucket, String objectKey, long maxBytes) {
         String normalizedBucket = requireStorageName(bucket, "bucket");
         String normalizedObjectKey = requireStorageName(objectKey, "objectKey");
         try (var stream = buildClient().getObject(GetObjectArgs.builder()
                 .bucket(normalizedBucket)
                 .object(normalizedObjectKey)
                 .build())) {
-            return stream.readAllBytes();
+            byte[] buffer = new byte[1024 * 1024];
+            ByteArrayOutputStream out = new ByteArrayOutputStream();
+            long total = 0;
+            int read;
+            while ((read = stream.read(buffer)) != -1) {
+                total += read;
+                if (total > maxBytes) {
+                    throw new IllegalArgumentException("对象过大，超过读取上限 " + maxBytes + " 字节: "
+                            + normalizedBucket + "/" + normalizedObjectKey);
+                }
+                out.write(buffer, 0, read);
+            }
+            return out.toByteArray();
+        } catch (IllegalArgumentException ex) {
+            throw ex;
         } catch (Exception ex) {
             throw storageFailure("read", normalizedBucket + "/" + normalizedObjectKey, ex);
         }

backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskService.java

@@ -16,6 +16,7 @@ import com.nanri.aiimage.modules.imagevideo.model.dto.ImageVideoWorkflowResultRe
 import com.nanri.aiimage.modules.imagevideo.model.dto.ImageVideoWorkflowRunRequest;
 import com.nanri.aiimage.modules.imagevideo.model.entity.ImageVideoAsyncTaskEntity;
 import com.nanri.aiimage.modules.imagevideo.model.vo.ImageVideoAsyncTaskVo;
+import com.nanri.aiimage.common.service.DistributedJobLockService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.context.event.ApplicationReadyEvent;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -66,6 +67,7 @@ public class ImageVideoAsyncTaskService {
     private final ObjectMapper objectMapper;
     private final TaskExecutor taskQueueExecutor;
     private final InstanceMetadata instanceMetadata;
+    private final com.nanri.aiimage.common.service.DistributedJobLockService distributedJobLockService;
 
     public ImageVideoAsyncTaskService(
             ImageVideoAsyncTaskMapper taskMapper,
@@ -74,7 +76,8 @@ public class ImageVideoAsyncTaskService {
             ImageVideoArchiveService archiveService,
             ObjectMapper objectMapper,
             @Qualifier("taskQueueExecutor") TaskExecutor taskQueueExecutor,
-            InstanceMetadata instanceMetadata) {
+            InstanceMetadata instanceMetadata,
+            com.nanri.aiimage.common.service.DistributedJobLockService distributedJobLockService) {
         this.taskMapper = taskMapper;
         this.cozeService = cozeService;
         this.workflowConfigService = workflowConfigService;
@@ -82,6 +85,7 @@ public class ImageVideoAsyncTaskService {
         this.objectMapper = objectMapper;
         this.taskQueueExecutor = taskQueueExecutor;
         this.instanceMetadata = instanceMetadata;
+        this.distributedJobLockService = distributedJobLockService;
     }
 
     public ImageVideoAsyncTaskVo submitDouyinCopy(DouyinCopyRequest request) {
@@ -130,26 +134,41 @@ public class ImageVideoAsyncTaskService {
 
     @Scheduled(fixedDelayString = "${aiimage.image-video.async-task-dispatch-delay-ms:1000}")
     public void dispatchPendingTasks() {
-        List<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>()
+        // 双节点互斥：只影响扫描入队动作（CAS claim 已兜底不重复执行），省掉另一节点的每秒空扫
-                .eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.PENDING.name())
+        DistributedJobLockService.LockHandle jobLock =
-                .and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId())
+                distributedJobLockService.tryLock("image-video:dispatch", java.time.Duration.ofSeconds(5));
-                        .or().isNull(ImageVideoAsyncTaskEntity::getOwnerInstanceId)
+        if (jobLock == null) {
-                        .or().eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, ""))
+            return;
-                .orderByAsc(ImageVideoAsyncTaskEntity::getId)
+        }
-                .last("LIMIT " + DISPATCH_BATCH_SIZE));
+        try (jobLock) {
-        tasks.forEach(task -> taskQueueExecutor.execute(() -> executeTask(task.getId())));
+            List<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>()
+                    .eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.PENDING.name())
+                    .and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId())
+                            .or().isNull(ImageVideoAsyncTaskEntity::getOwnerInstanceId)
+                            .or().eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, ""))
+                    .orderByAsc(ImageVideoAsyncTaskEntity::getId)
+                    .last("LIMIT " + DISPATCH_BATCH_SIZE));
+            tasks.forEach(task -> taskQueueExecutor.execute(() -> executeTask(task.getId())));
+        }
     }
 
     @Scheduled(fixedDelayString = "${aiimage.image-video.async-task-poll-delay-ms:5000}")
     public void pollWaitingTasks() {
-        List<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>()
+        DistributedJobLockService.LockHandle jobLock =
-                .eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.WAITING.name())
+                distributedJobLockService.tryLock("image-video:poll", java.time.Duration.ofSeconds(30));
-                .and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId())
+        if (jobLock == null) {
-                        .or().isNull(ImageVideoAsyncTaskEntity::getOwnerInstanceId)
+            return;
-                        .or().eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, ""))
+        }
-                .orderByAsc(ImageVideoAsyncTaskEntity::getUpdatedAt)
+        try (jobLock) {
-                .last("LIMIT " + POLL_BATCH_SIZE));
+            List<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>()
-        tasks.forEach(task -> taskQueueExecutor.execute(() -> pollTask(task.getId())));
+                    .eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.WAITING.name())
+                    .and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId())
+                            .or().isNull(ImageVideoAsyncTaskEntity::getOwnerInstanceId)
+                            .or().eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, ""))
+                    .orderByAsc(ImageVideoAsyncTaskEntity::getUpdatedAt)
+                    .last("LIMIT " + POLL_BATCH_SIZE));
+            tasks.forEach(task -> taskQueueExecutor.execute(() -> pollTask(task.getId())));
+        }
     }
 
     @EventListener(ApplicationReadyEvent.class)

backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java

@@ -50,6 +50,7 @@ import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopIndexService;
 import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import com.nanri.aiimage.common.service.DistributedJobLockService;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.dao.DuplicateKeyException;
 import org.springframework.scheduling.annotation.Scheduled;
@@ -108,6 +109,7 @@ public class PublishTaskService {
     private final OssStorageService ossStorageService;
     private final ObjectMapper objectMapper;
     private final TransactionTemplate transactionTemplate;
+    private final DistributedJobLockService distributedJobLockService;
     private final InstanceMetadata instanceMetadata;
     private final TaskProgressLightAssembler taskProgressLightAssembler;
 
@@ -420,32 +422,42 @@ public class PublishTaskService {
 
     @Scheduled(fixedDelayString = "${aiimage.publish.stale-scan-delay-ms:60000}")
     public void failStaleTasks() {
-        LocalDateTime threshold = LocalDateTime.now().minusMinutes(Math.max(5, staleTimeoutMinutes));
+        // 双节点互斥：避免 ownerInstanceId 为空的 stale 行被两节点同时扫出双写错误信息；
-        List<FileTaskEntity> staleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
+        // Redis 不可用时 fail-open，兜底靠下方 per-task 锁
-                .eq(FileTaskEntity::getModuleType, MODULE_TYPE)
+        DistributedJobLockService.LockHandle jobLock =
-                .eq(FileTaskEntity::getStatus, STATUS_RUNNING)
+                distributedJobLockService.tryLock("publish:stale-check", java.time.Duration.ofMinutes(2));
-                .and(owner -> owner
+        if (jobLock == null) {
-                        .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) IS NULL")
+            log.info("[publish] stale-check skipped, another instance holds the distributed lock");
-                        .or()
+            return;
-                        .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = ''")
+        }
-                        .or()
+        try (jobLock) {
-                        .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = {0}", currentInstanceId()))
+            LocalDateTime threshold = LocalDateTime.now().minusMinutes(Math.max(5, staleTimeoutMinutes));
-                .lt(FileTaskEntity::getUpdatedAt, threshold)
+            List<FileTaskEntity> staleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
-                .orderByAsc(FileTaskEntity::getUpdatedAt)
+                    .eq(FileTaskEntity::getModuleType, MODULE_TYPE)
-                .last("limit 100"));
+                    .eq(FileTaskEntity::getStatus, STATUS_RUNNING)
-        for (FileTaskEntity candidate : staleTasks) {
+                    .and(owner -> owner
-            if (taskFileJobService.countUnfinishedAssembleJobs(candidate.getId(), MODULE_TYPE) > 0L) {
+                            .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) IS NULL")
-                continue;
+                            .or()
-            }
+                            .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = ''")
-            try (TaskDistributedLockService.LockHandle lock =
+                            .or()
-                         taskDistributedLockService.acquire(MODULE_TYPE, candidate.getId(), 0L)) {
+                            .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = {0}", currentInstanceId()))
-                if (lock == null) {
+                    .lt(FileTaskEntity::getUpdatedAt, threshold)
+                    .orderByAsc(FileTaskEntity::getUpdatedAt)
+                    .last("limit 100"));
+            for (FileTaskEntity candidate : staleTasks) {
+                if (taskFileJobService.countUnfinishedAssembleJobs(candidate.getId(), MODULE_TYPE) > 0L) {
                     continue;
                 }
-                transactionTemplate.executeWithoutResult(status -> failStaleTaskLocked(candidate.getId(), threshold));
+                try (TaskDistributedLockService.LockHandle lock =
-            } catch (Exception ex) {
+                             taskDistributedLockService.acquire(MODULE_TYPE, candidate.getId(), 0L)) {
-                log.warn("[publish] stale task cleanup failed taskId={} msg={}",
+                    if (lock == null) {
-                        candidate.getId(), ex.getMessage());
+                        continue;
+                    }
+                    transactionTemplate.executeWithoutResult(status -> failStaleTaskLocked(candidate.getId(), threshold));
+                } catch (Exception ex) {
+                    log.warn("[publish] stale task cleanup failed taskId={} msg={}",
+                            candidate.getId(), ex.getMessage());
+                }
             }
         }
     }

backend-java/src/main/java/com/nanri/aiimage/modules/task/controller/TaskHeartbeatController.java

@@ -33,4 +33,17 @@ public class TaskHeartbeatController {
             @Valid @RequestBody(required = false) TaskHeartbeatRequest request) {
         return ApiResponse.success(taskHeartbeatService.heartbeat(taskId, request));
     }
+
+    @PostMapping("/{taskId}/interrupted")
+    @Operation(
+            summary = "上报客户端异常中断",
+            description = "客户端重启后发现自己上次进程崩溃时调用：将仍在 RUNNING 的任务立即标为终态"
+                    + "（file_task→FAILED，brand_crawl_tasks→cancelled），替代最长 30 分钟的 stale 兜底。幂等，非 RUNNING 状态不修改。")
+    public ApiResponse<TaskHeartbeatVo> interrupted(
+            @Parameter(description = "任务 ID", required = true, example = "200")
+            @PathVariable Long taskId,
+            @RequestBody(required = false) TaskHeartbeatRequest request) {
+        String reason = request == null ? null : request.getPhase();
+        return ApiResponse.success(taskHeartbeatService.markInterrupted(taskId, reason));
+    }
 }

backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java

@@ -117,6 +117,56 @@ public class TaskHeartbeatService {
         return fileTaskMapper.selectOne(fileQuery);
     }
 
+    /**
+     * 客户端崩溃恢复：客户端重启后发现自己上次进程级崩溃（journal 残留），
+     * 调此接口把上次还在 RUNNING 的任务立即标终态，替代最长等 30 分钟的 stale 兜底。
+     * file_task RUNNING → FAILED；brand_crawl_tasks running → cancelled；其余状态不动（幂等）。
+     */
+    public TaskHeartbeatVo markInterrupted(Long taskId, String reason) {
+        if (taskId == null || taskId <= 0) {
+            log.warn("[task-interrupted] ignored invalid taskId={}", taskId);
+            return TaskHeartbeatVo.notAlive(null, null, "invalid taskId");
+        }
+        String safeReason = (reason == null || reason.isBlank())
+                ? "客户端异常中断，任务已自动失败"
+                : "客户端异常中断: " + reason.trim();
+        FileTaskEntity fileTask = selectFileTask(taskId);
+        if (fileTask != null) {
+            String status = fileTask.getStatus();
+            int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper<FileTaskEntity>()
+                    .eq(FileTaskEntity::getId, taskId)
+                    .eq(FileTaskEntity::getStatus, STATUS_RUNNING)
+                    .set(FileTaskEntity::getStatus, "FAILED")
+                    .set(FileTaskEntity::getErrorMessage, safeReason)
+                    .set(FileTaskEntity::getFinishedAt, LocalDateTime.now()));
+            if (updated > 0) {
+                log.warn("[task-interrupted] file task marked failed by client restart taskId={} moduleType={} reason={}",
+                        taskId, fileTask.getModuleType(), safeReason);
+                return TaskHeartbeatVo.notAlive(fileTask.getModuleType(), "FAILED", "marked failed");
+            }
+            log.info("[task-interrupted] file task not in RUNNING, skipped taskId={} status={}", taskId, status);
+            return TaskHeartbeatVo.notAlive(fileTask.getModuleType(), status, "task is not running");
+        }
+        BrandCrawlTaskEntity brandTask = selectBrandTask(taskId);
+        if (brandTask != null) {
+            String status = brandTask.getStatus();
+            if ("running".equalsIgnoreCase(status) || "pending".equalsIgnoreCase(status)) {
+                brandTask.setStatus("cancelled");
+                brandTask.setErrorMessage(safeReason);
+                int updated = brandCrawlTaskMapper.updateById(brandTask);
+                if (updated > 0) {
+                    log.warn("[task-interrupted] brand task marked cancelled by client restart taskId={} reason={}",
+                            taskId, safeReason);
+                    return TaskHeartbeatVo.notAlive(MODULE_BRAND, "cancelled", "marked cancelled");
+                }
+            }
+            log.info("[task-interrupted] brand task not in running/pending, skipped taskId={} status={}", taskId, status);
+            return TaskHeartbeatVo.notAlive(MODULE_BRAND, status, "task is not running");
+        }
+        log.warn("[task-interrupted] task not found taskId={}", taskId);
+        return TaskHeartbeatVo.notAlive(null, null, "task not found");
+    }
+
     private BrandCrawlTaskEntity selectBrandTask(Long taskId) {
         LambdaQueryWrapper<BrandCrawlTaskEntity> brandQuery = new LambdaQueryWrapper<BrandCrawlTaskEntity>()
                 .eq(BrandCrawlTaskEntity::getId, taskId)

frontend-vue/src/pages/brand/components/BrandApiSecretSettingsButton.vue

@@ -163,7 +163,7 @@ import {
   type ApiSecretModuleKey,
   type ApiSecretRetention,
 } from '@/shared/utils/api-secret-store'
-import { getPywebviewApi, type ProxyMode } from '@/shared/bridges/pywebview'
+import { getPywebviewApi, type ProxyMode, type DesktopConfigUpdate } from '@/shared/bridges/pywebview'
 
 type SecretState = {
   value: string
@@ -275,6 +275,39 @@ function clearSecret(moduleKey: ApiSecretModuleKey) {
   ElMessage.success('已清空密钥')
 }
 
+function proxyUserId() {
+  if (typeof window === 'undefined') return '0'
+  return window.localStorage.getItem('uid') || '0'
+}
+
+/** 代理地址按登录用户隔离：proxy_users[uid]，各自计费各自复用；
+ *  未登录（uid=0）回退全局 proxy_url（兼容旧版本已保存的配置）。 */
+function readUserProxy(config: Record<string, unknown> | null | undefined) {
+  const uid = proxyUserId()
+  if (uid !== '0') {
+    const users = (config?.proxy_users ?? {}) as Record<string, unknown>
+    const own = (users[uid] ?? {}) as Record<string, unknown>
+    return {
+      url: typeof own.proxy_url === 'string' ? own.proxy_url : '',
+      mode: Number(own.proxy_mode) === 2 ? 2 : 1,
+    }
+  }
+  return {
+    url: typeof config?.proxy_url === 'string' ? config.proxy_url : '',
+    mode: Number(config?.proxy_mode) === 2 ? 2 : 1,
+  }
+}
+
+function userProxyPatch(nextProxyUrl: string, nextProxyMode: ProxyMode) {
+  const uid = proxyUserId()
+  if (uid === '0') {
+    return { proxy_url: nextProxyUrl, proxy_mode: nextProxyMode }
+  }
+  const users: Record<string, unknown> = {}
+  users[uid] = { proxy_url: nextProxyUrl, proxy_mode: nextProxyMode }
+  return { proxy_users: users }
+}
+
 async function loadProxyConfig() {
   const requestId = ++proxyLoadRequestId
   proxyLoading.value = true
@@ -294,10 +327,9 @@ async function loadProxyConfig() {
   try {
     const config = await api.read_config()
     if (requestId !== proxyLoadRequestId || !dialogVisible.value) return
-    const nextUrl = typeof config?.proxy_url === 'string' ? config.proxy_url : ''
+    const own = readUserProxy(config as Record<string, unknown>)
-    const nextMode: ProxyMode = Number(config?.proxy_mode) === 2 ? 2 : 1
+    proxyUrl.value = own.url
-    proxyUrl.value = nextUrl
+    proxyMode.value = (own.mode === 2 ? 2 : 1) as ProxyMode
-    proxyMode.value = nextMode
     proxyReady.value = true
   } catch (error) {
     if (requestId !== proxyLoadRequestId || !dialogVisible.value) return
@@ -331,10 +363,8 @@ async function saveAll() {
     if (shouldSaveProxy) {
       const api = getPywebviewApi()
       if (!api?.save_config) throw new Error('当前客户端未提供代理配置保存能力')
-      await api.save_config({
+      // 按登录用户保存：proxy_users[uid]（未登录回退全局 proxy_url 字段）
-        proxy_url: nextProxyUrl,
+      await api.save_config(userProxyPatch(nextProxyUrl, nextProxyMode) as DesktopConfigUpdate)
-        proxy_mode: nextProxyMode,
-      })
       proxyUrl.value = nextProxyUrl
       proxyDirty.value = false
     }

frontend-vue/src/pages/login/DesktopLoginPage.vue

@@ -333,6 +333,17 @@ async function submitLogin() {
       } catch {
         /* 忽略 */
       }
+      // 同步登录用户到桌面客户端配置：Python 端按 uid 选用该用户自己的
+      // 代理密钥/代理池（各自计费、各自复用省钱）。网页形态无 pywebview
+      // 桥，静默跳过。
+      try {
+        const bridge = (window as unknown as { pywebview?: { api?: { save_config?: (data: Record<string, unknown>) => Promise<unknown> } } }).pywebview
+        if (bridge?.api && typeof bridge.api.save_config === 'function') {
+          void bridge.api.save_config({ current_uid: String(data.userId) })
+        }
+      } catch {
+        /* 忽略 */
+      }
     }
     if (data.username) {
       try {
@@ -365,6 +376,15 @@ onMounted(() => {
     } catch {
       /* 忽略 */
     }
+    // 同步清掉客户端的登录用户标记：Python 端回退到全局/默认代理池
+    try {
+      const bridge = (window as unknown as { pywebview?: { api?: { save_config?: (data: Record<string, unknown>) => Promise<unknown> } } }).pywebview
+      if (bridge?.api && typeof bridge.api.save_config === 'function') {
+        void bridge.api.save_config({ current_uid: '' })
+      }
+    } catch {
+      /* 忽略 */
+    }
     // 显式登出/切号后不再自动登录（保留记住的账号，仅关闭自动登录）
     autoLogin.value = false
     lsSet(AUTO_LOGIN_KEY, '0')

frontend-vue/src/shared/bridges/pywebview.ts

@@ -56,6 +56,8 @@ export interface DesktopConfig {
 export interface DesktopConfigUpdate {
   proxy_url?: string;
   proxy_mode?: ProxyMode;
+  current_uid?: string;
+  proxy_users?: Record<string, { proxy_url?: string; proxy_mode?: number }>;
   aliprice_usename?: string;
   aliprice_pwd?: string;
 }

frontend-vue/src/shared/composables/useVersionUpdate.ts

@@ -8,6 +8,7 @@
 
 import { computed, ref } from 'vue'
 import { getPywebviewApi } from '@/shared/bridges/pywebview'
+import { compareVersions, normVersion } from '@/shared/utils/version-compare'
 import {
   normalizeUpdateProgress,
   type UpdateProgress,
@@ -32,10 +33,6 @@ function bindProgressListener() {
   })
 }
 
-function normVersion(value: unknown): string {
-  return String(value ?? '').trim().replace(/^[vV]/, '')
-}
-
 async function fetchJson(url: string): Promise<Record<string, unknown>> {
   const resp = await window.fetch(url, { credentials: 'same-origin' })
   if (!resp.ok) throw new Error(`HTTP ${resp.status}`)
@@ -116,11 +113,16 @@ export function useVersionUpdate() {
           : `线上最新版本 v${latest.version}，暂无下载地址`
         return
       }
-      hasUpdate.value = local !== latest.version
+      // 只有线上版本严格高于本机版本才算"有更新"：本机领先（灰度/回滚/漏发版）
+      // 时不能反过来提示更新，否则会把用户往低版本上引导。
+      const diff = compareVersions(latest.version, local)
+      hasUpdate.value = diff > 0
       canDownload.value = hasUpdate.value && Boolean(latest.fileUrl)
       hint.value = hasUpdate.value
         ? `发现新版本 v${latest.version}${latest.fileUrl ? '' : '，暂无下载地址'}`
-        : `已是最新版本 v${local}`
+        : diff < 0
+          ? `当前版本 v${local} 高于线上发布版本 v${latest.version}，无需更新`
+          : `已是最新版本 v${local}`
     } catch {
       hasUpdate.value = false
       canDownload.value = false

frontend-vue/src/shared/utils/version-compare.ts

@@ -0,0 +1,26 @@
+/**
+ * 客户端版本号归一化与比较（桌面端更新检测用）。
+ *
+ * 只比数字段、缺失段按 0 补齐（3.0.7 == 3.0.7.0）；非纯数字段（如
+ * 3.0.71-beta 的 beta）按 0 处理，避免 NaN 让比较结果乱序。
+ */
+
+/** 归一化版本号：去空白、去前缀 v/V，便于 'v3.0.71' 与 '3.0.71' 相等比较。 */
+export function normVersion(value: unknown): string {
+  return String(value ?? '').trim().replace(/^[vV]/, '')
+}
+
+/** 返回 >0 表示 a 比 b 新，<0 表示 a 比 b 旧，0 表示相同。 */
+export function compareVersions(a: string, b: string): number {
+  const pa = normVersion(a).split('.')
+  const pb = normVersion(b).split('.')
+  const len = Math.max(pa.length, pb.length)
+  for (let i = 0; i < len; i++) {
+    const na = parseInt(pa[i] ?? '', 10)
+    const nb = parseInt(pb[i] ?? '', 10)
+    const va = Number.isFinite(na) ? na : 0
+    const vb = Number.isFinite(nb) ? nb : 0
+    if (va !== vb) return va - vb
+  }
+  return 0
+}

frontend-vue/tests/version-compare.test.ts

@@ -0,0 +1,39 @@
+import { test } from 'node:test'
+import assert from 'node:assert/strict'
+
+import { compareVersions, normVersion } from '../src/shared/utils/version-compare.ts'
+
+test('版本比较：线上高于本地时返回正数（应提示更新）', () => {
+  assert.ok(compareVersions('3.0.71', '3.0.70') > 0)
+  assert.ok(compareVersions('3.1.0', '3.0.71') > 0)
+})
+
+test('版本比较：线上低于本地时返回负数（不得提示更新）', () => {
+  // 现网真实故障场景：本地已升到 3.0.71，线上发布记录仍是 3.0.70，
+  // 旧逻辑 local !== latest 会误报"发现新版本 v3.0.70"并诱导降级。
+  assert.ok(compareVersions('3.0.70', '3.0.71') < 0)
+  assert.ok(compareVersions('2.9.9', '3.0.0') < 0)
+})
+
+test('版本比较：完全相同（含 v 前缀、段数补齐）返回 0', () => {
+  assert.equal(compareVersions('3.0.71', '3.0.71'), 0)
+  assert.equal(compareVersions('v3.0.71', '3.0.71'), 0)
+  assert.equal(compareVersions('3.0.7', '3.0.7.0'), 0)
+})
+
+test('版本比较：按数字段而非字符串比较（3.0.9 < 3.0.10）', () => {
+  assert.ok(compareVersions('3.0.9', '3.0.10') < 0)
+  assert.ok(compareVersions('3.0.10', '3.0.9') > 0)
+})
+
+test('版本比较：非数字段按 0 处理，不得产生 NaN 乱序', () => {
+  assert.equal(compareVersions('3.0.71-beta', '3.0.71'), 0)
+  assert.ok(compareVersions('3.0.72', '3.0.71-beta') > 0)
+})
+
+test('版本归一化：去空白与 v/V 前缀', () => {
+  assert.equal(normVersion(' v3.0.71 '), '3.0.71')
+  assert.equal(normVersion('V3.0.71'), '3.0.71')
+  assert.equal(normVersion(null), '')
+  assert.equal(normVersion(undefined), '')
+})