Merge branch 'wk1'

Co-Authored-By: Claude Fable 5.1 <[email protected]>

apps/server/src/modules/acquisition-organize/handlers.ts

@@ -76,6 +76,7 @@ export function registerAoTasks(registry: TaskRegistry, d: AoDeps) {
     defineTask({
       name: "evaluateAcquisition",
       version: 1,
+      lane: "network",
       inputSchema: z.object({
         subscriptionId: id.optional(),
         acquisitionSlotId: id.optional(),
@@ -91,6 +92,7 @@ export function registerAoTasks(registry: TaskRegistry, d: AoDeps) {
     defineTask({
       name: "submitDownload",
       version: 1,
+      lane: "network",
       inputSchema: z.object({ acquisitionPlanId: id, taskId: id.optional() }),
       retryPolicy: NETWORK_RETRY,
       handler: async (t) => {
@@ -117,6 +119,7 @@ export function registerAoTasks(registry: TaskRegistry, d: AoDeps) {
     defineTask({
       name: "verifyAndImport",
       version: 1,
+      lane: "file",
       inputSchema: z.object({ downloadTaskId: id.optional(), recordId: id.optional(), taskId: id.optional() }),
       retryPolicy: FILE_RETRY,
       handler: async (t) => {
@@ -174,6 +177,7 @@ export function registerAoTasks(registry: TaskRegistry, d: AoDeps) {
     defineTask({
       name: "cleanupDownloadCache",
       version: 1,
+      lane: "file",
       inputSchema: z.object({ downloaderInstanceId: id.optional(), taskId: id.optional() }),
       retryPolicy: FILE_RETRY,
       handler: async (t) => {
@@ -186,6 +190,7 @@ export function registerAoTasks(registry: TaskRegistry, d: AoDeps) {
     defineTask({
       name: "reconcileLibrary",
       version: 1,
+      lane: "maintenance",
       inputSchema: z.object({ libraryRootId: id.optional(), reconcileRunId: id.optional(), taskId: id.optional() }),
       retryPolicy: FILE_RETRY,
       handler: async (t) => {

apps/server/src/modules/agent/tasks.ts

@@ -37,6 +37,7 @@ export function registerAgentTasks(registry: TaskRegistry, env: () => AgentEnv)
     defineTask({
       name: "runAgent",
       version: 1,
+      lane: "agent",
       /** upstreamId:覆盖默认上游(W7-A);缺省用 agent_run.upstream_id,再缺省 settings.pi.defaultUpstreamId */
       inputSchema: z.object({ agentRunId: id, upstreamId: id.optional() }),
       retryPolicy: DB_RETRY,
@@ -175,6 +176,7 @@ export function registerAgentTasks(registry: TaskRegistry, env: () => AgentEnv)
     defineTask({
       name: "resolveDecision",
       version: 1,
+      lane: "agent",
       inputSchema: z.object({ decisionId: id, trigger: z.enum(["created", "updated"]).default("updated") }),
       retryPolicy: DB_RETRY,
       handler: async (t) => {

apps/server/src/modules/auth-settings/index.ts

@@ -50,6 +50,7 @@ export function createAuthSettingsModule(opts: AuthSettingsModuleOptions = {}):
       if (everyMs <= 0) return;
       registerScheduledJob(ctx, {
         key: "auth.cleanup",
+        lane: "maintenance",
         defaults: {
           name: "登录会话清理",
           description: "删除 7 天前的登录尝试和 30 天前失效的会话",

apps/server/src/modules/cache/index.ts

@@ -181,6 +181,7 @@ export function createCacheModule(opts: CacheModuleOptions = {}): ServerModule &
     start(ctx: ServerContext) {
       registerScheduledJob(ctx, {
         key: "cache.sweep",
+        lane: "maintenance",
         defaults: {
           name: "缓存清理",
           description: "删除过期的 TMDB 与识别缓存",
@@ -192,6 +193,7 @@ export function createCacheModule(opts: CacheModuleOptions = {}): ServerModule &
       });
       registerScheduledJob(ctx, {
         key: "cache.images_clean",
+        lane: "maintenance",
         defaults: {
           name: "图片缓存清理",
           description: "删除过期的海报缓存",
@@ -206,6 +208,7 @@ export function createCacheModule(opts: CacheModuleOptions = {}): ServerModule &
       });
       registerScheduledJob(ctx, {
         key: "cache.images_lru",
+        lane: "maintenance",
         defaults: {
           name: "图片缓存容量控制",
           description: "海报缓存超过容量上限时，删掉最久没用的",

apps/server/src/modules/file-manager/index.ts

@@ -58,6 +58,7 @@ export function createFileManagerModule(overrides: FmOverrides = {}): ServerModu
         defineTask({
           name: "purgeTrash",
           version: 1,
+          lane: "maintenance",
           inputSchema: z.object({}).passthrough(),
           retryPolicy: { maximumAttempts: 3, initialInterval: "5s", backoffCoefficient: 2, maximumInterval: "60s" },
           handler: async (t) =>
@@ -70,6 +71,7 @@ export function createFileManagerModule(overrides: FmOverrides = {}): ServerModu
         defineTask({
           name: IDENTIFY_REMOTE_TASK,
           version: 1,
+          lane: "network",
           inputSchema: z.object({ body: FilesIdentifyRequest, actorId: z.string() }),
           retryPolicy: { maximumAttempts: 1 },
           handler: async (t) =>

apps/server/src/modules/pi/index.ts

@@ -60,6 +60,7 @@ export function createPiModule(
         defineTask({
           name: "refreshPiTokens",
           version: 1,
+          lane: "network",
           inputSchema: z.object({}).passthrough(),
           retryPolicy: { maximumAttempts: 2, initialInterval: "10s", backoffCoefficient: 2, maximumInterval: "60s" },
           handler: async (t) =>

apps/server/src/modules/schedules/registry.ts

@@ -48,6 +48,12 @@ export interface ScheduledJobDef {
   /** 稳定 key(表主键),如 feeds.poll */
   key: string;
   kind?: ScheduleKind;
+  /**
+   * 执行道(WK-1):"maintenance" = 在调度器里内联干重活的维护任务(清理、回填、保留期裁剪),
+   * 与 io worker 的维护道共用并发、文件道 / 网络道有活时让路(立即执行不让路)。
+   * 省略 = 轻量调度(只入队任务或很快结束),任务本身按它的 lane 执行。
+   */
+  lane?: "maintenance";
   defaults: ScheduledJobDefaults;
   run: JobRunFn;
 }

apps/server/src/modules/schedules/scheduler.ts

@@ -2,6 +2,7 @@ import { SCHEDULE_CHANGED_EVENT } from "@umt/contracts";
 import { type DB, type Kysely, type ScheduledJobRow, sql } from "@umt/db";
 import type { Logger } from "pino";
 import type { EventBusPort } from "../../http/sse.ts";
+import { laneControllerOf } from "../../worker/lanes.ts";
 import type { ServerContext } from "../types.ts";
 import { runAgentCheck } from "./agent-checks.ts";
 import type { JobRunFn, JobRunResult, ScheduleRegistry } from "./registry.ts";
@@ -158,10 +159,20 @@ export class Scheduler {
       .limit(50)
       .execute();
     const started: string[] = [];
+    const lanes = laneControllerOf(this.ctx);
     for (const job of due) {
       const fn = this.runFor(job);
       if (!fn) continue;
-      const p = this.execute(job, fn).finally(() => this.inflight.delete(job.id));
+      // 内联维护任务进维护道:占不到并发或正在让路就留到下一轮(不改 next_run_at);立即执行不让路
+      let release: (() => void) | null = null;
+      if (lanes && this.registry.get(job.id)?.lane === "maintenance") {
+        release = await lanes.enterMaintenance(job.run_requested_at !== null);
+        if (!release) continue;
+      }
+      const p = this.execute(job, fn).finally(() => {
+        release?.();
+        this.inflight.delete(job.id);
+      });
       this.inflight.set(job.id, p);
       started.push(job.id);
     }

apps/server/src/modules/sources-identity/index.ts

@@ -93,6 +93,7 @@ export function createSourcesIdentityModule(overrides: RuntimeOverrides = {}): S
       // FIX-5 存量元数据回刷:部署后首次启动即运行,跑完自动停用;可在计划页手动再执行
       registerScheduledJob(ctx, {
         key: BACKFILL_LANGUAGES_JOB,
+        lane: "maintenance",
         defaults: {
           name: "元数据语言回刷",
           description: "按语言设置重新拉一遍作品的标题、简介和每集信息；跑完自动停用，可手动再执行",

apps/server/src/modules/sources-identity/tasks.ts

@@ -29,6 +29,7 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu
     defineTask({
       name: "ingestFeed",
       version: 1,
+      lane: "network",
       inputSchema: z.object({ feedSourceId: id, manual: z.boolean().optional() }),
       retryPolicy: NETWORK_RETRY,
       handler: (t) => t.step.run("ingest", () => ingestFeed(rt(), t.input.feedSourceId, { manual: !!t.input.manual })),
@@ -36,6 +37,7 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu
     defineTask({
       name: "resolveObservation",
       version: 1,
+      lane: "network",
       inputSchema: z
         .object({
           observationId: id.nullish(),
@@ -60,6 +62,7 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu
     defineTask({
       name: AUTO_ENRICH_TASK,
       version: 1,
+      lane: "network",
       inputSchema: z.object({ poolEntryId: id }),
       retryPolicy: { maximumAttempts: 1 },
       handler: (t) => t.step.run("enrich", () => runAutoEnrich(rt().ctx, t.input.poolEntryId, t.runId, t.abortSignal)),
@@ -67,6 +70,7 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu
     defineTask({
       name: "refreshMetadata",
       version: 1,
+      lane: "network",
       inputSchema: z.object({ mediaEntityId: id, full: z.boolean().optional() }),
       retryPolicy: NETWORK_RETRY,
       handler: (t) =>
@@ -75,6 +79,7 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu
     defineTask({
       name: "searchSubscription",
       version: 1,
+      lane: "network",
       inputSchema: z.object({ subscriptionId: id, manual: z.boolean().optional() }),
       retryPolicy: NETWORK_RETRY,
       handler: (t) =>
@@ -83,6 +88,7 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu
     defineTask({
       name: "snapshotSite",
       version: 1,
+      lane: "network",
       inputSchema: z.object({
         siteId: id,
         kind: z.enum(["periodic", "backfill", "manual"]).default("periodic"),
@@ -97,6 +103,7 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu
     defineTask({
       name: "refreshSiteAccount",
       version: 1,
+      lane: "network",
       inputSchema: z.object({ accountId: id, mode: z.enum(["login", "userData"]).default("login") }),
       // 网络失败只记 last_error 不抛错;不自动重试(站点登录频繁重试有封号风险)
       retryPolicy: { maximumAttempts: 1 },

apps/server/src/modules/system/health.ts

@@ -1,12 +1,13 @@
 import { readdir, stat } from "node:fs/promises";
 import { checkRoot, DEFAULT_ROOT_SENTINEL, type DiskUsage, diskUsage } from "@umt/adapters";
-import type {
-  HealthLevel,
-  SystemDisk,
-  SystemHealth,
-  SystemHealthHistory,
-  SystemMount,
-  SystemProcess,
+import {
+  type HealthLevel,
+  type SystemDisk,
+  type SystemHealth,
+  type SystemHealthHistory,
+  type SystemMount,
+  type SystemProcess,
+  SystemWorkerLane,
 } from "@umt/contracts";
 import { type DB, type Kysely, type Selectable, sql } from "@umt/db";
 import { CURRENT_HOST_WINDOW_MS, HEARTBEAT_ROW_RETENTION_MS, HEARTBEAT_STALE_MS } from "./heartbeat.ts";
@@ -61,7 +62,13 @@ function toProcess(r: HeartbeatRow, now: Date): SystemProcess {
   const beat = toDate(r.last_beat_at);
   const age = Math.max(0, now.getTime() - beat.getTime());
   const extra = (r.extra ?? {}) as Record<string, unknown>;
-  const conc = (r.concurrency ?? {}) as { io?: number; agent?: number };
+  const conc = (r.concurrency ?? {}) as Record<string, unknown>;
+  const lanes = Array.isArray(extra.lanes)
+    ? extra.lanes.flatMap((l) => {
+        const p = SystemWorkerLane.safeParse(l);
+        return p.success ? [p.data] : [];
+      })
+    : [];
   const sse = extra.sseConnections;
   return {
     instanceId: r.instance_id,
@@ -76,10 +83,12 @@ function toProcess(r: HeartbeatRow, now: Date): SystemProcess {
     stoppedAt: r.stopped_at ? toDate(r.stopped_at).toISOString() : null,
     eventLoopLagMs: r.event_loop_lag_ms,
     eventLoopLagMaxMs: r.event_loop_lag_max_ms,
-    concurrency: {
-      ...(typeof conc.io === "number" ? { io: conc.io } : {}),
-      ...(typeof conc.agent === "number" ? { agent: conc.agent } : {}),
-    },
+    concurrency: Object.fromEntries(
+      (["io", "agent", "file", "network", "maintenance"] as const).flatMap((k) =>
+        typeof conc[k] === "number" ? [[k, conc[k]]] : [],
+      ),
+    ),
+    lanes,
     runningRuns: r.running_runs,
     sseConnections: typeof sse === "number" ? sse : null,
     extra,

apps/server/src/modules/system/heartbeat.ts

@@ -51,7 +51,8 @@ export function createLagSampler(intervalMs = 1000) {
 }
 
 export interface HeartbeatSample {
-  concurrency?: { io?: number; agent?: number };
+  /** io(各道合计)/ agent,以及 io 组的 file / network / maintenance */
+  concurrency?: Partial<Record<"io" | "agent" | "file" | "network" | "maintenance", number>>;
   runningRuns?: number | null;
   sseConnections?: number | null;
   extra?: Record<string, unknown>;

apps/server/src/modules/system/index.ts

@@ -185,6 +185,7 @@ export function createSystemModule(opts: SystemModuleOptions = {}): ServerModule
         for (const j of RETENTION_JOBS) {
           registerScheduledJob(ctx, {
             key: j.jobId,
+            lane: "maintenance",
             defaults: {
               name: j.name,
               description: j.description,
@@ -208,6 +209,7 @@ export function createSystemModule(opts: SystemModuleOptions = {}): ServerModule
       if (everyMs <= 0) return;
       registerScheduledJob(ctx, {
         key: "system.heartbeat_cleanup",
+        lane: "maintenance",
         defaults: {
           name: "进程心跳清理",
           description: "删除过期的进程状态记录",